shifathosn
/
pdfcraft
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

M2: cancel obsolete in-flight requests (#702) 

* render: cancel obsolete in-flight requests

* #623: recheck cancellation after busy publication

* #623: record rendering session in roadmap

* M2: a request cancelled during the parse doesn't leave its worker unable to render

The abort was cached as the worker's parse result, so every later request on that worker failed as
unparsable. Also drops ROADMAP edits that described other PRs.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Brandon Thomas <echelon@gmail.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
cyfung1031 2026-10-10 19:22:24 +09:00 committed by GitHub
parent 7fafadc3a2
commit 742c42ab63
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 246 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

257
crates/render/src/raster.rs
View File

@ -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<AtomicBool>,
valid: Arc<AtomicBool>,
}
#[cfg(all(test, not(target_arch = "wasm32")))]
#[derive(Clone)]
struct BusyPublishGate {
request: RenderRequest,
entered: std::sync::mpsc::Sender<()>,
release: Arc<Mutex<Receiver<()>>>,
}
/// 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<Option<BusyPublishGate>>,
#[cfg(all(test, not(target_arch = "wasm32")))]
parser_stage_entries: Mutex<Vec<RenderRequest>>,
#[cfg(all(test, not(target_arch = "wasm32")))]
shared_render_delay: Mutex<Option<std::time::Duration>>,
}
@ -495,6 +509,24 @@ fn lock<T>(m: &Mutex<T>) -> 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<RenderRequest>) {
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<std::sync::atomic::AtomicBool>) -> InterpreterSettings {
InterpreterSettings { cancelled: Some(stop.clone()), ..config.settings() }
@ -843,7 +888,6 @@ fn worker(
stop: Arc<std::sync::atomic::AtomicBool>,
) {
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::*;

29
crates/render/src/raster/work_queue.rs
View File

@ -27,14 +27,20 @@ impl From<RenderRequest> 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<Arc<AtomicBool>>,
valid: Vec<Arc<AtomicBool>>,
}
fn valid(token: &Option<Arc<AtomicBool>>) -> bool {
token.as_ref().is_none_or(|v| v.load(Ordering::Acquire))
fn valid(tokens: &[Arc<AtomicBool>]) -> 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<RenderRequest>) {
pub(super) fn replace(&self, requests: Vec<RenderRequest>) -> Vec<RenderRequest> {
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<RenderRequest> {
@ -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<Arc<AtomicBool>>) -> 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<Arc<AtomicBool>>) -> 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();
}