fix(analysis): bound concurrent OCR inferences with a shared semaphore
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2
backend/Cargo.lock
generated
2
backend/Cargo.lock
generated
@@ -1558,7 +1558,7 @@ checksum = "c41e0c4fef86961ac6d6f8a82609f55f31b05e4fce149ac5710e439df7619ba4"
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "mangalord"
|
name = "mangalord"
|
||||||
version = "0.90.1"
|
version = "0.90.2"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"anyhow",
|
"anyhow",
|
||||||
"argon2",
|
"argon2",
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "mangalord"
|
name = "mangalord"
|
||||||
version = "0.90.1"
|
version = "0.90.2"
|
||||||
edition = "2021"
|
edition = "2021"
|
||||||
default-run = "mangalord"
|
default-run = "mangalord"
|
||||||
|
|
||||||
|
|||||||
@@ -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
|
/// 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
|
/// from storage, run OCR on the blocking pool, and persist the lines. Mirrors
|
||||||
/// [`crate::analysis::daemon::RealAnalyzeDispatcher`] but with no network I/O.
|
/// [`crate::analysis::daemon::RealAnalyzeDispatcher`] but with no network I/O.
|
||||||
@@ -151,6 +163,10 @@ pub struct OcrAnalyzeDispatcher {
|
|||||||
pub storage: Arc<dyn Storage>,
|
pub storage: Arc<dyn Storage>,
|
||||||
pub engine: Arc<dyn OcrEngine>,
|
pub engine: Arc<dyn OcrEngine>,
|
||||||
pub max_image_bytes: usize,
|
pub max_image_bytes: usize,
|
||||||
|
/// Caps concurrent CPU-bound OCR inferences across all workers (see
|
||||||
|
/// [`ocr_concurrency_limit`]). Shared via the `Arc<dyn AnalyzeDispatcher>`,
|
||||||
|
/// so one permit pool covers every worker.
|
||||||
|
pub ocr_permits: Arc<tokio::sync::Semaphore>,
|
||||||
}
|
}
|
||||||
|
|
||||||
#[async_trait]
|
#[async_trait]
|
||||||
@@ -174,7 +190,13 @@ impl AnalyzeDispatcher for OcrAnalyzeDispatcher {
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
// OCR inference is CPU-bound and synchronous — keep it off the async
|
// 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 engine = Arc::clone(&self.engine);
|
||||||
let lines = tokio::task::spawn_blocking(move || engine.recognize(&bytes))
|
let lines = tokio::task::spawn_blocking(move || engine.recognize(&bytes))
|
||||||
.await
|
.await
|
||||||
@@ -267,4 +289,23 @@ mod tests {
|
|||||||
"expected a decode error, got: {err}"
|
"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);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -443,11 +443,19 @@ async fn spawn_analysis_daemon(
|
|||||||
cfg.ocr_max_decode_pixels,
|
cfg.ocr_max_decode_pixels,
|
||||||
)
|
)
|
||||||
.context("build ocrs engine")?;
|
.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 {
|
let dispatcher = Arc::new(crate::analysis::ocr::OcrAnalyzeDispatcher {
|
||||||
db: db.clone(),
|
db: db.clone(),
|
||||||
storage,
|
storage,
|
||||||
engine: Arc::new(engine),
|
engine: Arc::new(engine),
|
||||||
max_image_bytes: cfg.max_image_bytes,
|
max_image_bytes: cfg.max_image_bytes,
|
||||||
|
ocr_permits: Arc::new(tokio::sync::Semaphore::new(permits)),
|
||||||
});
|
});
|
||||||
// In-process engine is always ready — no gate.
|
// In-process engine is always ready — no gate.
|
||||||
(dispatcher, None)
|
(dispatcher, None)
|
||||||
|
|||||||
@@ -56,6 +56,7 @@ fn ocr_dispatcher(
|
|||||||
storage,
|
storage,
|
||||||
engine: StubOcrEngine::new(lines),
|
engine: StubOcrEngine::new(lines),
|
||||||
max_image_bytes: 8 * 1024 * 1024,
|
max_image_bytes: 8 * 1024 * 1024,
|
||||||
|
ocr_permits: Arc::new(tokio::sync::Semaphore::new(1)),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
{
|
{
|
||||||
"name": "mangalord-frontend",
|
"name": "mangalord-frontend",
|
||||||
"version": "0.90.1",
|
"version": "0.90.2",
|
||||||
"private": true,
|
"private": true,
|
||||||
"type": "module",
|
"type": "module",
|
||||||
"scripts": {
|
"scripts": {
|
||||||
|
|||||||
Reference in New Issue
Block a user