diff --git a/CLAUDE.md b/CLAUDE.md index 8a6209655..24c6769ad 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -46,6 +46,7 @@ cargo run --release --features glm52 -- --model-path models/GLM5.2 - `PEGAINFER_NVCC_JOBS` — override parallel nvcc job count - `PEGAINFER_KV_FP8` — gemma4 opt-in fp8 KV: `local` stores the sliding family's K/V as e4m3 at scale 1.0 (lossy; halves the local pool; refuses an enabled prefix cache; unset = byte-identical serving) - `PEGAINFER_PREFIX_CACHE` — gemma4 opt-in conversation prefix cache: `K` entries of captured prompt state resume multi-turn prompts (pre-allocated page budget; unset = off, byte-identical serving) +- `PEGAINFER_ADMIT_COALESCE_MS` — gemma4 opt-in admission coalesce door: `N` ms in `1..=2000` (`off`/`0`/unset = admit on sight), holds arrivals that would invade a live decode batch so a window's arrivals land as one admission burst; refuses the async prefill lane; merging into one mixed step needs the chunked walk or a sub-budget prompt - `PEGAINFER_ASYNC_PREFILL` — gemma4 opt-in overlap lane: `green:NN` prefills live-batch admissions on an SM-capped stream to protect decode tails (`shared` for comparison; unset = off; bad values refuse to start) - `PEGAINFER_MIX_CHUNK_TOKENS` — gemma4 opt-in chunked walk: a mixed admission computes at most `N` prompt rows per step (`64 <= N <` the serving ceiling; unset = whole-prompt steps; bad values refuse to start) - `PEGAINFER_MAX_CONTEXT` — gemma4 serving ceiling raise (default 8192, up to the checkpoint's 262144; a raise past the default needs `PEGAINFER_MIX_CHUNK_TOKENS` and refuses the async lane) diff --git a/docs/index.md b/docs/index.md index 63a5950b1..021ca864a 100644 --- a/docs/index.md +++ b/docs/index.md @@ -64,7 +64,7 @@ Organized by domain (model line / subsystem / playbook / lesson) instead of by l | Path | TL;DR | | --- | --- | | `models/gemma4/tokenizer.md` | Gemma 4 tokenizer/chat-template contracts, gated against a Hugging Face reference (token ids plus all five chat renders, content flattened to strings): BOS comes only from the standalone `chat_template.jinja` (which opens a thought channel and accepts a native system role), EOS is declared three times with three values, the published defaults are sampled rather than greedy, image/audio tokens encode straight from user text so text-only serving must reject them at admission, and one divergence stays open — the server's default content format adds a trailing space to system turns. | -| `models/gemma4/serving.md` | What the engine promises under load: iteration level scheduling with a ceiling on prompt plus output (8192 by default), prompts prefilled whole at a step boundary by default, requests beyond the configured decode slots (16 by default) queued rather than refused, and the two KV families budgeted separately (7.27 GiB sliding + 2.00 GiB global at 12B with the chunk knob off; the sliding budget shrinks to window plus segment under it). The default configuration needs a 48 GiB card: 32.2 GiB resident before the first request. Clients must send `` themselves or the model degenerates. A row is bit-identical when its companions' content and lengths change under a fixed batch width trajectory, and moves when arrivals or retirements change that trajectory, so greedy output is reproducible for a workload rather than across workloads. An opt-in conversation prefix cache (`PEGAINFER_PREFIX_CACHE=K`) resumes multi-turn prompts from captured prompt state at a pre-allocated page cost. An opt-in chunked walk (`PEGAINFER_MIX_CHUNK_TOKENS=N`) walks admissions through shared segment steps with round-by-round page reservation, and `PEGAINFER_MAX_CONTEXT` raises the ceiling to the checkpoint's 262144 while `PEGAINFER_DECODE_SLOTS` trades concurrency for the memory that buys. Open: no cross-request prefix sharing, single GPU. | +| `models/gemma4/serving.md` | What the engine promises under load: iteration level scheduling with a ceiling on prompt plus output (8192 by default), prompts prefilled whole at a step boundary by default, requests beyond the configured decode slots (16 by default) queued rather than refused, and the two KV families budgeted separately (7.27 GiB sliding + 2.00 GiB global at 12B with the chunk knob off; the sliding budget shrinks to window plus segment under it). The default configuration needs a 48 GiB card: 32.2 GiB resident before the first request. Clients must send `` themselves or the model degenerates. A row is bit-identical when its companions' content and lengths change under a fixed batch width trajectory, and moves when arrivals or retirements change that trajectory, so greedy output is reproducible for a workload rather than across workloads. An opt-in conversation prefix cache (`PEGAINFER_PREFIX_CACHE=K`) resumes multi-turn prompts from captured prompt state at a pre-allocated page cost. An opt-in chunked walk (`PEGAINFER_MIX_CHUNK_TOKENS=N`) walks admissions through shared segment steps with round-by-round page reservation, and `PEGAINFER_MAX_CONTEXT` raises the ceiling to the checkpoint's 262144 while `PEGAINFER_DECODE_SLOTS` trades concurrency for the memory that buys. `PEGAINFER_ADMIT_COALESCE_MS=N` batches arrivals that would interrupt a live decode batch into one admission burst, defaults off, and refuses the async lane. Open: no cross-request prefix sharing, single GPU. | | `models/gemma4/hf-golden.md` | Three Hugging Face references for 12B. The base fixture carries layer-boundary activations at both ends of both layer types plus top-64 logprobs, over a single-token, a nine-token and a 1024-token (exactly the sliding window) case, and pins three facts the forward path has to match — the embedding scale is bf16 62.0 rather than `sqrt(3840)`, text attention is causal, and `layer_scalar` applies to the layer output after both residual adds. The window fixture goes past the window (1023/1024/1025/4096, teacher-forced) under both sdpa and eager. The long-context fixture takes the same comparison to 16384/32768 for the raised ceiling, sdpa-only, with the window fixture's dual-backend floor on loan. Regeneration is byte-identical and checked with sha256. | ## models / glm52 diff --git a/docs/models/gemma4/serving.md b/docs/models/gemma4/serving.md index f89f5b2a1..7cfe99951 100644 --- a/docs/models/gemma4/serving.md +++ b/docs/models/gemma4/serving.md @@ -110,6 +110,12 @@ The cache brings its own page budget, added to the pool lines above at startup. At `PEGAINFER_PREFIX_CACHE=16` the idle footprint measured **39242 MiB** against the 33034 MiB baseline — the difference is the pre-allocated cache budget. +## The admission coalesce door (opt-in) + +`PEGAINFER_ADMIT_COALESCE_MS=N` (`1..=2000`; unset, `off` or `0` admits on sight) holds arrivals that would invade a live decode batch, then releases a window's arrivals as one back-to-back admission burst. It prices the number of admission interruptions, not their size: whole prompts beyond the 512-row gather budget still take separate weight scans unless `PEGAINFER_MIX_CHUNK_TOKENS` enables the chunked walk. An idle engine admits immediately, and a shallow roster skips the door when `(active + pending) * 2 < slots`. + +A deep roster releases when the window expires or the pending queue reaches `min(4, slots - active)`, with a floor of one. A full cohort releases before `N`; the timeout release lands no earlier than `N`, at the first intake turn after the window elapses — the engine drains its submission channel before each intake, so there is no hard bound on how much later. That cohort is a capacity bound over the currently free slots, not a cross-completion batch. The door refuses to combine with `PEGAINFER_ASYNC_PREFILL`, whose single in-flight prefill could only be delayed by it. Measured under sustained load, c16 median TPOT improves about 8.5% for about +288 ms median TTFT; c8 pays about 5.7% throughput and about +19 ms TTFT. P99 ITL is flat to slightly worse everywhere, so the door remains off by default. + ## The async prefill lane (opt-in) When `PEGAINFER_ASYNC_PREFILL` is unset, serving uses the normal mixed-step path; when set, a live-batch admission's prefill moves onto its own stream so decode steps keep replaying while the prompt computes. Dense and routed checkpoints share this path at the default context ceiling. `green:NN` pins the lane to roughly NN% of the SMs via a Green Context — the cap is the mechanism: a `shared` lane's full-width prefill grids starve decode steps, and is kept only for comparison. An unrecognized value or an unviable SM partition refuses to start rather than silently degrading. diff --git a/pegainfer-gemma4/src/engine.rs b/pegainfer-gemma4/src/engine.rs index 96cfe478b..ddfc6ac0a 100644 --- a/pegainfer-gemma4/src/engine.rs +++ b/pegainfer-gemma4/src/engine.rs @@ -46,6 +46,7 @@ const MIX_CHUNK_TOKENS_ENV: &str = "PEGAINFER_MIX_CHUNK_TOKENS"; const MAX_CONTEXT_ENV: &str = "PEGAINFER_MAX_CONTEXT"; const DECODE_SLOTS_ENV: &str = "PEGAINFER_DECODE_SLOTS"; const KV_FP8_ENV: &str = "PEGAINFER_KV_FP8"; +const ADMIT_COALESCE_ENV: &str = "PEGAINFER_ADMIT_COALESCE_MS"; const MIN_CONTEXT: usize = 1024; const MIN_CHUNK_TOKENS: usize = 64; const CEILING_DOMAIN: usize = i32::MAX as usize; @@ -194,6 +195,62 @@ fn parse_kv_fp8(raw: Option<&str>) -> Result { Some(value) => anyhow::bail!("PEGAINFER_KV_FP8 supports only \"local\", got {value:?}"), } } +fn admit_coalesce_ms() -> Result> { + read_env(ADMIT_COALESCE_ENV)?.map_or(Ok(None), |raw| parse_admit_coalesce_ms(&raw)) +} + +fn parse_admit_coalesce_ms(raw: &str) -> Result> { + let value = raw.trim().to_ascii_lowercase(); + match value.as_str() { + "" | "0" | "off" => Ok(None), + other => match other.parse::() { + Ok(ms) if (1..=2000).contains(&ms) => Ok(Some(std::time::Duration::from_millis(ms))), + _ => anyhow::bail!( + "{ADMIT_COALESCE_ENV}={raw:?} not recognized (off | N ms, 1 <= N <= 2000)" + ), + }, + } +} + +/// Holds arrivals that would invade a live decode batch so one window's +/// arrivals land as a back-to-back burst of admissions: the stream's tail +/// gap prices the number of interruptions. One mixed step merges extra +/// prompts only with chunking or while its leader is under +/// `MIX_GATHER_ROWS`. The cohort bounds free-slot capacity, not a batch +/// across completions; idle engines admit on sight and shallow batches skip. +struct CoalesceDoor { + window: std::time::Duration, + since: Option, +} + +impl CoalesceDoor { + fn new(window: std::time::Duration) -> Self { + Self { + window, + since: None, + } + } + + fn opens( + &mut self, + pending: usize, + active: usize, + slots: usize, + now: std::time::Instant, + ) -> bool { + if pending == 0 || active == 0 || (active + pending) * 2 < slots { + self.since = None; + return true; + } + let cohort = MIX_MAX_PROMPTS.min(slots.saturating_sub(active)).max(1); + let since = *self.since.get_or_insert(now); + let open = pending >= cohort || now.duration_since(since) >= self.window; + if open { + self.since = None; + } + open + } +} pub(crate) fn start(model_path: &Path, options: &EngineLoadOptions) -> Result { let dir = model_path @@ -235,6 +292,7 @@ pub(crate) fn start(model_path: &Path, options: &EngineLoadOptions) -> Result = VecDeque::new(); let mut active: Vec = Vec::new(); let mut disconnected = false; + let mut door = state.coalesce_door(); 'engine: loop { loop { match submit_rx.try_recv() { @@ -283,7 +341,12 @@ pub(crate) fn start(model_path: &Path, options: &EngineLoadOptions) -> Result, } impl EngineState { + fn coalesce_door(&self) -> Option { + self.admit_coalesce.map(CoalesceDoor::new) + } + + fn intake_turn( + &mut self, + door: &mut Option, + pending: &mut VecDeque, + active: &mut Vec, + now: std::time::Instant, + ) -> bool { + let open = door + .as_mut() + .is_none_or(|door| door.opens(pending.len(), active.len(), self.slots, now)); + if open { + self.admit_from_queue(pending, active); + } + open + } + fn reserve_with_eviction( &mut self, kv: &mut GemmaKv, @@ -901,8 +987,14 @@ impl EngineState { let max_context = serving_context(config.max_position_embeddings)?; let lane_mode = async_prefill_mode()?; let mix_chunk = mix_chunk_tokens(max_context)?; + let admit_coalesce = admit_coalesce_ms()?; let slots = decode_slots()?; let local_kv_storage = kv_fp8_storage()?; + anyhow::ensure!( + admit_coalesce.is_none() || lane_mode.is_none(), + "{ADMIT_COALESCE_ENV} and {ASYNC_PREFILL_ENV} cannot combine: the lane flies one \ + prefill at a time, so the door could only delay it" + ); if max_context > MAX_CONTEXT { anyhow::ensure!( mix_chunk.is_some(), @@ -1036,6 +1128,7 @@ impl EngineState { mix_chunk, max_context, slots, + admit_coalesce, }) } @@ -2275,6 +2368,103 @@ mod knob_tests { assert!(parse_kv_fp8(Some("global")).is_err()); } + #[test] + fn admit_coalesce_parses_or_refuses() { + for off in ["off", "0", ""] { + assert_eq!(parse_admit_coalesce_ms(off).unwrap(), None); + } + assert_eq!( + parse_admit_coalesce_ms("300").unwrap(), + Some(std::time::Duration::from_millis(300)) + ); + assert_eq!( + parse_admit_coalesce_ms("1").unwrap(), + Some(std::time::Duration::from_millis(1)) + ); + assert_eq!( + parse_admit_coalesce_ms("2000").unwrap(), + Some(std::time::Duration::from_millis(2000)) + ); + for bad in ["0x", "2001", "abc"] { + assert!(parse_admit_coalesce_ms(bad).is_err(), "{bad:?} must refuse"); + } + } + + fn test_door() -> CoalesceDoor { + CoalesceDoor::new(std::time::Duration::from_millis(100)) + } + + #[test] + fn coalesce_door_idle_opens_and_clears_the_timer() { + let now = std::time::Instant::now(); + let mut door = test_door(); + door.since = Some(now); + assert!(door.opens(1, 0, 8, now)); + assert_eq!(door.since, None); + } + + #[test] + fn coalesce_door_empty_queue_opens() { + let now = std::time::Instant::now(); + let mut door = test_door(); + assert!(door.opens(0, 4, 8, now)); + assert_eq!(door.since, None); + } + + #[test] + fn coalesce_door_shallow_batch_opens() { + let now = std::time::Instant::now(); + let mut door = test_door(); + assert!(door.opens(1, 1, 8, now)); + assert_eq!(door.since, None); + } + + #[test] + fn coalesce_door_deep_under_cohort_batch_closes_and_pins_the_timer() { + let now = std::time::Instant::now(); + let mut door = test_door(); + assert!(!door.opens(1, 3, 8, now)); + assert_eq!(door.since, Some(now)); + assert!(!door.opens(1, 3, 8, now + std::time::Duration::from_millis(1))); + assert_eq!(door.since, Some(now)); + } + + #[test] + fn coalesce_door_full_cohort_opens_and_clears() { + let now = std::time::Instant::now(); + let mut door = test_door(); + door.since = Some(now); + assert!(door.opens(4, 4, 8, now)); + assert_eq!(door.since, None); + } + + #[test] + fn coalesce_door_elapsed_window_opens() { + let now = std::time::Instant::now(); + let mut door = test_door(); + let window = door.window; + assert!(!door.opens(1, 3, 8, now)); + assert!(door.opens(1, 3, 8, now + window)); + assert_eq!(door.since, None); + } + + #[test] + fn coalesce_door_full_roster_opens_for_one_capacity_bounded_arrival() { + let now = std::time::Instant::now(); + let mut door = test_door(); + assert!(door.opens(1, 8, 8, now)); + assert_eq!(door.since, None); + } + + #[test] + fn coalesce_door_freed_slots_rederive_the_cohort() { + let now = std::time::Instant::now(); + let mut door = test_door(); + assert!(door.opens(2, 6, 8, now)); + assert!(!door.opens(2, 5, 8, now)); + assert_eq!(door.since, Some(now)); + } + #[test] fn chunk_mode_parses_or_refuses() { for off in ["", "0", "off", " OFF "] { @@ -2385,6 +2575,7 @@ mod lane_tests { struct Drained { tokens: usize, cached: usize, + scheduled: usize, finish: FinishReason, ids: Vec, } @@ -2392,6 +2583,7 @@ mod lane_tests { fn drain(rx: &mut TokenStreamReceiver, name: &str) -> Drained { let mut tokens = 0; let mut cached = 0; + let mut scheduled = 0; let mut ids = Vec::new(); loop { match rx.blocking_recv().map(|(_, event)| event) { @@ -2399,12 +2591,16 @@ mod lane_tests { tokens += 1; ids.push(id); } - Some(TokenEvent::Scheduled { cached_tokens, .. }) => cached = cached_tokens, + Some(TokenEvent::Scheduled { cached_tokens, .. }) => { + cached = cached_tokens; + scheduled += 1; + } Some(TokenEvent::PromptTokens { .. } | TokenEvent::KvTransfer { .. }) => {} Some(TokenEvent::Finished { finish_reason, .. }) => { return Drained { tokens, cached, + scheduled, finish: finish_reason, ids, }; @@ -2484,13 +2680,14 @@ mod lane_tests { } } - const SERVING_KNOBS: [&str; 6] = [ - "PEGAINFER_ASYNC_PREFILL", - "PEGAINFER_PREFIX_CACHE", - "PEGAINFER_MIX_CHUNK_TOKENS", - "PEGAINFER_MAX_CONTEXT", - "PEGAINFER_DECODE_SLOTS", - "PEGAINFER_KV_FP8", + const SERVING_KNOBS: [&str; 7] = [ + super::ASYNC_PREFILL_ENV, + super::PREFIX_CACHE_ENV, + super::MIX_CHUNK_TOKENS_ENV, + super::MAX_CONTEXT_ENV, + super::DECODE_SLOTS_ENV, + super::ADMIT_COALESCE_ENV, + super::KV_FP8_ENV, ]; /// Clear every serving knob, set `overrides`, and hand back the guard @@ -2711,9 +2908,8 @@ mod lane_tests { super::EngineState::load(&dir, 0, policy, 0x5EED, true).expect("engine state") } - /// A raise without the chunk knob, and a raise with the overlap lane, - /// both refuse before the multi-GiB load — the startup policy the - /// serving doc promises. + /// Invalid knob combinations refuse before the multi-GiB load — the + /// startup policy the serving doc promises. #[test] #[ignore = "requires the pinned 12B checkpoint and --test-threads=1"] fn the_raise_refuses_without_its_prerequisites() { @@ -2741,6 +2937,207 @@ mod lane_tests { format!("{err:#}").contains("unsupported over"), "unexpected refusal: {err:#}" ); + let err = load(&[ + (super::ADMIT_COALESCE_ENV, "300"), + (super::ASYNC_PREFILL_ENV, "green:35"), + ]) + .err() + .expect("the coalesce door and async lane must refuse"); + assert!( + format!("{err:#}").contains("the door could only delay it"), + "unexpected refusal: {err:#}" + ); + } + + /// Drive the production intake turn with an injected clock: closed turns + /// preserve the live stream, cohort and timeout releases each drain one + /// burst, and an open door still respects slot capacity. + #[test] + #[ignore = "requires the pinned 12B checkpoint, a GPU, and --test-threads=1"] + fn the_coalesce_door_releases_one_admission_burst() { + let dir = crate::testkit::model_path(); + let policy = super::generation_policy(&dir).expect("policy"); + let _env = scoped_engine_env(&[ + (super::ADMIT_COALESCE_ENV, "2000"), + (super::DECODE_SLOTS_ENV, "4"), + ]); + let mut state = + super::EngineState::load(&dir, 0, policy, 0x5EED, true).expect("engine state"); + let mut pending = std::collections::VecDeque::new(); + let mut active = Vec::new(); + let mut door = state.coalesce_door(); + let now = std::time::Instant::now(); + let (incumbent, mut incumbent_rx) = walk_request(ids(40, 1), 64); + pending.push_back((incumbent, pegainfer_frontend::engine::KvPrefix::none())); + assert!( + state.intake_turn(&mut door, &mut pending, &mut active, now), + "closed-wait setup: an idle roster opens" + ); + assert_eq!(active.len(), 1, "closed-wait setup: the roster is live"); + let mut incumbent_tokens = 0; + while let Ok((_, event)) = incumbent_rx.try_recv() { + if matches!(event, TokenEvent::Token { .. }) { + incumbent_tokens += 1; + } + } + + let (second, mut second_rx) = walk_request(ids(40, 2), 4); + let (third, mut third_rx) = walk_request(ids(40, 3), 4); + pending.push_back((second, pegainfer_frontend::engine::KvPrefix::none())); + pending.push_back((third, pegainfer_frontend::engine::KvPrefix::none())); + assert!( + !state.intake_turn(&mut door, &mut pending, &mut active, now), + "closed wait: two arrivals stay behind the door" + ); + assert!( + second_rx.try_recv().is_err(), + "closed wait: the second arrival is not scheduled" + ); + assert!( + third_rx.try_recv().is_err(), + "closed wait: the third arrival is not scheduled" + ); + // The decode pipeline emits one step behind, so give the token a + // bounded number of rounds to surface. + let before_closed_decode = incumbent_tokens; + for _ in 0..3 { + state.decode_round(&mut active); + while let Ok((_, event)) = incumbent_rx.try_recv() { + if matches!(event, TokenEvent::Token { .. }) { + incumbent_tokens += 1; + } + } + if incumbent_tokens > before_closed_decode { + break; + } + } + assert!( + incumbent_tokens > before_closed_decode, + "closed wait: decode advances the incumbent between intake turns" + ); + + let (fourth, mut fourth_rx) = walk_request(ids(40, 4), 4); + pending.push_back((fourth, pegainfer_frontend::engine::KvPrefix::none())); + assert!( + state.intake_turn(&mut door, &mut pending, &mut active, now), + "cohort release: the third arrival opens the door" + ); + assert!( + pending.is_empty(), + "cohort release: one turn drains the queue" + ); + assert_eq!(active.len(), 4, "cohort release: every waiter is admitted"); + while active.len() > 1 { + state.decode_round(&mut active); + } + let second = drain(&mut second_rx, "second"); + let third = drain(&mut third_rx, "third"); + let fourth = drain(&mut fourth_rx, "fourth"); + assert_eq!( + (second.scheduled, third.scheduled, fourth.scheduled), + (1, 1, 1), + "cohort release: every waiter carries exactly one Scheduled event" + ); + assert_eq!( + active.len(), + 1, + "cohort release: the incumbent keeps its stream" + ); + + let timeout_start = now + std::time::Duration::from_secs(3); + let (timeout_a, mut timeout_a_rx) = walk_request(ids(40, 5), 4); + let (timeout_b, mut timeout_b_rx) = walk_request(ids(40, 6), 4); + pending.push_back((timeout_a, pegainfer_frontend::engine::KvPrefix::none())); + pending.push_back((timeout_b, pegainfer_frontend::engine::KvPrefix::none())); + assert!( + !state.intake_turn(&mut door, &mut pending, &mut active, timeout_start), + "timeout release: a fresh sub-cohort starts closed" + ); + let timeout_release = timeout_start + state.admit_coalesce.expect("door enabled"); + assert!( + state.intake_turn(&mut door, &mut pending, &mut active, timeout_release), + "timeout release: the window opens the next intake turn" + ); + assert!( + pending.is_empty(), + "timeout release: one turn drains the waiters" + ); + while active.len() > 1 { + state.decode_round(&mut active); + } + assert_eq!( + ( + drain(&mut timeout_a_rx, "timeout a").scheduled, + drain(&mut timeout_b_rx, "timeout b").scheduled, + ), + (1, 1), + "timeout release: every waiter carries one Scheduled event" + ); + + // The injected clock stays monotonic past the timeout release. + let free_slot_now = timeout_release + std::time::Duration::from_secs(1); + let (short, mut short_rx) = walk_request(ids(40, 7), 2); + let (long_a, mut long_a_rx) = walk_request(ids(40, 8), 8); + let (long_b, mut long_b_rx) = walk_request(ids(40, 9), 8); + pending.push_back((short, pegainfer_frontend::engine::KvPrefix::none())); + pending.push_back((long_a, pegainfer_frontend::engine::KvPrefix::none())); + pending.push_back((long_b, pegainfer_frontend::engine::KvPrefix::none())); + assert!( + state.intake_turn(&mut door, &mut pending, &mut active, free_slot_now), + "one free slot setup: a full cohort opens" + ); + assert_eq!(active.len(), 4, "one free slot setup: the roster is full"); + let (replacement, mut replacement_rx) = walk_request(ids(40, 10), 4); + pending.push_back((replacement, pegainfer_frontend::engine::KvPrefix::none())); + assert!( + state.intake_turn(&mut door, &mut pending, &mut active, free_slot_now), + "one free slot: the capacity-bounded floor opens the turn" + ); + assert_eq!( + pending.len(), + 1, + "one free slot: a full roster admits nothing" + ); + assert!( + replacement_rx.try_recv().is_err(), + "one free slot: the replacement is not scheduled while full" + ); + state.decode_round(&mut active); + assert_eq!( + active.len(), + 3, + "one free slot: exactly one incumbent finishes" + ); + assert!( + state.intake_turn(&mut door, &mut pending, &mut active, free_slot_now), + "one free slot: the next turn opens for the replacement" + ); + assert!( + pending.is_empty(), + "one free slot: the replacement is admitted" + ); + while !active.is_empty() { + state.decode_round(&mut active); + } + assert_eq!( + drain(&mut replacement_rx, "replacement").scheduled, + 1, + "one free slot: the replacement carries one Scheduled event" + ); + let short = drain(&mut short_rx, "short incumbent"); + let long_a = drain(&mut long_a_rx, "long incumbent a"); + let long_b = drain(&mut long_b_rx, "long incumbent b"); + let incumbent = drain(&mut incumbent_rx, "incumbent"); + assert_eq!( + incumbent_tokens + incumbent.tokens, + 64, + "one free slot: the original incumbent keeps its whole stream" + ); + assert_eq!( + (short.tokens, long_a.tokens, long_b.tokens), + (2, 8, 8), + "one free slot: every incumbent keeps its stream" + ); } /// The slots boundary, driven at the roster edge the engine loop owns diff --git a/scripts/gemma4_gates.sh b/scripts/gemma4_gates.sh index d54f4d1e6..51a5645c5 100755 --- a/scripts/gemma4_gates.sh +++ b/scripts/gemma4_gates.sh @@ -61,6 +61,7 @@ GATES_DENSE_AND_ROUTED=( # reads a weight, so that one needs the config and nothing else. GATES_SERVING_CONTRACT=( "gpu,ckpt engine::lane_tests::the_gathered_lifecycle_completes" + "gpu,ckpt engine::lane_tests::the_coalesce_door_releases_one_admission_burst" "gpu,ckpt,prompts engine::lane_tests::the_raised_ceiling_and_slots_hold_at_the_roster_edge" "gpu,ckpt,prompts engine::lane_tests::the_full_roster_keeps_its_pipeline_under_a_queue" "gpu,ckpt,prompts engine::lane_tests::an_idle_refill_drops_the_retired_fingerprint"