From d550cc62a1fb2f212f02c97528e81c6ff18cd0e2 Mon Sep 17 00:00:00 2001 From: albertlast Date: Sat, 25 Jul 2026 19:25:28 +0200 Subject: [PATCH] feat(pipeline): run pipeline stages in parallel MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pages were processed strictly one stage at a time. This lets stages overlap and lets a stage fold several pages into one model call. Two config knobs control it, both `0` = auto: - `max_inflight_pages` — pages moving through the pipeline at once. Set to `1` to restore fully sequential behaviour. - `max_batch_pages` — pages a single stage folds into one model call. The larger VRAM lever, so lower this one first. Engines opt in rather than out. The `Engine` trait gains `max_workers`, `max_batch` and `run_batch`, all defaulting to no concurrency, so an engine that shares one GPU context or a `&mut` model is untouched. `run_batch` returns one result per input in order, so a failure is attributed to the page that caused it instead of failing the group. Supporting changes: - `Registry::get` dedupes concurrent misses behind a per-engine lock, so parallel stages hitting a cold engine no longer each load the model and allocate its GPU memory. - Stage threads are pooled and park between runs instead of exiting: candle caches cuDNN handles in a `thread_local!` that is unsafe to tear down. - Translation batching is skipped when a custom system prompt is set, since such a prompt describes the single-page `[N]` block format. - Settings UI for both knobs, translated into all nine locales. Co-Authored-By: Claude Opus 5 --- Cargo.lock | 2 + Cargo.toml | 3 + crates/koharu-app/Cargo.toml | 2 + crates/koharu-app/bin/pipeline.rs | 1 + crates/koharu-app/src/config.rs | 75 ++ crates/koharu-app/src/llm.rs | 465 +++++++- crates/koharu-app/src/pipeline/engine.rs | 78 ++ .../src/pipeline/engines/llm_translate.rs | 120 +- .../src/pipeline/engines/manga_ocr.rs | 156 ++- .../src/pipeline/engines/renderer.rs | 10 +- .../src/pipeline/engines/yuzumarker_font.rs | 169 ++- crates/koharu-app/src/pipeline/mod.rs | 908 ++++++++++++-- crates/koharu-core/src/protocol.rs | 9 + crates/koharu-llm/src/prompt.rs | 12 +- crates/koharu-rpc/src/mcp/mod.rs | 1 + crates/koharu-rpc/src/routes/meta.rs | 1 + crates/koharu-rpc/src/routes/pipelines.rs | 10 +- ui/components/SettingsDialog.tsx | 194 ++- ui/lib/api/default/default.msw.ts | 1 + ui/lib/api/schemas/metaInfo.ts | 9 +- ui/lib/api/schemas/pipelineConfig.ts | 26 +- ui/lib/api/schemas/pipelineConfigPatch.ts | 17 +- ui/openapi.json | 1058 +++++++++++++---- ui/public/locales/en-US/translation.json | 9 + ui/public/locales/es-ES/translation.json | 9 + ui/public/locales/ja-JP/translation.json | 9 + ui/public/locales/ko-KR/translation.json | 9 + ui/public/locales/pt-BR/translation.json | 9 + ui/public/locales/ru-RU/translation.json | 9 + ui/public/locales/tr-TR/translation.json | 9 + ui/public/locales/zh-CN/translation.json | 9 + ui/public/locales/zh-TW/translation.json | 9 + 32 files changed, 2927 insertions(+), 481 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 7df5f2870..9083136ff 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4213,6 +4213,7 @@ version = "0.61.2" dependencies = [ "anyhow", "arc-swap", + "async-channel", "async-trait", "atomicwrites", "base64 0.22.1", @@ -4232,6 +4233,7 @@ dependencies = [ "koharu-runtime", "koharu-secrets", "lru", + "num_cpus", "parking_lot", "petgraph", "postcard", diff --git a/Cargo.toml b/Cargo.toml index 04fbebbe4..a5d294aef 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -89,6 +89,9 @@ tempfile = "3.24" once_cell = "1.21" libloading = "0.8.9" num_cpus = "1.17" +# MPMC: lets every worker on a multi-worker pipeline stage pull from one +# shared queue, so pages are work-stolen rather than round-robined. +async-channel = "2.5" rustfft = "6.4" cudarc = { version = "0.19.4", features = [ "cuda-version-from-build-system", diff --git a/crates/koharu-app/Cargo.toml b/crates/koharu-app/Cargo.toml index e832a9eeb..d1fd1742a 100644 --- a/crates/koharu-app/Cargo.toml +++ b/crates/koharu-app/Cargo.toml @@ -27,6 +27,7 @@ koharu-runtime = { workspace = true } koharu-secrets = { workspace = true } anyhow = { workspace = true } arc-swap = { workspace = true } +async-channel = { workspace = true } async-trait = { workspace = true } atomicwrites = { workspace = true } base64 = { workspace = true } @@ -38,6 +39,7 @@ fs4 = { workspace = true } image = { workspace = true } inventory = { workspace = true } lru = { workspace = true } +num_cpus = { workspace = true } parking_lot = { workspace = true } petgraph = { workspace = true } postcard = { workspace = true } diff --git a/crates/koharu-app/bin/pipeline.rs b/crates/koharu-app/bin/pipeline.rs index 2f14d9b8a..16d31337b 100644 --- a/crates/koharu-app/bin/pipeline.rs +++ b/crates/koharu-app/bin/pipeline.rs @@ -222,6 +222,7 @@ async fn run() -> Result<()> { let spec = koharu_app::pipeline::PipelineSpec { scope: koharu_app::pipeline::Scope::Pages(vec![page_id]), steps, + limits: Default::default(), options: koharu_app::PipelineRunOptions { target_language: Some(cli.target_lang.clone()), system_prompt: cli.system_prompt.clone(), diff --git a/crates/koharu-app/src/config.rs b/crates/koharu-app/src/config.rs index dadf810ea..4baac9c16 100644 --- a/crates/koharu-app/src/config.rs +++ b/crates/koharu-app/src/config.rs @@ -79,6 +79,19 @@ pub struct PipelineConfig { pub translator: String, pub inpainter: String, pub renderer: String, + /// Maximum pages moving through the pipeline at once. `0` = auto (one per + /// stage, so every stage can stay busy). + /// + /// Set to `1` to restore fully sequential processing: one page finishes + /// every step before the next starts, and no stage ever batches. That is + /// the escape hatch if parallelism causes trouble. + pub max_inflight_pages: usize, + /// Upper bound on pages any single stage folds into one model call. + /// `0` = auto, `1` = disable batching but keep stage overlap. + /// + /// Batch size is the larger VRAM lever of the two, so try lowering this + /// before `max_inflight_pages`. + pub max_batch_pages: usize, } impl Default for PipelineConfig { @@ -92,6 +105,8 @@ impl Default for PipelineConfig { translator: "llm".to_string(), inpainter: "lama-manga".to_string(), renderer: "koharu-renderer".to_string(), + max_inflight_pages: 0, + max_batch_pages: 0, } } } @@ -236,6 +251,12 @@ pub fn apply_patch(config: &mut AppConfig, patch: koharu_core::ConfigPatch) { if let Some(v) = p.renderer { config.pipeline.renderer = v; } + if let Some(v) = p.max_inflight_pages { + config.pipeline.max_inflight_pages = v; + } + if let Some(v) = p.max_batch_pages { + config.pipeline.max_batch_pages = v; + } } if let Some(providers) = patch.providers { let mut new_providers = Vec::with_capacity(providers.len()); @@ -449,6 +470,60 @@ mod tests { assert_eq!(config.pipeline.ocr, PipelineConfig::default().ocr); } + #[test] + fn apply_patch_sets_parallelism_limits_including_zero() { + let mut config = AppConfig::default(); + config.pipeline.max_inflight_pages = 4; + config.pipeline.max_batch_pages = 2; + + apply_patch( + &mut config, + ConfigPatch { + pipeline: Some(PipelineConfigPatch { + max_inflight_pages: Some(1), + ..Default::default() + }), + ..Default::default() + }, + ); + + // Set field applied, unmentioned field untouched. + assert_eq!(config.pipeline.max_inflight_pages, 1); + assert_eq!(config.pipeline.max_batch_pages, 2); + + // `0` is a real value (auto), not "leave alone" — the patch is sparse + // via `Option`, so the UI must be able to set auto back. + apply_patch( + &mut config, + ConfigPatch { + pipeline: Some(PipelineConfigPatch { + max_inflight_pages: Some(0), + max_batch_pages: Some(0), + ..Default::default() + }), + ..Default::default() + }, + ); + + assert_eq!(config.pipeline.max_inflight_pages, 0); + assert_eq!(config.pipeline.max_batch_pages, 0); + } + + #[test] + fn parallelism_limits_survive_a_config_round_trip() { + let config: AppConfig = toml::from_str( + r#" + [pipeline] + max_inflight_pages = 2 + max_batch_pages = 1 + "#, + ) + .unwrap(); + + assert_eq!(config.pipeline.max_inflight_pages, 2); + assert_eq!(config.pipeline.max_batch_pages, 1); + } + #[test] fn apply_patch_normalizes_invalid_pipeline_engine_names() { let mut config = AppConfig::default(); diff --git a/crates/koharu-app/src/llm.rs b/crates/koharu-app/src/llm.rs index f237e0f4a..3c0b994b3 100644 --- a/crates/koharu-app/src/llm.rs +++ b/crates/koharu-app/src/llm.rs @@ -13,7 +13,7 @@ use std::sync::Arc; -use anyhow::Result; +use anyhow::{Result, bail}; use koharu_core::{ LlmCatalog, LlmCatalogModel, LlmLoadRequest, LlmProviderCatalog, LlmProviderCatalogStatus, LlmState, LlmStateStatus, LlmTarget, LlmTargetKind, @@ -41,7 +41,9 @@ pub enum State { ReadyLocal(Llm), ReadyProvider { target: LlmTarget, - provider: Box, + /// `Arc`, not `Box`, so `translate_texts` can clone it out and drop + /// the state lock *before* awaiting the provider's HTTP call. + provider: Arc, }, Failed { target: Option, @@ -134,7 +136,10 @@ impl Model { target: LlmTarget, provider: Box, ) -> Result<()> { - *self.state.write().await = State::ReadyProvider { target, provider }; + *self.state.write().await = State::ReadyProvider { + target, + provider: Arc::from(provider), + }; self.emit_state().await; Ok(()) } @@ -197,6 +202,58 @@ impl Model { let _ = self.state_tx.send(self.snapshot().await); } + /// Run one generation against whichever backend is loaded. + /// + /// Remote providers are stateless, so the provider handle is cloned out + /// under a *read* lock and the lock is released before the HTTP call is + /// awaited — otherwise every translation in the process serializes on the + /// state lock, and `snapshot()` / `ready()` block for the whole request. + /// Only the local llama.cpp path takes the write lock, which it genuinely + /// needs: `Llm::generate` is `&mut self` and a context is single-use. + async fn generate_raw( + &self, + body: &str, + target_language: Language, + custom_system_prompt: Option<&str>, + ) -> Result { + enum Route { + Remote(Arc, String), + Local, + } + + let route = { + let guard = self.state.read().await; + match &*guard { + State::ReadyProvider { target, provider } => { + Route::Remote(provider.clone(), target.model_id.clone()) + } + State::ReadyLocal(_) => Route::Local, + State::Loading { .. } => bail!("LLM is still loading"), + State::Failed { error, .. } => bail!("LLM failed to load: {error}"), + State::Empty => bail!("no LLM loaded"), + } + }; + + match route { + Route::Remote(provider, model_id) => { + provider + .translate(body, target_language, &model_id, custom_system_prompt) + .await + } + Route::Local => { + let mut guard = self.state.write().await; + match &mut *guard { + State::ReadyLocal(llm) => { + let opts = llm.id().default_generate_options(); + llm.generate(body, &opts, target_language, custom_system_prompt) + } + // Raced with offload/reload between the read and write lock. + _ => bail!("no local LLM loaded"), + } + } + } + } + /// Translate a batch of source strings. Each source becomes a tagged /// `[N]...` block; the response is parsed back into per-block /// translations. Output length matches input length (possibly with empty @@ -215,26 +272,9 @@ impl Model { .unwrap_or(Language::English); let body = format_sources(sources); - let mut guard = self.state.write().await; - let translation = match &mut *guard { - State::ReadyLocal(llm) => { - let opts = llm.id().default_generate_options(); - llm.generate(&body, &opts, target_language, custom_system_prompt) - } - State::ReadyProvider { target, provider } => { - provider - .translate( - &body, - target_language, - &target.model_id, - custom_system_prompt, - ) - .await - } - State::Loading { .. } => Err(anyhow::anyhow!("LLM is still loading")), - State::Failed { error, .. } => Err(anyhow::anyhow!("LLM failed to load: {error}")), - State::Empty => Err(anyhow::anyhow!("no LLM loaded")), - }?; + let translation = self + .generate_raw(&body, target_language, custom_system_prompt) + .await?; let translation = strip_thinking_block(&translation); let out = match parse_tagged_blocks(translation, sources.len())? { @@ -246,6 +286,90 @@ impl Model { .map(|s| strip_wrapping_quotes(s.trim())) .collect()) } + + /// Translate several pages in a single request, tagging every block with + /// its page (`[bPAGE-BLOCK]`) so the response can be validated. + /// + /// Returns one `Vec` per input page, each the same length as that + /// page's sources. If the response fails validation the pages are retried + /// individually — slower, but it never lands a translation on the wrong + /// bubble. + /// + /// Falls back to the per-page path (and the untouched `[N]` wire format) + /// when there is nothing to gain or too much to risk: a single page, or a + /// user-supplied system prompt that describes the old tag scheme. + pub async fn translate_pages( + &self, + pages: &[Vec], + target_language: Option<&str>, + custom_system_prompt: Option<&str>, + ) -> Result>> { + let has_custom_prompt = custom_system_prompt.is_some_and(|p| !p.trim().is_empty()); + if pages.len() <= 1 || has_custom_prompt { + return self + .translate_each(pages, target_language, custom_system_prompt) + .await; + } + + let page_lens: Vec = pages.iter().map(|p| p.len()).collect(); + if page_lens.iter().all(|&n| n == 0) { + return Ok(pages.iter().map(|_| Vec::new()).collect()); + } + + let language = target_language + .and_then(Language::parse) + .unwrap_or(Language::English); + let refs: Vec<&[String]> = pages.iter().map(|p| p.as_slice()).collect(); + let body = format_sources_batched(&refs); + + let parsed = match self + .generate_raw(&body, language, custom_system_prompt) + .await + { + Ok(translation) => { + let translation = strip_thinking_block(&translation); + parse_batched_blocks(translation, &page_lens) + } + Err(err) => Err(err), + }; + + match parsed { + Ok(blocks) => Ok(blocks + .into_iter() + .map(|page| { + page.into_iter() + .map(|s| strip_wrapping_quotes(s.trim())) + .collect() + }) + .collect()), + Err(err) => { + tracing::warn!( + pages = pages.len(), + "batched translation rejected, retrying per page: {err:#}" + ); + self.translate_each(pages, target_language, custom_system_prompt) + .await + } + } + } + + /// Translate each page as its own request, preserving per-page failure + /// isolation: one page erroring doesn't lose the others' translations. + async fn translate_each( + &self, + pages: &[Vec], + target_language: Option<&str>, + custom_system_prompt: Option<&str>, + ) -> Result>> { + let mut out = Vec::with_capacity(pages.len()); + for sources in pages { + out.push( + self.translate_texts(sources, target_language, custom_system_prompt) + .await?, + ); + } + Ok(out) + } } // --------------------------------------------------------------------------- @@ -436,6 +560,13 @@ pub fn provider_config_from_settings( // Tag formatting + response parsing // --------------------------------------------------------------------------- +/// A parsed block tag, both indices 0-based. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct BlockTag { + page: usize, + block: usize, +} + fn format_sources(sources: &[String]) -> String { sources .iter() @@ -445,21 +576,65 @@ fn format_sources(sources: &[String]) -> String { .join("\n") } -fn parse_block_tag(text: &str) -> Option<(usize, usize)> { - let bytes = text.as_bytes(); - if bytes.first()? != &b'[' { +/// Tag each block with its page as well as its index, so a response that +/// drops or repeats a tag can be *detected* rather than silently shifting +/// every later translation onto the wrong bubble — and, past a page boundary, +/// onto the wrong page. +fn format_sources_batched(pages: &[&[String]]) -> String { + let mut lines = Vec::new(); + for (page_idx, sources) in pages.iter().enumerate() { + for (block_idx, text) in sources.iter().enumerate() { + lines.push(format!("[b{}-{}]{}", page_idx + 1, block_idx + 1, text)); + } + } + lines.join("\n") +} + +/// Parse a leading block tag, accepting both the single-page `[N]` form +/// (implicitly page 1) and the batched `[bPAGE-BLOCK]` form. Returns the byte +/// length of the tag and the indices it names. +/// +/// Accepting both is deliberate: a model that ignores the batched instruction +/// and replies with flat `[N]` still parses, and is then caught by +/// [`parse_batched_blocks`]'s validation rather than being mis-assigned. +fn parse_block_tag(text: &str) -> Option<(usize, BlockTag)> { + if !text.starts_with('[') { return None; } let end = text[1..].find(']')?; - let num_str = &text[1..1 + end]; - let id_1based: usize = num_str.parse().ok()?; - if id_1based == 0 { + let body = &text[1..1 + end]; + let len = 1 + end + 1; + + if let Some(rest) = body.strip_prefix(['b', 'B']) { + let (page, block) = rest.split_once('-')?; + let page: usize = page.parse().ok()?; + let block: usize = block.parse().ok()?; + if page == 0 || block == 0 { + return None; + } + return Some(( + len, + BlockTag { + page: page - 1, + block: block - 1, + }, + )); + } + + let block: usize = body.parse().ok()?; + if block == 0 { return None; } - Some((1 + end + 1, id_1based - 1)) + Some(( + len, + BlockTag { + page: 0, + block: block - 1, + }, + )) } -fn find_next_tag(text: &str) -> Option<(usize, usize, usize)> { +fn find_next_tag(text: &str) -> Option<(usize, usize, BlockTag)> { let mut line_start = 0; while line_start <= text.len() { let line = &text[line_start..]; @@ -469,8 +644,8 @@ fn find_next_tag(text: &str) -> Option<(usize, usize, usize)> { .take_while(|&&byte| matches!(byte, b' ' | b'\t')) .count(); let offset = line_start + indent; - if let Some((len, id)) = parse_block_tag(&text[offset..]) { - return Some((offset, len, id)); + if let Some((len, tag)) = parse_block_tag(&text[offset..]) { + return Some((offset, len, tag)); } let Some(next_newline) = line.find('\n') else { break; @@ -480,26 +655,91 @@ fn find_next_tag(text: &str) -> Option<(usize, usize, usize)> { None } -fn parse_tagged_blocks(translation: &str, expected_blocks: usize) -> Result>> { - if find_next_tag(translation).is_none() { - return Ok(None); - } - let mut blocks = vec![String::new(); expected_blocks]; +/// Split a response into `(tag, content)` pairs in the order they appear. +fn scan_blocks(translation: &str) -> Vec<(BlockTag, String)> { + let mut found = Vec::new(); let mut cursor = translation; - let mut found_any = false; - while let Some((offset, len, id)) = find_next_tag(cursor) { - found_any = true; + while let Some((offset, len, tag)) = find_next_tag(cursor) { cursor = &cursor[offset + len..]; let content_end = find_next_tag(cursor) .map(|(next_offset, _, _)| next_offset) .unwrap_or(cursor.len()); - let content = cursor[..content_end].trim().to_string(); - if id < expected_blocks { - blocks[id] = content; - } + found.push((tag, cursor[..content_end].trim().to_string())); cursor = &cursor[content_end..]; } - Ok(found_any.then_some(blocks)) + found +} + +fn parse_tagged_blocks(translation: &str, expected_blocks: usize) -> Result>> { + let found = scan_blocks(translation); + if found.is_empty() { + return Ok(None); + } + let mut blocks = vec![String::new(); expected_blocks]; + for (tag, content) in found { + if tag.page == 0 && tag.block < expected_blocks { + blocks[tag.block] = content; + } + } + Ok(Some(blocks)) +} + +/// Parse a batched response into per-page blocks. +/// +/// Strict on purpose: every expected `(page, block)` must appear exactly once +/// and no unknown id may appear. `parse_tagged_blocks` can afford to be lenient +/// because a missing block just leaves one bubble empty; here a single dropped +/// tag would shift text across a page boundary, which is invisible to the user +/// and corrupts a page that looked fine. Callers retry rejected batches +/// page-by-page, trading a little speed for correctness. +fn parse_batched_blocks(translation: &str, page_lens: &[usize]) -> Result>> { + let found = scan_blocks(translation); + if found.is_empty() { + bail!("response contained no tagged blocks"); + } + + let mut slots: Vec>> = + page_lens.iter().map(|&len| vec![None; len]).collect(); + + for (tag, content) in found { + let page = slots + .get_mut(tag.page) + .ok_or_else(|| anyhow::anyhow!("response referenced unknown page {}", tag.page + 1))?; + let slot = page.get_mut(tag.block).ok_or_else(|| { + anyhow::anyhow!( + "response referenced unknown block {} on page {}", + tag.block + 1, + tag.page + 1 + ) + })?; + if slot.is_some() { + bail!( + "response repeated block [b{}-{}]", + tag.page + 1, + tag.block + 1 + ); + } + *slot = Some(content); + } + + slots + .into_iter() + .enumerate() + .map(|(page_idx, page)| { + page.into_iter() + .enumerate() + .map(|(block_idx, slot)| { + slot.ok_or_else(|| { + anyhow::anyhow!( + "response missing block [b{}-{}]", + page_idx + 1, + block_idx + 1 + ) + }) + }) + .collect::>>() + }) + .collect() } fn split_legacy_lines(translation: &str, expected_blocks: usize) -> Vec { @@ -536,3 +776,136 @@ fn strip_wrapping_quotes(text: &str) -> String { } trimmed.to_string() } + +#[cfg(test)] +mod tests { + use super::*; + + fn pages(spec: &[&[&str]]) -> Vec> { + spec.iter() + .map(|p| p.iter().map(|s| s.to_string()).collect()) + .collect() + } + + // --- wire format ------------------------------------------------------ + + #[test] + fn single_page_wire_format_is_unchanged() { + let sources = vec!["one".to_string(), "two".to_string()]; + assert_eq!(format_sources(&sources), "[1]one\n[2]two"); + } + + #[test] + fn batched_format_qualifies_every_tag_with_its_page() { + let owned = pages(&[&["a", "b"], &["c"]]); + let refs: Vec<&[String]> = owned.iter().map(|p| p.as_slice()).collect(); + assert_eq!(format_sources_batched(&refs), "[b1-1]a\n[b1-2]b\n[b2-1]c"); + } + + #[test] + fn batched_round_trip() { + let owned = pages(&[&["a", "b"], &["c"]]); + let refs: Vec<&[String]> = owned.iter().map(|p| p.as_slice()).collect(); + let echoed = format_sources_batched(&refs); + let parsed = parse_batched_blocks(&echoed, &[2, 1]).expect("round trip"); + assert_eq!(parsed, vec![vec!["a", "b"], vec!["c"]]); + } + + // --- tag parsing ------------------------------------------------------ + + #[test] + fn parses_both_tag_forms() { + assert_eq!( + parse_block_tag("[3]hi").map(|(_, t)| t), + Some(BlockTag { page: 0, block: 2 }) + ); + assert_eq!( + parse_block_tag("[b2-3]hi").map(|(_, t)| t), + Some(BlockTag { page: 1, block: 2 }) + ); + // Zero indices and malformed bodies are not tags. + assert!(parse_block_tag("[0]x").is_none()); + assert!(parse_block_tag("[b0-1]x").is_none()); + assert!(parse_block_tag("[b1-0]x").is_none()); + assert!(parse_block_tag("[b1]x").is_none()); + assert!(parse_block_tag("[bx-y]x").is_none()); + assert!(parse_block_tag("no tag").is_none()); + } + + // --- validation: each of these must be REJECTED, not mis-assigned ----- + + #[test] + fn rejects_missing_block() { + let err = parse_batched_blocks("[b1-1]a\n[b2-1]c", &[2, 1]).unwrap_err(); + assert!(err.to_string().contains("missing"), "{err}"); + } + + #[test] + fn rejects_duplicated_block() { + let err = parse_batched_blocks("[b1-1]a\n[b1-1]again\n[b2-1]c", &[1, 1]).unwrap_err(); + assert!(err.to_string().contains("repeated"), "{err}"); + } + + #[test] + fn rejects_unknown_page() { + let err = parse_batched_blocks("[b1-1]a\n[b9-1]ghost", &[1, 1]).unwrap_err(); + assert!(err.to_string().contains("unknown page"), "{err}"); + } + + #[test] + fn rejects_unknown_block() { + let err = parse_batched_blocks("[b1-1]a\n[b1-7]ghost\n[b2-1]c", &[1, 1]).unwrap_err(); + assert!(err.to_string().contains("unknown block"), "{err}"); + } + + #[test] + fn rejects_flat_tags_when_batching() { + // A model that ignores the batched format and replies with `[1] [2]` + // would otherwise pile every block onto page 1. + let err = parse_batched_blocks("[1]a\n[2]b", &[1, 1]).unwrap_err(); + assert!(err.to_string().contains("unknown block"), "{err}"); + } + + #[test] + fn rejects_response_with_no_tags() { + let err = parse_batched_blocks("just prose", &[1, 1]).unwrap_err(); + assert!(err.to_string().contains("no tagged blocks"), "{err}"); + } + + // --- ordering + content ---------------------------------------------- + + #[test] + fn accepts_tags_returned_out_of_order() { + // Order on the wire doesn't matter; the tag says where each block goes. + let parsed = parse_batched_blocks("[b2-1]c\n[b1-2]b\n[b1-1]a", &[2, 1]).unwrap(); + assert_eq!(parsed, vec![vec!["a", "b"], vec!["c"]]); + } + + #[test] + fn handles_pages_with_differing_and_zero_block_counts() { + let parsed = + parse_batched_blocks("[b1-1]a\n[b1-2]b\n[b1-3]c\n[b3-1]d", &[3, 0, 1]).unwrap(); + assert_eq!(parsed, vec![vec!["a", "b", "c"], vec![], vec!["d"]]); + } + + #[test] + fn keeps_multiline_block_content() { + let parsed = parse_batched_blocks("[b1-1]line one\nline two\n[b2-1]c", &[1, 1]).unwrap(); + assert_eq!(parsed, vec![vec!["line one\nline two"], vec!["c"]]); + } + + // --- single-page path is untouched ------------------------------------ + + #[test] + fn single_page_parse_stays_lenient() { + // Unlike the batched parser, a missing block here just leaves that + // bubble empty rather than failing the page. + let parsed = parse_tagged_blocks("[1]a\n[3]c", 3).unwrap().unwrap(); + assert_eq!(parsed, vec!["a", "", "c"]); + } + + #[test] + fn single_page_parse_reports_untagged_response() { + assert!(parse_tagged_blocks("no tags here", 2).unwrap().is_none()); + } +} diff --git a/crates/koharu-app/src/pipeline/engine.rs b/crates/koharu-app/src/pipeline/engine.rs index d9c874cf9..a4023ce06 100644 --- a/crates/koharu-app/src/pipeline/engine.rs +++ b/crates/koharu-app/src/pipeline/engine.rs @@ -69,11 +69,64 @@ pub struct PipelineRunOptions { // Engine trait // --------------------------------------------------------------------------- +/// Runtime facts the driver hands each engine so it can size its own +/// concurrency. Built once per pipeline run. +#[derive(Debug, Clone, Copy)] +pub struct ConcurrencyHint { + /// Suggested worker count for CPU-bound, lock-free stages. + pub cpu_workers: usize, + /// True when the loaded translator is a remote HTTP provider (network + /// bound, safe to fan out) rather than a local llama.cpp context + /// (single-context, `&mut self`, cannot be parallelized). + pub translator_is_remote: bool, + /// True when a user-supplied system prompt is in effect. Such a prompt + /// describes the single-page `[N]` block format and cannot be assumed to + /// teach the batched `[bP-N]` form, so cross-page translation batching + /// must be disabled. + pub custom_system_prompt: bool, + /// Hard cap on pages folded into one model call. 1 disables batching. + pub max_batch_pages: usize, +} + #[async_trait] pub trait Engine: Send + Sync + 'static { /// Run the engine on one page. Return the ops to apply. /// Empty `Vec` = nothing changed (still a success). async fn run(&self, ctx: EngineCtx<'_>) -> Result>; + + /// How many pages this engine can safely process concurrently. + /// + /// Default 1. Only override when the engine holds no exclusive lock and + /// contends for a resource that actually parallelizes — CPU cores or the + /// network. Fanning out a stage that shares one GPU or one `&mut` model + /// buys nothing and multiplies peak memory. + fn max_workers(&self, _hint: &ConcurrencyHint) -> usize { + 1 + } + + /// How many pages this engine can fold into a single model call. + /// + /// Default 1 (no batching). Override only where the underlying model + /// genuinely batches — a real tensor batch or a single combined request — + /// not where the "batch" API just loops internally. + fn max_batch(&self, _hint: &ConcurrencyHint) -> usize { + 1 + } + + /// Run the engine over several pages at once. + /// + /// Returns **one result per input context, in the same order**, so a + /// failure is attributed to the page that caused it rather than failing + /// the whole group. The default implementation simply runs them in + /// sequence, so engines that don't override `max_batch` never see a + /// batch larger than 1 and need not implement this. + async fn run_batch(&self, ctxs: Vec>) -> Vec>> { + let mut out = Vec::with_capacity(ctxs.len()); + for ctx in ctxs { + out.push(self.run(ctx).await); + } + out + } } // --------------------------------------------------------------------------- @@ -100,12 +153,18 @@ inventory::collect!(EngineInfo); pub struct Registry { engines: RwLock>>, + /// One lock per engine id, held across the `load` await so that + /// concurrent misses for the same engine don't each allocate a copy of + /// the model (and its GPU memory). Only ever guards loading; lookups of + /// already-cached engines never touch it. + loading: tokio::sync::Mutex>>>, } impl Default for Registry { fn default() -> Self { Self { engines: RwLock::new(HashMap::new()), + loading: tokio::sync::Mutex::new(HashMap::new()), } } } @@ -116,6 +175,11 @@ impl Registry { } /// Get or load an engine instance by id. + /// + /// Safe to call concurrently for the same id: only one caller performs the + /// load, the rest wait and observe the cached instance. Without this, + /// parallel pipeline stages hitting a cold engine would each load the + /// model and allocate its GPU memory, then discard all but one. pub async fn get( &self, id: &str, @@ -126,6 +190,20 @@ impl Registry { return Ok(engine); } let info = Self::find(id)?; + + // Take this engine's load lock. Scoped so the map lock isn't held + // across the load itself. + let load_lock = { + let mut loading = self.loading.lock().await; + loading.entry(info.id).or_default().clone() + }; + let _guard = load_lock.lock().await; + + // Re-check: another caller may have loaded it while we waited. + if let Some(engine) = self.engines.read().get(info.id).cloned() { + return Ok(engine); + } + let loaded = async { (info.load)(runtime, cpu).await } .instrument(tracing::info_span!("engine_load", engine = id)) .await?; diff --git a/crates/koharu-app/src/pipeline/engines/llm_translate.rs b/crates/koharu-app/src/pipeline/engines/llm_translate.rs index d33933276..88d46c100 100644 --- a/crates/koharu-app/src/pipeline/engines/llm_translate.rs +++ b/crates/koharu-app/src/pipeline/engines/llm_translate.rs @@ -7,9 +7,17 @@ use async_trait::async_trait; use koharu_core::{NodeDataPatch, NodeId, NodePatch, Op, PageId, Scene, TextData, TextDataPatch}; use crate::pipeline::artifacts::Artifact; -use crate::pipeline::engine::{Engine, EngineCtx, EngineInfo}; +use crate::pipeline::engine::{ConcurrencyHint, Engine, EngineCtx, EngineInfo}; use crate::pipeline::engines::support::text_nodes; +/// Pages folded into one request when the translator supports it. Bounded +/// because the whole group shares a context window — and because a rejected +/// batch costs a full re-translation of every page in it. +const MAX_BATCH_PAGES: usize = 4; + +/// Concurrent in-flight requests against a remote provider. +const REMOTE_WORKERS: usize = 4; + pub struct Model; #[async_trait] @@ -30,24 +38,102 @@ impl Engine for Model { ) .await?; - let mut ops = Vec::with_capacity(targets.len()); - for ((node_id, _), translation) in targets.into_iter().zip(translations) { - ops.push(Op::UpdateNode { - page: ctx.page, - id: node_id, - patch: NodePatch { - data: Some(NodeDataPatch::Text(TextDataPatch { - translation: Some(Some(translation)), - ..Default::default() - })), - transform: None, - visible: None, - }, - prev: NodePatch::default(), - }); + Ok(translation_ops(ctx.page, targets, translations)) + } + + /// Remote providers are network-bound and stateless, so requests overlap. + /// A local llama.cpp context is `&mut` and single-use — fanning it out + /// would only queue on the state lock. + fn max_workers(&self, hint: &ConcurrencyHint) -> usize { + if hint.translator_is_remote { + REMOTE_WORKERS + } else { + 1 + } + } + + /// Batching here is one combined *request*, not a tensor batch. It only + /// pays against a remote provider, where the win is collapsing N network + /// round-trips into one. A custom system prompt disables it: that prompt + /// documents the single-page `[N]` tags and can't be assumed to teach the + /// page-qualified form the response is validated against. + fn max_batch(&self, hint: &ConcurrencyHint) -> usize { + if hint.translator_is_remote && !hint.custom_system_prompt { + MAX_BATCH_PAGES.min(hint.max_batch_pages) + } else { + 1 + } + } + + async fn run_batch(&self, ctxs: Vec>) -> Vec>> { + if ctxs.len() <= 1 { + let mut out = Vec::with_capacity(ctxs.len()); + for ctx in ctxs { + out.push(self.run(ctx).await); + } + return out; } - Ok(ops) + + let per_page: Vec> = + ctxs.iter().map(collect_translation_targets).collect(); + let sources: Vec> = per_page + .iter() + .map(|targets| targets.iter().map(|(_, s)| s.clone()).collect()) + .collect(); + + // Options are uniform across a run, so page 0's are representative. + let options = ctxs[0].options; + let translated = ctxs[0] + .llm + .translate_pages( + &sources, + options.target_language.as_deref(), + options.system_prompt.as_deref(), + ) + .await; + + match translated { + Ok(pages) => ctxs + .iter() + .zip(per_page) + .zip(pages) + .map(|((ctx, targets), translations)| { + Ok(translation_ops(ctx.page, targets, translations)) + }) + .collect(), + // The whole request failed (network, no LLM loaded, …). Report it + // against every page in the batch rather than silently dropping + // pages the driver still expects an answer for. + Err(err) => ctxs + .iter() + .map(|_| Err(anyhow::anyhow!("{err:#}"))) + .collect(), + } + } +} + +fn translation_ops( + page: PageId, + targets: Vec<(NodeId, String)>, + translations: Vec, +) -> Vec { + let mut ops = Vec::with_capacity(targets.len()); + for ((node_id, _), translation) in targets.into_iter().zip(translations) { + ops.push(Op::UpdateNode { + page, + id: node_id, + patch: NodePatch { + data: Some(NodeDataPatch::Text(TextDataPatch { + translation: Some(Some(translation)), + ..Default::default() + })), + transform: None, + visible: None, + }, + prev: NodePatch::default(), + }); } + ops } fn collect_translation_targets(ctx: &EngineCtx<'_>) -> Vec<(NodeId, String)> { diff --git a/crates/koharu-app/src/pipeline/engines/manga_ocr.rs b/crates/koharu-app/src/pipeline/engines/manga_ocr.rs index 43c00d657..0c878f751 100644 --- a/crates/koharu-app/src/pipeline/engines/manga_ocr.rs +++ b/crates/koharu-app/src/pipeline/engines/manga_ocr.rs @@ -9,46 +9,150 @@ use koharu_ml::comic_text_detector::crop_text_block_bbox; use koharu_ml::manga_ocr::MangaOcr; use crate::pipeline::artifacts::Artifact; -use crate::pipeline::engine::{Engine, EngineCtx, EngineInfo}; +use crate::pipeline::engine::{ConcurrencyHint, Engine, EngineCtx, EngineInfo}; use crate::pipeline::engines::support::{load_source_image, text_node_to_region, text_nodes}; +/// Upper bound on pages folded into one forward pass. +/// +/// The real unit here is *crops*, not pages — `MangaOcr::inference` cats every +/// crop into one `[N,3,H,W]` tensor, and a single page already yields 10-30 of +/// them. So the batch is capped by [`MAX_BATCH_CROPS`] and this only bounds +/// how far ahead the stage will look. +const MAX_BATCH_PAGES: usize = 4; + +/// Crop budget for one forward pass. Activation memory scales linearly with +/// this, so it is the actual guard against OOM on smaller GPUs. +const MAX_BATCH_CROPS: usize = 64; + pub struct Model(MangaOcr); +/// Crops for one page, paired with the nodes they came from. +struct PageCrops { + nodes: Vec, + crops: Vec, +} + +impl Model { + fn page_crops(&self, ctx: &EngineCtx<'_>) -> Result { + let texts = text_nodes(ctx.scene, ctx.page); + if texts.is_empty() { + return Ok(PageCrops { + nodes: Vec::new(), + crops: Vec::new(), + }); + } + let image = load_source_image(ctx.scene, ctx.page, ctx.blobs)?; + let mut nodes = Vec::with_capacity(texts.len()); + let mut crops = Vec::with_capacity(texts.len()); + for (node_id, transform, text) in &texts { + let region = text_node_to_region(transform, text); + crops.push(crop_text_block_bbox(&image, ®ion)); + nodes.push(*node_id); + } + Ok(PageCrops { nodes, crops }) + } + + /// One inference call per page — the fallback whenever a combined forward + /// pass isn't safe or didn't work. + fn infer_each( + &self, + ctxs: &[EngineCtx<'_>], + per_page: Vec>, + ) -> Vec>> { + ctxs.iter() + .zip(per_page) + .map(|(ctx, page)| match page { + Ok(PageCrops { nodes, crops }) if !crops.is_empty() => self + .0 + .inference(&crops) + .map(|texts| ocr_ops(ctx.page, &nodes, texts)), + Ok(_) => Ok(Vec::new()), + Err(err) => Err(err), + }) + .collect() + } +} + #[async_trait] impl Engine for Model { async fn run(&self, ctx: EngineCtx<'_>) -> Result> { - let texts = text_nodes(ctx.scene, ctx.page); - if texts.is_empty() { + let PageCrops { nodes, crops } = self.page_crops(&ctx)?; + if crops.is_empty() { return Ok(Vec::new()); } - let image = load_source_image(ctx.scene, ctx.page, ctx.blobs)?; - let crops: Vec = texts + let recognised = self.0.inference(&crops)?; + Ok(ocr_ops(ctx.page, &nodes, recognised)) + } + + /// `MangaOcr::inference` is a genuine tensor batch — every crop is cat'd + /// into one tensor and put through a single forward pass — so crops from + /// several pages cost far less together than separately. + fn max_batch(&self, hint: &ConcurrencyHint) -> usize { + MAX_BATCH_PAGES.min(hint.max_batch_pages) + } + + async fn run_batch(&self, ctxs: Vec>) -> Vec>> { + // Crop each page independently so one unreadable page fails alone. + let per_page: Vec> = + ctxs.iter().map(|ctx| self.page_crops(ctx)).collect(); + + let flat: Vec = per_page .iter() - .map(|(_, transform, text)| { - let region = text_node_to_region(transform, text); - crop_text_block_bbox(&image, ®ion) - }) + .flatten() + .flat_map(|page| page.crops.iter().cloned()) .collect(); - let recognised = self.0.inference(&crops)?; - let mut ops = Vec::with_capacity(texts.len()); - for ((node_id, _, _), text) in texts.iter().zip(recognised) { - ops.push(Op::UpdateNode { - page: ctx.page, - id: *node_id, - patch: NodePatch { - data: Some(NodeDataPatch::Text(TextDataPatch { - text: Some(Some(text)), - ..Default::default() - })), - transform: None, - visible: None, - }, - prev: NodePatch::default(), - }); + // Over the crop budget, or nothing to batch: one call per page. + if flat.is_empty() || flat.len() > MAX_BATCH_CROPS { + return self.infer_each(&ctxs, per_page); } - Ok(ops) + + let recognised = match self.0.inference(&flat) { + Ok(texts) => texts, + // The batched forward failed as a unit; retry singly so only a + // genuinely bad page ends up reported. + Err(_) => return self.infer_each(&ctxs, per_page), + }; + + // Split the flat output back out by each page's crop count. Pages + // that errored during cropping consumed none of it. + let mut cursor = 0; + ctxs.iter() + .zip(per_page) + .map(|(ctx, page)| match page { + Ok(PageCrops { nodes, crops }) => { + let texts = recognised[cursor..cursor + crops.len()].to_vec(); + cursor += crops.len(); + Ok(ocr_ops(ctx.page, &nodes, texts)) + } + Err(err) => Err(err), + }) + .collect() + } +} + +fn ocr_ops( + page: koharu_core::PageId, + nodes: &[koharu_core::NodeId], + texts: Vec, +) -> Vec { + let mut ops = Vec::with_capacity(nodes.len()); + for (node_id, text) in nodes.iter().zip(texts) { + ops.push(Op::UpdateNode { + page, + id: *node_id, + patch: NodePatch { + data: Some(NodeDataPatch::Text(TextDataPatch { + text: Some(Some(text)), + ..Default::default() + })), + transform: None, + visible: None, + }, + prev: NodePatch::default(), + }); } + ops } inventory::submit! { diff --git a/crates/koharu-app/src/pipeline/engines/renderer.rs b/crates/koharu-app/src/pipeline/engines/renderer.rs index 09429cfad..281192749 100644 --- a/crates/koharu-app/src/pipeline/engines/renderer.rs +++ b/crates/koharu-app/src/pipeline/engines/renderer.rs @@ -15,7 +15,7 @@ use koharu_core::{ use koharu_llm::Language; use crate::pipeline::artifacts::Artifact; -use crate::pipeline::engine::{Engine, EngineCtx, EngineInfo}; +use crate::pipeline::engine::{ConcurrencyHint, Engine, EngineCtx, EngineInfo}; use crate::pipeline::engines::support::{ find_image_node, find_mask_node, image_dimensions, load_source_image, text_nodes, upsert_image_blob, @@ -26,6 +26,14 @@ pub struct Model; #[async_trait] impl Engine for Model { + /// Text shaping and rasterisation is pure CPU work on `&self`, and the + /// only shared mutable state is the font book — a short-lived mutex around + /// font lookup, not the raster loop. So pages scale across cores here, + /// unlike the GPU stages upstream. + fn max_workers(&self, hint: &ConcurrencyHint) -> usize { + hint.cpu_workers + } + async fn run(&self, ctx: EngineCtx<'_>) -> Result> { // Find the target surface: prefer inpainted, fall back to source. let base = match find_image_node(ctx.scene, ctx.page, ImageRole::Inpainted) { diff --git a/crates/koharu-app/src/pipeline/engines/yuzumarker_font.rs b/crates/koharu-app/src/pipeline/engines/yuzumarker_font.rs index 64ffa9b4f..8f785590e 100644 --- a/crates/koharu-app/src/pipeline/engines/yuzumarker_font.rs +++ b/crates/koharu-app/src/pipeline/engines/yuzumarker_font.rs @@ -8,56 +8,159 @@ use koharu_core::{FontPrediction, NodeDataPatch, NodePatch, Op, TextDataPatch}; use koharu_ml::font_detector::FontDetector; use crate::pipeline::artifacts::Artifact; -use crate::pipeline::engine::{Engine, EngineCtx, EngineInfo}; +use crate::pipeline::engine::{ConcurrencyHint, Engine, EngineCtx, EngineInfo}; use crate::pipeline::engines::support::{load_source_image, text_nodes}; +/// Upper bound on pages folded into one call. As with manga-ocr the real +/// unit is crops, bounded by [`MAX_BATCH_CROPS`]. +const MAX_BATCH_PAGES: usize = 4; + +/// Crop budget for one batched call. +const MAX_BATCH_CROPS: usize = 64; + pub struct Model(FontDetector); +/// Crops for one page, paired with the nodes they came from. +struct PageCrops { + nodes: Vec, + crops: Vec, +} + +impl Model { + fn page_crops(&self, ctx: &EngineCtx<'_>) -> Result { + let texts = text_nodes(ctx.scene, ctx.page); + if texts.is_empty() { + return Ok(PageCrops { + nodes: Vec::new(), + crops: Vec::new(), + }); + } + let image = load_source_image(ctx.scene, ctx.page, ctx.blobs)?; + let mut nodes = Vec::with_capacity(texts.len()); + let mut crops = Vec::with_capacity(texts.len()); + for (node_id, t, _) in &texts { + crops.push(image.crop_imm( + t.x.max(0.0) as u32, + t.y.max(0.0) as u32, + t.width.max(1.0) as u32, + t.height.max(1.0) as u32, + )); + nodes.push(*node_id); + } + Ok(PageCrops { nodes, crops }) + } + + fn infer( + &self, + page: koharu_core::PageId, + nodes: &[koharu_core::NodeId], + crops: &[DynamicImage], + ) -> Result> { + let mut preds = self.0.inference(crops, 1)?; + for p in &mut preds { + normalize_font_prediction(p); + } + Ok(font_ops(page, nodes, preds)) + } + + /// One inference call per page — the fallback whenever a combined call + /// isn't safe or didn't work. + fn infer_each( + &self, + ctxs: &[EngineCtx<'_>], + per_page: Vec>, + ) -> Vec>> { + ctxs.iter() + .zip(per_page) + .map(|(ctx, page)| match page { + Ok(PageCrops { nodes, crops }) if !crops.is_empty() => { + self.infer(ctx.page, &nodes, &crops) + } + Ok(_) => Ok(Vec::new()), + Err(err) => Err(err), + }) + .collect() + } +} + #[async_trait] impl Engine for Model { async fn run(&self, ctx: EngineCtx<'_>) -> Result> { - let texts = text_nodes(ctx.scene, ctx.page); - if texts.is_empty() { + let PageCrops { nodes, crops } = self.page_crops(&ctx)?; + if crops.is_empty() { return Ok(Vec::new()); } - let image = load_source_image(ctx.scene, ctx.page, ctx.blobs)?; - let crops: Vec = texts + self.infer(ctx.page, &nodes, &crops) + } + + /// `FontDetector::inference` already takes a crop slice and preprocesses + /// it in parallel before a single batched forward, so crops from several + /// pages cost less together than separately. + fn max_batch(&self, hint: &ConcurrencyHint) -> usize { + MAX_BATCH_PAGES.min(hint.max_batch_pages) + } + + async fn run_batch(&self, ctxs: Vec>) -> Vec>> { + let per_page: Vec> = + ctxs.iter().map(|ctx| self.page_crops(ctx)).collect(); + + let flat: Vec = per_page .iter() - .map(|(_, t, _)| { - image.crop_imm( - t.x.max(0.0) as u32, - t.y.max(0.0) as u32, - t.width.max(1.0) as u32, - t.height.max(1.0) as u32, - ) - }) + .flatten() + .flat_map(|page| page.crops.iter().cloned()) .collect(); - let mut preds = self.0.inference(&crops, 1)?; + if flat.is_empty() || flat.len() > MAX_BATCH_CROPS { + return self.infer_each(&ctxs, per_page); + } + + let mut preds = match self.0.inference(&flat, 1) { + Ok(preds) => preds, + Err(_) => return self.infer_each(&ctxs, per_page), + }; for p in &mut preds { normalize_font_prediction(p); } - let mut ops = Vec::with_capacity(texts.len()); - for ((node_id, _, _), pred) in texts.iter().zip(preds) { - ops.push(Op::UpdateNode { - page: ctx.page, - id: *node_id, - patch: NodePatch { - data: Some(NodeDataPatch::Text(TextDataPatch { - font_prediction: Some(Some(ml_prediction_to_core(pred))), - // Clear any previous style so the renderer re-derives. - style: Some(None), - ..Default::default() - })), - transform: None, - visible: None, - }, - prev: NodePatch::default(), - }); - } - Ok(ops) + let mut cursor = 0; + ctxs.iter() + .zip(per_page) + .map(|(ctx, page)| match page { + Ok(PageCrops { nodes, crops }) => { + let slice = preds[cursor..cursor + crops.len()].to_vec(); + cursor += crops.len(); + Ok(font_ops(ctx.page, &nodes, slice)) + } + Err(err) => Err(err), + }) + .collect() + } +} + +fn font_ops( + page: koharu_core::PageId, + nodes: &[koharu_core::NodeId], + preds: Vec, +) -> Vec { + let mut ops = Vec::with_capacity(nodes.len()); + for (node_id, pred) in nodes.iter().zip(preds) { + ops.push(Op::UpdateNode { + page, + id: *node_id, + patch: NodePatch { + data: Some(NodeDataPatch::Text(TextDataPatch { + font_prediction: Some(Some(ml_prediction_to_core(pred))), + // Clear any previous style so the renderer re-derives. + style: Some(None), + ..Default::default() + })), + transform: None, + visible: None, + }, + prev: NodePatch::default(), + }); } + ops } inventory::submit! { diff --git a/crates/koharu-app/src/pipeline/mod.rs b/crates/koharu-app/src/pipeline/mod.rs index 48e0a309c..4565babae 100644 --- a/crates/koharu-app/src/pipeline/mod.rs +++ b/crates/koharu-app/src/pipeline/mod.rs @@ -11,13 +11,13 @@ mod engines; pub use artifacts::Artifact; pub use engine::{ - BoxFuture, Engine, EngineCtx, EngineInfo, EngineLoadFn, PipelineRunOptions, Registry, - build_order, + BoxFuture, ConcurrencyHint, Engine, EngineCtx, EngineInfo, EngineLoadFn, PipelineRunOptions, + Registry, build_order, }; pub use engines::support; -use std::sync::Arc; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::sync::{Arc, OnceLock}; use anyhow::{Result, bail}; use koharu_core::{Op, PageId, PipelineStep}; @@ -95,6 +95,29 @@ pub struct PipelineSpec { pub scope: Scope, pub steps: Vec, pub options: PipelineRunOptions, + /// Concurrency caps. `Default` means auto — see [`PipelineLimits`]. + pub limits: PipelineLimits, +} + +/// Workers a CPU-bound, lock-free stage gets when sizing itself automatically. +/// +/// Capped well below the core count because the stages that use it run +/// alongside GPU stages and the HTTP server, and because each worker holds a +/// page's worth of pixels. Exposed so the settings UI can show what auto +/// resolves to on this machine. +pub fn auto_cpu_workers() -> usize { + num_cpus::get().clamp(1, 4) +} + +/// Caps on how much of the pipeline runs at once. Both `0` mean "auto". +#[derive(Debug, Clone, Copy, Default)] +pub struct PipelineLimits { + /// Pages allowed in the pipeline simultaneously. `1` restores fully + /// sequential processing (and forces every batch to size 1). + pub max_inflight_pages: usize, + /// Cap on pages any stage folds into one model call. `1` disables + /// batching while keeping stage overlap. + pub max_batch_pages: usize, } #[derive(Debug, Clone)] @@ -149,118 +172,171 @@ pub async fn run( let total_pages = pages.len().max(1); let total_steps = order.len().max(1); - let total_units = (total_pages * total_steps) as u64; - let mut completed: u64 = 0; - let mut warning_count: usize = 0; - - 'pages: for (page_index, page_id) in pages.iter().enumerate() { - for (seq, &i) in order.iter().enumerate() { - if cancel.load(Ordering::Relaxed) { - bail!("cancelled"); + + // No steps means no stages to wire up. The sequential driver silently did + // nothing here, so keep that rather than building an empty pipeline. + if order.is_empty() { + if let Some(sink) = progress.as_ref() { + sink(ProgressTick { + step: None, + step_id: String::new(), + step_index: 0, + total_steps, + page_index: total_pages.saturating_sub(1), + total_pages, + overall_percent: 100, + }); + } + return Ok(RunOutcome::default()); + } + + // --- sizing ----------------------------------------------------------- + + let max_batch_pages = match spec.limits.max_batch_pages { + 0 => usize::MAX, + n => n, + }; + let hint = ConcurrencyHint { + cpu_workers: auto_cpu_workers(), + translator_is_remote: llm + .current_target() + .await + .is_some_and(|t| t.kind != koharu_core::LlmTargetKind::Local), + custom_system_prompt: spec + .options + .system_prompt + .as_deref() + .is_some_and(|p| !p.trim().is_empty()), + max_batch_pages, + }; + // One page in flight means the pipeline is sequential again, so no stage + // can ever accumulate a batch either. + let inflight = match spec.limits.max_inflight_pages { + 0 => order.len().max(1), + n => n, + }; + let batching_possible = inflight > 1; + + // --- load engines ----------------------------------------------------- + // + // Up front rather than lazily: `max_workers` / `max_batch` are engine + // methods, so an instance is needed before its stage can be sized. A load + // failure is kept non-fatal — that stage warns and drops each page, which + // is what the sequential driver did per page. + + let mut stages: Vec = Vec::with_capacity(order.len()); + for &i in &order { + if cancel.load(Ordering::Relaxed) { + bail!("cancelled"); + } + let info = infos[i]; + match registry.get(info.id, &runtime, cpu).await { + Ok(engine) => { + let workers = engine.max_workers(&hint).max(1); + let batch = if batching_possible { + engine.max_batch(&hint).clamp(1, max_batch_pages) + } else { + 1 + }; + stages.push(StageSpec { + info, + engine: Some(engine), + workers, + batch, + }); } - let info = infos[i]; - - if let Some(sink) = progress.as_ref() { - let percent = ((completed * 100) / total_units).min(100) as u8; - sink(ProgressTick { - step: step_for(info), - step_id: info.id.to_string(), - step_index: seq, - total_steps, - page_index, - total_pages, - overall_percent: percent, + Err(err) => { + tracing::warn!(engine = info.id, "engine failed to load: {err:#}"); + stages.push(StageSpec { + info, + engine: None, + workers: 1, + batch: 1, }); - tokio::task::yield_now().await; } + } + } - // The page must still exist (user may have deleted it mid-run). - if !session.scene.read().pages.contains_key(page_id) { - // Skip the remaining steps for a deleted page and credit all - // of them against total_units so progress still reaches 100%. - completed += (total_steps - seq) as u64; - continue 'pages; - } + let tracker = Arc::new(Tracker::new(total_pages, total_steps)); + let ctx = Arc::new(StageContext { + session, + runtime, + llm, + renderer, + cancel: cancel.clone(), + options: spec.options, + progress, + warnings, + tracker: tracker.clone(), + }); - let engine = match registry.get(info.id, &runtime, cpu).await { - Ok(e) => e, - Err(err) => { - // Engine *load* failure: same recovery as a run failure. - report_step_failure( - info.id, - page_id, - seq, - page_index, - total_pages, - total_steps, - &err, - &mut warning_count, - warnings.as_ref(), - ); - completed += (total_steps - seq) as u64; - continue 'pages; - } - }; - let scene_snap = session.scene_snapshot(); - let ctx = EngineCtx { - scene: &scene_snap, - page: *page_id, - blobs: &session.blobs, - runtime: &runtime, - cancel: &cancel, - options: &spec.options, - llm: &llm, - renderer: &renderer, - }; - let step_result = async { engine.run(ctx).await } - .instrument(tracing::info_span!("step", engine = info.id, page = %page_id)) - .await; - let ops = match step_result { - Ok(ops) => ops, - Err(err) => { - report_step_failure( - info.id, - page_id, - seq, - page_index, - total_pages, - total_steps, - &err, - &mut warning_count, - warnings.as_ref(), - ); - // Subsequent steps on this page almost always consume the - // failed step's artifact; skip the rest and move on. - completed += (total_steps - seq) as u64; - continue 'pages; - } - }; - completed += 1; - if ops.is_empty() { - continue; - } - let batch = Op::Batch { - ops, - label: format!("{}: page {}", info.id, page_id), - }; - if let Err(err) = session.apply(batch) { - report_step_failure( - info.id, - page_id, - seq, - page_index, - total_pages, - total_steps, - &err, - &mut warning_count, - warnings.as_ref(), - ); - continue 'pages; - } + // --- wire the stages together ----------------------------------------- + + // Built back-to-front so each stage already knows the channel it forwards + // into: `senders` holds them in reverse, so the last one pushed is always + // the immediate downstream stage. + let mut senders: Vec> = Vec::with_capacity(stages.len()); + let mut workers = Vec::new(); + + for (seq, stage) in stages.iter().enumerate().rev() { + let (tx, rx) = async_channel::bounded::(stage.batch.max(2)); + // The final stage has nowhere to forward to; its items are dropped, + // which is what releases their in-flight permits. + let forward = senders.last().cloned(); + + for _ in 0..stage.workers { + workers.push(spawn_stage_worker( + ctx.clone(), + stage.clone_for_worker(), + seq, + rx.clone(), + forward.clone(), + )); } + senders.push(tx); } - if let Some(sink) = progress.as_ref() { + // Last pushed is stage 0's sender — the head of the pipeline. + let head = senders.pop().expect("at least one stage"); + // Drop the driver's remaining copies so each stage's channel closes once + // every upstream worker has finished with it. + drop(senders); + + // --- feed pages ------------------------------------------------------- + + let permits = Arc::new(tokio::sync::Semaphore::new(inflight)); + for (page_index, page_id) in pages.iter().enumerate() { + if cancel.load(Ordering::Relaxed) { + break; + } + let Ok(permit) = permits.clone().acquire_owned().await else { + break; + }; + if head + .send(Item { + page_index, + page_id: *page_id, + _permit: permit, + }) + .await + .is_err() + { + break; + } + } + drop(head); + + // Every worker returns its thread to the pool once its channel is closed + // and drained. + for worker in workers { + worker.join(); + } + + if cancel.load(Ordering::Relaxed) { + bail!("cancelled"); + } + + if let Some(sink) = ctx.progress.as_ref() { sink(ProgressTick { step: None, step_id: String::new(), @@ -271,7 +347,410 @@ pub async fn run( overall_percent: 100, }); } - Ok(RunOutcome { warning_count }) + Ok(RunOutcome { + warning_count: tracker.warnings.load(Ordering::Relaxed), + }) +} + +// --------------------------------------------------------------------------- +// Stage plumbing +// --------------------------------------------------------------------------- + +/// A page travelling through the pipeline. The permit rides along so that +/// dropping an item anywhere — success, failure, or cancellation — releases +/// its in-flight slot without any explicit bookkeeping. +struct Item { + page_index: usize, + page_id: PageId, + _permit: tokio::sync::OwnedSemaphorePermit, +} + +struct StageSpec { + info: &'static EngineInfo, + /// `None` when the engine failed to load; the stage then warns per page. + engine: Option>, + workers: usize, + batch: usize, +} + +impl StageSpec { + fn clone_for_worker(&self) -> StageSpec { + StageSpec { + info: self.info, + engine: self.engine.clone(), + workers: self.workers, + batch: self.batch, + } + } +} + +/// Everything a stage worker needs that is shared across all stages. +struct StageContext { + session: Arc, + runtime: Arc, + llm: Arc, + renderer: Arc, + cancel: Arc, + options: PipelineRunOptions, + progress: Option, + warnings: Option, + tracker: Arc, +} + +/// Progress bookkeeping shared by every worker. +struct Tracker { + /// Steps completed-or-skipped per page, indexed by page index. + steps_done: Vec, + completed: std::sync::atomic::AtomicU64, + warnings: std::sync::atomic::AtomicUsize, + total_pages: usize, + total_steps: usize, + total_units: u64, +} + +impl Tracker { + fn new(total_pages: usize, total_steps: usize) -> Self { + Self { + steps_done: (0..total_pages) + .map(|_| std::sync::atomic::AtomicUsize::new(0)) + .collect(), + completed: std::sync::atomic::AtomicU64::new(0), + warnings: std::sync::atomic::AtomicUsize::new(0), + total_pages, + total_steps, + total_units: (total_pages * total_steps) as u64, + } + } + + fn credit(&self, page_index: usize, steps: usize) { + if let Some(slot) = self.steps_done.get(page_index) { + slot.fetch_add(steps, Ordering::Relaxed); + } + self.completed.fetch_add(steps as u64, Ordering::Relaxed); + } + + /// Lowest page index not yet finished through every stage. + /// + /// Reported as `current_page` so the UI's "Image N/M" stays monotonic even + /// though pages are in flight simultaneously — and so the frontend only + /// treats a page as done once all of its ops really are in the scene. + fn frontier(&self) -> usize { + self.steps_done + .iter() + .position(|d| d.load(Ordering::Relaxed) < self.total_steps) + .unwrap_or(self.total_pages.saturating_sub(1)) + } + + fn percent(&self) -> u8 { + let done = self.completed.load(Ordering::Relaxed); + ((done * 100) / self.total_units.max(1)).min(100) as u8 + } +} + +/// A unit of stage work handed to a pooled thread, run on that thread's +/// long-lived runtime. +type StageJob = Box; + +/// Threads that run stage work and are never allowed to finish. +/// +/// Each stage still gets a thread to itself for the length of a run: model +/// inference is synchronous and multi-second, so on a shared tokio worker pool +/// several concurrent stages would starve the HTTP server and the SSE stream. A +/// private thread also pins any GPU context to one thread, which is what +/// `comic-text-bubble-detector` already does for itself. +/// +/// What the pool adds is that the thread *parks* when its stage ends instead of +/// exiting, because both of the things a stage thread owns are unsafe to tear +/// down: +/// +/// * `candle` caches its cuDNN handles in a `thread_local!` (they are neither +/// `Send` nor `Sync`), and `cudarc`'s `Drop for Cudnn` unwraps `cudnnDestroy`. +/// A thread that ends therefore destroys those handles and turns any teardown +/// error into a panic inside a destructor — a hard crash, most easily hit by +/// cancelling a run, which ends every stage thread at once. +/// * Hyper's pooled connections belong to the runtime that opened them, and the +/// LLM client is one `Arc` shared by every stage. A +/// runtime that is dropped takes its connections with it, so a *different* +/// worker's next request fails with "error sending request". +/// +/// Parking keeps the cuDNN handles and the CUDA context warm for the next run +/// as a side benefit. +struct StagePool { + jobs: async_channel::Sender, + /// Receiver kept alive here so the channel never closes and the threads + /// never fall out of their loop. + rx: async_channel::Receiver, + /// Threads currently parked on `recv`, claimed by submitters. + idle: Arc, +} + +static STAGE_POOL: OnceLock = OnceLock::new(); + +impl StagePool { + fn get() -> &'static StagePool { + STAGE_POOL.get_or_init(|| { + let (jobs, rx) = async_channel::unbounded::(); + StagePool { + jobs, + rx, + idle: Arc::new(AtomicUsize::new(0)), + } + }) + } + + /// Hand `job` to a parked thread, spawning a fresh one when none is free. + /// + /// The claim decrements before the job is queued so two stages can never + /// count on the same parked thread. Over-spawning is harmless — the extra + /// thread just parks — but under-spawning would deadlock a pipeline whose + /// upstream stage is blocked on a bounded channel, so a failed claim always + /// spawns. + fn submit(&'static self, job: StageJob) { + let claimed = self + .idle + .fetch_update(Ordering::AcqRel, Ordering::Acquire, |n| n.checked_sub(1)) + .is_ok(); + if !claimed { + self.spawn_thread(); + } + // Unbounded, and the receiver lives in the pool, so this never blocks + // and never fails. + let _ = self.jobs.try_send(job); + } + + fn spawn_thread(&'static self) { + let rx = self.rx.clone(); + let idle = self.idle.clone(); + std::thread::spawn(move || { + // Built on first use and then kept for the life of the thread. + let mut rt: Option = None; + // This thread was spawned to serve a job that is already queued, so + // it is not idle on the way into the first `recv`. + let mut park_counted = false; + loop { + if park_counted { + idle.fetch_add(1, Ordering::Release); + } + park_counted = true; + let Ok(job) = rx.recv_blocking() else { return }; + + if rt.is_none() { + match tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + { + Ok(built) => rt = Some(built), + Err(err) => { + // Dropping the job releases its completion signal, + // so the driver stops waiting on a stage that will + // never run rather than hanging. + tracing::error!("stage runtime failed: {err:#}"); + continue; + } + } + } + let rt = rt.as_ref().expect("runtime built above"); + + // A panicking stage must not take the thread down with it: that + // would destroy exactly the cuDNN state this pool exists to + // keep alive. The driver sees the dropped signal and moves on, + // which is what a panicked `JoinHandle` gave it before. + let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| job(rt))); + } + }); + } +} + +/// Waits for one stage worker to finish. The pooled thread outlives the run, so +/// completion is signalled by the job dropping its sender — on panic too — +/// rather than by joining a thread. +struct StageHandle(std::sync::mpsc::Receiver<()>); + +impl StageHandle { + fn join(self) { + // `Err` means the worker panicked or was dropped before signalling, + // which counts as finished just as `JoinHandle::join` did. + let _ = self.0.recv(); + } +} + +/// Run one stage on a pooled thread with its own current-thread runtime. +fn spawn_stage_worker( + ctx: Arc, + stage: StageSpec, + seq: usize, + rx: async_channel::Receiver, + forward: Option>, +) -> StageHandle { + let (done, wait) = std::sync::mpsc::channel::<()>(); + StagePool::get().submit(Box::new(move |rt| { + let _done = done; + rt.block_on(stage_loop(ctx, stage, seq, rx, forward)); + })); + StageHandle(wait) +} + +/// Take one page, then greedily add whatever is *already* sitting in the +/// channel, up to `max`. +/// +/// Never waits for a batch to fill. An idle queue yields a batch of one, so a +/// pipeline that isn't backed up behaves exactly like the sequential driver +/// and adds no latency; batching only kicks in where work has actually piled +/// up. `None` means the channel closed and the stage is finished. +async fn drain_batch(rx: &async_channel::Receiver, max: usize) -> Option> { + let first = rx.recv().await.ok()?; + let mut items = vec![first]; + while items.len() < max { + match rx.try_recv() { + Ok(item) => items.push(item), + Err(_) => break, + } + } + Some(items) +} + +async fn stage_loop( + ctx: Arc, + stage: StageSpec, + seq: usize, + rx: async_channel::Receiver, + forward: Option>, +) { + let tracker = &ctx.tracker; + let remaining_steps = tracker.total_steps - seq; + + while let Some(items) = drain_batch(&rx, stage.batch).await { + if ctx.cancel.load(Ordering::Relaxed) { + return; + } + + // One tick per batch, not per page: the tick is derived from the + // tracker, so repeating it for each page in a batch would put + // identical frames on the SSE bus. + if let Some(sink) = ctx.progress.as_ref() { + sink(ProgressTick { + step: step_for(stage.info), + step_id: stage.info.id.to_string(), + step_index: seq, + total_steps: tracker.total_steps, + page_index: tracker.frontier(), + total_pages: tracker.total_pages, + overall_percent: tracker.percent(), + }); + } + // Give this thread's runtime a chance to flush the frame before the + // next long, fully synchronous inference call monopolises it. + tokio::task::yield_now().await; + + // Engine never loaded: warn once per page, same as the sequential + // driver did when a lazy load failed. + let Some(engine) = stage.engine.as_ref() else { + for item in items { + report_step_failure( + stage.info.id, + &item.page_id, + seq, + item.page_index, + tracker.total_pages, + tracker.total_steps, + &anyhow::anyhow!("engine failed to load"), + &ctx.tracker.warnings, + ctx.warnings.as_ref(), + ); + tracker.credit(item.page_index, remaining_steps); + } + continue; + }; + + // Drop pages the user deleted mid-run before touching the engine. + let scene_snap = ctx.session.scene_snapshot(); + let mut live = Vec::with_capacity(items.len()); + for item in items { + if scene_snap.pages.contains_key(&item.page_id) { + live.push(item); + } else { + tracker.credit(item.page_index, remaining_steps); + } + } + if live.is_empty() { + continue; + } + + let ctxs: Vec> = live + .iter() + .map(|item| EngineCtx { + scene: &scene_snap, + page: item.page_id, + blobs: &ctx.session.blobs, + runtime: &ctx.runtime, + cancel: &ctx.cancel, + options: &ctx.options, + llm: &ctx.llm, + renderer: &ctx.renderer, + }) + .collect(); + + let results = async { engine.run_batch(ctxs).await } + .instrument(tracing::info_span!( + "step", + engine = stage.info.id, + pages = live.len() + )) + .await; + + for (item, result) in live.into_iter().zip(results) { + match result { + Ok(ops) => { + if !ops.is_empty() { + let batch = Op::Batch { + ops, + label: format!("{}: page {}", stage.info.id, item.page_id), + }; + if let Err(err) = ctx.session.apply(batch) { + report_step_failure( + stage.info.id, + &item.page_id, + seq, + item.page_index, + tracker.total_pages, + tracker.total_steps, + &err, + &ctx.tracker.warnings, + ctx.warnings.as_ref(), + ); + tracker.credit(item.page_index, remaining_steps); + continue; + } + } + tracker.credit(item.page_index, 1); + // Not forwarding is how a page stops: the last stage has + // no downstream, and a dropped item frees its permit. + if let Some(tx) = forward.as_ref() + && tx.send(item).await.is_err() + { + return; + } + } + Err(err) => { + report_step_failure( + stage.info.id, + &item.page_id, + seq, + item.page_index, + tracker.total_pages, + tracker.total_steps, + &err, + &ctx.tracker.warnings, + ctx.warnings.as_ref(), + ); + // Later steps almost always consume this step's artifact, + // so the page stops here and its remaining steps are + // credited to keep progress reaching 100%. + tracker.credit(item.page_index, remaining_steps); + } + } + } + } } #[allow(clippy::too_many_arguments)] @@ -283,7 +762,7 @@ fn report_step_failure( total_pages: usize, total_steps: usize, err: &anyhow::Error, - warning_count: &mut usize, + warning_count: &std::sync::atomic::AtomicUsize, sink: Option<&WarningSink>, ) { let _ = total_steps; @@ -293,7 +772,7 @@ fn report_step_failure( step_index, "pipeline step failed: {err:#}" ); - *warning_count += 1; + warning_count.fetch_add(1, Ordering::Relaxed); if let Some(sink) = sink { sink(WarningTick { step_id: engine_id.to_string(), @@ -367,4 +846,211 @@ mod tests { && engine.produces.iter().map(String::as_str).eq(["TextBoxes"]) })); } + + // --- batching policy: take what's queued, never wait ------------------- + + #[tokio::test] + async fn idle_queue_yields_a_batch_of_one() { + let (tx, rx) = async_channel::unbounded::(); + tx.send(1).await.unwrap(); + // Four more are allowed, but nothing else is queued, so the stage must + // not block waiting for them. + assert_eq!(drain_batch(&rx, 4).await, Some(vec![1])); + } + + #[tokio::test] + async fn backed_up_queue_is_drained_up_to_the_cap() { + let (tx, rx) = async_channel::unbounded::(); + for i in 1..=6 { + tx.send(i).await.unwrap(); + } + assert_eq!(drain_batch(&rx, 4).await, Some(vec![1, 2, 3, 4])); + // The remainder stays queued for the next pass. + assert_eq!(drain_batch(&rx, 4).await, Some(vec![5, 6])); + } + + #[tokio::test] + async fn batch_of_one_never_groups() { + let (tx, rx) = async_channel::unbounded::(); + for i in 1..=3 { + tx.send(i).await.unwrap(); + } + // `max_batch_pages = 1` / a non-batching engine must stay one-at-a-time + // even when the queue is full. + assert_eq!(drain_batch(&rx, 1).await, Some(vec![1])); + assert_eq!(drain_batch(&rx, 1).await, Some(vec![2])); + } + + #[tokio::test] + async fn closed_and_drained_channel_ends_the_stage() { + let (tx, rx) = async_channel::unbounded::(); + tx.send(1).await.unwrap(); + drop(tx); + // Items already queued are still delivered after close. + assert_eq!(drain_batch(&rx, 4).await, Some(vec![1])); + assert_eq!(drain_batch(&rx, 4).await, None); + } + + // --- progress accounting --------------------------------------------- + + #[test] + fn frontier_tracks_the_lowest_unfinished_page() { + let t = Tracker::new(3, 2); + assert_eq!(t.frontier(), 0); + + // Page 1 finishing first must not advance the frontier past page 0 — + // this is what keeps the UI's "Image N/M" monotonic while pages run + // out of order. + t.credit(1, 2); + assert_eq!(t.frontier(), 0); + + t.credit(0, 1); + assert_eq!(t.frontier(), 0); + t.credit(0, 1); + assert_eq!(t.frontier(), 2); + } + + #[test] + fn frontier_never_decreases_under_interleaved_completion() { + let t = Tracker::new(4, 3); + let mut last = t.frontier(); + // Deliberately out-of-order completions. + for (page, steps) in [(2, 3), (0, 1), (3, 3), (0, 2), (1, 3)] { + t.credit(page, steps); + let now = t.frontier(); + assert!(now >= last, "frontier went backwards: {last} -> {now}"); + last = now; + } + } + + #[test] + fn percent_reaches_100_when_every_step_is_credited() { + let t = Tracker::new(2, 3); + assert_eq!(t.percent(), 0); + t.credit(0, 3); + assert_eq!(t.percent(), 50); + // A page that fails early still credits its remaining steps, so a run + // with failures reaches 100 rather than stalling short. + t.credit(1, 3); + assert_eq!(t.percent(), 100); + } + + #[test] + fn percent_is_clamped_if_over_credited() { + let t = Tracker::new(1, 1); + t.credit(0, 5); + assert_eq!(t.percent(), 100); + } + + // --- stage pool: threads must outlive the stages they run -------------- + + /// One global pool, so these tests must not run against each other. + static POOL_TEST_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(()); + + thread_local! { + /// Stands in for candle's cuDNN handle: state that exists only as long + /// as the thread holding it does. + static STAGE_LOCAL_MARK: std::cell::Cell = const { std::cell::Cell::new(false) }; + } + + /// Submit one job and wait for its result. + fn run_on_pool(job: impl FnOnce() -> T + Send + 'static) -> T { + let (tx, rx) = std::sync::mpsc::channel(); + StagePool::get().submit(Box::new(move |_rt| { + let _ = tx.send(job()); + })); + rx.recv().expect("pooled job never reported") + } + + /// A job signals completion when its body ends, which is a moment before + /// its thread re-parks. Waiting for the park means the next submit claims + /// that thread rather than spawning another. + fn wait_for_parked_thread() { + for _ in 0..1_000 { + if StagePool::get().idle.load(Ordering::Acquire) > 0 { + return; + } + std::thread::sleep(std::time::Duration::from_millis(1)); + } + } + + /// The whole point of the pool: a finished stage parks its thread instead + /// of ending it, so candle's `thread_local!` cuDNN handles are never + /// destroyed by an unwrapping `Drop` and hyper's pooled connections keep + /// the runtime they were opened on. + #[test] + fn pooled_thread_keeps_thread_locals_across_stages() { + let _guard = POOL_TEST_LOCK.lock().unwrap_or_else(|e| e.into_inner()); + + wait_for_parked_thread(); + let marked = run_on_pool(|| { + STAGE_LOCAL_MARK.with(|m| m.set(true)); + std::thread::current().id() + }); + + // Land on that thread again. Had it ended with its stage, the mark + // would have gone with it. + for _ in 0..200 { + wait_for_parked_thread(); + let (id, still_marked) = run_on_pool(|| { + ( + std::thread::current().id(), + STAGE_LOCAL_MARK.with(|m| m.get()), + ) + }); + if id == marked { + assert!(still_marked, "pooled thread lost its thread-local state"); + return; + } + } + panic!("stage thread {marked:?} never took another job; it did not survive its stage"); + } + + /// A panicking stage must not take its thread — and so that thread's CUDA + /// state — down with it. + #[test] + fn panicking_stage_does_not_kill_its_pooled_thread() { + let _guard = POOL_TEST_LOCK.lock().unwrap_or_else(|e| e.into_inner()); + + let (done, wait) = std::sync::mpsc::channel::<()>(); + StagePool::get().submit(Box::new(move |_rt| { + let _done = done; + panic!("stage blew up"); + })); + // The dropped sender releases the driver whether the stage returned or + // unwound, which is what a panicked `JoinHandle` gave it before. + assert!( + wait.recv().is_err(), + "panicking stage should not signal success" + ); + + // The pool still serves work. + assert_eq!(run_on_pool(|| 7), 7); + } + + /// Concurrent stages must never be handed the same parked thread: an + /// upstream stage blocked on a bounded channel would deadlock waiting on a + /// downstream stage that has nowhere to run. + #[test] + fn concurrent_stages_each_get_their_own_thread() { + let _guard = POOL_TEST_LOCK.lock().unwrap_or_else(|e| e.into_inner()); + + let barrier = Arc::new(std::sync::Barrier::new(4)); + let mut waits = Vec::new(); + + for _ in 0..4 { + let barrier = barrier.clone(); + let (done, wait) = std::sync::mpsc::channel::<()>(); + StagePool::get().submit(Box::new(move |_rt| { + let _done = done; + // Only completes if all four are running at once. + barrier.wait(); + })); + waits.push(wait); + } + + for wait in waits { + let _ = wait.recv(); + } + } } diff --git a/crates/koharu-core/src/protocol.rs b/crates/koharu-core/src/protocol.rs index 9598b8673..d1ad75bc8 100644 --- a/crates/koharu-core/src/protocol.rs +++ b/crates/koharu-core/src/protocol.rs @@ -31,6 +31,9 @@ pub struct FontFaceInfo { pub struct MetaInfo { pub version: String, pub ml_device: String, + /// Workers a CPU-bound pipeline stage gets under auto sizing. Depends on + /// the host's core count, so only the server can report it. + pub cpu_workers: usize, } // --------------------------------------------------------------------------- @@ -231,6 +234,12 @@ pub struct PipelineConfigPatch { pub translator: Option, pub inpainter: Option, pub renderer: Option, + /// Pages allowed in the pipeline at once. `0` = auto (one per step), `1` = + /// fully sequential. + pub max_inflight_pages: Option, + /// Cap on pages a stage folds into one model call. `0` = auto, `1` = + /// no batching. + pub max_batch_pages: Option, } #[derive(Debug, Clone, Serialize, Deserialize, JsonSchema, ToSchema)] diff --git a/crates/koharu-llm/src/prompt.rs b/crates/koharu-llm/src/prompt.rs index b89ad82f3..9fab2035a 100644 --- a/crates/koharu-llm/src/prompt.rs +++ b/crates/koharu-llm/src/prompt.rs @@ -47,7 +47,11 @@ pub struct PromptRenderer { eos_token: String, } -pub const BLOCK_TAG_INSTRUCTIONS: &str = "The input uses numbered tags like [1], [2], etc. to mark each text block. Translate only the text after each tag. Keep every tag exactly unchanged, including numbers and order. Output the same tags followed by the translated text. Do not merge, split, or reorder blocks."; +/// Deliberately format-agnostic: the same instruction covers the single-page +/// `[N]` tags and the page-qualified `[bPAGE-BLOCK]` tags used when several +/// pages share one request. "add, or omit" matters — a dropped or invented +/// tag is what shifts translations onto the wrong bubble. +pub const BLOCK_TAG_INSTRUCTIONS: &str = "The input marks each text block with a bracketed tag such as [1] or [b2-3]. Translate only the text after each tag. Reproduce every tag exactly as given, unchanged and in the same order. Output the same tags, each followed by the translated text. Do not merge, split, reorder, add, or omit blocks."; pub fn system_prompt(target_language: Language) -> String { format!( @@ -133,8 +137,12 @@ mod tests { fn system_prompt_mentions_target_language_and_block_rules() { let prompt = system_prompt(Language::Korean); assert!(prompt.contains("natural Korean")); - assert!(prompt.contains("[1], [2]")); + // Both tag forms must be described: single-page `[N]` and the + // page-qualified `[bPAGE-BLOCK]` used for batched requests. + assert!(prompt.contains("[1]")); + assert!(prompt.contains("[b2-3]")); assert!(prompt.contains("Do not merge")); + assert!(prompt.contains("omit")); } #[test] diff --git a/crates/koharu-rpc/src/mcp/mod.rs b/crates/koharu-rpc/src/mcp/mod.rs index 0544cd58c..7ff3f26b0 100644 --- a/crates/koharu-rpc/src/mcp/mod.rs +++ b/crates/koharu-rpc/src/mcp/mod.rs @@ -185,6 +185,7 @@ impl KoharuServer { None => Scope::WholeProject, }, steps: input.steps, + limits: Default::default(), options: PipelineRunOptions { target_language: input.target_language, system_prompt: input.system_prompt, diff --git a/crates/koharu-rpc/src/routes/meta.rs b/crates/koharu-rpc/src/routes/meta.rs index 68e74bd89..0985bfeac 100644 --- a/crates/koharu-rpc/src/routes/meta.rs +++ b/crates/koharu-rpc/src/routes/meta.rs @@ -19,6 +19,7 @@ async fn get_meta(State(app): State) -> ApiResult> { Ok(Json(MetaInfo { version: app.version.to_string(), ml_device: device_label(&app), + cpu_workers: koharu_app::pipeline::auto_cpu_workers(), })) } diff --git a/crates/koharu-rpc/src/routes/pipelines.rs b/crates/koharu-rpc/src/routes/pipelines.rs index 23f50d327..b8f6e0f9d 100644 --- a/crates/koharu-rpc/src/routes/pipelines.rs +++ b/crates/koharu-rpc/src/routes/pipelines.rs @@ -10,7 +10,7 @@ use std::sync::atomic::AtomicBool; use axum::Json; use axum::extract::State; use koharu_app::pipeline::{ - self, PipelineRunOptions, PipelineSpec, ProgressTick, Scope, WarningTick, + self, PipelineLimits, PipelineRunOptions, PipelineSpec, ProgressTick, Scope, WarningTick, }; use koharu_core::{ AppEvent, JobFinishedEvent, JobStatus, JobSummary, JobWarningEvent, NodeId, PageId, @@ -75,12 +75,20 @@ async fn start_pipeline( for id in &req.steps { pipeline::Registry::find(id).map_err(|e| ApiError::bad_request(format!("{e:#}")))?; } + let limits = { + let cfg = app.config.load(); + PipelineLimits { + max_inflight_pages: cfg.pipeline.max_inflight_pages, + max_batch_pages: cfg.pipeline.max_batch_pages, + } + }; let spec = PipelineSpec { scope: match req.pages { Some(pages) => Scope::Pages(pages), None => Scope::WholeProject, }, steps: req.steps, + limits, options: PipelineRunOptions { target_language: req.target_language, system_prompt: req.system_prompt, diff --git a/ui/components/SettingsDialog.tsx b/ui/components/SettingsDialog.tsx index 978e88058..2abc6d63f 100644 --- a/ui/components/SettingsDialog.tsx +++ b/ui/components/SettingsDialog.tsx @@ -115,6 +115,8 @@ function appConfigToPatch(cfg: AppConfig): ConfigPatch { translator: cfg.pipeline.translator, inpainter: cfg.pipeline.inpainter, renderer: cfg.pipeline.renderer, + maxInflightPages: cfg.pipeline.max_inflight_pages ?? 0, + maxBatchPages: cfg.pipeline.max_batch_pages ?? 0, } } if (cfg.providers) { @@ -174,10 +176,13 @@ export function SettingsDialog({ const [httpConnectTimeoutDraft, setHttpConnectTimeoutDraft] = useState('') const [httpReadTimeoutDraft, setHttpReadTimeoutDraft] = useState('') const [httpMaxRetriesDraft, setHttpMaxRetriesDraft] = useState('') + const [maxInflightPagesDraft, setMaxInflightPagesDraft] = useState('') + const [maxBatchPagesDraft, setMaxBatchPagesDraft] = useState('') const [storageSettingsError, setStorageSettingsError] = useState(null) const [isSavingStorageSettings, setIsSavingStorageSettings] = useState(false) const [engineCatalog, setEngineCatalog] = useState(null) const [appVersion, setAppVersion] = useState() + const [cpuWorkers, setCpuWorkers] = useState() const updater = useUpdater() useEffect(() => { @@ -204,6 +209,7 @@ export function SettingsDialog({ const meta = await getMeta() if (cancelled) return setAppVersion(meta.version) + setCpuWorkers(meta.cpuWorkers) } catch { return } @@ -227,6 +233,8 @@ export function SettingsDialog({ ) setHttpReadTimeoutDraft(String(appConfig.http?.read_timeout ?? DEFAULT_HTTP_READ_TIMEOUT)) setHttpMaxRetriesDraft(String(appConfig.http?.max_retries ?? DEFAULT_HTTP_MAX_RETRIES)) + setMaxInflightPagesDraft(String(appConfig.pipeline?.max_inflight_pages ?? 0)) + setMaxBatchPagesDraft(String(appConfig.pipeline?.max_batch_pages ?? 0)) setStorageSettingsError(null) }, [appConfig]) @@ -243,6 +251,43 @@ export function SettingsDialog({ } } + // What `max_inflight_pages = 0` resolves to: the driver allows one page per + // step, and a full run is every configured stage — the same list MenuBar + // sends when translating a project. + const autoInflightPages = useMemo(() => { + const p = appConfig?.pipeline + if (!p) return 0 + return [ + p.detector, + p.segmenter, + p.bubble_segmenter, + p.font_detector, + p.ocr, + p.translator, + p.inpainter, + p.renderer, + ].filter(Boolean).length + }, [appConfig]) + + /// Parse on blur, snapping an unusable entry back to what is stored rather + /// than persisting garbage. Unlike the storage settings these need no + /// restart, so they save as soon as the field is left. + const commitPipelineLimit = (field: 'max_inflight_pages' | 'max_batch_pages', raw: string) => { + if (!appConfig?.pipeline) return + const current = appConfig.pipeline[field] ?? 0 + const parsed = Number.parseInt(raw.trim(), 10) + const next = Number.isInteger(parsed) && parsed >= 0 ? parsed : current + + if (field === 'max_inflight_pages') setMaxInflightPagesDraft(String(next)) + else setMaxBatchPagesDraft(String(next)) + + if (next === current) return + void persistConfig({ + ...appConfig, + pipeline: { ...appConfig.pipeline, [field]: next }, + }) + } + const upsertProvider = (id: string, updater: (p: ProviderConfig) => ProviderConfig) => { if (!appConfig) return const providers = [...(appConfig.providers ?? [])] @@ -399,32 +444,48 @@ export function SettingsDialog({ )} {tab === 'ai' && } {tab === 'runtime' && ( - { - setDataPathDraft(v) - setStorageSettingsError(null) - }} - onHttpConnectTimeoutChange={(v) => { - setHttpConnectTimeoutDraft(v) - setStorageSettingsError(null) - }} - onHttpReadTimeoutChange={(v) => { - setHttpReadTimeoutDraft(v) - setStorageSettingsError(null) - }} - onHttpMaxRetriesChange={(v) => { - setHttpMaxRetriesDraft(v) - setStorageSettingsError(null) - }} - onApply={() => void handleApplyStorageSettings()} - /> +
+ { + setDataPathDraft(v) + setStorageSettingsError(null) + }} + onHttpConnectTimeoutChange={(v) => { + setHttpConnectTimeoutDraft(v) + setStorageSettingsError(null) + }} + onHttpReadTimeoutChange={(v) => { + setHttpReadTimeoutDraft(v) + setStorageSettingsError(null) + }} + onHttpMaxRetriesChange={(v) => { + setHttpMaxRetriesDraft(v) + setStorageSettingsError(null) + }} + onApply={() => void handleApplyStorageSettings()} + /> + + commitPipelineLimit('max_inflight_pages', maxInflightPagesDraft) + } + onMaxBatchPagesChange={setMaxBatchPagesDraft} + onMaxBatchPagesCommit={() => + commitPipelineLimit('max_batch_pages', maxBatchPagesDraft) + } + /> +
)} {tab === 'keybinds' && } {tab === 'about' && ( @@ -1263,6 +1324,87 @@ function StoragePane({ ) } +// ── Parallel processing ─────────────────────────────────────────── + +function ParallelPane({ + maxInflightPages, + maxBatchPages, + autoInflightPages, + cpuWorkers, + onMaxInflightPagesChange, + onMaxInflightPagesCommit, + onMaxBatchPagesChange, + onMaxBatchPagesCommit, +}: { + maxInflightPages: string + maxBatchPages: string + autoInflightPages: number + cpuWorkers: number | undefined + onMaxInflightPagesChange: (v: string) => void + onMaxInflightPagesCommit: () => void + onMaxBatchPagesChange: (v: string) => void + onMaxBatchPagesCommit: () => void +}) { + const { t } = useTranslation() + // `0` is auto, so that is when the resolved value is worth showing. + const inflightIsAuto = maxInflightPages.trim() === '0' + const batchIsAuto = maxBatchPages.trim() === '0' + + return ( +
+
+
+ + onMaxInflightPagesChange(e.target.value)} + onBlur={onMaxInflightPagesCommit} + /> +

+ {t('settings.maxInflightPagesDescription')} +

+ {inflightIsAuto && autoInflightPages > 0 && ( +

+ {t('settings.maxInflightPagesAuto', { pages: autoInflightPages })} +

+ )} +
+ +
+ + onMaxBatchPagesChange(e.target.value)} + onBlur={onMaxBatchPagesCommit} + /> +

