From 55bba60c0aba80b24a4478d90047485adddb46b9 Mon Sep 17 00:00:00 2001 From: sch Date: Mon, 1 Jun 2026 19:14:03 +0800 Subject: [PATCH 01/11] feat: callback rate limiting and jitter to isolate EAS contention - Add WorkerPoolSize, RateLimitQPS, JitterMaxMs, DropOnBackpressure to CallBackConfig - Refactor CallBackControllerHandler with configurable pool/buffer size - Add backpressure detection: drop on full channel instead of blocking - Add rate limiter (golang.org/x/time/rate) before callback EAS calls - Add random jitter delay to spread callback load - Expose CallbackPending() for metrics --- go.mod | 1 + go.sum | 2 ++ recconf/recconf.go | 18 +++++++----- service/call_back_service.go | 20 ++++++++++++++ web/callback_controller_handler.go | 44 +++++++++++++++++++++++------- 5 files changed, 68 insertions(+), 17 deletions(-) diff --git a/go.mod b/go.mod index 5f14a54b..e6f34c4d 100644 --- a/go.mod +++ b/go.mod @@ -52,6 +52,7 @@ require ( github.com/google/uuid v1.3.0 github.com/mmcloughlin/geohash v0.10.0 go.uber.org/automaxprocs v1.5.2 + golang.org/x/time v0.11.0 ) require ( diff --git a/go.sum b/go.sum index 91fb8025..244f3931 100644 --- a/go.sum +++ b/go.sum @@ -769,6 +769,8 @@ golang.org/x/text v0.30.0 h1:yznKA/E9zq54KzlzBEAWn1NXSQ8DIp/NYMy88xJjl4k= golang.org/x/text v0.30.0/go.mod h1:yDdHFIX9t+tORqspjENWgzaCVXgk0yYnYuSZ8UzzBVM= golang.org/x/time v0.0.0-20180412165947-fbb02b2291d2/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= +golang.org/x/time v0.11.0 h1:/bpjEDfN9tkoN/ryeYHnv5hcMlc8ncjMcM4XBk5NWV0= +golang.org/x/time v0.11.0/go.mod h1:CDIdPxbZBQxdj6cxyCIdrNogrJKMJ7pr37NYpMcMDSg= golang.org/x/tools v0.0.0-20180221164845-07fd8470d635/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= golang.org/x/tools v0.0.0-20180525024113-a5b4c53f6e8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= golang.org/x/tools v0.0.0-20180828015842-6cd1fcedba52/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= diff --git a/recconf/recconf.go b/recconf/recconf.go index f9f1d1bc..d64e1fc6 100644 --- a/recconf/recconf.go +++ b/recconf/recconf.go @@ -919,13 +919,17 @@ type AdjustCountConfig struct { Weight int } type CallBackConfig struct { - DataSource DataSourceConfig - RankConf RankConfig - RawFeatures bool - RawFeaturesRate int - ItemSize int - ItemSizeRate int - UseUserFeatures bool + DataSource DataSourceConfig + RankConf RankConfig + RawFeatures bool + RawFeaturesRate int + ItemSize int + ItemSizeRate int + UseUserFeatures bool + WorkerPoolSize int `json:"WorkerPoolSize"` + RateLimitQPS float64 `json:"RateLimitQPS"` + JitterMaxMs int `json:"JitterMaxMs"` + DropOnBackpressure bool `json:"DropOnBackpressure"` } type EmbeddingConfig struct { DataSource DataSourceConfig diff --git a/service/call_back_service.go b/service/call_back_service.go index 13fa524f..2a09445a 100644 --- a/service/call_back_service.go +++ b/service/call_back_service.go @@ -1,6 +1,7 @@ package service import ( + gocontext "context" "encoding/json" "fmt" "math/rand" @@ -21,8 +22,20 @@ import ( "github.com/alibaba/pairec/v2/service/feature" "github.com/alibaba/pairec/v2/service/rank" "github.com/alibaba/pairec/v2/utils" + "golang.org/x/time/rate" ) +var callbackLimiters sync.Map + +func getCallbackLimiter(scene string, qps float64) *rate.Limiter { + if v, ok := callbackLimiters.Load(scene); ok { + return v.(*rate.Limiter) + } + limiter := rate.NewLimiter(rate.Limit(qps), int(qps)) + actual, _ := callbackLimiters.LoadOrStore(scene, limiter) + return actual.(*rate.Limiter) +} + type CallBackService struct { recallService *RecallService featureService *feature.FeatureService @@ -239,6 +252,13 @@ func (r *CallBackService) Rank(context *context.RecommendContext) { } } } + if callBackConfig.RateLimitQPS > 0 { + limiter := getCallbackLimiter(scene_name, callBackConfig.RateLimitQPS) + _ = limiter.Wait(gocontext.Background()) + } + if callBackConfig.JitterMaxMs > 0 { + time.Sleep(time.Duration(rand.Intn(callBackConfig.JitterMaxMs)) * time.Millisecond) + } // run 返回原始的值，然后处理返回数据// 注册配置 ret, err := algorithm.Run(newAlgoName, algoData.GetFeatures()) if err != nil { diff --git a/web/callback_controller_handler.go b/web/callback_controller_handler.go index 9cadd4be..3f571a59 100644 --- a/web/callback_controller_handler.go +++ b/web/callback_controller_handler.go @@ -3,8 +3,9 @@ package web import ( "fmt" "sync" + "sync/atomic" - plog "github.com/alibaba/pairec/v2/log" + "github.com/alibaba/pairec/v2/log" "github.com/alibaba/pairec/v2/utils" ) @@ -14,14 +15,20 @@ var once sync.Once type CallBackControllerHandler struct { controllerCh chan *CallBackController poolSize int + pending atomic.Int64 } -func NewCallBackControllerHandler() *CallBackControllerHandler { +func NewCallBackControllerHandler(poolSize int, bufferSize int) *CallBackControllerHandler { + if poolSize <= 0 { + poolSize = 20 + } + if bufferSize <= 0 { + bufferSize = 5000 + } handler := &CallBackControllerHandler{ - controllerCh: make(chan *CallBackController, 5000), - poolSize: 20, + controllerCh: make(chan *CallBackController, bufferSize), + poolSize: poolSize, } - handler.start() return handler } @@ -31,19 +38,36 @@ func (h *CallBackControllerHandler) start() { go func() { for controller := range h.controllerCh { controller.doCallbackLog() + h.pending.Add(-1) } }() } } +func (h *CallBackControllerHandler) Pending() int64 { + return h.pending.Load() +} + +func CallbackPending() int64 { + if callBackControllerHandler == nil { + return 0 + } + return callBackControllerHandler.Pending() +} + func Send(controller *CallBackController) { if callBackControllerHandler == nil { once.Do(func() { - callBackControllerHandler = NewCallBackControllerHandler() + callBackControllerHandler = NewCallBackControllerHandler(20, 5000) }) } - callBackControllerHandler.controllerCh <- controller + select { + case callBackControllerHandler.controllerCh <- controller: + callBackControllerHandler.pending.Add(1) + default: + log.Warning("callback channel full, dropping request") + } } // SendDirect bypasses the HTTP self-call path used by the external @@ -67,7 +91,7 @@ func SendDirect(param *CallBackParam) { // so a panic here would crash the request goroutine. Logging the error // keeps the misuse visible without taking down the request. if param == nil { - plog.Error("event=SendDirect\terror=nil param") + log.Error("event=SendDirect\terror=nil param") return } @@ -77,13 +101,13 @@ func SendDirect(param *CallBackParam) { // at a nil algoData. Owning the invariant here makes SendDirect spec- // equivalent to the HTTP entry, so any future caller is automatically safe. if len(param.ItemList) == 0 { - plog.Info(fmt.Sprintf("requestId=%s\tevent=SendDirect\tmsg=empty item list, skip", param.RequestId)) + log.Info(fmt.Sprintf("requestId=%s\tevent=SendDirect\tmsg=empty item list, skip", param.RequestId)) return } if callBackControllerHandler == nil { once.Do(func() { - callBackControllerHandler = NewCallBackControllerHandler() + callBackControllerHandler = NewCallBackControllerHandler(20, 5000) }) } From 5020fe02c1bdcabd9898637e8c5e5487dc68f3c5 Mon Sep 17 00:00:00 2001 From: sch Date: Mon, 1 Jun 2026 19:15:43 +0800 Subject: [PATCH 02/11] perf: reduce alloc in AlgoDataGenerator - share userFeatures, preallocate slices - AlgoDataGenerator.AddFeatures: store userFeatures reference instead of copying per item - AlgoDataGenerator.GeneratorAlgoData: merge userFeatures at generation time - EasyrecAlgoDataGenerator: preallocate contextFeatures slice capacity to 128 --- service/rank/algo_data.go | 31 +++++++++++++++++++++---------- 1 file changed, 21 insertions(+), 10 deletions(-) diff --git a/service/rank/algo_data.go b/service/rank/algo_data.go index 031f10ef..c8348ada 100644 --- a/service/rank/algo_data.go +++ b/service/rank/algo_data.go @@ -86,8 +86,9 @@ func (d *EasyrecAlgoData) GetFeatures() interface{} { } type AlgoDataGenerator struct { - requestData []map[string]interface{} - requestItem []*module.Item + requestData []map[string]interface{} + requestItem []*module.Item + userFeatures map[string]interface{} } // SetItemFeatures implements IAlgoDataGenerator. @@ -105,22 +106,31 @@ func (g *AlgoDataGenerator) AddFeatures(item *module.Item, itemFeatures map[stri if item != nil { g.requestItem = append(g.requestItem, item) } - - features := make(map[string]interface{}, len(itemFeatures)+len(userFeatures)) - for k, v := range userFeatures { - features[k] = v + if g.userFeatures == nil { + g.userFeatures = userFeatures } + + features := make(map[string]interface{}, len(itemFeatures)) for k, v := range itemFeatures { features[k] = v } - g.requestData = append(g.requestData, features) } func (g *AlgoDataGenerator) GeneratorAlgoData() IAlgoData { copydata := make([]map[string]interface{}, len(g.requestData)) + for i, itemFeatures := range g.requestData { + merged := make(map[string]interface{}, len(itemFeatures)+len(g.userFeatures)) + for k, v := range g.userFeatures { + merged[k] = v + } + for k, v := range itemFeatures { + merged[k] = v + } + copydata[i] = merged + } + copyItems := make([]*module.Item, len(g.requestItem)) copy(copyItems, g.requestItem) - copy(copydata, g.requestData) algoData := &AlgoData{ AlgoDataBase: &AlgoDataBase{ @@ -183,8 +193,8 @@ type EasyrecAlgoDataGenerator struct { func NewEasyrecAlgoDataGenerator(contextFeatures []string) *EasyrecAlgoDataGenerator { generator := &EasyrecAlgoDataGenerator{ - requestItem: make([]*module.Item, 0, 100), - contextFeatures: make(map[string][]interface{}, 8), + requestItem: make([]*module.Item, 0, 128), + contextFeatures: make(map[string][]interface{}, len(contextFeatures)+8), parseFeature: true, } @@ -195,6 +205,7 @@ func NewEasyrecAlgoDataGenerator(contextFeatures []string) *EasyrecAlgoDataGener valueType: reflect.TypeOf(""), } generator.itemFeatures = append(generator.itemFeatures, feature) + generator.contextFeatures[featureName] = make([]interface{}, 0, 128) } } From a3b24b519c345f550f035989daaaa6d5326bdc10 Mon Sep 17 00:00:00 2001 From: sch Date: Mon, 1 Jun 2026 19:16:32 +0800 Subject: [PATCH 03/11] perf: transfer slice ownership instead of copy in EasyrecAlgoDataGenerator GeneratorAlgoData and GeneratorAlgoDataDebugWithLevel now transfer requestItem slice ownership directly instead of allocating + copying, since the generator resets immediately after. --- service/rank/algo_data.go | 20 ++++++++++---------- 1 file changed, 10 insertions(+), 10 deletions(-) diff --git a/service/rank/algo_data.go b/service/rank/algo_data.go index c8348ada..14d45b74 100644 --- a/service/rank/algo_data.go +++ b/service/rank/algo_data.go @@ -285,14 +285,15 @@ func (g *EasyrecAlgoDataGenerator) AddFeatures(item *module.Item, itemFeatures m } func (g *EasyrecAlgoDataGenerator) GeneratorAlgoData() IAlgoData { - copyItems := make([]*module.Item, len(g.requestItem)) - copy(copyItems, g.requestItem) + // Transfer ownership instead of copy — generator resets immediately after + items := g.requestItem + g.requestItem = make([]*module.Item, 0, 128) builder := easyrec.NewEasyrecRequestBuilder() for k, v := range g.userFeatures { builder.AddUserFeature(k, v) } - for _, item := range g.requestItem { + for _, item := range items { builder.AddItemId(string(item.Id)) } for k, v := range g.contextFeatures { @@ -306,13 +307,12 @@ func (g *EasyrecAlgoDataGenerator) GeneratorAlgoData() IAlgoData { algoData := &EasyrecAlgoData{ AlgoDataBase: &AlgoDataBase{ - Items: copyItems, + Items: items, AlgoResult: make(map[string][]response.AlgoResponse), }, easyrecRequest: builder.EasyrecRequest(), } - g.requestItem = g.requestItem[:0] return algoData } @@ -321,8 +321,9 @@ func (g *EasyrecAlgoDataGenerator) GeneratorAlgoDataDebug() IAlgoData { } func (g *EasyrecAlgoDataGenerator) GeneratorAlgoDataDebugWithLevel(level int, meta map[string]string) IAlgoData { - copyItems := make([]*module.Item, len(g.requestItem)) - copy(copyItems, g.requestItem) + // Transfer ownership instead of copy — generator resets immediately after + items := g.requestItem + g.requestItem = make([]*module.Item, 0, 128) builder := easyrec.NewEasyrecRequestBuilderDebugWithLevel(level) if meta != nil && len(meta) > 0 { @@ -332,7 +333,7 @@ func (g *EasyrecAlgoDataGenerator) GeneratorAlgoDataDebugWithLevel(level int, me for k, v := range g.userFeatures { builder.AddUserFeature(k, v) } - for _, item := range g.requestItem { + for _, item := range items { builder.AddItemId(string(item.Id)) } for k, v := range g.contextFeatures { @@ -346,13 +347,12 @@ func (g *EasyrecAlgoDataGenerator) GeneratorAlgoDataDebugWithLevel(level int, me algoData := &EasyrecAlgoData{ AlgoDataBase: &AlgoDataBase{ - Items: copyItems, + Items: items, AlgoResult: make(map[string][]response.AlgoResponse), }, easyrecRequest: builder.EasyrecRequest(), } - g.requestItem = g.requestItem[:0] return algoData } From ee0c2d0eb9eda94c1f2b05e1dbe3e349ef3c9cf5 Mon Sep 17 00:00:00 2001 From: sch Date: Mon, 1 Jun 2026 19:16:49 +0800 Subject: [PATCH 04/11] perf: increase default batchCount from 100 to 200 to reduce EAS calls Reduces the number of parallel EAS invocations per rank request. Still configurable via RankConfig.BatchCount for per-scene tuning. --- service/rank/rank_service.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/service/rank/rank_service.go b/service/rank/rank_service.go index 2a541303..bff7debc 100644 --- a/service/rank/rank_service.go +++ b/service/rank/rank_service.go @@ -160,7 +160,7 @@ func (r *RankService) Rank(user *module.User, items []*module.Item, context *con rankConfig.RankScore = rankscore } - batchCount := 100 + batchCount := 200 if rankConfig.BatchCount > 0 { batchCount = rankConfig.BatchCount } From 29b526269936c1779f392a5d394db861b42cef66 Mon Sep 17 00:00:00 2001 From: sch Date: Mon, 1 Jun 2026 19:17:30 +0800 Subject: [PATCH 05/11] perf: reuse proto marshal buffer via sync.Pool in EasyrecRequest Use proto.Buffer from sync.Pool to avoid allocating a new byte slice on every proto.Marshal call. Buffer is returned after BytesPredict. --- algorithm/eas/easyrec_request.go | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/algorithm/eas/easyrec_request.go b/algorithm/eas/easyrec_request.go index 1fb8ba12..8ba38fc0 100644 --- a/algorithm/eas/easyrec_request.go +++ b/algorithm/eas/easyrec_request.go @@ -5,6 +5,7 @@ import ( "fmt" "os" "strings" + "sync" "github.com/alibaba/pairec/v2/algorithm/eas/easyrec" "github.com/alibaba/pairec/v2/config" @@ -12,6 +13,13 @@ import ( proto "github.com/golang/protobuf/proto" ) +// marshalBufPool reuses proto.Buffer to reduce marshal allocations +var marshalBufPool = sync.Pool{ + New: func() interface{} { + return proto.NewBuffer(make([]byte, 0, 4096)) + }, +} + type EasyrecRequest struct { EasRequest EasClient *eas.PredictClient @@ -24,7 +32,14 @@ func (r *EasyrecRequest) Invoke(requestData interface{}) (response interface{}, return } - data, _ := proto.Marshal(request) + buf := marshalBufPool.Get().(*proto.Buffer) + buf.Reset() + if err = buf.Marshal(request); err != nil { + marshalBufPool.Put(buf) + return + } + data := buf.Bytes() + if config.AppConfig.WarmUpData { warmupFunc := func(data []byte) { if file, err := os.OpenFile("warm_up.bin", os.O_CREATE|os.O_TRUNC|os.O_RDWR, 0664); err == nil { @@ -36,6 +51,7 @@ func (r *EasyrecRequest) Invoke(requestData interface{}) (response interface{}, } body, err := r.EasClient.BytesPredict(data) + marshalBufPool.Put(buf) if err != nil { return } From 8e38ab35fc5bc651b364f445901124ab42ba1b89 Mon Sep 17 00:00:00 2001 From: sch Date: Mon, 1 Jun 2026 19:28:46 +0800 Subject: [PATCH 06/11] feat: expose callback_channel_pending prometheus gauge Wire CallbackPending() into metrics via function pointer to avoid import cycles. Gauge reports current channel backlog for alerting. --- service/metrics/metrics.go | 44 ++++++++++++++++++++---------- web/callback_controller_handler.go | 5 ++++ 2 files changed, 35 insertions(+), 14 deletions(-) diff --git a/service/metrics/metrics.go b/service/metrics/metrics.go index 35ea8fc9..d476dd3d 100644 --- a/service/metrics/metrics.go +++ b/service/metrics/metrics.go @@ -11,24 +11,28 @@ import ( ) var ( - SizeNotEnoughTotal *prometheus.CounterVec - RecTotal *prometheus.CounterVec - RecallCount *prometheus.CounterVec - RecallCountTotal *prometheus.CounterVec - RecallItemsPercentage *prometheus.GaugeVec - RecallDurSecs *prometheus.HistogramVec - FilterDurSecs *prometheus.HistogramVec - GeneralRankDurSecs *prometheus.HistogramVec - LoadFeatureDurSecs *prometheus.HistogramVec - RankDurSecs *prometheus.HistogramVec - SortDurSecs *prometheus.HistogramVec - RecDurSecs *prometheus.HistogramVec - FallbackTotal *prometheus.CounterVec + SizeNotEnoughTotal *prometheus.CounterVec + RecTotal *prometheus.CounterVec + RecallCount *prometheus.CounterVec + RecallCountTotal *prometheus.CounterVec + RecallItemsPercentage *prometheus.GaugeVec + RecallDurSecs *prometheus.HistogramVec + FilterDurSecs *prometheus.HistogramVec + GeneralRankDurSecs *prometheus.HistogramVec + LoadFeatureDurSecs *prometheus.HistogramVec + RankDurSecs *prometheus.HistogramVec + SortDurSecs *prometheus.HistogramVec + RecDurSecs *prometheus.HistogramVec + FallbackTotal *prometheus.CounterVec + CallbackChannelPending prometheus.GaugeFunc enabled = false once sync.Once ) +// CallbackPendingFunc is set by the web package to avoid import cycle +var CallbackPendingFunc func() int64 + var CustomRegister = prometheus.NewRegistry() func Enabled() bool { @@ -51,7 +55,8 @@ func Load(conf *recconf.RecommendConfig) { LoadFeatureDurSecs, RankDurSecs, SortDurSecs, - FallbackTotal) + FallbackTotal, + CallbackChannelPending) }) enabled = conf.PrometheusConfig.Enable @@ -173,4 +178,15 @@ func initMetrics(conf *recconf.RecommendConfig) { Name: "fallback_total", Help: "How many times of fallback recommend.", }, []string{"scene"}) + + CallbackChannelPending = prometheus.NewGaugeFunc(prometheus.GaugeOpts{ + Subsystem: subsystem, + Name: "callback_channel_pending", + Help: "Number of pending callback requests in channel.", + }, func() float64 { + if CallbackPendingFunc != nil { + return float64(CallbackPendingFunc()) + } + return 0 + }) } diff --git a/web/callback_controller_handler.go b/web/callback_controller_handler.go index 3f571a59..e70ce323 100644 --- a/web/callback_controller_handler.go +++ b/web/callback_controller_handler.go @@ -6,12 +6,17 @@ import ( "sync/atomic" "github.com/alibaba/pairec/v2/log" + "github.com/alibaba/pairec/v2/service/metrics" "github.com/alibaba/pairec/v2/utils" ) var callBackControllerHandler *CallBackControllerHandler var once sync.Once +func init() { + metrics.CallbackPendingFunc = CallbackPending +} + type CallBackControllerHandler struct { controllerCh chan *CallBackController poolSize int From 0ca66a96aa1b729296db06e3299b20690ba16aa3 Mon Sep 17 00:00:00 2001 From: sch Date: Mon, 1 Jun 2026 23:06:27 +0800 Subject: [PATCH 07/11] fix: address code review findings on callback optimization High: Wire CallBackConfig.WorkerPoolSize and DropOnBackpressure into runtime via InitHandler() called from configloader. Without config, falls back to 20 workers + blocking (no drop). Medium: Rate limiter burst clamped to min 1, so sub-1 QPS values (e.g. 0.5) now correctly throttle instead of silently passing through. Low: Pending counter incremented BEFORE enqueue and decremented AFTER processing, eliminating transient negative values under high concurrency. --- configloader.go | 13 +++++++++ service/call_back_service.go | 6 ++++- web/callback_controller_handler.go | 43 +++++++++++++++++++++--------- 3 files changed, 49 insertions(+), 13 deletions(-) diff --git a/configloader.go b/configloader.go index cb679345..69f3db49 100644 --- a/configloader.go +++ b/configloader.go @@ -39,6 +39,7 @@ import ( "github.com/alibaba/pairec/v2/service/recall/berecall" "github.com/alibaba/pairec/v2/sort" "github.com/alibaba/pairec/v2/utils" + "github.com/alibaba/pairec/v2/web" ) var ( @@ -118,6 +119,7 @@ func (l *ConfigLoader) reloadConfig(config *recconf.RecommendConfig) { general_rank.LoadGeneralRankWithConfig(config) pipeline.LoadPipelineConfigs(config) fallback.LoadFallbackConfig(config) + initCallbackHandler(config) } func (l *ConfigLoader) loadConfigFromConfigServer() (*recconf.RecommendConfig, error) { @@ -175,3 +177,14 @@ func ListenConfig(configName string) { go loader.loopLoadConfig() } + +// initCallbackHandler initializes the callback worker pool from config. +// Uses the first CallBackConfig found for pool settings (pool is global). +func initCallbackHandler(config *recconf.RecommendConfig) { + for _, cbConf := range config.CallBackConfs { + poolSize := cbConf.WorkerPoolSize + drop := cbConf.DropOnBackpressure + web.InitHandler(poolSize, 5000, drop) + return + } +} diff --git a/service/call_back_service.go b/service/call_back_service.go index 2a09445a..02bf1b61 100644 --- a/service/call_back_service.go +++ b/service/call_back_service.go @@ -31,7 +31,11 @@ func getCallbackLimiter(scene string, qps float64) *rate.Limiter { if v, ok := callbackLimiters.Load(scene); ok { return v.(*rate.Limiter) } - limiter := rate.NewLimiter(rate.Limit(qps), int(qps)) + burst := int(qps) + if burst < 1 { + burst = 1 + } + limiter := rate.NewLimiter(rate.Limit(qps), burst) actual, _ := callbackLimiters.LoadOrStore(scene, limiter) return actual.(*rate.Limiter) } diff --git a/web/callback_controller_handler.go b/web/callback_controller_handler.go index e70ce323..04af4c2b 100644 --- a/web/callback_controller_handler.go +++ b/web/callback_controller_handler.go @@ -18,12 +18,13 @@ func init() { } type CallBackControllerHandler struct { - controllerCh chan *CallBackController - poolSize int - pending atomic.Int64 + controllerCh chan *CallBackController + poolSize int + pending atomic.Int64 + dropOnBackpressure bool } -func NewCallBackControllerHandler(poolSize int, bufferSize int) *CallBackControllerHandler { +func NewCallBackControllerHandler(poolSize int, bufferSize int, dropOnBackpressure bool) *CallBackControllerHandler { if poolSize <= 0 { poolSize = 20 } @@ -31,8 +32,9 @@ func NewCallBackControllerHandler(poolSize int, bufferSize int) *CallBackControl bufferSize = 5000 } handler := &CallBackControllerHandler{ - controllerCh: make(chan *CallBackController, bufferSize), - poolSize: poolSize, + controllerCh: make(chan *CallBackController, bufferSize), + poolSize: poolSize, + dropOnBackpressure: dropOnBackpressure, } handler.start() return handler @@ -60,18 +62,35 @@ func CallbackPending() int64 { return callBackControllerHandler.Pending() } +// InitHandler initializes the callback handler with config values. +// Must be called during service startup after config is loaded. +func InitHandler(poolSize int, bufferSize int, dropOnBackpressure bool) { + once.Do(func() { + callBackControllerHandler = NewCallBackControllerHandler(poolSize, bufferSize, dropOnBackpressure) + }) +} + func Send(controller *CallBackController) { if callBackControllerHandler == nil { once.Do(func() { - callBackControllerHandler = NewCallBackControllerHandler(20, 5000) + callBackControllerHandler = NewCallBackControllerHandler(20, 5000, false) }) } - select { - case callBackControllerHandler.controllerCh <- controller: - callBackControllerHandler.pending.Add(1) - default: - log.Warning("callback channel full, dropping request") + // Increment pending BEFORE enqueue to avoid transient negative values + h := callBackControllerHandler + if h.dropOnBackpressure { + h.pending.Add(1) + select { + case h.controllerCh <- controller: + // already counted + default: + h.pending.Add(-1) + log.Warning("callback channel full, dropping request") + } + } else { + h.pending.Add(1) + h.controllerCh <- controller // block if full } } From 46aec3f84c917a4d046fbf3b8ad6d4d63fc2d67f Mon Sep 17 00:00:00 2001 From: sch Date: Mon, 1 Jun 2026 23:13:59 +0800 Subject: [PATCH 08/11] fix: address P1/P3 from code review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P1: limiter.Wait now has 2s timeout — prevents indefinite goroutine blocking when QPS is very low. On timeout, skips the EAS call entirely (callback is best-effort). P3: proto.Buffer pool caps at 1MB — oversized buffers from outlier requests are discarded instead of retained in the pool. --- algorithm/eas/easyrec_request.go | 7 ++++++- service/call_back_service.go | 8 +++++++- 2 files changed, 13 insertions(+), 2 deletions(-) diff --git a/algorithm/eas/easyrec_request.go b/algorithm/eas/easyrec_request.go index 8ba38fc0..af414d5e 100644 --- a/algorithm/eas/easyrec_request.go +++ b/algorithm/eas/easyrec_request.go @@ -51,7 +51,12 @@ func (r *EasyrecRequest) Invoke(requestData interface{}) (response interface{}, } body, err := r.EasClient.BytesPredict(data) - marshalBufPool.Put(buf) + // Cap oversized buffers to avoid retaining large allocations in pool + if cap(buf.Bytes()) > 1<<20 { + marshalBufPool.Put(proto.NewBuffer(make([]byte, 0, 4096))) + } else { + marshalBufPool.Put(buf) + } if err != nil { return } diff --git a/service/call_back_service.go b/service/call_back_service.go index 02bf1b61..d21d9209 100644 --- a/service/call_back_service.go +++ b/service/call_back_service.go @@ -258,7 +258,13 @@ func (r *CallBackService) Rank(context *context.RecommendContext) { } if callBackConfig.RateLimitQPS > 0 { limiter := getCallbackLimiter(scene_name, callBackConfig.RateLimitQPS) - _ = limiter.Wait(gocontext.Background()) + ctx, cancel := gocontext.WithTimeout(gocontext.Background(), 2*time.Second) + if err := limiter.Wait(ctx); err != nil { + cancel() + log.Warning(fmt.Sprintf("requestId=%s\tcallback rate limit timeout, skipping EAS call", context.RecommendId)) + return + } + cancel() } if callBackConfig.JitterMaxMs > 0 { time.Sleep(time.Duration(rand.Intn(callBackConfig.JitterMaxMs)) * time.Millisecond) From 0c3480142b594b14835050e41ba911b8207dd782 Mon Sep 17 00:00:00 2001 From: sch Date: Mon, 1 Jun 2026 23:19:53 +0800 Subject: [PATCH 09/11] fix: deterministic config, init race, and pool hygiene 1. Config: aggregate max WorkerPoolSize across all scenes instead of taking random first map entry. DropOnBackpressure is OR'd. 2. Init race: InitHandler and Send share one sync.Once (initOnce). Whoever calls first wins. If InitHandler is called with real config before any Send, config takes effect. If traffic arrives first, defaults are used (same Once prevents config from overriding later). 3. Pool: oversized buffers (>1MB) are simply not returned to pool, letting GC reclaim them. No unnecessary allocation of replacement. --- algorithm/eas/easyrec_request.go | 6 ++---- configloader.go | 15 ++++++++++----- web/callback_controller_handler.go | 7 ++++--- 3 files changed, 16 insertions(+), 12 deletions(-) diff --git a/algorithm/eas/easyrec_request.go b/algorithm/eas/easyrec_request.go index af414d5e..48b5381b 100644 --- a/algorithm/eas/easyrec_request.go +++ b/algorithm/eas/easyrec_request.go @@ -51,10 +51,8 @@ func (r *EasyrecRequest) Invoke(requestData interface{}) (response interface{}, } body, err := r.EasClient.BytesPredict(data) - // Cap oversized buffers to avoid retaining large allocations in pool - if cap(buf.Bytes()) > 1<<20 { - marshalBufPool.Put(proto.NewBuffer(make([]byte, 0, 4096))) - } else { + // Only return normal-sized buffers to pool; let oversized ones be GC'd + if cap(buf.Bytes()) <= 1<<20 { marshalBufPool.Put(buf) } if err != nil { diff --git a/configloader.go b/configloader.go index 69f3db49..f4afc5be 100644 --- a/configloader.go +++ b/configloader.go @@ -179,12 +179,17 @@ func ListenConfig(configName string) { } // initCallbackHandler initializes the callback worker pool from config. -// Uses the first CallBackConfig found for pool settings (pool is global). +// Aggregates across all scenes: uses max WorkerPoolSize, OR of DropOnBackpressure. func initCallbackHandler(config *recconf.RecommendConfig) { + var poolSize int + var drop bool for _, cbConf := range config.CallBackConfs { - poolSize := cbConf.WorkerPoolSize - drop := cbConf.DropOnBackpressure - web.InitHandler(poolSize, 5000, drop) - return + if cbConf.WorkerPoolSize > poolSize { + poolSize = cbConf.WorkerPoolSize + } + if cbConf.DropOnBackpressure { + drop = true + } } + web.InitHandler(poolSize, 5000, drop) } diff --git a/web/callback_controller_handler.go b/web/callback_controller_handler.go index 04af4c2b..b5000a3c 100644 --- a/web/callback_controller_handler.go +++ b/web/callback_controller_handler.go @@ -11,7 +11,7 @@ import ( ) var callBackControllerHandler *CallBackControllerHandler -var once sync.Once +var initOnce sync.Once func init() { metrics.CallbackPendingFunc = CallbackPending @@ -65,14 +65,15 @@ func CallbackPending() int64 { // InitHandler initializes the callback handler with config values. // Must be called during service startup after config is loaded. func InitHandler(poolSize int, bufferSize int, dropOnBackpressure bool) { - once.Do(func() { + initOnce.Do(func() { callBackControllerHandler = NewCallBackControllerHandler(poolSize, bufferSize, dropOnBackpressure) }) } func Send(controller *CallBackController) { if callBackControllerHandler == nil { - once.Do(func() { + // Fallback: config not yet loaded, use defaults + initOnce.Do(func() { callBackControllerHandler = NewCallBackControllerHandler(20, 5000, false) }) } From ec07a1ba4e179c83b13c202e1727a5e6dbd5ed91 Mon Sep 17 00:00:00 2001 From: sch Date: Mon, 1 Jun 2026 23:22:00 +0800 Subject: [PATCH 10/11] fix: address gemini-code-assist PR review comments 1. Replace limiter.Wait (blocking) with limiter.Allow() (non-blocking). Callback is best-effort; no reason to block shared worker pool. Exceeding rate limit simply skips the EAS call. 2. Remove data race: eliminate bare nil check on callBackControllerHandler outside sync.Once. Now initOnce.Do is always called in Send path. 3. Support config hot reload: getCallbackLimiter calls SetLimit/SetBurst on cached limiter when QPS config changes. --- service/call_back_service.go | 18 +++++++++--------- web/callback_controller_handler.go | 10 ++++------ 2 files changed, 13 insertions(+), 15 deletions(-) diff --git a/service/call_back_service.go b/service/call_back_service.go index d21d9209..5aba0f7b 100644 --- a/service/call_back_service.go +++ b/service/call_back_service.go @@ -1,7 +1,6 @@ package service import ( - gocontext "context" "encoding/json" "fmt" "math/rand" @@ -28,13 +27,17 @@ import ( var callbackLimiters sync.Map func getCallbackLimiter(scene string, qps float64) *rate.Limiter { - if v, ok := callbackLimiters.Load(scene); ok { - return v.(*rate.Limiter) - } burst := int(qps) if burst < 1 { burst = 1 } + if v, ok := callbackLimiters.Load(scene); ok { + limiter := v.(*rate.Limiter) + // Support hot reload: update limit if config changed + limiter.SetLimit(rate.Limit(qps)) + limiter.SetBurst(burst) + return limiter + } limiter := rate.NewLimiter(rate.Limit(qps), burst) actual, _ := callbackLimiters.LoadOrStore(scene, limiter) return actual.(*rate.Limiter) @@ -258,13 +261,10 @@ func (r *CallBackService) Rank(context *context.RecommendContext) { } if callBackConfig.RateLimitQPS > 0 { limiter := getCallbackLimiter(scene_name, callBackConfig.RateLimitQPS) - ctx, cancel := gocontext.WithTimeout(gocontext.Background(), 2*time.Second) - if err := limiter.Wait(ctx); err != nil { - cancel() - log.Warning(fmt.Sprintf("requestId=%s\tcallback rate limit timeout, skipping EAS call", context.RecommendId)) + if !limiter.Allow() { + log.Warning(fmt.Sprintf("requestId=%s\tcallback rate limited, skipping EAS call", context.RecommendId)) return } - cancel() } if callBackConfig.JitterMaxMs > 0 { time.Sleep(time.Duration(rand.Intn(callBackConfig.JitterMaxMs)) * time.Millisecond) diff --git a/web/callback_controller_handler.go b/web/callback_controller_handler.go index b5000a3c..9bd69138 100644 --- a/web/callback_controller_handler.go +++ b/web/callback_controller_handler.go @@ -71,12 +71,10 @@ func InitHandler(poolSize int, bufferSize int, dropOnBackpressure bool) { } func Send(controller *CallBackController) { - if callBackControllerHandler == nil { - // Fallback: config not yet loaded, use defaults - initOnce.Do(func() { - callBackControllerHandler = NewCallBackControllerHandler(20, 5000, false) - }) - } + // initOnce ensures exactly one initialization, whether from InitHandler or here + initOnce.Do(func() { + callBackControllerHandler = NewCallBackControllerHandler(20, 5000, false) + }) // Increment pending BEFORE enqueue to avoid transient negative values h := callBackControllerHandler From 0daefaf2e1a87d09d5da9e21a9f92f453b7ccfab Mon Sep 17 00:00:00 2001 From: sch Date: Tue, 2 Jun 2026 09:34:37 +0800 Subject: [PATCH 11/11] chore: go mod tidy after upstream merge --- go.mod | 2 ++ go.sum | 1 + web/callback_controller_handler.go | 4 ++-- 3 files changed, 5 insertions(+), 2 deletions(-) diff --git a/go.mod b/go.mod index e6f34c4d..7d39c03d 100644 --- a/go.mod +++ b/go.mod @@ -58,6 +58,8 @@ require ( require ( github.com/alibabacloud-go/alibabacloud-gateway-spi v0.0.5 // indirect github.com/alibabacloud-go/darabonba-openapi/v2 v2.1.13 // indirect + github.com/alibabacloud-go/endpoint-util v1.1.0 // indirect + github.com/alibabacloud-go/openapi-util v0.1.1 // indirect github.com/alibabacloud-go/paifeaturestore-20230621/v6 v6.0.5 // indirect github.com/alibabacloud-go/pairecservice-20221213/v3 v3.3.0 // indirect github.com/alibabacloud-go/pairecservice-20221213/v4 v4.0.0 // indirect diff --git a/go.sum b/go.sum index 244f3931..b3ae5687 100644 --- a/go.sum +++ b/go.sum @@ -70,6 +70,7 @@ github.com/alibabacloud-go/tea-utils v1.3.1/go.mod h1:EI/o33aBfj3hETm4RLiAxF/ThQ github.com/alibabacloud-go/tea-utils v1.4.5 h1:h0/6Xd2f3bPE4XHTvkpjwxowIwRCJAJOqY6Eq8f3zfA= github.com/alibabacloud-go/tea-utils v1.4.5/go.mod h1:KNcT0oXlZZxOXINnZBs6YvgOd5aYp9U67G+E3R8fcQw= github.com/alibabacloud-go/tea-utils/v2 v2.0.5/go.mod h1:dL6vbUT35E4F4bFTHL845eUloqaerYBYPsdWR2/jhe4= +github.com/alibabacloud-go/tea-utils/v2 v2.0.6/go.mod h1:qxn986l+q33J5VkialKMqT/TTs3E+U9MJpd001iWQ9I= github.com/alibabacloud-go/tea-utils/v2 v2.0.7 h1:WDx5qW3Xa5ZgJ1c8NfqJkF6w+AU5wB8835UdhPr6Ax0= github.com/alibabacloud-go/tea-utils/v2 v2.0.7/go.mod h1:qxn986l+q33J5VkialKMqT/TTs3E+U9MJpd001iWQ9I= github.com/aliyun/alibaba-cloud-sdk-go v1.63.92 h1:qespx4b6EexlXkvQUow9x0v1GnWUJYGU5FWYw3a4Wlg= diff --git a/web/callback_controller_handler.go b/web/callback_controller_handler.go index 9bd69138..849b2ef7 100644 --- a/web/callback_controller_handler.go +++ b/web/callback_controller_handler.go @@ -129,8 +129,8 @@ func SendDirect(param *CallBackParam) { } if callBackControllerHandler == nil { - once.Do(func() { - callBackControllerHandler = NewCallBackControllerHandler(20, 5000) + initOnce.Do(func() { + callBackControllerHandler = NewCallBackControllerHandler(20, 5000, false) }) }