From 5b76e0cc3772752d90e4bfa614be259981c833b7 Mon Sep 17 00:00:00 2001 From: MechaCat02 Date: Wed, 1 Jul 2026 07:14:19 +0200 Subject: [PATCH] fix(analysis): bound concurrent OCR inferences with a shared semaphore Co-Authored-By: Claude Opus 4.8 --- backend/Cargo.lock | 2 +- backend/Cargo.toml | 2 +- backend/src/analysis/ocr.rs | 43 ++++++++++++++++++++++++++++++++++- backend/src/app.rs | 8 +++++++ backend/tests/analysis_ocr.rs | 1 + frontend/package.json | 2 +- 6 files changed, 54 insertions(+), 4 deletions(-) diff --git a/backend/Cargo.lock b/backend/Cargo.lock index c89fd73..fd6e23c 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -1558,7 +1558,7 @@ checksum = "c41e0c4fef86961ac6d6f8a82609f55f31b05e4fce149ac5710e439df7619ba4" [[package]] name = "mangalord" -version = "0.90.1" +version = "0.90.2" dependencies = [ "anyhow", "argon2", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 75148f2..a791cb9 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "mangalord" -version = "0.90.1" +version = "0.90.2" edition = "2021" default-run = "mangalord" diff --git a/backend/src/analysis/ocr.rs b/backend/src/analysis/ocr.rs index fa8cbba..f840fc8 100644 --- a/backend/src/analysis/ocr.rs +++ b/backend/src/analysis/ocr.rs @@ -143,6 +143,18 @@ impl OcrEngine for OcrsEngine { } } +/// Bound on concurrent CPU-bound OCR inferences, given the number of analysis +/// workers and the host's available parallelism. +/// +/// Each worker's `dispatch` fires one `spawn_blocking` OCR run, so without a +/// bound `ANALYSIS_WORKERS` blocking tasks can run at once. OCR is fully +/// CPU-bound, so running more than there are cores just thrashes the scheduler +/// (and balloons the blocking pool). Cap at the core count, but never below 1 +/// and never above the worker count (more permits than workers is pointless). +pub fn ocr_concurrency_limit(workers: usize, cores: usize) -> usize { + workers.min(cores.max(1)).max(1) +} + /// Production dispatcher for the OCR backend: load the page, read its image /// from storage, run OCR on the blocking pool, and persist the lines. Mirrors /// [`crate::analysis::daemon::RealAnalyzeDispatcher`] but with no network I/O. @@ -151,6 +163,10 @@ pub struct OcrAnalyzeDispatcher { pub storage: Arc, pub engine: Arc, pub max_image_bytes: usize, + /// Caps concurrent CPU-bound OCR inferences across all workers (see + /// [`ocr_concurrency_limit`]). Shared via the `Arc`, + /// so one permit pool covers every worker. + pub ocr_permits: Arc, } #[async_trait] @@ -174,7 +190,13 @@ impl AnalyzeDispatcher for OcrAnalyzeDispatcher { ); } // OCR inference is CPU-bound and synchronous — keep it off the async - // worker's runtime thread. + // worker's runtime thread, and gate it behind the shared permit pool so + // ANALYSIS_WORKERS > cores can't oversubscribe the blocking pool. + let _permit = self + .ocr_permits + .acquire() + .await + .map_err(|e| anyhow::anyhow!("OCR semaphore closed: {e}"))?; let engine = Arc::clone(&self.engine); let lines = tokio::task::spawn_blocking(move || engine.recognize(&bytes)) .await @@ -267,4 +289,23 @@ mod tests { "expected a decode error, got: {err}" ); } + + #[test] + fn ocr_concurrency_limit_caps_at_cores() { + // More workers than cores → clamp to cores (don't oversubscribe). + assert_eq!(ocr_concurrency_limit(8, 4), 4); + } + + #[test] + fn ocr_concurrency_limit_caps_at_workers() { + // Fewer workers than cores → only `workers` ever run anyway. + assert_eq!(ocr_concurrency_limit(2, 16), 2); + } + + #[test] + fn ocr_concurrency_limit_never_zero() { + // Degenerate inputs still yield at least one permit. + assert_eq!(ocr_concurrency_limit(0, 0), 1); + assert_eq!(ocr_concurrency_limit(1, 0), 1); + } } diff --git a/backend/src/app.rs b/backend/src/app.rs index 868a89c..fd2afa9 100644 --- a/backend/src/app.rs +++ b/backend/src/app.rs @@ -443,11 +443,19 @@ async fn spawn_analysis_daemon( cfg.ocr_max_decode_pixels, ) .context("build ocrs engine")?; + // Cap concurrent CPU-bound OCR runs across all workers so a high + // ANALYSIS_WORKERS can't oversubscribe the blocking pool. + let cores = std::thread::available_parallelism() + .map(|n| n.get()) + .unwrap_or(1); + let permits = + crate::analysis::ocr::ocr_concurrency_limit(cfg.workers, cores); let dispatcher = Arc::new(crate::analysis::ocr::OcrAnalyzeDispatcher { db: db.clone(), storage, engine: Arc::new(engine), max_image_bytes: cfg.max_image_bytes, + ocr_permits: Arc::new(tokio::sync::Semaphore::new(permits)), }); // In-process engine is always ready — no gate. (dispatcher, None) diff --git a/backend/tests/analysis_ocr.rs b/backend/tests/analysis_ocr.rs index e3c40bc..6a11dfb 100644 --- a/backend/tests/analysis_ocr.rs +++ b/backend/tests/analysis_ocr.rs @@ -56,6 +56,7 @@ fn ocr_dispatcher( storage, engine: StubOcrEngine::new(lines), max_image_bytes: 8 * 1024 * 1024, + ocr_permits: Arc::new(tokio::sync::Semaphore::new(1)), } } diff --git a/frontend/package.json b/frontend/package.json index 637de24..ba0db7f 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -1,6 +1,6 @@ { "name": "mangalord-frontend", - "version": "0.90.1", + "version": "0.90.2", "private": true, "type": "module", "scripts": {