diff --git a/pkg/epp/framework/plugins/flowcontrol/fairness/program-aware/README.md b/pkg/epp/framework/plugins/flowcontrol/fairness/program-aware/README.md index e1fb65b4fa..66c5d33fd0 100644 --- a/pkg/epp/framework/plugins/flowcontrol/fairness/program-aware/README.md +++ b/pkg/epp/framework/plugins/flowcontrol/fairness/program-aware/README.md @@ -17,7 +17,9 @@ Choose this policy when: * **Identifies programs** via the fairness ID header carried on each request. * **Tracks per-program metrics** through the request lifecycle: queue wait time, dispatched count, in-flight count, last completion time, and (LAS) attained service in weighted tokens. -* **Selects a queue to dispatch from** using the configured strategy. Currently `las` (Least Attained Service) is supported; programs with the lowest accumulated service score highest. +* **Selects a queue to dispatch from** using the configured strategy. Two are supported: + * `las` (Least Attained Service): programs with the lowest accumulated service score highest. + * `turn-priority`: scoring engages only under contention. With no flow waiting nothing is dispatched; with exactly one waiting flow it is dispatched directly. When two or more flows are waiting, each scores by its head request's `prio = turn_number + turnPriorityTimeWeight * time_elapsed_in_seconds`, highest first. The elapsed term ages waiting flows up so lower-turn flows are not starved. Turn number is the program's dispatched-request count plus the waiting request itself: each program (fairness ID) has its own counter, advanced once per dispatch. The counter resets when idle program state is evicted: a program idle past the eviction TTL has likely lost its KV cache, so on return it restarts from turn one and re-earns priority from a cold state. * **Decays attained service** for inactive programs so a long-idle program is not penalized indefinitely. Both wall-clock half-life and per-Pick factor decay are supported. * **Evicts idle program state** on a periodic sweep so per-program memory and Prometheus label series do not accumulate forever. @@ -51,11 +53,12 @@ flowControl: | Field | Default | Description | |---|---|---| -| `strategy` | `las` | Scoring strategy. Only `las` is supported. | +| `strategy` | `las` | Scoring strategy: `las` or `turn-priority`. | | `lasWeightService` | `0.8` | Weight on the inverted attained-service signal. Higher values prioritize underserved programs more aggressively. | | `lasWeightHeadWait` | `0.2` | Weight on the head-of-queue age. Acts as a tiebreaker on cold start when programs have equal attained service. | | `lasDecayFactor` | `0.99997` | Per-Pick decay factor applied to inactive programs when `lasHalfLifeSeconds` is `0`. Must be in `(0, 1]`. Coupled to Pick rate. | | `lasHalfLifeSeconds` | `0` | Wall-clock half-life of attained service for inactive programs. When `> 0` it overrides `lasDecayFactor`. | +| `turnPriorityTimeWeight` | `0.5` | (`turn-priority`) Weight on the elapsed-seconds term of the head-request priority score. | | `evictionTtlSeconds` | `3600` | A program with no completion in this window is evicted from the metrics map. | | `evictionSweepSeconds` | `300` | How often the eviction sweep runs. Must be `> 0`. | diff --git a/pkg/epp/framework/plugins/flowcontrol/fairness/program-aware/plugin.go b/pkg/epp/framework/plugins/flowcontrol/fairness/program-aware/plugin.go index 0312af4001..dca3c7a4f3 100644 --- a/pkg/epp/framework/plugins/flowcontrol/fairness/program-aware/plugin.go +++ b/pkg/epp/framework/plugins/flowcontrol/fairness/program-aware/plugin.go @@ -32,6 +32,8 @@ type Config struct { LASWeightHeadWait float64 `json:"lasWeightHeadWait,omitempty"` LASDecayFactor float64 `json:"lasDecayFactor,omitempty"` LASHalfLifeSeconds float64 `json:"lasHalfLifeSeconds,omitempty"` + + TurnPriorityTimeWeight float64 `json:"turnPriorityTimeWeight,omitempty"` } func DefaultConfig() Config { @@ -43,6 +45,8 @@ func DefaultConfig() Config { LASWeightHeadWait: 0.2, LASDecayFactor: 0.99997, LASHalfLifeSeconds: 0, + + TurnPriorityTimeWeight: 0.5, } } @@ -65,6 +69,9 @@ func (c Config) validate() error { if c.LASHalfLifeSeconds < 0 { return fmt.Errorf("lasHalfLifeSeconds must be >= 0, got %v", c.LASHalfLifeSeconds) } + if c.TurnPriorityTimeWeight < 0 { + return fmt.Errorf("turnPriorityTimeWeight must be >= 0, got %v", c.TurnPriorityTimeWeight) + } return nil } diff --git a/pkg/epp/framework/plugins/flowcontrol/fairness/program-aware/plugin_test.go b/pkg/epp/framework/plugins/flowcontrol/fairness/program-aware/plugin_test.go index fc972b565f..59e31952c8 100644 --- a/pkg/epp/framework/plugins/flowcontrol/fairness/program-aware/plugin_test.go +++ b/pkg/epp/framework/plugins/flowcontrol/fairness/program-aware/plugin_test.go @@ -32,6 +32,13 @@ func TestFactory_LASConfig(t *testing.T) { require.NotNil(t, p) } +func TestFactory_TurnPriorityConfig(t *testing.T) { + cfg := `{"strategy":"turn-priority","turnPriorityTimeWeight":0.5}` + p, err := ProgramAwarePluginFactory("test", decoder(cfg), nil) + require.NoError(t, err) + require.NotNil(t, p) +} + func TestFactory_UnknownStrategy(t *testing.T) { _, err := ProgramAwarePluginFactory("test", decoder(`{"strategy":"wfq"}`), nil) require.Error(t, err) @@ -39,12 +46,13 @@ func TestFactory_UnknownStrategy(t *testing.T) { func TestFactory_InvalidConfig(t *testing.T) { cases := map[string]string{ - "negative ttl": `{"evictionTtlSeconds":-1}`, - "zero sweep": `{"evictionSweepSeconds":0}`, - "negative weight": `{"lasWeightService":-0.1}`, - "decay factor > 1": `{"lasDecayFactor":1.5}`, - "decay factor 0": `{"lasDecayFactor":0}`, - "negative half life": `{"lasHalfLifeSeconds":-1}`, + "negative ttl": `{"evictionTtlSeconds":-1}`, + "zero sweep": `{"evictionSweepSeconds":0}`, + "negative weight": `{"lasWeightService":-0.1}`, + "decay factor > 1": `{"lasDecayFactor":1.5}`, + "decay factor 0": `{"lasDecayFactor":0}`, + "negative half life": `{"lasHalfLifeSeconds":-1}`, + "negative turn weight": `{"turnPriorityTimeWeight":-0.1}`, } for name, cfg := range cases { t.Run(name, func(t *testing.T) { diff --git a/pkg/epp/framework/plugins/flowcontrol/fairness/program-aware/strategy.go b/pkg/epp/framework/plugins/flowcontrol/fairness/program-aware/strategy.go index fe77cd997f..4a5e0fc729 100644 --- a/pkg/epp/framework/plugins/flowcontrol/fairness/program-aware/strategy.go +++ b/pkg/epp/framework/plugins/flowcontrol/fairness/program-aware/strategy.go @@ -36,8 +36,10 @@ func newStrategy(cfg Config) (Strategy, error) { decayFactor: cfg.LASDecayFactor, halfLifeSeconds: cfg.LASHalfLifeSeconds, }, nil + case turnPriorityStrategyName: + return &turnPriorityStrategy{timeWeight: cfg.TurnPriorityTimeWeight}, nil default: - return nil, fmt.Errorf("unknown scoring strategy %q: only \"las\" is supported", cfg.Strategy) + return nil, fmt.Errorf("unknown scoring strategy %q: only \"las\" and \"turn-priority\" are supported", cfg.Strategy) } } diff --git a/pkg/epp/framework/plugins/flowcontrol/fairness/program-aware/strategy_turnpriority.go b/pkg/epp/framework/plugins/flowcontrol/fairness/program-aware/strategy_turnpriority.go new file mode 100644 index 0000000000..ecc9ab37b2 --- /dev/null +++ b/pkg/epp/framework/plugins/flowcontrol/fairness/program-aware/strategy_turnpriority.go @@ -0,0 +1,94 @@ +package programaware + +import ( + "math" + "time" + + "github.com/prometheus/client_golang/prometheus" + + "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" +) + +const turnPriorityStrategyName = "turn-priority" + +var _ Strategy = &turnPriorityStrategy{} + +// turnPriorityStrategy scores each flow by its head request's priority +// +// prio = turnNumber + timeWeight * elapsedSeconds +// +// and picks the highest. The elapsed term ages waiting flows up so a flow with +// a lower turn number is not starved indefinitely. The strategy keeps no +// accumulated per-program state. +type turnPriorityStrategy struct { + timeWeight float64 +} + +func (s *turnPriorityStrategy) Name() string { return turnPriorityStrategyName } + +func (s *turnPriorityStrategy) Pick(_ int, queues map[string]QueueInfo) flowcontrol.FlowQueueAccessor { + // Collect the flows with pending work. Scoring only orders flows against + // each other, so it is meaningful only under contention: with no waiting + // flow there is nothing to dispatch, and with exactly one the choice is + // forced. Score only when two or more flows are waiting. + waiting := make([]QueueInfo, 0, len(queues)) + for _, qi := range queues { + if qi.Len == 0 || qi.Queue.Peek() == nil { + continue + } + waiting = append(waiting, qi) + } + + switch len(waiting) { + case 0: + return nil + case 1: + return waiting[0].Queue + } + + var best flowcontrol.FlowQueueAccessor + bestScore := math.Inf(-1) + now := time.Now() + + for _, qi := range waiting { + head := qi.Queue.Peek() + elapsed := now.Sub(head.EnqueueTime()).Seconds() + if elapsed < 0 { + elapsed = 0 + } + score := float64(turnNumberFor(qi.Metrics)) + s.timeWeight*elapsed + if score > bestScore { + bestScore = score + best = qi.Queue + } + } + + return best +} + +func (s *turnPriorityStrategy) OnPreRequest(_ *ProgramMetrics, _ *fwksched.InferenceRequest) {} + +func (s *turnPriorityStrategy) OnCompleted(_ *ProgramMetrics, _ *fwksched.InferenceRequest, _ *fwkrc.Response) { +} + +func (s *turnPriorityStrategy) EvictProgram(_ string) {} + +func (s *turnPriorityStrategy) Collectors() []prometheus.Collector { return nil } + +// turnNumberFor returns the turn number of a program's head request: the count +// of requests already dispatched for that program plus the waiting request +// itself (dispatchedCount+1). +// +// The counter is program-scoped: each program (fairness ID) has its own count, +// and every dispatch advances it by one. The count resets to zero when idle +// program state is evicted: a program that goes idle past the eviction TTL has +// likely lost its KV cache, so on return it restarts from turn one and re-earns +// priority from a cold state. +func turnNumberFor(metrics *ProgramMetrics) int64 { + if metrics == nil { + return 1 + } + return metrics.DispatchedCount() + 1 +} diff --git a/pkg/epp/framework/plugins/flowcontrol/fairness/program-aware/strategy_turnpriority_test.go b/pkg/epp/framework/plugins/flowcontrol/fairness/program-aware/strategy_turnpriority_test.go new file mode 100644 index 0000000000..fde86478a2 --- /dev/null +++ b/pkg/epp/framework/plugins/flowcontrol/fairness/program-aware/strategy_turnpriority_test.go @@ -0,0 +1,95 @@ +package programaware + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestTurnPriority_Name(t *testing.T) { + assert.Equal(t, "turn-priority", (&turnPriorityStrategy{}).Name()) +} + +func TestTurnPriority_Pick_PrefersLongerWait(t *testing.T) { + s := &turnPriorityStrategy{timeWeight: 0.5} + now := time.Now() + + // Both programs have zero dispatches (equal turn), so the longer-waiting + // head wins on the elapsed term. + idA, qA := makeInfo("alpha", now.Add(-5*time.Second)) + idB, qB := makeInfo("beta", now.Add(-1*time.Second)) + queues := map[string]QueueInfo{idA: qA, idB: qB} + + got := s.Pick(0, queues) + require.NotNil(t, got) + assert.Equal(t, "alpha", got.FlowKey().ID) +} + +func TestTurnPriority_Pick_PrefersHigherTurn(t *testing.T) { + s := &turnPriorityStrategy{timeWeight: 0.5} + now := time.Now() + + // Equal wait, but beta has more dispatched requests, so its turn number is + // higher and it wins. + idA, qA := makeInfo("alpha", now.Add(-1*time.Second)) + idB, qB := makeInfo("beta", now.Add(-1*time.Second)) + for i := 0; i < 10; i++ { + qB.Metrics.RecordDispatched(time.Time{}) + } + queues := map[string]QueueInfo{idA: qA, idB: qB} + + got := s.Pick(0, queues) + require.NotNil(t, got) + assert.Equal(t, "beta", got.FlowKey().ID) +} + +func TestTurnPriority_TurnNumberFromDispatchedCount(t *testing.T) { + // Waiting request is the next turn: dispatchedCount + 1. + assert.Equal(t, int64(1), turnNumberFor(&ProgramMetrics{})) + assert.Equal(t, int64(1), turnNumberFor(nil)) + + m := &ProgramMetrics{} + m.RecordDispatched(time.Time{}) + m.RecordDispatched(time.Time{}) + assert.Equal(t, int64(3), turnNumberFor(m)) +} + +func TestTurnPriority_Pick_SkipsEmptyQueues(t *testing.T) { + s := &turnPriorityStrategy{timeWeight: 0.5} + queues := map[string]QueueInfo{ + "empty": {Queue: makeQueue("empty", 0, time.Time{}), Metrics: &ProgramMetrics{}, Len: 0}, + } + assert.Nil(t, s.Pick(0, queues)) +} + +func TestTurnPriority_Pick_SingleWaitingFlowDispatchesDirectly(t *testing.T) { + s := &turnPriorityStrategy{timeWeight: 0.5} + + // A just-enqueued head scores ~0; with only one waiting flow it must still + // be picked, proving the single-flow path bypasses scoring. + id, qi := makeInfo("only", time.Now()) + queues := map[string]QueueInfo{ + id: qi, + "empty": {Queue: makeQueue("empty", 0, time.Time{}), Metrics: &ProgramMetrics{}, Len: 0}, + } + + got := s.Pick(0, queues) + require.NotNil(t, got) + assert.Equal(t, "only", got.FlowKey().ID) +} + +func TestTurnPriority_Pick_ZeroWeightIgnoresElapsed(t *testing.T) { + s := &turnPriorityStrategy{timeWeight: 0} + now := time.Now() + + // With timeWeight 0 and equal turn numbers, all scores tie; a flow is still + // selected rather than nil. + idA, qA := makeInfo("alpha", now.Add(-5*time.Second)) + idB, qB := makeInfo("beta", now.Add(-1*time.Second)) + queues := map[string]QueueInfo{idA: qA, idB: qB} + + got := s.Pick(0, queues) + require.NotNil(t, got) +}