PSD: parallel RLE for the merged image; layer channels decode and gather on all cores (#2754)
Saving a 36 MP document spent most of its time packing the merged image's RLE rows one after another on one thread (packbits over ~144 MB), and importing it decoded each layer's channels sequentially. The merged-image row packer now runs in parallel chunks that concatenate in order (rows are independent; errors surface like the sequential path's), the layer gather runs in row bands on all cores before the per-channel RLE, the layer's channels decode side by side, and the full-surface clone per layer is gone. Measured on a 6000×4000 8-bit document with a photo, a paint layer and a Levels adjustment: save 2.86 s -> 2.55-2.64 s, open's import stage 1.12 s -> 1.07 s (psd_to_document 745 -> 564 ms). The remaining save time is dominated by the merged composite render itself (~0.7 s).
This commit is contained in:
Elvis33LE
GitHub
parent
995bc89c40
commit
fc0a579a4a
|
|
@ -298,6 +298,10 @@ fn upsert(raw: &mut Vec<([u8; 4], Vec<u8>)>, key: &[u8; 4], data: Vec<u8>, 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<ChannelData>) {
|
||||
fn pixel_channels(&self, s: &Surface, name: &str) -> (PsdRect, Vec<ChannelData>, Vec<String>) {
|
||||
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<u8>)> =
|
||||
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<ChannelData>) {
|
||||
|
|
@ -410,10 +424,11 @@ impl Ex {
|
|||
|
||||
fn record(&mut self, l: &Layer, extra: Vec<TaggedBlock>, 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));
|
||||
|
|
|
|||
|
|
@ -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<Vec<u8>> {
|
||||
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<i16> = (0..self.cc as i16).chain([CHANNEL_TRANSPARENCY]).collect();
|
||||
let planes: Vec<Option<Vec<u8>>> = 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<Option<Vec<u8>>> = crate::pixels::par_map(ids.to_vec(), |id| self.channel_plane_decoded(rec, id));
|
||||
let mut planes: Vec<Option<Vec<u8>>> = 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
|
||||
|
|
|
|||
|
|
@ -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<u8> = 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}");
|
||||
}
|
||||
}
|
||||
|
|
@ -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<Vec<u8>> {
|
|||
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<u8>,
|
||||
packed: Vec<u8>,
|
||||
}
|
||||
let pack_chunk = |first: usize, count: usize| -> Result<Chunk> {
|
||||
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<Chunk> = if threads == 1 {
|
||||
vec![pack_chunk(0, rows)?]
|
||||
} else {
|
||||
std::thread::scope(|scope| -> Result<Vec<Chunk>> {
|
||||
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<u8> = 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<Vec<u8>> {
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue