diff --git a/crates/render/src/raster.rs b/crates/render/src/raster.rs index 908bc2ad..fb59c44e 100644 --- a/crates/render/src/raster.rs +++ b/crates/render/src/raster.rs @@ -450,11 +450,21 @@ enum BusyStage { } #[cfg(not(target_arch = "wasm32"))] -#[derive(Clone, Copy)] +#[derive(Clone)] struct Busy { request: RenderRequest, since: std::time::Instant, stage: BusyStage, + cancelled: Arc, + valid: Arc, +} + +#[cfg(all(test, not(target_arch = "wasm32")))] +#[derive(Clone)] +struct BusyPublishGate { + request: RenderRequest, + entered: std::sync::mpsc::Sender<()>, + release: Arc>>, } /// State shared by the pool and its workers. @@ -488,6 +498,10 @@ struct Shared { #[cfg(all(test, not(target_arch = "wasm32")))] render_started: std::sync::atomic::AtomicUsize, #[cfg(all(test, not(target_arch = "wasm32")))] + busy_publish_gate: Mutex>, + #[cfg(all(test, not(target_arch = "wasm32")))] + parser_stage_entries: Mutex>, + #[cfg(all(test, not(target_arch = "wasm32")))] shared_render_delay: Mutex>, } @@ -495,6 +509,24 @@ fn lock(m: &Mutex) -> std::sync::MutexGuard<'_, T> { m.lock().unwrap_or_else(|poisoned| poisoned.into_inner()) } +#[cfg(not(target_arch = "wasm32"))] +fn same_request(a: RenderRequest, b: RenderRequest) -> bool { + a.page == b.page && a.kind == b.kind && a.tile == b.tile && a.scale.to_bits() == b.scale.to_bits() && a.tag == b.tag +} + +#[cfg(not(target_arch = "wasm32"))] +impl Shared { + fn cancel_in_flight(&self, obsolete: &[RenderRequest]) { + let busy = lock(&self.busy); + for busy in busy.iter().flatten() { + if obsolete.iter().copied().any(|request| same_request(request, busy.request)) { + busy.cancelled.store(true, Ordering::Release); + busy.valid.store(false, Ordering::Release); + } + } + } +} + #[cfg(not(target_arch = "wasm32"))] fn set_busy_stage(shared: &Shared, id: usize, stage: BusyStage) -> bool { let mut slots = lock(&shared.busy); @@ -704,13 +736,19 @@ impl RenderPool { } /// Replace desired work (most urgent first), retaining matching in-flight/completed jobs. - /// Obsolete results are released before they reach the caller. In-flight work is not killed. + /// Obsolete results are released before they reach the caller, and active obsolete renders + /// are asked to stop at their next cancellation point. /// /// The contract: a result is delivered only while its request is in the latest queue. /// A caller still waiting for a page (or text, or a thumbnail) must keep listing it in /// every call, or the finished result is dropped to save memory. pub fn set_queue(&self, requests: Vec) { - self.shared.queue.replace(requests); + #[cfg(not(target_arch = "wasm32"))] + let obsolete = self.shared.queue.replace(requests); + #[cfg(target_arch = "wasm32")] + let _ = self.shared.queue.replace(requests); + #[cfg(not(target_arch = "wasm32"))] + self.shared.cancel_in_flight(&obsolete); self.wake_workers(); } @@ -765,12 +803,14 @@ impl RenderPool { // Lock order: `busy`, then `stop`. let stop = lock(&self.shared.stop); for (id, slot) in busy.iter_mut().enumerate() { - if let Some(busy) = *slot + if let Some(busy) = slot.clone() && busy.stage != BusyStage::Parsing && busy.since.elapsed() > self.stuck_after { *slot = None; // the worker sees this and exits when it returns // Told to stop at its next operator; its replacement retries the page. + busy.cancelled.store(true, Ordering::Release); + busy.valid.store(false, Ordering::Release); if let Some(s) = stop.get(id) { s.store(true, std::sync::atomic::Ordering::Relaxed); } @@ -820,13 +860,18 @@ impl RenderPool { impl Drop for RenderPool { fn drop(&mut self) { + #[cfg(not(target_arch = "wasm32"))] + for busy in lock(&self.shared.busy).iter().flatten() { + busy.cancelled.store(true, Ordering::Release); + busy.valid.store(false, Ordering::Release); + } for s in lock(&self.shared.stop).iter() { s.store(true, std::sync::atomic::Ordering::Relaxed); } } } -/// A worker's interpreter settings: the document's, stopping once `stop` is set. +/// A worker's interpreter settings: the document's, stopping once the request is cancelled. #[cfg(not(target_arch = "wasm32"))] fn worker_settings(config: &RenderConfig, stop: &Arc) -> InterpreterSettings { InterpreterSettings { cancelled: Some(stop.clone()), ..config.settings() } @@ -843,7 +888,6 @@ fn worker( stop: Arc, ) { let warnings = Arc::new(Mutex::new(Vec::new())); - let settings = warning_settings(&worker_settings(&config, &stop), warnings.clone()); // A local cell keeps Pdf borrows used by RenderCache stable. Drop both on an // epoch change/panic before publishing an error or taking another request. loop { @@ -863,6 +907,13 @@ fn worker( if shared.queue.cancel_obsolete(req) { continue; } + #[cfg(all(test, not(target_arch = "wasm32")))] + if let Some(gate) = lock(&shared.busy_publish_gate).clone().filter(|gate| same_request(gate.request, req)) { + let _ = gate.entered.send(()); + let _ = lock(&gate.release).recv(); + } + let request_cancelled = Arc::new(AtomicBool::new(false)); + let request_valid = Arc::new(AtomicBool::new(true)); if lock(&shared.stuck).contains(&req.into()) { let error = format!("page {} was skipped earlier because rendering failed or took too long", req.page + 1); let page = RenderedPage { @@ -889,32 +940,76 @@ fn worker( }, ); if let Some(slot) = lock(&shared.busy).get_mut(id) { - *slot = Some(Busy { request: req, since: std::time::Instant::now(), stage }); + *slot = Some(Busy { + request: req, + since: std::time::Instant::now(), + stage, + cancelled: request_cancelled.clone(), + valid: request_valid.clone(), + }); } - let (pdf, cache, shared_generation) = parsed.get_or_init(|| { - let load = || { - #[cfg(test)] - { - shared.parses.fetch_add(1, Ordering::Relaxed); - let delay = *lock(&shared.slow_parse); - if let Some(delay) = delay { - std::thread::sleep(delay); + if shared.queue.cancel_obsolete(req) { + if lock(&shared.busy).get_mut(id).is_none_or(|slot| slot.take().is_none()) { + return; + } + continue; + } + #[cfg(all(test, not(target_arch = "wasm32")))] + lock(&shared.parser_stage_entries).push(req); + // A request cancelled while it waited for the parser must not leave its abort + // behind as this worker's parse: every later request would then fail as unparsable. + let aborted = || request_cancelled.load(Ordering::Acquire) || stop.load(Ordering::Acquire); + if parsed.get().is_none() { + let init = { + let load = || { + #[cfg(test)] + { + shared.parses.fetch_add(1, Ordering::Relaxed); + let delay = *lock(&shared.slow_parse); + if let Some(delay) = delay { + std::thread::sleep(delay); + } + } + shared.stats.parser_builds.fetch_add(1, Ordering::Relaxed); + parse(&bytes, config.password.as_deref()) + }; + match shared.parser.acquire(load, || { + request_cancelled.load(Ordering::Acquire) + || stop.load(Ordering::Acquire) + || matches!(wake.try_recv(), Err(std::sync::mpsc::TryRecvError::Disconnected)) + }) { + Some(pdf) => { + let shared_generation = pdf.is_some(); + (pdf, RenderCache::new(), shared_generation) + } + None => { + let pdf = if !request_cancelled.load(Ordering::Acquire) + && !stop.load(Ordering::Acquire) + && set_busy_stage(&shared, id, BusyStage::Parsing) + { + load().map(Arc::new) + } else { + None + }; + (pdf, RenderCache::new(), false) } } - shared.stats.parser_builds.fetch_add(1, Ordering::Relaxed); - parse(&bytes, config.password.as_deref()) }; - match shared.parser.acquire(load, || matches!(wake.try_recv(), Err(std::sync::mpsc::TryRecvError::Disconnected))) { - Some(pdf) => { - let shared_generation = pdf.is_some(); - (pdf, RenderCache::new(), shared_generation) - } - None => { - let pdf = if set_busy_stage(&shared, id, BusyStage::Parsing) { load().map(Arc::new) } else { None }; - (pdf, RenderCache::new(), false) + if aborted() { + if lock(&shared.busy).get_mut(id).is_none_or(|slot| slot.take().is_none()) { + return; } + continue; } - }); + let _ = parsed.set(init); + } + // Set just above when it was empty. + let Some((pdf, cache, shared_generation)) = parsed.get() else { + if lock(&shared.busy).get_mut(id).is_none_or(|slot| slot.take().is_none()) { + return; + } + continue; + }; // Acquisition may have waited or parsed privately after shared // fallback. Retire before any render work if the watchdog gave // this request to a replacement while initialization ran. @@ -934,7 +1029,7 @@ fn worker( } break Some(( finish(req, start, Err(("the shared parser was retired".into(), false)), Vec::new()), - Some(shared.parser.valid.clone()), + vec![request_valid, shared.parser.valid.clone()], )); } #[cfg(test)] @@ -949,9 +1044,13 @@ fn worker( if let Some((page, delay)) = slow && page == req.page { - std::thread::sleep(delay); + let deadline = std::time::Instant::now() + delay; + while std::time::Instant::now() < deadline && !request_cancelled.load(Ordering::Acquire) { + std::thread::sleep(std::time::Duration::from_millis(1)); + } } } + let settings = warning_settings(&worker_settings(&config, &request_cancelled), warnings.clone()); lock(&warnings).clear(); let r = catch_unwind(AssertUnwindSafe(|| { #[cfg(test)] @@ -975,19 +1074,22 @@ fn worker( if panicked { shared.parser.abandon(); lock(&shared.stuck).insert(req.into()); - break Some((finish(req, start, r, request_warnings), None)); + break Some((finish(req, start, r, request_warnings), vec![request_valid])); + } + let mut valid = vec![request_valid]; + if *shared_generation { + valid.push(shared.parser.valid.clone()); } - let valid = shared_generation.then(|| shared.parser.valid.clone()); if *shared_generation && !shared.parser.valid.load(Ordering::Acquire) { break Some((finish(req, start, r, request_warnings), valid)); } - if out.send_valid(finish(req, start, r, request_warnings), valid).is_err() { + if out.send_valid_with_tokens(finish(req, start, r, request_warnings), valid).is_err() { return; } } }; // drop the parser and its borrowing cache before delivering reset_result if let Some((page, valid)) = reset_result - && out.send_valid(page, valid).is_err() + && out.send_valid_with_tokens(page, valid).is_err() { return; } @@ -1621,6 +1723,7 @@ mod tests { } // What the canvas sends next: the page it is still waiting for, then another one. 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 mut answers = Vec::new(); let t = std::time::Instant::now(); while answers.len() < 2 || lock(&pool.shared.busy).iter().any(Option::is_some) { @@ -1639,6 +1742,96 @@ mod tests { assert_eq!(pool.shared.render_started.load(Ordering::Relaxed), 2, "each page rendered once"); } + #[cfg(not(target_arch = "wasm32"))] + #[test] + fn replacing_an_in_flight_request_does_not_delay_the_replacement() { + use super::*; + use std::time::{Duration, Instant}; + + let pool = RenderPool::new(Arc::new(ONE_PAGE_TWICE.to_vec()), 1, RenderConfig::default()); + let a = RenderRequest { page: 0, scale: 0.5, tag: 101, ..Default::default() }; + let b = RenderRequest { page: 1, scale: 0.5, tag: 102, ..Default::default() }; + *lock(&pool.shared.slow_page) = Some((a.page, Duration::from_secs(10))); + pool.set_queue(vec![a]); + + let started = Instant::now(); + while pool.shared.render_started.load(Ordering::Relaxed) == 0 { + assert!(started.elapsed() < Duration::from_secs(2), "request A never started"); + std::thread::sleep(Duration::from_millis(1)); + } + + let replaced = Instant::now(); + pool.set_queue(vec![b]); + assert!(lock(&pool.shared.busy).iter().flatten().any(|busy| busy.cancelled.load(Ordering::Acquire)), "request A was not cancelled"); + let page = receive_before_deadline(&pool); + assert_eq!(page.request, b); + assert!(page.error.is_none(), "replacement request failed: {:?}", page.error); + assert!(replaced.elapsed() < Duration::from_secs(3), "obsolete request A delayed B"); + + let deadline = Instant::now() + Duration::from_millis(100); + while Instant::now() < deadline { + assert!(pool.try_recv().is_none(), "obsolete request produced a result"); + std::thread::sleep(Duration::from_millis(5)); + } + } + + /// A request cancelled while its worker waited for the shared parse must not leave that + /// worker unable to render: the abort is not a parse result. + #[cfg(not(target_arch = "wasm32"))] + #[test] + fn a_request_cancelled_while_waiting_for_the_parse_leaves_its_worker_usable() { + use super::*; + use std::time::{Duration, Instant}; + + let pool = RenderPool::new(Arc::new(ONE_PAGE_TWICE.to_vec()), 2, RenderConfig::default()); + *lock(&pool.shared.slow_parse) = Some(Duration::from_millis(500)); + let a = RenderRequest { page: 0, scale: 0.5, tag: 301, ..Default::default() }; + let c = RenderRequest { page: 1, scale: 0.5, tag: 302, ..Default::default() }; + pool.set_queue(vec![a, c]); + let started = Instant::now(); + while lock(&pool.shared.busy).iter().flatten().count() < 2 { + assert!(started.elapsed() < Duration::from_secs(2), "both requests never started"); + std::thread::sleep(Duration::from_millis(1)); + } + *lock(&pool.shared.slow_parse) = None; + for round in 0..4u64 { + let x = RenderRequest { page: 0, scale: 0.5, tag: 400 + 2 * round, ..Default::default() }; + let y = RenderRequest { page: 1, scale: 0.5, tag: 401 + 2 * round, ..Default::default() }; + pool.set_queue(vec![x, y]); + for _ in 0..2 { + let page = receive_before_deadline(&pool); + assert!(page.request == x || page.request == y, "round {round}: obsolete {:?}", page.request); + assert!(page.error.is_none(), "round {round}: {:?}", page.error); + } + } + } + + #[cfg(not(target_arch = "wasm32"))] + #[test] + fn request_removed_before_busy_publication_skips_parser_acquisition() { + use std::{sync::mpsc, time::Duration}; + + let pool = RenderPool::new(Arc::new(ONE_PAGE_TWICE.to_vec()), 1, RenderConfig::default()); + let a = RenderRequest { page: 0, scale: 0.5, tag: 201, ..Default::default() }; + let b = RenderRequest { page: 1, scale: 0.5, tag: 202, ..Default::default() }; + let (entered_tx, entered_rx) = mpsc::channel(); + let (release_tx, release_rx) = mpsc::channel(); + *lock(&pool.shared.busy_publish_gate) = Some(BusyPublishGate { request: a, entered: entered_tx, release: Arc::new(Mutex::new(release_rx)) }); + + pool.set_queue(vec![a]); + entered_rx.recv_timeout(Duration::from_secs(2)).expect("request A passed its first obsolescence check"); + assert!(!lock(&pool.shared.busy).iter().flatten().any(|busy| same_request(busy.request, a))); + pool.set_queue(vec![b]); + release_tx.send(()).unwrap(); + + let page = receive_before_deadline(&pool); + assert_eq!(page.request, b); + assert!(page.error.is_none(), "replacement request failed: {:?}", page.error); + assert_eq!(*lock(&pool.shared.parser_stage_entries), vec![b], "obsolete A must not enter parser acquisition"); + assert_eq!(pool.shared.render_started.load(Ordering::Relaxed), 1, "only replacement B is rasterized"); + assert!(pool.try_recv().is_none(), "obsolete request A did not publish a stale result"); + } + #[test] fn re_sent_queues_render_each_page_once() { use super::*; diff --git a/crates/render/src/raster/work_queue.rs b/crates/render/src/raster/work_queue.rs index fd61c074..eaa0cbab 100644 --- a/crates/render/src/raster/work_queue.rs +++ b/crates/render/src/raster/work_queue.rs @@ -27,14 +27,20 @@ impl From for Key { } } +impl Key { + fn request(self) -> RenderRequest { + RenderRequest { page: self.page, kind: self.kind, tile: self.tile, scale: f32::from_bits(self.scale), tag: self.tag } + } +} + struct Completed { page: RenderedPage, bytes: usize, - valid: Option>, + valid: Vec>, } -fn valid(token: &Option>) -> bool { - token.as_ref().is_none_or(|v| v.load(Ordering::Acquire)) +fn valid(tokens: &[Arc]) -> bool { + tokens.iter().all(|token| token.load(Ordering::Acquire)) } #[cfg(any(test, not(target_arch = "wasm32")))] @@ -124,21 +130,23 @@ impl Default for WorkQueue { } impl WorkQueue { - pub(super) fn replace(&self, requests: Vec) { + pub(super) fn replace(&self, requests: Vec) -> Vec { let mut state = lock(&self.state); if state.closed { - return; + return Vec::new(); } let mut desired = HashSet::with_capacity(requests.len()); let requests: Vec<_> = requests.into_iter().filter(|r| desired.insert(Key::from(*r))).collect(); state.results.retain(|r| desired.contains(&Key::from(r.page.request)) && valid(&r.valid)); state.bytes = state.results.iter().fold(0usize, |n, r| n.saturating_add(r.bytes)); let completed: HashSet<_> = state.results.iter().map(|r| Key::from(r.page.request)).collect(); + let obsolete = state.active.iter().filter(|key| !desired.contains(key)).copied().map(Key::request).collect(); state.desired = requests.iter().enumerate().map(|(priority, req)| (Key::from(*req), priority)).collect(); state.pending = requests.into_iter().rev().filter(|r| !state.active.contains(&Key::from(*r)) && !completed.contains(&Key::from(*r))).collect(); drop(state); self.changed.notify_all(); + obsolete } pub(super) fn pop(&self) -> Option { @@ -179,6 +187,11 @@ impl WorkQueue { /// using the pool's fallback parser, including when failure occurs after enqueueing. #[cfg(any(test, not(target_arch = "wasm32")))] pub(super) fn send_valid(&self, page: RenderedPage, token: Option>) -> Result<(), ()> { + self.send_valid_with_tokens(page, token.into_iter().collect()) + } + + #[cfg(any(test, not(target_arch = "wasm32")))] + pub(super) fn send_valid_with_tokens(&self, page: RenderedPage, tokens: Vec>) -> Result<(), ()> { let key = Key::from(page.request); let bytes = result_bytes(&page); let mut state = lock(&self.state); @@ -191,7 +204,7 @@ impl WorkQueue { state.active.remove(&key); return Ok(()); } - if !valid(&token) { + if !valid(&tokens) { state.retry(page.request); self.changed.notify_all(); return Ok(()); @@ -199,7 +212,7 @@ impl WorkQueue { if state.results.len() < self.max_count && (state.results.is_empty() || state.bytes.saturating_add(bytes) <= self.max_bytes) { state.active.remove(&key); state.bytes = state.bytes.saturating_add(bytes); - state.results.push_back(Completed { page, bytes, valid: token }); + state.results.push_back(Completed { page, bytes, valid: tokens }); drop(state); self.changed.notify_all(); return Ok(()); @@ -236,7 +249,7 @@ impl WorkQueue { } } state.bytes = state.bytes.saturating_add(bytes); - state.results.push_front(Completed { page, bytes, valid: None }); + state.results.push_front(Completed { page, bytes, valid: Vec::new() }); drop(state); self.changed.notify_all(); }