diff --git a/crates/io/src/psd_export.rs b/crates/io/src/psd_export.rs index b5e9db31..9db3d479 100644 --- a/crates/io/src/psd_export.rs +++ b/crates/io/src/psd_export.rs @@ -298,6 +298,10 @@ fn upsert(raw: &mut Vec<([u8; 4], Vec)>, key: &[u8; 4], data: Vec, befor } } +fn h_of(r: photocraft_geom::Rect) -> usize { + r.height() as usize +} + fn psd_mode(m: ColorMode) -> PsdMode { match m { ColorMode::Grayscale => PsdMode::Grayscale, @@ -335,18 +339,28 @@ impl Ex { (-1..self.cc as i16).map(|id| self.encode(id, &[], 0, 0)).collect() } - fn pixel_channels(&mut self, s: &Surface, name: &str) -> (PsdRect, Vec) { + fn pixel_channels(&self, s: &Surface, name: &str) -> (PsdRect, Vec, Vec) { + let mut warnings = Vec::new(); let s = if s.format() != self.fmt { - self.warnings.push(format!("layer \"{name}\": pixels converted from {:?} to {:?}", s.format(), self.fmt)); + warnings.push(format!("layer \"{name}\": pixels converted from {:?} to {:?}", s.format(), self.fmt)); s.convert(self.fmt) } else { s.clone() }; let r = s.content_bounds(); if r.is_empty() { - return (PsdRect::default(), self.empty_channels()); + return (PsdRect::default(), self.empty_channels(), warnings); + } + // The interleaved gather is the sequential pass for full-canvas layers, so it runs + // in row bands on all cores and the band results concatenate in order. + let row_len = r.width() as usize; + let bands = crate::pixels::bands(h_of(r)); + let parts = + crate::pixels::par_map(bands, |rows| s.to_interleaved(photocraft_geom::Rect::new(r.x0, r.y0 + rows.start as i32, r.x1, r.y0 + rows.end as i32))); + let mut bytes = Vec::with_capacity(row_len * h_of(r) * s.format().bytes_per_pixel()); + for part in parts { + bytes.extend_from_slice(&part); } - let mut bytes = s.to_interleaved(r); if self.fmt.mode == ColorMode::Lab && self.fmt.sample == SampleType::U16 { crate::pixels::lab16_chroma(&mut bytes, self.cc + 1, false); } @@ -359,7 +373,7 @@ impl Ex { let order: Vec<(i16, &Vec)> = std::iter::once((-1, &planes[self.cc])).chain(planes.iter().take(self.cc).enumerate().map(|(c, p)| (c as i16, p))).collect(); let ch = crate::pixels::par_map(order, |(id, plane)| self.encode(id, plane, w, h)); - (to_psd_rect(r), ch) + (to_psd_rect(r), ch, warnings) } fn mask(&self, m: &LayerMask, vector: Option<(f32, f32)>) -> (MaskData, Option) { @@ -410,10 +424,11 @@ impl Ex { fn record(&mut self, l: &Layer, extra: Vec, pixels: Option<&Surface>) -> LayerRecord { let id = self.layer_id(l); - let (rect, mut channels) = match pixels { + let (rect, mut channels, warnings) = match pixels { Some(s) => self.pixel_channels(s, &l.name), - None => (PsdRect::default(), self.empty_channels()), + None => (PsdRect::default(), self.empty_channels(), Vec::new()), }; + self.warnings.extend(warnings); let mut mask = MaskData::None; // Vector mask density/feather live in the mask parameters, with or without a user mask. let vector = l.vector_mask.as_ref().map(|vm| (vm.density, vm.feather)); diff --git a/crates/io/src/psd_import.rs b/crates/io/src/psd_import.rs index 323fe600..444c2d6e 100644 --- a/crates/io/src/psd_import.rs +++ b/crates/io/src/psd_import.rs @@ -163,6 +163,13 @@ impl Ctx<'_> { } } + /// [`Self::channel_plane`] without the warnings sink, so a layer's channels decode + /// concurrently (their RLE rows are the bulk of the open time on full-canvas layers). + fn channel_plane_decoded(&self, rec: &LayerRecord, id: i16) -> Option> { + rec.channel(id)?; + rec.decode_channel(id, self.file.header.depth, self.file.header.version).ok() + } + fn record_surface(&mut self, rec: &LayerRecord, name: &str) -> Surface { let r = rec.rect; if r.is_empty() || r.size().is_err() { @@ -170,15 +177,28 @@ impl Ctx<'_> { } let (w, h) = r.size().unwrap_or((0, 0)); let s = self.fmt.sample; - let mut planes = Vec::with_capacity(self.cc + 1); - for c in 0..self.cc { - planes.push(self.channel_plane(rec, c as i16, name)); - } - planes.push(self.channel_plane(rec, CHANNEL_TRANSPARENCY, name)); + let ids: Vec = (0..self.cc as i16).chain([CHANNEL_TRANSPARENCY]).collect(); + let planes: Vec>> = if cfg!(target_arch = "wasm32") { + ids.iter().map(|&id| self.channel_plane(rec, id, name)).collect() + } else { + // The channels decode independently; their RLE rows are the bulk of the open + // time for full-canvas layers, so they run side by side and the warnings land + // in the same order afterwards. + // par_map reuses the crate's worker pool (one OS thread per channel per layer + // would spawn thousands on a file with many layers). + let decoded: Vec>> = crate::pixels::par_map(ids.to_vec(), |id| self.channel_plane_decoded(rec, id)); + let mut planes: Vec>> = Vec::with_capacity(decoded.len()); + for (&id, p) in ids.iter().zip(decoded) { + if p.is_none() && rec.channel(id).is_some() { + self.warn(format!("layer \"{name}\": channel {id} could not be decoded; treated as empty")); + } + planes.push(p); + } + planes + }; // A channel that is in the file but could not be decoded leaves the layer empty, as the // warning says. Filling in for it would paint an opaque black layer over the document. - let ids = (0..self.cc as i16).chain([CHANNEL_TRANSPARENCY]); - if ids.zip(&planes).any(|(id, p)| p.is_none() && rec.channel(id).is_some()) { + if ids.iter().zip(planes.iter()).any(|(&id, p)| p.is_none() && rec.channel(id).is_some()) { return Surface::new(self.fmt); } // Only decoded channel data justifies a buffer: the decoders bound their output by the diff --git a/crates/io/tests/psd_roundtrip_bytes.rs b/crates/io/tests/psd_roundtrip_bytes.rs new file mode 100644 index 00000000..dccd5785 --- /dev/null +++ b/crates/io/tests/psd_roundtrip_bytes.rs @@ -0,0 +1,28 @@ +//! The parallel RLE/decode paths must not change output bytes: a document exported by this +//! branch re-exports byte-equal after a decode/encode cycle, and matches main's exporter +//! (verified by the corpus round-trip floors in `corpus.rs`; here the byte-equality of the +//! codec layer itself is pinned). + +use photocraft_psd::compression::{Compression, PlaneLayout, decode_planes, encode_planes}; +use photocraft_psd::header::Version; + +#[test] +fn encode_decode_planes_round_trip_exactly() { + let mut data: u64 = 0x0123_4567_89ab_cdef; + let mut next = move || { + data ^= data << 13; + data ^= data >> 7; + data ^= data << 17; + data + }; + for (version, planes, w, h, depth) in + [(Version::Psd, 1usize, 640usize, 480usize, 8u16), (Version::Psb, 1, 15000, 10000, 8), (Version::Psd, 4, 513, 65, 8), (Version::Psb, 3, 64, 2049, 16)] + { + let layout = PlaneLayout { planes, width: w, height: h, depth, version }; + let n = layout.decoded_len().expect("size"); + let pattern: Vec = if h % 2 == 0 { vec![42u8; n] } else { (0..n).map(|i| ((i as u64 ^ next()) % 251) as u8).collect() }; + let encoded = encode_planes(Compression::Rle, &pattern, &layout).expect("encode"); + let decoded = decode_planes(Compression::Rle, &encoded, &layout).expect("decode"); + assert_eq!(decoded, pattern, "{version:?} {planes}x{w}x{h} d{depth}"); + } +} diff --git a/crates/psd/src/compression.rs b/crates/psd/src/compression.rs index d47408d2..7ad2296f 100644 --- a/crates/psd/src/compression.rs +++ b/crates/psd/src/compression.rs @@ -8,7 +8,7 @@ use std::io::{Read, Write}; use crate::error::{PsdError, Result}; use crate::header::{Version, row_bytes}; -use crate::io::{Reader, WriteExt}; +use crate::io::Reader; /// Maximum number of decoded bytes a single decode call may produce: 8 GiB on /// 64-bit targets, 2 GiB elsewhere (the same budget as `photocraft-codecs`' @@ -345,21 +345,62 @@ fn encode_rle(decoded: &[u8], layout: &PlaneLayout) -> Result> { let rows = layout.rows()?; let rb = layout.row_bytes(); let cs = layout.count_size(); - let mut out = vec![0u8; rows * cs]; - let mut enc = Vec::new(); - for row in 0..rows { - let start = enc.len(); - packbits::encode(&decoded[row * rb..(row + 1) * rb], &mut enc); - let n = enc.len() - start; - if cs == 2 { - let n = u16::try_from(n).map_err(|_| PsdError::LimitExceeded("RLE row exceeds 65535 bytes (use PSB)"))?; - out[row * 2..row * 2 + 2].copy_from_slice(&n.to_be_bytes()); - } else { - let n = u32::try_from(n).map_err(|_| PsdError::LimitExceeded("RLE row too large"))?; - out[row * 4..row * 4 + 4].copy_from_slice(&n.to_be_bytes()); - } + if rows == 0 { + return Ok(Vec::new()); + } + // Rows are independent, so they pack in parallel chunks and concatenate in order. + let threads = if cfg!(target_arch = "wasm32") { 1 } else { std::thread::available_parallelism().map_or(1, |v| v.get()).clamp(1, 32) }; + let chunk_rows = rows.div_ceil(threads).max(1); + struct Chunk { + counts: Vec, + packed: Vec, + } + let pack_chunk = |first: usize, count: usize| -> Result { + let mut chunk = Chunk { counts: vec![0u8; count * cs], packed: Vec::new() }; + for row in first..first + count { + let start = chunk.packed.len(); + packbits::encode(&decoded[row * rb..(row + 1) * rb], &mut chunk.packed); + let n = chunk.packed.len() - start; + if cs == 2 { + let n = u16::try_from(n).map_err(|_| PsdError::LimitExceeded("RLE row exceeds 65535 bytes (use PSB)"))?; + chunk.counts[(row - first) * 2..(row - first) * 2 + 2].copy_from_slice(&n.to_be_bytes()); + } else { + let n = u32::try_from(n).map_err(|_| PsdError::LimitExceeded("RLE row too large"))?; + chunk.counts[(row - first) * 4..(row - first) * 4 + 4].copy_from_slice(&n.to_be_bytes()); + } + } + Ok(chunk) + }; + let chunk_list: Vec = if threads == 1 { + vec![pack_chunk(0, rows)?] + } else { + std::thread::scope(|scope| -> Result> { + let handles: Vec<_> = (0..threads) + .map(|t| { + let first = t * chunk_rows; + let count = chunk_rows.min(rows.saturating_sub(first)); + scope.spawn(move || pack_chunk(first, count)) + }) + .collect(); + // Join every worker first, then surface the first error like the sequential + // path would instead of aborting the scope with a thread panic. + let mut all = Vec::with_capacity(threads); + for h in handles { + all.push(h.join().map_err(|_| PsdError::LimitExceeded("an RLE worker panicked"))??); + } + Ok(all) + })? + }; + let packed_total: usize = chunk_list.iter().map(|c| c.packed.len()).sum(); + let mut out = vec![0u8; rows * cs + packed_total]; + let mut counts_at = 0usize; + let mut packed_at = rows * cs; + for chunk in &chunk_list { + out[counts_at..counts_at + chunk.counts.len()].copy_from_slice(&chunk.counts); + counts_at += chunk.counts.len(); + out[packed_at..packed_at + chunk.packed.len()].copy_from_slice(&chunk.packed); + packed_at += chunk.packed.len(); } - out.put(&enc); Ok(out) } @@ -830,3 +871,79 @@ mod tests { assert!(zip_decompress(&[1, 2, 3, 4], 4).is_err()); } } + +#[cfg(test)] +mod parallel_rle_equivalence { + use super::*; + + /// The parallel row packer's output must be byte-equal to a straightforward sequential + /// encoding of the same plane, across depths, row counts around the chunking and data + /// shapes (runs, noise, all-default). + #[test] + fn parallel_rle_matches_sequential() { + let mut data: u64 = 0x9e37_79b9_7f4a_7c15; + let mut next = move || { + data ^= data << 13; + data ^= data >> 7; + data ^= data << 17; + data + }; + let layout_for = |rows: usize, rb: usize, version: Version| PlaneLayout { planes: 1, width: rb, height: rows, depth: 8, version }; + for (rows, rb, version) in [ + (0usize, 4usize, Version::Psd), + (1, 4, Version::Psd), + (7, 4, Version::Psb), + (64, 4, Version::Psd), + (65, 4, Version::Psb), // chunk boundary + (513, 4, Version::Psd), // more rows than threads on most runners + (2049, 4, Version::Psb), + (2, 4, Version::Psd), // fewer rows than threads + (33, 3000, Version::Psd), + (31, 20000, Version::Psb), + ] { + let pattern: Vec = match rows % 3 { + 0 => vec![7u8; rows * rb], // long runs + 1 => (0..rows * rb).map(|i| (i * 31 % 251) as u8).collect(), // noise-ish + _ => vec![0u8; rows * rb], // all default + }; + let pattern = if pattern.len() != rows * rb { vec![7u8; rows * rb] } else { pattern }; + let mut data = pattern.clone(); + if rows % 4 == 0 { + // sprinkle runs into the noise + for chunk in data.chunks_mut(97) { + if let Some(f) = chunk.first_mut() { + *f = 200; + } + } + } + let _ = next(); + let layout = layout_for(rows, rb, version); + let parallel = encode_rle(&data, &layout).expect("parallel encode"); + let sequential = encode_rle_sequential(&data, &layout).expect("sequential encode"); + assert_eq!(parallel, sequential, "rows {rows} rb {rb}"); + } + } + + /// The pre-parallel encoder, kept as the oracle. + fn encode_rle_sequential(decoded: &[u8], layout: &PlaneLayout) -> Result> { + let rows = layout.rows()?; + let rb = layout.row_bytes(); + let cs = layout.count_size(); + let mut out = vec![0u8; rows * cs]; + let mut enc = Vec::new(); + for row in 0..rows { + let start = enc.len(); + packbits::encode(&decoded[row * rb..(row + 1) * rb], &mut enc); + let n = enc.len() - start; + if cs == 2 { + let n = u16::try_from(n).map_err(|_| PsdError::LimitExceeded("RLE row exceeds 65535 bytes (use PSB)"))?; + out[row * 2..row * 2 + 2].copy_from_slice(&n.to_be_bytes()); + } else { + let n = u32::try_from(n).map_err(|_| PsdError::LimitExceeded("RLE row too large"))?; + out[row * 4..row * 4 + 4].copy_from_slice(&n.to_be_bytes()); + } + } + out.extend_from_slice(&enc); + Ok(out) + } +}