fix(analysis): hold the OCR permit for the full blocking inference
The OCR dispatcher acquired a borrowed semaphore permit, then ran the CPU-bound inference in `spawn_blocking`. On cancellation (graceful shutdown) the future dropped the permit while the detached blocking task kept running, briefly oversubscribing the blocking pool past ANALYSIS_WORKERS. Extract `run_ocr_blocking`, which acquires an *owned* permit and moves it into the blocking task so the concurrency slot's lifetime matches the actual work. Covered by a cancellation-invariant test asserting the permit stays held after the caller future is aborted. Co-Authored-By: Claude Opus 4.8 (1M context) <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.93.5"
|
version = "0.93.6"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"anyhow",
|
"anyhow",
|
||||||
"argon2",
|
"argon2",
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "mangalord"
|
name = "mangalord"
|
||||||
version = "0.93.5"
|
version = "0.93.6"
|
||||||
edition = "2021"
|
edition = "2021"
|
||||||
default-run = "mangalord"
|
default-run = "mangalord"
|
||||||
|
|
||||||
|
|||||||
@@ -155,6 +155,37 @@ pub fn ocr_concurrency_limit(workers: usize, cores: usize) -> usize {
|
|||||||
workers.min(cores.max(1)).max(1)
|
workers.min(cores.max(1)).max(1)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Run a CPU-bound OCR closure on the blocking pool while holding an
|
||||||
|
/// **owned** permit for the entire duration of the work.
|
||||||
|
///
|
||||||
|
/// The permit is acquired with `acquire_owned` and moved *into* the
|
||||||
|
/// blocking task rather than being held by the caller's future. This
|
||||||
|
/// matters on cancellation: if the dispatcher future is dropped (graceful
|
||||||
|
/// shutdown), a borrowed permit would be released the instant the future
|
||||||
|
/// unwinds — but `spawn_blocking` work is not cancellable and keeps
|
||||||
|
/// running detached, so the concurrency bound (ANALYSIS_WORKERS) would be
|
||||||
|
/// briefly exceeded. Moving the permit into the task ties the slot's
|
||||||
|
/// lifetime to the actual CPU work.
|
||||||
|
async fn run_ocr_blocking<T, F>(
|
||||||
|
permits: Arc<tokio::sync::Semaphore>,
|
||||||
|
f: F,
|
||||||
|
) -> anyhow::Result<T>
|
||||||
|
where
|
||||||
|
F: FnOnce() -> T + Send + 'static,
|
||||||
|
T: Send + 'static,
|
||||||
|
{
|
||||||
|
let permit = permits
|
||||||
|
.acquire_owned()
|
||||||
|
.await
|
||||||
|
.map_err(|e| anyhow::anyhow!("OCR semaphore closed: {e}"))?;
|
||||||
|
tokio::task::spawn_blocking(move || {
|
||||||
|
let _permit = permit; // held until f() returns, even if the caller is cancelled
|
||||||
|
f()
|
||||||
|
})
|
||||||
|
.await
|
||||||
|
.map_err(|e| anyhow::anyhow!("OCR task join error: {e}"))
|
||||||
|
}
|
||||||
|
|
||||||
/// 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.
|
||||||
@@ -192,15 +223,10 @@ 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, and gate it behind the shared permit pool so
|
// worker's runtime thread, and gate it behind the shared permit pool so
|
||||||
// ANALYSIS_WORKERS > cores can't oversubscribe the blocking pool.
|
// 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 =
|
||||||
.await
|
run_ocr_blocking(Arc::clone(&self.ocr_permits), move || engine.recognize(&bytes))
|
||||||
.map_err(|e| anyhow::anyhow!("OCR task join error: {e}"))??;
|
.await??;
|
||||||
let analysis = lines_to_analysis(lines);
|
let analysis = lines_to_analysis(lines);
|
||||||
repo::page_analysis::persist_analysis(&self.db, page_id, &analysis, OCR_MODEL_LABEL).await?;
|
repo::page_analysis::persist_analysis(&self.db, page_id, &analysis, OCR_MODEL_LABEL).await?;
|
||||||
Ok(())
|
Ok(())
|
||||||
@@ -235,6 +261,55 @@ pub mod test_support {
|
|||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn run_ocr_blocking_holds_permit_until_work_completes_despite_cancellation() {
|
||||||
|
use std::sync::mpsc;
|
||||||
|
use tokio::sync::{Notify, Semaphore};
|
||||||
|
|
||||||
|
let permits = Arc::new(Semaphore::new(1));
|
||||||
|
let (release_tx, release_rx) = mpsc::channel::<()>();
|
||||||
|
let started = Arc::new(Notify::new());
|
||||||
|
|
||||||
|
// Model the dispatch path: acquire an owned permit and run blocking
|
||||||
|
// work that we hold open via a channel.
|
||||||
|
let sem = Arc::clone(&permits);
|
||||||
|
let started2 = Arc::clone(&started);
|
||||||
|
let caller = tokio::spawn(async move {
|
||||||
|
run_ocr_blocking(sem, move || {
|
||||||
|
// Signal that the blocking task now holds the permit, then
|
||||||
|
// block until the test releases us.
|
||||||
|
started2.notify_one();
|
||||||
|
release_rx.recv().ok();
|
||||||
|
})
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
});
|
||||||
|
|
||||||
|
// Wait until the blocking task is running and owns the permit.
|
||||||
|
started.notified().await;
|
||||||
|
assert_eq!(permits.available_permits(), 0, "permit taken by blocking work");
|
||||||
|
|
||||||
|
// Cancel the caller future (simulates graceful shutdown). The
|
||||||
|
// blocking task is not cancellable and keeps running detached; with
|
||||||
|
// an *owned* permit the slot must stay occupied. A borrowed permit
|
||||||
|
// would have been released here, regressing the concurrency bound.
|
||||||
|
caller.abort();
|
||||||
|
let _ = caller.await;
|
||||||
|
assert_eq!(
|
||||||
|
permits.available_permits(),
|
||||||
|
0,
|
||||||
|
"owned permit must remain held by the still-running blocking task"
|
||||||
|
);
|
||||||
|
|
||||||
|
// Let the blocking work finish; the permit is then returned.
|
||||||
|
release_tx.send(()).unwrap();
|
||||||
|
let permit = tokio::time::timeout(std::time::Duration::from_secs(5), permits.acquire())
|
||||||
|
.await
|
||||||
|
.expect("permit should be released once blocking work completes")
|
||||||
|
.unwrap();
|
||||||
|
drop(permit);
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn lines_to_analysis_maps_lines_in_order_and_leaves_rest_empty() {
|
fn lines_to_analysis_maps_lines_in_order_and_leaves_rest_empty() {
|
||||||
let v = lines_to_analysis(vec!["Hello".to_string(), "world!".to_string()]);
|
let v = lines_to_analysis(vec!["Hello".to_string(), "world!".to_string()]);
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
{
|
{
|
||||||
"name": "mangalord-frontend",
|
"name": "mangalord-frontend",
|
||||||
"version": "0.93.5",
|
"version": "0.93.6",
|
||||||
"private": true,
|
"private": true,
|
||||||
"type": "module",
|
"type": "module",
|
||||||
"scripts": {
|
"scripts": {
|
||||||
|
|||||||
Reference in New Issue
Block a user