From a4c38414ad7cf84b06bc5a533dcac2beabed5fbf Mon Sep 17 00:00:00 2001 From: plotarmordev <299844489+plotarmordev@users.noreply.github.com> Date: Thu, 10 Sep 2026 14:13:15 +0800 Subject: [PATCH] feat(perf): admit explicit deadlines up to one hour --- crates/grill-perf/src/model.rs | 41 +++++++--- crates/grill-perf/tests/cli.rs | 140 +++++++++++++++++++++++++++++++++ docs/performance/CONTRACT.md | 14 +++- docs/performance/README.md | 8 +- 4 files changed, 189 insertions(+), 14 deletions(-) diff --git a/crates/grill-perf/src/model.rs b/crates/grill-perf/src/model.rs index 35e7dd9..dc49afa 100644 --- a/crates/grill-perf/src/model.rs +++ b/crates/grill-perf/src/model.rs @@ -194,14 +194,32 @@ impl Workload { return Err("exact output, required prefix evidence, and thinking controls need the explicit vllm-fixed-v1 request profile".into()); } let l = &self.limits; - if l.total_ms == 0 - || l.total_ms > 600_000 - || l.idle_ms == 0 - || l.idle_ms > l.total_ms - || !(1024..=8 * 1024 * 1024).contains(&l.response_bytes) - || l.wave_buffer_bytes > 512 * 1024 * 1024 - { - return Err("invalid deadline or buffering limits".into()); + if l.total_ms == 0 || l.total_ms > 3_600_000 { + return Err(format!( + "limits.total_ms={} must be in 1..=3600000", + l.total_ms + )); + } + if l.idle_ms == 0 { + return Err(format!("limits.idle_ms={} must be positive", l.idle_ms)); + } + if l.idle_ms > l.total_ms { + return Err(format!( + "limits.idle_ms={} must be <= limits.total_ms={}", + l.idle_ms, l.total_ms + )); + } + if !(1024..=8 * 1024 * 1024).contains(&l.response_bytes) { + return Err(format!( + "limits.response_bytes={} must be in 1024..=8388608", + l.response_bytes + )); + } + if l.wave_buffer_bytes > 512 * 1024 * 1024 { + return Err(format!( + "limits.wave_buffer_bytes={} must be <= 536870912", + l.wave_buffer_bytes + )); } let mut names = std::collections::HashSet::new(); let mut attempts = 0u64; @@ -233,10 +251,11 @@ impl Workload { } else { 6 * l.response_bytes + 512 * 1024 } + fill_bytes; - if per_request * cell.concurrency as usize > l.wave_buffer_bytes { + let required = per_request * cell.concurrency as usize; + if required > l.wave_buffer_bytes { return Err(format!( - "cell {} exceeds the admitted wave buffer bound", - cell.id + "cell {} requires limits.wave_buffer_bytes >= {required}, supplied {}; concurrency={}, stream={}, response_bytes={}, fill_bytes={fill_bytes}", + cell.id, l.wave_buffer_bytes, cell.concurrency, r.stream, l.response_bytes )); } let n = u64::from(cell.trials) + u64::from(cell.warmup_trials); diff --git a/crates/grill-perf/tests/cli.rs b/crates/grill-perf/tests/cli.rs index abea496..0bf55b6 100644 --- a/crates/grill-perf/tests/cli.rs +++ b/crates/grill-perf/tests/cli.rs @@ -1358,6 +1358,146 @@ fn timeout_settles_partial_evidence_without_starting_more_waves() { assert_eq!(server.count.load(Ordering::SeqCst), 1); } +#[test] +fn longer_deadline_declarations_complete_without_waiting_for_the_budget_and_load() { + let temp = Temp::new(); + let server = Server::new(normal); + for total in [600_000, 2_700_000, 3_600_000] { + let name = total.to_string(); + let mut w = workload(1, 0, 1); + w["limits"]["total_ms"] = json!(total); + w["limits"]["idle_ms"] = json!(total); + successful(&run(&temp, &server, &name, &w)); + assert_eq!(wave(&temp, &name, 0)["attempts"][0]["status"], "complete"); + successful( + &cli() + .arg("compare") + .arg(temp.path(&name)) + .arg(temp.path(&name)) + .arg("--json") + .output() + .unwrap(), + ); + } + let compared = cli() + .arg("compare") + .arg(temp.path("600000")) + .arg(temp.path("2700000")) + .arg("--json") + .output() + .unwrap(); + assert_eq!(compared.status.code(), Some(1)); +} + +#[test] +fn deadline_and_buffer_admission_names_the_failed_field_before_dispatch() { + let temp = Temp::new(); + let server = Server::new(normal); + for (name, field, supplied, related) in [ + ("total-zero", "total_ms", 0, "3600000"), + ("total-over", "total_ms", 3_600_001, "3600000"), + ("idle-zero", "idle_ms", 0, "positive"), + ("idle-over", "idle_ms", 3001, "limits.total_ms=3000"), + ("response-under", "response_bytes", 1023, "8388608"), + ( + "response-over", + "response_bytes", + 8 * 1024 * 1024 + 1, + "1024", + ), + ( + "wave-over", + "wave_buffer_bytes", + 512 * 1024 * 1024 + 1, + "536870912", + ), + ("wave-under", "wave_buffer_bytes", 1, "cell cell"), + ] { + let mut w = workload(1, 0, 1); + w["limits"][field] = json!(supplied); + let output = run(&temp, &server, name, &w); + assert_eq!(output.status.code(), Some(1)); + let error = String::from_utf8_lossy(&output.stderr); + assert!(error.contains(&format!("limits.{field}")), "{error}"); + assert!(error.contains(&supplied.to_string()), "{error}"); + assert!(error.contains(related), "{error}"); + assert!(!temp.path(name).exists()); + } + assert_eq!(server.count.load(Ordering::SeqCst), 0); +} + +#[test] +fn idle_still_expires_and_body_traffic_cannot_reset_total() { + for (name, total, idle, traffic, expected) in [ + ("idle", 3_600_000, 200, false, "idle_timeout"), + ("tie", 200, 200, false, "total_timeout"), + ("traffic", 1200, 400, true, "total_timeout"), + ] { + let temp = Temp::new(); + let (ready, releases) = std::sync::mpsc::sync_channel(1); + let server = Server::new(move |mut s, _, _| { + header(&mut s, "text/event-stream"); + let (release, wait) = std::sync::mpsc::sync_channel(1); + ready.send(release).unwrap(); + if traffic { + let deadline = Instant::now() + Duration::from_secs(10); + while Instant::now() < deadline { + if s.write_all(b": heartbeat\n\n").is_err() { + break; + } + match wait.recv_timeout(Duration::from_millis(20)) { + Ok(()) | Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => break, + Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {} + } + } + } else { + wait.recv_timeout(Duration::from_secs(10)).unwrap(); + } + }); + let mut w = workload(1, 0, 2); + w["limits"]["total_ms"] = json!(total); + w["limits"]["idle_ms"] = json!(idle); + let input = temp.path("input.json"); + fs::write(&input, serde_json::to_vec(&w).unwrap()).unwrap(); + let mut child = cli() + .arg("run") + .arg(input) + .args([ + "--endpoint", + &server.endpoint, + "--model", + "fixture-model", + "--local-http", + "--json", + "--out", + ]) + .arg(temp.path(name)) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + server.wait_for_request(&mut child); + let release = releases.recv_timeout(Duration::from_secs(3)).unwrap(); + let output = child.wait_with_output().unwrap(); + // The collector closes the connection at expiry; the heartbeat writer + // may already have observed that close instead of waiting for release. + let _ = release.send(()); + assert_eq!(output.status.code(), Some(2)); + let receipt = wave(&temp, name, 0); + let attempt = &receipt["attempts"][0]; + assert_eq!(attempt["status"], expected); + assert!(attempt["timing"]["first_generated_text_us"].is_null()); + if traffic { + assert!(attempt["timing"]["first_body_us"].is_number()); + assert!(attempt["timing"]["settle_us"].as_u64().unwrap() >= total as u64 * 1000); + let raw = fs::read(temp.path(name).join("wave-000000/response-0000.bin")).unwrap(); + assert!(raw.starts_with(b": heartbeat\n\n")); + } + assert_eq!(server.count.load(Ordering::SeqCst), 1); + assert!(!temp.path(name).join("wave-000001").exists()); + } +} + fn cancellation_case(signal: i32) { let temp = Temp::new(); let (ready, releases) = std::sync::mpsc::sync_channel(2); diff --git a/docs/performance/CONTRACT.md b/docs/performance/CONTRACT.md index 59c0f6e..5f554fb 100644 --- a/docs/performance/CONTRACT.md +++ b/docs/performance/CONTRACT.md @@ -53,7 +53,19 @@ IDs are unique short ASCII identifiers. Response limits are 1 KiB..8 MiB per request. SSE line and event caps are 256 KiB, and JSON nesting is limited to 64. Per-request total/idle deadlines are explicit, -positive, at most ten minutes, with idle no greater than total. +positive, with `total_ms` at most 3,600,000 (one hour) and idle no greater than +total. This finite ceiling is an admission policy, not a runtime guarantee. +Selected budgets remain explicit workload identity; existing declarations are +not increased automatically. Admission errors identify the failed deadline or +buffer field and relation, before dispatch. + +Both clocks start at dispatch. Only nonempty body chunks reset idle; headers, +empty chunks and server-side progress do not. SSE comments can reset idle without +generated text. Total never resets, and completion parsing/settlement must fit +it. Cancellation takes precedence over total expiry, then idle expiry. Raising +total alone cannot prevent a shorter idle timeout during quiet prefill. +Longer selected budgets can lengthen active-wave drain and cooperative pause +latency. No retry, hidden override or buffer expansion accompanies the ceiling. Admission checks the wave-buffer allowance against concurrency times `2*response_bytes + 6*256KiB + 512KiB + fill_bytes` for streaming, or diff --git a/docs/performance/README.md b/docs/performance/README.md index e025971..80e75f4 100644 --- a/docs/performance/README.md +++ b/docs/performance/README.md @@ -59,8 +59,12 @@ three measured trials per size) instead of one request per size; output is exactly 8 tokens rather than a cap of 8 so runs stay length matched; the salt is per attempt; thinking is disabled through `chat_template_kwargs.thinking` only, so check `first_generated_channel` is `answer` in the receipts. -Both deadlines are set to the ten-minute `total_ms` ceiling because prefill -sends no bytes before the first token; check larger sizes fit it. +Both selected deadlines remain ten minutes because prefill may send no bytes +before the first token; they are not the admission ceiling. The supported +`total_ms` ceiling is one hour, with positive `idle_ms <= total_ms`. Raise both +explicit declarations when a longer quiet prefill is intended; raising total +alone does not fix idle expiry. This is a bounded policy, not a server-runtime +guarantee, and longer budgets can lengthen cooperative pause/wave drain. For a smaller compatibility probe, use [`recipe-smoke.json`](../../crates/grill-perf/examples/recipe-smoke.json):