From f72c2c2c67c6a4e895b3e1faa1d65d03e97d3233 Mon Sep 17 00:00:00 2001 From: Luke Van Drie Date: Mon, 20 Jul 2026 16:04:38 -0700 Subject: [PATCH 1/6] feat(flowcontrol): eviction-terminated listener and victim bound on the tracker RequestEvictor gains the confirmation signal for reclamation pacing (an at-most-once listener fired when an evicted request's stream terminates), a victim-priority peek, and a strict per-decision victim priority bound on EvictN so multi-revocation decisions cannot churn the demand band itself. Signed-off-by: Luke Van Drie --- .../flowcontrol/eviction/request_evictor.go | 90 ++++++++++++++- .../eviction/request_evictor_test.go | 109 +++++++++++++++++- pkg/epp/flowcontrol/integration_test.go | 2 +- 3 files changed, 191 insertions(+), 10 deletions(-) diff --git a/pkg/epp/flowcontrol/eviction/request_evictor.go b/pkg/epp/flowcontrol/eviction/request_evictor.go index 6cd1e39e5a..7b85f6e4ea 100644 --- a/pkg/epp/flowcontrol/eviction/request_evictor.go +++ b/pkg/epp/flowcontrol/eviction/request_evictor.go @@ -19,6 +19,7 @@ package eviction import ( "context" "net" + "sync" "time" "sigs.k8s.io/controller-runtime/pkg/log" @@ -37,6 +38,17 @@ type RequestEvictor struct { queue *EvictionQueue evictor Evictor evictionRegistry *EvictionRegistry + + // mu guards the fields below. + mu sync.Mutex + // pendingEvictions holds the IDs of requests whose eviction has been issued but whose stream has + // not yet terminated. Populated by EvictN before the eviction signal is sent; consumed exactly + // once by cleanupRequest when the stream terminates. + pendingEvictions map[string]struct{} + // evictionTerminatedListener, if set, is invoked with the request ID each time an evicted + // request's stream terminates. It may be called from ext_proc handler goroutines and must be + // safe for concurrent use. + evictionTerminatedListener func(requestID string) } // NewRequestEvictor creates a RequestEvictor with the given policies and evictor. @@ -53,9 +65,30 @@ func NewRequestEvictor( queue: NewEvictionQueue(ordering, filter), evictor: evictor, evictionRegistry: registry, + pendingEvictions: make(map[string]struct{}), } } +// SetEvictionTerminatedListener registers a callback invoked once per evicted request when its +// stream terminates. This is the confirmation signal for eviction pacing: the request is dead from +// the EPP's perspective, even though the engine may not have freed its resources yet. +// Must be called before the RequestEvictor is in use. +func (p *RequestEvictor) SetEvictionTerminatedListener(listener func(requestID string)) { + p.mu.Lock() + defer p.mu.Unlock() + p.evictionTerminatedListener = listener +} + +// PeekVictimPriority returns the priority of the next request that would be evicted, or false if +// no evictable requests exist. It does not modify the queue. +func (p *RequestEvictor) PeekVictimPriority() (priority int, ok bool) { + item := p.queue.Peek() + if item == nil { + return 0, false + } + return item.Priority, true +} + // EvictionRegistry returns the shared eviction registry. // The ext_proc Process() goroutine uses this to look up eviction channels for dispatched requests. func (p *RequestEvictor) EvictionRegistry() *EvictionRegistry { @@ -142,11 +175,14 @@ func (p *RequestEvictor) ResponseBody( "inFlight", p.queue.InFlightLen()) } -// EvictN attempts to evict up to n requests from the eviction queue. -// Each request is only removed from tracking after a successful eviction. If the eviction fails, -// the request remains in the queue for a future eviction attempt. +// EvictN attempts to evict up to n requests from the eviction queue, in victim-policy order, +// stopping at the first victim whose priority is not strictly below priorityBound. The bound +// enforces strict priority dominance across a whole multi-revocation decision: the victim head +// check alone does not cover later victims, which come off the heap at priorities at or above the +// head's. Each request is only removed from tracking after a successful eviction. If the eviction +// fails, the request remains in the queue for a future eviction attempt. // Returns the request IDs that were successfully evicted. -func (p *RequestEvictor) EvictN(ctx context.Context, n int) ([]string, error) { +func (p *RequestEvictor) EvictN(ctx context.Context, n int, priorityBound int) ([]string, error) { logger := log.FromContext(ctx) evicted := make([]string, 0, n) @@ -157,8 +193,23 @@ func (p *RequestEvictor) EvictN(ctx context.Context, n int) ([]string, error) { } item := items[0] + if item.Priority >= priorityBound { + // The heap orders victims lowest-priority first, so no remaining victim can be below the + // bound either. Re-tracking races a concurrent completion: if cleanupRequest ran between + // the pop and this Track, the entry is re-inserted dead and survives until a future + // decision evicts it, which then stalls the pacing gate for one ConfirmationTimeout. The + // window is sub-microsecond and the cost is bounded, so it is tolerated. + p.queue.Track(item) + break + } + + // Mark before the eviction signal is sent so that cleanupRequest observes the marker no matter + // how quickly the stream terminates. + p.markPendingEviction(item.RequestID) if err := p.evictor.Evict(ctx, item); err != nil { logger.Error(err, "Failed to evict request, re-tracking", "requestID", item.RequestID, "targetURL", item.TargetURL) + p.unmarkPendingEviction(item.RequestID) + // Same tolerated re-track-vs-completion window as the bound branch above. p.queue.Track(item) continue } @@ -179,12 +230,43 @@ func (p *RequestEvictor) Stats() (inFlight int, evictable int) { // cleanupRequest removes a request from all tracking structures. // If the evictor supports cleanup (e.g., ImmediateResponseEvictor), it also // cleans up evictor-internal state to prevent unbounded map growth. +// If the request had been evicted, the eviction-terminated listener is notified exactly once. func (p *RequestEvictor) cleanupRequest(requestID string) { p.queue.Untrack(requestID) p.evictionRegistry.Deregister(requestID) if c, ok := p.evictor.(EvictorWithCleanup); ok { c.Cleanup(requestID) } + if listener := p.consumePendingEviction(requestID); listener != nil { + listener(requestID) + } +} + +// markPendingEviction records that an eviction has been issued for the request. +func (p *RequestEvictor) markPendingEviction(requestID string) { + p.mu.Lock() + defer p.mu.Unlock() + p.pendingEvictions[requestID] = struct{}{} +} + +// unmarkPendingEviction removes the pending-eviction marker after a failed eviction attempt. +func (p *RequestEvictor) unmarkPendingEviction(requestID string) { + p.mu.Lock() + defer p.mu.Unlock() + delete(p.pendingEvictions, requestID) +} + +// consumePendingEviction removes the pending-eviction marker if present and returns the listener +// to notify, or nil if the request was not evicted or was already consumed. Consuming under the +// same lock as marking guarantees at-most-once notification across the idempotent cleanup paths. +func (p *RequestEvictor) consumePendingEviction(requestID string) func(requestID string) { + p.mu.Lock() + defer p.mu.Unlock() + if _, ok := p.pendingEvictions[requestID]; !ok { + return nil + } + delete(p.pendingEvictions, requestID) + return p.evictionTerminatedListener } // EvictorWithCleanup is an optional interface for evictors that maintain per-request state diff --git a/pkg/epp/flowcontrol/eviction/request_evictor_test.go b/pkg/epp/flowcontrol/eviction/request_evictor_test.go index 18d5b2687a..e9cc409ab1 100644 --- a/pkg/epp/flowcontrol/eviction/request_evictor_test.go +++ b/pkg/epp/flowcontrol/eviction/request_evictor_test.go @@ -102,7 +102,7 @@ func TestRequestEvictor_EvictN_ClosesEvictChannel(t *testing.T) { evictCh := re.EvictionRegistry().Get("req-1") require.NotNil(t, evictCh) - evicted, err := re.EvictN(ctx, 1) + evicted, err := re.EvictN(ctx, 1, 0) require.NoError(t, err) require.Equal(t, []string{"req-1"}, evicted) @@ -123,7 +123,7 @@ func TestRequestEvictor_EvictN_ReTracksOnFailure(t *testing.T) { ctx := context.Background() re.PreRequest(ctx, makeInferenceRequest("req-1", -1), makeSchedulingResult()) - evicted, err := re.EvictN(ctx, 1) + evicted, err := re.EvictN(ctx, 1, 0) require.NoError(t, err) assert.Empty(t, evicted) @@ -149,7 +149,7 @@ func TestRequestEvictor_RaceBetweenEvictAndCompletion(t *testing.T) { go func() { defer wg.Done() for range 5 { - _, _ = re.EvictN(ctx, 1) + _, _ = re.EvictN(ctx, 1, 0) time.Sleep(time.Millisecond) } }() @@ -204,7 +204,7 @@ func TestRequestEvictor_CleanupCallsEvictorCleanup(t *testing.T) { re.PreRequest(ctx, makeInferenceRequest("req-1", -1), makeSchedulingResult()) // Evict to create a sync.Once entry in the evictor. - _, _ = re.EvictN(ctx, 1) + _, _ = re.EvictN(ctx, 1, 0) // Complete the request — this should call cleanupRequest which calls evictor.Cleanup. // After cleanup, the sync.Once entry for "req-1" should be removed. @@ -222,7 +222,7 @@ func TestRequestEvictor_CleanupCallsEvictorCleanup(t *testing.T) { evictCh := re.EvictionRegistry().Get("req-1") require.NotNil(t, evictCh) - evicted, err := re.EvictN(ctx, 1) + evicted, err := re.EvictN(ctx, 1, 0) require.NoError(t, err) require.Len(t, evicted, 1) @@ -252,3 +252,102 @@ type failingEvictor struct{} func (e *failingEvictor) Evict(_ context.Context, _ *flowcontrol.EvictionItem) error { return assert.AnError } + +func TestRequestEvictor_EvictionTerminatedListener_NotifiedOncePerEvictedRequest(t *testing.T) { + t.Parallel() + re := NewRequestEvictor(&testOrdering{}, &acceptAllFilter{}, NewImmediateResponseEvictor()) + + var mu sync.Mutex + notified := make(map[string]int) + re.SetEvictionTerminatedListener(func(requestID string) { + mu.Lock() + defer mu.Unlock() + notified[requestID]++ + }) + + ctx := context.Background() + re.PreRequest(ctx, makeInferenceRequest("req-evicted", -1), makeSchedulingResult()) + re.PreRequest(ctx, makeInferenceRequest("req-natural", -1), makeSchedulingResult()) + + evicted, err := re.EvictN(ctx, 1, 0) + require.NoError(t, err) + require.Len(t, evicted, 1) + evictedID := evicted[0] + + // Stream termination for the evicted request notifies; the idempotent second cleanup does not. + re.ResponseBody(ctx, makeInferenceRequest(evictedID, -1), &requestcontrol.Response{EndOfStream: true}, nil) + re.ResponseBody(ctx, makeInferenceRequest(evictedID, -1), &requestcontrol.Response{EndOfStream: true}, nil) + + // A naturally completing request must not notify. + naturalID := "req-natural" + if evictedID == naturalID { + naturalID = "req-evicted" + } + re.ResponseBody(ctx, makeInferenceRequest(naturalID, -1), &requestcontrol.Response{EndOfStream: true}, nil) + + mu.Lock() + defer mu.Unlock() + assert.Equal(t, map[string]int{evictedID: 1}, notified, + "listener should fire exactly once, only for the evicted request") +} + +func TestRequestEvictor_FailedEviction_DoesNotNotifyListener(t *testing.T) { + t.Parallel() + re := NewRequestEvictor(&testOrdering{}, &acceptAllFilter{}, &failingEvictor{}) + + var mu sync.Mutex + var notifications []string + re.SetEvictionTerminatedListener(func(requestID string) { + mu.Lock() + defer mu.Unlock() + notifications = append(notifications, requestID) + }) + + ctx := context.Background() + re.PreRequest(ctx, makeInferenceRequest("req-1", -1), makeSchedulingResult()) + + evicted, err := re.EvictN(ctx, 1, 0) + require.NoError(t, err) + assert.Empty(t, evicted, "failed eviction should evict nothing") + + // The request later completes naturally: still no notification. + re.ResponseBody(ctx, makeInferenceRequest("req-1", -1), &requestcontrol.Response{EndOfStream: true}, nil) + + mu.Lock() + defer mu.Unlock() + assert.Empty(t, notifications, "a failed eviction must not mark the request as evicted") +} + +func TestRequestEvictor_PeekVictimPriority(t *testing.T) { + t.Parallel() + re := NewRequestEvictor(&testOrdering{}, &acceptAllFilter{}, &NoOpEvictor{}) + + _, ok := re.PeekVictimPriority() + assert.False(t, ok, "no victims when nothing is tracked") + + ctx := context.Background() + re.PreRequest(ctx, makeInferenceRequest("req-1", -2), makeSchedulingResult()) + + priority, ok := re.PeekVictimPriority() + require.True(t, ok) + assert.Equal(t, -2, priority) +} + +func TestRequestEvictor_EvictN_PriorityBound(t *testing.T) { + t.Parallel() + re := NewRequestEvictor(&testOrdering{}, &acceptAllFilter{}, NewImmediateResponseEvictor()) + + ctx := context.Background() + re.PreRequest(ctx, makeInferenceRequest("req-low", -5), makeSchedulingResult()) + re.PreRequest(ctx, makeInferenceRequest("req-high", -1), makeSchedulingResult()) + + // Demand at priority -1: only victims strictly below -1 may be evicted, even though n allows + // two. The -1 victim would be same-band churn against the demand. + evicted, err := re.EvictN(ctx, 2, -1) + require.NoError(t, err) + assert.Equal(t, []string{"req-low"}, evicted, "only victims strictly below the bound may be evicted") + + inFlight, evictable := re.Stats() + assert.Equal(t, 1, inFlight, "the at-bound victim must remain tracked") + assert.Equal(t, 1, evictable, "the at-bound victim must remain evictable for future decisions") +} diff --git a/pkg/epp/flowcontrol/integration_test.go b/pkg/epp/flowcontrol/integration_test.go index c3b03697a4..bc8d6311bc 100644 --- a/pkg/epp/flowcontrol/integration_test.go +++ b/pkg/epp/flowcontrol/integration_test.go @@ -695,7 +695,7 @@ func TestEvictionPipeline(t *testing.T) { sheddableCh := reg.Get("shed-1") require.NotNil(t, sheddableCh, "sheddable request should have an eviction channel") - evictedIDs, err := requestEvictor.EvictN(ctx, 1) + evictedIDs, err := requestEvictor.EvictN(ctx, 1, 0) require.NoError(t, err) require.Equal(t, []string{"shed-1"}, evictedIDs) From 792e77477934320ae1e1dd182c60d90c45df891e Mon Sep 17 00:00:00 2001 From: Luke Van Drie Date: Mon, 20 Jul 2026 16:04:38 -0700 Subject: [PATCH 2/6] feat(flowcontrol): reclamation controller for demand-driven in-flight eviction Implements the controller from docs/flow-control-eviction.md (PR #2061): on HoL blocking, scan blocked bands for queued demand strictly above the victim head, size revocations as a saturation deficit against a mean-footprint credit with pending-reclaim debits, and pace by confirmation (stream termination) plus a grace covering the sensor's visibility lag, with a timeout so a hung stream cannot wedge the gate. Bands with unattainable ceilings are excluded from demand. Adds the flowControl.enableEviction API field and its translation. New metrics: revocations issued and terminal outcomes, reclaim target, pending reclaim, confirmation latency. Signed-off-by: Luke Van Drie --- .../v1alpha1/endpointpickerconfig_types.go | 12 + pkg/epp/flowcontrol/controller/config.go | 86 +++- pkg/epp/flowcontrol/controller/config_test.go | 79 +++- pkg/epp/flowcontrol/controller/controller.go | 32 ++ .../flowcontrol/controller/controller_test.go | 1 + .../controller/internal/processor.go | 9 + .../controller/internal/processor_test.go | 3 +- .../controller/internal/reclamation.go | 336 ++++++++++++++++ .../controller/internal/reclamation_test.go | 375 ++++++++++++++++++ pkg/epp/metrics/llm_d_router_metrics.go | 46 +++ pkg/epp/metrics/metrics.go | 42 ++ 11 files changed, 1008 insertions(+), 13 deletions(-) create mode 100644 pkg/epp/flowcontrol/controller/internal/reclamation.go create mode 100644 pkg/epp/flowcontrol/controller/internal/reclamation_test.go diff --git a/apix/config/v1alpha1/endpointpickerconfig_types.go b/apix/config/v1alpha1/endpointpickerconfig_types.go index 3926f924e5..3b8112abe5 100644 --- a/apix/config/v1alpha1/endpointpickerconfig_types.go +++ b/apix/config/v1alpha1/endpointpickerconfig_types.go @@ -401,6 +401,14 @@ type FlowControlConfig struct { // SaturationDetector specifies which saturation detector plugin to use for both Admission and // Flow Control. If omitted, "utilization-detector" is used by default. SaturationDetector *SaturationDetectorConfig `json:"saturationDetector,omitempty"` + + // +optional + // EnableEviction enables demand-driven in-flight eviction. When higher-priority requests are + // blocked by pool saturation, lower-priority in-flight requests (priority < 0) may be + // terminated to reclaim capacity. Pacing and sizing self-configure from the selected + // saturation detector. See docs/flow-control-eviction.md. + // Defaults to false. + EnableEviction bool `json:"enableEviction,omitempty"` } func (fcc *FlowControlConfig) String() string { @@ -445,6 +453,10 @@ func (fcc *FlowControlConfig) String() string { parts = append(parts, fmt.Sprintf("SaturationDetector: %v", fcc.SaturationDetector)) } + if fcc.EnableEviction { + parts = append(parts, "EnableEviction: true") + } + return "{" + strings.Join(parts, ", ") + "}" } diff --git a/pkg/epp/flowcontrol/controller/config.go b/pkg/epp/flowcontrol/controller/config.go index b154d4a8be..8596306df9 100644 --- a/pkg/epp/flowcontrol/controller/config.go +++ b/pkg/epp/flowcontrol/controller/config.go @@ -28,6 +28,16 @@ const ( defaultExpiryCleanupInterval = 1 * time.Second // defaultEnqueueChannelBufferSize is the default size of a worker's incoming request buffer. defaultEnqueueChannelBufferSize = 100 + // defaultMaxRevocationsPerDecision caps revocations per reclamation decision, bounding the + // damage from mean-footprint misestimation. + defaultMaxRevocationsPerDecision = 2 + // defaultEvictionConfirmationGrace covers the reclaiming stage (engine abort, KV GC, scrape, + // staleness window) for the default utilization detector. Deployments pairing eviction with the + // concurrency detector can lower it substantially. + defaultEvictionConfirmationGrace = 500 * time.Millisecond + // defaultEvictionConfirmationTimeout bounds how long an unconfirmed revocation can hold the + // reclamation pacing gate closed. + defaultEvictionConfirmationTimeout = 10 * time.Second ) // Config holds the configuration for the `FlowController`. @@ -46,6 +56,33 @@ type Config struct { // serial execution loop and allowing the system to handle short bursts of traffic without blocking. // Optional: Defaults to `defaultEnqueueChannelBufferSize` (100). EnqueueChannelBufferSize int + + // EnableEviction enables demand-driven in-flight eviction: when higher-priority requests are + // blocked by pool saturation, lower-priority in-flight requests may be terminated to reclaim + // capacity. Requires the eviction plumbing to be wired (see Deps.InFlightEvictor). + // See docs/flow-control-eviction.md. + // Optional: Defaults to false. + EnableEviction bool + + // MaxRevocationsPerDecision caps how many revocations a single reclamation decision may issue. + // Not exposed through the API configuration: benchmark data shows sizing is deficit-bound, so + // this cap rarely binds and is not worth a user-facing knob. + // Optional: Defaults to `defaultMaxRevocationsPerDecision` (2). + MaxRevocationsPerDecision int + + // EvictionConfirmationGrace is how long a confirmed revocation's pending-reclaim debit keeps + // suppressing further reclamation, covering the saturation signal's confirmation-to-visibility + // lag. Not exposed through the API configuration: the EPP wiring derives it from the selected + // saturation detector, since the correct value is a property of the sensor, not a preference. + // Optional: Defaults to `defaultEvictionConfirmationGrace` (500ms), the conservative value for + // scraped sensors. + EvictionConfirmationGrace time.Duration + + // EvictionConfirmationTimeout bounds how long an unconfirmed revocation can hold the + // reclamation pacing gate closed before being treated as confirmed. Not exposed through the + // API configuration. + // Optional: Defaults to `defaultEvictionConfirmationTimeout` (10s). + EvictionConfirmationTimeout time.Duration } func (c *Config) String() string { @@ -64,11 +101,14 @@ type ConfigOption func(*Config) // NewConfigFromAPI creates a new Config from the API configuration. func NewConfigFromAPI(apiConfig *configapi.FlowControlConfig) (*Config, error) { - opts := make([]ConfigOption, 0, 1) + opts := make([]ConfigOption, 0, 4) if apiConfig != nil { if apiConfig.DefaultRequestTTL != nil { opts = append(opts, WithDefaultRequestTTL(apiConfig.DefaultRequestTTL.Duration)) } + if apiConfig.EnableEviction { + opts = append(opts, WithEnableEviction(true)) + } } return NewConfig(opts...) } @@ -76,8 +116,11 @@ func NewConfigFromAPI(apiConfig *configapi.FlowControlConfig) (*Config, error) { // NewConfig creates a new Config with the given options, applying defaults and validation. func NewConfig(opts ...ConfigOption) (*Config, error) { c := &Config{ - ExpiryCleanupInterval: defaultExpiryCleanupInterval, - EnqueueChannelBufferSize: defaultEnqueueChannelBufferSize, + ExpiryCleanupInterval: defaultExpiryCleanupInterval, + EnqueueChannelBufferSize: defaultEnqueueChannelBufferSize, + MaxRevocationsPerDecision: defaultMaxRevocationsPerDecision, + EvictionConfirmationGrace: defaultEvictionConfirmationGrace, + EvictionConfirmationTimeout: defaultEvictionConfirmationTimeout, } for _, opt := range opts { @@ -111,6 +154,34 @@ func WithEnqueueChannelBufferSize(size int) ConfigOption { } } +// WithEnableEviction enables demand-driven in-flight eviction. +func WithEnableEviction(enabled bool) ConfigOption { + return func(c *Config) { + c.EnableEviction = enabled + } +} + +// WithMaxRevocationsPerDecision sets the per-decision revocation cap. +func WithMaxRevocationsPerDecision(n int) ConfigOption { + return func(c *Config) { + c.MaxRevocationsPerDecision = n + } +} + +// WithEvictionConfirmationGrace sets the post-confirmation grace period. +func WithEvictionConfirmationGrace(d time.Duration) ConfigOption { + return func(c *Config) { + c.EvictionConfirmationGrace = d + } +} + +// WithEvictionConfirmationTimeout sets the confirmation timeout. +func WithEvictionConfirmationTimeout(d time.Duration) ConfigOption { + return func(c *Config) { + c.EvictionConfirmationTimeout = d + } +} + // validate checks the configuration for validity. func (c *Config) validate() error { if c.DefaultRequestTTL < 0 { @@ -122,5 +193,14 @@ func (c *Config) validate() error { if c.EnqueueChannelBufferSize < 0 { return fmt.Errorf("EnqueueChannelBufferSize cannot be negative, but got %d", c.EnqueueChannelBufferSize) } + if c.MaxRevocationsPerDecision < 1 { + return fmt.Errorf("MaxRevocationsPerDecision must be at least 1, but got %d", c.MaxRevocationsPerDecision) + } + if c.EvictionConfirmationGrace < 0 { + return fmt.Errorf("EvictionConfirmationGrace cannot be negative, but got %v", c.EvictionConfirmationGrace) + } + if c.EvictionConfirmationTimeout <= 0 { + return fmt.Errorf("EvictionConfirmationTimeout must be positive, but got %v", c.EvictionConfirmationTimeout) + } return nil } diff --git a/pkg/epp/flowcontrol/controller/config_test.go b/pkg/epp/flowcontrol/controller/config_test.go index 89773a3c18..5d91059ba1 100644 --- a/pkg/epp/flowcontrol/controller/config_test.go +++ b/pkg/epp/flowcontrol/controller/config_test.go @@ -41,9 +41,12 @@ func TestNewConfig(t *testing.T) { opts: nil, expectErr: false, expectedCfg: Config{ - DefaultRequestTTL: 0, - ExpiryCleanupInterval: defaultExpiryCleanupInterval, - EnqueueChannelBufferSize: defaultEnqueueChannelBufferSize, + DefaultRequestTTL: 0, + ExpiryCleanupInterval: defaultExpiryCleanupInterval, + EnqueueChannelBufferSize: defaultEnqueueChannelBufferSize, + MaxRevocationsPerDecision: defaultMaxRevocationsPerDecision, + EvictionConfirmationGrace: defaultEvictionConfirmationGrace, + EvictionConfirmationTimeout: defaultEvictionConfirmationTimeout, }, }, { @@ -53,9 +56,12 @@ func TestNewConfig(t *testing.T) { }, expectErr: false, expectedCfg: Config{ - DefaultRequestTTL: 10 * time.Second, - ExpiryCleanupInterval: defaultExpiryCleanupInterval, - EnqueueChannelBufferSize: defaultEnqueueChannelBufferSize, + DefaultRequestTTL: 10 * time.Second, + ExpiryCleanupInterval: defaultExpiryCleanupInterval, + EnqueueChannelBufferSize: defaultEnqueueChannelBufferSize, + MaxRevocationsPerDecision: defaultMaxRevocationsPerDecision, + EvictionConfirmationGrace: defaultEvictionConfirmationGrace, + EvictionConfirmationTimeout: defaultEvictionConfirmationTimeout, }, }, { @@ -67,9 +73,12 @@ func TestNewConfig(t *testing.T) { }, expectErr: false, expectedCfg: Config{ - DefaultRequestTTL: 10 * time.Second, - ExpiryCleanupInterval: 2 * time.Second, - EnqueueChannelBufferSize: 50, + DefaultRequestTTL: 10 * time.Second, + ExpiryCleanupInterval: 2 * time.Second, + EnqueueChannelBufferSize: 50, + MaxRevocationsPerDecision: defaultMaxRevocationsPerDecision, + EvictionConfirmationGrace: defaultEvictionConfirmationGrace, + EvictionConfirmationTimeout: defaultEvictionConfirmationTimeout, }, }, { @@ -100,6 +109,45 @@ func TestNewConfig(t *testing.T) { }, expectErr: true, }, + { + name: "WithEvictionOptions_ShouldUpdateConfig", + opts: []ConfigOption{ + WithEnableEviction(true), + WithMaxRevocationsPerDecision(5), + WithEvictionConfirmationGrace(50 * time.Millisecond), + WithEvictionConfirmationTimeout(30 * time.Second), + }, + expectErr: false, + expectedCfg: Config{ + ExpiryCleanupInterval: defaultExpiryCleanupInterval, + EnqueueChannelBufferSize: defaultEnqueueChannelBufferSize, + EnableEviction: true, + MaxRevocationsPerDecision: 5, + EvictionConfirmationGrace: 50 * time.Millisecond, + EvictionConfirmationTimeout: 30 * time.Second, + }, + }, + { + name: "ZeroMaxRevocationsPerDecision_ShouldError", + opts: []ConfigOption{ + WithMaxRevocationsPerDecision(0), + }, + expectErr: true, + }, + { + name: "NegativeEvictionConfirmationGrace_ShouldError", + opts: []ConfigOption{ + WithEvictionConfirmationGrace(-1 * time.Second), + }, + expectErr: true, + }, + { + name: "ZeroEvictionConfirmationTimeout_ShouldError", + opts: []ConfigOption{ + WithEvictionConfirmationTimeout(0), + }, + expectErr: true, + }, } for _, tc := range testCases { @@ -175,6 +223,19 @@ func TestNewConfigFromAPI(t *testing.T) { }, expectedErr: "DefaultRequestTTL cannot be negative", }, + { + name: "EnableEviction_ShouldTranslate_WithInternalDefaults", + apiConfig: &configapi.FlowControlConfig{ + EnableEviction: true, + }, + assertion: func(t *testing.T, cfg *Config) { + assert.True(t, cfg.EnableEviction, "EnableEviction should be translated") + // Pacing and sizing parameters are internal: defaulted here, derived at wiring time. + assert.Equal(t, defaultMaxRevocationsPerDecision, cfg.MaxRevocationsPerDecision) + assert.Equal(t, defaultEvictionConfirmationGrace, cfg.EvictionConfirmationGrace) + assert.Equal(t, defaultEvictionConfirmationTimeout, cfg.EvictionConfirmationTimeout) + }, + }, } for _, tc := range testCases { diff --git a/pkg/epp/flowcontrol/controller/controller.go b/pkg/epp/flowcontrol/controller/controller.go index 2bcb2293be..8f5de80d65 100644 --- a/pkg/epp/flowcontrol/controller/controller.go +++ b/pkg/epp/flowcontrol/controller/controller.go @@ -60,6 +60,7 @@ type processorFactory func( cleanupSweepInterval time.Duration, enqueueChannelBufferSize int, logger logr.Logger, + reclamation *internal.ReclamationController, ) processor var _ processor = &internal.Processor{} @@ -105,6 +106,11 @@ type Deps struct { UsageLimitPolicy flowcontrol.UsageLimitPolicy Clock clock.WithTicker ProcessorFactory processorFactory + + // InFlightEvictor enables demand-driven in-flight eviction when Config.EnableEviction is set. + // Satisfied by *eviction.RequestEvictor. The FlowController registers the reclamation + // controller's Confirm as the evictor's eviction-terminated listener. + InFlightEvictor internal.InFlightEvictor } // NewFlowController creates and starts a new FlowController instance. @@ -147,6 +153,7 @@ func NewFlowController( cleanupSweepInterval time.Duration, enqueueChannelBufferSize int, logger logr.Logger, + reclamation *internal.ReclamationController, ) processor { return internal.NewProcessor( ctx, @@ -160,12 +167,36 @@ func NewFlowController( cleanupSweepInterval, enqueueChannelBufferSize, logger, + reclamation, ) } } else { fc.processorFactory = deps.ProcessorFactory } + // Demand-driven in-flight eviction is active only when both the config enables it and the + // eviction plumbing was wired in. The reclamation controller's Confirm becomes the evictor's + // eviction-terminated listener, closing the confirmation loop. + var reclamation *internal.ReclamationController + if config.EnableEviction && deps.InFlightEvictor != nil { + reclamation = internal.NewReclamationController( + internal.ReclamationConfig{ + MaxRevocationsPerDecision: config.MaxRevocationsPerDecision, + ConfirmationGrace: config.EvictionConfirmationGrace, + ConfirmationTimeout: config.EvictionConfirmationTimeout, + }, + deps.InFlightEvictor, + deps.Clock, + fc.logger, + poolName, + ) + deps.InFlightEvictor.SetEvictionTerminatedListener(reclamation.Confirm) + fc.logger.V(logutil.DEFAULT).Info("Demand-driven in-flight eviction enabled.", + "maxRevocationsPerDecision", config.MaxRevocationsPerDecision, + "confirmationGrace", config.EvictionConfirmationGrace, + "confirmationTimeout", config.EvictionConfirmationTimeout) + } + // Construct a new worker, but do not start its goroutine yet. fc.processor = fc.processorFactory( fc.parentCtx, @@ -178,6 +209,7 @@ func NewFlowController( fc.config.ExpiryCleanupInterval, fc.config.EnqueueChannelBufferSize, fc.logger, + reclamation, ) fc.logger.V(logutil.DEFAULT).Info("Starting the Processor.") diff --git a/pkg/epp/flowcontrol/controller/controller_test.go b/pkg/epp/flowcontrol/controller/controller_test.go index 68e4dd3db7..ecac47448f 100644 --- a/pkg/epp/flowcontrol/controller/controller_test.go +++ b/pkg/epp/flowcontrol/controller/controller_test.go @@ -287,6 +287,7 @@ func (f *mockProcessorFactory) new( _ time.Duration, _ int, _ logr.Logger, + _ *internal.ReclamationController, ) processor { if f.processor != nil { return f.processor diff --git a/pkg/epp/flowcontrol/controller/internal/processor.go b/pkg/epp/flowcontrol/controller/internal/processor.go index d8c32d257e..121cf1f222 100644 --- a/pkg/epp/flowcontrol/controller/internal/processor.go +++ b/pkg/epp/flowcontrol/controller/internal/processor.go @@ -75,6 +75,10 @@ type Processor struct { cleanupSweepInterval time.Duration logger logr.Logger + // reclamation, when non-nil, enables demand-driven in-flight eviction on HoL blocking. + // See docs/flow-control-eviction.md. + reclamation *ReclamationController + // lifecycleCtx controls the processor's lifetime. Monitored by Submit* methods for safe shutdown. lifecycleCtx context.Context @@ -106,6 +110,7 @@ func NewProcessor( cleanupSweepInterval time.Duration, enqueueChannelBufferSize int, logger logr.Logger, + reclamation *ReclamationController, ) *Processor { return &Processor{ registry: registry, @@ -119,6 +124,7 @@ func NewProcessor( logger: logger, lifecycleCtx: ctx, enqueueChan: make(chan *FlowItem, enqueueChannelBufferSize), + reclamation: reclamation, } } @@ -377,6 +383,9 @@ func (p *Processor) dispatchCycle(ctx context.Context) bool { if saturation >= usageLimit { p.logger.V(logutil.DEBUG).Info("Priority band is saturated; enforcing HoL blocking.", "priority", priority, "saturation", saturation, "usageLimit", usageLimit) + if p.reclamation != nil { + p.maybeReclaim(ctx, saturation, priorities, ceilings, i) + } // Stop the dispatch cycle entirely to respect strict policy decision and prevent priority inversion where // lower-priority work might exacerbate the saturation affecting high-priority work. return false diff --git a/pkg/epp/flowcontrol/controller/internal/processor_test.go b/pkg/epp/flowcontrol/controller/internal/processor_test.go index 20be2f473b..921dc172d0 100644 --- a/pkg/epp/flowcontrol/controller/internal/processor_test.go +++ b/pkg/epp/flowcontrol/controller/internal/processor_test.go @@ -142,7 +142,8 @@ func newTestHarness(t *testing.T, expiryCleanupInterval time.Duration) *testHarn h.clock, expiryCleanupInterval, 100, - h.logger) + h.logger, + nil) require.NotNil(t, h.processor, "NewProcessor should not return nil") t.Cleanup(func() { h.Stop() }) diff --git a/pkg/epp/flowcontrol/controller/internal/reclamation.go b/pkg/epp/flowcontrol/controller/internal/reclamation.go new file mode 100644 index 0000000000..95a109d051 --- /dev/null +++ b/pkg/epp/flowcontrol/controller/internal/reclamation.go @@ -0,0 +1,336 @@ +/* +Copyright 2026 The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package internal + +import ( + "context" + "math" + "sync" + "time" + + "github.com/go-logr/logr" + "k8s.io/utils/clock" + + logutil "github.com/llm-d/llm-d-router/pkg/common/observability/logging" + "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/flowcontrol" + "github.com/llm-d/llm-d-router/pkg/epp/metrics" +) + +// InFlightEvictor is the narrow view of the eviction subsystem the reclamation controller needs. +// It is satisfied by *eviction.RequestEvictor. +type InFlightEvictor interface { + // EvictN evicts up to n requests in victim-policy order, stopping at the first victim whose + // priority is not strictly below priorityBound, and returns the IDs actually evicted. + EvictN(ctx context.Context, n int, priorityBound int) ([]string, error) + // Stats returns the number of tracked in-flight requests and how many are evictable. + Stats() (inFlight int, evictable int) + // PeekVictimPriority returns the priority of the next victim, or false if none are evictable. + PeekVictimPriority() (priority int, ok bool) + // SetEvictionTerminatedListener registers the confirmation callback, invoked once per evicted + // request when its stream terminates. + SetEvictionTerminatedListener(listener func(requestID string)) +} + +// ReclamationConfig holds the tuning parameters for the ReclamationController. +// See docs/flow-control-eviction.md for the derivation of each parameter. +type ReclamationConfig struct { + // MaxRevocationsPerDecision caps how many revocations a single decision may issue. It bounds the + // damage from mean-footprint misestimation. + MaxRevocationsPerDecision int + + // ConfirmationGrace is the wait after the last confirmation before the next decision may run. It + // covers the reclaiming stage: engine abort, KV GC, metrics scrape, and staleness window. + ConfirmationGrace time.Duration + + // ConfirmationTimeout bounds how long an unconfirmed revocation can hold the pacing gate closed. + // After it expires the revocation is treated as confirmed and a health metric is incremented. + ConfirmationTimeout time.Duration +} + +// revocation is one issued, not-yet-confirmed eviction. +type revocation struct { + // credit is the pending-reclaim debit taken for this revocation, in saturation-gauge units, + // retained at its issue-time value until the debit expires. + credit float64 + issuedAt time.Time +} + +// coolingDebit is a confirmed revocation's debit held through the reclaiming stage: the stream is +// dead, but the freed capacity is not yet visible in the gauge (engine GC, scrape, staleness). The +// debit expires ConfirmationGrace after confirmation. +type coolingDebit struct { + credit float64 + expiresAt time.Time +} + +// ReclamationController decides when in-flight eviction fires and how many leases it revokes. +// It implements the sizing and pacing rules from docs/flow-control-eviction.md: +// +// - Sizing: deficit in saturation-gauge units against a mean-footprint credit estimate, capped by +// MaxRevocationsPerDecision. +// - Pacing: confirmation-gated. A new decision may run only once every previously issued +// revocation is confirmed (its ext_proc stream terminated) and ConfirmationGrace has elapsed, +// so the controller never acts twice on a gauge that has not absorbed its prior actions. +// +// Concurrency: Reclaim and GateOpen are called only from the processor's single-writer run loop. +// Confirm is called from ext_proc handler goroutines. The mutex is held across EvictN inside +// Reclaim so a confirmation arriving mid-decision blocks until its revocation is registered; the +// evictor never calls back into the controller synchronously, so this cannot deadlock. +type ReclamationController struct { + evictor InFlightEvictor + cfg ReclamationConfig + clock clock.PassiveClock + logger logr.Logger + poolName string + + mu sync.Mutex + outstanding map[string]revocation + cooling []coolingDebit + pendingReclaim float64 +} + +// NewReclamationController constructs a controller. The caller is responsible for registering +// Confirm as the evictor's eviction-terminated listener. +func NewReclamationController( + cfg ReclamationConfig, + evictor InFlightEvictor, + clk clock.PassiveClock, + logger logr.Logger, + poolName string, +) *ReclamationController { + return &ReclamationController{ + evictor: evictor, + cfg: cfg, + clock: clk, + logger: logger.WithName("reclamation-controller"), + poolName: poolName, + outstanding: make(map[string]revocation), + } +} + +// GateOpen reports whether a new reclamation decision may run: stop-and-wait, open only when +// nothing is outstanding and nothing is cooling, i.e. every prior revocation has confirmed and had +// ConfirmationGrace to become gauge-visible. It first retires timed-out revocations and expired +// cooling debits. This is the cheap early-exit the dispatch cycle calls on every HoL break. +// +// A sliding-window relaxation (issue while outstanding < W) was prototyped and benchmarked with no +// measurable benefit: sizing is deficit-bound at the ceiling boundary, so the epoch rate, not the +// gate, limits reclaim throughput. See docs/flow-control-eviction.md, Alternatives. +func (c *ReclamationController) GateOpen(now time.Time) bool { + c.mu.Lock() + defer c.mu.Unlock() + c.sweepTimeoutsLocked(now) + c.expireCoolingLocked(now) + return len(c.outstanding) == 0 && len(c.cooling) == 0 +} + +// VictimPriority returns the priority of the next lease the selector would revoke. +func (c *ReclamationController) VictimPriority() (int, bool) { + return c.evictor.PeekVictimPriority() +} + +// Reclaim runs one sizing decision and issues the resulting revocations. The caller must have +// observed GateOpen() == true in the same dispatch cycle. +// +// All quantities are dimensionless, in the saturation gauge's own units: +// +// reclaimTarget = saturation - ceiling(blocked band with eligible demand) +// credit = saturation / inFlight (mean lease footprint estimate) +// n = max(1, ceil((reclaimTarget - pendingReclaim) / credit)), +// capped at MaxRevocationsPerDecision and at the evictable lease count +// +// demandPriority is the eligible demand band's priority, passed to the actuator as a strict upper +// bound on victim priority so that no victim in the decision sits at or above the demand band +// (the victim-head check in the trigger does not cover later victims of a multi-revocation +// decision). +func (c *ReclamationController) Reclaim(ctx context.Context, saturation, ceiling float64, demandPriority int) { + inFlight, evictable := c.evictor.Stats() + if inFlight <= 0 || evictable <= 0 { + return + } + credit := saturation / float64(inFlight) + if credit <= 0 || math.IsNaN(credit) || math.IsInf(credit, 0) { + return + } + + c.mu.Lock() + defer c.mu.Unlock() + + reclaimTarget := saturation - ceiling + netTarget := reclaimTarget - c.pendingReclaim + metrics.RecordFlowControlReclaimTarget(c.poolName, reclaimTarget) + // Under stop-and-wait pacing the gate opens only once every debit has expired, so + // pendingReclaim is zero here and this branch never executes through the dispatch path. The + // subtraction makes sizing safe independently of pacing: defense in depth for out-of-band + // calls, and the invariant any relaxed pacing scheme would rely on. + if netTarget < 0 || (netTarget == 0 && c.pendingReclaim > 0) { + return + } + + // The dispatch gate blocks on saturation >= ceiling, so unblocking requires reclaiming strictly + // more than the deficit. At the boundary (netTarget == 0 with nothing pending) the quotient + // rounds to zero, which would leave the band blocked with no reclamation; issue at least one + // revocation instead. + n := max(1, int(math.Ceil(netTarget/credit))) + if n > c.cfg.MaxRevocationsPerDecision { + n = c.cfg.MaxRevocationsPerDecision + } + if n > evictable { + n = evictable + } + + // The mutex is held across EvictN so that a confirmation racing the registration below blocks in + // Confirm until the revocation is present in the outstanding set. + evicted, err := c.evictor.EvictN(ctx, n, demandPriority) + if err != nil { + c.logger.Error(err, "Revocation batch failed", "requested", n) + return + } + now := c.clock.Now() + for _, requestID := range evicted { + c.outstanding[requestID] = revocation{credit: credit, issuedAt: now} + c.pendingReclaim += credit + } + metrics.RecordFlowControlRevocationsIssued(c.poolName, len(evicted)) + metrics.RecordFlowControlPendingReclaim(c.poolName, c.pendingReclaim) + + c.logger.V(logutil.DEBUG).Info("Revocations issued", + "saturation", saturation, "ceiling", ceiling, + "reclaimTarget", reclaimTarget, "credit", credit, + "requested", n, "issued", len(evicted)) +} + +// Confirm records that an evicted request's stream has terminated. Safe for concurrent use; called +// from ext_proc handler goroutines via the evictor's eviction-terminated listener. IDs that are +// not outstanding (e.g. already retired by timeout) are ignored. +func (c *ReclamationController) Confirm(requestID string) { + c.mu.Lock() + defer c.mu.Unlock() + r, ok := c.outstanding[requestID] + if !ok { + return + } + delete(c.outstanding, requestID) + now := c.clock.Now() + c.startCoolingLocked(r, now) + metrics.RecordFlowControlRevocations(c.poolName, metrics.RevocationOutcomeConfirmed, 1) + metrics.RecordFlowControlRevocationConfirmationDuration(c.poolName, now.Sub(r.issuedAt)) +} + +// sweepTimeoutsLocked retires revocations whose confirmation never arrived within +// ConfirmationTimeout, so a hung stream cannot hold the pacing gate closed forever. +func (c *ReclamationController) sweepTimeoutsLocked(now time.Time) { + for requestID, r := range c.outstanding { + if now.Sub(r.issuedAt) < c.cfg.ConfirmationTimeout { + continue + } + delete(c.outstanding, requestID) + c.startCoolingLocked(r, now) + metrics.RecordFlowControlRevocations(c.poolName, metrics.RevocationOutcomeTimedOut, 1) + c.logger.V(logutil.DEFAULT).Info("Revocation confirmation timed out; treating as confirmed", + "requestID", requestID, "timeout", c.cfg.ConfirmationTimeout) + } +} + +// startCoolingLocked moves a confirmed revocation's debit into the reclaiming (cooling) stage: the +// stream is dead, but the freed capacity is assumed gauge-invisible for another ConfirmationGrace, +// so the debit keeps suppressing re-sizing until then. +func (c *ReclamationController) startCoolingLocked(r revocation, now time.Time) { + c.cooling = append(c.cooling, coolingDebit{credit: r.credit, expiresAt: now.Add(c.cfg.ConfirmationGrace)}) +} + +// expireCoolingLocked releases cooling debits whose grace has elapsed. +func (c *ReclamationController) expireCoolingLocked(now time.Time) { + kept := c.cooling[:0] + changed := false + for _, d := range c.cooling { + if now.Before(d.expiresAt) { + kept = append(kept, d) + continue + } + changed = true + c.pendingReclaim -= d.credit + } + c.cooling = kept + if changed { + if c.pendingReclaim < 0 { + c.pendingReclaim = 0 + } + metrics.RecordFlowControlPendingReclaim(c.poolName, c.pendingReclaim) + } +} + +// maybeReclaim evaluates the eviction trigger on a HoL-blocking break at band index breakIdx. +// The pacing gate is checked before any queue scan, so the common saturated-but-gated case costs a +// single comparison. Demand is eligible when a blocked band holds queued requests whose priority +// is strictly greater than the current victim head's priority (no same-band churn). +// +// priorities is ordered highest first, so once a band's priority is not strictly greater than the +// victim's, no later band can be eligible and the scan stops. +func (p *Processor) maybeReclaim( + ctx context.Context, + saturation float64, + priorities []int, + ceilings []float64, + breakIdx int, +) { + if !p.reclamation.GateOpen(p.clock.Now()) { + return + } + victimPriority, ok := p.reclamation.VictimPriority() + if !ok { + return // Nothing is evictable. + } + for j := breakIdx; j < len(priorities); j++ { + if priorities[j] <= victimPriority { + return + } + // A zero ceiling is a policy statement that the band must not dispatch regardless of load. + // No amount of reclamation can unblock it (saturation >= 0 always holds), so treating its + // queue as demand would revoke leases in a loop for no benefit. + if ceilings[j] <= 0 { + continue + } + // Each band's ceiling is checked directly; monotonicity across bands is not assumed. An + // unblocked band needs no eviction because it will dispatch on a subsequent cycle. + if saturation < ceilings[j] { + continue + } + band, err := p.registry.PriorityBandAccessor(priorities[j]) + if err != nil { + continue + } + if !bandHasQueuedItems(band) { + continue + } + p.reclamation.Reclaim(ctx, saturation, ceilings[j], priorities[j]) + return + } +} + +// bandHasQueuedItems reports whether any flow queue in the band is non-empty. +func bandHasQueuedItems(band flowcontrol.PriorityBandAccessor) bool { + found := false + band.IterateQueues(func(q flowcontrol.FlowQueueAccessor) bool { + if q.Len() > 0 { + found = true + return false + } + return true + }) + return found +} diff --git a/pkg/epp/flowcontrol/controller/internal/reclamation_test.go b/pkg/epp/flowcontrol/controller/internal/reclamation_test.go new file mode 100644 index 0000000000..af84582b36 --- /dev/null +++ b/pkg/epp/flowcontrol/controller/internal/reclamation_test.go @@ -0,0 +1,375 @@ +/* +Copyright 2026 The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package internal + +import ( + "context" + "fmt" + "sync" + "testing" + "time" + + "github.com/go-logr/logr" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + testclock "k8s.io/utils/clock/testing" + + fwkdl "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/datalayer" +) + +// fakeInFlightEvictor is a configurable InFlightEvictor for controller tests. +type fakeInFlightEvictor struct { + mu sync.Mutex + inFlight int + evictable int + victimPriority int + hasVictim bool + evictErr error + + evictCalls []int // n per EvictN call + boundCalls []int // priorityBound per EvictN call + nextID int + listener func(requestID string) +} + +func (f *fakeInFlightEvictor) EvictN(_ context.Context, n int, priorityBound int) ([]string, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.evictCalls = append(f.evictCalls, n) + f.boundCalls = append(f.boundCalls, priorityBound) + if f.evictErr != nil { + return nil, f.evictErr + } + if n > f.evictable { + n = f.evictable + } + ids := make([]string, 0, n) + for range n { + ids = append(ids, fmt.Sprintf("req-%d", f.nextID)) + f.nextID++ + } + f.evictable -= n + f.inFlight -= n + return ids, nil +} + +func (f *fakeInFlightEvictor) Stats() (int, int) { + f.mu.Lock() + defer f.mu.Unlock() + return f.inFlight, f.evictable +} + +func (f *fakeInFlightEvictor) PeekVictimPriority() (int, bool) { + f.mu.Lock() + defer f.mu.Unlock() + return f.victimPriority, f.hasVictim +} + +func (f *fakeInFlightEvictor) SetEvictionTerminatedListener(listener func(requestID string)) { + f.mu.Lock() + defer f.mu.Unlock() + f.listener = listener +} + +func (f *fakeInFlightEvictor) totalEvictCalls() []int { + f.mu.Lock() + defer f.mu.Unlock() + if len(f.evictCalls) == 0 { + return nil + } + out := make([]int, len(f.evictCalls)) + copy(out, f.evictCalls) + return out +} + +func newTestReclamationController( + evictor *fakeInFlightEvictor, + clk *testclock.FakeClock, + cfg ReclamationConfig, +) *ReclamationController { + return NewReclamationController(cfg, evictor, clk, logr.Discard(), "test-pool") +} + +var testReclamationConfig = ReclamationConfig{ + MaxRevocationsPerDecision: 2, + ConfirmationGrace: 100 * time.Millisecond, + ConfirmationTimeout: 5 * time.Second, +} + +func TestReclamationController_Sizing(t *testing.T) { + t.Parallel() + + testCases := []struct { + name string + inFlight int + evictable int + saturation float64 + ceiling float64 + expectedN []int // expected EvictN call args; empty means no call + }{ + { + // credit = 1.2/6 = 0.2; deficit = 0.2; n = 1. + name: "SmallDeficit_EvictsOne", + inFlight: 6, evictable: 6, saturation: 1.2, ceiling: 1.0, + expectedN: []int{1}, + }, + { + // credit = 2.0/4 = 0.5; deficit = 1.0; n = 2 = cap. + name: "DeepOverload_CappedByMaxPerDecision", + inFlight: 4, evictable: 4, saturation: 2.0, ceiling: 1.0, + expectedN: []int{2}, + }, + { + // n would be 2 but only 1 lease is evictable. + name: "CappedByEvictable", + inFlight: 4, evictable: 1, saturation: 2.0, ceiling: 1.0, + expectedN: []int{1}, + }, + { + // Saturation below the ceiling: no deficit, no eviction. + name: "NoDeficit_NoEviction", + inFlight: 6, evictable: 6, saturation: 0.9, ceiling: 1.0, + expectedN: nil, + }, + { + // The dispatch gate blocks at saturation == ceiling, but the deficit is exactly zero. + // The controller must still issue one revocation or the band deadlocks until churn. + name: "ExactBoundary_EvictsOne", + inFlight: 10, evictable: 10, saturation: 1.0, ceiling: 1.0, + expectedN: []int{1}, + }, + { + // No tracked leases: credit is undefined and there are no victims. + name: "ZeroLeases_NoEviction", + inFlight: 0, evictable: 0, saturation: 1.5, ceiling: 1.0, + expectedN: nil, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + clk := testclock.NewFakeClock(time.Now()) + evictor := &fakeInFlightEvictor{inFlight: tc.inFlight, evictable: tc.evictable} + rc := newTestReclamationController(evictor, clk, testReclamationConfig) + + rc.Reclaim(context.Background(), tc.saturation, tc.ceiling, 0) + + assert.Equal(t, tc.expectedN, evictor.totalEvictCalls(), "EvictN calls should match expected sizing") + }) + } +} + +func TestReclamationController_GateLifecycle(t *testing.T) { + t.Parallel() + clk := testclock.NewFakeClock(time.Now()) + evictor := &fakeInFlightEvictor{inFlight: 4, evictable: 4} + rc := newTestReclamationController(evictor, clk, testReclamationConfig) + + require.True(t, rc.GateOpen(clk.Now()), "gate should be open initially") + + // Issue a decision: deficit 1.0, credit 0.5, n = 2. + rc.Reclaim(context.Background(), 2.0, 1.0, 0) + require.Equal(t, []int{2}, evictor.totalEvictCalls(), "decision should issue revocations") + assert.False(t, rc.GateOpen(clk.Now()), "gate should be closed while revocations are outstanding") + + // One confirmation is not enough. + rc.Confirm("req-0") + assert.False(t, rc.GateOpen(clk.Now()), "gate should stay closed until all revocations confirm") + + // All confirmed, but grace has not elapsed. + rc.Confirm("req-1") + assert.False(t, rc.GateOpen(clk.Now()), "gate should stay closed during the grace window") + + // Grace elapsed: gate reopens. + clk.Step(testReclamationConfig.ConfirmationGrace + time.Millisecond) + assert.True(t, rc.GateOpen(clk.Now()), "gate should reopen after all confirmations plus grace") +} + +func TestReclamationController_ConfirmUnknownID_Ignored(t *testing.T) { + t.Parallel() + clk := testclock.NewFakeClock(time.Now()) + evictor := &fakeInFlightEvictor{inFlight: 4, evictable: 4} + rc := newTestReclamationController(evictor, clk, testReclamationConfig) + + require.NotPanics(t, func() { rc.Confirm("req-unknown") }) + assert.True(t, rc.GateOpen(clk.Now()), "unknown confirmations must not perturb the gate") +} + +func TestReclamationController_ConfirmationTimeout_ReopensGate(t *testing.T) { + t.Parallel() + clk := testclock.NewFakeClock(time.Now()) + evictor := &fakeInFlightEvictor{inFlight: 4, evictable: 4} + rc := newTestReclamationController(evictor, clk, testReclamationConfig) + + rc.Reclaim(context.Background(), 2.0, 1.0, 0) + require.False(t, rc.GateOpen(clk.Now()), "gate closed while outstanding") + + // Confirmation never arrives; the timeout retires the revocations. + clk.Step(testReclamationConfig.ConfirmationTimeout + time.Millisecond) + require.False(t, rc.GateOpen(clk.Now()), + "first check after timeout retires entries but grace restarts from retirement") + clk.Step(testReclamationConfig.ConfirmationGrace + time.Millisecond) + assert.True(t, rc.GateOpen(clk.Now()), "gate should reopen after timeout retirement plus grace") + + // A late confirmation for a timed-out revocation is a no-op: no second outcome, and the gate + // stays open. + require.NotPanics(t, func() { rc.Confirm("req-0") }) + assert.True(t, rc.GateOpen(clk.Now()), "late confirmation must not re-close the gate") +} + +func TestReclamationController_PendingDebit_SuppressesRepeatDecision(t *testing.T) { + t.Parallel() + clk := testclock.NewFakeClock(time.Now()) + evictor := &fakeInFlightEvictor{inFlight: 4, evictable: 4} + rc := newTestReclamationController(evictor, clk, testReclamationConfig) + + rc.Reclaim(context.Background(), 2.0, 1.0, 0) + require.Equal(t, []int{2}, evictor.totalEvictCalls()) + + // Even if a caller bypassed the gate, the pending debit covers the deficit and no further + // revocations are issued for the same signal. + rc.Reclaim(context.Background(), 2.0, 1.0, 0) + assert.Equal(t, []int{2}, evictor.totalEvictCalls(), + "a second decision against the same stale signal must not issue more revocations") +} + +// --- Integration: dispatchCycle -> maybeReclaim --- + +// withReclamation attaches a reclamation controller backed by the fake evictor to the harness +// processor. +func withReclamation(h *testHarness, evictor *fakeInFlightEvictor, cfg ReclamationConfig) { + h.processor.reclamation = NewReclamationController(cfg, evictor, h.clock, logr.Discard(), "test-pool") +} + +func TestDispatchCycle_HoLBlock_TriggersReclamation(t *testing.T) { + t.Parallel() + h := newTestHarness(t, testCleanupTick) + evictor := &fakeInFlightEvictor{inFlight: 6, evictable: 3, victimPriority: -1, hasVictim: true} + withReclamation(h, evictor, testReclamationConfig) + + q := h.addQueue(testFlow) // Priority 10. + require.NoError(t, q.Add(h.newTestItem("req-blocked", testFlow, testTTL))) + + h.saturationDetector.SaturationFunc = func(context.Context, []fwkdl.Endpoint) float64 { return 1.2 } + + dispatched := h.processor.dispatchCycle(h.ctx) + + assert.False(t, dispatched, "saturated cycle must not dispatch") + // credit = 1.2/6 = 0.2; deficit = 0.2; n = 1. + assert.Equal(t, []int{1}, evictor.totalEvictCalls(), "HoL blocking with eligible demand should revoke") +} + +func TestDispatchCycle_NoQueuedDemand_NoReclamation(t *testing.T) { + t.Parallel() + h := newTestHarness(t, testCleanupTick) + evictor := &fakeInFlightEvictor{inFlight: 6, evictable: 3, victimPriority: -1, hasVictim: true} + withReclamation(h, evictor, testReclamationConfig) + + h.addQueue(testFlow) // Registered band, but its queue is empty. + + h.saturationDetector.SaturationFunc = func(context.Context, []fwkdl.Endpoint) float64 { return 1.2 } + + h.processor.dispatchCycle(h.ctx) + assert.Empty(t, evictor.totalEvictCalls(), "no queued demand means nothing to reclaim for") +} + +func TestDispatchCycle_VictimNotStrictlyLower_NoReclamation(t *testing.T) { + t.Parallel() + h := newTestHarness(t, testCleanupTick) + // Victim priority equals the blocked band's priority: same-band churn is forbidden. + evictor := &fakeInFlightEvictor{inFlight: 6, evictable: 3, victimPriority: testFlow.Priority, hasVictim: true} + withReclamation(h, evictor, testReclamationConfig) + + q := h.addQueue(testFlow) + require.NoError(t, q.Add(h.newTestItem("req-blocked", testFlow, testTTL))) + + h.saturationDetector.SaturationFunc = func(context.Context, []fwkdl.Endpoint) float64 { return 1.2 } + + h.processor.dispatchCycle(h.ctx) + assert.Empty(t, evictor.totalEvictCalls(), "demand must dominate the victim priority strictly") +} + +func TestDispatchCycle_NoVictims_NoReclamation(t *testing.T) { + t.Parallel() + h := newTestHarness(t, testCleanupTick) + evictor := &fakeInFlightEvictor{inFlight: 0, evictable: 0, hasVictim: false} + withReclamation(h, evictor, testReclamationConfig) + + q := h.addQueue(testFlow) + require.NoError(t, q.Add(h.newTestItem("req-blocked", testFlow, testTTL))) + + h.saturationDetector.SaturationFunc = func(context.Context, []fwkdl.Endpoint) float64 { return 1.2 } + + h.processor.dispatchCycle(h.ctx) + assert.Empty(t, evictor.totalEvictCalls(), "no evictable leases means no decision") +} + +func TestDispatchCycle_GateClosed_SkipsDemandScan(t *testing.T) { + t.Parallel() + h := newTestHarness(t, testCleanupTick) + evictor := &fakeInFlightEvictor{inFlight: 6, evictable: 6, victimPriority: -1, hasVictim: true} + withReclamation(h, evictor, testReclamationConfig) + + q := h.addQueue(testFlow) + require.NoError(t, q.Add(h.newTestItem("req-blocked", testFlow, testTTL))) + + h.saturationDetector.SaturationFunc = func(context.Context, []fwkdl.Endpoint) float64 { return 1.2 } + + // First cycle issues; the gate then closes until confirmation. + h.processor.dispatchCycle(h.ctx) + require.Equal(t, []int{1}, evictor.totalEvictCalls()) + + // Repeated cycles against the same stale gauge must not issue more revocations. + for range 10 { + h.processor.dispatchCycle(h.ctx) + } + assert.Equal(t, []int{1}, evictor.totalEvictCalls(), "gate must suppress repeat decisions until confirmation") +} + +func TestMaybeReclaim_ZeroCeilingBand_NotEligibleDemand(t *testing.T) { + t.Parallel() + h := newTestHarness(t, testCleanupTick) + evictor := &fakeInFlightEvictor{inFlight: 6, evictable: 6, victimPriority: -1, hasVictim: true} + withReclamation(h, evictor, testReclamationConfig) + + q := h.addQueue(testFlow) + require.NoError(t, q.Add(h.newTestItem("req-blocked", testFlow, testTTL))) + + // A zero-ceiling band is permanently blocked by policy; no amount of reclamation can unblock + // it, so its queue must not count as eviction demand. + h.processor.maybeReclaim(h.ctx, 1.0, []int{testFlow.Priority}, []float64{0}, 0) + assert.Empty(t, evictor.totalEvictCalls(), "a zero-ceiling band must not trigger reclamation") + + // Positive control: the same demand against an attainable ceiling reclaims. + h.processor.maybeReclaim(h.ctx, 1.0, []int{testFlow.Priority}, []float64{0.9}, 0) + assert.Equal(t, []int{1}, evictor.totalEvictCalls(), "an attainable blocked ceiling must reclaim") +} + +func TestReclaim_PassesDemandPriorityAsVictimBound(t *testing.T) { + t.Parallel() + clk := testclock.NewFakeClock(time.Now()) + evictor := &fakeInFlightEvictor{inFlight: 4, evictable: 4} + rc := newTestReclamationController(evictor, clk, testReclamationConfig) + + rc.Reclaim(context.Background(), 2.0, 1.0, -1) + + evictor.mu.Lock() + defer evictor.mu.Unlock() + require.Equal(t, []int{-1}, evictor.boundCalls, + "the demand band's priority must reach the actuator as the victim bound") +} diff --git a/pkg/epp/metrics/llm_d_router_metrics.go b/pkg/epp/metrics/llm_d_router_metrics.go index 26e9d06f32..eda5ac9829 100644 --- a/pkg/epp/metrics/llm_d_router_metrics.go +++ b/pkg/epp/metrics/llm_d_router_metrics.go @@ -364,6 +364,52 @@ var ( }, []string{"outcome", "priority", "inference_pool"}, ) + + llmdFlowControlRevocationsIssuedTotal = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Subsystem: LLMDRouterEndpointPickerSubsystem, + Name: "flow_control_revocations_issued_total", + Help: metricsutil.HelpMsgWithStability("Total number of in-flight eviction revocations issued.", compbasemetrics.ALPHA), + }, + []string{"inference_pool"}, + ) + + llmdFlowControlRevocationsTotal = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Subsystem: LLMDRouterEndpointPickerSubsystem, + Name: "flow_control_revocations_total", + Help: metricsutil.HelpMsgWithStability("Total number of in-flight eviction revocations by terminal outcome (confirmed, timed_out). Every issued revocation eventually increments exactly one outcome.", compbasemetrics.ALPHA), + }, + []string{"outcome", "inference_pool"}, + ) + + llmdFlowControlReclaimTarget = prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Subsystem: LLMDRouterEndpointPickerSubsystem, + Name: "flow_control_reclaim_target", + Help: metricsutil.HelpMsgWithStability("Last computed reclamation deficit, in saturation-gauge units.", compbasemetrics.ALPHA), + }, + []string{"inference_pool"}, + ) + + llmdFlowControlPendingReclaim = prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Subsystem: LLMDRouterEndpointPickerSubsystem, + Name: "flow_control_pending_reclaim", + Help: metricsutil.HelpMsgWithStability("Sum of outstanding and cooling pending-reclaim debits, in saturation-gauge units.", compbasemetrics.ALPHA), + }, + []string{"inference_pool"}, + ) + + llmdFlowControlRevocationConfirmationDuration = prometheus.NewHistogramVec( + prometheus.HistogramOpts{ + Subsystem: LLMDRouterEndpointPickerSubsystem, + Name: "flow_control_revocation_confirmation_seconds", + Help: metricsutil.HelpMsgWithStability("Time from revocation issue to confirmed stream termination.", compbasemetrics.ALPHA), + Buckets: []float64{0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10}, + }, + []string{"inference_pool"}, + ) ) // --- llm-d Inference Model Rewrite Metrics --- diff --git a/pkg/epp/metrics/metrics.go b/pkg/epp/metrics/metrics.go index 044d557b88..24d6058d02 100644 --- a/pkg/epp/metrics/metrics.go +++ b/pkg/epp/metrics/metrics.go @@ -473,6 +473,11 @@ func Register(customCollectors ...prometheus.Collector) { metrics.Registry.MustRegister(flowControlRequestEnqueueDuration) metrics.Registry.MustRegister(llmdFlowControlRequestEnqueueDuration) metrics.Registry.MustRegister(llmdFlowControlRequestsTotal) + metrics.Registry.MustRegister(llmdFlowControlRevocationsIssuedTotal) + metrics.Registry.MustRegister(llmdFlowControlRevocationsTotal) + metrics.Registry.MustRegister(llmdFlowControlReclaimTarget) + metrics.Registry.MustRegister(llmdFlowControlPendingReclaim) + metrics.Registry.MustRegister(llmdFlowControlRevocationConfirmationDuration) metrics.Registry.MustRegister(inferenceModelRewriteDecisionsTotal) metrics.Registry.MustRegister(llmdInferenceModelRewriteDecisionsTotal) metrics.Registry.MustRegister(DataLayerPollErrorsTotal) @@ -543,6 +548,11 @@ func Reset() { flowControlDispatchCycleDuration.Reset() llmdFlowControlDispatchCycleDuration.Reset() llmdFlowControlRequestsTotal.Reset() + llmdFlowControlRevocationsIssuedTotal.Reset() + llmdFlowControlRevocationsTotal.Reset() + llmdFlowControlReclaimTarget.Reset() + llmdFlowControlPendingReclaim.Reset() + llmdFlowControlRevocationConfirmationDuration.Reset() inferenceModelRewriteDecisionsTotal.Reset() llmdInferenceModelRewriteDecisionsTotal.Reset() DataLayerPollErrorsTotal.Reset() @@ -874,6 +884,38 @@ func IncFlowControlRequestsTotal(outcome, priority, inferencePool string) { llmdFlowControlRequestsTotal.WithLabelValues(outcome, priority, inferencePool).Inc() } +// Terminal revocation outcomes for the flow control revocations counter. Every issued revocation +// eventually increments exactly one outcome. +const ( + RevocationOutcomeConfirmed = "confirmed" + RevocationOutcomeTimedOut = "timed_out" +) + +// RecordFlowControlRevocationsIssued counts revocations at issue time. +func RecordFlowControlRevocationsIssued(inferencePool string, n int) { + llmdFlowControlRevocationsIssuedTotal.WithLabelValues(inferencePool).Add(float64(n)) +} + +// RecordFlowControlRevocations increments the revocation counter for a terminal outcome. +func RecordFlowControlRevocations(inferencePool, outcome string, n int) { + llmdFlowControlRevocationsTotal.WithLabelValues(outcome, inferencePool).Add(float64(n)) +} + +// RecordFlowControlReclaimTarget records the last computed reclamation deficit. +func RecordFlowControlReclaimTarget(inferencePool string, target float64) { + llmdFlowControlReclaimTarget.WithLabelValues(inferencePool).Set(target) +} + +// RecordFlowControlPendingReclaim records the capacity debited for unconfirmed revocations. +func RecordFlowControlPendingReclaim(inferencePool string, pending float64) { + llmdFlowControlPendingReclaim.WithLabelValues(inferencePool).Set(pending) +} + +// RecordFlowControlRevocationConfirmationDuration records issue-to-confirmation latency. +func RecordFlowControlRevocationConfirmationDuration(inferencePool string, duration time.Duration) { + llmdFlowControlRevocationConfirmationDuration.WithLabelValues(inferencePool).Observe(duration.Seconds()) +} + // RecordInferenceModelRewriteDecision records the routing decision for InferenceModelRewrite. // The rewrite name and target come from configuration; only the source model name is // request-derived and needs bounding (a generic rule matches arbitrary model names). From 9014b0ae81eb0b2d4c8983ef8007c2d8a4e98f4c Mon Sep 17 00:00:00 2001 From: Luke Van Drie Date: Mon, 20 Jul 2026 16:04:39 -0700 Subject: [PATCH 3/6] feat(epp): wire in-flight eviction behind flowControl.enableEviction The confirmation grace is derived at wiring time from the selected saturation detector (dispatch-tick scale for the concurrency detector; refresh interval + staleness + engine reclaim budget for scraped sensors); remaining controller parameters are internal. The Director tracks dispatches and stream termination; the ext_proc server receives the evict-channel lookup. Signed-off-by: Luke Van Drie Co-authored-by: Rishabh Saini Signed-off-by: Luke Van Drie --- cmd/epp/runner/runner.go | 105 ++++++++++++++++++++++++----- pkg/epp/requestcontrol/director.go | 25 +++++++ pkg/epp/server/runserver.go | 6 ++ 3 files changed, 121 insertions(+), 15 deletions(-) diff --git a/cmd/epp/runner/runner.go b/cmd/epp/runner/runner.go index c2fb54efbe..5ecaa9fd39 100644 --- a/cmd/epp/runner/runner.go +++ b/cmd/epp/runner/runner.go @@ -56,8 +56,10 @@ import ( "github.com/llm-d/llm-d-router/pkg/epp/flowcontrol" "github.com/llm-d/llm-d-router/pkg/epp/flowcontrol/contracts" fccontroller "github.com/llm-d/llm-d-router/pkg/epp/flowcontrol/controller" + fceviction "github.com/llm-d/llm-d-router/pkg/epp/flowcontrol/eviction" fcregistry "github.com/llm-d/llm-d-router/pkg/epp/flowcontrol/registry" fwkdl "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/datalayer" + fwkfc "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/flowcontrol" fwkplugin "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/plugin" attrconcurrency "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/attribute/concurrency" attrlatency "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/attribute/latency" @@ -70,6 +72,8 @@ import ( sourcemetrics "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/source/metrics" srcmodels "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/source/models" sourcenotifications "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/source/notifications" + evictfiltering "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/flowcontrol/eviction/filtering" + evictordering "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/flowcontrol/eviction/ordering" "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/flowcontrol/fairness/globalstrict" programaware "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/flowcontrol/fairness/program-aware" "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/flowcontrol/fairness/roundrobin" @@ -448,9 +452,12 @@ func (r *Runner) setup(ctx context.Context, cfg *rest.Config, opts *runserver.Op endpointCandidates := contracts.EndpointCandidates(requestcontrol.NewDatastoreEndpointCandidates(ds, requestcontrol.WithDisableEndpointSubsetFilter(opts.DisableEndpointSubsetFilter))) - endpointCandidates, admissionController, priorityBandControlPlane := r.initAdmissionControl(ctx, opts, eppConfig, endpointCandidates) + endpointCandidates, admissionController, priorityBandControlPlane, requestEvictor := r.initAdmissionControl(ctx, opts, eppConfig, endpointCandidates) director := requestcontrol.NewDirectorWithConfig(ds, scheduler, admissionController, endpointCandidates, r.requestControlConfig) + if requestEvictor != nil { + director.SetRequestEvictor(requestEvictor) + } serverRunner := &runserver.ExtProcServerRunner{ GrpcPort: opts.GRPCPort, @@ -471,6 +478,9 @@ func (r *Runner) setup(ctx context.Context, cfg *rest.Config, opts *runserver.Op GRPCMaxSendMsgSize: opts.GRPCMaxSendMsgSize, EnableGRPCStreamMetrics: opts.EnableGRPCStreamMetrics, } + if requestEvictor != nil { + serverRunner.EvictChannelLookup = requestEvictor.EvictionRegistry() + } if err := serverRunner.SetupWithManager(mgr); err != nil { setupLog.Error(err, "Failed to setup EPP controllers") @@ -876,28 +886,87 @@ func (r *Runner) initAdmissionControl( opts *runserver.Options, eppConfig *config.Config, endpointCandidates contracts.EndpointCandidates, -) (contracts.EndpointCandidates, requestcontrol.AdmissionController, contracts.PriorityBandControlPlane) { +) (contracts.EndpointCandidates, requestcontrol.AdmissionController, contracts.PriorityBandControlPlane, *fceviction.RequestEvictor) { if !r.featureGates[flowcontrol.FeatureGate] { setupLog.Info("Experimental Flow Control layer is disabled, using legacy admission control") return endpointCandidates, requestcontrol.NewLegacyAdmissionController(eppConfig.SaturationDetector, endpointCandidates), + nil, nil } endpointCandidates = requestcontrol.NewCachedEndpointCandidates(ctx, endpointCandidates, 50*time.Millisecond) setupLog.Info("Initializing experimental Flow Control layer") registry := fcregistry.NewFlowRegistry(eppConfig.FlowControlConfig.Registry, setupLog) - fc := fccontroller.NewFlowController( - ctx, - opts.PoolName, - eppConfig.FlowControlConfig.Controller, - fccontroller.Deps{ - Registry: registry, - SaturationDetector: eppConfig.SaturationDetector, - EndpointCandidates: endpointCandidates, - UsageLimitPolicy: eppConfig.FlowControlConfig.UsageLimitPolicy, - }, - ) - return endpointCandidates, requestcontrol.NewFlowControlAdmissionController(fc, opts.PoolName), registry + + deps := fccontroller.Deps{ + Registry: registry, + SaturationDetector: eppConfig.SaturationDetector, + EndpointCandidates: endpointCandidates, + UsageLimitPolicy: eppConfig.FlowControlConfig.UsageLimitPolicy, + } + + var requestEvictor *fceviction.RequestEvictor + if eppConfig.FlowControlConfig.Controller.EnableEviction { + var err error + requestEvictor, err = buildRequestEvictor() + if err != nil { + setupLog.Error(err, "Failed to build eviction plumbing; in-flight eviction disabled") + } else { + deps.InFlightEvictor = requestEvictor + grace := deriveEvictionConfirmationGrace(eppConfig.SaturationDetector, opts) + eppConfig.FlowControlConfig.Controller.EvictionConfirmationGrace = grace + setupLog.Info("In-flight eviction plumbing initialized", + "filter", evictfiltering.SheddableFilterType, + "ordering", evictordering.PriorityThenTimeOrderingType, + "confirmationGrace", grace) + } + } + + fc := fccontroller.NewFlowController(ctx, opts.PoolName, eppConfig.FlowControlConfig.Controller, deps) + return endpointCandidates, requestcontrol.NewFlowControlAdmissionController(fc, opts.PoolName), registry, requestEvictor +} + +const ( + // evictionEngineReclaimBudget covers the engine-side abort and KV garbage-collection time + // between a revoked request's stream termination and the freed capacity appearing in scraped + // metrics. + evictionEngineReclaimBudget = 250 * time.Millisecond + // evictionLeadingSensorGrace covers scheduling jitter for sensors whose gauge updates in the + // same event chain as the confirmation (a few dispatch ticks). + evictionLeadingSensorGrace = 10 * time.Millisecond +) + +// deriveEvictionConfirmationGrace computes the reclamation pacing grace from the paired saturation +// detector's confirmation-to-visibility lag. The grace is a property of the sensor, not a +// preference, so it is derived rather than configured (see docs/flow-control-eviction.md). +func deriveEvictionConfirmationGrace(detector fwkfc.SaturationDetector, opts *runserver.Options) time.Duration { + if detector != nil && detector.TypedName().Type == concurrency.ConcurrencyDetectorType { + // The concurrency detector's counters decrement when the stream terminates: the gauge leads + // physical reclamation, and only dispatch-loop jitter separates confirmation from visibility. + return evictionLeadingSensorGrace + } + // Scraped sensors (utilization detector, or any custom detector, conservatively): the freed + // capacity is visible only after the engine reclaims it and the next non-stale scrape lands. + return opts.RefreshMetricsInterval + opts.MetricsStalenessThreshold + evictionEngineReclaimBudget +} + +// buildRequestEvictor assembles the in-flight eviction plumbing: the victim filter and ordering +// policies, the ImmediateResponse eviction mechanism, and the tracking queue that ties them +// together. See docs/flow-control-eviction.md. +func buildRequestEvictor() (*fceviction.RequestEvictor, error) { + orderingPlugin, err := evictordering.PriorityThenTimeOrderingFactory(evictordering.PriorityThenTimeOrderingType, nil, nil) + if err != nil { + return nil, fmt.Errorf("failed to create eviction ordering policy: %w", err) + } + filterPlugin, err := evictfiltering.SheddableFilterFactory(evictfiltering.SheddableFilterType, nil, nil) + if err != nil { + return nil, fmt.Errorf("failed to create eviction filter policy: %w", err) + } + return fceviction.NewRequestEvictor( + orderingPlugin.(fwkfc.EvictionOrderingPolicy), + filterPlugin.(fwkfc.EvictionFilterPolicy), + fceviction.NewImmediateResponseEvictor(), + ), nil } // runWithFileDiscovery handles the execution path when a discovery plugin is configured. @@ -974,8 +1043,11 @@ func (r *Runner) runWithFileDiscovery(ctx context.Context, opts *runserver.Optio requestcontrol.WithDisableEndpointSubsetFilter(opts.DisableEndpointSubsetFilter))) // File-discovery mode has no InferenceObjective reconciler to drive the // control plane; static bands from config apply at registry construction. - endpointCandidates, admissionController, _ := r.initAdmissionControl(ctx, opts, eppConfig, endpointCandidates) + endpointCandidates, admissionController, _, requestEvictor := r.initAdmissionControl(ctx, opts, eppConfig, endpointCandidates) director := requestcontrol.NewDirectorWithConfig(ds, scheduler, admissionController, endpointCandidates, r.requestControlConfig) + if requestEvictor != nil { + director.SetRequestEvictor(requestEvictor) + } gknn := common.GKNN{ NamespacedName: types.NamespacedName{Name: poolName, Namespace: namespace}, @@ -998,6 +1070,9 @@ func (r *Runner) runWithFileDiscovery(ctx context.Context, opts *runserver.Optio GRPCMaxSendMsgSize: opts.GRPCMaxSendMsgSize, EnableGRPCStreamMetrics: opts.EnableGRPCStreamMetrics, } + if requestEvictor != nil { + serverRunner.EvictChannelLookup = requestEvictor.EvictionRegistry() + } r.customCollectors = append(r.customCollectors, collectors.NewInferencePoolMetricsCollector(ds)) metrics.Register(r.customCollectors...) diff --git a/pkg/epp/requestcontrol/director.go b/pkg/epp/requestcontrol/director.go index 4e33e980f0..28afef87eb 100644 --- a/pkg/epp/requestcontrol/director.go +++ b/pkg/epp/requestcontrol/director.go @@ -43,6 +43,7 @@ import ( "github.com/llm-d/llm-d-router/pkg/epp/datalayer" "github.com/llm-d/llm-d-router/pkg/epp/datastore" "github.com/llm-d/llm-d-router/pkg/epp/flowcontrol/contracts" + "github.com/llm-d/llm-d-router/pkg/epp/flowcontrol/eviction" fwkdl "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/datalayer" fwkrc "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/requestcontrol" fwkrh "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/requesthandling" @@ -210,6 +211,10 @@ type Director struct { // streaming response path. The request context key avoids coupling independent streams that reuse the same // x-request-id header. responseBodyQueues sync.Map + + // requestEvictor, when set, tracks dispatched requests for demand-driven in-flight eviction. + // See docs/flow-control-eviction.md. + requestEvictor *eviction.RequestEvictor } // getInferenceObjective fetches the inferenceObjective from the datastore otherwise creates a new one based on reqCtx. @@ -483,9 +488,19 @@ func (d *Director) prepareRequest(ctx context.Context, reqCtx *handlers.RequestC d.runPreRequestPlugins(ctx, reqCtx.SchedulingRequest, result) + if d.requestEvictor != nil { + d.requestEvictor.PreRequest(ctx, reqCtx.SchedulingRequest, result) + } + return reqCtx, nil } +// SetRequestEvictor wires the in-flight eviction tracker into the request lifecycle. +// Must be called before the Director serves traffic. +func (d *Director) SetRequestEvictor(re *eviction.RequestEvictor) { + d.requestEvictor = re +} + func (d *Director) toSchedulerEndpoints(endpoints []fwkdl.Endpoint) []fwksched.Endpoint { result := make([]fwksched.Endpoint, len(endpoints)) for i, endpoint := range endpoints { @@ -522,6 +537,16 @@ func (d *Director) HandleResponseHeader(ctx context.Context, reqCtx *handlers.Re func (d *Director) HandleResponseBody(ctx context.Context, reqCtx *handlers.RequestContext, endOfStream bool) *handlers.RequestContext { logger := log.FromContext(ctx).WithValues("stage", "bodyChunk") logger.V(logutil.TRACE).Info("Entering HandleResponseBodyChunk") + + // The eviction tracker must observe stream termination even when no streaming plugins are + // registered, so this runs before the early return below. + if endOfStream && d.requestEvictor != nil { + d.requestEvictor.ResponseBody(ctx, reqCtx.SchedulingRequest, &fwkrc.Response{ + RequestID: reqCtx.Request.Headers[reqcommon.RequestIDHeaderKey], + EndOfStream: true, + }, reqCtx.TargetPod) + } + if len(d.requestControlPlugins.responseStreamingPlugins) == 0 { logger.V(logutil.TRACE).Info("Exiting HandleResponseBodyChunk") return reqCtx diff --git a/pkg/epp/server/runserver.go b/pkg/epp/server/runserver.go index fdd5b84bf8..3f31f1e43f 100644 --- a/pkg/epp/server/runserver.go +++ b/pkg/epp/server/runserver.go @@ -66,6 +66,9 @@ type ExtProcServerRunner struct { GRPCMaxRecvMsgSize int GRPCMaxSendMsgSize int EnableGRPCStreamMetrics bool + // EvictChannelLookup, when set, enables the ext_proc server to terminate in-flight requests + // selected for eviction by flow control. See docs/flow-control-eviction.md. + EvictChannelLookup handlers.EvictChannelLookup } // NewDefaultExtProcServerRunner creates a runner with default values. @@ -214,6 +217,9 @@ func (r *ExtProcServerRunner) AsRunnable(logger logr.Logger) manager.Runnable { poolCap = 4 * 1024 * 1024 // gRPC default 4MB } extProcServer := handlers.NewStreamingServer(r.Datastore, r.Director, r.ParserRegistry, poolCap) + if r.EvictChannelLookup != nil { + extProcServer.SetEvictChannelLookup(r.EvictChannelLookup) + } extProcPb.RegisterExternalProcessorServer(srv, extProcServer) if r.HealthChecking { From 442627f7defa426769fdba445511ec590ae0a3a8 Mon Sep 17 00:00:00 2001 From: Luke Van Drie Date: Mon, 20 Jul 2026 16:52:36 -0700 Subject: [PATCH 4/6] chore(flowcontrol): attribute new files to the llm-d Authors Signed-off-by: Luke Van Drie --- pkg/epp/flowcontrol/controller/internal/reclamation.go | 2 +- pkg/epp/flowcontrol/controller/internal/reclamation_test.go | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/pkg/epp/flowcontrol/controller/internal/reclamation.go b/pkg/epp/flowcontrol/controller/internal/reclamation.go index 95a109d051..766b60945f 100644 --- a/pkg/epp/flowcontrol/controller/internal/reclamation.go +++ b/pkg/epp/flowcontrol/controller/internal/reclamation.go @@ -1,5 +1,5 @@ /* -Copyright 2026 The Kubernetes Authors. +Copyright 2026 The llm-d Authors. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. diff --git a/pkg/epp/flowcontrol/controller/internal/reclamation_test.go b/pkg/epp/flowcontrol/controller/internal/reclamation_test.go index af84582b36..cca8f94248 100644 --- a/pkg/epp/flowcontrol/controller/internal/reclamation_test.go +++ b/pkg/epp/flowcontrol/controller/internal/reclamation_test.go @@ -1,5 +1,5 @@ /* -Copyright 2026 The Kubernetes Authors. +Copyright 2026 The llm-d Authors. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. From 166b75847dd609378efbb05d17f9068d174c967a Mon Sep 17 00:00:00 2001 From: Luke Van Drie Date: Mon, 20 Jul 2026 17:11:51 -0700 Subject: [PATCH 5/6] feat(flowcontrol): label issued revocations by demand priority Operators tuning holdback and priorities need to see which band's demand drives destruction; the terminal-outcome counter stays unlabeled because it measures actuator health, not band economics. Adds an emission test covering the eviction metrics. Signed-off-by: Luke Van Drie --- .../controller/internal/reclamation.go | 3 +- pkg/epp/metrics/llm_d_router_metrics.go | 4 +-- pkg/epp/metrics/metrics.go | 7 +++-- pkg/epp/metrics/metrics_test.go | 30 +++++++++++++++++++ .../llm_d_flow_control_pending_reclaim_metric | 3 ++ .../llm_d_flow_control_reclaim_target_metric | 3 ++ ...m_d_flow_control_revocations_issued_metric | 3 ++ .../llm_d_flow_control_revocations_metric | 4 +++ 8 files changed, 51 insertions(+), 6 deletions(-) create mode 100644 pkg/epp/metrics/testdata/llm_d_flow_control_pending_reclaim_metric create mode 100644 pkg/epp/metrics/testdata/llm_d_flow_control_reclaim_target_metric create mode 100644 pkg/epp/metrics/testdata/llm_d_flow_control_revocations_issued_metric create mode 100644 pkg/epp/metrics/testdata/llm_d_flow_control_revocations_metric diff --git a/pkg/epp/flowcontrol/controller/internal/reclamation.go b/pkg/epp/flowcontrol/controller/internal/reclamation.go index 766b60945f..f2bb7ac0f2 100644 --- a/pkg/epp/flowcontrol/controller/internal/reclamation.go +++ b/pkg/epp/flowcontrol/controller/internal/reclamation.go @@ -19,6 +19,7 @@ package internal import ( "context" "math" + "strconv" "sync" "time" @@ -205,7 +206,7 @@ func (c *ReclamationController) Reclaim(ctx context.Context, saturation, ceiling c.outstanding[requestID] = revocation{credit: credit, issuedAt: now} c.pendingReclaim += credit } - metrics.RecordFlowControlRevocationsIssued(c.poolName, len(evicted)) + metrics.RecordFlowControlRevocationsIssued(c.poolName, strconv.Itoa(demandPriority), len(evicted)) metrics.RecordFlowControlPendingReclaim(c.poolName, c.pendingReclaim) c.logger.V(logutil.DEBUG).Info("Revocations issued", diff --git a/pkg/epp/metrics/llm_d_router_metrics.go b/pkg/epp/metrics/llm_d_router_metrics.go index eda5ac9829..eb231991fd 100644 --- a/pkg/epp/metrics/llm_d_router_metrics.go +++ b/pkg/epp/metrics/llm_d_router_metrics.go @@ -369,9 +369,9 @@ var ( prometheus.CounterOpts{ Subsystem: LLMDRouterEndpointPickerSubsystem, Name: "flow_control_revocations_issued_total", - Help: metricsutil.HelpMsgWithStability("Total number of in-flight eviction revocations issued.", compbasemetrics.ALPHA), + Help: metricsutil.HelpMsgWithStability("Total number of in-flight eviction revocations issued, labeled by the demand band's priority.", compbasemetrics.ALPHA), }, - []string{"inference_pool"}, + []string{"priority", "inference_pool"}, ) llmdFlowControlRevocationsTotal = prometheus.NewCounterVec( diff --git a/pkg/epp/metrics/metrics.go b/pkg/epp/metrics/metrics.go index 24d6058d02..20c23ceeb5 100644 --- a/pkg/epp/metrics/metrics.go +++ b/pkg/epp/metrics/metrics.go @@ -891,9 +891,10 @@ const ( RevocationOutcomeTimedOut = "timed_out" ) -// RecordFlowControlRevocationsIssued counts revocations at issue time. -func RecordFlowControlRevocationsIssued(inferencePool string, n int) { - llmdFlowControlRevocationsIssuedTotal.WithLabelValues(inferencePool).Add(float64(n)) +// RecordFlowControlRevocationsIssued counts revocations at issue time, labeled by the demand +// band's priority. +func RecordFlowControlRevocationsIssued(inferencePool, priority string, n int) { + llmdFlowControlRevocationsIssuedTotal.WithLabelValues(priority, inferencePool).Add(float64(n)) } // RecordFlowControlRevocations increments the revocation counter for a terminal outcome. diff --git a/pkg/epp/metrics/metrics_test.go b/pkg/epp/metrics/metrics_test.go index ca0c00a685..aded10f76f 100644 --- a/pkg/epp/metrics/metrics_test.go +++ b/pkg/epp/metrics/metrics_test.go @@ -1460,3 +1460,33 @@ func TestInferenceModelRewriteDecisionsTotalMetric(t *testing.T) { require.NoError(t, err) require.Equal(t, 1.0, valNew) } + +func TestFlowControlEvictionMetrics(t *testing.T) { + RecordFlowControlRevocationsIssued("pool-evict", "10", 2) + RecordFlowControlRevocations("pool-evict", RevocationOutcomeConfirmed, 1) + RecordFlowControlRevocations("pool-evict", RevocationOutcomeTimedOut, 1) + RecordFlowControlReclaimTarget("pool-evict", 0.1) + RecordFlowControlPendingReclaim("pool-evict", 0.05) + RecordFlowControlRevocationConfirmationDuration("pool-evict", 5*time.Millisecond) + + for name, testdata := range map[string]string{ + "llm_d_epp_flow_control_revocations_issued_total": "testdata/llm_d_flow_control_revocations_issued_metric", + "llm_d_epp_flow_control_revocations_total": "testdata/llm_d_flow_control_revocations_metric", + "llm_d_epp_flow_control_reclaim_target": "testdata/llm_d_flow_control_reclaim_target_metric", + "llm_d_epp_flow_control_pending_reclaim": "testdata/llm_d_flow_control_pending_reclaim_metric", + } { + want, err := os.Open(testdata) + if err != nil { + t.Fatal(err) + } + defer want.Close() + if err := promtestutil.GatherAndCompare(metrics.Registry, want, name); err != nil { + t.Error(err) + } + } + + if got := promtestutil.CollectAndCount(llmdFlowControlRevocationConfirmationDuration, + "llm_d_epp_flow_control_revocation_confirmation_seconds"); got != 1 { + t.Errorf("confirmation duration histogram series = %d, want 1", got) + } +} diff --git a/pkg/epp/metrics/testdata/llm_d_flow_control_pending_reclaim_metric b/pkg/epp/metrics/testdata/llm_d_flow_control_pending_reclaim_metric new file mode 100644 index 0000000000..ed87a92ca6 --- /dev/null +++ b/pkg/epp/metrics/testdata/llm_d_flow_control_pending_reclaim_metric @@ -0,0 +1,3 @@ +# HELP llm_d_epp_flow_control_pending_reclaim [ALPHA] Sum of outstanding and cooling pending-reclaim debits, in saturation-gauge units. +# TYPE llm_d_epp_flow_control_pending_reclaim gauge +llm_d_epp_flow_control_pending_reclaim{inference_pool="pool-evict"} 0.05 diff --git a/pkg/epp/metrics/testdata/llm_d_flow_control_reclaim_target_metric b/pkg/epp/metrics/testdata/llm_d_flow_control_reclaim_target_metric new file mode 100644 index 0000000000..92f75401f2 --- /dev/null +++ b/pkg/epp/metrics/testdata/llm_d_flow_control_reclaim_target_metric @@ -0,0 +1,3 @@ +# HELP llm_d_epp_flow_control_reclaim_target [ALPHA] Last computed reclamation deficit, in saturation-gauge units. +# TYPE llm_d_epp_flow_control_reclaim_target gauge +llm_d_epp_flow_control_reclaim_target{inference_pool="pool-evict"} 0.1 diff --git a/pkg/epp/metrics/testdata/llm_d_flow_control_revocations_issued_metric b/pkg/epp/metrics/testdata/llm_d_flow_control_revocations_issued_metric new file mode 100644 index 0000000000..1c32b2bcbf --- /dev/null +++ b/pkg/epp/metrics/testdata/llm_d_flow_control_revocations_issued_metric @@ -0,0 +1,3 @@ +# HELP llm_d_epp_flow_control_revocations_issued_total [ALPHA] Total number of in-flight eviction revocations issued, labeled by the demand band's priority. +# TYPE llm_d_epp_flow_control_revocations_issued_total counter +llm_d_epp_flow_control_revocations_issued_total{inference_pool="pool-evict",priority="10"} 2 diff --git a/pkg/epp/metrics/testdata/llm_d_flow_control_revocations_metric b/pkg/epp/metrics/testdata/llm_d_flow_control_revocations_metric new file mode 100644 index 0000000000..04ee53dc85 --- /dev/null +++ b/pkg/epp/metrics/testdata/llm_d_flow_control_revocations_metric @@ -0,0 +1,4 @@ +# HELP llm_d_epp_flow_control_revocations_total [ALPHA] Total number of in-flight eviction revocations by terminal outcome (confirmed, timed_out). Every issued revocation eventually increments exactly one outcome. +# TYPE llm_d_epp_flow_control_revocations_total counter +llm_d_epp_flow_control_revocations_total{inference_pool="pool-evict",outcome="confirmed"} 1 +llm_d_epp_flow_control_revocations_total{inference_pool="pool-evict",outcome="timed_out"} 1 From bf5b4e76ccbdd4183cdca17f71ad5fdc3d706e1b Mon Sep 17 00:00:00 2001 From: Luke Van Drie Date: Mon, 20 Jul 2026 17:13:48 -0700 Subject: [PATCH 6/6] test(flowcontrol): eviction dynamics benchmark Closed-loop scenario benchmark for the reclamation controller's pacing (BenchmarkEvictionDynamics): a synthetic pool with configurable sensor lag drives the real FlowController, registry, RequestEvictor, and ReclamationController end to end. Measures time-to-first-relief, HP wait percentiles, goodput, and over-eviction against an analytic minimum, across sensor profiles, grace values, per-decision caps, burst shapes, and eviction-off baselines. Opt-in via -bench, skipped under -short; EVICTION_BENCH_TABLE emits a markdown summary and EVICTION_BENCH_FULL runs the full matrix. Reproduces the Validation results in docs/flow-control-eviction.md (PR #2061). Signed-off-by: Luke Van Drie --- pkg/epp/flowcontrol/benchmark/benchmark.go | 27 +- pkg/epp/flowcontrol/benchmark/evictionsim.go | 399 ++++++++++++++++ .../flowcontrol/benchmark/evictionsim_test.go | 428 ++++++++++++++++++ 3 files changed, 844 insertions(+), 10 deletions(-) create mode 100644 pkg/epp/flowcontrol/benchmark/evictionsim.go create mode 100644 pkg/epp/flowcontrol/benchmark/evictionsim_test.go diff --git a/pkg/epp/flowcontrol/benchmark/benchmark.go b/pkg/epp/flowcontrol/benchmark/benchmark.go index a171bbf051..650e819c87 100644 --- a/pkg/epp/flowcontrol/benchmark/benchmark.go +++ b/pkg/epp/flowcontrol/benchmark/benchmark.go @@ -218,15 +218,9 @@ func setupRegistry( return reg } -// setupBenchmarkHarness creates the standard SUT environment. -func setupBenchmarkHarness( - ctx context.Context, - b *testing.B, - p priorityLevels, - limit egressConcurrencyLimit, - customDetector testDetector, - customCfg *controller.Config, -) (*controller.FlowController, testDetector) { +// buildPolicyDefaults constructs the standard fairness (global-strict) and ordering (FCFS) policy +// defaults used by all benchmark registries. +func buildPolicyDefaults(ctx context.Context, b *testing.B) registry.PriorityBandPolicyDefaults { b.Helper() handle := testutils.NewTestHandle(ctx) @@ -242,11 +236,24 @@ func setupBenchmarkHarness( } handle.AddPlugin(registry.DefaultOrderingPolicyRef, oPolicy) - defaults := registry.PriorityBandPolicyDefaults{ + return registry.PriorityBandPolicyDefaults{ OrderingPolicy: oPolicy.(flowcontrol.OrderingPolicy), FairnessPolicy: fPolicy.(flowcontrol.FairnessPolicy), } +} + +// setupBenchmarkHarness creates the standard SUT environment. +func setupBenchmarkHarness( + ctx context.Context, + b *testing.B, + p priorityLevels, + limit egressConcurrencyLimit, + customDetector testDetector, + customCfg *controller.Config, +) (*controller.FlowController, testDetector) { + b.Helper() + defaults := buildPolicyDefaults(ctx, b) reg := setupRegistry(b, defaults, p) detector := customDetector diff --git a/pkg/epp/flowcontrol/benchmark/evictionsim.go b/pkg/epp/flowcontrol/benchmark/evictionsim.go new file mode 100644 index 0000000000..9ebbfda4b0 --- /dev/null +++ b/pkg/epp/flowcontrol/benchmark/evictionsim.go @@ -0,0 +1,399 @@ +/* +Copyright 2026 The llm-d Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +// This file implements a closed-loop simulator for the eviction/reclamation dynamics benchmark +// (BenchmarkEvictionDynamics). It models an inference pool as a fixed number of capacity units +// occupied by sheddable leases, with configurable sensor lag between a lease's termination and +// the moment the saturation gauge reflects the freed capacity. All flow control components are +// real (FlowController, registry, RequestEvictor, ReclamationController); only the pool and its +// sensor are synthetic. See docs/flow-control-eviction.md for the pacing design under test. +package benchmark + +import ( + "context" + "fmt" + "math" + "sync" + "sync/atomic" + "testing" + "time" + + "k8s.io/apimachinery/pkg/types" + + reqcommon "github.com/llm-d/llm-d-router/pkg/common/request" + "github.com/llm-d/llm-d-router/pkg/epp/flowcontrol/contracts/mocks" + "github.com/llm-d/llm-d-router/pkg/epp/flowcontrol/controller" + "github.com/llm-d/llm-d-router/pkg/epp/flowcontrol/eviction" + fwkdl "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/datalayer" + "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/flowcontrol" + fwkrc "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/requestcontrol" + fwksched "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/scheduling" + evictfiltering "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/flowcontrol/eviction/filtering" + evictordering "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/flowcontrol/eviction/ordering" + "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/flowcontrol/usagelimits" +) + +// sensorProfile models where the saturation sensor sits relative to reality when a lease frees. +type sensorProfile struct { + name string + // termLatency is the delay between the eviction signal (EvictCh close) and the stream's + // termination (the confirmation event). + termLatency time.Duration + // freeVisibilityLag is the additional delay between termination and the freed capacity + // appearing in the gauge (engine GC + metrics scrape + staleness for a utilization-style + // sensor; ~0 for a concurrency-style sensor whose counters decrement at termination). + freeVisibilityLag time.Duration +} + +var ( + profileConcurrency = sensorProfile{name: "conc", termLatency: 3 * time.Millisecond, freeVisibilityLag: 0} + profileUtilization = sensorProfile{name: "util", termLatency: 3 * time.Millisecond, freeVisibilityLag: 200 * time.Millisecond} +) + +// burstShape describes the high-priority demand applied to the saturated pool. +type burstShape struct { + name string + hpCount int // total HP requests issued (rate-shaped when window > 0) + arrivalGap time.Duration // spacing between HP arrivals (0 = simultaneous burst) + // serviceTime is how long a dispatched HP request occupies its capacity unit before freeing it + // through the same lagged path as evictions. 0 = sticky: held until scenario end. + serviceTime time.Duration + // window, when non-zero, switches the scenario to timed mode: HP arrivals are generated for + // the window duration and goodput is measured over it, instead of waiting for an analytic + // admission count. + window time.Duration +} + +// evictionScenario is one coordinate of the benchmark matrix. +type evictionScenario struct { + profile sensorProfile + grace time.Duration + maxRevoc int + burst burstShape + // evictionOff disables reclamation for baseline arms. + evictionOff bool + // churnMean, when non-zero, gives sheddable leases a natural completion time drawn + // deterministically from a truncated exponential-like ladder with this mean. + churnMean time.Duration + capacity int // C, pool capacity in lease units + preload int // S, sheddable leases occupying the pool at burst start + hpCeiling float64 // h, the HP band's dispatch ceiling +} + +func (sc evictionScenario) name() string { + mode := fmt.Sprintf("g%d/k%d", sc.grace.Milliseconds(), sc.maxRevoc) + if sc.evictionOff { + if sc.churnMean > 0 { + mode = "off-churn" + } else { + mode = "off-nochurn" + } + } + return fmt.Sprintf("%s/%s/%s", sc.profile.name, mode, sc.burst.name) +} + +// maxUnitsBelowCeiling returns the largest integer unit count strictly below h*C, i.e. the highest +// occupancy at which the HP band can still dispatch. +func maxUnitsBelowCeiling(capacity int, ceiling float64) int { + return int(math.Ceil(ceiling*float64(capacity))) - 1 +} + +// minEvictionsRequired computes the analytic minimum number of sheddable evictions needed for a +// sticky HP burst: preloaded S leases in a pool of capacity C, K HP arrivals, HP ceiling h. The +// k-th admission requires occupancy <= maxUnitsBelowCeiling just before it; admitted HP requests +// are sticky (never free). Returns (minEvictions, admittableHP). +func minEvictionsRequired(preload, hpCount, capacity int, ceiling float64) (minEvictions, admittable int) { + maxBelow := maxUnitsBelowCeiling(capacity, ceiling) + if maxBelow < 0 { + return 0, 0 // Ceiling of zero admits nothing. + } + // With every sheddable lease evicted, the k-th admission needs k-1 <= maxBelow. + admittable = min(hpCount, maxBelow+1) + if admittable == 0 { + return 0, 0 + } + // The last admission needs preload - E + (admittable-1) <= maxBelow. + minEvictions = max(0, preload+admittable-1-maxBelow) + minEvictions = min(minEvictions, preload) + return minEvictions, admittable +} + +// syntheticPool is the synthetic pool + sensor. visibleUnits is the gauge numerator: sheddable +// leases whose free is not yet sensor-visible, plus dispatched HP footprint. +type syntheticPool struct { + // Embedded to satisfy the plugin.Plugin surface of SaturationDetector (same pattern as + // benchDetector); only Saturation is ever called. + flowcontrol.SaturationDetector + + capacity int64 + profile sensorProfile + evictor *eviction.RequestEvictor + epMeta *fwkdl.EndpointMetadata + + visibleUnits atomic.Int64 + maxObservedUnits atomic.Int64 + + // hpWaiting counts HP requests currently blocked in EnqueueAndWait; sampled at eviction-signal + // time to attribute evictions fired with no waiting demand. + hpWaiting atomic.Int64 + maxHPWaiting atomic.Int64 + + evictionsSignaled atomic.Int64 + evictionsConfirmed atomic.Int64 + evictionsWhileIdle atomic.Int64 + + leaseCtx context.Context + leaseCancel context.CancelFunc + wg sync.WaitGroup +} + +var _ flowcontrol.SaturationDetector = (*syntheticPool)(nil) + +func newSyntheticPool(sc evictionScenario, evictor *eviction.RequestEvictor) *syntheticPool { + ctx, cancel := context.WithCancel(context.Background()) + return &syntheticPool{ + capacity: int64(sc.capacity), + profile: sc.profile, + evictor: evictor, + epMeta: &fwkdl.EndpointMetadata{ + NamespacedName: types.NamespacedName{Name: "sim-pod", Namespace: "default"}, + Address: "10.0.0.1", + Port: "8000", + }, + leaseCtx: ctx, + leaseCancel: cancel, + } +} + +func (p *syntheticPool) Saturation(context.Context, []fwkdl.Endpoint) float64 { + return float64(p.visibleUnits.Load()) / float64(p.capacity) +} + +// addUnits adjusts the gauge and maintains the overshoot watermark. +func (p *syntheticPool) addUnits(delta int64) { + v := p.visibleUnits.Add(delta) + for { + cur := p.maxObservedUnits.Load() + if v <= cur || p.maxObservedUnits.CompareAndSwap(cur, v) { + return + } + } +} + +// PreloadSheddable occupies the pool with n sheddable leases tracked in the real RequestEvictor, +// each with its own lifecycle goroutine reacting to eviction (or natural churn for baselines). +// churnAfter, when non-nil, returns the lease's natural completion delay (0 = never completes). +func (p *syntheticPool) PreloadSheddable(n int, churnAfter func(i int) time.Duration) { + for i := range n { + id := fmt.Sprintf("shed-%d", i) + req := &fwksched.InferenceRequest{ + RequestID: id, + Headers: map[string]string{reqcommon.RequestIDHeaderKey: id}, + Objectives: fwksched.RequestObjectives{Priority: -1}, + } + result := &fwksched.SchedulingResult{ + PrimaryProfileName: "decode", + ProfileResults: map[string]*fwksched.ProfileRunResult{ + "decode": {TargetEndpoints: []fwksched.Endpoint{ + fwksched.NewEndpoint(p.epMeta, fwkdl.NewMetrics(), nil), + }}, + }, + } + p.evictor.PreRequest(p.leaseCtx, req, result) + evictCh := p.evictor.EvictionRegistry().Get(id) + p.addUnits(1) + + var churn time.Duration + if churnAfter != nil { + churn = churnAfter(i) + } + p.wg.Add(1) + go p.leaseLifecycle(req, evictCh, churn) + } +} + +// leaseLifecycle waits for eviction, natural churn, or teardown, and walks the freed capacity +// through the sensor lag stages. +func (p *syntheticPool) leaseLifecycle(req *fwksched.InferenceRequest, evictCh chan struct{}, churnAfter time.Duration) { + defer p.wg.Done() + + var churnCh <-chan time.Time + if churnAfter > 0 { + timer := time.NewTimer(churnAfter) + defer timer.Stop() + churnCh = timer.C + } + + select { + case <-evictCh: + p.evictionsSignaled.Add(1) + if p.hpWaiting.Load() == 0 { + p.evictionsWhileIdle.Add(1) + } + if !sleepCtx(p.leaseCtx, p.profile.termLatency) { + return + } + // Stream termination: the real cleanup path, which fires the confirmation listener. + p.evictor.ResponseBody(p.leaseCtx, req, &fwkrc.Response{EndOfStream: true}, p.epMeta) + p.evictionsConfirmed.Add(1) + case <-churnCh: + // Natural completion frees capacity through the same termination path. + p.evictor.ResponseBody(p.leaseCtx, req, &fwkrc.Response{EndOfStream: true}, p.epMeta) + case <-p.leaseCtx.Done(): + return + } + + if !sleepCtx(p.leaseCtx, p.profile.freeVisibilityLag) { + return + } + p.addUnits(-1) +} + +// HPEnqueued and HPFinishedWaiting bracket a client's EnqueueAndWait call. +func (p *syntheticPool) HPEnqueued() { + w := p.hpWaiting.Add(1) + for { + cur := p.maxHPWaiting.Load() + if w <= cur || p.maxHPWaiting.CompareAndSwap(cur, w) { + return + } + } +} + +func (p *syntheticPool) HPFinishedWaiting() { p.hpWaiting.Add(-1) } + +// HPDispatched claims the dispatched request's footprint immediately (the sensor-lag dimension +// applies to frees only, which is conservative for the pacing question under test). If +// serviceTime > 0 the footprint is released through the lagged path after service completes. +func (p *syntheticPool) HPDispatched(serviceTime time.Duration) { + p.addUnits(1) + if serviceTime <= 0 { + return // Sticky: held until scenario teardown. + } + p.wg.Add(1) + go func() { + defer p.wg.Done() + if !sleepCtx(p.leaseCtx, serviceTime) { + return + } + if !sleepCtx(p.leaseCtx, p.profile.freeVisibilityLag) { + return + } + p.addUnits(-1) + }() +} + +// Close tears the pool down and fails the benchmark if lease goroutines leak. +func (p *syntheticPool) Close(b *testing.B) { + b.Helper() + p.leaseCancel() + done := make(chan struct{}) + go func() { + p.wg.Wait() + close(done) + }() + select { + case <-done: + case <-time.After(5 * time.Second): + b.Fatal("syntheticPool teardown timed out: lease goroutines leaked") + } +} + +// sleepCtx sleeps for d unless ctx is cancelled first; returns false on cancellation. A zero or +// negative d returns true immediately. +func sleepCtx(ctx context.Context, d time.Duration) bool { + if d <= 0 { + return ctx.Err() == nil + } + timer := time.NewTimer(d) + defer timer.Stop() + select { + case <-timer.C: + return true + case <-ctx.Done(): + return false + } +} + +// hpBenchRequest is a FlowControlRequest for the high-priority flow with a unique ID. +type hpBenchRequest struct { + id string + key flowcontrol.FlowKey +} + +func (r *hpBenchRequest) FlowKey() flowcontrol.FlowKey { return r.key } +func (r *hpBenchRequest) ByteSize() uint64 { return 1024 } +func (r *hpBenchRequest) InitialEffectiveTTL() time.Duration { return 5 * time.Minute } +func (r *hpBenchRequest) ID() string { return r.id } +func (r *hpBenchRequest) GetMetadata() map[string]any { return nil } +func (r *hpBenchRequest) InferencePoolName() string { return "bench-pool" } +func (r *hpBenchRequest) ModelName() string { return "bench-model" } +func (r *hpBenchRequest) TargetModelName() string { return "bench-target" } +func (r *hpBenchRequest) InferenceRequest() *fwksched.InferenceRequest { return nil } +func (r *hpBenchRequest) ReceivedTimestamp() time.Time { return time.Now() } + +// setupEvictionBenchHarness builds the real SUT stack for one scenario: registry with a single +// priority-0 band, constant HP ceiling, the synthetic pool as the saturation detector, and the +// full eviction plumbing wired through controller Deps. +func setupEvictionBenchHarness( + ctx context.Context, + b *testing.B, + sc evictionScenario, +) (*controller.FlowController, *syntheticPool) { + b.Helper() + + defaults := buildPolicyDefaults(ctx, b) + reg := setupRegistry(b, defaults, 1) // Single band at priority 0. + + orderingPlugin, err := evictordering.PriorityThenTimeOrderingFactory(evictordering.PriorityThenTimeOrderingType, nil, nil) + if err != nil { + b.Fatalf("Failed to create eviction ordering policy: %v", err) + } + filterPlugin, err := evictfiltering.SheddableFilterFactory(evictfiltering.SheddableFilterType, nil, nil) + if err != nil { + b.Fatalf("Failed to create eviction filter policy: %v", err) + } + requestEvictor := eviction.NewRequestEvictor( + orderingPlugin.(flowcontrol.EvictionOrderingPolicy), + filterPlugin.(flowcontrol.EvictionFilterPolicy), + eviction.NewImmediateResponseEvictor(), + ) + + pool := newSyntheticPool(sc, requestEvictor) + + cfg := &controller.Config{ + DefaultRequestTTL: 5 * time.Minute, + ExpiryCleanupInterval: 1 * time.Hour, // Effectively disabled. + EnqueueChannelBufferSize: 2000, + EnableEviction: !sc.evictionOff, + MaxRevocationsPerDecision: sc.maxRevoc, + EvictionConfirmationGrace: sc.grace, + EvictionConfirmationTimeout: 10 * time.Second, + } + + deps := controller.Deps{ + Registry: reg, + SaturationDetector: pool, + EndpointCandidates: &mocks.MockEndpointCandidates{}, + UsageLimitPolicy: usagelimits.NewConstPolicy("evict-bench", sc.hpCeiling), + } + if !sc.evictionOff { + deps.InFlightEvictor = requestEvictor + } + + fc := controller.NewFlowController(ctx, "eviction-bench", cfg, deps) + return fc, pool +} diff --git a/pkg/epp/flowcontrol/benchmark/evictionsim_test.go b/pkg/epp/flowcontrol/benchmark/evictionsim_test.go new file mode 100644 index 0000000000..d4690ceed3 --- /dev/null +++ b/pkg/epp/flowcontrol/benchmark/evictionsim_test.go @@ -0,0 +1,428 @@ +/* +Copyright 2026 The llm-d Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package benchmark + +import ( + "context" + "fmt" + "math" + "os" + "sort" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/llm-d/llm-d-router/pkg/epp/flowcontrol/types" + "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/flowcontrol" +) + +func TestMinEvictionsRequired(t *testing.T) { + t.Parallel() + testCases := []struct { + name string + preload, hp, cap int + ceiling float64 + wantMin, wantAdmits int + }{ + // C=20, h=0.9: maxBelow = 17 (occupancy must be <= 17 to admit). + {"FullPool_SingleHP", 20, 1, 20, 0.9, 3, 1}, + {"FullPool_Batch32", 20, 32, 20, 0.9, 20, 18}, + {"FullPool_Batch10", 20, 10, 20, 0.9, 12, 10}, + {"HalfPool_SingleHP", 10, 1, 20, 0.9, 0, 1}, + {"IntegralCeiling", 18, 1, 18, 1.0, 1, 1}, // maxBelow = 17; need 18-E <= 17 -> E=1. + {"FractionalCeiling", 20, 1, 20, 0.95, 2, 1}, + {"ZeroCeiling", 20, 5, 20, 0.0, 0, 0}, + {"EmptyPool", 0, 5, 20, 0.9, 0, 5}, + } + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + gotMin, gotAdmits := minEvictionsRequired(tc.preload, tc.hp, tc.cap, tc.ceiling) + if gotMin != tc.wantMin || gotAdmits != tc.wantAdmits { + t.Fatalf("minEvictionsRequired(%d,%d,%d,%v) = (%d,%d), want (%d,%d)", + tc.preload, tc.hp, tc.cap, tc.ceiling, gotMin, gotAdmits, tc.wantMin, tc.wantAdmits) + } + }) + } +} + +// scenarioResult holds one scenario run's measurements. +type scenarioResult struct { + ttfr time.Duration // burst start -> first HP dispatch (0 = none dispatched) + p50, p95 time.Duration + admitted int64 + elapsed time.Duration + evictSignaled int64 + evictConfirmed int64 + evictHPIdle int64 + overEvict int64 // confirmed - analytic minimum (sticky scenarios only; else -1) + maxHPWaiting int64 + overshoot int64 + timedOut bool +} + +const scenarioTimeout = 8 * time.Second + +// runScenario executes one full scenario against a fresh harness and returns its measurements. +func runScenario(b *testing.B, sc evictionScenario) scenarioResult { + b.Helper() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + fc, pool := setupEvictionBenchHarness(ctx, b, sc) + + var churnFn func(i int) time.Duration + if sc.churnMean > 0 { + n := sc.preload + mean := sc.churnMean + // Deterministic exponential quantile ladder so churn is identical across runs. + churnFn = func(i int) time.Duration { + u := (float64(i) + 0.5) / float64(n) + return time.Duration(-float64(mean) * math.Log(1.0-u)) + } + } + pool.PreloadSheddable(sc.preload, churnFn) + + minEvict, admittable := minEvictionsRequired(sc.preload, sc.burst.hpCount, sc.capacity, sc.hpCeiling) + timed := sc.burst.window > 0 + + var admitted atomic.Int64 + var ttfrNanos atomic.Int64 + var waitsMu sync.Mutex + waits := make([]time.Duration, 0, sc.burst.hpCount) + + clientCtx, clientCancel := context.WithCancel(ctx) + defer clientCancel() + var clientWG sync.WaitGroup + burstStart := time.Now() + + launch := func(i int) { + clientWG.Add(1) + go func() { + defer clientWG.Done() + req := &hpBenchRequest{id: fmt.Sprintf("hp-%d", i), key: flowcontrol.FlowKey{ID: "hp-flow", Priority: 0}} + pool.HPEnqueued() + t0 := time.Now() + outcome, err := fc.EnqueueAndWait(clientCtx, req) + wait := time.Since(t0) + pool.HPFinishedWaiting() + if err != nil || outcome != types.QueueOutcomeDispatched { + return + } + pool.HPDispatched(sc.burst.serviceTime) + ttfrNanos.CompareAndSwap(0, time.Since(burstStart).Nanoseconds()) + admitted.Add(1) + waitsMu.Lock() + waits = append(waits, wait) + waitsMu.Unlock() + }() + } + + timedOut := false + if timed { + gap := sc.burst.window / time.Duration(sc.burst.hpCount) + ticker := time.NewTicker(gap) + windowEnd := time.NewTimer(sc.burst.window) + i := 0 + genLoop: + for { + select { + case <-ticker.C: + if i < sc.burst.hpCount { + launch(i) + i++ + } + case <-windowEnd.C: + break genLoop + } + } + ticker.Stop() + } else { + for i := range sc.burst.hpCount { + launch(i) + if sc.burst.arrivalGap > 0 { + time.Sleep(sc.burst.arrivalGap) + } + } + // Baselines with no churn admit nothing; observe a fixed window instead of the full timeout. + waitBudget := scenarioTimeout + target := int64(admittable) + if sc.evictionOff && sc.churnMean == 0 { + waitBudget = 2 * time.Second + target = math.MaxInt64 + } + deadline := time.Now().Add(waitBudget) + for admitted.Load() < target { + if time.Now().After(deadline) { + timedOut = target != math.MaxInt64 + break + } + time.Sleep(time.Millisecond) + } + } + elapsed := time.Since(burstStart) + + // Unblock any still-queued HP clients and collect. + clientCancel() + clientWG.Wait() + + sort.Slice(waits, func(i, j int) bool { return waits[i] < waits[j] }) + percentile := func(p float64) time.Duration { + if len(waits) == 0 { + return 0 + } + idx := int(p * float64(len(waits)-1)) + return waits[idx] + } + + res := scenarioResult{ + ttfr: time.Duration(ttfrNanos.Load()), + p50: percentile(0.50), + p95: percentile(0.95), + admitted: admitted.Load(), + elapsed: elapsed, + evictSignaled: pool.evictionsSignaled.Load(), + evictConfirmed: pool.evictionsConfirmed.Load(), + evictHPIdle: pool.evictionsWhileIdle.Load(), + overEvict: -1, + maxHPWaiting: pool.maxHPWaiting.Load(), + // Overshoot beyond the higher of the preload watermark and the ceiling: any rise above it + // means dispatch outran the gauge. + overshoot: max(0, pool.maxObservedUnits.Load()- + max(int64(sc.preload), int64(math.Ceil(sc.hpCeiling*float64(sc.capacity))))), + timedOut: timedOut, + } + if !timed && !sc.evictionOff { + res.overEvict = res.evictConfirmed - int64(minEvict) + } + + // Teardown: stop the controller, then the pool (leases, HP release timers). + cancel() + pool.Close(b) + time.Sleep(50 * time.Millisecond) // Drain processor goroutines (package precedent). + return res +} + +// --- Aggregation over b.N iterations --- + +type scenarioAgg struct { + n int + sum scenarioResult + anyTO bool + sumTTFR time.Duration +} + +func (a *scenarioAgg) add(r scenarioResult) { + a.n++ + a.sumTTFR += r.ttfr + a.sum.p50 += r.p50 + a.sum.p95 += r.p95 + a.sum.admitted += r.admitted + a.sum.elapsed += r.elapsed + a.sum.evictSignaled += r.evictSignaled + a.sum.evictConfirmed += r.evictConfirmed + a.sum.evictHPIdle += r.evictHPIdle + a.sum.overEvict += r.overEvict + a.sum.maxHPWaiting += r.maxHPWaiting + a.sum.overshoot += r.overshoot + a.anyTO = a.anyTO || r.timedOut +} + +// evictionRow is one line of the shareable results table. +type evictionRow struct { + name string + ttfrMS, p50MS, p95MS, goodput float64 + evictConfirmed, overEvict, evictHPIdle, maxQueue float64 + timedOut bool +} + +var ( + evictionTableMu sync.Mutex + evictionTable []evictionRow +) + +func (a *scenarioAgg) report(b *testing.B, name string) { + n := float64(a.n) + ms := func(d time.Duration) float64 { return math.Round(float64(d)/float64(time.Millisecond)/n*10) / 10 } + row := evictionRow{ + name: name, + ttfrMS: ms(a.sumTTFR), + p50MS: ms(a.sum.p50), + p95MS: ms(a.sum.p95), + goodput: math.Round(float64(a.sum.admitted)/a.sum.elapsed.Seconds()*10) / 10, + evictConfirmed: float64(a.sum.evictConfirmed) / n, + overEvict: float64(a.sum.overEvict) / n, + evictHPIdle: float64(a.sum.evictHPIdle) / n, + maxQueue: float64(a.sum.maxHPWaiting) / n, + timedOut: a.anyTO, + } + + b.ReportMetric(row.ttfrMS, "ttfr-ms") + b.ReportMetric(row.p50MS, "p50-ms") + b.ReportMetric(row.p95MS, "p95-ms") + b.ReportMetric(row.goodput, "goodput-rps") + b.ReportMetric(row.evictConfirmed, "evict-confirmed") + if row.overEvict >= 0 { + b.ReportMetric(row.overEvict, "over-evict") + } + b.ReportMetric(row.evictHPIdle, "evict-hp-idle") + b.ReportMetric(float64(a.sum.overshoot)/n, "overshoot") + if a.anyTO { + b.ReportMetric(1, "timed-out") + } + + evictionTableMu.Lock() + evictionTable = append(evictionTable, row) + evictionTableMu.Unlock() +} + +// --- The matrix --- + +var ( + burstSingle = burstShape{name: "single", hpCount: 1} + burstBatch32 = burstShape{name: "batch32", hpCount: 32} + burstSustained = burstShape{name: "sustained", hpCount: 300, serviceTime: 250 * time.Millisecond, window: 3 * time.Second} +) + +// matchedGrace is the grace a deployment would pair with each sensor profile per the design doc. +func matchedGrace(p sensorProfile) time.Duration { + if p.freeVisibilityLag == 0 { + return 50 * time.Millisecond + } + return 500 * time.Millisecond +} + +func evictionScenarios(full bool) []evictionScenario { + base := evictionScenario{capacity: 20, preload: 20, hpCeiling: 0.9, maxRevoc: 2} + profiles := []sensorProfile{profileConcurrency, profileUtilization} + graces := []time.Duration{5 * time.Millisecond, 50 * time.Millisecond, 200 * time.Millisecond, 500 * time.Millisecond} + ks := []int{1, 2, 5} + bursts := []burstShape{burstSingle, burstBatch32, burstSustained} + + var out []evictionScenario + seen := map[string]bool{} + add := func(sc evictionScenario) { + if name := sc.name(); !seen[name] { + seen[name] = true + out = append(out, sc) + } + } + + for _, p := range profiles { + if full { + for _, g := range graces { + for _, k := range ks { + for _, bu := range bursts { + sc := base + sc.profile, sc.grace, sc.maxRevoc, sc.burst = p, g, k, bu + add(sc) + } + } + } + } else { + // Informative diagonal: grace sweep at k=2/batch32; k sweep at matched grace/batch32; + // single + sustained at matched settings. + for _, g := range graces { + sc := base + sc.profile, sc.grace, sc.burst = p, g, burstBatch32 + add(sc) + } + for _, k := range ks { + sc := base + sc.profile, sc.grace, sc.maxRevoc, sc.burst = p, matchedGrace(p), k, burstBatch32 + add(sc) + } + for _, bu := range []burstShape{burstSingle, burstSustained} { + sc := base + sc.profile, sc.grace, sc.burst = p, matchedGrace(p), bu + add(sc) + } + // Over-eviction exposure: low grace on shallow demand. Against the lagged sensor, debits + // expire before frees become gauge-visible (expect over-eviction); against the leading + // sensor, near-zero grace should stay clean (expect none). + lowGrace := 5 * time.Millisecond + if p.freeVisibilityLag > 0 { + lowGrace = 50 * time.Millisecond + } + { + sc := base + sc.profile, sc.grace, sc.burst = p, lowGrace, burstSingle + add(sc) + } + } + // Baselines: eviction off, with and without natural churn. + for _, churn := range []time.Duration{0, 500 * time.Millisecond} { + sc := base + sc.profile, sc.grace, sc.burst = p, matchedGrace(p), burstBatch32 + sc.evictionOff, sc.churnMean = true, churn + add(sc) + } + } + return out +} + +// BenchmarkEvictionDynamics measures the reclamation controller's pacing behavior end-to-end. +// +// Invocation: +// +// go test ./pkg/epp/flowcontrol/benchmark/ -run '^$' -bench EvictionDynamics -benchtime=1x -count=5 +// +// Set EVICTION_BENCH_FULL=1 for the full matrix and EVICTION_BENCH_TABLE=1 to print a markdown +// summary table after the run. +func BenchmarkEvictionDynamics(b *testing.B) { + if testing.Short() { + b.Skip("eviction dynamics benchmark skipped in -short mode") + } + + for _, sc := range evictionScenarios(os.Getenv("EVICTION_BENCH_FULL") != "") { + b.Run(sc.name(), func(b *testing.B) { + agg := &scenarioAgg{} + for range b.N { + agg.add(runScenario(b, sc)) + } + agg.report(b, sc.name()) + }) + } + + if os.Getenv("EVICTION_BENCH_TABLE") != "" { + printEvictionTable() + } +} + +func printEvictionTable() { + evictionTableMu.Lock() + defer evictionTableMu.Unlock() + + var sb strings.Builder + sb.WriteString("\n| scenario | ttfr(ms) | p50(ms) | p95(ms) | goodput(r/s) | evicted | over-evict | evict-hp-idle | maxq |\n") + sb.WriteString("|---|---|---|---|---|---|---|---|---|\n") + for _, r := range evictionTable { + over := "n/a" + if r.overEvict >= 0 { + over = fmt.Sprintf("%.1f", r.overEvict) + } + name := r.name + if r.timedOut { + name += " (TIMEOUT)" + } + fmt.Fprintf(&sb, "| %s | %.1f | %.1f | %.1f | %.1f | %.1f | %s | %.1f | %.0f |\n", + name, r.ttfrMS, r.p50MS, r.p95MS, r.goodput, r.evictConfirmed, over, r.evictHPIdle, r.maxQueue) + } + fmt.Println(sb.String()) +}