mirror of
https://github.com/storytold/pdfcraft.git
synced 2026-11-09 04:22:20 +00:00
perf(optimize): work out pictures on threads and skip needless page reads (#6230)
This commit is contained in:
quincy-jc authored and 
GitHub committed
1 parent
25aef86a1d
commit
d0b2a32744
13 files changed
+870
-103
No files matched your search
Generated
+15
@@ -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"
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<f32> = 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);
|
||||
|
||||
@@ -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
|
||||
@@ -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
|
||||
|
||||
@@ -243,14 +243,22 @@ fn merge_page(
|
||||
images: &mut ImageBudget,
|
||||
) -> Result<usize, OptimizeError> {
|
||||
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<Op> = 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]
|
||||
|
||||
+323
-83
@@ -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<ObjRef, f64> {
|
||||
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<ObjRef, f64>, HashSet<u32>, 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<ObjRef, f64>, HashSet<u32>, 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<PagesWalked, OptimizeError> {
|
||||
let mut walk = Walk {
|
||||
doc,
|
||||
path: Vec::new(),
|
||||
@@ -269,7 +278,10 @@ fn walk_pages(doc: &Document, pages: &[ObjRef]) -> (HashMap<ObjRef, f64>, 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<ObjRef, f64>, 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<Held> = 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<i64>, Option<i64>), bits: Option<Vec<u8>>, report: &mut Report| {
|
||||
let side = |v: Option<i64>| 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<u32, usize> = 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<ObjRef> {
|
||||
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<u32, usize>, 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<u64> {
|
||||
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<Vec<u8>>,
|
||||
},
|
||||
}
|
||||
|
||||
/// 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<Held>) {
|
||||
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<u8> = (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<ObjRef> = (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<Vec<u8>> = 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.
|
||||
|
||||
@@ -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<usize>,
|
||||
/// 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,
|
||||
|
||||
@@ -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>) -> 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<T: Sync, R: Send>(
|
||||
items: &[T],
|
||||
workers: usize,
|
||||
work: impl Fn(&T) -> R + Sync,
|
||||
tick: &mut dyn FnMut(usize) -> bool,
|
||||
) -> Result<Vec<Option<R>>, OptimizeError> {
|
||||
let mut out: Vec<Option<R>> = 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<u64> = (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::<Vec<_>>());
|
||||
}
|
||||
|
||||
#[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<u32> = (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}");
|
||||
}
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
@@ -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<u8> = (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:?}");
|
||||
}
|
||||
@@ -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();
|
||||
|
||||
Reference in new issue
Block a user