diff --git a/experimental/sgl-router/src/config/cli.rs b/experimental/sgl-router/src/config/cli.rs index cc30b7188..66de72441 100644 --- a/experimental/sgl-router/src/config/cli.rs +++ b/experimental/sgl-router/src/config/cli.rs @@ -193,6 +193,19 @@ pub struct Cli { /// the published queue sums across a worker's DP ranks. #[arg(long)] pub worker_queue_limit: Option, + /// Saturation floor for `--worker-queue-limit` diversions: when no + /// cache candidate survives both the queue limit and hard admission, + /// at least one was over the limit, AND no worker in the routable + /// fleet has a fresh queue reading strictly below this floor, the + /// diverted request would wait wherever it lands, so it stays with the + /// least-pressured prefix owner instead — same wait, but prefilled + /// from cache instead of a full cold prefill that evicts other + /// prefixes and manufactures the next round of misses. Unset disables + /// the pin. Requires `--worker-queue-limit` (there is no diversion to + /// cancel without it) and must be at most the limit; scale with + /// `--dp-size` like the limit. + #[arg(long)] + pub saturation_queue_floor: Option, // ---- score composition ---- /// Policies to sum, spelled exactly as `--policy` spells them and each @@ -380,12 +393,35 @@ impl Cli { || self.cache_candidate_ratio.is_some() || self.cache_candidate_max_workers.is_some() || self.cache_switch_margin_tokens.is_some() - || self.worker_queue_limit.is_some(); + || self.worker_queue_limit.is_some() + || self.saturation_queue_floor.is_some(); // Value checks before the policy check: a value that is wrong under // every policy should say so, rather than pointing at --policy. if self.worker_queue_limit == Some(0) { return Err(anyhow!("--worker-queue-limit must be at least 1")); } + if let Some(floor) = self.saturation_queue_floor { + // The floor modifies the gate's diversion; without the gate + // there is no diversion to cancel and the knob would sit dead. + let Some(limit) = self.worker_queue_limit else { + return Err(anyhow!( + "--saturation-queue-floor requires --worker-queue-limit (there is no \ + diversion to cancel without it)" + )); + }; + if floor == 0 { + return Err(anyhow!("--saturation-queue-floor must be at least 1")); + } + // floor <= limit keeps the saturation label readable: a floor + // above the limit would declare the fleet saturated while + // workers the gate still admits exist. + if floor > limit { + return Err(anyhow!( + "--saturation-queue-floor ({floor}) must be at most --worker-queue-limit \ + ({limit})" + )); + } + } if tuned_cache_candidates && self.policy != PolicyKind::CacheAware { return Err(anyhow!( "cache candidate tuning flags require --policy cache_aware" @@ -610,6 +646,7 @@ impl Cli { .cache_switch_margin_tokens .unwrap_or(d.cache_switch_margin_tokens), worker_queue_limit: self.worker_queue_limit.or(d.worker_queue_limit), + saturation_queue_floor: self.saturation_queue_floor.or(d.saturation_queue_floor), }) } else { None @@ -1972,6 +2009,74 @@ mod tests { assert!(err.contains("--worker-queue-limit"), "got: {err}"); } + #[test] + fn saturation_queue_floor_requires_the_queue_gate_and_stays_below_it() { + let config = cfg_of( + "--policy cache_aware --kv-indexer-endpoint http://indexer:50051 \ + --worker-queue-limit 4 --saturation-queue-floor 2", + ) + .unwrap(); + assert_eq!( + config + .model + .affinity + .expect("cache-aware needs affinity config") + .saturation_queue_floor, + Some(2) + ); + + // Unset, the pin is disabled and the gate behaves as before. + let defaults = cfg_of( + "--policy cache_aware --kv-indexer-endpoint http://indexer:50051 \ + --worker-queue-limit 4", + ) + .unwrap(); + assert_eq!( + defaults + .model + .affinity + .expect("default affinity config") + .saturation_queue_floor, + None + ); + + // Without the gate there is no diversion to cancel. + let err = cfg_of( + "--policy cache_aware --kv-indexer-endpoint http://indexer:50051 \ + --saturation-queue-floor 2", + ) + .expect_err("the floor modifies the gate's diversion") + .to_string(); + assert!(err.contains("--saturation-queue-floor"), "got: {err}"); + assert!(err.contains("--worker-queue-limit"), "got: {err}"); + + // A floor above the limit would declare saturation while workers + // the gate still admits exist. + let err = cfg_of( + "--policy cache_aware --kv-indexer-endpoint http://indexer:50051 \ + --worker-queue-limit 4 --saturation-queue-floor 5", + ) + .expect_err("floor must not exceed the limit") + .to_string(); + assert!(err.contains("at most"), "got: {err}"); + + let err = cfg_of( + "--policy cache_aware --kv-indexer-endpoint http://indexer:50051 \ + --worker-queue-limit 4 --saturation-queue-floor 0", + ) + .expect_err("a zero floor would reject every queue reading") + .to_string(); + assert!(err.contains("--saturation-queue-floor"), "got: {err}"); + + let err = cfg_of("--policy power_of_two --worker-queue-limit 4 --saturation-queue-floor 2") + .expect_err("the pin only governs cache-affinity selection") + .to_string(); + assert!( + err.contains("cache candidate tuning flags require --policy cache_aware"), + "got: {err}" + ); + } + #[test] fn cache_candidate_cli_rejects_invalid_bounds() { for (args, expected) in [ diff --git a/experimental/sgl-router/src/config/types.rs b/experimental/sgl-router/src/config/types.rs index 9ce039c86..5860cf87f 100644 --- a/experimental/sgl-router/src/config/types.rs +++ b/experimental/sgl-router/src/config/types.rs @@ -534,7 +534,31 @@ pub struct AffinityConfig { /// Note the firing point scales with `dp_size`: the sample sums `waiting` /// across a worker's DP ranks while a request lands on one of them, so /// scale the limit with `--dp-size` on DP-attention deployments. + /// + /// The companion `saturation_queue_floor` cancels the gate's diversions + /// when they have no payoff (nothing in the fleet reads below the + /// floor). pub worker_queue_limit: Option, + /// Saturation pin (`--saturation-queue-floor`): cancels queue-gate + /// diversions that have no payoff. When no cache candidate survives + /// both `worker_queue_limit` and hard admission, at least one was over + /// the limit, AND no worker in the routable fleet has a fresh queue + /// reading strictly below this floor, the diverted request would wait + /// wherever it lands — so it pins to the least-pressured prefix owner + /// instead of cold-prefilling on a non-owner (which evicts other + /// prefixes and manufactures the next round of misses). `None` — the + /// default — preserves the pure gate behavior. + /// + /// Polarity note: a worker with no fresh sample does NOT count as idle + /// — the opposite of the gate's fail-open, and deliberately so. The + /// gate keeps affinity because that is the safe default action; the + /// pin asks whether a *provably better* destination exists, and an + /// unknown queue is not proof. Both polarities leave the request with + /// its prefix owner when the signal is missing. + /// + /// The CLI enforces `floor <= worker_queue_limit` and requires the + /// gate; like the limit, scale the floor with `dp_size`. + pub saturation_queue_floor: Option, } impl Default for AffinityConfig { @@ -558,6 +582,7 @@ impl Default for AffinityConfig { cache_candidate_max_workers: 32, cache_switch_margin_tokens: 1_024, worker_queue_limit: None, + saturation_queue_floor: None, } } } diff --git a/experimental/sgl-router/src/policies/admission.rs b/experimental/sgl-router/src/policies/admission.rs index 35be201f7..76fb09a6d 100644 --- a/experimental/sgl-router/src/policies/admission.rs +++ b/experimental/sgl-router/src/policies/admission.rs @@ -11,6 +11,11 @@ //! that reads [`EngineWorkerLoad::num_waiting_reqs`] rather than the native //! monitor fields: a worker already making requests wait cannot win on cache //! affinity. It fails open on a missing sample — see [`queue_gate_admits`]. +//! The optional saturation floor (`--saturation-queue-floor`) cancels a +//! diversion that has no payoff: when no candidate survives the gate and hard +//! admission, at least one was gate-rejected, and no worker in the routable +//! fleet reads below the floor, the request pins to the least-pressured +//! prefix owner instead of cold-prefilling on a non-owner. use crate::policies::engine_load::{EngineLoadSnapshot, EngineWorkerLoad, NativeCacheWorkerLoad}; use crate::policies::power_of_two::select_with_snapshot; @@ -113,6 +118,11 @@ pub enum DecisionReason { BackupPressureGuard, RangeFallback, CapacityFallbackPowerOfTwo, + /// Fleet saturated: no cache candidate survived the queue gate and + /// capacity admission, at least one was rejected by the gate, and no + /// fleet worker has a fresh queue reading below the saturation floor — + /// so the request pinned to a prefix owner instead of diverting. + SaturationPin, } #[derive(Clone)] @@ -198,6 +208,11 @@ pub(crate) fn fleet_is_all_queued( } /// Selects a worker from bounded cache candidates and records guard coverage. +/// `fleet` is the worker set this request could actually be routed to (the +/// model's healthy prefill pool). Only the saturation pin reads it, and it +/// must be the routable fleet rather than the router-wide load table: that +/// table also holds decode peers and other models' workers, none of which a +/// diversion could reach. pub fn resolve_cache_candidates( proposal: &CacheCandidateProposal, request_input_tokens: u64, @@ -212,24 +227,36 @@ pub fn resolve_cache_candidates( // the gate answer per candidate is what splits the set, so asking twice // would re-read the snapshot for every candidate on every request. let mut evaluated: Vec<&CacheCandidate> = Vec::with_capacity(proposal.candidates.len()); - let mut queue_gate_rejected_candidates = 0u64; + // Kept, not just counted: the saturation pin ranks these by pressure when + // it fires. Stays unallocated while the gate is disabled, because nothing + // is ever rejected then. + let mut queue_gate_rejected: Vec<&CacheCandidate> = Vec::new(); let mut queue_gate_best_rejected_blocks = 0u32; for candidate in &proposal.candidates { if queue_gate_admits(snapshot, &candidate.worker, queue_limit) { evaluated.push(candidate); } else { - queue_gate_rejected_candidates += 1; + queue_gate_rejected.push(candidate); queue_gate_best_rejected_blocks = queue_gate_best_rejected_blocks.max(candidate.matched_prefix_blocks); } } + let queue_gate_rejected_candidates = queue_gate_rejected.len() as u64; // Second tier, mirroring `range_fallback`: when the gate removed every // candidate AND nowhere in the fleet is unqueued, diversion cannot dodge // a wait, so returning no decision would trade the whole prefix for // nothing. Re-admit the ungated set. While an unqueued worker still // exists the gate keeps its teeth and the request leaves the prefix. - let queue_gate_fell_back = - evaluated.is_empty() && queue_gate_rejected_candidates > 0 && fleet_all_queued; + // + // A configured saturation floor supersedes this tier rather than stacking + // with it: the floor names a weaker, tunable saturation condition and + // pins with a pressure-only ranking, where re-admission would re-rank by + // uncached work first. Both keep the prefix; only one may decide which + // owner, so the explicit knob wins and this tier covers the unset case. + let queue_gate_fell_back = evaluated.is_empty() + && queue_gate_rejected_candidates > 0 + && fleet_all_queued + && proposal.saturation_queue_floor.is_none(); if queue_gate_fell_back { evaluated.extend(proposal.candidates.iter()); } @@ -253,8 +280,60 @@ pub fn resolve_cache_candidates( .copied() .min_by_key(|candidate| candidate.uncached_tokens) else { + // Saturation pin: no candidate survived the gate and hard + // admission (and at least one was gate-rejected), but diverting + // only pays when a meaningfully idle destination exists. With a + // floor configured and no fresh queue reading below it, the request + // would wait wherever it lands — so waiting at a prefix owner + // dominates: same wait, prefill from cache instead of a full cold + // prefill that evicts other prefixes and manufactures the next + // round of misses. Saturation suspends the gate, not the tiebreak: + // pin to the least-pressured rejected owner, skipping any that also + // fail hard admission (a capacity-exhausted owner cannot take the + // request). + let pinned = if queue_gate_rejected.is_empty() { + None + } else { + proposal.saturation_queue_floor.and_then(|floor| { + if snapshot + .any_fresh_queue_below(fleet.iter().map(|worker| worker.url.as_str()), floor) + { + return None; + } + // Ranked over its own lookup, not `loads`: `loads` covers the + // gate-ADMITTED set, which is empty precisely when the pin + // fires. An empty lookup reports no engine coverage, so every + // comparison would fall back to router-local load — tie at + // zero for every owner, decided by worker id. The pin ranks + // the rejected owners, so it must see the rejected owners. + let pin_loads = FreshLoadLookup::new( + Some(snapshot), + queue_gate_rejected + .iter() + .map(|candidate| &candidate.worker), + ); + queue_gate_rejected + .iter() + .copied() + .filter(|candidate| { + is_cache_candidate_admitted(candidate, request_input_tokens, &pin_loads) + }) + .min_by(|left, right| { + pin_loads + .compare_prefill_pressure(&left.worker, &right.worker) + .then_with(|| left.worker.id.0.cmp(&right.worker.id.0)) + }) + }) + }; return CacheCandidateResolution { - decision: None, + decision: pinned.map(|pinned| FinalDecision { + selected: Arc::clone(&pinned.worker), + primary: Arc::clone(&pinned.worker), + backup: None, + reason: DecisionReason::SaturationPin, + candidate_range_id: pinned.candidate_range_id.clone(), + load_snapshot_version: snapshot.version, + }), prefill_pressure_source: loads.prefill_pressure_source(), admission_evaluated_candidates, admission_rejected_candidates, @@ -1132,6 +1211,7 @@ mod tests { pressure_abs_threshold_ms: None, pressure_rel_threshold: 1.5, worker_queue_limit: None, + saturation_queue_floor: None, }; let loads = snapshot(&[ (&congested, 1, 1_000, 10, 10_000), @@ -1179,6 +1259,19 @@ mod tests { } } + fn saturation_proposal( + candidates: Vec, + limit: Option, + floor: Option, + ) -> CacheCandidateProposal { + CacheCandidateProposal { + candidates, + worker_queue_limit: limit, + saturation_queue_floor: floor, + ..Default::default() + } + } + #[test] fn queue_gate_diverts_off_an_owner_over_the_limit() { let owner = worker("owner"); @@ -1398,6 +1491,178 @@ mod tests { assert!(!resolution.queue_gate_fell_back); } + #[test] + fn saturation_pin_keeps_affinity_with_the_least_pressured_owner() { + let calm_owner = worker("calm_owner"); + let busy_owner = worker("busy_owner"); + // A fleet worker that is NOT a cache candidate: the saturation check + // reads fleet-wide fresh samples, not just the candidate set. It is + // over the floor, so it does not break the saturation claim. + let fleet_only = worker("fleet_only"); + let proposal = saturation_proposal( + vec![ + // The busy owner holds the deeper prefix and would win + // without the gate; both owners are over the limit. + candidate(&busy_owner, 10, 9), + candidate(&calm_owner, 60, 4), + ], + Some(4), + Some(2), + ); + let loads = snapshot(&[ + (&busy_owner, 1, 9, 10, 10_000), + (&calm_owner, 1, 5, 10, 10_000), + (&fleet_only, 1, 3, 10, 10_000), + ]); + let fleet = vec![ + Arc::clone(&busy_owner), + Arc::clone(&calm_owner), + Arc::clone(&fleet_only), + ]; + + let resolution = resolve_cache_candidates(&proposal, 100, &loads, &fleet); + + let decision = resolution + .decision + .expect("nothing reads below the floor: pin to a prefix owner"); + assert_eq!(decision.reason, DecisionReason::SaturationPin); + assert_eq!( + decision.selected.id, calm_owner.id, + "saturation suspends the gate, not the tiebreak" + ); + assert_eq!(decision.primary.id, calm_owner.id); + assert!(decision.backup.is_none()); + assert_eq!(decision.load_snapshot_version, loads.version); + assert_eq!(resolution.queue_gate_rejected_candidates, 2); + assert_eq!(resolution.queue_gate_best_rejected_blocks, 9); + assert_eq!(resolution.admission_rejected_candidates, 0); + } + + #[test] + fn no_saturation_floor_preserves_queue_gate_exhaustion() { + let owner = worker("owner"); + let other = worker("other"); + let proposal = saturation_proposal( + vec![candidate(&owner, 10, 9), candidate(&other, 60, 4)], + Some(4), + None, + ); + let loads = snapshot(&[(&owner, 1, 9, 10, 10_000), (&other, 1, 5, 10, 10_000)]); + let fleet = vec![Arc::clone(&owner), Arc::clone(&other)]; + + let resolution = resolve_cache_candidates(&proposal, 100, &loads, &fleet); + + // With no floor the queue gate's own second tier applies: every owner + // is over the limit and nowhere in the fleet is unqueued, so the + // prefix is kept rather than traded for a wait that cannot be dodged. + // The pin is what a floor buys; without one this is the behaviour. + assert!(resolution.queue_gate_fell_back); + assert_eq!( + resolution.decision.as_ref().map(|d| d.reason), + Some(DecisionReason::CacheCandidate) + ); + assert_eq!(resolution.queue_gate_rejected_candidates, 2); + assert_eq!(resolution.queue_gate_best_rejected_blocks, 9); + assert_eq!(resolution.admission_rejected_candidates, 0); + } + + #[test] + fn saturation_pin_yields_to_a_provably_idle_fleet_worker() { + let owner = worker("owner"); + // Not a candidate — but a fresh reading below the floor anywhere in + // the fleet means diverting can pay, so the pin must not fire. + let idle_elsewhere = worker("idle_elsewhere"); + let proposal = saturation_proposal(vec![candidate(&owner, 10, 9)], Some(4), Some(2)); + let loads = snapshot(&[ + (&owner, 1, 9, 10, 10_000), + (&idle_elsewhere, 1, 0, 10, 10_000), + ]); + let fleet = vec![Arc::clone(&owner), Arc::clone(&idle_elsewhere)]; + + let resolution = resolve_cache_candidates(&proposal, 100, &loads, &fleet); + + assert!(resolution.decision.is_none()); + assert_eq!(resolution.queue_gate_rejected_candidates, 1); + assert_eq!(resolution.queue_gate_best_rejected_blocks, 9); + } + + #[test] + fn saturation_pin_treats_an_unknown_queue_as_not_idle() { + let owner = worker("owner"); + // In the fleet, routable, and never published a sample. + let unsampled = worker("unsampled"); + let proposal = saturation_proposal(vec![candidate(&owner, 10, 9)], Some(4), Some(2)); + // The snapshot holds only the over-limit owner. `unsampled` is a + // real destination whose queue is unknown, not proof of a better + // one — opposite of the gate's fail-open. + let loads = snapshot(&[(&owner, 1, 9, 10, 10_000)]); + let fleet = vec![Arc::clone(&owner), Arc::clone(&unsampled)]; + + let resolution = resolve_cache_candidates(&proposal, 100, &loads, &fleet); + + let decision = resolution + .decision + .expect("an unknown queue must not read as below the floor"); + assert_eq!(decision.reason, DecisionReason::SaturationPin); + assert_eq!(decision.selected.id, owner.id); + } + + #[test] + fn saturation_pin_ignores_idle_workers_outside_the_routable_fleet() { + let owner = worker("owner"); + // Present in the router-wide load table but not routable for this + // request: a PD decode peer, another model's worker, or a worker the + // registry no longer reports healthy. Decode peers idle near zero + // waiting, so scanning the whole table would veto the pin on every + // PD deployment. + let off_fleet_idle = worker("off_fleet_idle"); + let proposal = saturation_proposal(vec![candidate(&owner, 10, 9)], Some(4), Some(2)); + let loads = snapshot(&[ + (&owner, 1, 9, 10, 10_000), + (&off_fleet_idle, 1, 0, 10, 10_000), + ]); + let fleet = vec![Arc::clone(&owner)]; + + let resolution = resolve_cache_candidates(&proposal, 100, &loads, &fleet); + + let decision = resolution + .decision + .expect("an unroutable idle worker is not a destination a diversion could reach"); + assert_eq!(decision.reason, DecisionReason::SaturationPin); + assert_eq!(decision.selected.id, owner.id); + } + + #[test] + fn saturation_pin_skips_a_capacity_exhausted_owner() { + let full = worker("full"); + let admitted_owner = worker("admitted_owner"); + let proposal = saturation_proposal( + vec![candidate(&full, 10, 9), candidate(&admitted_owner, 60, 4)], + Some(4), + Some(2), + ); + // Both owners are over the queue limit, and `full` is the pressure + // minimum — but it is also KV-exhausted (used + request exceeds + // capacity), so it cannot take the request even pinned. + let loads = snapshot(&[ + (&full, 1, 5, 10_000, 10_000), + (&admitted_owner, 1, 9, 10, 10_000), + ]); + let fleet = vec![Arc::clone(&full), Arc::clone(&admitted_owner)]; + + let resolution = resolve_cache_candidates(&proposal, 100, &loads, &fleet); + + let decision = resolution + .decision + .expect("the capacity-admitted owner can be pinned"); + assert_eq!(decision.reason, DecisionReason::SaturationPin); + assert_eq!(decision.selected.id, admitted_owner.id); + // `full` is booked under the gate, not capacity — the pin's own + // capacity filter must not pollute the audit counters. + assert_eq!(resolution.queue_gate_rejected_candidates, 2); + assert_eq!(resolution.admission_rejected_candidates, 0); + } + #[test] fn range_fallback_prefers_an_unqueued_worker_over_a_shallower_queueing_one() { // The primary fails KV-capacity admission so selection reaches the diff --git a/experimental/sgl-router/src/policies/cache_aware.rs b/experimental/sgl-router/src/policies/cache_aware.rs index 067306aa4..e434426d1 100644 --- a/experimental/sgl-router/src/policies/cache_aware.rs +++ b/experimental/sgl-router/src/policies/cache_aware.rs @@ -110,6 +110,7 @@ impl CacheAwarePolicy { pressure_abs_threshold_ms: self.config.pressure_abs_threshold_ms, pressure_rel_threshold: self.config.pressure_rel_threshold, worker_queue_limit: self.config.worker_queue_limit, + saturation_queue_floor: self.config.saturation_queue_floor, }) } diff --git a/experimental/sgl-router/src/policies/engine_load.rs b/experimental/sgl-router/src/policies/engine_load.rs index a42090bdd..208014be8 100644 --- a/experimental/sgl-router/src/policies/engine_load.rs +++ b/experimental/sgl-router/src/policies/engine_load.rs @@ -105,6 +105,33 @@ impl EngineLoadSnapshot { self.workers.get(worker_url) } + /// True when some worker in `fleet_urls` has a fresh queue reading + /// strictly below `floor` — a destination where a diverted request would + /// provably wait behind fewer than `floor` others. Workers with no fresh + /// sample do not count: the saturation pin asks whether a provably better + /// destination exists, and an unknown queue is not proof. This is the + /// opposite polarity from the queue gate's fail-open, and the two + /// compose: both leave the request with its prefix owner when the signal + /// is missing. + /// + /// The caller passes the routable fleet rather than letting this scan the + /// whole table: the table is router-wide, so it also holds decode peers, + /// other models' workers, and workers no longer healthy enough to receive + /// this request. None of those is a destination a diversion could reach, + /// and a decode peer idling at zero waiting would otherwise veto the pin + /// on every PD deployment. + pub fn any_fresh_queue_below<'u>( + &self, + fleet_urls: impl IntoIterator, + floor: u64, + ) -> bool { + fleet_urls.into_iter().any(|url| { + self.workers + .get(url) + .is_some_and(|load| load.num_waiting_reqs < floor) + }) + } + /// Returns only complete, fresh native Cache-Aware monitor data. pub fn fresh_native_cache_load_for_url( &self, diff --git a/experimental/sgl-router/src/policies/mod.rs b/experimental/sgl-router/src/policies/mod.rs index 1e5b827a1..5b0474c93 100644 --- a/experimental/sgl-router/src/policies/mod.rs +++ b/experimental/sgl-router/src/policies/mod.rs @@ -356,6 +356,14 @@ pub struct CacheCandidateProposal { /// waiting requests cannot win on cache affinity. `None` disables the /// gate. See [`crate::config::AffinityConfig::worker_queue_limit`]. pub worker_queue_limit: Option, + /// Saturation pin: when no candidate survives the gate and hard + /// admission, at least one was queue-gate-rejected, and no worker in + /// the routable fleet has a fresh queue reading strictly below this + /// floor, the request pins to the least-pressured rejected prefix + /// owner instead of diverting — the diversion cannot dodge a wait and + /// would forfeit the matched prefix. `None` disables the pin. See + /// [`crate::config::AffinityConfig::saturation_queue_floor`]. + pub saturation_queue_floor: Option, } /// Prefill proposal returned as either a pair or a Cache-Aware candidate set. diff --git a/experimental/sgl-router/src/policies/selection.rs b/experimental/sgl-router/src/policies/selection.rs index f5ddea637..40e7e799b 100644 --- a/experimental/sgl-router/src/policies/selection.rs +++ b/experimental/sgl-router/src/policies/selection.rs @@ -70,6 +70,8 @@ pub(crate) struct PrefillSelectionInputs<'a> { pub session_affinity_mode: SessionAffinityMode, /// `--worker-queue-limit`. `None` disables the queue gate entirely. pub worker_queue_limit: Option, + /// `--saturation-queue-floor`. `None` disables the saturation pin. + pub saturation_queue_floor: Option, } /// The queue-gate blind warn is sampled: it fires on a per-request path, and @@ -78,6 +80,13 @@ pub(crate) struct PrefillSelectionInputs<'a> { const QUEUE_GATE_BLIND_LOG_SAMPLE: u64 = 64; static QUEUE_GATE_BLIND_LOG_COUNTER: AtomicU64 = AtomicU64::new(0); +/// The saturation-pin info log is sampled for the same reason as the +/// queue-gate blind warn: it fires on a per-request path and the condition +/// (a saturated fleet) persists for many requests, so 1-in-64 surfaces it +/// without log flooding. +const SATURATION_PIN_LOG_SAMPLE: u64 = 64; +static SATURATION_PIN_LOG_COUNTER: AtomicU64 = AtomicU64::new(0); + /// Maps the queue-gate audit of a Cache-Aware selection that produced no /// winner onto its decision label. Pure so every boundary is pinned by unit /// tests rather than inferred from the ladder that calls it: @@ -354,21 +363,52 @@ impl<'a> Selector<'a> { prefill_pressure_source = cache_decision.prefill_pressure_source, "cache candidate winner", ); - inputs - .metrics - .record_policy_decision("cache_aware", "cache_candidate"); - inputs.metrics.record_cache_aware_decision( - &inputs.model_id.0, - if cache_decision.queue_gate_fell_back { - // The gate removed every owner and nowhere in the fleet is - // unqueued, so the prefix was kept rather than traded for a - // wait that cannot be dodged. Booked as saturation, never as - // a plain hit. - CacheAwareDecision::AllQueued - } else { - CacheAwareDecision::CacheHit - }, + inputs.metrics.record_policy_decision( + "cache_aware", + prefill_policy_reason( + PolicyKind::CacheAware, + ProposalKind::CacheAffinity, + decision.reason, + inputs.session_id.is_some_and(|value| !value.is_empty()), + true, + ), ); + if decision.reason == DecisionReason::SaturationPin { + // The pin books the saturation label because it always means + // affinity was kept under a queueing fleet. It does not retire + // the off-owner draw in `run`: when every gate-rejected owner + // also fails capacity admission the pin yields no decision, and + // the fallback records the same label from an off-owner landing. + inputs + .metrics + .record_cache_aware_decision(&inputs.model_id.0, CacheAwareDecision::AllQueued); + if SATURATION_PIN_LOG_COUNTER + .fetch_add(1, AtomicOrdering::Relaxed) + .is_multiple_of(SATURATION_PIN_LOG_SAMPLE) + { + tracing::info!( + model = %&inputs.model_id.0, + worker = %decision.selected.url, + saturation_queue_floor = inputs.saturation_queue_floor, + worker_queue_limit = inputs.worker_queue_limit, + "fleet saturated, keeping affinity with a queueing prefix owner \ + instead of diverting", + ); + } + } else { + inputs.metrics.record_cache_aware_decision( + &inputs.model_id.0, + if cache_decision.queue_gate_fell_back { + // The gate removed every owner and nowhere in the fleet is + // unqueued, so the prefix was kept rather than traded for a + // wait that cannot be dodged. Booked as saturation, never + // as a plain hit. + CacheAwareDecision::AllQueued + } else { + CacheAwareDecision::CacheHit + }, + ); + } Some(decision.selected) } @@ -525,6 +565,7 @@ fn prefill_policy_reason( PolicyKind::CacheAware => match (proposal, decision) { (_, DecisionReason::CacheCandidate) | (ProposalKind::CacheAffinity, DecisionReason::Primary) => "cache_candidate", + (_, DecisionReason::SaturationPin) => "saturation_pin", (_, DecisionReason::Primary) => "no_cache_candidate", (_, DecisionReason::BackupPrimaryAdmission) => "no_cache_candidate_admission_backup", (_, DecisionReason::BackupPressureGuard) => "no_cache_candidate_pressure_backup", @@ -540,6 +581,7 @@ fn prefill_policy_reason( DecisionReason::BackupPressureGuard => "pressure_backup", DecisionReason::RangeFallback => "range_fallback", DecisionReason::CapacityFallbackPowerOfTwo => "capacity_fallback_power_of_two", + DecisionReason::SaturationPin => "saturation_pin", }, } } @@ -784,6 +826,7 @@ mod tests { load_snapshot, workers, worker_queue_limit: None, + saturation_queue_floor: None, ttft_slo_ms: None, tps_slo: None, session_affinity_mode: SessionAffinityMode::Bucket, diff --git a/experimental/sgl-router/src/server/metrics.rs b/experimental/sgl-router/src/server/metrics.rs index 68af70db3..770d1635c 100644 --- a/experimental/sgl-router/src/server/metrics.rs +++ b/experimental/sgl-router/src/server/metrics.rs @@ -66,15 +66,22 @@ //! `sgl_router_diverted_overlap_blocks` — read it against the overlap of //! all selections: a diverted curve skewing high means the gate is trading //! large cached prefixes for short waits. -//! - `all_queued` — the queue gate removed every owner AND every worker in -//! the prefill fleet is queueing, so no diversion could dodge a wait. This -//! is the saturation signal for traffic the gate acted on, keyed on -//! saturation rather than on where the request landed: usually the request -//! kept its prefix, but when the re-admitted owners are also out of KV -//! capacity it lands off-owner and still books here. Reporting that case as `cache_miss` would hide the -//! saturation in the one state where it matters most. It deliberately -//! does NOT spell `cache_hit*`: a `decision=~"cache_hit.*"` hit-rate query -//! must not absorb it, or a fully saturated fleet reads as a healthy one. +//! - `all_queued` — the queue gate removed every owner and no diversion could +//! dodge a wait. Two conditions draw it. Without `--saturation-queue-floor` +//! it means every worker in the prefill fleet is queueing at or above +//! `--worker-queue-limit`. With a floor set, the saturation pin also draws +//! it on the weaker condition the floor names: no fleet worker reads +//! strictly below the floor. Since the floor may be lower than the limit, a +//! floor well under the limit widens this label to fleets that still hold +//! gate-admissible workers — read it against the configured floor, not as +//! "every worker is over the limit". It is keyed on saturation rather than +//! on where the request landed: usually the request kept its prefix, but +//! when the owners it would keep are also out of KV capacity it lands +//! off-owner and still books here. Reporting that case as `cache_miss` +//! would hide the saturation in the one state where it matters most. It +//! deliberately does NOT spell `cache_hit*`: a `decision=~"cache_hit.*"` +//! hit-rate query must not absorb it, or a fully saturated fleet reads as a +//! healthy one. //! //! The four `sgl_router_worker*` gauges and `sgl_router_workers` are sampled //! at scrape time from the live [`crate::workers::WorkerRegistry`] (passed to diff --git a/experimental/sgl-router/src/server/routes/chat.rs b/experimental/sgl-router/src/server/routes/chat.rs index 6b82fb108..2fbf82aa7 100644 --- a/experimental/sgl-router/src/server/routes/chat.rs +++ b/experimental/sgl-router/src/server/routes/chat.rs @@ -636,6 +636,12 @@ pub async fn chat_completions( .affinity .as_ref() .and_then(|config| config.worker_queue_limit); + let saturation_queue_floor = ctx + .config + .model + .affinity + .as_ref() + .and_then(|config| config.saturation_queue_floor); // Each Bucket retry rebuilds the proposal and reruns Admission/Guard. let worker = select_prefill_worker(&PrefillSelectionInputs { policy: policy.as_ref(), @@ -655,6 +661,7 @@ pub async fn chat_completions( tps_slo, session_affinity_mode, worker_queue_limit, + saturation_queue_floor, }) .map_err(|reason| policy_selection_failed(&ctx, &model_str, reason))?;