pshock13/pdfcraft
1
0 Fork 0
mirror of https://github.com/storytold/pdfcraft.git synced 2026-10-15 03:37:03 -04:00
Code Issues Packages Projects Releases Wiki Activity

M3: make render watchdog regressions deterministic (#699)

Browse Source 
* M3: gate watchdog tests without timed sleeps

* test(render): replace watchdog timing delays with gates

* Render tests: the panic test doesn't wait on the gate's finished barrier

Only an abandoned render reaches it, so waiting there hung under load.

Co-Authored-By: Claude Opus 5.5 <[email protected]>

---------

Co-authored-by: Brandon Thomas <[email protected]>
Co-authored-by: Claude Opus 5.5 <[email protected]>
This commit is contained in:
cyfung1031Claude Opus 5.5Brandon Thomas
 authored and GitHub committed 2026-10-10 08:33:40 -04:00
1 parent 9cd07e4793
commit 5d59060486
1 file changed
+175 -103
Download Patch File Download Diff File Expand all files Collapse all files
crates/render/src/raster.rs
+175 -103
View File
@@ -467,6 +467,24 @@ struct BusyPublishGate {
release: Arc<Mutex<Receiver<()>>>, release: Arc<Mutex<Receiver<()>>>,
} }
#[cfg(all(test, not(target_arch = "wasm32")))]
#[derive(Clone)]
struct RenderGate {
page: usize,
shared_generation: bool,
claimed: Arc<std::sync::atomic::AtomicBool>,
entered: Arc<std::sync::Barrier>,
release: Arc<std::sync::Barrier>,
finished: Arc<std::sync::Barrier>,
}
#[cfg(all(test, not(target_arch = "wasm32")))]
#[derive(Clone)]
struct ParseGate {
entered: Arc<std::sync::Barrier>,
release: Arc<std::sync::Barrier>,
}
/// State shared by the pool and its workers. /// State shared by the pool and its workers.
#[derive(Default)] #[derive(Default)]
struct Shared { struct Shared {
@@ -483,9 +501,18 @@ struct Shared {
/// can receive its answer (the pool was dropped, or the watchdog gave up on the render). /// can receive its answer (the pool was dropped, or the watchdog gave up on the render).
/// Lock `busy` first when holding both. /// Lock `busy` first when holding both.
stop: Mutex<Vec<Arc<std::sync::atomic::AtomicBool>>>, stop: Mutex<Vec<Arc<std::sync::atomic::AtomicBool>>>,
/// Test hook: make one page slow. /// Test hook: hold one worker at the render boundary until the test releases it.
#[cfg(test)] #[cfg(all(test, not(target_arch = "wasm32")))]
render_gates: Mutex<Vec<RenderGate>>,
/// Test hook: hold parser initialization until the test releases it.
#[cfg(all(test, not(target_arch = "wasm32")))]
parse_gate: Mutex<Option<ParseGate>>,
/// Test hook: make rendering this page take this long (cancellable).
#[cfg(all(test, not(target_arch = "wasm32")))]
slow_page: Mutex<Option<(usize, std::time::Duration)>>, slow_page: Mutex<Option<(usize, std::time::Duration)>>,
/// Test hook: make parser initialization take this long.
#[cfg(all(test, not(target_arch = "wasm32")))]
slow_parse: Mutex<Option<std::time::Duration>>,
/// Test hook: count parser initializations, including failed parses. /// Test hook: count parser initializations, including failed parses.
#[cfg(all(test, not(target_arch = "wasm32")))] #[cfg(all(test, not(target_arch = "wasm32")))]
parses: std::sync::atomic::AtomicUsize, parses: std::sync::atomic::AtomicUsize,
@@ -494,8 +521,6 @@ struct Shared {
#[cfg(all(test, not(target_arch = "wasm32")))] #[cfg(all(test, not(target_arch = "wasm32")))]
panic_page: Mutex<Option<usize>>, panic_page: Mutex<Option<usize>>,
#[cfg(all(test, not(target_arch = "wasm32")))] #[cfg(all(test, not(target_arch = "wasm32")))]
slow_parse: Mutex<Option<std::time::Duration>>,
#[cfg(all(test, not(target_arch = "wasm32")))]
render_started: std::sync::atomic::AtomicUsize, render_started: std::sync::atomic::AtomicUsize,
#[cfg(all(test, not(target_arch = "wasm32")))] #[cfg(all(test, not(target_arch = "wasm32")))]
busy_publish_gate: Mutex<Option<BusyPublishGate>>, busy_publish_gate: Mutex<Option<BusyPublishGate>>,
@@ -965,6 +990,10 @@ fn worker(
#[cfg(test)] #[cfg(test)]
{ {
shared.parses.fetch_add(1, Ordering::Relaxed); shared.parses.fetch_add(1, Ordering::Relaxed);
if let Some(gate) = lock(&shared.parse_gate).clone() {
gate.entered.wait();
gate.release.wait();
}
let delay = *lock(&shared.slow_parse); let delay = *lock(&shared.slow_parse);
if let Some(delay) = delay { if let Some(delay) = delay {
std::thread::sleep(delay); std::thread::sleep(delay);
@@ -1050,6 +1079,20 @@ fn worker(
} }
} }
} }
#[cfg(all(test, not(target_arch = "wasm32")))]
let render_gate = lock(&shared.render_gates)
.iter()
.find(|gate| {
gate.page == req.page
&& gate.shared_generation == *shared_generation
&& gate.claimed.compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire).is_ok()
})
.cloned();
#[cfg(all(test, not(target_arch = "wasm32")))]
if let Some(gate) = &render_gate {
gate.entered.wait();
gate.release.wait();
}
let settings = warning_settings(&worker_settings(&config, &request_cancelled), warnings.clone()); let settings = warning_settings(&worker_settings(&config, &request_cancelled), warnings.clone());
lock(&warnings).clear(); lock(&warnings).clear();
let r = catch_unwind(AssertUnwindSafe(|| { let r = catch_unwind(AssertUnwindSafe(|| {
@@ -1068,6 +1111,10 @@ fn worker(
// and started a replacement: drop the late result and retire. // and started a replacement: drop the late result and retire.
let abandoned = lock(&shared.busy).get_mut(id).is_none_or(|slot| slot.take().is_none()); let abandoned = lock(&shared.busy).get_mut(id).is_none_or(|slot| slot.take().is_none());
if abandoned { if abandoned {
#[cfg(all(test, not(target_arch = "wasm32")))]
if let Some(gate) = render_gate {
gate.finished.wait();
}
return; return;
} }
let panicked = matches!(r, Err((_, true))); let panicked = matches!(r, Err((_, true)));
@@ -1098,6 +1145,25 @@ fn worker(
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
#[cfg(not(target_arch = "wasm32"))]
fn render_gate(page: usize, shared_generation: bool) -> super::RenderGate {
let barrier = || std::sync::Arc::new(std::sync::Barrier::new(2));
super::RenderGate {
page,
shared_generation,
claimed: Arc::new(std::sync::atomic::AtomicBool::new(false)),
entered: barrier(),
release: barrier(),
finished: barrier(),
}
}
#[cfg(not(target_arch = "wasm32"))]
fn parse_gate() -> super::ParseGate {
let barrier = || std::sync::Arc::new(std::sync::Barrier::new(2));
super::ParseGate { entered: barrier(), release: barrier() }
}
#[cfg(not(target_arch = "wasm32"))] #[cfg(not(target_arch = "wasm32"))]
fn receive_before_deadline(pool: &RenderPool) -> RenderedPage { fn receive_before_deadline(pool: &RenderPool) -> RenderedPage {
// Generous: a loaded CI machine runs many render tests at once. // Generous: a loaded CI machine runs many render tests at once.
@@ -1233,12 +1299,20 @@ mod tests {
fn page_panic_retires_shared_generation_and_healthy_pages_use_private_parsers() { fn page_panic_retires_shared_generation_and_healthy_pages_use_private_parsers() {
let pool = RenderPool::new(Arc::new(ONE_PAGE_TWICE.to_vec()), 3, RenderConfig::default()); let pool = RenderPool::new(Arc::new(ONE_PAGE_TWICE.to_vec()), 3, RenderConfig::default());
*lock(&pool.shared.panic_page) = Some(0); *lock(&pool.shared.panic_page) = Some(0);
*lock(&pool.shared.slow_page) = Some((1, std::time::Duration::from_millis(50))); // Hold page 1 at the shared render boundary until page 0 has crashed, so it can't finish
// in the shared generation first (then only one parse would happen).
let gate = render_gate(1, true);
*lock(&pool.shared.render_gates) = vec![gate.clone()];
pool.set_queue(vec![ pool.set_queue(vec![
RenderRequest { page: 0, scale: 1.0, ..Default::default() }, RenderRequest { page: 0, scale: 1.0, ..Default::default() },
RenderRequest { page: 1, scale: 1.0, ..Default::default() }, RenderRequest { page: 1, scale: 1.0, ..Default::default() },
]); ]);
let mut pages = [receive_before_deadline(&pool), receive_before_deadline(&pool)]; let crashed = receive_before_deadline(&pool);
if gate.claimed.load(Ordering::Acquire) {
gate.entered.wait();
gate.release.wait();
}
let mut pages = [crashed, receive_before_deadline(&pool)];
pages.sort_by_key(|page| page.request.page); pages.sort_by_key(|page| page.request.page);
assert!(pages[0].error.as_deref().is_some_and(|message| message.contains("crashed"))); assert!(pages[0].error.as_deref().is_some_and(|message| message.contains("crashed")));
assert!(pages[1].error.is_none(), "{:?}", pages[1].error); assert!(pages[1].error.is_none(), "{:?}", pages[1].error);
@@ -1271,7 +1345,7 @@ mod tests {
#[cfg(not(target_arch = "wasm32"))] #[cfg(not(target_arch = "wasm32"))]
#[test] #[test]
fn a_slow_parse_is_not_a_timeout_and_its_pages_render() { fn a_slow_parse_is_not_a_timeout_and_its_pages_render() {
use std::time::{Duration, Instant}; use std::time::Duration;
// Shared, then the private parsers used after a shared failure. // Shared, then the private parsers used after a shared failure.
for private in [false, true] { for private in [false, true] {
let mut pool = RenderPool::new(Arc::new(ONE_PAGE_TWICE.to_vec()), 2, RenderConfig::default()); let mut pool = RenderPool::new(Arc::new(ONE_PAGE_TWICE.to_vec()), 2, RenderConfig::default());
@@ -1279,33 +1353,31 @@ mod tests {
*lock(&pool.shared.parser.state) = ParserState::Private; *lock(&pool.shared.parser.state) = ParserState::Private;
pool.shared.parser.valid.store(false, Ordering::Release); pool.shared.parser.valid.store(false, Ordering::Release);
} }
// The parse takes 2 s against a 10 ms limit: the watchdog must leave it alone. let gate = parse_gate();
*lock(&pool.shared.slow_parse) = Some(Duration::from_secs(2)); *lock(&pool.shared.parse_gate) = Some(gate.clone());
pool.set_stuck_after(Duration::from_millis(10)); pool.set_stuck_after(Duration::ZERO);
pool.set_queue(vec![ let requests = if private {
RenderRequest { page: 0, scale: 1.0, tag: 60, ..Default::default() }, vec![RenderRequest { page: 0, scale: 1.0, tag: 60, ..Default::default() }]
RenderRequest { page: 1, scale: 1.0, tag: 61, ..Default::default() }, } else {
]); vec![
let started = Instant::now(); RenderRequest { page: 0, scale: 1.0, tag: 60, ..Default::default() },
while pool.shared.parses.load(Ordering::Relaxed) == 0 { RenderRequest { page: 1, scale: 1.0, tag: 61, ..Default::default() },
assert!(started.elapsed() < Duration::from_secs(5), "parsing did not start"); ]
std::thread::sleep(Duration::from_millis(2)); };
} pool.set_queue(requests.clone());
// Run the watchdog well past its limit while the parse is still sleeping. gate.entered.wait();
for _ in 0..20 { assert!(pool.try_recv().is_none(), "the watchdog must leave parsing alone");
assert!(pool.try_recv().is_none(), "nothing may be given up on while parsing");
std::thread::sleep(Duration::from_millis(10));
}
assert!(lock(&pool.shared.stuck).is_empty()); assert!(lock(&pool.shared.stuck).is_empty());
assert_eq!(*lock(&pool.retired_workers), 0); assert_eq!(*lock(&pool.retired_workers), 0);
assert!(!matches!(*lock(&pool.shared.parser.state), ParserState::Failed)); assert!(!matches!(*lock(&pool.shared.parser.state), ParserState::Failed));
// Rendering is timed from the end of parsing: give it a limit it can't miss. // Rendering gets its own deadline after initialization completes.
pool.set_stuck_after(Duration::from_secs(60)); pool.set_stuck_after(Duration::from_secs(60));
let mut pages = [receive_before_deadline(&pool), receive_before_deadline(&pool)]; gate.release.wait();
let mut pages: Vec<_> = (0..requests.len()).map(|_| receive_before_deadline(&pool)).collect();
pages.sort_by_key(|page| page.request.page); pages.sort_by_key(|page| page.request.page);
for (page, tag) in pages.iter().zip([60, 61]) { for (page, request) in pages.iter().zip(&requests) {
assert!(page.error.is_none(), "{:?}", page.error); assert!(page.error.is_none(), "{:?}", page.error);
assert_eq!((page.width, page.height, page.request.tag), (100, 50, tag)); assert_eq!((page.width, page.height, page.request.tag), (100, 50, request.tag));
} }
if !private { if !private {
assert_eq!(pool.shared.parses.load(Ordering::Relaxed), 1, "one shared parse"); assert_eq!(pool.shared.parses.load(Ordering::Relaxed), 1, "one shared parse");
@@ -1319,22 +1391,32 @@ mod tests {
fn shared_timeouts_retry_healthy_pages_privately_before_blacklisting() { fn shared_timeouts_retry_healthy_pages_privately_before_blacklisting() {
use std::time::Duration; use std::time::Duration;
let mut pool = RenderPool::new(Arc::new(ONE_PAGE_TWICE.to_vec()), 2, RenderConfig::default()); let mut pool = RenderPool::new(Arc::new(ONE_PAGE_TWICE.to_vec()), 2, RenderConfig::default());
// Limits and delays far apart, so a loaded machine can't flip the outcome: a healthy pool.set_stuck_after(Duration::ZERO);
// page renders privately well within 1 s, the injected delays are 3 s and 6 s. let shared_bad = render_gate(0, true);
pool.set_stuck_after(Duration::from_secs(1)); let shared_healthy = render_gate(1, true);
// Both shared workers wait as if behind the same cold decode. Only page0 let private_bad = render_gate(0, false);
// remains pathological in an independent parser; page1 must recover. *lock(&pool.shared.render_gates) = vec![shared_bad.clone(), shared_healthy.clone(), private_bad.clone()];
*lock(&pool.shared.shared_render_delay) = Some(Duration::from_secs(3));
*lock(&pool.shared.slow_page) = Some((0, Duration::from_secs(6)));
pool.set_queue(vec![ pool.set_queue(vec![
RenderRequest { page: 0, scale: 1.0, tag: 40, ..Default::default() }, RenderRequest { page: 0, scale: 1.0, tag: 40, ..Default::default() },
RenderRequest { page: 1, scale: 1.0, tag: 41, ..Default::default() }, RenderRequest { page: 1, scale: 1.0, tag: 41, ..Default::default() },
]); ]);
let mut pages = [receive_before_deadline(&pool), receive_before_deadline(&pool)]; shared_bad.entered.wait();
pages.sort_by_key(|page| page.request.page); shared_healthy.entered.wait();
assert!(pages[0].error.as_deref().is_some_and(|message| message.contains("took longer"))); assert!(pool.try_recv().is_none(), "shared timeouts retry instead of answering yet");
assert!(pages[1].error.is_none(), "{:?}", pages[1].error); shared_bad.release.wait();
assert_eq!((pages[1].width, pages[1].height, pages[1].request.tag), (100, 50, 41)); shared_healthy.release.wait();
shared_bad.finished.wait();
shared_healthy.finished.wait();
private_bad.entered.wait();
let healthy = pool.results.recv_timeout(Duration::from_secs(30)).expect("healthy private retry");
assert!(healthy.error.is_none(), "{:?}", healthy.error);
assert_eq!((healthy.request.page, healthy.width, healthy.height, healthy.request.tag), (1, 100, 50, 41));
pool.watchdog();
let bad = pool.results.recv_timeout(Duration::from_secs(30)).expect("the held private retry times out");
assert!(bad.error.as_deref().is_some_and(|message| message.contains("took longer")));
assert_eq!(bad.request.page, 0);
private_bad.release.wait();
private_bad.finished.wait();
assert!(lock(&pool.shared.stuck).contains(&RequestFailureKey::from(RenderRequest { page: 0, scale: 1.0, tag: 40, ..Default::default() }))); assert!(lock(&pool.shared.stuck).contains(&RequestFailureKey::from(RenderRequest { page: 0, scale: 1.0, tag: 40, ..Default::default() })));
assert!(!lock(&pool.shared.stuck).contains(&RequestFailureKey::from(RenderRequest { page: 1, scale: 1.0, tag: 41, ..Default::default() }))); assert!(!lock(&pool.shared.stuck).contains(&RequestFailureKey::from(RenderRequest { page: 1, scale: 1.0, tag: 41, ..Default::default() })));
assert!((2..=3).contains(&pool.shared.parses.load(Ordering::Relaxed)), "one shared parser and one or two reusable private parsers"); assert!((2..=3).contains(&pool.shared.parses.load(Ordering::Relaxed)), "one shared parser and one or two reusable private parsers");
@@ -1346,10 +1428,18 @@ mod tests {
fn exhausted_private_retry_budget_reports_pending_and_future_requests() { fn exhausted_private_retry_budget_reports_pending_and_future_requests() {
use std::time::Duration; use std::time::Duration;
let mut pool = RenderPool::new(Arc::new(ONE_PAGE_TWICE.to_vec()), 1, RenderConfig::default()); let mut pool = RenderPool::new(Arc::new(ONE_PAGE_TWICE.to_vec()), 1, RenderConfig::default());
pool.set_stuck_after(Duration::from_millis(300)); pool.set_stuck_after(Duration::ZERO);
*lock(&pool.shared.slow_page) = Some((0, Duration::from_secs(3))); let shared = render_gate(0, true);
let private = render_gate(0, false);
*lock(&pool.shared.render_gates) = vec![shared.clone(), private.clone()];
let bad = RenderRequest { page: 0, scale: 1.0, tag: 50, ..Default::default() }; let bad = RenderRequest { page: 0, scale: 1.0, tag: 50, ..Default::default() };
let healthy = RenderRequest { page: 1, scale: 1.0, tag: 51, ..Default::default() }; let healthy = RenderRequest { page: 1, scale: 1.0, tag: 51, ..Default::default() };
pool.set_queue(vec![bad]);
shared.entered.wait();
assert!(pool.try_recv().is_none(), "shared timeout schedules the private retry");
shared.release.wait();
shared.finished.wait();
private.entered.wait();
pool.set_queue(vec![bad, healthy]); pool.set_queue(vec![bad, healthy]);
let bad_result = receive_before_deadline(&pool); let bad_result = receive_before_deadline(&pool);
assert_eq!(bad_result.request, bad); assert_eq!(bad_result.request, bad);
@@ -1367,6 +1457,8 @@ mod tests {
assert!(result.error.as_deref().is_some_and(|message| message.contains("all render workers"))); assert!(result.error.as_deref().is_some_and(|message| message.contains("all render workers")));
assert_eq!(pool.shared.parses.load(Ordering::Relaxed), 2, "exhaustion performs no parser work on the consumer"); assert_eq!(pool.shared.parses.load(Ordering::Relaxed), 2, "exhaustion performs no parser work on the consumer");
let workers = std::mem::take(&mut *lock(&pool._workers)); let workers = std::mem::take(&mut *lock(&pool._workers));
private.release.wait();
private.finished.wait();
for worker in workers { for worker in workers {
worker.join().unwrap(); worker.join().unwrap();
} }
@@ -1396,24 +1488,23 @@ mod tests {
#[cfg(not(target_arch = "wasm32"))] #[cfg(not(target_arch = "wasm32"))]
#[test] #[test]
fn replacing_a_request_during_initialization_skips_its_obsolete_render() { fn replacing_a_request_during_initialization_skips_its_obsolete_render() {
use std::time::{Duration, Instant}; use std::time::Duration;
let pool = RenderPool::new(Arc::new(ONE_PAGE_TWICE.to_vec()), 1, RenderConfig::default()); let mut pool = RenderPool::new(Arc::new(ONE_PAGE_TWICE.to_vec()), 1, RenderConfig::default());
*lock(&pool.shared.slow_parse) = Some(Duration::from_millis(200)); let gate = parse_gate();
*lock(&pool.shared.slow_page) = Some((0, Duration::from_secs(10))); *lock(&pool.shared.parse_gate) = Some(gate.clone());
pool.set_stuck_after(Duration::ZERO);
pool.set_queue(vec![RenderRequest { page: 0, scale: 1.0, tag: 10, ..Default::default() }]); pool.set_queue(vec![RenderRequest { page: 0, scale: 1.0, tag: 10, ..Default::default() }]);
let deadline = Instant::now() + Duration::from_secs(2); gate.entered.wait();
while pool.shared.parses.load(Ordering::Relaxed) == 0 {
assert!(Instant::now() < deadline, "first request did not start parsing");
std::thread::sleep(Duration::from_millis(2));
}
let replaced = Instant::now();
pool.set_queue(vec![RenderRequest { page: 1, scale: 1.0, tag: 11, ..Default::default() }]); pool.set_queue(vec![RenderRequest { page: 1, scale: 1.0, tag: 11, ..Default::default() }]);
assert!(pool.try_recv().is_none(), "parsing is still held");
pool.set_stuck_after(Duration::from_secs(60));
gate.release.wait();
let page = receive_before_deadline(&pool); let page = receive_before_deadline(&pool);
assert_eq!((page.request.page, page.request.tag), (1, 11)); assert_eq!((page.request.page, page.request.tag), (1, 11));
assert!(page.error.is_none(), "{:?}", page.error); assert!(page.error.is_none(), "{:?}", page.error);
assert_eq!((page.width, page.height), (100, 50)); assert_eq!((page.width, page.height), (100, 50));
assert!(replaced.elapsed() < Duration::from_secs(8), "obsolete page (10 s) delayed the replacement request");
assert_eq!(pool.shared.parses.load(Ordering::Relaxed), 1, "the useful parser is retained"); assert_eq!(pool.shared.parses.load(Ordering::Relaxed), 1, "the useful parser is retained");
assert_eq!(pool.shared.render_started.load(Ordering::Relaxed), 1, "obsolete page was skipped before rendering");
} }
#[cfg(not(target_arch = "wasm32"))] #[cfg(not(target_arch = "wasm32"))]
@@ -1678,34 +1769,27 @@ mod tests {
// contention gets one isolated attempt, covered by the tests above. // contention gets one isolated attempt, covered by the tests above.
*lock(&pool.shared.parser.state) = ParserState::Private; *lock(&pool.shared.parser.state) = ParserState::Private;
pool.shared.parser.valid.store(false, Ordering::Release); pool.shared.parser.valid.store(false, Ordering::Release);
pool.set_stuck_after(std::time::Duration::from_millis(1000)); pool.set_stuck_after(std::time::Duration::ZERO);
*lock(&pool.shared.slow_page) = Some((0, std::time::Duration::from_secs(4))); let gate = render_gate(0, false);
*lock(&pool.shared.render_gates) = vec![gate.clone()];
let req = |page| RenderRequest { page, kind: RequestKind::Pixels, tile: None, scale: 0.5, tag: 0 }; let req = |page| RenderRequest { page, kind: RequestKind::Pixels, tile: None, scale: 0.5, tag: 0 };
pool.set_queue(vec![req(0)]); pool.set_queue(vec![req(0)]);
let recv = |pool: &RenderPool| { gate.entered.wait();
let t = std::time::Instant::now(); let first = pool.try_recv().expect("the watchdog answers while the render is held");
loop {
if let Some(p) = pool.try_recv() {
return p;
}
assert!(t.elapsed() < std::time::Duration::from_secs(8), "no result");
std::thread::sleep(std::time::Duration::from_millis(5));
}
};
let started = std::time::Instant::now();
let first = recv(&pool);
assert!(started.elapsed() < std::time::Duration::from_millis(3000), "the watchdog answers before the render ends");
assert!(first.error.as_deref().is_some_and(|e| e.contains("took longer")), "{:?}", first.error); assert!(first.error.as_deref().is_some_and(|e| e.contains("took longer")), "{:?}", first.error);
// A zero allowance would give up on the healthy page 2 as well.
pool.set_stuck_after(std::time::Duration::from_secs(60));
// The single worker is stuck, yet page 2 still renders (on the replacement worker). // The single worker is stuck, yet page 2 still renders (on the replacement worker).
pool.set_queue(vec![req(1)]); pool.set_queue(vec![req(1)]);
let second = recv(&pool); let second = receive_before_deadline(&pool);
assert!(second.error.is_none() && second.request.page == 1, "{:?}", second.error); assert!(second.error.is_none() && second.request.page == 1, "{:?}", second.error);
// Asking for the stuck page again fails fast instead of trapping another worker. // Asking for the stuck page again fails fast instead of trapping another worker.
pool.set_queue(vec![req(0)]); pool.set_queue(vec![req(0)]);
let again = recv(&pool); let again = receive_before_deadline(&pool);
assert!(again.error.as_deref().is_some_and(|e| e.contains("skipped earlier")), "{:?}", again.error); assert!(again.error.as_deref().is_some_and(|e| e.contains("skipped earlier")), "{:?}", again.error);
// The stuck worker's late result is dropped, not delivered twice. // The released worker notices its slot was cleared; its late result is dropped.
std::thread::sleep(std::time::Duration::from_millis(4200)); gate.release.wait();
gate.finished.wait();
assert!(pool.try_recv().is_none()); assert!(pool.try_recv().is_none());
} }
@@ -1713,32 +1797,22 @@ mod tests {
fn a_request_being_rendered_is_not_rendered_again() { fn a_request_being_rendered_is_not_rendered_again() {
use super::*; use super::*;
let pool = RenderPool::new(Arc::new(ONE_PAGE_TWICE.to_vec()), 2, RenderConfig::default()); let pool = RenderPool::new(Arc::new(ONE_PAGE_TWICE.to_vec()), 2, RenderConfig::default());
*lock(&pool.shared.slow_page) = Some((0, std::time::Duration::from_millis(1000))); let gate = render_gate(0, true);
*lock(&pool.shared.render_gates) = vec![gate.clone()];
let req = |page| RenderRequest { page, kind: RequestKind::Pixels, tile: None, scale: 0.5, tag: 0 }; let req = |page| RenderRequest { page, kind: RequestKind::Pixels, tile: None, scale: 0.5, tag: 0 };
pool.set_queue(vec![req(0)]); pool.set_queue(vec![req(0)]);
let t = std::time::Instant::now(); gate.entered.wait();
while !lock(&pool.shared.busy).iter().flatten().any(|busy| busy.request.page == 0) {
assert!(t.elapsed() < std::time::Duration::from_secs(8), "page 1 never started");
std::thread::sleep(std::time::Duration::from_millis(1));
}
// What the canvas sends next: the page it is still waiting for, then another one. // What the canvas sends next: the page it is still waiting for, then another one.
pool.set_queue(vec![req(0), req(1)]); pool.set_queue(vec![req(0), req(1)]);
assert!(lock(&pool.shared.busy).iter().flatten().any(|busy| { busy.request == req(0) && !busy.cancelled.load(Ordering::Acquire) })); let other = receive_before_deadline(&pool);
let mut answers = Vec::new(); assert!(other.error.is_none(), "{:?}", other.error);
let t = std::time::Instant::now(); assert_eq!(other.request.page, 1);
while answers.len() < 2 || lock(&pool.shared.busy).iter().any(Option::is_some) {
assert!(t.elapsed() < std::time::Duration::from_secs(8), "answers so far: {answers:?}");
if let Some(p) = pool.try_recv() {
assert!(p.error.is_none(), "{:?}", p.error);
answers.push(p.request.page);
}
std::thread::sleep(std::time::Duration::from_millis(1));
}
while let Some(p) = pool.try_recv() {
answers.push(p.request.page);
}
// Page 2 went to the idle worker instead of waiting behind a second copy of page 1. // Page 2 went to the idle worker instead of waiting behind a second copy of page 1.
assert_eq!(answers, vec![1, 0]); assert_eq!(pool.shared.render_started.load(Ordering::Relaxed), 2, "each page rendered once");
gate.release.wait();
let first = receive_before_deadline(&pool);
assert!(first.error.is_none(), "{:?}", first.error);
assert_eq!(first.request.page, 0);
assert_eq!(pool.shared.render_started.load(Ordering::Relaxed), 2, "each page rendered once"); assert_eq!(pool.shared.render_started.load(Ordering::Relaxed), 2, "each page rendered once");
} }
@@ -2758,21 +2832,19 @@ trailer << /Root 1 0 R >>
// give up on (and stop) a second worker too. // give up on (and stop) a second worker too.
*lock(&pool.shared.parser.state) = ParserState::Private; *lock(&pool.shared.parser.state) = ParserState::Private;
pool.shared.parser.valid.store(false, Ordering::Release); pool.shared.parser.valid.store(false, Ordering::Release);
pool.set_stuck_after(std::time::Duration::from_millis(100)); pool.set_stuck_after(std::time::Duration::ZERO);
*lock(&pool.shared.slow_page) = Some((0, std::time::Duration::from_millis(1500))); let gate = render_gate(0, false);
*lock(&pool.shared.render_gates) = vec![gate.clone()];
pool.set_queue(vec![RenderRequest { page: 0, kind: RequestKind::Pixels, tile: None, scale: 0.5, tag: 0 }]); pool.set_queue(vec![RenderRequest { page: 0, kind: RequestKind::Pixels, tile: None, scale: 0.5, tag: 0 }]);
let t = std::time::Instant::now(); gate.entered.wait();
let first = loop { let first = pool.try_recv().expect("the watchdog answers for the held render");
if let Some(p) = pool.try_recv() {
break p;
}
assert!(t.elapsed() < std::time::Duration::from_secs(30), "no answer");
std::thread::sleep(std::time::Duration::from_millis(5));
};
assert!(first.error.as_deref().is_some_and(|e| e.contains("took longer")), "{:?}", first.error); assert!(first.error.as_deref().is_some_and(|e| e.contains("took longer")), "{:?}", first.error);
// The abandoned worker is told to stop at its next operator; its replacement isn't. // The abandoned worker is told to stop at its next operator; its replacement isn't.
let stop: Vec<bool> = lock(&pool.shared.stop).iter().map(|s| s.load(std::sync::atomic::Ordering::Relaxed)).collect(); let stop: Vec<bool> = lock(&pool.shared.stop).iter().map(|s| s.load(std::sync::atomic::Ordering::Relaxed)).collect();
assert_eq!(stop, vec![true, false]); assert_eq!(stop, vec![true, false]);
gate.release.wait();
gate.finished.wait();
assert!(pool.try_recv().is_none(), "the abandoned render must not publish a late result");
} }
#[test] #[test]