From e0314bd47026ea17e362a9cd9ab6a45528984877 Mon Sep 17 00:00:00 2001 From: juliankang4 <200967947+juliankang4@users.noreply.github.com> Date: Wed, 14 Oct 2026 23:47:59 +0900 Subject: [PATCH 1/7] serve: the prompt reader, the panic text, the base64 encoder and the models listing each have one owner prompt_ids reads with the tokenizer it is given, so a second vocabulary can read a prompt as /completion does; special puts that tokenizer's BOS first where it adds one, the rule the callers spelled with add_bos before. The panic message reader is api::panic_text, which the worker and the decision server call in place of their two copies. media::encode_base64 is the one base64 encoder: the media tests' copy and serve_client's are gone, and serve_client::base64 re-exports it for the gates. models::listing takes a list of Listed models, the servers each passing their one. No answer changes. --- crates/gpu-gates/src/serve_client.rs | 27 ++-------- crates/serve/src/api.rs | 65 ++++++++++++++--------- crates/serve/src/decide.rs | 31 ++++++----- crates/serve/src/media.rs | 58 ++++++++++---------- crates/serve/src/models.rs | 79 +++++++++++++++++++--------- crates/serve/src/worker.rs | 9 +--- 6 files changed, 146 insertions(+), 123 deletions(-) diff --git a/crates/gpu-gates/src/serve_client.rs b/crates/gpu-gates/src/serve_client.rs index 8a8d5396a..261288807 100644 --- a/crates/gpu-gates/src/serve_client.rs +++ b/crates/gpu-gates/src/serve_client.rs @@ -24,6 +24,9 @@ use serde_json::{Value, json}; use crate::record::{self, Kind, Log, Record}; use crate::{GateError, die_with_parent, verdict}; +/// An image file's bytes as the base64 a `data:` URL carries. +pub use serve::media::encode_base64 as base64; + /// The longest gap between two reads of the server's stderr in /// [`Served::address`]: the `server` record's wall ends at the read that found /// the line, so a gate's own `poll` (seconds) must not set its resolution. @@ -244,30 +247,6 @@ pub fn parse_ids(s: &str) -> Result, GateError> { .collect() } -/// The bytes as standard base64 with padding, the alphabet `serve::media`'s strict decoder -/// reads. -pub fn base64(bytes: &[u8]) -> String { - const A: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"; - let mut out = String::with_capacity(bytes.len().div_ceil(3) * 4); - for c in bytes.chunks(3) { - let b = [c[0], *c.get(1).unwrap_or(&0), *c.get(2).unwrap_or(&0)]; - let n = u32::from(b[0]) << 16 | u32::from(b[1]) << 8 | u32::from(b[2]); - out.push(A[(n >> 18) as usize & 63] as char); - out.push(A[(n >> 12) as usize & 63] as char); - out.push(if c.len() > 1 { - A[(n >> 6) as usize & 63] as char - } else { - '=' - }); - out.push(if c.len() > 2 { - A[n as usize & 63] as char - } else { - '=' - }); - } - out -} - /// One clause's line, `check : PASS|FAIL`, folded into `ok`. pub fn check(ok: &mut bool, name: &str, pass: bool) { println!("check {name}: {}", verdict(pass)); diff --git a/crates/serve/src/api.rs b/crates/serve/src/api.rs index b0496b506..f48731fe1 100644 --- a/crates/serve/src/api.rs +++ b/crates/serve/src/api.rs @@ -76,6 +76,7 @@ use crate::genloop::{self, Event, GenError, GenParams, Outcome, Slot, Timings, m use crate::glmxml::ArgTypes; use crate::http::{self, EventStream, Request}; use crate::media::{self, ExpandError, MediaError, Prompt, SharedMediaModel}; +use crate::models::Listed; use crate::qwenxml::ParamKinds; use crate::reasoning::{ReasoningFormat, ThinkEntry}; use crate::sampling; @@ -491,6 +492,14 @@ pub(crate) fn relock(m: &Mutex) -> MutexGuard<'_, T> { m.lock().unwrap_or_else(|e| e.into_inner()) } +/// A caught panic's message, as the failure it ends the server with names it. +pub(crate) fn panic_text(p: &(dyn std::any::Any + Send)) -> String { + p.downcast_ref::<&str>() + .map(|s| (*s).to_owned()) + .or_else(|| p.downcast_ref::().cloned()) + .unwrap_or_else(|| "a panic without a message".to_owned()) +} + /// The arrival tickets of the board ([`crate::sched::Board`]): drawn in /// arrival order, the turn passed on as each takes its slot. /// @@ -1779,10 +1788,12 @@ fn health(state: &State, req: &Request, w: &mut TcpStream) -> io::Result { fn models(state: &State) -> Value { crate::models::listing( - &state.alias, + vec![Listed { + id: &state.alias, + meta: json!({ "n_vocab": state.info.n_vocab, "n_ctx_train": state.info.ctx_max }), + max_model_len: Some(state.info.ctx_max), + }], state.start_unix, - json!({ "n_vocab": state.info.n_vocab, "n_ctx_train": state.info.ctx_max }), - Some(state.info.ctx_max), ) } @@ -2194,7 +2205,7 @@ fn tokenize(state: &State, b: &Map) -> Result { let t = &*state.tok; let ids = match b.get("content") { None => return Err(invalid("'content' is required")), - Some(v) => prompt_ids(state, "content", v, add_special && t.add_bos())?, + Some(v) => prompt_ids(t, "content", v, add_special)?, }; let tokens: Vec = if with_pieces { ids.iter() @@ -2211,12 +2222,12 @@ fn tokenize(state: &State, b: &Map) -> Result { fn detokenize(state: &State, b: &Map) -> Result { let ids = match b.get("tokens") { None => return Err(invalid("'tokens' is required")), - Some(v) => token_ids(state, v)?, + Some(v) => token_ids(&*state.tok, v)?, }; Ok(json!({ "content": state.tok.decode(&ids) })) } -fn token_ids(state: &State, v: &Value) -> Result, ApiError> { +fn token_ids(tok: &dyn Tokenizer, v: &Value) -> Result, ApiError> { let Value::Array(a) = v else { return Err(invalid("tokens must be an array of token ids")); }; @@ -2224,7 +2235,7 @@ fn token_ids(state: &State, v: &Value) -> Result, ApiError> { .map(|x| { x.as_u64() .and_then(|id| u32::try_from(id).ok()) - .filter(|&id| (id as usize) < state.info.n_vocab) + .filter(|&id| (id as usize) < tok.n_vocab()) .ok_or_else(|| invalid(format!("invalid token id {x}"))) }) .collect() @@ -3141,7 +3152,7 @@ fn completion_ids(state: &State, v: Option<&Value>) -> Result, ApiError match v { Some(o) if carries_media(o) => Err(no_media()), Some(v @ (Value::String(_) | Value::Array(_))) => { - prompt_ids(state, "prompt", v, state.tok.add_bos()) + prompt_ids(&*state.tok, "prompt", v, true) } _ => Err(invalid("'prompt' must be a string or an array of tokens")), } @@ -3151,29 +3162,31 @@ fn no_media() -> ApiError { invalid("Multimodal data provided, but model does not support multimodal requests.") } -/// The field `k`, a string or an array of strings and token ids, as token ids: -/// each string encoded, each id checked against the vocabulary. `bos` puts the -/// tokenizer's BOS first when it starts with text. Any other type is a 400 -/// naming `k` and the type it got. -fn prompt_ids(state: &State, k: &str, v: &Value, bos: bool) -> Result, ApiError> { - let e = &*state.tok; +/// The field `k`, a string or an array of strings and token ids, as `tok`'s token ids, as +/// llama-server's `tokenize_mixed` reads it: each string encoded, each id checked against the +/// vocabulary. `special` puts the vocabulary's BOS first ([`Tokenizer::add_bos`]) when the input +/// starts with text. Any other type is a 400 naming `k` and the type it got. +fn prompt_ids( + tok: &dyn Tokenizer, + k: &str, + v: &Value, + special: bool, +) -> Result, ApiError> { let mut ids = Vec::new(); - match v { - Value::String(s) => { - if bos { - ids.push(e.bos()); - } - ids.extend(e.encode(s)); + let text = |ids: &mut Vec, s: &str, special: bool| { + if special && tok.add_bos() { + ids.push(tok.bos()); } + ids.extend(tok.encode(s)); + }; + match v { + Value::String(s) => text(&mut ids, s, special), Value::Array(a) => { - if bos && a.first().is_some_and(Value::is_string) { - ids.push(e.bos()); - } - for x in a { + for (i, x) in a.iter().enumerate() { match x { - Value::String(s) => ids.extend(e.encode(s)), + Value::String(s) => text(&mut ids, s, special && i == 0), o if carries_media(o) => return Err(no_media()), - other => ids.extend(token_ids(state, &Value::Array(vec![other.clone()]))?), + other => ids.extend(token_ids(tok, &Value::Array(vec![other.clone()]))?), } } } diff --git a/crates/serve/src/decide.rs b/crates/serve/src/decide.rs index 59b1aa0f6..b018dbc8f 100644 --- a/crates/serve/src/decide.rs +++ b/crates/serve/src/decide.rs @@ -38,11 +38,12 @@ use serde_json::{Value, json}; use crate::api::{ Answers, End, Ended, JSON, accept, cors_preflight, engine_object, error_body, exit_shutdown, - fatal_health, install_signals, keep_alive, key_denied, relock, stopping, stopping_message, - wait_end, + fatal_health, install_signals, keep_alive, key_denied, panic_text, relock, stopping, + stopping_message, wait_end, }; use crate::flag::ApiKeys; use crate::http::{self, Request}; +use crate::models::Listed; use crate::{EngineFailure, EngineProps, FATAL_LINGER, ServeError}; /// One answered request. @@ -610,14 +611,7 @@ fn decide(s: &Shared, body: &str) -> Reply { Ok(Err(DecideError::Refused(m))) => Reply::error(400, "invalid_request_error", &m), Ok(Err(DecideError::NotSupported(m))) => Reply::error(501, "not_supported_error", &m), Ok(Err(DecideError::Engine(m))) => s.fail(m), - Err(p) => { - let what = p - .downcast_ref::<&str>() - .map(|s| (*s).to_owned()) - .or_else(|| p.downcast_ref::().cloned()) - .unwrap_or_else(|| "a panic without a message".to_owned()); - s.fail(format!("the decider panicked: {what}")) - } + Err(p) => s.fail(format!("the decider panicked: {}", panic_text(&*p))), } } @@ -660,10 +654,12 @@ impl DecideServer { routes: seated.routes, props: props_of(decider.props()).map_err(invalid)?, models: crate::models::listing( - &seated.name, + vec![Listed { + id: &seated.name, + meta: json!({ "n_ctx": seated.n_ctx }), + max_model_len: Some(seated.n_ctx), + }], created, - json!({ "n_ctx": seated.n_ctx }), - Some(seated.n_ctx), ), }; let listener = TcpListener::bind(addr)?; @@ -806,7 +802,14 @@ mod tests { fixed: Fixed { routes: ROUTES, props: json!({}), - models: crate::models::listing("m.gguf", 0, json!({}), None), + models: crate::models::listing( + vec![Listed { + id: "m.gguf", + meta: json!({}), + max_model_len: None, + }], + 0, + ), }, api_keys, engine: "the decider of m.gguf".to_owned(), diff --git a/crates/serve/src/media.rs b/crates/serve/src/media.rs index cfd0a3775..48d491f91 100644 --- a/crates/serve/src/media.rs +++ b/crates/serve/src/media.rs @@ -624,6 +624,27 @@ fn base64(text: &str) -> Result, MediaError> { Ok(out) } +/// `bytes` as standard base64 with padding (RFC 4648 §4), the text [`base64`] reads back. +#[must_use] +pub fn encode_base64(bytes: &[u8]) -> String { + const A: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"; + let mut out = String::with_capacity(bytes.len().div_ceil(3) * 4); + for c in bytes.chunks(3) { + let n = c + .iter() + .enumerate() + .fold(0u32, |n, (i, &b)| n | u32::from(b) << (16 - 8 * i)); + for i in 0..4 { + out.push(if i <= c.len() { + char::from(A[(n >> (18 - 6 * i) & 63) as usize]) + } else { + '=' + }); + } + } + out +} + /// At most the first 64 characters of a value an error names, so a refusal never echoes a payload. fn short(s: &str) -> String { match s.char_indices().nth(64) { @@ -641,8 +662,8 @@ mod tests { use super::{ ExpandError, FileKind, Held, ImageKey, MediaError, MediaModel, MediaSpan, NumberedMedia, - Part, PartJoin, Prepared, Rgb8, VisionError, common_prefix, expand_prompt, flatten, - keep_whole_spans, load, parse_part, prepare, + Part, PartJoin, Prepared, Rgb8, VisionError, common_prefix, encode_base64, expand_prompt, + flatten, keep_whole_spans, load, parse_part, prepare, }; /// A 2×1 RGB PNG of the colour (200, 100, 50). @@ -704,26 +725,6 @@ mod tests { format!("data:{mime};base64,{b64}") } - /// Standard base64 with padding (RFC 4648 §4), to build payloads. - fn encode(bytes: &[u8]) -> String { - const A: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"; - let mut s = String::new(); - for c in bytes.chunks(3) { - let n = c - .iter() - .enumerate() - .fold(0u32, |n, (i, &b)| n | u32::from(b) << (16 - 8 * i)); - for i in 0..4 { - s.push(if i <= c.len() { - char::from(A[(n >> (18 - 6 * i) & 63) as usize]) - } else { - '=' - }); - } - } - s - } - fn err(r: Result) -> MediaError { match r { Ok(v) => panic!("accepted: {v:?}"), @@ -1009,7 +1010,7 @@ mod tests { fn jpeg_and_its_jpg_alias_load() { let jpeg = [0xFF, 0xD8, 0xFF, 0xE0, 0, 16, b'J', b'F', b'I', b'F']; for mime in ["image/jpeg", "image/jpg"] { - let f = load(&data(mime, &encode(&jpeg))).unwrap(); + let f = load(&data(mime, &encode_base64(&jpeg))).unwrap(); assert_eq!( (f.kind, f.bytes.as_slice()), (FileKind::Jpeg, &jpeg[..]), @@ -1077,10 +1078,11 @@ mod tests { ("foobar", "Zm9vYmFy"), ] { assert_eq!(super::base64(b64).unwrap(), plain.as_bytes(), "{b64}"); + assert_eq!(encode_base64(plain.as_bytes()), b64, "{plain}"); } let all: Vec = (0..=255).collect(); for n in [254, 255, 256] { - assert_eq!(super::base64(&encode(&all[..n])).unwrap(), &all[..n]); + assert_eq!(super::base64(&encode_base64(&all[..n])).unwrap(), &all[..n]); } for (b64, want) in [ ("Zm9v\nYmFy", "char 10 at 4"), @@ -1117,7 +1119,7 @@ mod tests { } e => panic!("{e}"), } - match err(load(&data("image/png", &encode(&jpeg)))) { + match err(load(&data("image/png", &encode_base64(&jpeg)))) { MediaError::Magic { declared, found } => { assert_eq!((declared, found), (FileKind::Png, FileKind::Jpeg)); } @@ -1127,12 +1129,12 @@ mod tests { (&b"RIFF\x24\0\0\0WEBPVP8 "[..], FileKind::Webp), (b"GIF89a\x01\0\x01\0", FileKind::Gif), ] { - match err(load(&data("image/png", &encode(bytes)))) { + match err(load(&data("image/png", &encode_base64(bytes)))) { MediaError::Format(k) => assert_eq!(k, kind), e => panic!("{kind:?}: {e}"), } } - match err(load(&data("image/png", &encode(b"BM\x3a\0\0\0")))) { + match err(load(&data("image/png", &encode_base64(b"BM\x3a\0\0\0")))) { MediaError::Unknown(head) => assert_eq!(head, "424d3a000000"), e => panic!("{e}"), } @@ -1147,7 +1149,7 @@ mod tests { let patches = p.grid().unwrap(); assert_eq!((patches.plan.best_w, patches.plan.best_h), (2, 1)); let jpeg = [0xFF, 0xD8, 0xFF, 0xDB, 0, 4, 0, 0]; - let broken = load(&data("image/jpeg", &encode(&jpeg))).unwrap(); + let broken = load(&data("image/jpeg", &encode_base64(&jpeg))).unwrap(); assert!(matches!( err(prepare(&broken, &PLAIN)), MediaError::Decode(VisionError::Decode(_)) diff --git a/crates/serve/src/models.rs b/crates/serve/src/models.rs index f1fe1ca4e..59e87b0ab 100644 --- a/crates/serve/src/models.rs +++ b/crates/serve/src/models.rs @@ -1,27 +1,44 @@ -//! `GET /v1/models` (and `/models`): the one model a server serves, in llama.cpp's server's shape: +//! `GET /v1/models` (and `/models`): the models a server serves, in llama.cpp's server's shape: //! `data`, OpenAI's list, and `models` beside it (`tools/server/server-context.cpp` at `a4cb4c61`, //! `get_res_models`: its `name` and `model`, the entry's other fields there being placeholders). //! Every server of this crate answers through [`listing`]. use serde_json::{Value, json}; -/// The listing of the model `id`: `created` (unix seconds), `meta` as the server knows it, and -/// `max_model_len` when the server states one. -pub(crate) fn listing(id: &str, created: u64, meta: Value, max_model_len: Option) -> Value { - let mut entry = json!({ - "id": id, - "object": "model", - "created": created, - "owned_by": "bloomery", - "meta": meta, - }); - if let (Some(n), Value::Object(o)) = (max_model_len, &mut entry) { - o.insert("max_model_len".to_owned(), json!(n)); - } +/// One model of a listing: its id, `meta` as the server knows it, and `max_model_len` when the +/// server states one. +pub(crate) struct Listed<'a> { + pub id: &'a str, + pub meta: Value, + pub max_model_len: Option, +} + +/// The listing of `models`, in order, each `created` at the same unix second. +pub(crate) fn listing(models: Vec>, created: u64) -> Value { + let names: Vec = models + .iter() + .map(|m| json!({ "name": m.id, "model": m.id })) + .collect(); + let data: Vec = models + .into_iter() + .map(|m| { + let mut entry = json!({ + "id": m.id, + "object": "model", + "created": created, + "owned_by": "bloomery", + "meta": m.meta, + }); + if let (Some(n), Value::Object(o)) = (m.max_model_len, &mut entry) { + o.insert("max_model_len".to_owned(), json!(n)); + } + entry + }) + .collect(); json!({ - "models": [{ "name": id, "model": id }], + "models": names, "object": "list", - "data": [entry], + "data": data, }) } @@ -30,18 +47,32 @@ mod tests { use super::*; #[test] - fn both_lists_name_the_one_model() { - let v = listing("m.gguf", 7, json!({ "n_ctx": 16 }), Some(16)); + fn both_lists_name_every_model_in_order() { + let v = listing( + vec![ + Listed { + id: "m.gguf", + meta: json!({ "n_ctx": 16 }), + max_model_len: Some(16), + }, + Listed { + id: "e.gguf", + meta: json!({}), + max_model_len: None, + }, + ], + 7, + ); assert_eq!(v["data"][0]["id"], "m.gguf"); assert_eq!(v["data"][0]["max_model_len"], 16); + assert_eq!(v["data"][1]["id"], "e.gguf"); + assert!(v["data"][1].get("max_model_len").is_none()); assert_eq!( v["models"], - json!([{ "name": "m.gguf", "model": "m.gguf" }]) - ); - assert!( - listing("m", 7, json!({}), None)["data"][0] - .get("max_model_len") - .is_none() + json!([ + { "name": "m.gguf", "model": "m.gguf" }, + { "name": "e.gguf", "model": "e.gguf" }, + ]) ); } } diff --git a/crates/serve/src/worker.rs b/crates/serve/src/worker.rs index 80b7032d8..86176154f 100644 --- a/crates/serve/src/worker.rs +++ b/crates/serve/src/worker.rs @@ -48,7 +48,7 @@ use std::time::{Duration, Instant}; use serde_json::Value; -use crate::api::{Answers, End, EngineFailure, Stop, Waited, client_gone, relock}; +use crate::api::{Answers, End, EngineFailure, Stop, Waited, client_gone, panic_text, relock}; use crate::engine::{ Drafted, EngineError, Sampler, SamplerFactory, Saved, SlotPass, SlotRow, StateError, }; @@ -402,12 +402,7 @@ pub(crate) fn serve(slot: Slot, sh: Arc) { // and drops the engine it holds, which no stop waits for. sh.end_loop(); if let Err(p) = ran { - let what = p - .downcast_ref::<&str>() - .map(|s| (*s).to_owned()) - .or_else(|| p.downcast_ref::().cloned()) - .unwrap_or_else(|| "a panic without a message".to_owned()); - let error = format!("the engine thread panicked: {what}"); + let error = format!("the engine thread panicked: {}", panic_text(&*p)); *relock(&sh.fatal) = Some(error.clone()); drop(relock(&sh.board).kill(&error)); // The running requests' channels close: their HTTP threads answer From dc8398271c3b5536d402e6df6bdecf3ae7c59105 Mon Sep 17 00:00:00 2001 From: juliankang4 <200967947+juliankang4@users.noreply.github.com> Date: Wed, 14 Oct 2026 23:58:10 +0900 Subject: [PATCH 2/7] serve: the first engine failure is the one /health, the board and the exit name Shared::record_fatal adopts a failure only when none was adopted before: it sets the reason /health and every later request give, closes the waiting requests' and actions' channels on it, and hands it back for Shared::end_on, which sends it to the end channel once the requests that met it have their answers. A later failure changes none of the three and sends nothing. The worker's failure paths and its caught panic go through the pair. Before this a later failure overwrote the reason, killed the board again with its own, and sent a second end, so /health, a waiting request and the process exit could each name a different failure once a second thread can fail the server. --- crates/serve/src/worker.rs | 105 +++++++++++++++++++++++++++++-------- 1 file changed, 82 insertions(+), 23 deletions(-) diff --git a/crates/serve/src/worker.rs b/crates/serve/src/worker.rs index 86176154f..12a6ff93e 100644 --- a/crates/serve/src/worker.rs +++ b/crates/serve/src/worker.rs @@ -154,7 +154,8 @@ pub(crate) struct Shared { /// thread. pub work: Condvar, pub stats: Mutex, - /// Set by the engine error that ended the server; the reason `/health` gives. + /// The first failure that ended the server ([`Shared::record_fatal`]); + /// the reason `/health` gives. pub fatal: Mutex>, /// The orderly stop's cause, once it began ([`Shared::begin_stop`]): the /// reason every request that needs the engine is refused with. @@ -207,6 +208,32 @@ impl Shared { } } + /// Adopts `f` as the failure that ends the server unless one already + /// was: the first is kept. `/health` and every later request see its + /// error, and the waiting requests' and actions' channels close on it, + /// so their HTTP threads answer with it. `Some(f)` when `f` was adopted, + /// which [`Shared::end_on`] then sends once the requests that met it have + /// their answers; `None` when an earlier failure was, whose own sender + /// ends the server. + #[must_use] + pub(crate) fn record_fatal(&self, f: EngineFailure) -> Option { + let mut fatal = relock(&self.fatal); + if fatal.is_some() { + return None; + } + *fatal = Some(f.error.clone()); + drop(relock(&self.board).kill(&f.error)); + Some(f) + } + + /// Ends the server on the failure [`Shared::record_fatal`] adopted; + /// nothing for one it did not. + pub(crate) fn end_on(&self, adopted: Option) { + if let Some(f) = adopted { + let _ = self.end.send(End::Engine(f)); + } + } + pub(crate) fn next_id(&self) -> u64 { self.ids.fetch_add(1, Ordering::SeqCst) } @@ -402,16 +429,14 @@ pub(crate) fn serve(slot: Slot, sh: Arc) { // and drops the engine it holds, which no stop waits for. sh.end_loop(); if let Err(p) = ran { - let error = format!("the engine thread panicked: {}", panic_text(&*p)); - *relock(&sh.fatal) = Some(error.clone()); - drop(relock(&sh.board).kill(&error)); + let adopted = sh.record_fatal(EngineFailure { + engine: "unknown: its thread panicked".to_owned(), + error: format!("the engine thread panicked: {}", panic_text(&*p)), + }); // The running requests' channels close: their HTTP threads answer // with the reason. drop(w); - let _ = sh.end.send(End::Engine(EngineFailure { - engine: "unknown: its thread panicked".to_owned(), - error, - })); + sh.end_on(adopted); } } @@ -510,24 +535,20 @@ impl Worker { Ok(()) } - /// Records an engine failure: `/health` and every later request see it, - /// and what waits is refused. - fn fail(&mut self, e: &EngineError) -> EngineFailure { - let f = EngineFailure { + /// Records an engine failure ([`Shared::record_fatal`]): unless an + /// earlier failure ended the server, `/health` and every later request + /// see it, and what waits is refused with it. + fn fail(&mut self, e: &EngineError) -> Option { + self.sh.record_fatal(EngineFailure { engine: self.slot.engine.describe(), error: e.to_string(), - }; - *relock(&self.sh.fatal) = Some(f.error.clone()); - // The waiting requests' and actions' channels close: their HTTP - // threads answer with the reason. - drop(relock(&self.sh.board).kill(&f.error)); - f + }) } /// Ends the server once the requests that met the failure have their /// answers. - fn end(&self, f: EngineFailure) -> Dead { - let _ = self.sh.end.send(End::Engine(f)); + fn end(&self, adopted: Option) -> Dead { + self.sh.end_on(adopted); Dead } @@ -1412,12 +1433,50 @@ mod tests { use std::sync::{Arc, mpsc}; use std::time::Duration; - use super::{Shared, serve}; - use crate::api::relock; + use super::{Acted, Action, Shared, serve}; + use crate::api::{End, EngineFailure, relock}; use crate::genloop::Slot; - use crate::sched::{Board, FifoPicker}; + use crate::sched::{Board, FifoPicker, Refusal, Reserve}; use crate::{MockEngine, sampling}; + /// The first failure ends the server, once: a later one, from the + /// engine thread or from a request's own thread, changes neither the + /// reason `/health` and the board give nor the failure the server ends + /// with. + #[test] + fn the_first_failure_is_the_one_that_ends_the_server() { + let (end, ended) = mpsc::channel(); + let sh = Shared::new( + Board::new(1, 4, Box::new(FifoPicker)), + end, + sampling::reference_factory(), + ); + let failure = |error: &str| EngineFailure { + engine: "mock".to_owned(), + error: error.to_owned(), + }; + let first = sh.record_fatal(failure("first")); + assert!(sh.record_fatal(failure("second")).is_none()); + sh.end_on(first); + assert_eq!(relock(&sh.fatal).as_deref(), Some("first")); + let action = Action { + run: Box::new(|_: &mut Slot| Acted { + failure: None, + reply: Box::new(|| {}), + }), + drops: false, + }; + assert_eq!( + relock(&sh.board).reserve(Reserve::All, action).err(), + Some(Refusal::Dead("first".to_owned())) + ); + match ended.try_recv() { + Ok(End::Engine(f)) => assert_eq!(f.error, "first"), + _ => panic!("no engine end"), + } + assert!(ended.try_recv().is_err(), "the server ends once"); + } + /// The halt reads the stop's cause, which [`Shared::begin_stop`] sets /// before it stores the stop flag: a stop flag with no cause is that /// order broken, and the engine thread ends on a panic naming it — a From 4d6addf7f161e7840be1dde3520edb308eeac234 Mon Sep 17 00:00:00 2001 From: juliankang4 <200967947+juliankang4@users.noreply.github.com> Date: Thu, 15 Oct 2026 00:14:30 +0900 Subject: [PATCH 3/7] gpu-gates: the reserves a plan holds beside its tensors have one owner, beside image_plan's reserve_on, reserve_in, with_reserve, note and refusal move to beside.rs, keyed by what the reserve is for: Beside names each reserve a plan holds on its stage card beside its own tensors, the image encoder's tower its one field in this commit. The moved functions give the names, bytes and lines image_plan gave: "image encoder" as the reserve, the verdict note and the plan refusal word for word, so by the code no seat's plan or record moves. image_plan keeps the media load's per-position table term, and vision_seat re-exports only that. gate-paths gains a beside.rs row with the gates that reached the reserve, and image_plan's row narrows to the gates that still reach the table term. --- crates/gpu-gates/src/beside.rs | 141 ++++++++++++++++++ crates/gpu-gates/src/bin/gate_qwen3_serve.rs | 11 +- crates/gpu-gates/src/bin/shared/ds41_place.rs | 4 +- .../src/bin/shared/glm5next_place.rs | 10 +- .../gpu-gates/src/bin/shared/qwen35_open.rs | 6 +- .../src/bin/shared/qwen3moe_place.rs | 16 +- .../src/bin/shared/serve_seats/decide.rs | 13 +- .../src/bin/shared/serve_seats/ds41.rs | 19 ++- .../src/bin/shared/serve_seats/glm.rs | 13 +- .../src/bin/shared/serve_seats/qwen3.rs | 3 +- crates/gpu-gates/src/image_plan.rs | 116 +------------- crates/gpu-gates/src/lib.rs | 1 + crates/gpu-gates/src/vision_seat.rs | 14 +- tools/gate-paths.tsv | 3 +- 14 files changed, 210 insertions(+), 160 deletions(-) create mode 100644 crates/gpu-gates/src/beside.rs diff --git a/crates/gpu-gates/src/beside.rs b/crates/gpu-gates/src/beside.rs new file mode 100644 index 000000000..64687ed82 --- /dev/null +++ b/crates/gpu-gates/src/beside.rs @@ -0,0 +1,141 @@ +//! What a plan holds beside its own tensors: a model the process serves beside the planned one (the +//! image encoder's tower) that sits on the stage card, as a named reserve of fixed bytes the plan's +//! fit counts ([`model::placement::Card::reserves`]). One function enters it into a placed plan's +//! machine and a whole-fit probe's ([`reserve_on`]), one reads it back ([`reserve_in`]), and a +//! plan's refusal ([`refusal`]) and a verdict's line ([`note`]) name it. Free of the encoder crates, +//! so a placement module every build includes can take it. + +use std::fmt::Display; + +use model::placement::{Card, Machine}; + +/// What the plan's reserve for the image encoder's tower is called, as the plan's terms name it +/// (beside `workstation`'s draft and tier-batch reserves). +pub const IMAGE_ENCODER: &str = "image encoder"; + +/// The bytes a plan reserves on its stage card beside its own tensors, by what they are for; +/// `None` for one that sits off the plan or is not served. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct Beside { + /// The image encoder's tower. + pub image: Option, +} + +impl Beside { + /// Each reserve by its name, in the order the stage card takes them. + fn named(self) -> impl Iterator { + [(IMAGE_ENCODER, self.image)] + .into_iter() + .filter_map(|(name, bytes)| Some((name, bytes?))) + } +} + +/// `machine` with `beside`'s reserves on its stage card. The one place they enter a machine: a +/// placed plan's builder and a whole-fit probe both call it, so both fit the same bytes. +pub fn reserve_on(machine: &mut Machine, beside: Beside) { + for (name, bytes) in beside.named() { + machine.cards[0].reserves.push((name.to_owned(), bytes)); + } +} + +/// The reserves [`reserve_on`] put on `machine`'s stage card: what a plan built on it refuses +/// beside ([`refusal`]). +#[must_use] +pub fn reserve_in(machine: &Machine) -> Beside { + let held = |name: &str| { + machine + .cards + .first()? + .reserves + .iter() + .find(|(n, _)| n == name) + .map(|&(_, bytes)| bytes) + }; + Beside { + image: held(IMAGE_ENCODER), + } +} + +/// A machine builder (`layers` to its [`Machine`]) with [`reserve_on`] applied to what it builds. +pub fn with_reserve(plain: impl Fn(usize) -> Machine, beside: Beside) -> impl Fn(usize) -> Machine { + move |layers| { + let mut machine = plain(layers); + reserve_on(&mut machine, beside); + machine + } +} + +/// What a verdict's line adds when its card carries the image encoder's reserve: the reserve by +/// name and bytes, beside the reserves term the verdict sums them in. Nothing for a card that +/// carries none, so a load without `--mmproj` prints its line as it did. +#[must_use] +pub fn note(card: &Card) -> String { + card.reserves + .iter() + .find(|(name, _)| name == IMAGE_ENCODER) + .map_or_else(String::new, |(name, bytes)| { + format!("; the reserves term holds the {name}'s {bytes} B") + }) +} + +/// A plan's refusal, beside the reserves that were in it: with none the refusal as it is. +pub fn refusal(beside: Beside, e: impl Display) -> String { + let held: Vec = beside + .named() + .map(|(name, bytes)| format!("the {name}'s {bytes} B as the stage card's {name} reserve")) + .collect(); + if held.is_empty() { + return e.to_string(); + } + format!("the plan holds {}: {e}", held.join(" and ")) +} + +#[cfg(test)] +mod tests { + use model::placement::workstation::plan_a; + + use super::*; + + const BYTES: u64 = 1_476_000_000; + + /// The whole-fit probe and the placed plan take the one reserve: the machines both build carry + /// the same named bytes, the card sets aside exactly that much more, and none changes nothing. + #[test] + fn the_probe_and_the_placed_plan_see_the_same_reserve() { + let image = Beside { image: Some(BYTES) }; + let plain = plan_a(40); + let mut probe = plan_a(40); + reserve_on(&mut probe, image); + let placed = with_reserve(plan_a, image)(40); + assert_eq!(probe, placed); + assert_eq!( + probe.cards[0].reserves.last(), + Some(&(IMAGE_ENCODER.to_owned(), BYTES)) + ); + assert_eq!( + probe.cards[0].reserve_bytes(), + plain.cards[0].reserve_bytes() + BYTES + ); + assert_eq!( + probe.cards[0].set_aside_bytes(), + plain.cards[0].set_aside_bytes() + BYTES + ); + assert_eq!(reserve_in(&probe), image); + assert_eq!(reserve_in(&plain), Beside::default()); + let mut none = plan_a(40); + reserve_on(&mut none, Beside::default()); + assert_eq!(none, plain); + assert_eq!(with_reserve(plan_a, Beside::default())(40), plain); + assert_eq!(note(&plain.cards[0]), ""); + assert_eq!( + note(&probe.cards[0]), + "; the reserves term holds the image encoder's 1476000000 B" + ); + assert_eq!(refusal(Beside::default(), "no fit"), "no fit"); + assert_eq!( + refusal(image, "no fit"), + "the plan holds the image encoder's 1476000000 B as the stage card's image encoder \ + reserve: no fit" + ); + } +} diff --git a/crates/gpu-gates/src/bin/gate_qwen3_serve.rs b/crates/gpu-gates/src/bin/gate_qwen3_serve.rs index 1cc25822e..86d6e5835 100644 --- a/crates/gpu-gates/src/bin/gate_qwen3_serve.rs +++ b/crates/gpu-gates/src/bin/gate_qwen3_serve.rs @@ -448,13 +448,14 @@ mod gate { use bloomery_gpu::arch::qwen3moe::ubatch::ubatch_size; use bloomery_gpu::arch::qwen3moe::{KvQ8, Open35, Qwen35moeModel, checkpoint_budget35}; use bloomery_gpu::linear::{KHeadMap, LinearShape}; + use bloomery_gpu_gates::beside; use bloomery_gpu_gates::ctx_ladder::{ ctx_line, default_line_held, field as field_u64, rung_above, rungs, target_of, }; use bloomery_gpu_gates::generate::Place; use bloomery_gpu_gates::record::{self, Fields}; use bloomery_gpu_gates::serve_client::{ - self, Answer, PrefixTurns, Served, agree, beside, chat_is_those_ids, check, curl, + self, Answer, PrefixTurns, Served, agree, chat_is_those_ids, check, curl, edit_and_extension, ids_of, json_of, metric, parse_ids, rendered, serve_cmd, server_log, tokenized, }; @@ -3643,7 +3644,7 @@ mod gate { let seat = &img.seat; let named = format!( "; the reserves term holds the {}'s {} B", - vision_seat::RESERVE, + beside::IMAGE_ENCODER, seat.reserve.unwrap_or(0) ); check( @@ -3651,7 +3652,7 @@ mod gate { "mmproj_whole_fit_names_the_reserve", match seat.reserve { Some(r) => r == img.bytes.total() && fit.ends_with(&named), - None => !fit.contains(vision_seat::RESERVE), + None => !fit.contains(beside::IMAGE_ENCODER), }, ); check( @@ -3862,7 +3863,7 @@ mod gate { println!("{line}"); let counted = format!( ", the {} reserve of {} B counted", - vision_seat::RESERVE, + beside::IMAGE_ENCODER, img.seat.reserve.unwrap_or(0) ); check( @@ -3870,7 +3871,7 @@ mod gate { "mmproj_ctx_default_names_the_reserve", match img.seat.reserve { Some(_) => line.contains(&counted), - None => !line.contains(vision_seat::RESERVE), + None => !line.contains(beside::IMAGE_ENCODER), }, ); Ok(()) diff --git a/crates/gpu-gates/src/bin/shared/ds41_place.rs b/crates/gpu-gates/src/bin/shared/ds41_place.rs index bb3d7fb65..656c61180 100644 --- a/crates/gpu-gates/src/bin/shared/ds41_place.rs +++ b/crates/gpu-gates/src/bin/shared/ds41_place.rs @@ -10,11 +10,11 @@ use std::num::NonZeroUsize; use bloomery_gpu_deepseek41::body::Deepseek41Model; use bloomery_gpu_gates::GateError; +use bloomery_gpu_gates::beside::{self, Beside}; use bloomery_gpu_gates::generate::{Place, PlaceWhy, TierRule, choose_ctx}; use bloomery_gpu_gates::gpu_census; use bloomery_gpu_gates::record::Record; use bloomery_gpu_gates::residency38::{Seqs, reserve_checkpoints}; -use bloomery_gpu_gates::vision_seat; use gguf::Split; use model::arch::deepseek41::hparams::Hparams; use model::arch::deepseek41::place::PlanInputs; @@ -105,7 +105,7 @@ pub fn ctx_choice( (Some((KERNEL_POSITIONS, "KERNEL_POSITIONS")), None), |c| -> Result, GateError> { let mut machine = place.machine(reserve, tier_batch)?(inputs.model.layers); - vision_seat::reserve_on(&mut machine, vision); + beside::reserve_on(&mut machine, Beside { image: vision }); reserve_checkpoints(&mut machine, Seqs::none(slots.get()), c)?; match inputs.plan_with_slots(&machine, c, levers, slots) { Ok(_) => Ok(Probe::Holds(())), diff --git a/crates/gpu-gates/src/bin/shared/glm5next_place.rs b/crates/gpu-gates/src/bin/shared/glm5next_place.rs index ba6c01017..ac6a3dd03 100644 --- a/crates/gpu-gates/src/bin/shared/glm5next_place.rs +++ b/crates/gpu-gates/src/bin/shared/glm5next_place.rs @@ -6,9 +6,9 @@ //! them here. use bloomery_gpu_gates::GateError; +use bloomery_gpu_gates::beside::{self, Beside}; use bloomery_gpu_gates::generate::{Place, TierRule, choose_ctx_of}; use bloomery_gpu_gates::residency38::glm_machine; -use bloomery_gpu_gates::vision_seat; use bloomery_gpu_glm5next::checkpoint_price; use model::arch::glm5next::hparams::Hparams; use model::arch::glm5next::place::{ @@ -110,15 +110,15 @@ pub fn plan_of<'a>( } /// A plan refusal on `machine` as the seat names it: the image encoder's -/// reserve named when the machine carries one ([`vision_seat::refusal`]). +/// reserve named when the machine carries one ([`beside::refusal`]). #[allow( dead_code, reason = "the serve seat names its refusals; the CLI and the routed legs read them bare" )] pub fn named(machine: &Machine, e: PlaceError) -> GateError { - match vision_seat::reserve_in(machine) { - Some(bytes) => GateError::from(vision_seat::refusal(Some(bytes), e)), - None => GateError::from(e), + match beside::reserve_in(machine) { + held if held == Beside::default() => GateError::from(e), + held => GateError::from(beside::refusal(held, e)), } } diff --git a/crates/gpu-gates/src/bin/shared/qwen35_open.rs b/crates/gpu-gates/src/bin/shared/qwen35_open.rs index 22f16b13e..42326038a 100644 --- a/crates/gpu-gates/src/bin/shared/qwen35_open.rs +++ b/crates/gpu-gates/src/bin/shared/qwen35_open.rs @@ -6,8 +6,8 @@ use bloomery_gpu::Gpu; use bloomery_gpu::arch::qwen3moe::{KvQ8, Open35, Qwen35moeModel}; use bloomery_gpu_gates::GateError; +use bloomery_gpu_gates::beside; use bloomery_gpu_gates::generate::Place; -use bloomery_gpu_gates::image_plan; use gguf::Split; use model::arch::qwen3moe::place as q3; use model::arch::qwen35moe::place as q35; @@ -77,7 +77,7 @@ pub fn open_with( /// [`open_with`] whose whole-fit machine goes through `seat` before the verdict: a seat that holds /// bytes of the card for something else (the image encoder on the stage card, -/// `vision_seat::reserve_on`) puts them there, so the verdict counts them. +/// `beside::reserve_on`) puts them there, so the verdict counts them. pub fn open_seated( path: &Path, ctx: usize, @@ -168,7 +168,7 @@ fn refuse_unless_whole_fits( seat: &dyn Fn(&mut Machine), ) -> Result<(), GateError> { let need = whole_verdict(split, o, card, seat)?; - let held = image_plan::note(&need.card); + let held = beside::note(&need.card); eprintln!("whole fit at ctx {}: {need}{held}", o.ctx); if need.fits() { return Ok(()); diff --git a/crates/gpu-gates/src/bin/shared/qwen3moe_place.rs b/crates/gpu-gates/src/bin/shared/qwen3moe_place.rs index c3dd2201f..4363086a5 100644 --- a/crates/gpu-gates/src/bin/shared/qwen3moe_place.rs +++ b/crates/gpu-gates/src/bin/shared/qwen3moe_place.rs @@ -36,6 +36,7 @@ use bloomery_gpu::model::{GpuModel, Slots}; use bloomery_gpu::rope_neox::q8_plane_lens; use bloomery_gpu::{Gpu, Qwen3moeModel}; use bloomery_gpu_gates::GateError; +use bloomery_gpu_gates::beside::{self, Beside}; use bloomery_gpu_gates::generate::{Place, PlaceWhy, choose_ctx, short_in}; use bloomery_gpu_gates::image_plan; use bloomery_gpu_gates::record::{self, Record}; @@ -239,7 +240,12 @@ fn laid( )); } reserve_checkpoints(&mut machine, seqs, slot_ctx)?; - image_plan::reserve_on(&mut machine, media.encoder); + beside::reserve_on( + &mut machine, + Beside { + image: media.encoder, + }, + ); Ok(machine) } @@ -251,7 +257,7 @@ fn slot_ctx(ctx: usize, slots: usize) -> usize { /// What a media load adds to the machines and verdicts of a load ([`laid`], [`whole_on`]): the /// image encoder's bytes when its tower sits on the stage card -/// (`vision_seat::EncoderSeat::reserve`, entered by [`image_plan::reserve_on`]), and the width +/// (`vision_seat::EncoderSeat::reserve`, entered by [`beside::reserve_on`]), and the width /// in f32 of the rotation table every sequence holds on a media load (`Open35::media`; /// [`image_plan::table_bytes_per_position`] a position). A text load adds neither. #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] @@ -618,7 +624,7 @@ impl Whole { fn line(&self, ctx: usize) -> String { match self { Whole::Fits(n) | Whole::Short(n) => { - format!("whole fit at ctx {ctx}: {n}{}", image_plan::note(&n.card)) + format!("whole fit at ctx {ctx}: {n}{}", beside::note(&n.card)) } Whole::Refused(why) => format!("whole fit at ctx {ctx}: {why}"), } @@ -640,7 +646,7 @@ impl Whole { /// out for the program ([`q3::machine`], its step arenas' scratch), on a /// unified machine with the checkpoints `seqs` pin at `slot_ctx` positions a /// slot ([`reserve_checkpoints`], `Machine::whole_card`) and the image -/// `encoder`'s bytes when its tower sits on the card ([`image_plan::reserve_on`], +/// `encoder`'s bytes when its tower sits on the card ([`beside::reserve_on`], /// the reserve a placed plan carries), and what the program holds past it /// (`load`, or /// the program's refusal), sized by `placement::whole_need`, the verdict's @@ -663,7 +669,7 @@ fn whole_on( }; let mut machine = q3::machine(spec, layers, 0); reserve_checkpoints(&mut machine, seqs, slot_ctx)?; - image_plan::reserve_on(&mut machine, encoder); + beside::reserve_on(&mut machine, Beside { image: encoder }); let card = machine .whole_card() .ok_or_else(|| format!("the whole load on {}: a machine with no card", spec.name))?; diff --git a/crates/gpu-gates/src/bin/shared/serve_seats/decide.rs b/crates/gpu-gates/src/bin/shared/serve_seats/decide.rs index a5ccd53de..dac7825b8 100644 --- a/crates/gpu-gates/src/bin/shared/serve_seats/decide.rs +++ b/crates/gpu-gates/src/bin/shared/serve_seats/decide.rs @@ -71,6 +71,7 @@ use std::time::Instant; use bloomery_gpu::GpuError; use bloomery_gpu::arch::qwen3moe::{Body35, PrefillPath, Qwen35moeModel}; use bloomery_gpu::media::{ImageSpan, MediaBody}; +use bloomery_gpu_gates::beside::{self, Beside}; use bloomery_gpu_gates::bind::arm_screen; use bloomery_gpu_gates::generate::{Place, choose_ctx}; use bloomery_gpu_gates::record::{self, Record}; @@ -210,11 +211,13 @@ fn open_qwen35( ctx: usize, image: Option<&EncoderSeat>, ) -> Result<(Box