From d0b2a32744cfda8edd9dd9abb5b5c79f424a4cdd Mon Sep 17 00:00:00 2001 From: quincy-jc Date: Mon, 9 Nov 2026 11:53:27 +1100 Subject: [PATCH] perf(optimize): work out pictures on threads and skip needless page reads (#6230) --- Cargo.lock | 15 + crates/cos/src/budget.rs | 5 + crates/engine/src/optimizer.rs | 27 +- crates/engine/src/tests.rs | 18 +- crates/optimize/Cargo.toml | 4 + crates/optimize/README.md | 6 + crates/optimize/src/fragments.rs | 44 +- crates/optimize/src/images.rs | 406 ++++++++++++++---- crates/optimize/src/lib.rs | 7 + crates/optimize/src/pool.rs | 262 +++++++++++ crates/optimize/tests/jpeg_color_transform.rs | 16 +- crates/optimize/tests/optimize.rs | 141 +++++- crates/ui-egui/src/workflows/optimize_ui.rs | 22 +- 13 files changed, 870 insertions(+), 103 deletions(-) create mode 100644 crates/optimize/src/pool.rs diff --git a/Cargo.lock b/Cargo.lock index ea3a743bf8..5bcb1d80dc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4592,6 +4592,7 @@ dependencies = [ "pdfcraft-render", "proptest", "thiserror 2.0.21", + "thread-priority", "zune-jpeg", ] @@ -6955,6 +6956,20 @@ dependencies = [ "syn 3.0.6", ] +[[package]] +name = "thread-priority" +version = "3.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8d2e834949be5111506bb252643498af1514f600d9e1dceedaa42afae155b67f" +dependencies = [ + "bitflags 2.13.2", + "cfg-if", + "libc", + "log", + "rustversion", + "windows", +] + [[package]] name = "tiff" version = "0.11.3" diff --git a/crates/cos/src/budget.rs b/crates/cos/src/budget.rs index 1a99885944..5271e591b7 100644 --- a/crates/cos/src/budget.rs +++ b/crates/cos/src/budget.rs @@ -209,6 +209,11 @@ impl ImageBudget { self.refused } + /// Charges `work` steps paid for out of another allowance, by a thread working on its own. + pub fn charge(&mut self, work: u64) { + self.work = self.work.saturating_sub(work); + } + /// The steps of decoding still allowed. pub fn work_left(&self) -> u64 { self.work diff --git a/crates/engine/src/optimizer.rs b/crates/engine/src/optimizer.rs index 30a14c5695..1a741c0912 100644 --- a/crates/engine/src/optimizer.rs +++ b/crates/engine/src/optimizer.rs @@ -12,11 +12,12 @@ use pdfcraft_cos::{SaveOptions, write_full}; use crate::{DocId, EditError, Hidden, OptimizeReport, Session, guard, optimize}; /// Where an optimization is. -#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +#[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum OptimizeStage { - /// Removing the user data chosen under Discard User Data. - #[default] + /// Removing the user data chosen under Discard User Data, if any was. Discarding, + /// Reading what the pages draw: `done` of `total` read. + Reading { done: usize, total: usize }, /// Merging the fragments of a page picture into one image. Fragments, /// About to process image `done` (0-based) of `total`. @@ -29,14 +30,21 @@ pub enum OptimizeStage { Writing, } +impl Default for OptimizeStage { + fn default() -> Self { + OptimizeStage::Reading { done: 0, total: 0 } + } +} + impl OptimizeStage { - /// How far along the whole run is, from 0 to 1. Images take most of the time; the rest are - /// small fixed steps. + /// How far along the whole run is, from 0 to 1, weighed by where the time goes: reading pages + /// and pictures. The rest are small fixed steps. pub fn fraction(self) -> f32 { + let part = |done: usize, total: usize| if total == 0 { 1.0 } else { done.min(total) as f32 / total as f32 }; match self { - OptimizeStage::Discarding => 0.0, - OptimizeStage::Fragments => 0.03, - OptimizeStage::Images { done, total } => 0.05 + 0.75 * if total == 0 { 1.0 } else { done.min(total) as f32 / total as f32 }, + OptimizeStage::Discarding | OptimizeStage::Fragments => 0.0, + OptimizeStage::Reading { done, total } => 0.25 * if total == 0 { 0.0 } else { part(done, total) }, + OptimizeStage::Images { done, total } => 0.25 + 0.55 * part(done, total), OptimizeStage::CleaningUp => 0.8, OptimizeStage::Merging => 0.85, OptimizeStage::Writing => 0.9, @@ -91,10 +99,10 @@ impl OptimizeJob { pdfcraft_forms::clear_password_values(&mut cos).map_err(|e| EditError::Write(e.to_string()))?; pdfcraft_xfa::clear_password_values(&mut cos).map_err(|e| EditError::Write(e.to_string()))?; let mut step = |stage| if progress(stage) { Ok(()) } else { Err(EditError::Cancelled) }; - step(OptimizeStage::Discarding)?; let discarded = if discard.is_empty() { Vec::new() } else { + step(OptimizeStage::Discarding)?; // Every ticked category, as `remove_hidden` decides between them: with form fields // ticked, discarding hidden text also takes the hidden text the flattened appearances // bring onto the page, which no scan before flattening counts. It refuses when its scan @@ -112,6 +120,7 @@ impl OptimizeJob { }; let report = optimize::optimize_with_progress(&mut cos, &settings, &mut |s| { step(match s { + optimize::Stage::Reading { done, total } => OptimizeStage::Reading { done, total }, optimize::Stage::Fragments => OptimizeStage::Fragments, optimize::Stage::Images { done, total } => OptimizeStage::Images { done, total }, optimize::Stage::CleanUp => OptimizeStage::CleaningUp, diff --git a/crates/engine/src/tests.rs b/crates/engine/src/tests.rs index efb196de56..2d8eab60db 100644 --- a/crates/engine/src/tests.rs +++ b/crates/engine/src/tests.rs @@ -4083,7 +4083,9 @@ fn an_optimize_job_reports_its_stages_and_can_be_cancelled() { }) .unwrap(); assert!(bytes.starts_with(b"%PDF-")); - assert_eq!(stages.first(), Some(&OptimizeStage::Discarding)); + assert_eq!(stages.first(), Some(&OptimizeStage::Reading { done: 0, total: 2 }), "it says what it does first"); + assert!(!stages.contains(&OptimizeStage::Discarding), "nothing was chosen to discard: {stages:?}"); + assert!(stages.iter().any(|st| matches!(st, OptimizeStage::Images { .. })), "{stages:?}"); assert_eq!(&stages[stages.len() - 3..], [OptimizeStage::CleaningUp, OptimizeStage::Merging, OptimizeStage::Writing]); let fractions: Vec = stages.iter().map(|st| st.fraction()).collect(); assert!(fractions.windows(2).all(|w| w[0] <= w[1]), "the bar never goes back: {fractions:?}"); @@ -4094,6 +4096,20 @@ fn an_optimize_job_reports_its_stages_and_can_be_cancelled() { "no images: no division by zero" ); + // What the bar shows before anything is reported says nothing of user data either. + assert_eq!(OptimizeStage::default(), OptimizeStage::Reading { done: 0, total: 0 }); + assert_eq!(OptimizeStage::default().fraction(), 0.0); + assert!(OptimizeStage::Reading { done: 2, total: 4 }.fraction() > 0.0, "reading pages moves the bar"); + + // Discarding says so when something is. This document has no metadata, which only stops it after + // the stage is reported. + let mut with_discard = Vec::new(); + let _ = s.optimize_job(id, &settings, &[crate::Hidden::Metadata]).unwrap().run(|st| { + with_discard.push(st); + true + }); + assert_eq!(with_discard.first(), Some(&OptimizeStage::Discarding)); + // Cancelled at the merge: nothing is written, the open document is untouched. let before = s.get(id).unwrap().bytes.clone(); let r = s.optimize_job(id, &settings, &[]).unwrap().run(|st| st != OptimizeStage::Merging); diff --git a/crates/optimize/Cargo.toml b/crates/optimize/Cargo.toml index d1a759d468..dd9802e303 100644 --- a/crates/optimize/Cargo.toml +++ b/crates/optimize/Cargo.toml @@ -22,6 +22,10 @@ image = { version = "0.25", default-features = false, features = ["jpeg"] } # The same JPEG decoder, used directly to read samples untransformed (/ColorTransform 0). zune-jpeg = "0.5" +# Lets other programs go first while a copy is being made. +[target.'cfg(not(target_arch = "wasm32"))'.dependencies] +thread-priority = "3.1" + [dev-dependencies] pdfcraft-render = { workspace = true } proptest.workspace = true diff --git a/crates/optimize/README.md b/crates/optimize/README.md index fd7c60ae85..4c0549b1c8 100644 --- a/crates/optimize/README.md +++ b/crates/optimize/README.md @@ -121,6 +121,12 @@ let report = optimize_with_progress(&mut doc, &settings, &mut |stage| keep_going 1,000 scanned pages' worth, plus more for each byte of the file, enough to decode a stream about 8 times at Flate's best. Images past it stay as they are, and `Report::images_refused` counts those the image step read. Streams it only stores with Flate (below) aren't counted. +- **Threads**: pictures are worked out on up to half the machine's processors (at most 8, one on + the web), and as many as 1 GiB holds of the largest picture. Those threads go at low priority + where the system allows it, so other programs go first; pictures done on the calling thread keep + its priority. The document only changes in order on one thread, and what a picture costs is paid + in that order, so the file written, and the pictures the run's decoding allowance refuses, are + the same with any number. - **Discard objects:** thumbnails, alternate images, document tags, print settings. - **Clean up:** Flate for streams with no filter; for streams with an outer ASCII85 or ASCIIHex layer, that layer is removed when the result is smaller (the inner filters keep their encoded diff --git a/crates/optimize/src/fragments.rs b/crates/optimize/src/fragments.rs index b5662a2bb6..34c8f9ca76 100644 --- a/crates/optimize/src/fragments.rs +++ b/crates/optimize/src/fragments.rs @@ -243,14 +243,22 @@ fn merge_page( images: &mut ImageBudget, ) -> Result { let Some(contents) = doc.get(page).as_dict().and_then(|d| d.get(b"Contents").cloned()) else { return Ok(0) }; + let resources = page_resources(doc, page).unwrap_or_default(); + let mut xobjects = resources.get(b"XObject").map(|x| doc.resolve(x)).and_then(|x| x.as_dict().cloned()).unwrap_or_default(); + // Reading a heavy page takes seconds, so not one that has no picture that could be a fragment. + if !xobjects.iter().any(|(name, _)| fragment_image(doc, &xobjects, name).is_some()) { + return Ok(0); + } // A page whose content isn't all known is left alone: rewriting it would drop the rest. let Some(data) = content_bytes(doc, &contents, &mut DecodeBudget::page()) else { return Ok(0) }; + // Nor one that doesn't draw enough things for a run, each fragment being a `Do`. + if data.windows(2).filter(|w| *w == b"Do").count() < MIN_FRAGMENTS { + return Ok(0); + } let parsed = parse(&data); if parsed.skipped > 0 { return Ok(0); } - let resources = page_resources(doc, page).unwrap_or_default(); - let mut xobjects = resources.get(b"XObject").map(|x| doc.resolve(x)).and_then(|x| x.as_dict().cloned()).unwrap_or_default(); let ops = parsed.ops; let mut out: Vec = Vec::with_capacity(ops.len()); let (mut i, mut merged, mut k) = (0, 0, 0); @@ -339,6 +347,9 @@ pub(crate) fn run( budget: &mut ImageBudget, ) -> Result<(), OptimizeError> { for p in pages { + if !progress(Stage::Reading { done: 0, total: pages.len() }) { + return Err(OptimizeError::Cancelled); + } report.fragments_merged += merge_page(doc, *p, report, progress, budget)?; } Ok(()) @@ -405,6 +416,35 @@ mod tests { assert_eq!(budget.refused(), 1); } + /// A page drawing too few things for a run isn't parsed for one, even with a picture that could + /// be a fragment. + #[test] + fn too_few_draws_for_a_run_read_no_groups() { + let mut doc = Document::new_empty(); + let mut image = Dict::new(); + for (k, v) in + [("Subtype", Object::name("Image")), ("Width", Object::Int(1)), ("Height", Object::Int(1)), ("BitsPerComponent", Object::Int(8))] + { + image.set(k.as_bytes().to_vec(), v); + } + image.set(b"ColorSpace".to_vec(), Object::name("DeviceGray")); + let image = doc.add(Object::Stream(Stream::from_raw(image, vec![0]))); + let drawn = "q 10 0 0 10 0 0 cm /Im Do Q\n".repeat(MIN_FRAGMENTS - 1); + let content = doc.add(Object::Stream(Stream::flate(Dict::new(), drawn.as_bytes()))); + let mut xobjects = Dict::new(); + xobjects.set(b"Im".to_vec(), Object::Ref(image)); + let mut resources = Dict::new(); + resources.set(b"XObject".to_vec(), Object::Dict(xobjects)); + let mut page = Dict::new(); + page.set(b"Contents".to_vec(), Object::Ref(content)); + page.set(b"Resources".to_vec(), Object::Dict(resources)); + let page = doc.add(Object::Dict(page)); + GROUP_READS.with(|c| c.set(0)); + let merged = merge_page(&mut doc, page, &mut Report::default(), &mut |_| true, &mut ImageBudget::with_work(u64::MAX)); + assert_eq!(merged.ok(), Some(0)); + assert_eq!(GROUP_READS.with(|c| c.get()), 0); + } + /// Many images drawn on top of each other are not one picture; looking for runs among them reads /// a bounded number of groups per operator, not the rest of the page from every group. #[test] diff --git a/crates/optimize/src/images.rs b/crates/optimize/src/images.rs index 264e1a2b9b..10388b8107 100644 --- a/crates/optimize/src/images.rs +++ b/crates/optimize/src/images.rs @@ -4,6 +4,8 @@ use std::collections::{HashMap, HashSet}; use std::sync::Arc; use pdfcraft_content::{Matrix, Pieces}; +use std::sync::atomic::{AtomicU64, Ordering}; + use pdfcraft_cos::{DecodeBudget, Dict, Document, ImageBudget, ObjRef, Object, Stream}; use crate::{Compression, ImageSettings, MonoCompression, MonoSettings, OptimizeError, Report, Settings, Stage}; @@ -250,13 +252,20 @@ fn remaps_device_spaces(doc: &Document, resources: &Dict) -> bool { /// decode within its budget or is damaged partway, every image is given `INFINITY` (never /// resampled), since its largest use may be among those missed. pub fn effective_resolutions(doc: &Document, pages: &[ObjRef]) -> HashMap { - walk_pages(doc, pages).0 + let Ok((ppi, ..)) = walk_pages(doc, pages, &mut |_| true) else { return HashMap::new() }; + ppi } +/// What a walk of the pages found: each image's resolution, the images under remapped device +/// spaces, and the references followed. +type PagesWalked = (HashMap, HashSet, Read); + /// [`effective_resolutions`], the images drawn or listed where `/DefaultGray` or `/DefaultRGB` is /// in force, by object number (all of them when a use may have been missed), and the references /// followed to reach them. -fn walk_pages(doc: &Document, pages: &[ObjRef]) -> (HashMap, HashSet, Read) { +/// +/// `tick` hears how many pages are done, and stops the walk by returning false. +fn walk_pages(doc: &Document, pages: &[ObjRef], tick: &mut dyn FnMut(usize) -> bool) -> Result { let mut walk = Walk { doc, path: Vec::new(), @@ -269,7 +278,10 @@ fn walk_pages(doc: &Document, pages: &[ObjRef]) -> (HashMap, HashSe defaults: HashSet::new(), read: Read::default(), }; - for p in pages { + for (done, p) in pages.iter().enumerate() { + if !tick(done) { + return Err(OptimizeError::Cancelled); + } let Some(dict) = doc.get(*p).as_dict().cloned() else { continue }; let page = pdfcraft_model::Page { obj: *p, dict }; walk.resources_above(*p); @@ -288,7 +300,7 @@ fn walk_pages(doc: &Document, pages: &[ObjRef]) -> (HashMap, HashSe walk.out.values_mut().for_each(|v| *v = f64::INFINITY); walk.defaults.extend(walk.out.keys().map(|r| r.num)); } - (walk.out, walk.defaults, walk.read) + Ok((walk.out, walk.defaults, walk.read)) } /// Whether image `r` is one [`one_bit`] may make grey: eight-bit `DeviceRGB`, not in an image codec. @@ -1330,7 +1342,7 @@ pub(crate) fn run( progress: &mut dyn FnMut(Stage) -> bool, budget: &mut ImageBudget, ) -> Result<(), OptimizeError> { - let (ppi, mut defaults, read) = walk_pages(doc, pages); + let (ppi, mut defaults, read) = walk_pages(doc, pages, &mut |done| progress(Stage::Reading { done, total: pages.len() }))?; // Black-and-white RGB becomes grey, which default colour spaces would draw otherwise, and an // indexed image's palette says which of its samples a resampled one keeps, which they may // change: one used where the walk didn't look may be drawn so. @@ -1347,93 +1359,73 @@ pub(crate) fn run( }) .collect(); let mut refs: Vec<(ObjRef, f64)> = ppi.into_iter().filter(|(r, _)| !masks.contains(r)).collect(); - refs.sort_by_key(|(r, _)| r.num); + refs.sort_by_key(|(r, _)| (r.num, r.generation)); let total = refs.len(); let mut jbig2 = false; let mut group: Vec = Vec::new(); - // Holds a JBIG2 image for `share`, which codes the group once it is full. - let mut hold = |doc: &mut Document, image: ObjRef, size: (Option, Option), bits: Option>, report: &mut Report| { - let side = |v: Option| v.and_then(|v| u32::try_from(v).ok()); - let (Some(bits), Some(width), Some(height)) = (bits, side(size.0), side(size.1)) else { return }; - hold_in(&mut group, Held { image, bits, width, height }, SHARED_BYTES, |group| share(doc, group, report)); + // By number: a repaired object answers to two generations. + let mut smask_uses: HashMap = HashMap::new(); + for (r, _) in &refs { + if let Some(m) = smask_of(doc, *r) { + *smask_uses.entry(m.num).or_default() += 1; + } + } + // Too few to be worth the threads, which keeps a small file's progress in step too. + let threads = if total < MIN_FOR_THREADS { + 1 + } else { + let largest = refs.iter().filter_map(|(r, _)| picture_pixels(doc, *r)).max().unwrap_or(0); + threads_for(crate::pool::workers(settings.workers), largest) }; - for (done, (r, ppi)) in refs.into_iter().enumerate() { - if !progress(Stage::Images { done, total }) { - return Err(OptimizeError::Cancelled); - } - let obj = doc.get(r); - let Object::Stream(s) = &*obj else { continue }; - report.images += 1; - let refused = budget.refused(); - // Packed from more bits a sample, it is lossless at one bit, but the grey or colour - // settings can make it smaller still (below). - let packed_from_more = s.dict.int(b"BitsPerComponent").is_some_and(|b| b > 1); - let (packed, mono) = match mono(doc, s, ppi, defaults.contains(&r.num), &settings.mono, budget) { - Some(m) if packed_from_more => (Some(m), None), - m => (None, m), - }; - if let Some((new, resampled, bits)) = mono { - report.image_bytes_before += s.raw.len(); - report.image_bytes_after += new.raw.len(); - if resampled { - report.images_resampled += 1; - } else { - report.images_recompressed += 1; + let mut done = 0; + while let Some(rest) = refs.get(done..).filter(|rest| !rest.is_empty()) { + let run = if threads > 1 { batch_len(doc, rest, &smask_uses, settings.mono.compression == MonoCompression::Jbig2) } else { 0 }; + let Some(batch) = rest.get(..run).filter(|batch| batch.len() > 1) else { + if !progress(Stage::Images { done, total }) { + return Err(OptimizeError::Cancelled); } - jbig2 |= new.dict.name(b"Filter") == Some(b"JBIG2Decode"); - let size = (new.dict.int(b"Width"), new.dict.int(b"Height")); - let bits = bits.filter(|_| shares(&new.dict)); - doc.set(r, Object::Stream(new)); - hold(doc, r, size, bits, report); - continue; - } - let n = s.dict.get(b"ColorSpace").and_then(|cs| components(doc, cs)); - let processed = n.and_then(|n| process(doc, s, ppi, if n == 1 { &settings.gray } else { &settings.color }, n, budget)); - // A dithered scan, say, downsampled and JPEG-compressed: the smaller wins, counting a soft - // mask resampled with it. - let smask_len = |smask: Option<&(ObjRef, Stream)>| match smask { - Some((_, m)) => Some(m.raw.len()), - None => s.dict.reference(b"SMask").map(|m| match &*doc.get(m) { - Object::Stream(m) => m.raw.len(), - _ => 0, - }), - }; - let total = |new: &Stream, smask: Option<&(ObjRef, Stream)>| Some(crate::stored_len(new)?.saturating_add(smask_len(smask).unwrap_or(0))); - let packed_wins = packed.as_ref().is_some_and(|(p, _, _)| { - processed.as_ref().is_none_or(|(new, smask)| total(p, None).zip(total(new, smask.as_ref())).is_some_and(|(a, b)| a <= b)) - }); - if let Some((new, resampled, bits)) = packed.filter(|_| packed_wins) { - jbig2 |= new.dict.name(b"Filter") == Some(b"JBIG2Decode"); - report.image_bytes_before += s.raw.len(); - report.image_bytes_after += new.raw.len(); - if resampled { - report.images_resampled += 1; - } else { - report.images_recompressed += 1; + if let Some((r, ppi)) = rest.first() { + let decision = decide(doc, *r, *ppi, defaults.contains(&r.num), settings, budget); + apply(doc, decision, report, &mut jbig2, &mut group); } - let size = (new.dict.int(b"Width"), new.dict.int(b"Height")); - let bits = bits.filter(|_| shares(&new.dict)); - doc.set(r, Object::Stream(new)); - hold(doc, r, size, bits, report); - continue; - } - let Some((new, smask)) = processed else { - // Read to be packed and again to be resampled, it is still one image, and it stayed. - report.images_refused += usize::from(budget.refused() > refused); + done += 1; continue; }; - let resized = new.dict.int(b"Width") != s.dict.int(b"Width") || new.dict.int(b"Height") != s.dict.int(b"Height"); - report.image_bytes_before += s.raw.len(); - report.image_bytes_after += new.raw.len(); - if resized { - report.images_resampled += 1; - } else { - report.images_recompressed += 1; - } - doc.set(r, Object::Stream(new)); - if let Some((mr, m)) = smask { - doc.set(mr, Object::Stream(m)); + // Each thread works within what's left now, and pays for what it used in the document's + // order below: a picture the order can't pay for is done again as if there were no threads, + // so the same pictures are refused with any number of them. Once more than what's left has + // been used, no more are started. + let allowed = budget.work_left(); + let used = AtomicU64::new(0); + let tried = { + let (doc, defaults) = (&*doc, &defaults); + crate::pool::map( + batch, + threads, + |(r, ppi)| { + if used.load(Ordering::Relaxed) >= allowed { + return None; + } + let mut own = ImageBudget::with_work(allowed); + let decision = decide(doc, *r, *ppi, defaults.contains(&r.num), settings, &mut own); + let cost = allowed.saturating_sub(own.work_left()); + used.fetch_add(cost, Ordering::Relaxed); + (own.refused() == 0).then_some((decision, cost)) + }, + &mut |finished| progress(Stage::Images { done: done.saturating_add(finished), total }), + )? + }; + for ((r, ppi), tried) in batch.iter().zip(tried) { + let decision = match tried.flatten() { + Some((decision, cost)) if cost <= budget.work_left() => { + budget.charge(cost); + decision + } + _ => decide(doc, *r, *ppi, defaults.contains(&r.num), settings, budget), + }; + apply(doc, decision, report, &mut jbig2, &mut group); } + done += batch.len(); } share(doc, &mut group, report); if !progress(Stage::Images { done: total, total }) { @@ -1445,6 +1437,166 @@ pub(crate) fn run( Ok(()) } +/// A picture's soft mask, if it names one. +fn smask_of(doc: &Document, r: ObjRef) -> Option { + match &*doc.get(r) { + Object::Stream(s) => s.dict.reference(b"SMask"), + _ => None, + } +} + +/// How many of `rest`, from the first, can be worked out beside each other. +fn batch_len(doc: &Document, rest: &[(ObjRef, f64)], smask_uses: &HashMap, holds_samples: bool) -> usize { + // One object under two generations: the second must see what the first left. Under one of them + // it can be a picture, and under the other another's soft mask. + let mut numbers = HashSet::new(); + let mut samples = 0usize; + rest.iter() + .take(PICTURES_AT_ONCE) + .take_while(|(r, _)| { + // Results wait for the end of the batch, samples kept to share a dictionary with them. + samples = samples.saturating_add(if holds_samples { held_samples(doc, *r) } else { 0 }); + // A soft mask two pictures share is changed by the first of them, and the second must + // see that, so they don't run beside others. + samples <= SHARED_BYTES + && !smask_uses.contains_key(&r.num) + && numbers.insert(r.num) + && smask_of(doc, *r).is_none_or(|m| smask_uses.get(&m.num) == Some(&1)) + }) + .count() +} + +/// The most bytes of 1-bit samples picture `r` could keep to share a dictionary: whole bytes a row. +fn held_samples(doc: &Document, r: ObjRef) -> usize { + let object = doc.get(r); + let Object::Stream(s) = &*object else { return 0 }; + let side = |key: &[u8]| s.dict.int(key).and_then(|v| u64::try_from(v).ok()); + let Some((w, h)) = side(b"Width").zip(side(b"Height")) else { return 0 }; + if w.saturating_mul(h) > SHARED_PIXELS { + return 0; + } + usize::try_from(w.div_ceil(8).saturating_mul(h)).unwrap_or(usize::MAX) +} + +/// Below this many pictures, one thread does them. +const MIN_FOR_THREADS: usize = 16; + +/// How many pictures run beside each other before what they used is paid for. +const PICTURES_AT_ONCE: usize = 32; + +/// What the pictures being worked out may hold in memory together. +const PICTURE_MEMORY: u64 = 1 << 30; + +/// What a picture holds while it is worked out, for each pixel: decoded, and resampled or encoded. +const PICTURE_BYTES_PER_PIXEL: u64 = 8; + +/// A picture's pixels, if it is small enough to be decoded at all. +fn picture_pixels(doc: &Document, r: ObjRef) -> Option { + let object = doc.get(r); + let Object::Stream(s) = &*object else { return None }; + let (w, h) = (u64::try_from(s.dict.int(b"Width")?).ok()?, u64::try_from(s.dict.int(b"Height")?).ok()?); + w.checked_mul(h).filter(|pixels| *pixels <= MAX_PIXELS) +} + +/// How many pictures of `largest` pixels to work on at once: `workers`, or as many as +/// [`PICTURE_MEMORY`] holds, and at least one. +fn threads_for(workers: usize, largest: u64) -> usize { + let fit = PICTURE_MEMORY / largest.saturating_mul(PICTURE_BYTES_PER_PIXEL).max(1); + workers.min(usize::try_from(fit).unwrap_or(usize::MAX)).max(1) +} + +/// What's to become of one picture. +enum Decision { + /// Not an image stream after all. + NotAnImage, + /// An image that stays as it is, `refused` if the allowance couldn't pay for it. + Stays { refused: bool }, + Replace { + image: ObjRef, + before: usize, + new: Stream, + resampled: bool, + smask: Option<(ObjRef, Stream)>, + /// Its 1-bit samples, if it may share a JBIG2 dictionary ([`share`]). + bits: Option>, + }, +} + +/// Do what was decided for one picture, and count it. +fn apply(doc: &mut Document, decision: Decision, report: &mut Report, jbig2: &mut bool, group: &mut Vec) { + let (image, before, new, resampled, smask, bits) = match decision { + Decision::NotAnImage => return, + Decision::Stays { refused } => { + report.images += 1; + report.images_refused += usize::from(refused); + return; + } + Decision::Replace { image, before, new, resampled, smask, bits } => (image, before, new, resampled, smask, bits), + }; + report.images += 1; + *jbig2 |= new.dict.name(b"Filter") == Some(b"JBIG2Decode"); + report.image_bytes_before += before; + report.image_bytes_after += new.raw.len(); + if resampled { + report.images_resampled += 1; + } else { + report.images_recompressed += 1; + } + let side = |key: &[u8]| new.dict.int(key).and_then(|v| u32::try_from(v).ok()); + let held = bits.zip(side(b"Width")).zip(side(b"Height")); + doc.set(image, Object::Stream(new)); + if let Some((mr, m)) = smask { + doc.set(mr, Object::Stream(m)); + } + // A JBIG2 image waits for `share`, which codes the group once it is full. + if let Some(((bits, width), height)) = held { + hold_in(group, Held { image, bits, width, height }, SHARED_BYTES, |group| share(doc, group, report)); + } +} + +/// Work out what's to become of picture `r`, changing nothing yet. +fn decide(doc: &Document, r: ObjRef, ppi: f64, defaults: bool, settings: &Settings, budget: &mut ImageBudget) -> Decision { + let obj = doc.get(r); + let Object::Stream(s) = &*obj else { return Decision::NotAnImage }; + let refused = budget.refused(); + // Packed from more bits a sample, it is lossless at one bit, but the grey or colour + // settings can make it smaller still (below). + let packed_from_more = s.dict.int(b"BitsPerComponent").is_some_and(|b| b > 1); + let (packed, mono) = match mono(doc, s, ppi, defaults, &settings.mono, budget) { + Some(m) if packed_from_more => (Some(m), None), + m => (None, m), + }; + if let Some((new, resampled, bits)) = mono { + let bits = bits.filter(|_| shares(&new.dict)); + return Decision::Replace { image: r, before: s.raw.len(), new, resampled, smask: None, bits }; + } + let n = s.dict.get(b"ColorSpace").and_then(|cs| components(doc, cs)); + let processed = n.and_then(|n| process(doc, s, ppi, if n == 1 { &settings.gray } else { &settings.color }, n, budget)); + // A dithered scan, say, downsampled and JPEG-compressed: the smaller wins, counting a soft + // mask resampled with it. + let smask_len = |smask: Option<&(ObjRef, Stream)>| match smask { + Some((_, m)) => Some(m.raw.len()), + None => s.dict.reference(b"SMask").map(|m| match &*doc.get(m) { + Object::Stream(m) => m.raw.len(), + _ => 0, + }), + }; + let total = |new: &Stream, smask: Option<&(ObjRef, Stream)>| Some(crate::stored_len(new)?.saturating_add(smask_len(smask).unwrap_or(0))); + let packed_wins = packed.as_ref().is_some_and(|(p, _, _)| { + processed.as_ref().is_none_or(|(new, smask)| total(p, None).zip(total(new, smask.as_ref())).is_some_and(|(a, b)| a <= b)) + }); + if let Some((new, resampled, bits)) = packed.filter(|_| packed_wins) { + let bits = bits.filter(|_| shares(&new.dict)); + return Decision::Replace { image: r, before: s.raw.len(), new, resampled, smask: None, bits }; + } + let Some((new, smask)) = processed else { + // Read to be packed and again to be resampled, it is still one image, and it stayed. + return Decision::Stays { refused: budget.refused() > refused }; + }; + let resampled = new.dict.int(b"Width") != s.dict.int(b"Width") || new.dict.int(b"Height") != s.dict.int(b"Height"); + Decision::Replace { image: r, before: s.raw.len(), new, resampled, smask, bits: None } +} + /// JBIG2Decode is PDF 1.4: a document whose header says less, and whose catalog doesn't say 1.4 /// or more already (directly or by reference), gets `/Version /1.4` in its catalog (ISO 32000-2 /// §7.7.2). A full save with object streams writes a 1.5 header anyway. @@ -1492,6 +1644,37 @@ mod dimension_tests { assert_eq!((count, group.len()), (SHARED_IMAGES, 0), "coded on the 16th"); } + #[test] + fn a_batch_holds_no_more_samples_than_a_group_may() { + let mut doc = Document::new_empty(); + let mut add = |width: i64, height: i64| { + let mut d = Dict::new(); + d.set(b"Width".to_vec(), Object::Int(width)); + d.set(b"Height".to_vec(), Object::Int(height)); + (doc.add(Object::Stream(Stream::from_raw(d, Vec::new()))), 72.0) + }; + // Eight megabytes of samples each, and one byte a pixel when one pixel wide. + let large: Vec<_> = (0..12).map(|_| add(8192, 8192)).collect(); + let thin: Vec<_> = (0..8).map(|_| add(1, 32 << 20)).collect(); + let small: Vec<_> = (0..40).map(|_| add(100, 100)).collect(); + // Past what may share a dictionary, so never held. + let huge: Vec<_> = (0..12).map(|_| add(16384, 16384)).collect(); + let none = HashMap::new(); + assert_eq!(batch_len(&doc, &large, &none, true), 8); + assert_eq!(batch_len(&doc, &thin, &none, true), 2); + assert_eq!(batch_len(&doc, &small, &none, true), PICTURES_AT_ONCE); + assert_eq!(batch_len(&doc, &huge, &none, true), huge.len()); + assert_eq!(batch_len(&doc, &large, &none, false), large.len(), "nothing is held but for JBIG2"); + } + + #[test] + fn threads_are_limited_by_the_memory_of_the_largest_picture() { + assert_eq!(threads_for(8, 1_000_000), 8); + assert_eq!(threads_for(8, 40_000_000), 3); + assert_eq!(threads_for(8, MAX_PIXELS), 1); + assert_eq!(threads_for(8, 0), 8); + } + /// Only what redaction can store again without the dictionary shares one: an image mask, or /// DeviceGray named directly, of at most 2^26 pixels. #[test] @@ -1609,6 +1792,63 @@ mod dimension_tests { assert_eq!(changed, 2, "the third is as it was"); } + /// An allowance that runs out refuses the same pictures with any number of threads as it does + /// one after another. + #[test] + fn the_allowance_refuses_the_same_pictures_with_any_number_of_threads() { + let picture = |shade: u32| { + let mut dict = Dict::new(); + dict.set(b"Type".to_vec(), Object::name("XObject")); + dict.set(b"Subtype".to_vec(), Object::name("Image")); + dict.set(b"Width".to_vec(), Object::Int(400)); + dict.set(b"Height".to_vec(), Object::Int(400)); + dict.set(b"BitsPerComponent".to_vec(), Object::Int(8)); + dict.set(b"ColorSpace".to_vec(), Object::name("DeviceGray")); + let samples: Vec = (0..400u32 * 400).map(|i| (i.wrapping_mul(shade + 3) % 251) as u8).collect(); + Stream::flate(dict, &samples) + }; + let build = || { + let mut doc = Document::new_empty(); + // A few of different sizes, so what's left when the allowance ends depends on the order. + let refs: Vec = (0..40u32).map(|i| doc.add(Object::Stream(picture(i)))).collect(); + let mut xobjects = Dict::new(); + let mut content = String::from("q 72 0 0 72 0 0 cm "); + for (i, r) in refs.iter().enumerate() { + xobjects.set(format!("I{i}").into_bytes(), Object::Ref(*r)); + content.push_str(&format!("/I{i} Do ")); + } + content.push('Q'); + let mut resources = Dict::new(); + resources.set(b"XObject".to_vec(), Object::Dict(xobjects)); + let content = doc.add(Object::Stream(Stream::flate(Dict::new(), content.as_bytes()))); + let mut page = Dict::new(); + page.set(b"Type".to_vec(), Object::name("Page")); + page.set(b"Resources".to_vec(), Object::Dict(resources)); + page.set(b"Contents".to_vec(), Object::Ref(content)); + let page = doc.add(Object::Dict(page)); + (doc, refs, page) + }; + let run_with = |work: u64, workers: usize| { + let (mut doc, refs, page) = build(); + let settings = Settings { workers: Some(workers), ..Settings::default() }; + let (mut report, mut budget) = (Report::default(), ImageBudget::with_work(work)); + run(&mut doc, &[page], &settings, &mut report, &mut |_| true, &mut budget).unwrap(); + let raws: Vec> = refs.iter().map(|r| if let Object::Stream(s) = &*doc.get(*r) { s.raw.to_vec() } else { Vec::new() }).collect(); + (format!("{report:?}"), budget.work_left(), raws) + }; + let all = u64::MAX - run_with(u64::MAX, 1).1; + // Past one batch of what threads run together, and part of the way through a picture. + let work = all / 40 * 37 + all / 80; + let one = run_with(work, 1); + assert!(one.0.contains("images_refused: 3"), "{}", one.0); + for workers in [2, 4, 8] { + let many = run_with(work, workers); + assert_eq!(many.0, one.0, "{workers} threads count the same"); + assert_eq!(many.1, one.1, "{workers} threads leave the same allowance"); + assert!(many.2 == one.2, "{workers} threads store the same pictures"); + } + } + /// Images that aren't JBIG2 share the run's allowance too, both on the way to being resampled /// and on the way to being packed to a bit a sample: a file of filter bombs would hold a run /// as long as one of JBIG2, and those past it stay as they are. diff --git a/crates/optimize/src/lib.rs b/crates/optimize/src/lib.rs index ef6423ab7d..275c6f32a6 100644 --- a/crates/optimize/src/lib.rs +++ b/crates/optimize/src/lib.rs @@ -32,6 +32,7 @@ mod audit; pub mod fragments; mod images; mod links; +mod pool; mod samples; mod unused; @@ -41,6 +42,7 @@ use pdfcraft_cos::{Dict, Document, ImageBudget, MAX_PAGE_DECODED, ObjRef, Object use pdfcraft_filters::{Filter, Params}; pub use images::effective_resolutions; +pub use pool::give_way_to_other_programs; pub use samples::{NotDecoded, Samples, decode_samples, encode_samples}; #[derive(Debug, thiserror::Error)] @@ -106,6 +108,8 @@ impl Compatibility { /// Where an optimization is, reported to [`optimize_with_progress`]'s callback. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum Stage { + /// Reading what pages draw: `done` of `total` read, none yet while looking for fragments. + Reading { done: usize, total: usize }, /// About to process image `done` (0-based) of `total`; `done == total` once all are done. Images { done: usize, total: usize }, /// Merging the fragments of a page picture into one image (only when a page has some). @@ -181,6 +185,8 @@ impl ImageSettings { pub struct Settings { /// Target PDF version (`Retain` by default). pub compatibility: Compatibility, + /// Threads for pictures: `None` picks from the machine. The result is the same with any number. + pub workers: Option, /// Put page pictures drawn as many stacked fragments back together first ([`fragments`]). pub merge_fragments: bool, pub color: ImageSettings, @@ -217,6 +223,7 @@ impl Default for Settings { /// Acrobat's Reduce File Size choices. fn default() -> Self { Self { + workers: None, compatibility: Compatibility::Retain, merge_fragments: true, color: ImageSettings::REDUCE, diff --git a/crates/optimize/src/pool.rs b/crates/optimize/src/pool.rs new file mode 100644 index 0000000000..57dcc7fdb1 --- /dev/null +++ b/crates/optimize/src/pool.rs @@ -0,0 +1,262 @@ +//! Work over many pages or images on several threads, the results in order, so that what is +//! written is the same however many threads did it. + +use std::panic::{AssertUnwindSafe, catch_unwind}; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::sync::mpsc::{RecvTimeoutError, channel}; +use std::time::Duration; + +use crate::OptimizeError; + +/// Threads beyond this add more memory than speed: each holds a page or an image being worked on. +const MAX_WORKERS: usize = 8; +/// How long Cancel can wait while every thread is busy with something long. +const CANCEL_CHECK: Duration = Duration::from_millis(100); + +/// Let other programs go first on this thread: the wait for a copy matters less than whatever else +/// is open. A system that refuses leaves it as it was. +pub fn give_way_to_other_programs() { + #[cfg(not(target_arch = "wasm32"))] + let _ = thread_priority::set_current_thread_priority(thread_priority::ThreadPriority::Min); +} + +/// How many threads to use: `asked`, or half the machine's processors up to [`MAX_WORKERS`], so a +/// bigger machine goes faster and every machine keeps room for other programs. One on the web, which +/// has no threads here. +pub(crate) fn workers(asked: Option) -> usize { + if cfg!(target_arch = "wasm32") { + return 1; + } + asked.unwrap_or_else(|| std::thread::available_parallelism().map_or(1, |n| (n.get() / 2).clamp(1, MAX_WORKERS))).max(1) +} + +/// `work` on each of `items`, results in the order of `items`. `tick` hears how many are done, and +/// stops the run by returning false. With more than one worker an item whose work panics gets +/// `None`, and the caller decides what becomes of it; with one, the panic goes on up as it does for +/// any other work. +pub(crate) fn map( + items: &[T], + workers: usize, + work: impl Fn(&T) -> R + Sync, + tick: &mut dyn FnMut(usize) -> bool, +) -> Result>, OptimizeError> { + let mut out: Vec> = Vec::with_capacity(items.len()); + if workers <= 1 || items.len() <= 1 { + for (done, item) in items.iter().enumerate() { + if !tick(done) { + return Err(OptimizeError::Cancelled); + } + out.push(Some(work(item))); + } + return Ok(out); + } + out.resize_with(items.len(), || None); + let (next, stop) = (AtomicUsize::new(0), AtomicBool::new(false)); + let mut stopped = false; + std::thread::scope(|scope| { + let (results, finished) = channel(); + for _ in 0..workers.min(items.len()) { + let results = results.clone(); + let (next, stop, work) = (&next, &stop, &work); + // A thread that can't start only leaves more for the others, or for below. + let _ = std::thread::Builder::new().name("pdfcraft-optimize-worker".into()).spawn_scoped(scope, move || { + give_way_to_other_programs(); + while !stop.load(Ordering::Relaxed) { + let i = next.fetch_add(1, Ordering::Relaxed); + let Some(item) = items.get(i) else { break }; + let result = catch_unwind(AssertUnwindSafe(|| work(item))).ok(); + if results.send((i, result)).is_err() { + break; + } + } + }); + } + drop(results); + let mut done = 0; + loop { + if !stopped && !tick(done) { + stopped = true; + stop.store(true, Ordering::Relaxed); + } + match finished.recv_timeout(CANCEL_CHECK) { + Ok((i, result)) => { + if let Some(slot) = out.get_mut(i) { + *slot = result; + } + done += 1; + } + Err(RecvTimeoutError::Timeout) => {} + Err(RecvTimeoutError::Disconnected) => break, + } + } + }); + if stopped { + return Err(OptimizeError::Cancelled); + } + // No thread could start, or the last of them stopped early: the rest are done here. + let claimed = next.load(Ordering::Relaxed).min(items.len()); + for (done, (item, slot)) in items.iter().zip(out.iter_mut()).enumerate().skip(claimed) { + if !tick(done) { + return Err(OptimizeError::Cancelled); + } + *slot = Some(work(item)); + } + Ok(out) +} + +#[cfg(test)] +mod tests { + use std::sync::Mutex; + + use super::*; + + #[test] + fn results_keep_their_order_whatever_finishes_first() { + let items: Vec = (0..200).collect(); + let out = map( + &items, + 4, + |i| { + // The early ones finish last. + std::thread::sleep(Duration::from_micros(200u64.saturating_sub(*i))); + i * 2 + }, + &mut |_| true, + ) + .unwrap(); + assert_eq!(out, items.iter().map(|i| Some(i * 2)).collect::>()); + } + + #[test] + fn one_worker_does_them_in_order_on_this_thread() { + let here = std::thread::current().id(); + let order = Mutex::new(Vec::new()); + let mut ticks = Vec::new(); + map( + &[1, 2, 3], + 1, + |i| { + assert_eq!(std::thread::current().id(), here); + order.lock().unwrap().push(*i); + }, + &mut |done| { + ticks.push(done); + true + }, + ) + .unwrap(); + assert_eq!(*order.lock().unwrap(), [1, 2, 3]); + assert_eq!(ticks, [0, 1, 2]); + } + + #[test] + fn several_workers_share_the_work() { + let threads = Mutex::new(std::collections::HashSet::new()); + map( + &[0; 64], + 4, + |_| { + threads.lock().unwrap().insert(std::thread::current().id()); + std::thread::sleep(Duration::from_millis(5)); + }, + &mut |_| true, + ) + .unwrap(); + assert!(threads.lock().unwrap().len() > 1, "it really ran on more than one"); + } + + #[test] + fn stopping_stops_and_says_so() { + let ran = AtomicUsize::new(0); + let r = map( + &[0; 500], + 4, + |_| { + ran.fetch_add(1, Ordering::Relaxed); + std::thread::sleep(Duration::from_millis(1)); + }, + &mut |done| done < 8, + ); + assert!(matches!(r, Err(OptimizeError::Cancelled))); + assert!(ran.load(Ordering::Relaxed) < 500, "the rest never started"); + } + + /// Cancel is heard while every thread is busy with something long, not only between items. + #[test] + fn a_stop_gets_through_while_every_thread_is_busy() { + let mut asked_while_busy = 0; + map(&[0; 2], 2, |_| std::thread::sleep(Duration::from_millis(600)), &mut |done| { + if done == 0 { + asked_while_busy += 1; + } + true + }) + .unwrap(); + assert!(asked_while_busy >= 3, "asked {asked_while_busy} times before anything finished"); + + let started = std::time::Instant::now(); + let mut asked = 0; + let r = map(&[0; 4], 2, |_| std::thread::sleep(Duration::from_millis(400)), &mut |_| { + asked += 1; + asked < 3 + }); + assert!(matches!(r, Err(OptimizeError::Cancelled))); + assert!(started.elapsed() < Duration::from_millis(1500), "the items not started never ran"); + } + + #[test] + fn an_item_that_panics_is_left_alone_and_the_rest_are_done() { + let items: Vec = (0..32).collect(); + let out = map( + &items, + 4, + |i| { + assert_ne!(*i, 7, "this one panics"); + *i + }, + &mut |_| true, + ) + .unwrap(); + for (i, got) in out.iter().enumerate() { + assert_eq!(*got, (i != 7).then_some(i as u32), "item {i}"); + } + } + + /// The workers give way, and the thread that asked keeps its own priority. + #[cfg(windows)] + #[test] + fn workers_give_way_to_other_programs() { + use thread_priority::get_current_thread_priority; + let here = get_current_thread_priority().unwrap(); + let theirs = Mutex::new(Vec::new()); + map( + &[0; 8], + 2, + |_| { + theirs.lock().unwrap().push(get_current_thread_priority().unwrap()); + std::thread::sleep(Duration::from_millis(2)); + }, + &mut |_| true, + ) + .unwrap(); + assert_eq!(get_current_thread_priority().unwrap(), here); + let theirs = theirs.into_inner().unwrap(); + assert_eq!(theirs.len(), 8); + assert!(theirs.iter().all(|p| *p != here), "{theirs:?} against {here:?}"); + } + + #[test] + fn nothing_to_do_is_fine() { + let none: [u8; 0] = []; + assert!(map(&none, 4, |_| 1, &mut |_| true).unwrap().is_empty()); + assert_eq!(map(&[5], 4, |i| *i, &mut |_| true).unwrap(), [Some(5)]); + } + + #[test] + fn the_web_and_small_machines_get_a_sensible_count() { + assert_eq!(workers(Some(3)), if cfg!(target_arch = "wasm32") { 1 } else { 3 }); + assert_eq!(workers(Some(0)), 1, "never none"); + let picked = workers(None); + assert!((1..=MAX_WORKERS).contains(&picked), "{picked}"); + } +} diff --git a/crates/optimize/tests/jpeg_color_transform.rs b/crates/optimize/tests/jpeg_color_transform.rs index c4dfabc9d9..4e08185d77 100644 --- a/crates/optimize/tests/jpeg_color_transform.rs +++ b/crates/optimize/tests/jpeg_color_transform.rs @@ -395,9 +395,9 @@ fn a_jpeg_header_that_takes_more_than_a_few_steps_to_read_is_left_as_it_is() { #[test] fn cancelling_works_from_the_first_call_with_a_jpeg_drawn_many_times() { // One JPEG with 8 MB of stray bytes after SOI, drawn 64 times at the same place (the case - // reported on #3012). Fragment merging reads its header for each group it tries, before the - // first progress call. The header scan stops at the first stray byte, so that call is the - // first image, and cancelling there stops Optimize. + // reported on #3012). Fragment merging reads its header for each group it tries. The header + // scan stops at the first stray byte, so the first image comes soon, and cancelling there stops + // Optimize. Cancelling while pages are read stops it sooner still. let (w, h) = (128, 8); let jpeg = after_soi(&rgb_jpeg(&photo(w, h), w, h), &vec![0; 8_000_000]); let mut doc = Document::new_empty(); @@ -411,8 +411,16 @@ fn cancelling_works_from_the_first_call_with_a_jpeg_drawn_many_times() { let mut stages = Vec::new(); let result = optimize_with_progress(&mut doc, &Settings::default(), &mut |stage| { stages.push(stage); + !matches!(stage, Stage::Images { .. }) + }); + assert!(matches!(result, Err(OptimizeError::Cancelled)), "{result:?}"); + assert_eq!(stages.last(), Some(&Stage::Images { done: 0, total: 1 })); + assert_eq!(stages.first(), Some(&Stage::Reading { done: 0, total: 1 }), "it can be stopped from the first call"); + let mut stages = 0; + let result = optimize_with_progress(&mut doc, &Settings::default(), &mut |_| { + stages += 1; false }); assert!(matches!(result, Err(OptimizeError::Cancelled)), "{result:?}"); - assert_eq!(stages, [Stage::Images { done: 0, total: 1 }]); + assert_eq!(stages, 1); } diff --git a/crates/optimize/tests/optimize.rs b/crates/optimize/tests/optimize.rs index 2785ddac6c..c4fb865198 100644 --- a/crates/optimize/tests/optimize.rs +++ b/crates/optimize/tests/optimize.rs @@ -2647,7 +2647,9 @@ fn progress_is_reported_per_image_and_a_refusal_cancels() { }; let mut seen = Vec::new(); let report = optimize_with_progress(&mut build(), &Settings::default(), &mut |s| { - seen.push(s); + if !matches!(s, Stage::Reading { .. }) { + seen.push(s); + } true }) .unwrap(); @@ -2665,7 +2667,10 @@ fn progress_is_reported_per_image_and_a_refusal_cancels() { // Stopping after the first image. let mut calls = 0; - let r = optimize_with_progress(&mut build(), &Settings::default(), &mut |_| { + let r = optimize_with_progress(&mut build(), &Settings::default(), &mut |s| { + if matches!(s, Stage::Reading { .. }) { + return true; + } calls += 1; calls < 2 }); @@ -4289,3 +4294,135 @@ fn a_picture_whose_mask_is_not_a_stencil_is_left_alone() { assert_eq!(stream(&doc, img), before, "{what}"); } } + +/// Pictures run on several threads: what comes out doesn't depend on how many. +#[test] +fn the_same_file_comes_out_with_any_number_of_threads() { + let build = || { + let mut doc = Document::new_empty(); + let shared_mask = image(&mut doc, 1200, 1200, 1, false, None); + // Under a soft mask they share, at different sizes on the page: the second must start from + // the mask the first left. Each needs the mask at a different size. + let first = image(&mut doc, 1200, 1200, 3, true, Some(shared_mask)); + let second = image(&mut doc, 1000, 1000, 3, true, Some(shared_mask)); + page(&mut doc, &[("Im0", first)], "q 144 0 0 144 100 100 cm /Im0 Do Q"); + page(&mut doc, &[("Im0", second)], "q 180 0 0 180 100 100 cm /Im0 Do Q"); + // Enough others for the threads. + for page_no in 0..22u32 { + let img = image(&mut doc, 500 + page_no, 500, 3, true, None); + page(&mut doc, &[("Im0", img)], "q 72 0 0 72 100 100 cm /Im0 Do Q"); + } + doc + }; + let run = |workers| { + let mut doc = build(); + let report = optimize(&mut doc, &Settings { workers: Some(workers), ..Settings::default() }).unwrap(); + assert_eq!(report.images, 24, "{report:?}"); + assert!(report.images_resampled + report.images_recompressed > 16, "{report:?}"); + (write_full(&doc, &SaveOptions::default()).unwrap(), format!("{report:?}")) + }; + let (one, one_report) = run(1); + for workers in [2, 4, 7] { + let (many, many_report) = run(workers); + assert_eq!(many_report, one_report, "{workers} threads count the same"); + assert!(many == one, "{workers} threads write the same bytes"); + } +} + +/// A table that disagrees with an object's header leaves it reachable under two generations, as +/// one object: each use decides from what the one before it left, with any number of threads. +#[test] +fn one_object_under_two_generations_comes_out_the_same_with_any_number_of_threads() { + let picture = |size: usize, extra: &str| { + let samples: Vec = (0..size * size).map(|i| ((i % size) * 255 / size) as u8 ^ ((i / size) as u8 & 7)).collect(); + let dict = format!( + "<< /Type /XObject /Subtype /Image /Width {size} /Height {size} /BitsPerComponent 8 /ColorSpace /DeviceGray{extra} /Length {} >>\nstream\n", + samples.len() + ); + [dict.as_bytes(), &samples, b"\nendstream"].concat() + }; + let build = || { + // 1 the catalog, 2 the pages, 3 the mask, 4 and 5 under it, 6 the picture, 7 a picture that is + // also 8's mask, then the others. + let mut objects = vec![b"<< /Type /Catalog /Pages 2 0 R >>".to_vec(), Vec::new()]; + objects.push(picture(1200, "")); + objects.push(picture(1200, " /SMask 3 0 R")); + objects.push(picture(1000, " /SMask 3 1 R")); + objects.push(picture(1200, "")); + objects.push(picture(1200, "")); + objects.push(picture(1000, " /SMask 7 1 R")); + // Small under one generation, large under the other: they want different sizes. + let mut draws = vec![ + ("4 0 R".to_string(), 144), + ("5 0 R".to_string(), 180), + ("6 0 R".to_string(), 72), + ("6 1 R".to_string(), 576), + ("7 0 R".to_string(), 72), + ("8 0 R".to_string(), 180), + ]; + for i in 0..22 { + objects.push(picture(500 + i, "")); + draws.push((format!("{} 0 R", objects.len()), 72)); + } + let mut kids = String::new(); + for (drawn, size) in &draws { + let content = format!("q {size} 0 0 {size} 10 10 cm /Im0 Do Q"); + objects.push(format!("<< /Length {} >>\nstream\n{content}\nendstream", content.len()).into_bytes()); + let contents = objects.len(); + objects.push( + format!( + "<< /Type /Page /Parent 2 0 R /MediaBox [0 0 612 792] /Contents {contents} 0 R /Resources << /XObject << /Im0 {drawn} >> >> >>" + ) + .into_bytes(), + ); + kids.push_str(&format!("{} 0 R ", objects.len())); + } + objects[1] = format!("<< /Type /Pages /Kids [{kids}] /Count {} >>", draws.len()).into_bytes(); + let mut bytes = b"%PDF-1.7\n".to_vec(); + let mut table = format!("xref\n0 {}\n0000000000 65535 f \n", objects.len() + 1); + for (i, object) in objects.iter().enumerate() { + let num = i + 1; + // Headers say generation 0, and the table says 1 of these three. + let generation = u8::from(num == 3 || num == 6 || num == 7); + table.push_str(&format!("{:010} {generation:05} n \n", bytes.len())); + bytes.extend(format!("{num} 0 obj\n").as_bytes()); + bytes.extend(object); + bytes.extend(b"\nendobj\n"); + } + let start = bytes.len(); + bytes.extend(table.as_bytes()); + bytes.extend(format!("trailer\n<< /Size {} /Root 1 0 R >>\nstartxref\n{start}\n%%EOF\n", objects.len() + 1).as_bytes()); + let doc = Document::open(Arc::new(bytes)).unwrap(); + for (num, generation) in [(3, 0), (3, 1), (6, 0), (6, 1), (7, 0), (7, 1)] { + assert!(matches!(&*doc.get(ObjRef { num, generation }), Object::Stream(_)), "{num} {generation} finds the object"); + } + doc + }; + let run = |workers| { + let mut doc = build(); + let report = optimize(&mut doc, &Settings { workers: Some(workers), ..Settings::default() }).unwrap(); + (write_full(&doc, &SaveOptions::default()).unwrap(), format!("{report:?}")) + }; + let (one, one_report) = run(1); + for workers in [2, 4, 7] { + let (many, many_report) = run(workers); + assert_eq!(many_report, one_report, "{workers} threads count the same"); + assert!(many == one, "{workers} threads write the same bytes"); + } +} + +/// Pages too small for a run aren't read, and that must not cost the smallest run there can be. +#[test] +fn the_fewest_strips_that_make_a_picture_are_merged() { + use pdfcraft_optimize::fragments::MIN_FRAGMENTS; + let keep = Settings { + gray: ImageSettings { downsample: false, target_ppi: 150.0, above_ppi: 225.0, compression: Compression::Retain }, + ..Settings::default() + }; + let (mut doc, _) = printed_page(MIN_FRAGMENTS as _, |_| false, |_, s| s); + let r = optimize(&mut doc, &keep).unwrap(); + assert_eq!((r.merged_images, r.fragments_merged), (1, MIN_FRAGMENTS), "{r:?}"); + let (mut doc, _) = printed_page((MIN_FRAGMENTS - 1) as _, |_| false, |_, s| s); + let r = optimize(&mut doc, &keep).unwrap(); + assert_eq!((r.merged_images, r.fragments_merged), (0, 0), "{r:?}"); +} diff --git a/crates/ui-egui/src/workflows/optimize_ui.rs b/crates/ui-egui/src/workflows/optimize_ui.rs index 6c6bedad00..89f82b00a0 100644 --- a/crates/ui-egui/src/workflows/optimize_ui.rs +++ b/crates/ui-egui/src/workflows/optimize_ui.rs @@ -80,6 +80,11 @@ pub struct OptimizeRun { fn stage_label(stage: OptimizeStage) -> String { match stage { OptimizeStage::Discarding => tl!("Optimizing… removing user data").to_string(), + OptimizeStage::Reading { done: 0, .. } => tl!("Optimizing… reading the pages").to_string(), + OptimizeStage::Reading { done, total } => crate::i18n::fmt( + tl!("Optimizing… reading page {d} of {t}"), + &[("d", &(done + 1).min(total.max(1)).to_string()), ("t", &total.max(1).to_string())], + ), OptimizeStage::Fragments => tl!("Optimizing… merging image fragments").to_string(), OptimizeStage::Images { done, total } => crate::i18n::fmt( tl!("Optimizing… image {d} of {t}"), @@ -418,7 +423,10 @@ impl PdfCraftApp { #[cfg(not(target_arch = "wasm32"))] if self.run_inline { work(); - } else if let Err(e) = std::thread::Builder::new().name("pdfcraft-optimize".into()).spawn(work) { + } else if let Err(e) = std::thread::Builder::new().name("pdfcraft-optimize".into()).spawn(move || { + pdfcraft_engine::optimize::give_way_to_other_programs(); + work(); + }) { // No worker: report the failure instead of waiting for a result that never comes. if let Ok(mut s) = progress.lock() { s.result = Some(Err(EditError::Optimize(e.to_string()))); @@ -504,7 +512,7 @@ mod tests { } h.get_by_label("Optimizing… image 4 of 10"); let shown = h.state().progress_notice.clone().unwrap(); - assert!((shown.fraction - 0.275).abs() < 1e-4, "{shown:?}"); + assert!((shown.fraction - 0.415).abs() < 1e-4, "{shown:?}"); assert!(shown.cancellable); h.get_by_label("Cancel").click(); @@ -520,6 +528,16 @@ mod tests { assert_eq!(h.state().toast.as_ref().map(|t| t.0.as_str()), Some("Optimization cancelled")); } + #[test] + fn the_progress_card_says_what_is_happening() { + let mut app = PdfCraftApp::new(); + app.set_option("language", "en").unwrap(); + assert_eq!(stage_label(OptimizeStage::default()), "Optimizing… reading the pages"); + assert_eq!(stage_label(OptimizeStage::Reading { done: 11, total: 51 }), "Optimizing… reading page 12 of 51"); + assert_eq!(stage_label(OptimizeStage::Reading { done: 51, total: 51 }), "Optimizing… reading page 51 of 51"); + assert_eq!(stage_label(OptimizeStage::Discarding), "Optimizing… removing user data"); + } + #[test] fn a_second_run_is_refused_while_one_is_going() { let mut app = PdfCraftApp::new();