diff --git a/algorithm/eas/easyrec_request.go b/algorithm/eas/easyrec_request.go index 1fb8ba12..48b5381b 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,10 @@ func (r *EasyrecRequest) Invoke(requestData interface{}) (response interface{}, } body, err := r.EasClient.BytesPredict(data) + // 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 { return } diff --git a/configloader.go b/configloader.go index cb679345..f4afc5be 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,19 @@ func ListenConfig(configName string) { go loader.loopLoadConfig() } + +// initCallbackHandler initializes the callback worker pool from config. +// 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 { + if cbConf.WorkerPoolSize > poolSize { + poolSize = cbConf.WorkerPoolSize + } + if cbConf.DropOnBackpressure { + drop = true + } + } + web.InitHandler(poolSize, 5000, drop) +} diff --git a/go.mod b/go.mod index 5f14a54b..7d39c03d 100644 --- a/go.mod +++ b/go.mod @@ -52,11 +52,14 @@ 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 ( 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 91fb8025..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= @@ -769,6 +770,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..5aba0f7b 100644 --- a/service/call_back_service.go +++ b/service/call_back_service.go @@ -21,8 +21,28 @@ 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 { + 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) +} + type CallBackService struct { recallService *RecallService featureService *feature.FeatureService @@ -239,6 +259,16 @@ func (r *CallBackService) Rank(context *context.RecommendContext) { } } } + if callBackConfig.RateLimitQPS > 0 { + limiter := getCallbackLimiter(scene_name, callBackConfig.RateLimitQPS) + if !limiter.Allow() { + log.Warning(fmt.Sprintf("requestId=%s\tcallback rate limited, skipping EAS call", context.RecommendId)) + return + } + } + 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/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/service/rank/algo_data.go b/service/rank/algo_data.go index 031f10ef..14d45b74 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) } } @@ -274,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 { @@ -295,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 } @@ -310,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 { @@ -321,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 { @@ -335,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 } 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 } diff --git a/web/callback_controller_handler.go b/web/callback_controller_handler.go index 9cadd4be..849b2ef7 100644 --- a/web/callback_controller_handler.go +++ b/web/callback_controller_handler.go @@ -3,25 +3,39 @@ 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/service/metrics" "github.com/alibaba/pairec/v2/utils" ) var callBackControllerHandler *CallBackControllerHandler -var once sync.Once +var initOnce sync.Once + +func init() { + metrics.CallbackPendingFunc = CallbackPending +} type CallBackControllerHandler struct { - controllerCh chan *CallBackController - poolSize int + controllerCh chan *CallBackController + poolSize int + pending atomic.Int64 + dropOnBackpressure bool } -func NewCallBackControllerHandler() *CallBackControllerHandler { +func NewCallBackControllerHandler(poolSize int, bufferSize int, dropOnBackpressure bool) *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, + dropOnBackpressure: dropOnBackpressure, } - handler.start() return handler } @@ -31,19 +45,52 @@ func (h *CallBackControllerHandler) start() { go func() { for controller := range h.controllerCh { controller.doCallbackLog() + h.pending.Add(-1) } }() } } -func Send(controller *CallBackController) { +func (h *CallBackControllerHandler) Pending() int64 { + return h.pending.Load() +} + +func CallbackPending() int64 { if callBackControllerHandler == nil { - once.Do(func() { - callBackControllerHandler = NewCallBackControllerHandler() - }) + return 0 } + 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) { + initOnce.Do(func() { + callBackControllerHandler = NewCallBackControllerHandler(poolSize, bufferSize, dropOnBackpressure) + }) +} - callBackControllerHandler.controllerCh <- controller +func Send(controller *CallBackController) { + // 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 + 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 + } } // SendDirect bypasses the HTTP self-call path used by the external @@ -67,7 +114,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 +124,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() + initOnce.Do(func() { + callBackControllerHandler = NewCallBackControllerHandler(20, 5000, false) }) }