feat(flowcontrol): demand-driven in-flight eviction

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, ", ") + "}"
 }
 

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...)

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,20 +101,26 @@ 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...)
 }
 
 // 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
 }