+ {t('settings.maxBatchPagesDescription')} +

+ {batchIsAuto && ( +

+ {t('settings.maxBatchPagesAuto')} +

+ )} +
+
+ + {cpuWorkers !== undefined && ( +

+ {t('settings.cpuWorkersInfo', { workers: cpuWorkers })} +

+ )} +
+ ) +} + // ── About ───────────────────────────────────────────────────────── function AboutPane({ diff --git a/ui/lib/api/default/default.msw.ts b/ui/lib/api/default/default.msw.ts index ba6cd39a6..6ff1fb17b 100644 --- a/ui/lib/api/default/default.msw.ts +++ b/ui/lib/api/default/default.msw.ts @@ -653,6 +653,7 @@ export const getGetCurrentLlmResponseMock = ( export const getGetMetaResponseMock = ( overrideResponse: Partial> = {}, ): MetaInfo => ({ + cpuWorkers: faker.number.int({ min: 0 }), mlDevice: faker.string.alpha({ length: { min: 10, max: 20 } }), version: faker.string.alpha({ length: { min: 10, max: 20 } }), ...overrideResponse, diff --git a/ui/lib/api/schemas/metaInfo.ts b/ui/lib/api/schemas/metaInfo.ts index 6c9d7329f..c1726213f 100644 --- a/ui/lib/api/schemas/metaInfo.ts +++ b/ui/lib/api/schemas/metaInfo.ts @@ -1,10 +1,15 @@ /** - * Generated by orval v8.8.1 🍺 + * Generated by orval v8.19.0 🍺 * Do not edit manually. - * OpenAPI spec version: 0.0.1 */ export interface MetaInfo { + /** + * Workers a CPU-bound pipeline stage gets under auto sizing. Depends on + * the host's core count, so only the server can report it. + * @minimum 0 + */ + cpuWorkers: number mlDevice: string version: string } diff --git a/ui/lib/api/schemas/pipelineConfig.ts b/ui/lib/api/schemas/pipelineConfig.ts index fcdd600fd..0977e91b0 100644 --- a/ui/lib/api/schemas/pipelineConfig.ts +++ b/ui/lib/api/schemas/pipelineConfig.ts @@ -1,19 +1,37 @@ /** - * Generated by orval v8.8.1 🍺 + * Generated by orval v8.19.0 🍺 * Do not edit manually. - * OpenAPI spec version: 0.0.1 */ /** * Engine selection for each pipeline stage. -Values are engine IDs (e.g. "pp-doclayout-v3", "comic-text-detector"). -Empty string means use default. + * Values are engine IDs (e.g. "pp-doclayout-v3", "comic-text-detector"). + * Empty string means use default. */ export interface PipelineConfig { bubble_segmenter?: string detector?: string font_detector?: string inpainter?: string + /** + * Upper bound on pages any single stage folds into one model call. + * `0` = auto, `1` = disable batching but keep stage overlap. + * + * Batch size is the larger VRAM lever of the two, so try lowering this + * before `max_inflight_pages`. + * @minimum 0 + */ + max_batch_pages?: number + /** + * Maximum pages moving through the pipeline at once. `0` = auto (one per + * stage, so every stage can stay busy). + * + * Set to `1` to restore fully sequential processing: one page finishes + * every step before the next starts, and no stage ever batches. That is + * the escape hatch if parallelism causes trouble. + * @minimum 0 + */ + max_inflight_pages?: number ocr?: string renderer?: string segmenter?: string diff --git a/ui/lib/api/schemas/pipelineConfigPatch.ts b/ui/lib/api/schemas/pipelineConfigPatch.ts index ddc1b406e..4c8f3edb6 100644 --- a/ui/lib/api/schemas/pipelineConfigPatch.ts +++ b/ui/lib/api/schemas/pipelineConfigPatch.ts @@ -1,7 +1,6 @@ /** - * Generated by orval v8.8.1 🍺 + * Generated by orval v8.19.0 🍺 * Do not edit manually. - * OpenAPI spec version: 0.0.1 */ export interface PipelineConfigPatch { @@ -13,6 +12,20 @@ export interface PipelineConfigPatch { fontDetector?: string | null /** @nullable */ inpainter?: string | null + /** + * Cap on pages a stage folds into one model call. `0` = auto, `1` = + * no batching. + * @minimum 0 + * @nullable + */ + maxBatchPages?: number | null + /** + * Pages allowed in the pipeline at once. `0` = auto (one per step), `1` = + * fully sequential. + * @minimum 0 + * @nullable + */ + maxInflightPages?: number | null /** @nullable */ ocr?: string | null /** @nullable */ diff --git a/ui/openapi.json b/ui/openapi.json index 7433e0084..f1cd920c3 100644 --- a/ui/openapi.json +++ b/ui/openapi.json @@ -650,7 +650,10 @@ "description": "Optional pipeline engine to run after the mask is updated.", "required": false, "schema": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] } }, { @@ -659,7 +662,10 @@ "description": "Bounding box for the pipeline run.", "required": false, "schema": { - "type": ["number", "null"], + "type": [ + "number", + "null" + ], "format": "float" } }, @@ -668,7 +674,10 @@ "in": "query", "required": false, "schema": { - "type": ["number", "null"], + "type": [ + "number", + "null" + ], "format": "float" } }, @@ -677,7 +686,10 @@ "in": "query", "required": false, "schema": { - "type": ["number", "null"], + "type": [ + "number", + "null" + ], "format": "float" } }, @@ -686,7 +698,10 @@ "in": "query", "required": false, "schema": { - "type": ["number", "null"], + "type": [ + "number", + "null" + ], "format": "float" } } @@ -979,7 +994,9 @@ "schemas": { "AddImageLayerResponse": { "type": "object", - "required": ["node"], + "required": [ + "node" + ], "properties": { "node": { "$ref": "#/components/schemas/NodeId" @@ -996,7 +1013,7 @@ } ], "default": { - "path": "C:\\Users\\Mayo\\AppData\\Local\\Koharu" + "path": "C:\\Users\\Bekeon\\AppData\\Local\\Koharu" } }, "http": { @@ -1022,7 +1039,9 @@ "detector": "pp-doclayout-v3", "font_detector": "yuzumarker-font-detection", "inpainter": "lama-manga", - "ocr": "paddle-ocr-vl-1.5", + "max_batch_pages": 0, + "max_inflight_pages": 0, + "ocr": "paddle-ocr-vl-1.6", "renderer": "koharu-renderer", "segmenter": "comic-text-detector-seg", "translator": "llm" @@ -1041,11 +1060,17 @@ "oneOf": [ { "type": "object", - "required": ["id", "kind", "event"], + "required": [ + "id", + "kind", + "event" + ], "properties": { "event": { "type": "string", - "enum": ["jobStarted"] + "enum": [ + "jobStarted" + ] }, "id": { "type": "string" @@ -1062,11 +1087,15 @@ }, { "type": "object", - "required": ["event"], + "required": [ + "event" + ], "properties": { "event": { "type": "string", - "enum": ["jobProgress"] + "enum": [ + "jobProgress" + ] } } } @@ -1080,11 +1109,15 @@ }, { "type": "object", - "required": ["event"], + "required": [ + "event" + ], "properties": { "event": { "type": "string", - "enum": ["jobWarning"] + "enum": [ + "jobWarning" + ] } } } @@ -1098,11 +1131,15 @@ }, { "type": "object", - "required": ["event"], + "required": [ + "event" + ], "properties": { "event": { "type": "string", - "enum": ["jobFinished"] + "enum": [ + "jobFinished" + ] } } } @@ -1115,11 +1152,15 @@ }, { "type": "object", - "required": ["event"], + "required": [ + "event" + ], "properties": { "event": { "type": "string", - "enum": ["downloadProgress"] + "enum": [ + "downloadProgress" + ] } } } @@ -1127,11 +1168,16 @@ }, { "type": "object", - "required": ["target", "event"], + "required": [ + "target", + "event" + ], "properties": { "event": { "type": "string", - "enum": ["llmLoading"] + "enum": [ + "llmLoading" + ] }, "target": { "$ref": "#/components/schemas/LlmTarget" @@ -1140,11 +1186,16 @@ }, { "type": "object", - "required": ["target", "event"], + "required": [ + "target", + "event" + ], "properties": { "event": { "type": "string", - "enum": ["llmLoaded"] + "enum": [ + "llmLoaded" + ] }, "target": { "$ref": "#/components/schemas/LlmTarget" @@ -1153,11 +1204,15 @@ }, { "type": "object", - "required": ["event"], + "required": [ + "event" + ], "properties": { "event": { "type": "string", - "enum": ["llmFailed"] + "enum": [ + "llmFailed" + ] }, "target": { "oneOf": [ @@ -1173,11 +1228,15 @@ }, { "type": "object", - "required": ["event"], + "required": [ + "event" + ], "properties": { "event": { "type": "string", - "enum": ["llmUnloaded"] + "enum": [ + "llmUnloaded" + ] } } }, @@ -1188,11 +1247,15 @@ }, { "type": "object", - "required": ["event"], + "required": [ + "event" + ], "properties": { "event": { "type": "string", - "enum": ["snapshot"] + "enum": [ + "snapshot" + ] } } } @@ -1206,14 +1269,23 @@ }, "CodexAuthAttemptStatus": { "type": "string", - "enum": ["pending", "succeeded", "failed"] + "enum": [ + "pending", + "succeeded", + "failed" + ] }, "CodexAuthStatus": { "type": "object", - "required": ["signedIn"], + "required": [ + "signedIn" + ], "properties": { "accountId": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] }, "login": { "oneOf": [ @@ -1232,7 +1304,13 @@ }, "CodexDeviceLogin": { "type": "object", - "required": ["loginId", "verificationUrl", "userCode", "intervalSeconds", "timeoutSeconds"], + "required": [ + "loginId", + "verificationUrl", + "userCode", + "intervalSeconds", + "timeoutSeconds" + ], "properties": { "intervalSeconds": { "type": "integer", @@ -1257,13 +1335,22 @@ }, "CodexDeviceLoginStatus": { "type": "object", - "required": ["loginId", "status"], + "required": [ + "loginId", + "status" + ], "properties": { "accountId": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] }, "error": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] }, "loginId": { "type": "string" @@ -1275,13 +1362,22 @@ }, "CodexImageGenerationOptions": { "type": "object", - "required": ["pageId", "prompt"], + "required": [ + "pageId", + "prompt" + ], "properties": { "instructions": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] }, "model": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] }, "pageId": { "$ref": "#/components/schemas/PageId" @@ -1290,16 +1386,24 @@ "type": "string" }, "quality": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] }, "size": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] } } }, "CodexImageGenerationResponse": { "type": "object", - "required": ["operationId"], + "required": [ + "operationId" + ], "properties": { "operationId": { "type": "string" @@ -1341,7 +1445,10 @@ ] }, "providers": { - "type": ["array", "null"], + "type": [ + "array", + "null" + ], "items": { "$ref": "#/components/schemas/ProviderPatch" }, @@ -1351,7 +1458,9 @@ }, "CreatePagesFromPathsRequest": { "type": "object", - "required": ["paths"], + "required": [ + "paths" + ], "properties": { "paths": { "type": "array", @@ -1366,7 +1475,9 @@ }, "CreatePagesResponse": { "type": "object", - "required": ["pages"], + "required": [ + "pages" + ], "properties": { "pages": { "type": "array", @@ -1378,7 +1489,9 @@ }, "CreateProjectRequest": { "type": "object", - "required": ["name"], + "required": [ + "name" + ], "properties": { "name": { "type": "string" @@ -1387,7 +1500,9 @@ }, "DataConfig": { "type": "object", - "required": ["path"], + "required": [ + "path" + ], "properties": { "path": { "type": "string" @@ -1398,13 +1513,21 @@ "type": "object", "properties": { "path": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] } } }, "DownloadProgress": { "type": "object", - "required": ["id", "filename", "downloaded", "status"], + "required": [ + "id", + "filename", + "downloaded", + "status" + ], "properties": { "downloaded": { "type": "integer", @@ -1421,7 +1544,10 @@ "$ref": "#/components/schemas/DownloadStatus" }, "total": { - "type": ["integer", "null"], + "type": [ + "integer", + "null" + ], "format": "int64", "minimum": 0 } @@ -1431,44 +1557,61 @@ "oneOf": [ { "type": "object", - "required": ["status"], + "required": [ + "status" + ], "properties": { "status": { "type": "string", - "enum": ["started"] + "enum": [ + "started" + ] } } }, { "type": "object", - "required": ["status"], + "required": [ + "status" + ], "properties": { "status": { "type": "string", - "enum": ["downloading"] + "enum": [ + "downloading" + ] } } }, { "type": "object", - "required": ["status"], + "required": [ + "status" + ], "properties": { "status": { "type": "string", - "enum": ["completed"] + "enum": [ + "completed" + ] } } }, { "type": "object", - "required": ["reason", "status"], + "required": [ + "reason", + "status" + ], "properties": { "reason": { "type": "string" }, "status": { "type": "string", - "enum": ["failed"] + "enum": [ + "failed" + ] } } } @@ -1539,7 +1682,11 @@ }, "EngineCatalogEntry": { "type": "object", - "required": ["id", "name", "produces"], + "required": [ + "id", + "name", + "produces" + ], "properties": { "id": { "type": "string" @@ -1557,21 +1704,34 @@ }, "ExportFormat": { "type": "string", - "enum": ["khr", "psd", "rendered", "inpainted"] + "enum": [ + "khr", + "psd", + "rendered", + "inpainted" + ] }, "ExportProjectRequest": { "type": "object", - "required": ["format"], + "required": [ + "format" + ], "properties": { "defaultFont": { - "type": ["string", "null"], + "type": [ + "string", + "null" + ], "description": "Optional global font override (from UI preferences)." }, "format": { "$ref": "#/components/schemas/ExportFormat" }, "pages": { - "type": ["array", "null"], + "type": [ + "array", + "null" + ], "items": { "$ref": "#/components/schemas/PageId" }, @@ -1581,13 +1741,21 @@ }, "FontFaceInfo": { "type": "object", - "required": ["familyName", "postScriptName", "source", "cached"], + "required": [ + "familyName", + "postScriptName", + "source", + "cached" + ], "properties": { "cached": { "type": "boolean" }, "category": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] }, "familyName": { "type": "string" @@ -1665,11 +1833,16 @@ }, "FontSource": { "type": "string", - "enum": ["system", "google"] + "enum": [ + "system", + "google" + ] }, "GoogleFontCatalog": { "type": "object", - "required": ["fonts"], + "required": [ + "fonts" + ], "properties": { "fonts": { "type": "array", @@ -1681,7 +1854,12 @@ }, "GoogleFontEntry": { "type": "object", - "required": ["family", "category", "subsets", "variants"], + "required": [ + "family", + "category", + "subsets", + "variants" + ], "properties": { "category": { "type": "string" @@ -1705,7 +1883,11 @@ }, "GoogleFontVariant": { "type": "object", - "required": ["style", "weight", "filename"], + "required": [ + "style", + "weight", + "filename" + ], "properties": { "filename": { "type": "string" @@ -1724,7 +1906,10 @@ "type": "object", "properties": { "epoch": { - "type": ["integer", "null"], + "type": [ + "integer", + "null" + ], "format": "int64", "description": "New epoch. `None` only for a no-op undo/redo at the stack boundary.", "minimum": 0 @@ -1758,17 +1943,26 @@ "type": "object", "properties": { "connectTimeout": { - "type": ["integer", "null"], + "type": [ + "integer", + "null" + ], "format": "int64", "minimum": 0 }, "maxRetries": { - "type": ["integer", "null"], + "type": [ + "integer", + "null" + ], "format": "int32", "minimum": 0 }, "readTimeout": { - "type": ["integer", "null"], + "type": [ + "integer", + "null" + ], "format": "int64", "minimum": 0 } @@ -1776,13 +1970,21 @@ }, "ImageData": { "type": "object", - "required": ["role", "blob", "naturalWidth", "naturalHeight"], + "required": [ + "role", + "blob", + "naturalWidth", + "naturalHeight" + ], "properties": { "blob": { "$ref": "#/components/schemas/BlobRef" }, "name": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] }, "naturalHeight": { "type": "integer", @@ -1818,34 +2020,57 @@ ] }, "name": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] }, "naturalHeight": { - "type": ["integer", "null"], + "type": [ + "integer", + "null" + ], "format": "int32", "minimum": 0 }, "naturalWidth": { - "type": ["integer", "null"], + "type": [ + "integer", + "null" + ], "format": "int32", "minimum": 0 }, "opacity": { - "type": ["number", "null"], + "type": [ + "number", + "null" + ], "format": "float" } } }, "ImageRole": { "type": "string", - "enum": ["source", "inpainted", "rendered", "custom"] + "enum": [ + "source", + "inpainted", + "rendered", + "custom" + ] }, "JobFinishedEvent": { "type": "object", - "required": ["id", "status"], + "required": [ + "id", + "status" + ], "properties": { "error": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] }, "id": { "type": "string" @@ -1857,14 +2082,27 @@ }, "JobStatus": { "type": "string", - "enum": ["running", "completed", "completed_with_errors", "cancelled", "failed"] + "enum": [ + "running", + "completed", + "completed_with_errors", + "cancelled", + "failed" + ] }, "JobSummary": { "type": "object", - "required": ["id", "kind", "status"], + "required": [ + "id", + "kind", + "status" + ], "properties": { "error": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] }, "id": { "type": "string" @@ -1880,7 +2118,13 @@ "JobWarningEvent": { "type": "object", "description": "A non-fatal step failure during a pipeline run. The pipeline recovers by\nskipping the rest of the current page's steps and moving on to the next\npage; the UI accumulates these into a list during the job.", - "required": ["jobId", "pageIndex", "totalPages", "stepId", "message"], + "required": [ + "jobId", + "pageIndex", + "totalPages", + "stepId", + "message" + ], "properties": { "jobId": { "type": "string" @@ -1905,7 +2149,9 @@ }, "ListDownloadsResponse": { "type": "object", - "required": ["downloads"], + "required": [ + "downloads" + ], "properties": { "downloads": { "type": "array", @@ -1917,7 +2163,9 @@ }, "ListOperationsResponse": { "type": "object", - "required": ["operations"], + "required": [ + "operations" + ], "properties": { "operations": { "type": "array", @@ -1929,7 +2177,9 @@ }, "ListProjectsResponse": { "type": "object", - "required": ["projects"], + "required": [ + "projects" + ], "properties": { "projects": { "type": "array", @@ -1941,7 +2191,10 @@ }, "LlmCatalog": { "type": "object", - "required": ["localModels", "providers"], + "required": [ + "localModels", + "providers" + ], "properties": { "localModels": { "type": "array", @@ -1959,7 +2212,11 @@ }, "LlmCatalogModel": { "type": "object", - "required": ["target", "name", "languages"], + "required": [ + "target", + "name", + "languages" + ], "properties": { "languages": { "type": "array", @@ -1979,22 +2236,33 @@ "type": "object", "properties": { "customSystemPrompt": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] }, "maxTokens": { - "type": ["integer", "null"], + "type": [ + "integer", + "null" + ], "format": "int32", "minimum": 0 }, "temperature": { - "type": ["number", "null"], + "type": [ + "number", + "null" + ], "format": "double" } } }, "LlmLoadRequest": { "type": "object", - "required": ["target"], + "required": [ + "target" + ], "properties": { "options": { "oneOf": [ @@ -2024,10 +2292,16 @@ ], "properties": { "baseUrl": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] }, "error": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] }, "hasApiKey": { "type": "boolean" @@ -2057,14 +2331,23 @@ }, "LlmProviderCatalogStatus": { "type": "string", - "enum": ["ready", "missing_configuration", "discovery_failed"] + "enum": [ + "ready", + "missing_configuration", + "discovery_failed" + ] }, "LlmState": { "type": "object", - "required": ["status"], + "required": [ + "status" + ], "properties": { "error": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] }, "status": { "$ref": "#/components/schemas/LlmStateStatus" @@ -2083,11 +2366,19 @@ }, "LlmStateStatus": { "type": "string", - "enum": ["empty", "loading", "ready", "failed"] + "enum": [ + "empty", + "loading", + "ready", + "failed" + ] }, "LlmTarget": { "type": "object", - "required": ["kind", "modelId"], + "required": [ + "kind", + "modelId" + ], "properties": { "kind": { "$ref": "#/components/schemas/LlmTargetKind" @@ -2096,17 +2387,26 @@ "type": "string" }, "providerId": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] } } }, "LlmTargetKind": { "type": "string", - "enum": ["local", "provider"] + "enum": [ + "local", + "provider" + ] }, "MaskData": { "type": "object", - "required": ["role", "blob"], + "required": [ + "role", + "blob" + ], "properties": { "blob": { "$ref": "#/components/schemas/BlobRef" @@ -2133,12 +2433,25 @@ }, "MaskRole": { "type": "string", - "enum": ["brushInpaint", "segment", "bubble"] + "enum": [ + "brushInpaint", + "segment", + "bubble" + ] }, "MetaInfo": { "type": "object", - "required": ["version", "mlDevice"], + "required": [ + "version", + "mlDevice", + "cpuWorkers" + ], "properties": { + "cpuWorkers": { + "type": "integer", + "description": "Workers a CPU-bound pipeline stage gets under auto sizing. Depends on\nthe host's core count, so only the server can report it.", + "minimum": 0 + }, "mlDevice": { "type": "string" }, @@ -2149,14 +2462,22 @@ }, "NamedFontPrediction": { "type": "object", - "required": ["index", "name", "probability", "serif"], + "required": [ + "index", + "name", + "probability", + "serif" + ], "properties": { "index": { "type": "integer", "minimum": 0 }, "language": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] }, "name": { "type": "string" @@ -2172,7 +2493,11 @@ }, "Node": { "type": "object", - "required": ["id", "visible", "kind"], + "required": [ + "id", + "visible", + "kind" + ], "properties": { "id": { "$ref": "#/components/schemas/NodeId" @@ -2192,7 +2517,9 @@ "oneOf": [ { "type": "object", - "required": ["text"], + "required": [ + "text" + ], "properties": { "text": { "$ref": "#/components/schemas/TextDataPatch" @@ -2201,7 +2528,9 @@ }, { "type": "object", - "required": ["image"], + "required": [ + "image" + ], "properties": { "image": { "$ref": "#/components/schemas/ImageDataPatch" @@ -2210,7 +2539,9 @@ }, { "type": "object", - "required": ["mask"], + "required": [ + "mask" + ], "properties": { "mask": { "$ref": "#/components/schemas/MaskDataPatch" @@ -2227,7 +2558,9 @@ "oneOf": [ { "type": "object", - "required": ["image"], + "required": [ + "image" + ], "properties": { "image": { "$ref": "#/components/schemas/ImageData" @@ -2236,7 +2569,9 @@ }, { "type": "object", - "required": ["text"], + "required": [ + "text" + ], "properties": { "text": { "$ref": "#/components/schemas/TextData" @@ -2245,7 +2580,9 @@ }, { "type": "object", - "required": ["mask"], + "required": [ + "mask" + ], "properties": { "mask": { "$ref": "#/components/schemas/MaskData" @@ -2278,7 +2615,10 @@ ] }, "visible": { - "type": ["boolean", "null"] + "type": [ + "boolean", + "null" + ] } } }, @@ -2286,11 +2626,15 @@ "oneOf": [ { "type": "object", - "required": ["updateProjectMeta"], + "required": [ + "updateProjectMeta" + ], "properties": { "updateProjectMeta": { "type": "object", - "required": ["patch"], + "required": [ + "patch" + ], "properties": { "patch": { "$ref": "#/components/schemas/ProjectMetaPatch" @@ -2304,11 +2648,16 @@ }, { "type": "object", - "required": ["addPage"], + "required": [ + "addPage" + ], "properties": { "addPage": { "type": "object", - "required": ["page", "at"], + "required": [ + "page", + "at" + ], "properties": { "at": { "type": "integer", @@ -2323,11 +2672,17 @@ }, { "type": "object", - "required": ["removePage"], + "required": [ + "removePage" + ], "properties": { "removePage": { "type": "object", - "required": ["id", "prev_page", "prev_index"], + "required": [ + "id", + "prev_page", + "prev_index" + ], "properties": { "id": { "$ref": "#/components/schemas/PageId" @@ -2345,11 +2700,16 @@ }, { "type": "object", - "required": ["updatePage"], + "required": [ + "updatePage" + ], "properties": { "updatePage": { "type": "object", - "required": ["id", "patch"], + "required": [ + "id", + "patch" + ], "properties": { "id": { "$ref": "#/components/schemas/PageId" @@ -2366,11 +2726,16 @@ }, { "type": "object", - "required": ["reorderPages"], + "required": [ + "reorderPages" + ], "properties": { "reorderPages": { "type": "object", - "required": ["order", "prev_order"], + "required": [ + "order", + "prev_order" + ], "properties": { "order": { "type": "array", @@ -2390,11 +2755,17 @@ }, { "type": "object", - "required": ["addNode"], + "required": [ + "addNode" + ], "properties": { "addNode": { "type": "object", - "required": ["page", "node", "at"], + "required": [ + "page", + "node", + "at" + ], "properties": { "at": { "type": "integer", @@ -2412,11 +2783,18 @@ }, { "type": "object", - "required": ["removeNode"], + "required": [ + "removeNode" + ], "properties": { "removeNode": { "type": "object", - "required": ["page", "id", "prev_node", "prev_index"], + "required": [ + "page", + "id", + "prev_node", + "prev_index" + ], "properties": { "id": { "$ref": "#/components/schemas/NodeId" @@ -2437,11 +2815,17 @@ }, { "type": "object", - "required": ["updateNode"], + "required": [ + "updateNode" + ], "properties": { "updateNode": { "type": "object", - "required": ["page", "id", "patch"], + "required": [ + "page", + "id", + "patch" + ], "properties": { "id": { "$ref": "#/components/schemas/NodeId" @@ -2461,11 +2845,17 @@ }, { "type": "object", - "required": ["reorderNodes"], + "required": [ + "reorderNodes" + ], "properties": { "reorderNodes": { "type": "object", - "required": ["page", "order", "prev_order"], + "required": [ + "page", + "order", + "prev_order" + ], "properties": { "order": { "type": "array", @@ -2488,11 +2878,16 @@ }, { "type": "object", - "required": ["batch"], + "required": [ + "batch" + ], "properties": { "batch": { "type": "object", - "required": ["ops", "label"], + "required": [ + "ops", + "label" + ], "properties": { "label": { "type": "string" @@ -2511,7 +2906,9 @@ }, "OpenProjectRequest": { "type": "object", - "required": ["id"], + "required": [ + "id" + ], "properties": { "id": { "type": "string", @@ -2521,7 +2918,13 @@ }, "Page": { "type": "object", - "required": ["id", "name", "width", "height", "nodes"], + "required": [ + "id", + "name", + "width", + "height", + "nodes" + ], "properties": { "height": { "type": "integer", @@ -2560,15 +2963,24 @@ "type": "object", "properties": { "height": { - "type": ["integer", "null"], + "type": [ + "integer", + "null" + ], "format": "int32", "minimum": 0 }, "name": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] }, "width": { - "type": ["integer", "null"], + "type": [ + "integer", + "null" + ], "format": "int32", "minimum": 0 } @@ -2594,9 +3006,21 @@ "type": "string", "default": "lama-manga" }, + "max_batch_pages": { + "type": "integer", + "description": "Upper bound on pages any single stage folds into one model call.\n`0` = auto, `1` = disable batching but keep stage overlap.\n\nBatch size is the larger VRAM lever of the two, so try lowering this\nbefore `max_inflight_pages`.", + "default": 0, + "minimum": 0 + }, + "max_inflight_pages": { + "type": "integer", + "description": "Maximum pages moving through the pipeline at once. `0` = auto (one per\nstage, so every stage can stay busy).\n\nSet to `1` to restore fully sequential processing: one page finishes\nevery step before the next starts, and no stage ever batches. That is\nthe escape hatch if parallelism causes trouble.", + "default": 0, + "minimum": 0 + }, "ocr": { "type": "string", - "default": "paddle-ocr-vl-1.5" + "default": "paddle-ocr-vl-1.6" }, "renderer": { "type": "string", @@ -2616,28 +3040,68 @@ "type": "object", "properties": { "bubbleSegmenter": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] }, "detector": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] }, "fontDetector": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] }, "inpainter": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] + }, + "maxBatchPages": { + "type": [ + "integer", + "null" + ], + "description": "Cap on pages a stage folds into one model call. `0` = auto, `1` =\nno batching.", + "minimum": 0 + }, + "maxInflightPages": { + "type": [ + "integer", + "null" + ], + "description": "Pages allowed in the pipeline at once. `0` = auto (one per step), `1` =\nfully sequential.", + "minimum": 0 }, "ocr": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] }, "renderer": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] }, "segmenter": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] }, "translator": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] } } }, @@ -2696,44 +3160,61 @@ "oneOf": [ { "type": "object", - "required": ["status"], + "required": [ + "status" + ], "properties": { "status": { "type": "string", - "enum": ["running"] + "enum": [ + "running" + ] } } }, { "type": "object", - "required": ["status"], + "required": [ + "status" + ], "properties": { "status": { "type": "string", - "enum": ["completed"] + "enum": [ + "completed" + ] } } }, { "type": "object", - "required": ["status"], + "required": [ + "status" + ], "properties": { "status": { "type": "string", - "enum": ["cancelled"] + "enum": [ + "cancelled" + ] } } }, { "type": "object", - "required": ["reason", "status"], + "required": [ + "reason", + "status" + ], "properties": { "reason": { "type": "string" }, "status": { "type": "string", - "enum": ["failed"] + "enum": [ + "failed" + ] } } } @@ -2741,11 +3222,21 @@ }, "PipelineStep": { "type": "string", - "enum": ["detect", "ocr", "inpaint", "llmGenerate", "render"] + "enum": [ + "detect", + "ocr", + "inpaint", + "llmGenerate", + "render" + ] }, "ProjectMeta": { "type": "object", - "required": ["name", "createdAt", "updatedAt"], + "required": [ + "name", + "createdAt", + "updatedAt" + ], "properties": { "createdAt": { "type": "string", @@ -2767,7 +3258,10 @@ "type": "object", "properties": { "name": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] }, "style": { "oneOf": [ @@ -2780,7 +3274,10 @@ ] }, "updatedAt": { - "type": ["string", "null"], + "type": [ + "string", + "null" + ], "format": "date-time" } } @@ -2789,13 +3286,20 @@ "type": "object", "properties": { "defaultFont": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] } } }, "ProjectSummary": { "type": "object", - "required": ["id", "name", "path"], + "required": [ + "id", + "name", + "path" + ], "properties": { "id": { "type": "string", @@ -2818,14 +3322,22 @@ }, "ProviderConfig": { "type": "object", - "required": ["id"], + "required": [ + "id" + ], "properties": { "api_key": { - "type": ["string", "null"], + "type": [ + "string", + "null" + ], "description": "Populated from credential storage on `load()`, never written to config.toml.\nSerializes as `\"[REDACTED]\"` in API responses." }, "base_url": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] }, "id": { "type": "string" @@ -2834,14 +3346,22 @@ }, "ProviderPatch": { "type": "object", - "required": ["id"], + "required": [ + "id" + ], "properties": { "apiKey": { - "type": ["string", "null"], + "type": [ + "string", + "null" + ], "description": "`\"[REDACTED]\"` → keep existing keyring secret; empty → clear; otherwise save." }, "baseUrl": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] }, "id": { "type": "string" @@ -2850,7 +3370,9 @@ }, "ProviderSecretRequest": { "type": "object", - "required": ["secret"], + "required": [ + "secret" + ], "properties": { "secret": { "type": "string" @@ -2859,7 +3381,10 @@ }, "PutMaskResponse": { "type": "object", - "required": ["node", "blob"], + "required": [ + "node", + "blob" + ], "properties": { "blob": { "$ref": "#/components/schemas/BlobRef" @@ -2871,11 +3396,20 @@ }, "ReadingOrder": { "type": "string", - "enum": ["rtl", "ltr", "custom"] + "enum": [ + "rtl", + "ltr", + "custom" + ] }, "Region": { "type": "object", - "required": ["x", "y", "width", "height"], + "required": [ + "x", + "y", + "width", + "height" + ], "properties": { "height": { "type": "integer", @@ -2901,7 +3435,10 @@ }, "Scene": { "type": "object", - "required": ["project", "pages"], + "required": [ + "project", + "pages" + ], "properties": { "pages": { "type": "object", @@ -2922,7 +3459,10 @@ "SceneSnapshot": { "type": "object", "description": "JSON-shaped scene snapshot for the UI (no postcard decoder in JS).", - "required": ["epoch", "scene"], + "required": [ + "epoch", + "scene" + ], "properties": { "epoch": { "type": "integer", @@ -2936,7 +3476,10 @@ }, "SnapshotEvent": { "type": "object", - "required": ["jobs", "downloads"], + "required": [ + "jobs", + "downloads" + ], "properties": { "downloads": { "type": "array", @@ -2954,7 +3497,9 @@ }, "StartDownloadRequest": { "type": "object", - "required": ["modelId"], + "required": [ + "modelId" + ], "properties": { "modelId": { "type": "string", @@ -2964,7 +3509,9 @@ }, "StartDownloadResponse": { "type": "object", - "required": ["operationId"], + "required": [ + "operationId" + ], "properties": { "operationId": { "type": "string", @@ -2974,13 +3521,21 @@ }, "StartPipelineRequest": { "type": "object", - "required": ["steps"], + "required": [ + "steps" + ], "properties": { "defaultFont": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] }, "pages": { - "type": ["array", "null"], + "type": [ + "array", + "null" + ], "items": { "$ref": "#/components/schemas/PageId" }, @@ -3015,13 +3570,22 @@ "description": "Engine ids (`inventory::submit!` ids) to run in order." }, "systemPrompt": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] }, "targetLanguage": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] }, "textNodeIds": { - "type": ["array", "null"], + "type": [ + "array", + "null" + ], "items": { "$ref": "#/components/schemas/NodeId" }, @@ -3031,7 +3595,9 @@ }, "StartPipelineResponse": { "type": "object", - "required": ["operationId"], + "required": [ + "operationId" + ], "properties": { "operationId": { "type": "string" @@ -3040,7 +3606,11 @@ }, "TextAlign": { "type": "string", - "enum": ["left", "center", "right"] + "enum": [ + "left", + "center", + "right" + ] }, "TextData": { "type": "object", @@ -3050,11 +3620,17 @@ "format": "float" }, "detectedFontSizePx": { - "type": ["number", "null"], + "type": [ + "number", + "null" + ], "format": "float" }, "detector": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] }, "fontPrediction": { "oneOf": [ @@ -3067,7 +3643,10 @@ ] }, "linePolygons": { - "type": ["array", "null"], + "type": [ + "array", + "null" + ], "items": { "type": "array", "items": { @@ -3093,7 +3672,10 @@ ] }, "rotationDeg": { - "type": ["number", "null"], + "type": [ + "number", + "null" + ], "format": "float" }, "sourceDirection": { @@ -3107,7 +3689,10 @@ ] }, "sourceLang": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] }, "sprite": { "oneOf": [ @@ -3142,10 +3727,16 @@ ] }, "text": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] }, "translation": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] } } }, @@ -3154,15 +3745,24 @@ "description": "For fields where \"set to None\" is meaningful (e.g. clearing a translation),\nthe outer `Option` is \"patch present\", the inner is \"value present\".", "properties": { "confidence": { - "type": ["number", "null"], + "type": [ + "number", + "null" + ], "format": "float" }, "detectedFontSizePx": { - "type": ["number", "null"], + "type": [ + "number", + "null" + ], "format": "float" }, "detector": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] }, "fontPrediction": { "oneOf": [ @@ -3175,7 +3775,10 @@ ] }, "linePolygons": { - "type": ["array", "null"], + "type": [ + "array", + "null" + ], "items": { "type": "array", "items": { @@ -3188,7 +3791,10 @@ } }, "lockLayoutBox": { - "type": ["boolean", "null"] + "type": [ + "boolean", + "null" + ] }, "renderedDirection": { "oneOf": [ @@ -3201,7 +3807,10 @@ ] }, "rotationDeg": { - "type": ["number", "null"], + "type": [ + "number", + "null" + ], "format": "float" }, "sourceDirection": { @@ -3215,7 +3824,10 @@ ] }, "sourceLang": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] }, "sprite": { "oneOf": [ @@ -3248,17 +3860,26 @@ ] }, "text": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] }, "translation": { - "type": ["string", "null"] + "type": [ + "string", + "null" + ] } } }, "TextDirection": { "type": "string", "description": "Reading axis of a text block.", - "enum": ["horizontal", "vertical"] + "enum": [ + "horizontal", + "vertical" + ] }, "TextShaderEffect": { "type": "object", @@ -3286,14 +3907,20 @@ "type": "boolean" }, "widthPx": { - "type": ["number", "null"], + "type": [ + "number", + "null" + ], "format": "float" } } }, "TextStyle": { "type": "object", - "required": ["fontFamilies", "color"], + "required": [ + "fontFamilies", + "color" + ], "properties": { "color": { "type": "array", @@ -3320,7 +3947,10 @@ } }, "fontSize": { - "type": ["number", "null"], + "type": [ + "number", + "null" + ], "format": "float" }, "stroke": { @@ -3347,7 +3977,10 @@ }, "TopFont": { "type": "object", - "required": ["index", "score"], + "required": [ + "index", + "score" + ], "properties": { "index": { "type": "integer", @@ -3361,7 +3994,12 @@ }, "Transform": { "type": "object", - "required": ["x", "y", "width", "height"], + "required": [ + "x", + "y", + "width", + "height" + ], "properties": { "height": { "type": "number", @@ -3387,4 +4025,4 @@ } } } -} +} \ No newline at end of file diff --git a/ui/public/locales/en-US/translation.json b/ui/public/locales/en-US/translation.json index 5cdb145ac..14ca9b64b 100644 --- a/ui/public/locales/en-US/translation.json +++ b/ui/public/locales/en-US/translation.json @@ -424,6 +424,15 @@ "restartApply": "Apply & Restart", "restartApplying": "Applying...", "restartRequiredDescription": "These settings are loaded at startup. Applying them will save the config and restart the app.", + "parallel": "Parallel Processing", + "parallelDescription": "How much of the pipeline runs at once. Saved immediately and applied to the next run — no restart needed.", + "maxInflightPages": "Pages In Flight", + "maxInflightPagesDescription": "Pages moving through the pipeline simultaneously. Set 0 for auto, or 1 to process strictly one page at a time — the escape hatch if parallel processing causes trouble.", + "maxInflightPagesAuto": "Auto: {{pages}} — one page per pipeline step.", + "maxBatchPages": "Batch Size", + "maxBatchPagesDescription": "Pages a single stage folds into one model call. Set 0 for auto, or 1 to keep stages overlapping without batching. This is the larger VRAM lever, so lower it before Pages In Flight.", + "maxBatchPagesAuto": "Auto: each engine decides its own batch.", + "cpuWorkersInfo": "CPU-bound stages use {{workers}} workers on this machine.", "engines": "Engines", "enginesDescription": "Select which ML engine to use for each pipeline stage.", "detector": "Detector", diff --git a/ui/public/locales/es-ES/translation.json b/ui/public/locales/es-ES/translation.json index 17084fbac..1daa83463 100644 --- a/ui/public/locales/es-ES/translation.json +++ b/ui/public/locales/es-ES/translation.json @@ -385,6 +385,15 @@ "restartApply": "Aplicar y reiniciar", "restartApplying": "Aplicando...", "restartRequiredDescription": "Estos ajustes se cargan al iniciar. Aplicarlos guardará la configuración y reiniciará la aplicación.", + "parallel": "Procesamiento Paralelo", + "parallelDescription": "Cuánta parte de la canalización se ejecuta a la vez. Se guarda al instante y se aplica a la siguiente ejecución: no requiere reiniciar.", + "maxInflightPages": "Páginas Simultáneas", + "maxInflightPagesDescription": "Páginas que avanzan por la canalización a la vez. Usa 0 para automático o 1 para procesar estrictamente una página cada vez: la salida de emergencia si el procesamiento paralelo causa problemas.", + "maxInflightPagesAuto": "Automático: {{pages}}, una página por cada paso de la canalización.", + "maxBatchPages": "Tamaño de Lote", + "maxBatchPagesDescription": "Páginas que una etapa agrupa en una sola llamada al modelo. Usa 0 para automático o 1 para mantener las etapas solapadas sin agrupar. Es la palanca de VRAM más grande, así que redúcelo antes que Páginas Simultáneas.", + "maxBatchPagesAuto": "Automático: cada motor decide su propio lote.", + "cpuWorkersInfo": "Las etapas limitadas por CPU usan {{workers}} procesos en esta máquina.", "engines": "Motores", "enginesDescription": "Selecciona qué motor ML usar para cada etapa del pipeline.", "detector": "Detector", diff --git a/ui/public/locales/ja-JP/translation.json b/ui/public/locales/ja-JP/translation.json index 2fe195ab4..13ca3312a 100644 --- a/ui/public/locales/ja-JP/translation.json +++ b/ui/public/locales/ja-JP/translation.json @@ -385,6 +385,15 @@ "restartApply": "適用して再起動", "restartApplying": "適用中...", "restartRequiredDescription": "これらの設定は起動時に読み込まれます。適用すると設定を保存してアプリを再起動します。", + "parallel": "並列処理", + "parallelDescription": "パイプラインを同時にどれだけ実行するか。すぐに保存され、次回の実行から適用されます（再起動は不要）。", + "maxInflightPages": "同時処理ページ数", + "maxInflightPagesDescription": "パイプラインを同時に流れるページ数。0 で自動、1 で厳密に 1 ページずつ処理します。並列処理で問題が起きた場合の回避策です。", + "maxInflightPagesAuto": "自動: {{pages}} — パイプラインの各ステップにつき 1 ページ。", + "maxBatchPages": "バッチサイズ", + "maxBatchPagesDescription": "1 回のモデル呼び出しにまとめるページ数。0 で自動、1 でバッチ処理せずステージの並行のみ維持します。VRAM への影響が大きいため、同時処理ページ数より先にこちらを下げてください。", + "maxBatchPagesAuto": "自動: 各エンジンがバッチサイズを決定します。", + "cpuWorkersInfo": "CPU 依存のステージはこのマシンで {{workers}} ワーカーを使用します。", "engines": "エンジン", "enginesDescription": "各パイプライン段階で使用するMLエンジンを選択します。", "detector": "検出器", diff --git a/ui/public/locales/ko-KR/translation.json b/ui/public/locales/ko-KR/translation.json index f9ec55422..ca7145244 100644 --- a/ui/public/locales/ko-KR/translation.json +++ b/ui/public/locales/ko-KR/translation.json @@ -411,6 +411,15 @@ "restartApply": "적용 후 재시작", "restartApplying": "적용 중...", "restartRequiredDescription": "이 설정은 시작 시 로드됩니다. 적용하면 구성을 저장하고 앱을 다시 시작합니다.", + "parallel": "병렬 처리", + "parallelDescription": "파이프라인을 한 번에 얼마나 실행할지 설정합니다. 즉시 저장되며 다음 실행부터 적용됩니다(재시작 불필요).", + "maxInflightPages": "동시 처리 페이지", + "maxInflightPagesDescription": "파이프라인을 동시에 통과하는 페이지 수입니다. 0은 자동, 1은 한 번에 한 페이지씩 엄격하게 처리합니다. 병렬 처리에 문제가 있을 때 사용하세요.", + "maxInflightPagesAuto": "자동: {{pages}} — 파이프라인 단계당 한 페이지.", + "maxBatchPages": "배치 크기", + "maxBatchPagesDescription": "한 단계가 하나의 모델 호출로 묶는 페이지 수입니다. 0은 자동, 1은 배치 없이 단계 중첩만 유지합니다. VRAM에 더 큰 영향을 주므로 동시 처리 페이지보다 먼저 낮추세요.", + "maxBatchPagesAuto": "자동: 각 엔진이 배치 크기를 결정합니다.", + "cpuWorkersInfo": "CPU 위주 단계는 이 컴퓨터에서 {{workers}}개의 워커를 사용합니다.", "engines": "엔진", "enginesDescription": "각 파이프라인 단계에서 사용할 ML 엔진을 선택합니다.", "detector": "검출기", diff --git a/ui/public/locales/pt-BR/translation.json b/ui/public/locales/pt-BR/translation.json index ad891bfc6..d51a82a4a 100644 --- a/ui/public/locales/pt-BR/translation.json +++ b/ui/public/locales/pt-BR/translation.json @@ -386,6 +386,15 @@ "restartApply": "Aplicar e reiniciar", "restartApplying": "Aplicando...", "restartRequiredDescription": "Essas configurações são carregadas na inicialização. Aplicá-las salvará a configuração e reiniciará o aplicativo.", + "parallel": "Processamento Paralelo", + "parallelDescription": "Quanto do pipeline é executado de uma vez. Salvo imediatamente e aplicado na próxima execução, sem reiniciar.", + "maxInflightPages": "Páginas Simultâneas", + "maxInflightPagesDescription": "Páginas percorrendo o pipeline ao mesmo tempo. Use 0 para automático ou 1 para processar estritamente uma página por vez: a saída de emergência se o processamento paralelo causar problemas.", + "maxInflightPagesAuto": "Automático: {{pages}}, uma página por etapa do pipeline.", + "maxBatchPages": "Tamanho do Lote", + "maxBatchPagesDescription": "Páginas que uma etapa agrupa em uma única chamada ao modelo. Use 0 para automático ou 1 para manter as etapas sobrepostas sem agrupar. Esta é a maior alavanca de VRAM, então reduza-a antes de Páginas Simultâneas.", + "maxBatchPagesAuto": "Automático: cada motor decide seu próprio lote.", + "cpuWorkersInfo": "Etapas limitadas por CPU usam {{workers}} workers nesta máquina.", "engines": "Motores", "enginesDescription": "Selecione qual motor ML usar para cada etapa do pipeline.", "detector": "Detector", diff --git a/ui/public/locales/ru-RU/translation.json b/ui/public/locales/ru-RU/translation.json index 62b06211f..f51a2bbd3 100644 --- a/ui/public/locales/ru-RU/translation.json +++ b/ui/public/locales/ru-RU/translation.json @@ -385,6 +385,15 @@ "restartApply": "Применить и перезапустить", "restartApplying": "Применение...", "restartRequiredDescription": "Эти настройки загружаются при запуске. Применение сохранит конфигурацию и перезапустит приложение.", + "parallel": "Параллельная обработка", + "parallelDescription": "Сколько конвейера выполняется одновременно. Сохраняется сразу и применяется к следующему запуску, перезапуск не нужен.", + "maxInflightPages": "Страниц одновременно", + "maxInflightPagesDescription": "Страницы, одновременно проходящие по конвейеру. 0 — автоматически, 1 — строго по одной странице за раз: аварийный вариант, если параллельная обработка вызывает проблемы.", + "maxInflightPagesAuto": "Автоматически: {{pages}} — по одной странице на шаг конвейера.", + "maxBatchPages": "Размер пакета", + "maxBatchPagesDescription": "Страницы, которые этап объединяет в один вызов модели. 0 — автоматически, 1 — без пакетов, но с наложением этапов. Это сильнее влияет на VRAM, поэтому уменьшайте его раньше, чем «Страниц одновременно».", + "maxBatchPagesAuto": "Автоматически: каждый движок сам определяет размер пакета.", + "cpuWorkersInfo": "Этапы, нагружающие процессор, используют {{workers}} рабочих потоков на этой машине.", "engines": "Движки", "enginesDescription": "Выберите ML-движок для каждого этапа конвейера.", "detector": "Детектор", diff --git a/ui/public/locales/tr-TR/translation.json b/ui/public/locales/tr-TR/translation.json index b062d6fe9..5e10bf93c 100644 --- a/ui/public/locales/tr-TR/translation.json +++ b/ui/public/locales/tr-TR/translation.json @@ -385,6 +385,15 @@ "restartApply": "Uygula ve Yeniden Başlat", "restartApplying": "Uygulanıyor...", "restartRequiredDescription": "Bu ayarlar başlangıçta yüklenir. Uygulamak yapılandırmayı kaydedip uygulamayı yeniden başlatır.", + "parallel": "Paralel İşleme", + "parallelDescription": "İşlem hattının ne kadarının aynı anda çalışacağı. Hemen kaydedilir ve bir sonraki çalıştırmada uygulanır, yeniden başlatma gerekmez.", + "maxInflightPages": "Eşzamanlı Sayfa", + "maxInflightPagesDescription": "İşlem hattında aynı anda ilerleyen sayfa sayısı. Otomatik için 0, kesin olarak tek seferde bir sayfa işlemek için 1 girin: paralel işleme sorun çıkarırsa acil çıkış.", + "maxInflightPagesAuto": "Otomatik: {{pages}}, işlem hattı adımı başına bir sayfa.", + "maxBatchPages": "Yığın Boyutu", + "maxBatchPagesDescription": "Bir aşamanın tek bir model çağrısında birleştirdiği sayfa sayısı. Otomatik için 0, yığınlama olmadan aşamaların örtüşmesini sürdürmek için 1 girin. VRAM üzerinde daha büyük etkisi vardır, bu yüzden Eşzamanlı Sayfa ayarından önce bunu düşürün.", + "maxBatchPagesAuto": "Otomatik: her motor kendi yığınına karar verir.", + "cpuWorkersInfo": "CPU ağırlıklı aşamalar bu makinede {{workers}} işçi kullanır.", "engines": "Motorlar", "enginesDescription": "Her işlem hattı aşaması için kullanılacak ML motorunu seçin.", "detector": "Algılayıcı", diff --git a/ui/public/locales/zh-CN/translation.json b/ui/public/locales/zh-CN/translation.json index 60bfeb1af..47fd63480 100644 --- a/ui/public/locales/zh-CN/translation.json +++ b/ui/public/locales/zh-CN/translation.json @@ -385,6 +385,15 @@ "restartApply": "应用并重启", "restartApplying": "正在应用...", "restartRequiredDescription": "这些设置会在启动时加载。应用后会保存配置并重启应用。", + "parallel": "并行处理", + "parallelDescription": "流水线同时运行的规模。立即保存并在下次运行时生效，无需重启。", + "maxInflightPages": "并行页数", + "maxInflightPagesDescription": "同时在流水线中处理的页数。0 表示自动，1 表示严格逐页处理，是并行处理出问题时的应急选项。", + "maxInflightPagesAuto": "自动：{{pages}}，每个流水线步骤一页。", + "maxBatchPages": "批处理大小", + "maxBatchPagesDescription": "单个阶段合并到一次模型调用的页数。0 表示自动，1 表示不批处理但保持阶段重叠。它对显存的影响更大，请先降低此项再调整并行页数。", + "maxBatchPagesAuto": "自动：由各引擎自行决定批大小。", + "cpuWorkersInfo": "受 CPU 限制的阶段在本机使用 {{workers}} 个工作线程。", "engines": "引擎", "enginesDescription": "为每个流水线阶段选择要使用的ML引擎。", "detector": "检测器", diff --git a/ui/public/locales/zh-TW/translation.json b/ui/public/locales/zh-TW/translation.json index 973fac145..bffc6f3eb 100644 --- a/ui/public/locales/zh-TW/translation.json +++ b/ui/public/locales/zh-TW/translation.json @@ -385,6 +385,15 @@ "restartApply": "套用並重新啟動", "restartApplying": "套用中...", "restartRequiredDescription": "這些設定會在啟動時載入。套用後會儲存設定並重新啟動應用程式。", + "parallel": "平行處理", + "parallelDescription": "流水線同時執行的規模。立即儲存並於下次執行時生效，無需重新啟動。", + "maxInflightPages": "平行頁數", + "maxInflightPagesDescription": "同時在流水線中處理的頁數。0 表示自動，1 表示嚴格逐頁處理，是平行處理出問題時的應急選項。", + "maxInflightPagesAuto": "自動：{{pages}}，每個流水線步驟一頁。", + "maxBatchPages": "批次大小", + "maxBatchPagesDescription": "單一階段合併為一次模型呼叫的頁數。0 表示自動，1 表示不批次處理但保持階段重疊。它對顯示記憶體的影響更大，請先調降此項再調整平行頁數。", + "maxBatchPagesAuto": "自動：由各引擎自行決定批次大小。", + "cpuWorkersInfo": "受 CPU 限制的階段在本機使用 {{workers}} 個工作執行緒。", "engines": "引擎", "enginesDescription": "為每個流水線階段選擇要使用的ML引擎。", "detector": "偵測器",