Something went wrong. Try again.
a tiny pet with an even tinier brain, living in your browser
Something went wrong. Try again.
33 kB · 791 lines
Rust
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792//! Opt-in real-model checks and warm serving measurements. No downloaded model//! or GPU is needed by ordinary cargo test. Run this test serially in release.//! Required env: ORT_DYLIB_PATH, SUBTOKENLING_MODEL_DIR, SUBTOKENLING_MODEL_ID,//! SUBTOKENLING_NATIVE_PROFILE=cpu|gpu|mixed, SUBTOKENLING_NATIVE_OUT (new JSON file).//! Optional: SUBTOKENLING_MODEL_DTYPE=q4|q8|fp32 (default q4),//! SUBTOKENLING_THREADS=4, SUBTOKENLING_GPU_DEVICE=0 (logical visible index),//! SUBTOKENLING_NATIVE_COUNTS=1,4,8,16, SUBTOKENLING_NATIVE_ROUNDS=3.//! Optional: SUBTOKENLING_NATIVE_STEADY_SECONDS=0 (0..300),//! SUBTOKENLING_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};use sha2::{Digest, Sha256};use std::{ collections::BTreeMap, env, path::Path, sync::{ Arc, Mutex, atomic::{AtomicBool, AtomicU32, Ordering}, }, time::Instant,};use tokenling_server::{ brain::{Brain, Generated, Generation, TOP_K, TOP_P}, host::{Generator, Host, Worker}, simulation::{self, HOUR, HostedPet, MIN}, store::Store, types::{PetState, Today}, world,};use tokio::io::{AsyncReadExt, AsyncWriteExt};
#[derive(Clone)]struct HttpClient;fn required(name: &str) -> String { env::var(name).unwrap_or_else(|_| panic!("set {name}"))}fn digest(s: &[u8]) -> String { format!("{:x}", Sha256::digest(s))}fn fixture(model: &str) -> PetState { let mut v: Value = serde_json::from_str(include_str!("../../test/fixtures/save-v1.json")).unwrap(); let now = simulation::now(); v["v"] = json!(2); v["hosting"] = json!({"mode":"local", "brainId":model,"timeZone":"UTC"}); v["created"] = json!(now - HOUR); v["pet"]["hatched"] = json!(now - 30.0 * MIN); v["pet"]["stage"] = json!("baby"); v["lastTick"] = json!(now); v["lastActive"] = json!(now); v["nextPoop"] = json!(now + 24.0 * HOUR); v["wokeAt"] = json!(now); v["grimeSince"] = json!(0); v["asleep"] = json!(false); v["sick"] = json!(false); v["poop"] = json!(0); v["stats"] = json!({"tummy":60,"fun":60,"energy":80,"clean":80}); v["offer"] = Value::Null; v["history"] = json!({}); let mut state: PetState = serde_json::from_value(v).unwrap(); state.today = Today { date: chrono::Utc::now().format("%Y-%m-%d").to_string(), log: tokenling_server::types::DayLog { log: vec![], weather: None, day: None, }, }; state.validate().unwrap(); state}
struct Measured { brain: Brain, backend: &'static str, zero: Instant, attempts: Arc<Mutex<Vec<Value>>>, temperature: Arc<AtomicU32>,}impl Generator for Measured { fn generate(&mut self, g: &Generation, cancelled: &AtomicBool) -> anyhow::Result<Generated> { 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 = 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(); let index = attempts.len(); attempts.push(json!({"backend":self.backend,"startMs":start,"endMs":null})); index }; let result = self .brain .generate_cancellable(&g, || cancelled.load(Ordering::Relaxed)); 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()),"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}) } }; result }}async fn api( _client: &HttpClient, base: &str, method: &str, path: &str, key: Option<&str>, input: Value,) -> Value { let address = base.strip_prefix("http://").unwrap(); let body = if method == "GET" || method == "DELETE" { String::new() } else { input.to_string() }; let auth = key.map_or(String::new(), |k| format!("Authorization: Bearer {k}\r\n")); let message = format!( "{method} {path} HTTP/1.1\r\nHost: {address}\r\nConnection: close\r\nContent-Type: application/json\r\nContent-Length: {}\r\n{auth}\r\n{body}", body.len() ); let bytes = tokio::time::timeout(std::time::Duration::from_secs(30), async { let mut socket = tokio::net::TcpStream::connect(address).await.unwrap(); socket.write_all(message.as_bytes()).await.unwrap(); let mut bytes = Vec::new(); socket.read_to_end(&mut bytes).await.unwrap(); bytes }) .await .expect("HTTP request timed out"); let boundary = bytes.windows(4).position(|w| w == b"\r\n\r\n").unwrap(); let headers = std::str::from_utf8(&bytes[..boundary]).unwrap(); let status: u16 = headers .lines() .next() .unwrap() .split_whitespace() .nth(1) .unwrap() .parse() .unwrap(); assert!( (200..300).contains(&status), "API returned {status}; diagnostic body omitted to protect keys/state" ); assert!( !headers .to_ascii_lowercase() .contains("transfer-encoding: chunked"), "test client expects Axum's fixed-length JSON response" ); serde_json::from_slice(&bytes[boundary + 4..]).unwrap()}fn percentile(values: &[f64], fraction: f64) -> f64 { let mut ordered = values.to_vec(); ordered.sort_by(f64::total_cmp); ordered[((ordered.len() as f64 * fraction).ceil() as usize).saturating_sub(1)]}
async fn steady_traffic( client: &HttpClient, base: &str, host: &Arc<Host>, state: &PetState, long_lines: &[String], attempts: &Arc<Mutex<Vec<Value>>>, 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 { 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!(Instant::now()<end+std::time::Duration::from_secs(30),"steady request stalled"); tokio::time::sleep(std::time::Duration::from_millis(5)).await; } if 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::<String, usize>::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::<Vec<_>>(),"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() { let model = required("SUBTOKENLING_MODEL_ID"); let model_dir = required("SUBTOKENLING_MODEL_DIR"); let dtype = env::var("SUBTOKENLING_MODEL_DTYPE") .unwrap_or_else(|_| "q4".into()) .parse() .unwrap(); let library = required("ORT_DYLIB_PATH"); let profile = required("SUBTOKENLING_NATIVE_PROFILE"); assert!( ["cpu", "gpu", "mixed"].contains(&profile.as_str()), "invalid profile" ); let output = required("SUBTOKENLING_NATIVE_OUT"); assert!( !Path::new(&output).exists(), "use a new output receipt path" ); let threads: usize = env::var("SUBTOKENLING_THREADS") .unwrap_or_else(|_| "4".into()) .parse() .unwrap(); let device: i32 = env::var("SUBTOKENLING_GPU_DEVICE") .unwrap_or_else(|_| "0".into()) .parse() .unwrap(); assert!(threads > 0 && device >= 0); let counts: Vec<usize> = env::var("SUBTOKENLING_NATIVE_COUNTS") .unwrap_or_else(|_| "1,4,8,16".into()) .split(',') .map(|s| s.parse().unwrap()) .collect(); let rounds: usize = env::var("SUBTOKENLING_NATIVE_ROUNDS") .unwrap_or_else(|_| "3".into()) .parse() .unwrap(); assert!(!counts.is_empty() && counts.iter().all(|n| (1..=64).contains(n)) && rounds > 0); let steady_seconds: u64 = env::var("SUBTOKENLING_NATIVE_STEADY_SECONDS") .unwrap_or_else(|_| "0".into()) .parse() .unwrap(); let steady_temperature: f32 = env::var("SUBTOKENLING_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") .commit(); let loaded = Instant::now(); let mut cpu = Brain::load(Path::new(&model_dir), dtype, threads, None).unwrap(); let cpu_load_ms = loaded.elapsed().as_secs_f64() * 1000.0; let loaded = Instant::now(); let mut gpu = if profile == "cpu" { None } else { Some(Brain::load(Path::new(&model_dir), dtype, 1, Some(device)).unwrap()) }; let gpu_load_ms = gpu .as_ref() .map(|_| loaded.elapsed().as_secs_f64() * 1000.0); let state = fixture(&model); let row = HostedPet::new(state.clone(), simulation::now()); let header = world::header("chat", &row.world(simulation::now(), "")); let mut cases = vec![ Generation { header: header.clone(), lines: vec!["Sam: hello little friend".into()], prefix: "Mochi:".into(), mode: "line".into(), max_new: 20, temperature: 0.0, top_k: TOP_K, top_p: TOP_P, ctx: 1024, }, Generation { header: header.clone(), lines: vec!["Sam: what time is it?".into()], prefix: "Mochi:".into(), mode: "line".into(), max_new: 20, temperature: 0.0, top_k: TOP_K, top_p: TOP_P, ctx: 1024, }, ]; let mut long = cases[0].clone(); while cpu.count(&cpu.fit(&long).unwrap()).unwrap() < 970 { long.lines .push("Sam: hello little friend, the flowers are bright and the room is cozy".into()); assert!(long.lines.len() < 100, "long fixture did not fit"); } cases.push(long); let mut direct = Vec::new(); for (case, g) in cases.iter().enumerate() { let reference = cpu.generate(g).unwrap(); assert!(reference.prompt_tokens + reference.generated_tokens <= 1024); let mut native_cpu = Vec::new(); let mut native_gpu = Vec::new(); for _ in 0..5 { let r = cpu.generate(g).unwrap(); assert_eq!(r.text, reference.text); native_cpu.push(serde_json::to_value(r).unwrap()); if let Some(gpu) = &mut gpu { let r = gpu.generate(g).unwrap(); assert_eq!( r.text, reference.text, "CPU/CUDA greedy text differs in fixture {case}" ); assert_eq!(r.reason, reference.reason); assert_eq!(r.prompt_tokens, reference.prompt_tokens); assert_eq!(r.generated_tokens, reference.generated_tokens); native_gpu.push(serde_json::to_value(r).unwrap()); } } // Keep fingerprints/timings, not raw prompt or generated pet text. for r in native_cpu.iter_mut().chain(native_gpu.iter_mut()) { r["promptHash"] = json!(digest(r["prompt"].as_str().unwrap().as_bytes())); r["outputHash"] = json!(digest(r["text"].as_str().unwrap().as_bytes())); r.as_object_mut().unwrap().remove("prompt"); r.as_object_mut().unwrap().remove("text"); } direct.push(json!({"case":case,"cpu":native_cpu,"gpu":native_gpu})); } let temporary = tempfile::tempdir().unwrap(); let attempts = Arc::new(Mutex::new(Vec::<Value>::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(), features: vec![], max_pets: 64, weather_seed: String::new(), cpu: Worker::new( "cpu", Measured { brain: cpu, backend: "cpu", zero, attempts: attempts.clone(), temperature: temperature.clone(), }, ), gpu: gpu.map(|brain| { Worker::new( "gpu", Measured { brain, 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); } let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let base = format!("http://{}", listener.local_addr().unwrap()); let app = host.clone().router(); let http = tokio::spawn(async move { axum::serve(listener, app).await.unwrap(); }); let scheduler = tokio::spawn(host.clone().run()); let client = HttpClient; let health = api(&client, &base, "GET", "/api/health", None, Value::Null).await; assert_eq!(health["gpu"].as_bool().unwrap(), profile != "cpu"); // Exercise an actual model-emitted lookup through HTTP and host continuation. let clock_id = uuid::Uuid::new_v4().to_string(); let clock_key = "a".repeat(64); let clock_device = uuid::Uuid::new_v4().to_string(); let mut clock_state = state.clone(); clock_state.chat = vec!["Sam: what time is it?".into()]; api( &client, &base, "POST", "/api/pets", Some(&clock_key), json!({"id":clock_id,"state":clock_state,"modelId":model}), ) .await; let before_clock = attempts.lock().unwrap().len(); api(&client,&base,"POST",&format!("/api/pets/{clock_id}/transfer"),Some(&clock_key),json!({"epoch":0,"operationId":uuid::Uuid::new_v4().to_string(),"mode":"remote","deviceId":clock_device,"state":clock_state})).await; let clock_deadline = Instant::now() + std::time::Duration::from_secs(30); loop { let snapshot = api( &client, &base, "GET", &format!("/api/pets/{clock_id}"), Some(&clock_key), Value::Null, ) .await; assert!( Instant::now() < clock_deadline, "clock continuation timed out" ); assert!(snapshot["lastError"].is_null()); if snapshot["thinking"] == false { assert!( snapshot["state"]["chat"] .as_array() .unwrap() .iter() .any(|s| s.as_str().unwrap().contains("[clock]{")), "host did not supply clock lookup result" ); break; } tokio::time::sleep(std::time::Duration::from_millis(5)).await; } let clock_attempts = attempts.lock().unwrap()[before_clock..].to_vec(); assert!(clock_attempts.len() >= 2 && clock_attempts.iter().any(|r| r["reason"] == "lookup")); api( &client, &base, "DELETE", &format!("/api/pets/{clock_id}"), Some(&clock_key), Value::Null, ) .await;
// Park a real, long inference while it is in flight. Cancellation must // discard its line without marking a healthy GPU unavailable. let cancel_id = uuid::Uuid::new_v4().to_string(); let cancel_key = "b".repeat(64); let cancel_device = uuid::Uuid::new_v4().to_string(); let mut cancel_state = state.clone(); cancel_state.chat = cases[2].lines.clone(); api( &client, &base, "POST", "/api/pets", Some(&cancel_key), json!({"id":cancel_id,"state":cancel_state,"modelId":model}), ) .await; let before_cancel = attempts.lock().unwrap().len(); api(&client,&base,"POST",&format!("/api/pets/{cancel_id}/transfer"),Some(&cancel_key),json!({"epoch":0,"operationId":uuid::Uuid::new_v4().to_string(),"mode":"remote","deviceId":cancel_device,"state":cancel_state})).await; let cancel_deadline = Instant::now() + std::time::Duration::from_secs(30); loop { let active = { let calls = attempts.lock().unwrap(); calls.len() > before_cancel && calls.last().unwrap()["endMs"].is_null() }; if active { break; } assert!( Instant::now() < cancel_deadline, "native inference did not start" ); tokio::time::sleep(std::time::Duration::from_millis(1)).await; } let parked=api(&client,&base,"POST",&format!("/api/pets/{cancel_id}/transfer"),Some(&cancel_key),json!({"epoch":1,"operationId":uuid::Uuid::new_v4().to_string(),"mode":"local","deviceId":cancel_device,"state":null})).await; while host.jobs.lock().unwrap().contains_key(&cancel_id) { assert!( Instant::now() < cancel_deadline, "cancelled inference did not release its lane" ); tokio::time::sleep(std::time::Duration::from_millis(5)).await; } let after = api( &client, &base, "GET", &format!("/api/pets/{cancel_id}"), Some(&cancel_key), Value::Null, ) .await; assert_eq!(after["authority"], "local"); assert_eq!( after["state"]["chat"], parked["state"]["chat"], "cancelled generation overwrote the local handoff" ); let cancellation_attempts = attempts.lock().unwrap()[before_cancel..].to_vec(); assert!(cancellation_attempts.iter().any(|r| r["cancelled"] == true)); if let Some(gpu) = &host.gpu { assert!( gpu.available.load(Ordering::Relaxed), "cancelling a turn must not disable CUDA" ); } api( &client, &base, "DELETE", &format!("/api/pets/{cancel_id}"), Some(&cancel_key), Value::Null, ) .await; let mut trials = Vec::new(); 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); loop { let snapshot = api(&client,&base,"GET",&format!("/api/pets/{id}"),Some(&key),Value::Null).await; assert!(Instant::now() < deadline, "native host turn timed out"); assert!(snapshot["lastError"].is_null(), "native host turn failed"); if !snapshot["thinking"].as_bool().unwrap() { assert!(snapshot["state"]["chat"] != serde_json::to_value(&state.chat).unwrap(), "native turn did not commit pet output"); break; } tokio::time::sleep(std::time::Duration::from_millis(5)).await; } 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(); if captured .iter() .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() .all(|v| v["ok"] == true && v["cancelled"] == false) ); let mut by_backend = BTreeMap::<String, usize>::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| { 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(), ) .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,"modelDtype":dtype,"modelHash":digest(&std::fs::read(Path::new(&model_dir).join("onnx").join(dtype.filename())).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", ) .unwrap();}