Skip to content

Commit eabf7c2

Browse files
loicmarchalloicmarchal
committed
feat: add priority based holdback strategies in flow control
Signed-off-by: Loic Marchal <lmarchal@redhat.com>
1 parent 8ae9991 commit eabf7c2

4 files changed

Lines changed: 774 additions & 0 deletions

File tree

‎cmd/epp/runner/runner.go‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -79,6 +79,7 @@ import (
7979
"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/flowcontrol/saturationdetector/concurrency"
8080
"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/flowcontrol/saturationdetector/utilization"
8181
"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/flowcontrol/usagelimits"
82+
"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/flowcontrol/usagelimits/priorityholdback"
8283
"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/requestcontrol/admitter/latencyslo"
8384
"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/requestcontrol/admitter/probabilisticadmitter"
8485
reqdataprodprefix "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/requestcontrol/dataproducer/approximateprefix"
@@ -589,6 +590,7 @@ func (r *Runner) registerInTreePlugins() {
589590
fwkplugin.Register(edf.EDFOrderingPolicyType, edf.EDFOrderingPolicyFactory)
590591
fwkplugin.Register(slodeadline.SLODeadlineOrderingPolicyType, slodeadline.SLODeadlineOrderingPolicyFactory)
591592
fwkplugin.Register(usagelimits.StaticUsageLimitPolicyType, usagelimits.StaticPolicyFactory)
593+
fwkplugin.Register(priorityholdback.PolicyType, priorityholdback.PolicyFactory)
592594

593595
// Register Request level data producer plugins as defaults for their respective data keys.
594596
fwkplugin.RegisterAsDefaultProducer(reqdataprodprefix.ApproxPrefixCachePluginType, reqdataprodprefix.ApproxPrefixCacheFactory, attrprefix.PrefixCacheMatchInfoDataKey)

‎pkg/epp/framework/plugins/flowcontrol/usagelimits/priorityholdback/config.go‎

Lines changed: 145 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,145 @@
1+
/*
2+
Copyright 2025 The Kubernetes Authors.
3+
4+
Licensed under the Apache License, Version 2.0 (the "License");
5+
you may not use this file except in compliance with the License.
6+
You may obtain a copy of the License at
7+
8+
http://www.apache.org/licenses/LICENSE-2.0
9+
10+
Unless required by applicable law or agreed to in writing, software
11+
distributed under the License is distributed on an "AS IS" BASIS,
12+
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
See the License for the specific language governing permissions and
14+
limitations under the License.
15+
*/
16+
17+
package priorityholdback
18+
19+
import (
20+
"errors"
21+
"fmt"
22+
23+
"k8s.io/utils/ptr"
24+
)
25+
26+
// strategy constants define the supported gating strategies.
27+
const (
28+
// strategyStepwiseSpread divides the [minCeiling, maxCeiling] range into equal steps based on
29+
// the count of active priorities, ignoring their absolute numerical difference.
30+
strategyStepwiseSpread = "stepwise-spread"
31+
32+
// strategyLinearProportional scales ceilings linearly based on numerical priority values
33+
// relative to the observed active range.
34+
strategyLinearProportional = "linear-proportional"
35+
)
36+
37+
const (
38+
// defaultMaxCeiling allows the highest priority to use full capacity.
39+
defaultMaxCeiling float64 = 1.0
40+
)
41+
42+
// apiConfig represents the external configuration schema for the priority holdback policy.
43+
// It is designed to be deserialized from JSON via the plugin's raw parameters.
44+
type apiConfig struct {
45+
// Strategy selects the gating algorithm used to compute per-priority admission ceilings.
46+
//
47+
// Required. Valid values: "stepwise-spread", "linear-proportional".
48+
Strategy *string `json:"strategy"`
49+
50+
// MinCeiling is the admission ceiling assigned to the lowest-priority traffic.
51+
// Determines how aggressively the lowest priority is gated as saturation rises.
52+
//
53+
// Required. Must be in [0.0, 1.0) and strictly less than MaxCeiling.
54+
MinCeiling *float64 `json:"minCeiling"`
55+
56+
// MaxCeiling is the admission ceiling assigned to the highest-priority traffic.
57+
// A value of 1.0 means the highest priority is only gated at full saturation.
58+
//
59+
// Defaults to 1.0 if unset. Must be in (0.0, 1.0] and strictly greater than MinCeiling.
60+
MaxCeiling *float64 `json:"maxCeiling,omitempty"`
61+
}
62+
63+
// config is the internal, fully-validated configuration used by the policy.
64+
type config struct {
65+
strategy string
66+
minCeiling float64
67+
maxCeiling float64
68+
}
69+
70+
// buildConfig applies the configuration lifecycle (defaulting and validation) and translates the
71+
// external schema into the internal domain model.
72+
// The provided apiConfig is copied to prevent mutation side-effects.
73+
func buildConfig(apiCfg *apiConfig) (*config, error) {
74+
var safeCfg apiConfig
75+
if apiCfg != nil {
76+
safeCfg = *apiCfg
77+
}
78+
79+
if err := checkRequired(&safeCfg); err != nil {
80+
return nil, fmt.Errorf("invalid priority holdback policy configuration: %w", err)
81+
}
82+
83+
applyDefaults(&safeCfg)
84+
85+
if err := validateConfig(&safeCfg); err != nil {
86+
return nil, fmt.Errorf("invalid priority holdback policy configuration: %w", err)
87+
}
88+
89+
return &config{
90+
strategy: *safeCfg.Strategy,
91+
minCeiling: *safeCfg.MinCeiling,
92+
maxCeiling: *safeCfg.MaxCeiling,
93+
}, nil
94+
}
95+
96+
// checkRequired verifies that mandatory fields are present before defaulting.
97+
func checkRequired(cfg *apiConfig) error {
98+
var errs []error
99+
100+
if cfg.Strategy == nil {
101+
errs = append(errs, fmt.Errorf("strategy is required"))
102+
}
103+
if cfg.MinCeiling == nil {
104+
errs = append(errs, fmt.Errorf("minCeiling is required"))
105+
}
106+
107+
return errors.Join(errs...)
108+
}
109+
110+
// applyDefaults populates unset optional fields with their standard defaults.
111+
func applyDefaults(cfg *apiConfig) {
112+
if cfg.MaxCeiling == nil {
113+
cfg.MaxCeiling = ptr.To(defaultMaxCeiling)
114+
}
115+
}
116+
117+
// validateConfig checks the constraints of the fully defaulted configuration.
118+
// It aggregates all validation failures rather than failing on the first error.
119+
func validateConfig(cfg *apiConfig) error {
120+
var errs []error
121+
122+
if cfg.Strategy != nil {
123+
switch *cfg.Strategy {
124+
case strategyStepwiseSpread, strategyLinearProportional:
125+
default:
126+
errs = append(errs, fmt.Errorf("unsupported strategy %q, must be one of: %q, %q",
127+
*cfg.Strategy, strategyStepwiseSpread, strategyLinearProportional))
128+
}
129+
}
130+
131+
if cfg.MinCeiling != nil && (*cfg.MinCeiling < 0.0 || *cfg.MinCeiling >= 1.0) {
132+
errs = append(errs, fmt.Errorf("minCeiling must be in [0.0, 1.0), got %f", *cfg.MinCeiling))
133+
}
134+
135+
if cfg.MaxCeiling != nil && (*cfg.MaxCeiling <= 0.0 || *cfg.MaxCeiling > 1.0) {
136+
errs = append(errs, fmt.Errorf("maxCeiling must be in (0.0, 1.0], got %f", *cfg.MaxCeiling))
137+
}
138+
139+
if cfg.MinCeiling != nil && cfg.MaxCeiling != nil && *cfg.MinCeiling >= *cfg.MaxCeiling {
140+
errs = append(errs, fmt.Errorf("minCeiling (%f) must be strictly less than maxCeiling (%f)",
141+
*cfg.MinCeiling, *cfg.MaxCeiling))
142+
}
143+
144+
return errors.Join(errs...)
145+
}

‎pkg/epp/framework/plugins/flowcontrol/usagelimits/priorityholdback/priority_holdback.go‎

Lines changed: 128 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,128 @@
1+
// Package priorityholdback implements a UsageLimitPolicy that computes differentiated admission
2+
// ceilings per priority level. Lower-priority traffic is gated at lower saturation thresholds,
3+
// reserving capacity for higher-priority work as the pool approaches saturation.
4+
//
5+
// Two strategies are supported:
6+
// - "stepwise-spread": equal ceiling spacing by rank, ignoring numerical priority values.
7+
// - "linear-proportional": ceiling spacing proportional to numerical priority differences.
8+
package priorityholdback
9+
10+
import (
11+
"context"
12+
"encoding/json"
13+
"fmt"
14+
15+
"github.com/llm-d/llm-d-router/pkg/epp/framework/interface/flowcontrol"
16+
"github.com/llm-d/llm-d-router/pkg/epp/framework/interface/plugin"
17+
)
18+
19+
// PolicyType is the registration type for the priority holdback usage limit policy.
20+
const PolicyType = "priority-holdback-policy"
21+
22+
// PolicyFactory creates a priorityHoldbackPolicy from JSON config.
23+
func PolicyFactory(name string, params *json.Decoder, _ plugin.Handle) (plugin.Plugin, error) {
24+
var apiCfg apiConfig
25+
if params != nil {
26+
if err := params.Decode(&apiCfg); err != nil {
27+
return nil, fmt.Errorf("failed to unmarshal priority holdback policy config: %w", err)
28+
}
29+
}
30+
cfg, err := buildConfig(&apiCfg)
31+
if err != nil {
32+
return nil, err
33+
}
34+
return newPriorityHoldbackPolicy(*cfg).withName(name), nil
35+
}
36+
37+
// priorityHoldbackPolicy gates lower-priority traffic as saturation rises. The gating strategy
38+
// is resolved to a function at construction time to avoid per-dispatch branching.
39+
type priorityHoldbackPolicy struct {
40+
name string
41+
cMax float64
42+
cMin float64
43+
computeFn func(cMin, cMax float64, priorities []int) (ceilings []float64)
44+
}
45+
46+
var _ flowcontrol.UsageLimitPolicy = &priorityHoldbackPolicy{}
47+
48+
func newPriorityHoldbackPolicy(cfg config) *priorityHoldbackPolicy {
49+
var fn func(cMin, cMax float64, priorities []int) (ceilings []float64)
50+
switch cfg.strategy {
51+
case strategyStepwiseSpread:
52+
fn = computeLimitStepwiseSpread
53+
case strategyLinearProportional:
54+
fn = computeLimitLinearProportional
55+
}
56+
return &priorityHoldbackPolicy{
57+
name: PolicyType,
58+
cMax: cfg.maxCeiling,
59+
cMin: cfg.minCeiling,
60+
computeFn: fn,
61+
}
62+
}
63+
64+
func (p *priorityHoldbackPolicy) withName(name string) *priorityHoldbackPolicy {
65+
if name != "" {
66+
p.name = name
67+
}
68+
return p
69+
}
70+
71+
func (p *priorityHoldbackPolicy) Name() string {
72+
return p.name
73+
}
74+
75+
func (p *priorityHoldbackPolicy) TypedName() plugin.TypedName {
76+
return plugin.TypedName{
77+
Type: PolicyType,
78+
Name: p.name,
79+
}
80+
}
81+
82+
// ComputeLimit returns an admission ceiling for each priority. With a single active priority,
83+
// holdback is bypassed (ceiling = cMax) to preserve work-conserving behavior.
84+
func (p *priorityHoldbackPolicy) ComputeLimit(_ context.Context, _ float64, priorities []int) (ceilings []float64) {
85+
if len(priorities) == 0 {
86+
return []float64{}
87+
}
88+
if len(priorities) == 1 {
89+
return []float64{p.cMax}
90+
}
91+
return p.computeFn(p.cMin, p.cMax, priorities)
92+
}
93+
94+
// computeLimitStepwiseSpread divides [cMin, cMax] into equal steps by rank.
95+
// c_i = cMax - i * (cMax - cMin) / (N - 1)
96+
func computeLimitStepwiseSpread(cMin, cMax float64, priorities []int) (ceilings []float64) {
97+
ceilings = make([]float64, len(priorities))
98+
spread := cMax - cMin
99+
n := float64(len(priorities) - 1)
100+
for i := range priorities {
101+
ceilings[i] = cMax - float64(i)*spread/n
102+
}
103+
return ceilings
104+
}
105+
106+
// computeLimitLinearProportional scales ceilings proportionally to numerical priority values.
107+
// r_i = (p_i - pMin) / (pMax - pMin)
108+
// c_i = cMin + r_i * (cMax - cMin)
109+
func computeLimitLinearProportional(cMin, cMax float64, priorities []int) (ceilings []float64) {
110+
ceilings = make([]float64, len(priorities))
111+
pMin := float64(priorities[len(priorities)-1])
112+
pMax := float64(priorities[0])
113+
pRange := pMax - pMin
114+
// All priorities share the same value; no differentiation possible.
115+
// Also avoid division by zero in the formula for r_i.
116+
if pRange == 0 {
117+
for i := range priorities {
118+
ceilings[i] = cMax
119+
}
120+
return ceilings
121+
}
122+
spread := cMax - cMin
123+
for i := range priorities {
124+
r := (float64(priorities[i]) - pMin) / pRange
125+
ceilings[i] = cMin + r*spread
126+
}
127+
return ceilings
128+
}

0 commit comments

Comments
 (0)