diff --git a/README.md b/README.md index fdddd58..76a8a90 100644 --- a/README.md +++ b/README.md @@ -89,6 +89,7 @@ Mochi: [clock]{21:40 night} it's so late!! [emote:sleepy] [want:snack] | `web/app/` | typed state, save format/storage, body simulation, care, animations, conversation, diary, model client, game, PWA and lifecycle controllers; see the module map in `docs/architecture.md` | | `web/app/hosting.ts`, `web/components/Hosting.tsx` | authenticated remote sync, both transfer directions, local recovery and the settings choice | | `server/` | Rust/Axum/SQLite host, native ONNX workers, simulation/prompt mirrors and ownership/API/parity tests | +| `server/src/routing.rs` | bounded timing estimates, queue completion planning and background aging | | `web/types.ts`, `web/runtime-types.ts` | save, scene and worker message types; the vendored runtime API boundary | | `web/brain.ts` | model worker; bundled separately as `brain.js` | | `web/scene.ts`, `web/sprites.ts`, `web/world.ts` | pixel LCD, sprite icons and the Python-compatible prompt format | diff --git a/docs/architecture.md b/docs/architecture.md index 4d6d6ed..de4b173 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -46,6 +46,7 @@ browser | `web/app/conversation.ts`, `web/app/text.ts` | React/idle turn scheduler, lookup continuations, stale replies and tag effects; text filtering and chat chip descriptions. | | `web/app/hosting.ts`, `web/components/Hosting.tsx` | Authenticated sync, staged local/remote transfers, stale-response rejection, recovery and mode controls. | | `server/src/` | Rust private API, SQLite ownership/receipts, background scheduler, body/prompt mirrors and native ONNX inference. | +| `server/src/routing.rs` | Numeric worker cost estimates, two-lane queue planning and age-based priority. | | `server/tests/`, `server/src/bin/` | ownership/API/body tests, Rust parity and native inference probe. | | `web/app/care.ts`, `web/app/diary.ts` | Hatching, care offers/answers and bedtime/missed diary pages. | | `web/app/animation.ts`, `web/app/canvas.ts`, `web/app/game.ts` | Animation state, disturbances, canvas attachment/cleanup and catch physics/input. | @@ -164,3 +165,28 @@ See `training.md` and `export-and-runtime.md`. Cache Storage, and the page asks for persistent storage so it is not evicted. - **Speed.** About 35 tokens/s on a 4-thread desktop WebAssembly session (see `export-and-runtime.md`); most turns are a few tokens. + +## Shared native/browser rules: design option + +A separate Rust rules crate could be used natively by the host and compiled +to WebAssembly for the browser. Prompt rendering, closed lists/lookups, body +math, tag effects and save validation are useful boundaries. Preact/canvas, +browser storage/lifecycle and browser inference remain in TypeScript; HTTP, +SQLite, queue routing and native inference remain in the host. Python still +needs independent training-format parity. This is a recommendation, not an +implemented migration or accepted change to the prompt contract. + +Start with prompt rendering and lookups, preserve byte-for-byte Python/native/ +WASM parity, then move simulation. Pass time/calendar information and random +draws explicitly so pure rules can run in both environments and replay tests. +Avoid sending the complete diary/transcript over the WASM boundary every +animation frame; return small body updates/effects and keep canvas frames in +JS. Save versions and unknown-field preservation must remain compatible. + +The browser bridge can use [wasm-bindgen](https://wasm-bindgen.github.io/wasm-bindgen/) +and [serde-wasm-bindgen](https://wasm-bindgen.github.io/wasm-bindgen/reference/arbitrary-data-with-serde.html). +The current Node/esbuild-only frontend build would gain a Rust/WASM build and +bindings generation step. Its new immutable asset must enter the app hash and +offline cache, and core initialization must finish before reading/running a +pet. The host crate as a whole is not the browser target: its SQLite, native +ONNX and network dependencies need to stay outside the shared rules crate. diff --git a/docs/testing.md b/docs/testing.md index 416c9ab..6bc6e14 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -200,6 +200,22 @@ explicit greedy sampling, not production stochastic or sustained capacity. See [the native-test recipe and variables](../server/README.md#checks-and-source-map). +## Queue-cost routing: CPU check receipt (2026-10-09) + +The routing candidate keeps nine whole-turn timing estimates per worker and +plans completion over queued work. Unit cases cover avoiding a late CPU tail, +larger queues using both lanes, unknown/busy shapes, unavailable GPU, and +background aging across context edits. Four pure routing tests and ten host +tests pass, including filtered diaries keeping a healthy GPU available. + +The real CPU test passed 1/4/8/16/32-pet short and varied-length bursts, +three rounds each, followed by a 120-second owner-traffic window with eight +pets at production temperature 0.85. All pets progressed and final handoffs +completed. Controlled bursts disable chatter so extra idle turns do not enter +the latency comparison. The steady phase permits ordinary diary/cancellation +behavior. This receipt validates CPU and rules; GPU routing and mixed steady +traffic are awaiting a shared-device window. It is not a serving-capacity claim. + ## Preview server ```sh diff --git a/server/README.md b/server/README.md index e4ca6f3..55ccd39 100644 --- a/server/README.md +++ b/server/README.md @@ -106,8 +106,12 @@ plays visual effects without applying care effects again. Catch starts only after acceptance; its result commits once with the game ID. An abandoned game expires without inventing a score. -CPU is resident. An optional native CUDA session prefers reactive turns, while -CPU prefers background work; a spare worker can handle either. Sessions run +CPU is resident. Before timing samples exist, an optional CUDA session prefers +reactive turns while CPU prefers background work. Once both workers have +observations for that request shape, the host compares predicted queue +completion times, including work already running. An idle CPU can wait when +the GPU is expected to finish a small queue sooner. Larger queues can still +use both lanes. Sessions run serially per worker. CUDA keeps KV tensors on the GPU through I/O binding; only logits return to the CPU for sampling. Each generation starts with an empty cache, and cancellation releases that request's bindings. TF32 is @@ -115,7 +119,20 @@ disabled so the GPU uses FP32 math for the expanded weights. Initialization failure leaves CPU serving available; a later GPU failure disables that worker and requeues its turn on CPU. -Queue-cost learning, batching and sustained serving capacity are not established. +Timing estimates are nine numeric cells per worker: reactive/idle/diary crossed +with three approximate prompt-size bands. Only successful, uncancelled whole +turns update the exponentially weighted mean (20% new sample). Prompt size +uses bounded UTF-8 byte counts divided by four, not exact tokenization. An +unseen shape keeps its original affinity; a busy unseen shape is treated as +having unknown completion time. No timings, record IDs or text are persisted +by the router. Estimates reset on process restart. + +The planner applies a 10% CPU improvement margin to limit timing noise. Queued +background work moves ahead of new reactive work after 30 seconds; running +generation is not preempted for this promotion. The wait survives context +edits and resets after a mode/epoch change or work-kind change. These constants +are generic implementation choices by reasoning, rather than measured service +latency guarantees. Batching and production sustained capacity are not established. This native ONNX path is distinct from the earlier PyTorch probe. The opt-in checks below measure its real private HTTP adapter and both workers. @@ -155,12 +172,21 @@ cargo test --release --locked --manifest-path server/Cargo.toml --test native -- test profile the CPU worker is present but disabled; production keeps CPU available. `TOKENLING_NATIVE_COUNTS` defaults to `1,4,8,16`, each a simultaneous burst whose requests queue behind available workers. `TOKENLING_NATIVE_ROUNDS` -defaults to `3`. An explicit greedy wrapper makes profile comparisons +defaults to `3`. Short and varied prompt-length bursts are tested separately. +An explicit greedy wrapper makes profile comparisons repeatable; production sampling remains at 0.85. Three prompt cases include a near-limit context, and each has five measured repetitions after a warmup. Checks include real clock continuation, cancellation during inference and complete-state parking. Receipts contain hashes/timings, never keys/saves. +`TOKENLING_NATIVE_STEADY_SECONDS` optionally runs eight sequential owner streams +after the controlled bursts (default `0`, maximum `300` seconds). +`TOKENLING_NATIVE_STEADY_TEMPERATURE` defaults to `0.85` for this phase. It +records per-pet progress, latency and generation counts and finishes current +requests before parking. Chatter is disabled for the controlled comparisons; +ordinary body/diary behavior still applies during the steady phase. Partial +numeric progress/failure receipts are written beside the requested output. + The separate [startup check](../test/host-startup.py) requires a CPU-only library and verifies actual server fallback, private access, CORS/no-store, a real CPU reply, transfer and deletion. Required flags are `--server`, @@ -172,6 +198,7 @@ stay outside Git. Both tests bind ephemeral loopback listeners. |---|---| | `src/main.rs` | Configuration, native workers, HTTP/CORS and startup | | `src/host.rs` | Private API, scheduler, generation/lookup loop and commit fencing | +| `src/routing.rs` | Bounded numeric timing estimates, queue plans and wait aging | | `src/store.rs` | SQLite state, key hashes, epochs and idempotent operations | | `src/types.rs`, `src/simulation.rs` | Saves, body, care, diary state and visual effects | | `src/world.rs` | Rust mirror of the prompt contract | diff --git a/server/src/host.rs b/server/src/host.rs index 604fda7..bed9b3d 100644 --- a/server/src/host.rs +++ b/server/src/host.rs @@ -1,6 +1,7 @@ //! Private capability API and background scheduler. The host owns all state. use crate::{ brain::{Brain, Generated, Generation, lookup_tail}, + routing::{self, Costs, Kind, Queue, Request as RouteRequest, Work}, simulation::{self, Command, HostedPet}, store::{AccessDenied, Conflict, Store}, types::{DiaryPage, PetState}, @@ -16,6 +17,7 @@ use axum::{ }; use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; +use std::time::Instant; use std::{ collections::HashMap, sync::{ @@ -38,6 +40,7 @@ pub struct Worker { pub engine: Arc>>, pub slot: Arc, pub available: Arc, + pub costs: Arc>, } impl Worker { pub fn new(name: &'static str, engine: impl Generator + 'static) -> Self { @@ -46,6 +49,7 @@ impl Worker { engine: Arc::new(Mutex::new(Box::new(engine))), slot: Arc::new(Semaphore::new(1)), available: Arc::new(AtomicBool::new(true)), + costs: Arc::new(Mutex::new(Costs::default())), } } } @@ -53,6 +57,9 @@ pub struct Lease { pub epoch: u64, pub context: u64, pub cancel: Arc, + pub lane: usize, + pub started: Instant, + pub estimated_ms: Option, } pub struct Host { pub db: Mutex, @@ -64,6 +71,7 @@ pub struct Host { pub gpu: Option, pub jobs: Mutex>, pub notify: Notify, + pub queue: Mutex, } impl Host { pub fn snapshot(&self, id: &str, row: HostedPet) -> Value { @@ -107,41 +115,76 @@ impl Host { } fn dispatch(self: &Arc) -> Result<()> { let t = simulation::now(); - let mut rows = self.db.lock().unwrap().advance(t, &self.weather_seed)?; - rows.sort_by(|a, b| { - b.1.pending - .is_some() - .cmp(&a.1.pending.is_some()) - .then_with(|| a.1.last_turn.total_cmp(&b.1.last_turn)) - }); + let rows = self.db.lock().unwrap().advance(t, &self.weather_seed)?; + let now = Instant::now(); + let mut ready = Vec::new(); + let mut remaining = [0.0; 2]; + let jobs = self.jobs.lock().unwrap(); + for job in jobs.values() { + remaining[job.lane] = job.estimated_ms.map_or(f64::INFINITY, |ms| { + (ms - job.started.elapsed().as_secs_f64() * 1000.0).max(ms * 0.1) + }); + } + let mut queue = self.queue.lock().unwrap(); for (id, row) in rows { if row.authority != "remote" || row.state.pet.is_none() || row.game.is_some() { continue; } - { - let jobs = self.jobs.lock().unwrap(); - if let Some(job) = jobs.get(&id) { - if row.epoch != job.epoch || row.context_revision != job.context { - job.cancel.store(true, Ordering::Relaxed); - } - continue; + if let Some(job) = jobs.get(&id) { + if row.epoch != job.epoch || row.context_revision != job.context { + job.cancel.store(true, Ordering::Relaxed); } + continue; } let Some(task) = choose_task(&row, t) else { continue; }; - let preferred: Vec<&Worker> = if matches!(task,Task::Chat(ref k) if k=="react") { - self.gpu.iter().chain(std::iter::once(&self.cpu)).collect() - } else { - std::iter::once(&self.cpu).chain(self.gpu.iter()).collect() + let work = task.work(&row, t, &self.weather_seed); + let age = queue.age_ms(&id, row.epoch, row.context_revision, work, now); + ready.push((id, row, task, RouteRequest { work, age_ms: age })); + } + queue + .waiting + .retain(|id, _| ready.iter().any(|r| &r.0 == id)); + drop(queue); + drop(jobs); + ready.sort_by(|a, b| { + routing::priority(a.3) + .cmp(&routing::priority(b.3)) + .then_with(|| b.3.age_ms.total_cmp(&a.3.age_ms)) + .then_with(|| a.1.last_turn.total_cmp(&b.1.last_turn)) + .then_with(|| a.0.cmp(&b.0)) + }); + let workers = [Some(&self.cpu), self.gpu.as_ref()]; + let available = workers.map(|w| { + w.is_some_and(|w| w.available.load(Ordering::Relaxed) && !w.engine.is_poisoned()) + }); + let free = workers.map(|w| w.is_some_and(|w| w.slot.available_permits() > 0)); + let cpu_costs = self.cpu.costs.lock().unwrap().clone(); + let gpu_costs = self + .gpu + .as_ref() + .map(|w| w.costs.lock().unwrap().clone()) + .unwrap_or_default(); + let requests: Vec<_> = ready.iter().map(|r| r.3).collect(); + let lanes = routing::plan( + [&cpu_costs, &gpu_costs], + remaining, + available, + free, + &requests, + ); + for ((id, row, task, request), lane) in ready.into_iter().zip(lanes) { + let Some(worker) = workers[lane].filter(|_| available[lane]) else { + continue; }; - let Some((worker, permit)) = preferred - .into_iter() - .filter(|w| w.available.load(Ordering::Relaxed)) - .find_map(|w| w.slot.clone().try_acquire_owned().ok().map(|p| (w, p))) - else { - break; + let Ok(permit) = worker.slot.clone().try_acquire_owned() else { + continue; }; + let work = request.work; + let estimate = worker.costs.lock().unwrap().estimate(work); + let costs = worker.costs.clone(); + let started = Instant::now(); let engine = worker.engine.clone(); let availability = worker.available.clone(); let backend = worker.name; @@ -166,8 +209,12 @@ impl Host { epoch: row.epoch, context: row.context_revision, cancel: cancel.clone(), + lane, + started, + estimated_ms: estimate, }, ); + self.queue.lock().unwrap().waiting.remove(&id); let host = self.clone(); let seed = self.weather_seed.clone(); tokio::spawn(async move { @@ -182,8 +229,18 @@ impl Host { .await; let outcome = result.unwrap_or_else(|_| Err(anyhow::anyhow!("brain worker stopped"))); - let gpu_failed = - backend == "gpu" && outcome.is_err() && !cancel.load(Ordering::Relaxed); + if outcome.is_ok() && !cancel.load(Ordering::Relaxed) { + costs + .lock() + .unwrap() + .observe(work, started.elapsed().as_secs_f64() * 1000.0); + } + let gpu_failed = backend == "gpu" + && outcome + .as_ref() + .err() + .is_some_and(|e| !e.is::()) + && !cancel.load(Ordering::Relaxed); if gpu_failed { availability.store(false, Ordering::Relaxed); } @@ -252,10 +309,48 @@ enum Task { Chat(String), Diary(String), } +impl Task { + fn work(&self, row: &HostedPet, t: f64, seed: &str) -> Work { + let (kind, bytes, cap) = match self { + Self::Chat(k) => ( + if k == "react" { + Kind::Reactive + } else { + Kind::Idle + }, + world::header("chat", &row.world(t, seed)).len() + + row.state.chat.iter().map(|s| s.len() + 1).sum::(), + if k == "react" { 956 } else { 468 }, + ), + Self::Diary(date) => { + let log = if date == &row.state.today.date { + Some(&row.state.today.log) + } else { + row.state.history.get(date) + }; + ( + Kind::Diary, + world::header("diary", &row.world(t, seed)).len() + + log.map_or(0, |d| d.log.iter().map(|s| s.len() + 1).sum::()), + 800, + ) + } + }; + Work::new(kind, bytes.div_ceil(4).min(cap)) + } +} enum Output { Chat(String), Diary(DiaryPage), } +#[derive(Debug)] +struct FilteredDiary; +impl std::fmt::Display for FilteredDiary { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "diary text was filtered") + } +} +impl std::error::Error for FilteredDiary {} fn choose_task(row: &HostedPet, t: f64) -> Option { if let Some(kind) = &row.pending { return Some(Task::Chat(kind.clone())); @@ -367,7 +462,9 @@ fn execute( } else { world::clean_text(&text) }; - ensure!(!text.is_empty(), "diary text was filtered"); + if text.is_empty() { + return Err(FilteredDiary.into()); + } Ok(Output::Diary(DiaryPage { date, day: st.day, diff --git a/server/src/lib.rs b/server/src/lib.rs index 52d1e06..6c6cfcf 100644 --- a/server/src/lib.rs +++ b/server/src/lib.rs @@ -1,5 +1,6 @@ pub mod brain; pub mod host; +pub mod routing; pub mod simulation; pub mod store; pub mod types; diff --git a/server/src/main.rs b/server/src/main.rs index 91559c5..25b9ad1 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -69,6 +69,7 @@ async fn main() -> Result<()> { gpu, jobs: Mutex::new(std::collections::HashMap::new()), notify: tokio::sync::Notify::new(), + queue: Mutex::new(Default::default()), }); let router = host .clone() diff --git a/server/src/routing.rs b/server/src/routing.rs new file mode 100644 index 0000000..02c9e8d --- /dev/null +++ b/server/src/routing.rs @@ -0,0 +1,287 @@ +//! Bounded, anonymous timing estimates and two-worker queue planning. +//! Costs retain only numeric aggregates. The waiting queue holds current record +//! IDs in memory; neither it nor timing estimates are saved or logged. +use std::collections::HashMap; +use std::time::Instant; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Kind { + Reactive, + Idle, + Diary, +} +impl Kind { + fn index(self) -> usize { + match self { + Self::Reactive => 0, + Self::Idle => 1, + Self::Diary => 2, + } + } +} +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct Work { + pub kind: Kind, + band: usize, +} +impl Work { + pub fn new(kind: Kind, estimated_tokens: usize) -> Self { + Self { + kind, + band: if estimated_tokens < 256 { + 0 + } else if estimated_tokens < 640 { + 1 + } else { + 2 + }, + } + } + fn index(self) -> usize { + self.kind.index() * 3 + self.band + } +} +#[derive(Clone, Copy, Default)] +struct Sample { + count: u64, + mean_ms: f64, +} +#[derive(Clone, Default)] +pub struct Costs { + samples: [Sample; 9], +} +impl Costs { + pub fn observe(&mut self, work: Work, ms: f64) { + if !ms.is_finite() || ms <= 0.0 { + return; + } + let sample = &mut self.samples[work.index()]; + sample.mean_ms = if sample.count == 0 { + ms + } else { + sample.mean_ms * 0.8 + ms * 0.2 + }; + sample.count = sample.count.saturating_add(1); + } + pub fn estimate(&self, work: Work) -> Option { + let exact = self.samples[work.index()]; + if exact.count > 0 { + return Some(exact.mean_ms); + } + // A different prompt band is deliberately not borrowed: CPU prefill + // scales much more sharply than CUDA. Unseen work retains affinity. + None + } + pub fn sample_count(&self, work: Work) -> u64 { + self.samples[work.index()].count + } +} + +pub struct Waiting { + pub epoch: u64, + pub context: u64, + pub work: Work, + pub since: Instant, +} +#[derive(Default)] +pub struct Queue { + pub waiting: HashMap, +} +impl Queue { + pub fn age_ms(&mut self, id: &str, epoch: u64, context: u64, work: Work, now: Instant) -> f64 { + let entry = self.waiting.entry(id.into()).or_insert(Waiting { + epoch, + context, + work, + since: now, + }); + if entry.epoch != epoch || entry.work.kind != work.kind { + *entry = Waiting { + epoch, + context, + work, + since: now, + }; + } + entry.context = context; + entry.work = work; + now.duration_since(entry.since).as_secs_f64() * 1000.0 + } +} +#[derive(Clone, Copy)] +pub struct Request { + pub work: Work, + pub age_ms: f64, +} +pub const BACKGROUND_MAX_WAIT_MS: f64 = 30000.0; + +pub fn priority(request: Request) -> u8 { + if request.work.kind != Kind::Reactive && request.age_ms >= BACKGROUND_MAX_WAIT_MS { + 0 + } else if request.work.kind == Kind::Reactive { + 1 + } else { + 2 + } +} + +// Returns a lane for every request, in the supplied priority/FIFO order. +// A lane can be busy: its planned request waits instead of spilling blindly. +pub fn plan( + costs: [&Costs; 2], + remaining: [f64; 2], + available: [bool; 2], + free: [bool; 2], + requests: &[Request], +) -> Vec { + let mut finish = remaining; + let mut cold_free = free; + requests + .iter() + .map(|request| { + let times = [ + costs[0].estimate(request.work), + costs[1].estimate(request.work), + ]; + let preferred = if request.work.kind == Kind::Reactive { + 1 + } else { + 0 + }; + let lane = if !available[0] { + 1 + } else if !available[1] { + 0 + } else if let [Some(cpu), Some(gpu)] = times { + // Prefer a clear improvement before sending work to a slower + // lane. The 10% margin limits oscillation from timing noise. + let cpu_end = finish[0] + cpu; + let gpu_end = finish[1] + gpu; + if cpu_end * 1.1 < gpu_end { 0 } else { 1 } + } else if cold_free[preferred] { + preferred + } else if cold_free[1 - preferred] { + 1 - preferred + } else { + preferred + }; + cold_free[lane] = false; + finish[lane] += times[lane].unwrap_or(remaining[lane].max(1.0)); + lane + }) + .collect() +} + +#[cfg(test)] +mod tests { + use super::*; + fn request() -> Request { + Request { + work: Work::new(Kind::Reactive, 100), + age_ms: 0.0, + } + } + fn costs() -> [Costs; 2] { + let mut c = [Costs::default(), Costs::default()]; + c[0].observe(request().work, 140.0); + c[1].observe(request().work, 30.0); + c + } + #[test] + fn small_queue_waits_for_gpu_and_large_queue_uses_both() { + let c = costs(); + assert_eq!( + plan( + [&c[0], &c[1]], + [0.0; 2], + [true; 2], + [true; 2], + &vec![request(); 4] + ), + vec![1; 4] + ); + let large = plan( + [&c[0], &c[1]], + [0.0; 2], + [true; 2], + [true; 2], + &vec![request(); 16], + ); + assert!(large.contains(&0) && large.contains(&1)); + assert_eq!( + plan( + [&c[0], &c[1]], + [0.0, 20.0], + [true; 2], + [true, false], + &vec![request(); 2] + ), + vec![1; 2] + ); + assert_eq!( + plan( + [&c[0], &c[1]], + [0.0, f64::INFINITY], + [true; 2], + [true, false], + &[request()] + ), + vec![0], + "a busy unseen GPU shape cannot promise an immediate finish" + ); + } + #[test] + fn unknown_shapes_keep_affinity_and_unavailable_gpu_uses_cpu() { + let c = costs(); + let requests = [ + request(), + Request { + work: Work::new(Kind::Diary, 700), + age_ms: 0.0, + }, + ]; + assert_eq!( + plan([&c[0], &c[1]], [0.0; 2], [true; 2], [true; 2], &requests), + vec![1, 0] + ); + assert_eq!( + plan( + [&c[0], &c[1]], + [0.0; 2], + [true, false], + [true, false], + &requests + ), + vec![0, 0] + ); + } + #[test] + fn estimates_are_finite_bounded_and_separate_by_work_shape() { + let mut c = Costs::default(); + let work = request().work; + c.observe(work, 100.0); + c.observe(work, 200.0); + c.observe(work, f64::NAN); + c.observe(work, -1.0); + assert_eq!(c.estimate(work), Some(120.0)); + assert_eq!(c.sample_count(work), 2); + assert_eq!(c.estimate(Work::new(Kind::Reactive, 900)), None); + assert_eq!(c.estimate(Work::new(Kind::Idle, 100)), None); + } + #[test] + fn background_ages_ahead_of_new_reactive_work_and_context_edits_keep_its_place() { + let work = Work::new(Kind::Diary, 100); + let start = Instant::now(); + let mut queue = Queue::default(); + assert_eq!(queue.age_ms("test-record", 1, 1, work, start), 0.0); + let later = start + std::time::Duration::from_secs(31); + let age = queue.age_ms("test-record", 1, 2, Work::new(Kind::Diary, 900), later); + assert_eq!(age, 31000.0); + assert!(priority(Request { work, age_ms: age }) < priority(request())); + assert_eq!( + queue.age_ms("test-record", 2, 3, work, later), + 0.0, + "a new executor starts a new queue wait" + ); + } +} diff --git a/server/tests/host.rs b/server/tests/host.rs index 178ac0d..7bd59f0 100644 --- a/server/tests/host.rs +++ b/server/tests/host.rs @@ -243,6 +243,21 @@ struct Scripted { calls: Arc, } struct BrokenGpu; +struct FilteredDiaryBrain; +impl Generator for FilteredDiaryBrain { + fn generate(&mut self, g: &Generation, _: &AtomicBool) -> anyhow::Result { + Ok(Generated { + text: tokenling_server::world::NOT_CUTE.as_str().into(), + prompt: g.header.clone() + &g.prefix, + reason: "end".into(), + prompt_tokens: 100, + prefill_ms: 1.0, + tps: 100.0, + generated_tokens: 5, + total_ms: 1.0, + }) + } +} impl Generator for BrokenGpu { fn generate(&mut self, _: &Generation, _: &AtomicBool) -> anyhow::Result { anyhow::bail!("GPU unavailable") @@ -273,6 +288,7 @@ async fn failed_gpu_requeues_the_turn_on_cpu_without_applying_it_twice() { gpu: Some(Worker::new("gpu", BrokenGpu)), jobs: Mutex::new(Default::default()), notify: tokio::sync::Notify::new(), + queue: Mutex::new(Default::default()), }); let scheduler = tokio::spawn(host.clone().run()); for _ in 0..100 { @@ -305,6 +321,71 @@ async fn failed_gpu_requeues_the_turn_on_cpu_without_applying_it_twice() { scheduler.abort(); } +#[tokio::test] +async fn a_filtered_diary_retries_its_pet_without_disabling_a_healthy_gpu() { + let temporary = tempfile::tempdir().unwrap(); + let mut db = Store::open(&temporary.path().join("pets.sqlite")).unwrap(); + let id = uuid::Uuid::new_v4().to_string(); + let key = "a".repeat(64); + let t = simulation::now(); + let mut state = pet(); + state.last_active = t - 60000.0; + state.history.insert( + (chrono::Utc::now() - chrono::Duration::days(1)) + .format("%Y-%m-%d") + .to_string(), + tokenling_server::types::DayLog { + log: vec!["had a cozy day".into(), "watched flowers".into()], + weather: None, + day: Some(1), + }, + ); + let mut row = HostedPet::new(state, t); + activate(&mut row); + row.pending = None; + row.next_idle = t + HOUR; + db.create(&id, &key, row, 64).unwrap(); + let host = Arc::new(Host { + db: Mutex::new(db), + model_id: "test-brain".into(), + features: vec![], + max_pets: 64, + weather_seed: String::new(), + cpu: Worker::new("cpu", FilteredDiaryBrain), + gpu: Some(Worker::new("gpu", FilteredDiaryBrain)), + jobs: Mutex::new(Default::default()), + notify: tokio::sync::Notify::new(), + queue: Mutex::new(Default::default()), + }); + host.cpu.available.store(false, Ordering::Relaxed); + let scheduler = tokio::spawn(host.clone().run()); + for _ in 0..100 { + if host + .db + .lock() + .unwrap() + .get(&id, &key) + .unwrap() + .last_error + .is_some() + { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + assert!( + host.db + .lock() + .unwrap() + .get(&id, &key) + .unwrap() + .last_error + .is_some() + ); + assert!(host.gpu.as_ref().unwrap().available.load(Ordering::Relaxed)); + scheduler.abort(); +} + #[test] fn forgetting_a_memory_cannot_delete_a_different_fact_after_eviction() { let t = simulation::now(); @@ -437,6 +518,7 @@ async fn mixed_workers_prefer_reactive_gpu_and_background_cpu_and_serialize_each )), jobs: Mutex::new(Default::default()), notify: tokio::sync::Notify::new(), + queue: Mutex::new(Default::default()), }); let scheduler = tokio::spawn(host.clone().run()); for _ in 0..100 { @@ -492,6 +574,88 @@ async fn mixed_workers_prefer_reactive_gpu_and_background_cpu_and_serialize_each assert!(host.gpu.as_ref().unwrap().available.load(Ordering::Relaxed)); scheduler.abort(); } + +#[tokio::test] +async fn measured_small_queue_waits_for_busy_gpu_instead_of_creating_a_cpu_tail() { + use tokenling_server::routing::{Kind, Work}; + let temporary = tempfile::tempdir().unwrap(); + let mut db = Store::open(&temporary.path().join("pets.sqlite")).unwrap(); + for name in ["Mochi", "Bean", "Pip"] { + let mut state = pet(); + state.pet.as_mut().unwrap().name = name.into(); + let mut row = HostedPet::new(state, simulation::now()); + activate(&mut row); + db.create(&uuid::Uuid::new_v4().to_string(), &"a".repeat(64), row, 64) + .unwrap(); + } + let started = Arc::new(Mutex::new(Vec::new())); + let released = Arc::new(AtomicBool::new(false)); + let host = Arc::new(Host { + db: Mutex::new(db), + model_id: "test-brain".into(), + features: vec![], + max_pets: 64, + weather_seed: String::new(), + cpu: Worker::new( + "cpu", + Gated { + backend: "cpu", + started: started.clone(), + released: released.clone(), + }, + ), + gpu: Some(Worker::new( + "gpu", + Gated { + backend: "gpu", + started: started.clone(), + released: released.clone(), + }, + )), + jobs: Mutex::new(Default::default()), + notify: tokio::sync::Notify::new(), + queue: Mutex::new(Default::default()), + }); + let work = Work::new(Kind::Reactive, 100); + host.cpu.costs.lock().unwrap().observe(work, 140.0); + host.gpu + .as_ref() + .unwrap() + .costs + .lock() + .unwrap() + .observe(work, 30.0); + let scheduler = tokio::spawn(host.clone().run()); + for _ in 0..100 { + if started.lock().unwrap().len() == 1 { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + tokio::time::sleep(std::time::Duration::from_millis(25)).await; + assert_eq!(started.lock().unwrap().len(), 1); + assert_eq!(started.lock().unwrap()[0].0, "gpu"); + assert_eq!( + host.cpu.slot.available_permits(), + 1, + "CPU stays free for this measured small queue" + ); + released.store(true, Ordering::Relaxed); + for _ in 0..100 { + if started.lock().unwrap().len() == 3 { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + assert_eq!(started.lock().unwrap().len(), 3); + assert!(started.lock().unwrap().iter().all(|r| r.0 == "gpu")); + assert_eq!( + host.cpu.costs.lock().unwrap().sample_count(work), + 1, + "seeded timing must not claim an unrun CPU turn" + ); + scheduler.abort(); +} async fn request( app: axum::Router, method: &str, @@ -537,6 +701,7 @@ async fn private_api_runs_background_turns_and_supports_both_transfer_directions gpu: None, jobs: Mutex::new(Default::default()), notify: tokio::sync::Notify::new(), + queue: Mutex::new(Default::default()), }); let app = host.clone().router(); let id = uuid::Uuid::new_v4().to_string(); diff --git a/server/tests/native.rs b/server/tests/native.rs index ef6f5b7..b5c7040 100644 --- a/server/tests/native.rs +++ b/server/tests/native.rs @@ -4,6 +4,8 @@ //! TOKENLING_NATIVE_PROFILE=cpu|gpu|mixed, TOKENLING_NATIVE_OUT (new JSON file). //! Optional: TOKENLING_THREADS=4, TOKENLING_GPU_DEVICE=0 (logical visible index), //! TOKENLING_NATIVE_COUNTS=1,4,8,16, TOKENLING_NATIVE_ROUNDS=3. +//! Optional: TOKENLING_NATIVE_STEADY_SECONDS=0 (0..300), +//! TOKENLING_NATIVE_STEADY_TEMPERATURE=0.85 (0..2). //! GPU profiles require an admitted allocation. The receipt contains only //! generic fixture hashes and aggregate timings, never access keys or saves. use serde_json::{Value, json}; @@ -14,7 +16,7 @@ use std::{ path::Path, sync::{ Arc, Mutex, - atomic::{AtomicBool, Ordering}, + atomic::{AtomicBool, AtomicU32, Ordering}, }, time::Instant, }; @@ -74,13 +76,14 @@ struct Measured { backend: &'static str, zero: Instant, attempts: Arc>>, + temperature: Arc, } impl Generator for Measured { fn generate(&mut self, g: &Generation, cancelled: &AtomicBool) -> anyhow::Result { let mut g = g.clone(); // Hold sampling constant for serving comparisons. This is an explicit // benchmark wrapper; the real host keeps its normal 0.85 temperature. - g.temperature = 0.0; + g.temperature = f32::from_bits(self.temperature.load(Ordering::Relaxed)); let start = self.zero.elapsed().as_secs_f64() * 1000.0; let index = { let mut attempts = self.attempts.lock().unwrap(); @@ -94,7 +97,7 @@ impl Generator for Measured { let end = self.zero.elapsed().as_secs_f64() * 1000.0; self.attempts.lock().unwrap()[index] = match &result { Ok(r) => { - json!({"backend":self.backend,"startMs":start,"endMs":end,"promptTokens":r.prompt_tokens,"generatedTokens":r.generated_tokens,"prefillMs":r.prefill_ms,"totalMs":r.total_ms,"reason":r.reason,"promptHash":digest(r.prompt.as_bytes()),"outputHash":digest(r.text.as_bytes()),"cancelled":cancelled.load(Ordering::Relaxed),"ok":true}) + json!({"backend":self.backend,"startMs":start,"endMs":end,"promptTokens":r.prompt_tokens,"generatedTokens":r.generated_tokens,"prefillMs":r.prefill_ms,"totalMs":r.total_ms,"reason":r.reason,"promptHash":digest(r.prompt.as_bytes()),"outputHash":digest(r.text.as_bytes()),"temperature":g.temperature,"cancelled":cancelled.load(Ordering::Relaxed),"ok":true}) } Err(_) => { json!({"backend":self.backend,"startMs":start,"endMs":end,"cancelled":cancelled.load(Ordering::Relaxed),"ok":false}) @@ -160,6 +163,137 @@ fn percentile(values: &[f64], fraction: f64) -> f64 { ordered[((ordered.len() as f64 * fraction).ceil() as usize).saturating_sub(1)] } +async fn steady_traffic( + client: &HttpClient, + base: &str, + host: &Arc, + state: &PetState, + long_lines: &[String], + attempts: &Arc>>, + seconds: u64, +) -> Value { + let mut pets = Vec::new(); + for index in 0..8 { + let id = uuid::Uuid::new_v4().to_string(); + let key = format!( + "{}{}", + uuid::Uuid::new_v4().simple(), + uuid::Uuid::new_v4().simple() + ); + let device = uuid::Uuid::new_v4().to_string(); + let mut st = state.clone(); + st.settings.chatter = false; + match index % 4 { + 2 => st.chat = long_lines.iter().take(12).cloned().collect(), + 3 => st.chat = long_lines.to_vec(), + _ => {} + } + api( + client, + base, + "POST", + "/api/pets", + Some(&key), + json!({"id":id,"state":st,"modelId":host.model_id}), + ) + .await; + api(client,base,"POST",&format!("/api/pets/{id}/transfer"),Some(&key),json!({"epoch":0,"operationId":uuid::Uuid::new_v4().to_string(),"mode":"remote","deviceId":device,"state":st})).await; + pets.push((id, key, device)); + } + let start = Instant::now(); + let end = start + std::time::Duration::from_secs(seconds); + let before = attempts.lock().unwrap().len(); + let mut writers = tokio::task::JoinSet::new(); + for (id, key, _) in &pets { + let (id, key, base, client) = (id.clone(), key.clone(), base.to_owned(), client.clone()); + writers.spawn(async move { + let mut latencies=Vec::new(); + let mut turn=0; + while Instant::now()=end {break;} + let began=Instant::now(); + let text=match turn%3 {0=>"hello little friend",1=>"what time is it?",_=>"how are you feeling today?"}; + api(&client,&base,"POST",&format!("/api/pets/{id}/commands"),Some(&key),json!({"epoch":1,"operationId":uuid::Uuid::new_v4().to_string(),"command":{"type":"message","text":text}})).await; + loop { + let s=api(&client,&base,"GET",&format!("/api/pets/{id}"),Some(&key),Value::Null).await; + assert!(s["lastError"].is_null(),"steady inference failed"); + if s["thinking"]==false {break;} + assert!(began.elapsed().as_secs()<30,"steady request stalled"); + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + latencies.push(began.elapsed().as_secs_f64()*1000.0); + turn+=1; + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + } + latencies + }); + } + let mut per_pet = Vec::new(); + while let Some(r) = writers.join_next().await { + per_pet.push(r.unwrap()); + } + let wall_ms = start.elapsed().as_secs_f64() * 1000.0; + let calls = attempts.lock().unwrap()[before..].to_vec(); + assert!( + calls + .iter() + .all(|r| r["ok"] == true || r["cancelled"] == true), + "an uncancelled steady generation failed" + ); + let latencies: Vec<_> = per_pet.iter().flatten().copied().collect(); + assert!( + per_pet.iter().all(|v| !v.is_empty()), + "a pet was starved during sustained traffic" + ); + let mut by_backend = BTreeMap::::new(); + let mut tokens = 0u64; + for r in &calls { + *by_backend + .entry(r["backend"].as_str().unwrap().to_owned()) + .or_default() += 1; + tokens += r["generatedTokens"].as_u64().unwrap(); + } + // Allow a diary already selected during the final drain to finish before + // parking; each pet remains under the ordinary one-generation fence. + for (id, key, device) in pets { + loop { + let s = api( + client, + base, + "GET", + &format!("/api/pets/{id}"), + Some(&key), + Value::Null, + ) + .await; + assert!(s["lastError"].is_null()); + if s["thinking"] == false { + break; + } + assert!(start.elapsed().as_secs() < seconds + 60); + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + api(client,base,"POST",&format!("/api/pets/{id}/transfer"),Some(&key),json!({"epoch":1,"operationId":uuid::Uuid::new_v4().to_string(),"mode":"local","deviceId":device,"state":null})).await; + api( + client, + base, + "DELETE", + &format!("/api/pets/{id}"), + Some(&key), + Value::Null, + ) + .await; + } + json!({"seconds":seconds,"wallMs":wall_ms,"completedTurns":latencies.len(),"perPetTurns":per_pet.iter().map(|v|v.len()).collect::>(),"p50Ms":percentile(&latencies,0.5),"p95Ms":percentile(&latencies,0.95),"maxMs":latencies.iter().copied().fold(0.0,f64::max),"generatedTokens":tokens,"outputTokensPerWallSecond":tokens as f64/(wall_ms/1000.0),"callsByBackend":by_backend,"attempts":calls}) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] #[ignore = "requires explicit native model/runtime and a reserved GPU for gpu/mixed"] async fn real_brain_and_concurrent_private_http() { @@ -195,6 +329,15 @@ async fn real_brain_and_concurrent_private_http() { .parse() .unwrap(); assert!(!counts.is_empty() && counts.iter().all(|n| (1..=64).contains(n)) && rounds > 0); + let steady_seconds: u64 = env::var("TOKENLING_NATIVE_STEADY_SECONDS") + .unwrap_or_else(|_| "0".into()) + .parse() + .unwrap(); + let steady_temperature: f32 = env::var("TOKENLING_NATIVE_STEADY_TEMPERATURE") + .unwrap_or_else(|_| "0.85".into()) + .parse() + .unwrap(); + assert!(steady_seconds <= 300 && (0.0..=2.0).contains(&steady_temperature)); ort::init_from(&library) .unwrap() .with_name("tokenling-native-test") @@ -275,6 +418,7 @@ async fn real_brain_and_concurrent_private_http() { let temporary = tempfile::tempdir().unwrap(); let attempts = Arc::new(Mutex::new(Vec::::new())); let zero = Instant::now(); + let temperature = Arc::new(AtomicU32::new(0.0f32.to_bits())); let host = Arc::new(Host { db: Mutex::new(Store::open(&temporary.path().join("pets.sqlite")).unwrap()), model_id: model.clone(), @@ -288,6 +432,7 @@ async fn real_brain_and_concurrent_private_http() { backend: "cpu", zero, attempts: attempts.clone(), + temperature: temperature.clone(), }, ), gpu: gpu.map(|brain| { @@ -298,11 +443,13 @@ async fn real_brain_and_concurrent_private_http() { backend: "gpu", zero, attempts: attempts.clone(), + temperature: temperature.clone(), }, ) }), jobs: Mutex::new(Default::default()), notify: tokio::sync::Notify::new(), + queue: Mutex::new(Default::default()), }); if profile == "gpu" { host.cpu.available.store(false, Ordering::Relaxed); @@ -448,42 +595,53 @@ async fn real_brain_and_concurrent_private_http() { ) .await; let mut trials = Vec::new(); - for n in counts { - for round in 0..rounds { - let mut pets = Vec::new(); - // Prepare all records inactive before scheduling a matched burst. - for _ in 0..n { - let id = uuid::Uuid::new_v4().to_string(); - let key = format!( - "{}{}", - uuid::Uuid::new_v4().simple(), - uuid::Uuid::new_v4().simple() - ); - let device = uuid::Uuid::new_v4().to_string(); - api( - &client, - &base, - "POST", - "/api/pets", - Some(&key), - json!({"id":id,"state":state,"modelId":model}), - ) - .await; - pets.push((id, key, device)); - } - let baseline = attempts.lock().unwrap().len(); - let start = Instant::now(); - let mut tasks = tokio::task::JoinSet::new(); - for (id, key, device) in &pets { - let (id, key, device, client, base, state) = ( - id.clone(), - key.clone(), - device.clone(), - client.clone(), - base.clone(), - state.clone(), - ); - tasks.spawn(async move { + for workload in ["short", "varied"] { + for n in &counts { + let n = *n; + for round in 0..rounds { + let mut pets = Vec::new(); + // Prepare all records inactive before scheduling a matched burst. + for index in 0..n { + let mut pet_state = state.clone(); + pet_state.settings.chatter = false; + if workload == "varied" { + match index % 4 { + 2 => pet_state.chat = cases[2].lines.iter().take(12).cloned().collect(), + 3 => pet_state.chat = cases[2].lines.clone(), + _ => {} + } + } + let id = uuid::Uuid::new_v4().to_string(); + let key = format!( + "{}{}", + uuid::Uuid::new_v4().simple(), + uuid::Uuid::new_v4().simple() + ); + let device = uuid::Uuid::new_v4().to_string(); + api( + &client, + &base, + "POST", + "/api/pets", + Some(&key), + json!({"id":id,"state":pet_state,"modelId":model}), + ) + .await; + pets.push((id, key, device, pet_state)); + } + let baseline = attempts.lock().unwrap().len(); + let start = Instant::now(); + let mut tasks = tokio::task::JoinSet::new(); + for (id, key, device, state) in &pets { + let (id, key, device, client, base, state) = ( + id.clone(), + key.clone(), + device.clone(), + client.clone(), + base.clone(), + state.clone(), + ); + tasks.spawn(async move { let began = Instant::now(); api(&client,&base,"POST",&format!("/api/pets/{id}/transfer"),Some(&key),json!({"epoch":0,"operationId":uuid::Uuid::new_v4().to_string(),"mode":"remote","deviceId":device,"state":state})).await; let deadline = began + std::time::Duration::from_secs(120); @@ -499,83 +657,123 @@ async fn real_brain_and_concurrent_private_http() { } began.elapsed().as_secs_f64()*1000.0 }); - } - let mut latencies = Vec::new(); - while let Some(result) = tasks.join_next().await { - latencies.push(result.unwrap()); - } - let wall_ms = start.elapsed().as_secs_f64() * 1000.0; - let captured = attempts.lock().unwrap()[baseline..].to_vec(); - assert!( - captured + } + let mut latencies = Vec::new(); + while let Some(result) = tasks.join_next().await { + latencies.push(result.unwrap()); + } + let wall_ms = start.elapsed().as_secs_f64() * 1000.0; + let captured = attempts.lock().unwrap()[baseline..].to_vec(); + if captured .iter() - .all(|v| v["ok"] == true && v["cancelled"] == false) - ); - let mut by_backend = BTreeMap::::new(); - let mut tokens = 0; - for attempt in &captured { - *by_backend - .entry(attempt["backend"].as_str().unwrap().to_owned()) - .or_default() += 1; - tokens += attempt["generatedTokens"].as_u64().unwrap(); - } - if profile == "cpu" { - assert!(!by_backend.contains_key("gpu")); - } - if profile == "gpu" { - assert!(!by_backend.contains_key("cpu")); - } - let overlaps = captured - .iter() - .filter(|a| a["backend"] == "cpu") - .flat_map(|a| { + .any(|v| v["ok"] != true || v["cancelled"] != false) + { + std::fs::write(format!("{output}.failed-trial.json"),serde_json::to_string_pretty(&json!({"profile":profile,"workload":workload,"pets":n,"round":round,"attempts":captured})).unwrap()).unwrap(); + } + assert!( captured .iter() - .filter(|b| b["backend"] == "gpu") - .map(move |b| (a, b)) - }) - .filter(|(a, b)| { - a["startMs"].as_f64().unwrap() < b["endMs"].as_f64().unwrap() - && b["startMs"].as_f64().unwrap() < a["endMs"].as_f64().unwrap() - }) - .count(); - if profile == "mixed" && n > 1 { - assert!(overlaps > 0, "CPU and GPU never ran concurrently"); - } - trials.push(json!({"pets":n,"round":round,"wallMs":wall_ms,"p50Ms":percentile(&latencies,0.5),"p95Ms":percentile(&latencies,0.95),"latenciesMs":latencies,"generatedTokens":tokens,"outputTokensPerWallSecond":tokens as f64/(wall_ms/1000.0),"callsByBackend":by_backend,"overlappingCpuGpuCalls":overlaps,"attempts":captured})); - for (id, key, device) in pets { - let before = api( - &client, - &base, - "GET", - &format!("/api/pets/{id}"), - Some(&key), - Value::Null, - ) - .await; - let parked = api(&client,&base,"POST",&format!("/api/pets/{id}/transfer"),Some(&key),json!({"epoch":before["epoch"],"operationId":uuid::Uuid::new_v4().to_string(),"mode":"local","deviceId":device,"state":null})).await; - assert_eq!(parked["authority"], "local"); - for field in ["pet", "owner", "memories", "shelf", "diary"] { - assert_eq!( - parked["state"][field], before["state"][field], - "handoff changed {field}" - ); + .all(|v| v["ok"] == true && v["cancelled"] == false) + ); + let mut by_backend = BTreeMap::::new(); + let mut tokens = 0; + for attempt in &captured { + *by_backend + .entry(attempt["backend"].as_str().unwrap().to_owned()) + .or_default() += 1; + tokens += attempt["generatedTokens"].as_u64().unwrap(); + } + if profile == "cpu" { + assert!(!by_backend.contains_key("gpu")); + } + if profile == "gpu" { + assert!(!by_backend.contains_key("cpu")); } - api( - &client, - &base, - "DELETE", - &format!("/api/pets/{id}"), - Some(&key), - Value::Null, + let overlaps = captured + .iter() + .filter(|a| a["backend"] == "cpu") + .flat_map(|a| { + captured + .iter() + .filter(|b| b["backend"] == "gpu") + .map(move |b| (a, b)) + }) + .filter(|(a, b)| { + a["startMs"].as_f64().unwrap() < b["endMs"].as_f64().unwrap() + && b["startMs"].as_f64().unwrap() < a["endMs"].as_f64().unwrap() + }) + .count(); + if profile == "mixed" && n >= 16 { + assert!(overlaps > 0, "CPU and GPU never ran concurrently"); + } + trials.push(json!({"workload":workload,"pets":n,"round":round,"wallMs":wall_ms,"p50Ms":percentile(&latencies,0.5),"p95Ms":percentile(&latencies,0.95),"latenciesMs":latencies,"generatedTokens":tokens,"outputTokensPerWallSecond":tokens as f64/(wall_ms/1000.0),"callsByBackend":by_backend,"overlappingCpuGpuCalls":overlaps,"attempts":captured})); + std::fs::write( + format!("{output}.progress.json"), + serde_json::to_string_pretty( + &json!({"profile":profile,"httpTrials":trials,"steadyStarted":false}), + ) + .unwrap(), ) - .await; + .unwrap(); + for (id, key, device, _) in pets { + let before = api( + &client, + &base, + "GET", + &format!("/api/pets/{id}"), + Some(&key), + Value::Null, + ) + .await; + let parked = api(&client,&base,"POST",&format!("/api/pets/{id}/transfer"),Some(&key),json!({"epoch":before["epoch"],"operationId":uuid::Uuid::new_v4().to_string(),"mode":"local","deviceId":device,"state":null})).await; + assert_eq!(parked["authority"], "local"); + for field in ["pet", "owner", "memories", "shelf", "diary"] { + assert_eq!( + parked["state"][field], before["state"][field], + "handoff changed {field}" + ); + } + api( + &client, + &base, + "DELETE", + &format!("/api/pets/{id}"), + Some(&key), + Value::Null, + ) + .await; + } } } } + let steady = if steady_seconds > 0 { + std::fs::write( + format!("{output}.progress.json"), + serde_json::to_string_pretty( + &json!({"profile":profile,"httpTrials":trials,"steadyStarted":true}), + ) + .unwrap(), + ) + .unwrap(); + temperature.store(steady_temperature.to_bits(), Ordering::Relaxed); + Some( + steady_traffic( + &client, + &base, + &host, + &state, + &cases[2].lines, + &attempts, + steady_seconds, + ) + .await, + ) + } else { + None + }; scheduler.abort(); http.abort(); - let receipt = json!({"profile":profile,"modelId":model,"modelHash":digest(&std::fs::read(Path::new(&model_dir).join("onnx/model_q4.onnx")).unwrap()),"libraryHash":digest(&std::fs::read(library).unwrap()),"cpuThreads":threads,"gpuDevice":if profile=="cpu"{None}else{Some(device)},"tf32":false,"gpuKvOnDevice":profile!="cpu","sampling":"greedy benchmark wrapper; production uses 0.85","cpuLoadMs":cpu_load_ms,"gpuLoadMs":gpu_load_ms,"direct":direct,"clockLookup":clock_attempts,"inflightTransferCancellation":cancellation_attempts,"httpTrials":trials}); + let receipt = json!({"profile":profile,"modelId":model,"modelHash":digest(&std::fs::read(Path::new(&model_dir).join("onnx/model_q4.onnx")).unwrap()),"libraryHash":digest(&std::fs::read(library).unwrap()),"cpuThreads":threads,"gpuDevice":if profile=="cpu"{None}else{Some(device)},"tf32":false,"gpuKvOnDevice":profile!="cpu","sampling":"greedy benchmark wrapper; production uses 0.85","cpuLoadMs":cpu_load_ms,"gpuLoadMs":gpu_load_ms,"direct":direct,"clockLookup":clock_attempts,"inflightTransferCancellation":cancellation_attempts,"httpTrials":trials,"steady":steady}); std::fs::write( output, serde_json::to_string_pretty(&receipt).unwrap() + "\n",