From 3fdd2e6f7470028bcdf448adb7cf491ded54de68 Mon Sep 17 00:00:00 2001 From: mohammadkhan Date: Mon, 13 Jul 2026 23:03:03 +0530 Subject: [PATCH 1/4] feat(concurrency-detector): Add hybrid mode to concurrency detector "hybrid" mode is a new concurrency detector mode that has been added which takes into account both concurrency based on token and request. Its saturation is computed as: `saturation = max(requestSaturation, tokenSaturation)` Using just request based concurrency is not enough as different workloads such as prefill heavy tends to consume more GPU compute, using just the tokenSaturation suffers from wrong estimation of output tokens which falls down in low ISL and high OSL workloads. Hybrid concurrency detector attempts to solve this problem by essentially capping whichever saturates first due to high ISL/OSL workload. Signed-off-by: mohammadkhan --- .../saturationdetector/concurrency/README.md | 2 +- .../saturationdetector/concurrency/config.go | 6 +- .../concurrency/detector.go | 73 +++++++++------- .../concurrency/detector_test.go | 87 +++++++++++++++++++ 4 files changed, 135 insertions(+), 33 deletions(-) diff --git a/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency/README.md b/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency/README.md index 8805b8b46a..51ddd30bea 100644 --- a/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency/README.md +++ b/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency/README.md @@ -34,7 +34,7 @@ The plugin internally tracks active concurrency by hooking into the request life The plugin accepts JSON parameters decoding to the following fields: -- `concurrencyMode` (`string`): Evaluation mode. Valid values are `"requests"` or `"tokens"`. (Default: `"requests"`) +- `concurrencyMode` (`string`): Evaluation mode. Valid values are `"requests"`, `"tokens"`, or `"hybrid"`. In `"hybrid"` mode both request and token accounting are evaluated and the more constraining dimension wins: pool saturation is the larger of the two ratios, and an endpoint is filtered out when either its request load or its token load reaches the limit. (Default: `"requests"`) - `maxConcurrency` (`int64`): Maximum requests in flight. Serves as the "ideal" request capacity for a single endpoint. Must be > 0. (Default: `100`) - `maxTokenConcurrency` (`int64`): Maximum tokens in flight. The "tokens" mode equivalent of `maxConcurrency`. Must be > 0. (Default: `1000000`) - `headroom` (`float64`): Allowed burst capacity above the ideal threshold, expressed as a fraction (e.g., `0.2` for 20%). Must be >= 0.0. (Default: `0.0`) diff --git a/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency/config.go b/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency/config.go index d5add71953..ee32283dbc 100644 --- a/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency/config.go +++ b/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency/config.go @@ -62,6 +62,8 @@ type apiConfig struct { // Valid values are: // - "requests": use discrete request counts for capacity accounting. // - "tokens": use estimated token counts for capacity accounting. + // - "hybrid": evaluate both request and token accounting and report whichever + // reaches saturation first (the more constraining of the two). // // Defaults to "requests" if unset. ConcurrencyMode *concurrencyMode `json:"concurrencyMode,omitempty"` @@ -85,6 +87,8 @@ const ( modeRequests concurrencyMode = "requests" // modeTokens uses token count for concurrency detection. modeTokens concurrencyMode = "tokens" + // modeHybrid uses both request and token counts, reporting whichever saturates first. + modeHybrid concurrencyMode = "hybrid" ) const ( @@ -164,7 +168,7 @@ func validateConfig(cfg *apiConfig) error { if cfg.ConcurrencyMode != nil { switch *cfg.ConcurrencyMode { - case modeRequests, modeTokens: + case modeRequests, modeTokens, modeHybrid: // Valid default: errs = append(errs, fmt.Errorf("unsupported concurrencyMode: %q", *cfg.ConcurrencyMode)) diff --git a/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency/detector.go b/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency/detector.go index 5dce0e6484..beeeeac388 100644 --- a/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency/detector.go +++ b/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency/detector.go @@ -121,49 +121,58 @@ func (d *detector) getLoad(m datalayer.AttributeMap) *attrconcurrency.InFlightLo // Saturation calculates the saturation level of the pool. // -// It returns an aggregate saturation signal where: +// It returns an aggregate saturation signal where each dimension is evaluated as: // -// Saturation = Total Inflight Requests / Total MaxConcurrency Capacity. +// Saturation = Total Inflight / Total Capacity. +// +// In "hybrid" mode both the request and token dimensions are computed and the larger +// (more saturated) of the two is returned, so the pool reports saturated as soon as +// either dimension is exhausted. func (d *detector) Saturation(_ context.Context, endpoints []datalayer.Endpoint) float64 { if len(endpoints) == 0 { return 1.0 } - var totalInflight, totalCapacity int64 + var reqInflight, reqCapacity, tokInflight, tokCapacity int64 for _, e := range endpoints { if e == nil { continue } - if d.config.mode == modeTokens { - totalCapacity += d.config.maxTokenConcurrency - } else { - totalCapacity += d.config.maxConcurrency - } + reqCapacity += d.config.maxConcurrency + tokCapacity += d.config.maxTokenConcurrency if e.GetMetadata() == nil { continue } load := d.getLoad(e.GetAttributes()) + reqInflight += load.Requests + tokInflight += load.Tokens + } - if d.config.mode == modeTokens { - totalInflight += load.Tokens - } else { - totalInflight += load.Requests - } + switch d.config.mode { + case modeTokens: + return ratio(tokInflight, tokCapacity) + case modeHybrid: + return max(ratio(reqInflight, reqCapacity), ratio(tokInflight, tokCapacity)) + default: + return ratio(reqInflight, reqCapacity) } +} - if totalCapacity == 0 { +// ratio computes inflight/capacity, failing closed (1.0) when capacity is zero. +func ratio(inflight, capacity int64) float64 { + if capacity == 0 { return 1.0 } - - return float64(totalInflight) / float64(totalCapacity) + return float64(inflight) / float64(capacity) } // Filter blocks traffic to specific endpoints that are physically saturated or exceeding their safety limits. // -// It applies a relaxed limit (MaxConcurrency * (1 + Headroom)) to allow for scheduling flexibility and burst tolerance. +// It applies a relaxed limit (Capacity * (1 + Headroom)) to allow for scheduling flexibility and burst tolerance. +// In "hybrid" mode an endpoint is dropped when either its request load or its token load reaches the limit. func (d *detector) Filter( _ context.Context, _ *fwksched.InferenceRequest, @@ -172,12 +181,8 @@ func (d *detector) Filter( // Pre-allocate assuming most endpoints will pass the filter to minimize allocations. filtered := make([]fwksched.Endpoint, 0, len(endpoints)) - var limit int64 - if d.config.mode == modeTokens { - limit = int64(float64(d.config.maxTokenConcurrency) * (1.0 + d.config.headroom)) - } else { - limit = int64(float64(d.config.maxConcurrency) * (1.0 + d.config.headroom)) - } + reqLimit := int64(float64(d.config.maxConcurrency) * (1.0 + d.config.headroom)) + tokLimit := int64(float64(d.config.maxTokenConcurrency) * (1.0 + d.config.headroom)) for _, e := range endpoints { if e == nil { @@ -185,15 +190,21 @@ func (d *detector) Filter( } load := d.getLoad(e) - if d.config.mode == modeTokens { - if load.Tokens < limit { - filtered = append(filtered, e) - } - } else { - if load.Requests < limit { - filtered = append(filtered, e) - } + if d.admits(load, reqLimit, tokLimit) { + filtered = append(filtered, e) } } return filtered } + +// admits reports whether an endpoint is below its safety limit for the active mode. +func (d *detector) admits(load *attrconcurrency.InFlightLoad, reqLimit, tokLimit int64) bool { + switch d.config.mode { + case modeTokens: + return load.Tokens < tokLimit + case modeHybrid: + return load.Requests < reqLimit && load.Tokens < tokLimit + default: + return load.Requests < reqLimit + } +} diff --git a/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency/detector_test.go b/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency/detector_test.go index 0126420a9c..76d78956ef 100644 --- a/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency/detector_test.go +++ b/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency/detector_test.go @@ -110,6 +110,11 @@ func TestConcurrencyDetectorFactory(t *testing.T) { configJSON: []byte(`{"headroom": -0.5}`), wantError: true, }, + { + name: "valid hybrid mode", + configJSON: []byte(`{"concurrencyMode": "hybrid", "maxConcurrency": 10, "maxTokenConcurrency": 100}`), + wantError: false, + }, { name: "invalid mode", configJSON: []byte(`{"concurrencyMode": "magic"}`), @@ -490,6 +495,88 @@ func TestDetector_TokenDeleteEndpoint(t *testing.T) { require.InDelta(t, 0.0, detector.Saturation(ctx, candidates), 1e-6, "expected clean state after DeleteEndpoint") } +// TestDetector_HybridSaturation verifies hybrid mode reports the more saturated dimension. +func TestDetector_HybridSaturation(t *testing.T) { + t.Parallel() + + config := config{ + mode: modeHybrid, + maxConcurrency: 10, + maxTokenConcurrency: 100, + } + + tests := []struct { + name string + requests int64 + tokens int64 + wantSaturation float64 + }{ + {name: "empty", requests: 0, tokens: 0, wantSaturation: 0.0}, + {name: "tokens_dominate", requests: 2, tokens: 80, wantSaturation: 0.8}, // max(0.2, 0.8) + {name: "requests_dominate", requests: 9, tokens: 10, wantSaturation: 0.9}, // max(0.9, 0.1) + {name: "request_dimension_saturates_first", requests: 10, tokens: 5, wantSaturation: 1.0}, + {name: "token_dimension_saturates_first", requests: 1, tokens: 100, wantSaturation: 1.0}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + reg := newLocalRegistry() + ctx := context.Background() + detector := newDetector("test-detector", config, logr.Discard()) + endpointName := "endpoint-a" + reg.update(fullEndpointName(endpointName), func(load *attrconcurrency.InFlightLoad) { + load.Requests = tc.requests + load.Tokens = tc.tokens + }) + + got := detector.Saturation(ctx, []datalayer.Endpoint{newFakeEndpoint(reg, endpointName)}) + require.InDelta(t, tc.wantSaturation, got, 1e-6, "hybrid saturation mismatch") + }) + } +} + +// TestDetector_HybridFilter verifies hybrid mode drops an endpoint when either dimension hits its limit. +func TestDetector_HybridFilter(t *testing.T) { + t.Parallel() + + // headroom 0.0 -> limits are exactly maxConcurrency (10) and maxTokenConcurrency (100). + config := config{ + mode: modeHybrid, + maxConcurrency: 10, + maxTokenConcurrency: 100, + } + ctx := context.Background() + + tests := []struct { + name string + requests int64 + tokens int64 + wantKept int + }{ + {name: "below_both_limits", requests: 5, tokens: 50, wantKept: 1}, + {name: "request_limit_reached", requests: 10, tokens: 50, wantKept: 0}, + {name: "token_limit_reached", requests: 5, tokens: 100, wantKept: 0}, + {name: "both_over", requests: 20, tokens: 200, wantKept: 0}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + reg := newLocalRegistry() + detector := newDetector("test-detector", config, logr.Discard()) + endpointName := "endpoint-a" + reg.update(fullEndpointName(endpointName), func(load *attrconcurrency.InFlightLoad) { + load.Requests = tc.requests + load.Tokens = tc.tokens + }) + + kept := detector.Filter(ctx, nil, []fwksched.Endpoint{newStubSchedulingEndpoint(reg, endpointName)}) + require.Len(t, kept, tc.wantKept, "hybrid filter mismatch") + }) + } +} + // TestDetector_ConcurrencyStress performs race condition check. func TestDetector_ConcurrencyStress(t *testing.T) { t.Parallel() From 5f9653485397375c6b2bcc9ba22d2976ed67a1ef Mon Sep 17 00:00:00 2001 From: mohammadkhan Date: Fri, 17 Jul 2026 14:28:03 +0530 Subject: [PATCH 2/4] ci: run formatting Signed-off-by: mohammadkhan --- .../flowcontrol/saturationdetector/concurrency/detector_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency/detector_test.go b/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency/detector_test.go index 76d78956ef..de3d344f3f 100644 --- a/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency/detector_test.go +++ b/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency/detector_test.go @@ -512,7 +512,7 @@ func TestDetector_HybridSaturation(t *testing.T) { wantSaturation float64 }{ {name: "empty", requests: 0, tokens: 0, wantSaturation: 0.0}, - {name: "tokens_dominate", requests: 2, tokens: 80, wantSaturation: 0.8}, // max(0.2, 0.8) + {name: "tokens_dominate", requests: 2, tokens: 80, wantSaturation: 0.8}, // max(0.2, 0.8) {name: "requests_dominate", requests: 9, tokens: 10, wantSaturation: 0.9}, // max(0.9, 0.1) {name: "request_dimension_saturates_first", requests: 10, tokens: 5, wantSaturation: 1.0}, {name: "token_dimension_saturates_first", requests: 1, tokens: 100, wantSaturation: 1.0}, From c655ab0d0b4fc6a604ddd2bfe3391d8d65224321 Mon Sep 17 00:00:00 2001 From: mohammadkhan Date: Mon, 20 Jul 2026 22:48:24 +0530 Subject: [PATCH 3/4] fix: calculate pool saturation as endpoint average Pool saturation is defined as average of saturation across each endpoint, this is exactly how it is computed but in multi-dimensional saturation cases such as hybrid, this wasn't true. This fixes it. Signed-off-by: mohammadkhan --- .../saturationdetector/concurrency/README.md | 6 ++- .../concurrency/detector.go | 21 ++++++-- .../concurrency/detector_test.go | 51 ++++++++++++++----- 3 files changed, 57 insertions(+), 21 deletions(-) diff --git a/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency/README.md b/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency/README.md index 51ddd30bea..5513590831 100644 --- a/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency/README.md +++ b/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency/README.md @@ -15,7 +15,9 @@ The detector implements the `SaturationDetector` interface to provide a utilizat In token mode, both numerator and denominator are evaluated in tokens: the aggregate inflight token count divided by the sum of all endpoints' MaxTokenConcurrency. -**Heterogeneous Deployments:** Because this detector calculates saturation globally as a single aggregate fraction, it utilizes an aggregate queueing model. In deployments with heterogeneous compute (e.g., mixing H100 and L4 nodes), this heavily biases the pool saturation metric toward the state of the larger nodes. Contrast this with the Utilization Detector, which evaluates saturation as an unweighted average of individual endpoint scores. +Hybrid mode is the exception: rather than one aggregate fraction, it evaluates each endpoint's saturation as the larger of its request and token ratios and reports the unweighted average across endpoints. This prevents distinct endpoints saturating on different dimensions from being masked by aggregate ratios that each remain low. + +**Heterogeneous Deployments:** Because this detector calculates saturation globally as a single aggregate fraction (in requests and tokens mode), it utilizes an aggregate queueing model. In deployments with heterogeneous compute (e.g., mixing H100 and L4 nodes), this heavily biases the pool saturation metric toward the state of the larger nodes. Contrast this with the Utilization Detector, which evaluates saturation as an unweighted average of individual endpoint scores. ### Role in Scheduling (The Traffic Shaper) The detector implements the `Filter` interface to protect individual endpoints. It removes endpoints from candidate lists if their local inflight count exceeds the safety limit: @@ -34,7 +36,7 @@ The plugin internally tracks active concurrency by hooking into the request life The plugin accepts JSON parameters decoding to the following fields: -- `concurrencyMode` (`string`): Evaluation mode. Valid values are `"requests"`, `"tokens"`, or `"hybrid"`. In `"hybrid"` mode both request and token accounting are evaluated and the more constraining dimension wins: pool saturation is the larger of the two ratios, and an endpoint is filtered out when either its request load or its token load reaches the limit. (Default: `"requests"`) +- `concurrencyMode` (`string`): Evaluation mode. Valid values are `"requests"`, `"tokens"`, or `"hybrid"`. In `"hybrid"` mode both request and token accounting are evaluated. Pool saturation is computed per endpoint as the larger of that endpoint's request and token ratios, then averaged across endpoints, so an endpoint saturated on either dimension is reflected even when distinct endpoints saturate on different dimensions. An endpoint is filtered out when either its request load or its token load reaches the limit. (Default: `"requests"`) - `maxConcurrency` (`int64`): Maximum requests in flight. Serves as the "ideal" request capacity for a single endpoint. Must be > 0. (Default: `100`) - `maxTokenConcurrency` (`int64`): Maximum tokens in flight. The "tokens" mode equivalent of `maxConcurrency`. Must be > 0. (Default: `1000000`) - `headroom` (`float64`): Allowed burst capacity above the ideal threshold, expressed as a fraction (e.g., `0.2` for 20%). Must be >= 0.0. (Default: `0.0`) diff --git a/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency/detector.go b/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency/detector.go index beeeeac388..62652eb9ca 100644 --- a/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency/detector.go +++ b/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency/detector.go @@ -121,24 +121,28 @@ func (d *detector) getLoad(m datalayer.AttributeMap) *attrconcurrency.InFlightLo // Saturation calculates the saturation level of the pool. // -// It returns an aggregate saturation signal where each dimension is evaluated as: +// In "requests" and "tokens" mode it returns an aggregate signal, evaluated as: // // Saturation = Total Inflight / Total Capacity. // -// In "hybrid" mode both the request and token dimensions are computed and the larger -// (more saturated) of the two is returned, so the pool reports saturated as soon as -// either dimension is exhausted. +// In "hybrid" mode saturation is instead evaluated per endpoint as +// max(requestRatio, tokenRatio) and averaged across endpoints. Evaluating each +// endpoint independently ensures an endpoint saturated on either dimension is +// reflected in the pool signal. func (d *detector) Saturation(_ context.Context, endpoints []datalayer.Endpoint) float64 { if len(endpoints) == 0 { return 1.0 } var reqInflight, reqCapacity, tokInflight, tokCapacity int64 + var endpointCount int + var hybridSatSum float64 for _, e := range endpoints { if e == nil { continue } + endpointCount++ reqCapacity += d.config.maxConcurrency tokCapacity += d.config.maxTokenConcurrency @@ -149,13 +153,20 @@ func (d *detector) Saturation(_ context.Context, endpoints []datalayer.Endpoint) load := d.getLoad(e.GetAttributes()) reqInflight += load.Requests tokInflight += load.Tokens + hybridSatSum += max( + ratio(load.Requests, d.config.maxConcurrency), + ratio(load.Tokens, d.config.maxTokenConcurrency), + ) } switch d.config.mode { case modeTokens: return ratio(tokInflight, tokCapacity) case modeHybrid: - return max(ratio(reqInflight, reqCapacity), ratio(tokInflight, tokCapacity)) + if endpointCount == 0 { + return 1.0 + } + return hybridSatSum / float64(endpointCount) default: return ratio(reqInflight, reqCapacity) } diff --git a/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency/detector_test.go b/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency/detector_test.go index de3d344f3f..4264d85443 100644 --- a/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency/detector_test.go +++ b/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency/detector_test.go @@ -495,7 +495,8 @@ func TestDetector_TokenDeleteEndpoint(t *testing.T) { require.InDelta(t, 0.0, detector.Saturation(ctx, candidates), 1e-6, "expected clean state after DeleteEndpoint") } -// TestDetector_HybridSaturation verifies hybrid mode reports the more saturated dimension. +// TestDetector_HybridSaturation verifies hybrid mode evaluates each endpoint as the more +// saturated of its request and token dimensions and averages the result across endpoints. func TestDetector_HybridSaturation(t *testing.T) { t.Parallel() @@ -505,17 +506,35 @@ func TestDetector_HybridSaturation(t *testing.T) { maxTokenConcurrency: 100, } + type endpointLoad struct { + requests int64 + tokens int64 + } + tests := []struct { name string - requests int64 - tokens int64 + endpoints map[string]endpointLoad wantSaturation float64 }{ - {name: "empty", requests: 0, tokens: 0, wantSaturation: 0.0}, - {name: "tokens_dominate", requests: 2, tokens: 80, wantSaturation: 0.8}, // max(0.2, 0.8) - {name: "requests_dominate", requests: 9, tokens: 10, wantSaturation: 0.9}, // max(0.9, 0.1) - {name: "request_dimension_saturates_first", requests: 10, tokens: 5, wantSaturation: 1.0}, - {name: "token_dimension_saturates_first", requests: 1, tokens: 100, wantSaturation: 1.0}, + {name: "empty", endpoints: map[string]endpointLoad{"endpoint-a": {0, 0}}, wantSaturation: 0.0}, + // Single endpoint: saturation is the more saturated of its two dimensions. + {name: "tokens_dominate", endpoints: map[string]endpointLoad{"endpoint-a": {2, 80}}, wantSaturation: 0.8}, // max(0.2, 0.8) + {name: "requests_dominate", endpoints: map[string]endpointLoad{"endpoint-a": {9, 10}}, wantSaturation: 0.9}, // max(0.9, 0.1) + {name: "request_dimension_saturates_first", endpoints: map[string]endpointLoad{"endpoint-a": {10, 5}}, wantSaturation: 1.0}, + {name: "token_dimension_saturates_first", endpoints: map[string]endpointLoad{"endpoint-a": {1, 100}}, wantSaturation: 1.0}, + // Distinct endpoints saturated on different dimensions each report 1.0, so the pool + // averages to 1.0. + { + name: "endpoints_saturated_on_different_dimensions", + endpoints: map[string]endpointLoad{"endpoint-a": {10, 0}, "endpoint-b": {0, 100}}, + wantSaturation: 1.0, + }, + // One saturated endpoint and one idle endpoint average to the midpoint. + { + name: "per_endpoint_average", + endpoints: map[string]endpointLoad{"endpoint-a": {10, 0}, "endpoint-b": {0, 0}}, + wantSaturation: 0.5, // mean(1.0, 0.0) + }, } for _, tc := range tests { @@ -524,13 +543,17 @@ func TestDetector_HybridSaturation(t *testing.T) { reg := newLocalRegistry() ctx := context.Background() detector := newDetector("test-detector", config, logr.Discard()) - endpointName := "endpoint-a" - reg.update(fullEndpointName(endpointName), func(load *attrconcurrency.InFlightLoad) { - load.Requests = tc.requests - load.Tokens = tc.tokens - }) - got := detector.Saturation(ctx, []datalayer.Endpoint{newFakeEndpoint(reg, endpointName)}) + candidates := make([]datalayer.Endpoint, 0, len(tc.endpoints)) + for name, l := range tc.endpoints { + reg.update(fullEndpointName(name), func(load *attrconcurrency.InFlightLoad) { + load.Requests = l.requests + load.Tokens = l.tokens + }) + candidates = append(candidates, newFakeEndpoint(reg, name)) + } + + got := detector.Saturation(ctx, candidates) require.InDelta(t, tc.wantSaturation, got, 1e-6, "hybrid saturation mismatch") }) } From 5a4c8c1848ed9b3281401059d10622d85906c728 Mon Sep 17 00:00:00 2001 From: mohammadkhan Date: Thu, 23 Jul 2026 18:29:01 +0530 Subject: [PATCH 4/4] docs: review and fix the code comments Signed-off-by: mohammadkhan --- .../flowcontrol/saturationdetector/concurrency/config.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency/config.go b/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency/config.go index ee32283dbc..26a19943f4 100644 --- a/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency/config.go +++ b/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency/config.go @@ -62,8 +62,8 @@ type apiConfig struct { // Valid values are: // - "requests": use discrete request counts for capacity accounting. // - "tokens": use estimated token counts for capacity accounting. - // - "hybrid": evaluate both request and token accounting and report whichever - // reaches saturation first (the more constraining of the two). + // - "hybrid": evaluate each endpoint as the more constraining of its request and + // token ratios; pool saturation is the average of these across endpoints. // // Defaults to "requests" if unset. ConcurrencyMode *concurrencyMode `json:"concurrencyMode,omitempty"` @@ -87,7 +87,7 @@ const ( modeRequests concurrencyMode = "requests" // modeTokens uses token count for concurrency detection. modeTokens concurrencyMode = "tokens" - // modeHybrid uses both request and token counts, reporting whichever saturates first. + // modeHybrid uses endpoint average saturation across both request and token counts. modeHybrid concurrencyMode = "hybrid" )