diff --git a/CHANGELOG.md b/CHANGELOG.md index eb5f6535..9153adcd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,19 @@ This changelog tracks the current release line. Historical phase notes and stale roadmap entries have been removed from the public documentation set. +## [Unreleased] + +- Parallel CPU region decodes on the row-streamed IDWT decode Tier-1 band by + band on the worker pool while the IDWT reads the bands, instead of decoding + every code-block of a component first. Each sub-band holds three bands at + the finest level and two at the others, or its whole window when that is + smaller, and the IDWT decodes a band itself when no worker has started it. + This applies to tiles the tile-wide route decodes: HTJ2K, and classic + code-blocks of one arithmetic segment. A 1024 x 1024 region of an + 18000 x 12000 RGB image decodes 3 to 11% faster at 12 threads and 9 to 18% + faster at 4, with about 3 MiB less peak heap. Output is unchanged, and a + failure reruns whole-window Tier-1 serially for the same error. + ## [0.13.1] - 2026-10-29 - `CpuBatchDecoder` region and region-plus-reduced requests no longer diff --git a/crates/j2k-native/src/j2c/build/allocation.rs b/crates/j2k-native/src/j2c/build/allocation.rs index 40356038..804e7e94 100644 --- a/crates/j2k-native/src/j2c/build/allocation.rs +++ b/crates/j2k-native/src/j2c/build/allocation.rs @@ -369,7 +369,7 @@ pub(crate) fn allocate_deferred_coefficients( crate::j2c::idwt::roi_idwt_peak_elements(storage, skipped_resolution_levels)?; storage.roi_idwt_charge_elements = idwt_elements; checked_include_elements::(&mut live_bytes, idwt_elements)?; - lay_out_region_coefficients(storage, clip_to_idwt_windows)? + lay_out_region_coefficients(storage, clip_to_idwt_windows, skipped_resolution_levels)? } else { checked_include_elements::(&mut live_bytes, idwt_workspace_elements)?; storage @@ -386,14 +386,22 @@ pub(crate) fn allocate_deferred_coefficients( Ok(()) } +/// Band slots each sub-band of the finest active level holds when Tier-1 +/// decodes bands ahead of the streamed IDWT on the worker pool, and each +/// sub-band of the other levels. The finest level's bands are the most work +/// and are each read for the shortest time, so they need the deepest ring. +pub(crate) const FINEST_LEVEL_BAND_SLOTS: u32 = 3; +pub(crate) const OTHER_LEVEL_BAND_SLOTS: u32 = 2; + /// Give each sub-band of a region decode the coefficient range for its /// stored rectangle: the required code-blocks, clipped to the IDWT windows /// unless `clip_to_idwt_windows` is false. With banded Tier-1, a range holds -/// one band of the rectangle's rows. Returns the coefficients the ranges -/// need. +/// one band of the rectangle's rows, and with parallel bands as many bands as +/// the sub-band has slots. Returns the coefficients the ranges need. fn lay_out_region_coefficients( storage: &mut DecompositionStorage<'_>, clip_to_idwt_windows: bool, + skipped_resolution_levels: u8, ) -> Result { let Some(roi_plan) = storage.roi_plan.as_ref() else { return Err(DecodingError::CodeBlockDecodeFailure.into()); @@ -433,9 +441,29 @@ fn lay_out_region_coefficients( }) .unwrap_or_else(|| IntRect::from_xywh(sub_band.rect.x0, sub_band.rect.y0, 0, 0)); // Banded Tier-1 holds one band of code-block rows at a time: as - // many rows as any band shares with the stored rectangle. - let rows = if storage.banded_tier1 { - storage.precincts[sub_band.precincts.clone()] + // many rows as any band shares with the stored rectangle. Parallel + // bands hold a ring of such bands. + let slots = if storage.parallel_bands { + let finest = storage.tile_decompositions.iter().any(|component| { + let active = component + .decompositions + .len() + .saturating_sub(usize::from(skipped_resolution_levels)); + active > 0 + && storage.decompositions[component.decompositions.start + active - 1] + .sub_bands + .contains(&sub_band_idx) + }); + if finest { + FINEST_LEVEL_BAND_SLOTS + } else { + OTHER_LEVEL_BAND_SLOTS + } + } else { + 1 + }; + let rows = if storage.banded_tier1 || storage.parallel_bands { + let band_rows = storage.precincts[sub_band.precincts.clone()] .iter() .flat_map(|precinct| &storage.code_blocks[precinct.code_blocks.clone()]) .map(|code_block| { @@ -446,7 +474,12 @@ fn lay_out_region_coefficients( .saturating_sub(code_block.rect.y0.max(stored.y0)) }) .max() - .unwrap_or(0) + .unwrap_or(0); + // A ring never holds more rows than the whole window. + band_rows + .checked_mul(slots) + .ok_or(ValidationError::ImageTooLarge)? + .min(stored.height().max(band_rows)) } else { stored.height() }; @@ -464,13 +497,15 @@ fn lay_out_region_coefficients( Ok(coefficient_count) } -/// After a banded Tier-1 failed, store every sub-band's whole rectangle -/// again, within the cap, so Tier-1 can rerun the old way. +/// After a banded Tier-1 failed, or parallel bands declined, store every +/// sub-band's whole rectangle again, within the cap, so Tier-1 can run the +/// old way. pub(crate) fn lay_out_whole_region_coefficients( storage: &mut DecompositionStorage<'_>, ) -> Result<()> { storage.banded_tier1 = false; - let coefficient_count = lay_out_region_coefficients(storage, true)?; + storage.parallel_bands = false; + let coefficient_count = lay_out_region_coefficients(storage, true, 0)?; let mut budget = reuse::ReallocationBudget::from_live_bytes(storage.structural_workspace_bytes)?; budget.reserve(&mut storage.coefficients, coefficient_count)?; diff --git a/crates/j2k-native/src/j2c/decode.rs b/crates/j2k-native/src/j2c/decode.rs index 7c0f49f6..e951efa5 100644 --- a/crates/j2k-native/src/j2c/decode.rs +++ b/crates/j2k-native/src/j2c/decode.rs @@ -39,10 +39,14 @@ use core::ops::Range; mod allocation; mod banded_tier1; mod direct_plan; +#[cfg(feature = "parallel")] +mod parallel_bands; mod peak; mod reuse; mod store; mod subband; +#[cfg(all(test, feature = "parallel"))] +pub(crate) use self::subband::MIN_BAND_WORK; mod subband_params; mod tier1; #[cfg(feature = "parallel")] @@ -687,6 +691,12 @@ pub(crate) struct DecodeDebugCounters { pub(crate) selection_scanned_packets: usize, /// Empty decomposition capacity discarded before region selection. pub(crate) selection_released_storage_bytes: usize, + /// Components whose bands decoded on the worker pool ahead of the + /// streamed IDWT, summed over decoded tiles. + pub(crate) pool_band_components: usize, + /// Components that gave up the pool for whole windows because a band + /// was left running with no worker to finish it. + pub(crate) pool_band_fallbacks: usize, /// The bytes the decode's budget starts from after `reset`, retained /// capacity included. pub(crate) decode_baseline_bytes: usize, @@ -883,6 +893,12 @@ fn prepare_region_storage( storage.banded_tier1 = storage.line_region_idwt && tile_bands(header, tile, tile_ctx, hooked, cpu_decode_parallelism); storage.lockstep_lines = storage.banded_tier1 && tile_ctx.stream_rows; + #[cfg(feature = "parallel")] + { + storage.parallel_bands = storage.line_region_idwt + && !storage.banded_tier1 + && parallel_bands::applies(header, tile, tile_ctx, hooked, cpu_decode_parallelism); + } build::allocate_deferred_coefficients( storage, header.skipped_resolution_levels, @@ -995,18 +1011,16 @@ fn decode_components<'a>( } // A tile-part COD or COC can give the tile its own coding style. for (idx, component_info) in tile.component_infos.iter().enumerate() { - if storage.banded_tier1 { - // Tier-1 decodes each band as the streamed IDWT reaches it. - banded_tier1::decode_component( - tile, - idx, - header, - tile_ctx, - storage, - component_info.wavelet_transform(), - profile_enabled, - profile_timings, - )?; + if decode_component_bands( + tile, + idx, + header, + tile_ctx, + storage, + component_info.wavelet_transform(), + profile_enabled, + profile_timings, + )? { record_workspace(tile_ctx, storage, 0)?; } else { if per_component { @@ -1079,6 +1093,53 @@ fn decode_components<'a>( Ok(()) } +/// Decode component `component_idx` of a tile that decodes in bands as the +/// streamed IDWT reads them, on the pool or serially. Returns `false` when it +/// takes the whole-window path instead. +#[expect( + clippy::too_many_arguments, + reason = "the banded stage shares decode_tile's state and profiling explicitly" +)] +fn decode_component_bands<'a>( + tile: &Tile<'a>, + component_idx: usize, + header: &Header<'_>, + tile_ctx: &mut TileDecodeContext, + storage: &mut DecompositionStorage<'a>, + transform: WaveletTransform, + profile_enabled: bool, + profile_timings: &mut DecodeProfileTimings, +) -> Result { + // Unless the route declines and the sub-bands hold whole windows again. + #[cfg(feature = "parallel")] + if storage.parallel_bands { + return parallel_bands::decode_component( + tile, + component_idx, + header, + tile_ctx, + storage, + transform, + profile_enabled, + profile_timings, + ); + } + if !storage.banded_tier1 { + return Ok(false); + } + banded_tier1::decode_component( + tile, + component_idx, + header, + tile_ctx, + storage, + transform, + profile_enabled, + profile_timings, + )?; + Ok(true) +} + /// Decode the code-blocks of `components` and sample the workspace after. #[expect( clippy::too_many_arguments, @@ -1397,6 +1458,10 @@ pub(crate) struct DecompositionStorage<'a> { /// the streamed IDWT reaches them. Each sub-band's `coefficients` then /// hold one band of its stored rectangle's rows. pub(crate) banded_tier1: bool, + /// Whether Tier-1 decodes each component's bands ahead of the streamed + /// IDWT on the worker pool (see [`parallel_bands`]). Each sub-band's + /// `coefficients` then hold a ring of bands. + pub(crate) parallel_bands: bool, /// Whether every component's IDWT rings, and a scratch row each, are /// live at once, because the tile streams its rows to a sink. pub(crate) lockstep_lines: bool, @@ -1437,6 +1502,7 @@ impl DecompositionStorage<'_> { self.last_idwt_level_in_channels = false; self.line_region_idwt = false; self.banded_tier1 = false; + self.parallel_bands = false; self.lockstep_lines = false; self.roi_idwt_charge_elements = 0; self.tier1_band = None; @@ -1529,6 +1595,15 @@ pub(crate) struct TileDecodeContext { /// compared against it. #[cfg(test)] pub(crate) whole_window_tier1: bool, + /// Test-only: the first pool worker to start a band abandons it, so the + /// band is left running with no worker to finish it. + #[cfg(all(test, feature = "parallel"))] + pub(crate) strand_pool_band: bool, + /// Test-only: the average work per band below which a component's + /// bands decode over whole windows instead of on the pool; zero, so + /// unit tests' tiny fixtures take the pool, unless a test sets it. + #[cfg(all(test, feature = "parallel"))] + pub(crate) band_work_cutoff: usize, /// Test-only capacity failure when building the optional band row index. #[cfg(test)] pub(crate) reject_band_index: bool, diff --git a/crates/j2k-native/src/j2c/decode/banded_tier1.rs b/crates/j2k-native/src/j2c/decode/banded_tier1.rs index 5001f06b..d64209f4 100644 --- a/crates/j2k-native/src/j2c/decode/banded_tier1.rs +++ b/crates/j2k-native/src/j2c/decode/banded_tier1.rs @@ -546,7 +546,7 @@ fn index_bands( /// windows, the old way, and return its error, which is the one the /// whole-window path reports. If the whole windows don't fit the cap, or the /// rerun succeeds, keep the banded error. -fn whole_window_error<'a>( +pub(super) fn whole_window_error<'a>( tile: &Tile<'a>, component_idx: usize, header: &Header<'_>, diff --git a/crates/j2k-native/src/j2c/decode/parallel_bands.rs b/crates/j2k-native/src/j2c/decode/parallel_bands.rs new file mode 100644 index 00000000..0be196c2 --- /dev/null +++ b/crates/j2k-native/src/j2c/decode/parallel_bands.rs @@ -0,0 +1,1210 @@ +// SPDX-License-Identifier: MIT OR Apache-2.0 + +//! Tier-1 decoded band by band on the worker pool, ahead of the streamed IDWT. +//! +//! A serial region decode on the streamed IDWT decodes each band of +//! code-block rows just before the IDWT reads it (see `banded_tier1`). With +//! several threads, the tile route instead decodes every code-block of a +//! component before the IDWT starts. Here the route's code-block rows are +//! bands that pool workers decode ahead of the IDWT, which runs on the +//! calling thread, so the two overlap: +//! +//! - Each sub-band holds a ring of band slots. Its band `k` can start once the +//! IDWT has read all of band `k - slots`, which frees that band's slot. +//! - Workers take the band the IDWT will read first. A worker that finds no +//! band it can start returns to the pool rather than wait, and the IDWT +//! sends workers again when it frees slots. +//! - The IDWT waits only for a band a worker is decoding. A band nobody has +//! started it decodes itself, so it never depends on a pool thread being +//! free, even when the pool is busy with other work. +//! +//! Every band is decoded once, including bands the IDWT never reads, so a +//! decode visits, decodes and rejects the code-blocks the whole-window route +//! does. Workers fail in any order, so a failure reruns whole-window Tier-1 +//! serially for the error the whole-window path reports. +//! +//! A running band always belongs to a worker that is still active. A wait +//! that finds a running band with no worker active would otherwise last +//! forever, so it gives up the pool instead, and the component decodes over +//! whole windows; a debug build asserts. + +use alloc::collections::BinaryHeap; +use alloc::vec::Vec; +use core::cmp::Reverse; +use core::mem::size_of; +use core::ops::Range; +use std::sync::{Condvar, Mutex, MutexGuard}; + +use super::banded_tier1; +use super::subband::{plan_band_route, routes_tile, BandRoute, BandRouting}; +use super::workspace::{TaskScratch, TaskWorkspace}; +use super::{ + clips_to_idwt_windows, CpuDecodeParallelism, DecodeProfileTimings, DecompositionStorage, + Header, Result, Tile, TileDecodeContext, +}; +use crate::error::{DecodeError, DecodingError, ValidationError}; +use crate::j2c::build::{self, CodeBlockCoding}; +use crate::j2c::codestream::WaveletTransform; +use crate::j2c::idwt::{ComponentLines, SubBandRows}; +use crate::j2c::rect::IntRect; +use crate::profile; + +/// Sub-band rows of support on each side of the rows a level's lifting +/// reads, in rows of that level, for ordering bands: a 9/7 lift reads two +/// rows past each side, and both of its stages do. +const SUPPORT_ROWS: i64 = 4; + +/// How often a wait for a running band checks that a worker is still active +/// to finish it. +const STRANDED_CHECK: std::time::Duration = std::time::Duration::from_millis(50); + +/// Whether a tile's components decode their bands on the worker pool: it +/// feeds the streamed IDWT, in stored windows clipped to it, under `Auto` +/// with two threads or more, every component has a level to reconstruct, and +/// the tile route decodes its code-blocks. The serial bands of +/// [`banded_tier1`] cover the other streamed tiles. +pub(super) fn applies( + header: &Header<'_>, + tile: &Tile<'_>, + tile_ctx: &TileDecodeContext, + hooked: bool, + parallelism: CpuDecodeParallelism, +) -> bool { + let bands = tile_ctx.line_region_idwt + && parallelism == CpuDecodeParallelism::Auto + && rayon::current_num_threads() >= 2 + && clips_to_idwt_windows(tile_ctx, hooked) + && tile.component_infos.iter().all(|component| { + component.num_decomposition_levels() > header.skipped_resolution_levels + }) + && routes_tile(tile); + #[cfg(test)] + let bands = bands && !tile_ctx.whole_window_tier1 && !tile_ctx.force_per_sub_band; + bands +} + +/// Decode component `component_idx`: its bands on the pool while the +/// streamed IDWT reads them, then the bands it never read. Returns `false` +/// when the route declines the component's bands; its sub-bands then hold +/// whole windows again, for the whole-window path. +#[expect( + clippy::too_many_arguments, + reason = "the banded stage shares decode_tile's state and profiling explicitly" +)] +pub(super) fn decode_component<'a>( + tile: &Tile<'a>, + component_idx: usize, + header: &Header<'_>, + tile_ctx: &mut TileDecodeContext, + storage: &mut DecompositionStorage<'a>, + transform: WaveletTransform, + profile_enabled: bool, + profile_timings: &mut DecodeProfileTimings, +) -> Result { + let stage_start = profile::profile_now(profile_enabled); + let routing = plan_band_route(tile, component_idx, tile_ctx, storage, header); + let route = match routing { + // Planning can fail in a later band before earlier blocks decode. + Err(error) => { + return Err(banded_tier1::whole_window_error( + tile, + component_idx, + header, + tile_ctx, + storage, + error, + )); + } + Ok(BandRouting::Route(route)) => route, + Ok(BandRouting::Serial) => { + // A ring's first slot holds a band, as the serial bands need. + banded_tier1::decode_component( + tile, + component_idx, + header, + tile_ctx, + storage, + transform, + profile_enabled, + profile_timings, + )?; + return Ok(true); + } + Ok(BandRouting::WholeWindows) => { + build::lay_out_whole_region_coefficients(storage)?; + storage.stale_coefficients = true; + return Ok(false); + } + }; + let mut tier1_us = 0; + let result = decode_route( + &route, + component_idx, + header, + tile_ctx, + storage, + transform, + profile_enabled, + &mut tier1_us, + ) + .and_then(|completed| { + if completed { + route.record(tile_ctx, storage)?; + } + Ok(completed) + }); + // Neither fallback's budget includes this plan's live allocations. + drop(route); + let elapsed = profile::elapsed_us(stage_start); + profile_timings.codeblock_us += tier1_us; + profile_timings.idwt_us += elapsed.saturating_sub(tier1_us); + match result { + Ok(true) => { + tile_ctx.debug_counters.pool_band_components += 1; + Ok(true) + } + Ok(false) => { + // A band was left running with no worker to finish it. + #[cfg(test)] + let simulated = tile_ctx.strand_pool_band; + #[cfg(not(test))] + let simulated = false; + debug_assert!(simulated, "a running band had no active worker"); + tile_ctx.debug_counters.pool_band_fallbacks += 1; + build::lay_out_whole_region_coefficients(storage)?; + storage.stale_coefficients = true; + Ok(false) + } + Err(parallel_error) => Err(banded_tier1::whole_window_error( + tile, + component_idx, + header, + tile_ctx, + storage, + parallel_error, + )), + } +} + +#[expect( + clippy::too_many_arguments, + reason = "the banded stage shares decode_tile's state and profiling explicitly" +)] +fn decode_route( + route: &BandRoute<'_>, + component_idx: usize, + header: &Header<'_>, + tile_ctx: &mut TileDecodeContext, + storage: &mut DecompositionStorage<'_>, + transform: WaveletTransform, + profile_enabled: bool, + tier1_us: &mut u128, +) -> Result { + let (rings, bands, active_levels) = plan_rings(route, storage, header, component_idx)?; + // The executor's records live beside the route's plan and scratch while + // the bands decode. + route + .budget() + .include_bytes(executor_bytes(&rings, &bands)?)?; + tile_ctx.idwt_output.placement = crate::j2c::idwt::IdwtPlacement::Coefficients; + tile_ctx.idwt_output.coefficients = Vec::new(); + tile_ctx.idwt_output.rect = IntRect::from_xywh(0, 0, 0, 0); + let mut lines = ComponentLines::prepare( + storage, + tile_ctx, + component_idx, + active_levels, + header, + transform, + )?; + #[cfg(test)] + let strand_once = tile_ctx.strand_pool_band; + #[cfg(not(test))] + let strand_once = false; + let mut coefficients = core::mem::take(&mut storage.coefficients); + let slots = split_slots(&rings, &bands, storage, &mut coefficients); + let result = slots.and_then(|slots| { + let run = |rows: &mut dyn SubBandRows| -> Result<()> { + lines.as_mut().map_or(Ok(()), |lines| lines.run(rows)) + }; + match route.coding() { + CodeBlockCoding::Classic => run_bands( + route, + &rings, + &bands, + slots, + &mut tile_ctx.classic_task_workspaces, + |index, output, workspace, scratch| { + route.decode_classic(index, output, workspace, scratch) + }, + run, + profile_enabled, + tier1_us, + strand_once, + ), + CodeBlockCoding::HighThroughput => run_bands( + route, + &rings, + &bands, + slots, + &mut tile_ctx.ht_task_workspaces, + |index, output, workspace, scratch| { + route.decode_ht(index, output, workspace, scratch) + }, + run, + profile_enabled, + tier1_us, + strand_once, + ), + } + }); + storage.coefficients = coefficients; + // Each band zeroes its slot before it decodes, and the last bands leave + // theirs written. + storage.stale_coefficients = true; + let Some(lines) = lines else { + return result; + }; + match result { + Ok(true) => lines.finish(tile_ctx, Ok(())).map(|()| true), + Ok(false) => { + // Return the rings and plane; the whole-window path refills them. + match lines.finish(tile_ctx, Err(DecodingError::CodeBlockDecodeFailure.into())) { + Ok(()) | Err(_) => Ok(false), + } + } + Err(error) => lines.finish(tile_ctx, Err(error)).map(|()| true), + } +} + +/// The bytes the executor holds for `rings` and `bands`, besides the slots: +/// the records, each band's status, slot, queue entry and decoded-list entry, +/// and each ring's cursors and slot list. +fn executor_bytes(rings: &[Ring], bands: &[Band]) -> Result { + let per_band = size_of::() + + size_of::() + + size_of::() + + 2 * size_of::>() + + size_of::, usize)>>(); + let per_ring = size_of::() + + 3 * size_of::() + + size_of::>() + + size_of::>(); + let slots: usize = rings.iter().map(|ring| ring.slots).sum(); + bands + .len() + .checked_mul(per_band) + .zip(rings.len().checked_mul(per_ring)) + .and_then(|(bands, rings)| bands.checked_add(rings)) + .and_then(|bytes| bytes.checked_add(slots.checked_mul(size_of::<&mut [f32]>())?)) + .ok_or(ValidationError::ImageTooLarge.into()) +} + +/// A sub-band's ring of band slots and its bands, in row order. +struct Ring { + sub_band: usize, + stored: IntRect, + slots: usize, + /// Whether the ring holds the sub-band's whole window, because its bands + /// fit in fewer rows than its slots would: each band then decodes into + /// its own rows of the window, and all of them can be in flight at once. + whole_window: bool, + bands: Range, +} + +/// A band: one strip of the route, in its ring's order. +struct Band { + strip: usize, + ring: usize, + /// The band's rows, in absolute sub-band coordinates. + rows: Range, + /// Where the IDWT first reads the band, in rows of the reconstructed + /// component, approximately: bands it reads sooner start sooner. + order: i64, + work: usize, +} + +/// The component's rings, its bands grouped by ring in row order, and the +/// number of levels it reconstructs. +fn plan_rings( + route: &BandRoute<'_>, + storage: &DecompositionStorage<'_>, + header: &Header<'_>, + component_idx: usize, +) -> Result<(Vec, Vec, usize)> { + let tile_decompositions = &storage.tile_decompositions[component_idx]; + let active_levels = tile_decompositions + .decompositions + .len() + .checked_sub(usize::from(header.skipped_resolution_levels)) + .filter(|levels| *levels > 0) + .ok_or(DecodingError::CodeBlockDecodeFailure)?; + let mut rings = Vec::new(); + rings + .try_reserve_exact(1 + 3 * active_levels) + .map_err(|_| DecodingError::HostAllocationFailed)?; + // A sub-band row at resolution `r` spans 2^(levels + 1 - r) rows of the + // reconstructed component, and the LL band's 2^levels. + let mut scales = Vec::new(); + scales + .try_reserve_exact(rings.capacity()) + .map_err(|_| DecodingError::HostAllocationFailed)?; + for resolution in 0..=active_levels { + let level = u8::try_from(resolution).map_err(|_| DecodingError::CodeBlockDecodeFailure)?; + let shift = if resolution == 0 { + active_levels + } else { + active_levels + 1 - resolution + }; + for sub_band in tile_decompositions.sub_band_iter(level, &storage.decompositions) { + let stored = storage.sub_bands[sub_band].storage_rect; + let coefficients = storage.sub_bands[sub_band].coefficients.len(); + let band_len = route + .strips() + .filter(|strip| strip.sub_band == sub_band) + .map(|strip| strip.rows * stored.width() as usize) + .max() + .unwrap_or(0); + let window = stored.width() as usize * stored.height() as usize; + let whole_window = band_len > 0 && coefficients >= window; + let slots = coefficients.checked_div(band_len).unwrap_or(0); + rings.push(Ring { + sub_band, + stored, + slots, + whole_window, + bands: 0..0, + }); + scales.push(1_i64 << shift.min(62)); + } + } + let mut bands = Vec::new(); + bands + .try_reserve_exact(route.strips().count()) + .map_err(|_| DecodingError::HostAllocationFailed)?; + for (index, ring) in rings.iter_mut().enumerate() { + let start = bands.len(); + for (strip_index, strip) in route + .strips() + .enumerate() + .filter(|(_, strip)| strip.sub_band == ring.sub_band) + { + let first = u32::try_from(strip.first_row) + .ok() + .and_then(|row| ring.stored.y0.checked_add(row)) + .ok_or(DecodingError::CodeBlockDecodeFailure)?; + let last = u32::try_from(strip.rows) + .ok() + .and_then(|rows| first.checked_add(rows)) + .ok_or(DecodingError::CodeBlockDecodeFailure)?; + bands.push(Band { + strip: strip_index, + ring: index, + rows: first..last, + order: (i64::from(first) - SUPPORT_ROWS) * scales[index], + work: strip.work, + }); + } + bands[start..].sort_unstable_by_key(|band| band.rows.start); + ring.bands = start..bands.len(); + if ring.whole_window { + ring.slots = ring.bands.len(); + } + if !ring.bands.is_empty() && ring.slots == 0 { + return Err(DecodingError::CodeBlockDecodeFailure.into()); + } + } + if bands.len() != route.strips().count() { + // A strip of a sub-band outside the component. + return Err(DecodingError::CodeBlockDecodeFailure.into()); + } + Ok((rings, bands, active_levels)) +} + +/// Each ring's free slots and the slot each band of a whole-window ring +/// owns. +type Slots<'c> = (Vec>, Vec>); + +/// Split each ring's coefficients into its slots: equal slots as long as its +/// longest band, or in a whole-window ring, each band's own rows. +fn split_slots<'c>( + rings: &[Ring], + bands: &[Band], + storage: &DecompositionStorage<'_>, + coefficients: &'c mut [f32], +) -> Result> { + let mut slots = Vec::new(); + slots + .try_reserve_exact(rings.len()) + .map_err(|_| DecodingError::HostAllocationFailed)?; + let mut owned = try_filled_with(bands.len(), || None)?; + // Sub-band ranges of a component don't overlap and ascend in ring order. + let mut rest = coefficients; + let mut consumed = 0usize; + for ring in rings { + let range = storage.sub_bands[ring.sub_band].coefficients.clone(); + let mut ring_slots = Vec::new(); + ring_slots + .try_reserve_exact(ring.slots) + .map_err(|_| DecodingError::HostAllocationFailed)?; + if ring.slots > 0 { + let skip = range + .start + .checked_sub(consumed) + .ok_or(DecodingError::CodeBlockDecodeFailure)?; + let tail = core::mem::take(&mut rest); + let (_, tail) = tail + .split_at_mut_checked(skip) + .ok_or(DecodingError::CodeBlockDecodeFailure)?; + let (mut ring_samples, tail) = tail + .split_at_mut_checked(range.len()) + .ok_or(DecodingError::CodeBlockDecodeFailure)?; + rest = tail; + consumed = range.end; + if ring.whole_window { + // Bands of a sub-band cover disjoint rows, in order. + let width = ring.stored.width() as usize; + let mut offset = 0usize; + for index in ring.bands.clone() { + let rows = &bands[index].rows; + let start = (rows.start - ring.stored.y0) as usize * width; + let len = (rows.end - rows.start) as usize * width; + let gap = start + .checked_sub(offset) + .ok_or(DecodingError::CodeBlockDecodeFailure)?; + let (_, tail) = core::mem::take(&mut ring_samples) + .split_at_mut_checked(gap) + .ok_or(DecodingError::CodeBlockDecodeFailure)?; + let (slot, tail) = tail + .split_at_mut_checked(len) + .ok_or(DecodingError::CodeBlockDecodeFailure)?; + owned[index] = Some(slot); + ring_samples = tail; + offset = start + len; + } + } else { + let slot_len = range + .len() + .checked_div(ring.slots) + .ok_or(DecodingError::CodeBlockDecodeFailure)?; + for _ in 0..ring.slots { + let (slot, tail) = core::mem::take(&mut ring_samples).split_at_mut(slot_len); + ring_slots.push(slot); + ring_samples = tail; + } + } + } + slots.push(ring_slots); + } + Ok((slots, owned)) +} + +#[derive(Clone, Copy, PartialEq, Eq)] +enum Status { + /// Not started; ready to start once its ring has a free slot. + Waiting, + /// Being decoded. + Running, + /// Decoded, holding its slot until the IDWT has read it. + Done, + /// Read, its slot back in its ring. + Freed, +} + +/// What the IDWT and the workers share, behind one lock. +struct State<'c, 'w, W> { + status: Vec, + /// The slot of each band that holds one and isn't in a thread's hands. + held: Vec>, + free_slots: Vec>, + /// The rows of the window each band of a whole-window ring decodes into, + /// until it starts. + owned: Vec>, + /// Bands that can start, soonest read first, then largest first. + /// Entries for bands since started are skipped. + ready: BinaryHeap, usize)>>, + /// Each ring's next band to make ready. + next_ready: Vec, + /// How many of each ring's bands, from its first, have been freed. + freed_prefix: Vec, + /// Bands that finished decoding, for the drain to free; bands the IDWT + /// has since read are skipped. + decoded: Vec, + running: usize, + freed: usize, + workspaces: Vec<&'w mut W>, + /// Workers on the pool. + active: usize, + error: Option, + /// Whether a wait found a running band with no worker active. + stranded: bool, +} + +struct Shared<'p, 'c, 'w, W> { + rings: &'p [Ring], + bands: &'p [Band], + max_workers: usize, + /// Test-only: the first worker to start a band abandons it. + #[cfg(test)] + strand_once: core::sync::atomic::AtomicBool, + state: Mutex>, + /// Signalled when a band is decoded or a decode fails. + decoded: Condvar, +} + +impl<'c, 'w, W> Shared<'_, 'c, 'w, W> { + fn lock(&self) -> Result>> { + self.state + .lock() + .map_err(|_| DecodingError::CodeBlockDecodeFailure.into()) + } + + /// Make ring `ring`'s next bands ready while it has free slots. + fn refill(&self, state: &mut State<'c, 'w, W>, ring: usize) { + let ring_bands = &self.rings[ring].bands; + while state.next_ready[ring] < ring_bands.end + && state.next_ready[ring] - ring_bands.start + < state.freed_prefix[ring] + self.rings[ring].slots + { + let index = state.next_ready[ring]; + let band = &self.bands[index]; + state + .ready + .push(Reverse((band.order, Reverse(band.work), index))); + state.next_ready[ring] += 1; + } + } + + /// Take the next ready band, marked running, with a slot to decode into. + fn start_next(&self, state: &mut State<'c, 'w, W>) -> Result> { + while let Some(Reverse((_, _, index))) = state.ready.pop() { + if state.status[index] == Status::Waiting { + return self.start(state, index).map(Some); + } + } + Ok(None) + } + + fn start(&self, state: &mut State<'c, 'w, W>, index: usize) -> Result<(usize, &'c mut [f32])> { + let slot = match state.owned[index].take() { + Some(slot) => slot, + None => state.free_slots[self.bands[index].ring] + .pop() + .ok_or(DecodingError::CodeBlockDecodeFailure)?, + }; + state.status[index] = Status::Running; + state.running += 1; + Ok((index, slot)) + } + + /// Band `index`, running, has decoded. + fn decoded(state: &mut State<'c, 'w, W>, index: usize) { + state.status[index] = Status::Done; + state.running -= 1; + state.decoded.push(index); + } + + /// Record band `index`'s decode, which its decoder ran outside the lock. + fn finish(&self, index: usize, slot: &'c mut [f32], result: Result<()>) -> Result<()> { + let mut state = self.lock()?; + match result { + Ok(()) => { + state.held[index] = Some(slot); + Self::decoded(&mut state, index); + } + Err(error) => { + state.error.get_or_insert(error); + } + } + drop(state); + self.decoded.notify_all(); + Ok(()) + } + + /// Return band `index`'s slot to its ring and make the ring's next band + /// ready. + fn free(&self, state: &mut State<'c, 'w, W>, index: usize, slot: &'c mut [f32]) { + let ring = self.bands[index].ring; + // A whole-window ring's slots belong to their bands. + if !self.rings[ring].whole_window { + state.free_slots[ring].push(slot); + } + state.status[index] = Status::Freed; + state.freed += 1; + let ring_bands = &self.rings[ring].bands; + while ring_bands.start + state.freed_prefix[ring] < ring_bands.end + && state.status[ring_bands.start + state.freed_prefix[ring]] == Status::Freed + { + state.freed_prefix[ring] += 1; + } + self.refill(state, ring); + } +} + +/// Decode `bands` into `slots` on up to `route.workers()` pool threads +/// while `run` drives the IDWT through them on this thread, then decode the +/// bands it didn't read. Returns `false` when a band was stranded. +#[expect( + clippy::too_many_arguments, + reason = "the executor takes the plan, its storage and its two drivers explicitly" +)] +fn run_bands( + route: &BandRoute<'_>, + rings: &[Ring], + bands: &[Band], + slots: Slots<'_>, + bank: &mut [TaskWorkspace], + decode: impl Fn(usize, &mut [f32], &mut W, &mut Vec) -> Result<()> + Sync, + run: impl FnOnce(&mut dyn SubBandRows) -> Result<()>, + profile_enabled: bool, + tier1_us: &mut u128, + strand_once: bool, +) -> Result { + #[cfg(not(test))] + let _ = strand_once; + let workspaces_needed = route.workers() + 1; + let mut workspaces: Vec<&mut W> = Vec::new(); + workspaces + .try_reserve_exact(workspaces_needed) + .map_err(|_| DecodingError::HostAllocationFailed)?; + workspaces.extend( + bank.iter_mut() + .take(workspaces_needed) + .map(|slot| &mut slot.workspace), + ); + if workspaces.len() != workspaces_needed { + return Err(DecodingError::CodeBlockDecodeFailure.into()); + } + let own_workspace = workspaces + .pop() + .ok_or(DecodingError::CodeBlockDecodeFailure)?; + let mut state = State { + status: try_filled(bands.len(), Status::Waiting)?, + held: try_filled_with(bands.len(), || None)?, + free_slots: slots.0, + owned: slots.1, + ready: BinaryHeap::new(), + next_ready: try_collect(rings.iter().map(|ring| ring.bands.start), rings.len())?, + freed_prefix: try_filled(rings.len(), 0)?, + decoded: Vec::new(), + running: 0, + freed: 0, + workspaces, + active: 0, + error: None, + stranded: false, + }; + state + .ready + .try_reserve(bands.len()) + .map_err(|_| DecodingError::HostAllocationFailed)?; + state + .decoded + .try_reserve_exact(bands.len()) + .map_err(|_| DecodingError::HostAllocationFailed)?; + let shared = Shared { + rings, + bands, + max_workers: route.workers(), + #[cfg(test)] + strand_once: core::sync::atomic::AtomicBool::new(strand_once), + state: Mutex::new(state), + decoded: Condvar::new(), + }; + { + let mut state = shared.lock()?; + for ring in 0..rings.len() { + shared.refill(&mut state, ring); + } + } + let decode = &decode; + let shared = &shared; + let scoped = rayon::in_place_scope(|scope| -> Result<()> { + let mut rows = PoolRows { + shared, + scope, + decode, + workspace: own_workspace, + clipped_scratch: Vec::new(), + current: try_filled_with(rings.len(), || None)?, + next: try_collect(rings.iter().map(|ring| ring.bands.start), rings.len())?, + profile_enabled, + tier1_us: 0, + }; + rows.send_workers()?; + let result = run(&mut rows).and_then(|()| rows.drain()); + *tier1_us = rows.tier1_us; + if result.is_err() { + // Let the workers stop at their next band. + if let Ok(mut state) = shared.lock() { + state.ready.clear(); + } + } + result + }); + let mut state = shared.lock()?; + if state.stranded { + return Ok(false); + } + let error = state.error.take(); + drop(state); + scoped?; + error.map_or(Ok(true), Err) +} + +fn try_filled(len: usize, value: T) -> Result> { + let mut values = Vec::new(); + values + .try_reserve_exact(len) + .map_err(|_| DecodingError::HostAllocationFailed)?; + values.resize(len, value); + Ok(values) +} + +fn try_collect(values: impl Iterator, len: usize) -> Result> { + let mut collected = Vec::new(); + collected + .try_reserve_exact(len) + .map_err(|_| DecodingError::HostAllocationFailed)?; + collected.extend(values); + Ok(collected) +} + +fn try_filled_with(len: usize, value: impl FnMut() -> T) -> Result> { + let mut values = Vec::new(); + values + .try_reserve_exact(len) + .map_err(|_| DecodingError::HostAllocationFailed)?; + values.extend(core::iter::repeat_with(value).take(len)); + Ok(values) +} + +/// A pool worker: decode ready bands until none is left to start. +fn work( + shared: &Shared<'_, '_, '_, W>, + decode: &(impl Fn(usize, &mut [f32], &mut W, &mut Vec) -> Result<()> + Sync), +) { + let Ok(mut state) = shared.lock() else { + return; + }; + let Some(workspace) = state.workspaces.pop() else { + state.active -= 1; + return; + }; + let mut clipped_scratch = Vec::new(); + loop { + let next = if state.error.is_some() { + Ok(None) + } else { + shared.start_next(&mut state) + }; + let (index, slot) = match next { + Ok(Some(started)) => started, + Ok(None) => break, + Err(error) => { + state.error.get_or_insert(error); + break; + } + }; + drop(state); + #[cfg(test)] + if shared + .strand_once + .swap(false, core::sync::atomic::Ordering::Relaxed) + { + // Leave the band running, as the stranding bug would. + if let Ok(mut state) = shared.lock() { + state.active -= 1; + } + return; + } + slot.fill(0.0); + let result = decode( + shared.bands[index].strip, + slot, + workspace, + &mut clipped_scratch, + ); + if shared.finish(index, slot, result).is_err() { + return; + } + state = match shared.lock() { + Ok(state) => state, + Err(_) => return, + }; + } + state.workspaces.push(workspace); + state.active -= 1; + drop(state); + shared.decoded.notify_all(); +} + +/// The IDWT's view of the bands: each sub-band's current band, which it +/// holds while it reads it. +struct PoolRows<'s, 'p, 'c, 'w, W, D> { + shared: &'p Shared<'p, 'c, 'w, W>, + scope: &'s rayon::Scope<'p>, + decode: &'p D, + workspace: &'w mut W, + clipped_scratch: Vec, + /// Each ring's band being read, with its slot. + current: Vec>, + /// Each ring's next band to read. + next: Vec, + profile_enabled: bool, + tier1_us: u128, +} + +impl<'p, 'c, W, D> PoolRows<'_, 'p, 'c, '_, W, D> +where + W: TaskScratch + 'p, + D: Fn(usize, &mut [f32], &mut W, &mut Vec) -> Result<()> + Sync, + 'c: 'p, +{ + /// Send pool workers, up to the plan's count, while bands are ready. + fn send_workers(&self) -> Result<()> { + let sent = { + let mut state = self.shared.lock()?; + let wanted = self.shared.max_workers.min(state.ready.len()); + let sent = wanted.saturating_sub(state.active); + state.active += sent; + sent + }; + for _ in 0..sent { + let (shared, decode) = (self.shared, self.decode); + self.scope.spawn(move |_| work(shared, decode)); + } + Ok(()) + } + + fn ring(&self, sub_band: usize) -> Result { + self.shared + .rings + .iter() + .position(|ring| ring.sub_band == sub_band) + .ok_or(DecodingError::CodeBlockDecodeFailure.into()) + } + + /// Hand back the band ring `ring` is reading. + fn release(&mut self, ring: usize) -> Result<()> { + if let Some((index, slot)) = self.current[ring].take() { + let mut state = self.shared.lock()?; + self.shared.free(&mut state, index, slot); + drop(state); + self.send_workers()?; + } + Ok(()) + } + + /// Band `index`, decoded, with its slot: wait for a worker decoding it, + /// or decode it here if nobody has started it. + fn acquire(&mut self, index: usize) -> Result<&'c mut [f32]> { + let mut state = self.shared.lock()?; + loop { + if let Some(error) = state.error.take() { + return Err(error); + } + match state.status[index] { + Status::Done => { + return state.held[index] + .take() + .ok_or(DecodingError::CodeBlockDecodeFailure.into()); + } + Status::Running => { + state = self + .shared + .decoded + .wait_timeout(state, STRANDED_CHECK) + .map_err(|_| DecodingError::CodeBlockDecodeFailure)? + .0; + // A failed band stays running; its error ends the wait. + if state.error.is_none() + && state.status[index] == Status::Running + && state.active == 0 + { + state.stranded = true; + return Err(DecodingError::CodeBlockDecodeFailure.into()); + } + } + Status::Waiting => { + let (index, slot) = self.shared.start(&mut state, index)?; + drop(state); + self.decode_here(index, slot)?; + return Ok(slot); + } + Status::Freed => return Err(DecodingError::CodeBlockDecodeFailure.into()), + } + } + } + + fn decode_here(&mut self, index: usize, slot: &mut [f32]) -> Result<()> { + let start = profile::profile_now(self.profile_enabled); + slot.fill(0.0); + let result = (self.decode)( + self.shared.bands[index].strip, + slot, + self.workspace, + &mut self.clipped_scratch, + ); + self.tier1_us += profile::elapsed_us(start); + result?; + let mut state = self.shared.lock()?; + Shared::decoded(&mut state, index); + drop(state); + Ok(()) + } + + /// After the IDWT, free its bands and decode every band it didn't read, + /// here and on the pool. + fn drain(&mut self) -> Result<()> { + for ring in 0..self.current.len() { + self.release(ring)?; + } + loop { + let mut state = self.shared.lock()?; + if let Some(error) = state.error.take() { + return Err(error); + } + // Free every decoded band, which lets its ring's next band start. + while let Some(index) = state.decoded.pop() { + if state.status[index] == Status::Done { + let slot = state.held[index] + .take() + .ok_or(DecodingError::CodeBlockDecodeFailure)?; + self.shared.free(&mut state, index, slot); + } + } + if state.freed == self.shared.bands.len() { + return Ok(()); + } + let next = self.shared.start_next(&mut state)?; + let running = state.running > 0; + match next { + Some((index, slot)) => { + drop(state); + self.send_workers()?; + self.decode_here(index, slot)?; + self.shared.lock()?.held[index] = Some(slot); + } + None if running => { + let mut state = self + .shared + .decoded + .wait_timeout(state, STRANDED_CHECK) + .map_err(|_| DecodingError::CodeBlockDecodeFailure)? + .0; + let stranded = state.error.is_none() && state.active == 0 && state.running > 0; + state.stranded |= stranded; + drop(state); + if stranded { + return Err(DecodingError::CodeBlockDecodeFailure.into()); + } + } + None => { + drop(state); + // Bands left that no ring can start. + return Err(DecodingError::CodeBlockDecodeFailure.into()); + } + } + } + } +} + +impl<'p, 'c, W, D> SubBandRows for PoolRows<'_, 'p, 'c, '_, W, D> +where + W: TaskScratch + 'p, + D: Fn(usize, &mut [f32], &mut W, &mut Vec) -> Result<()> + Sync, + 'c: 'p, +{ + fn prepare_row(&mut self, sub_band: usize, y: u32) -> Result<()> { + let ring = self.ring(sub_band)?; + let ring_bands = self.shared.rings[ring].bands.clone(); + let stored = self.shared.rings[ring].stored; + if y < stored.y0 || y >= stored.y1 { + // Outside the stored rows the sub-band reads as zero. + return Ok(()); + } + loop { + if let Some((index, _)) = self.current[ring] { + let rows = &self.shared.bands[index].rows; + if y < rows.start { + // Rows are read in order, so a band already left can't be + // read again. + return Err(DecodingError::CodeBlockDecodeFailure.into()); + } + if y < rows.end { + return Ok(()); + } + self.release(ring)?; + } + let index = self.next[ring]; + if index >= ring_bands.end || self.shared.bands[index].rows.start > y { + // No code-block covers the row, so it reads as zero. + return Ok(()); + } + let slot = self.acquire(index)?; + self.current[ring] = Some((index, slot)); + self.next[ring] = index + 1; + } + } + + fn samples(&self, sub_band: usize) -> (&[f32], IntRect, u32) { + let Ok(ring) = self.ring(sub_band) else { + return (&[], IntRect::from_xywh(0, 0, 0, 0), 0); + }; + let stored = self.shared.rings[ring].stored; + match &self.current[ring] { + Some((index, slot)) => { + let rows = &self.shared.bands[*index].rows; + ( + slot, + IntRect::from_ltrb(stored.x0, rows.start, stored.x1, rows.end), + stored.width(), + ) + } + None => (&[], IntRect::from_xywh(0, 0, 0, 0), stored.width()), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::j2c::decode::{DecodeAllocationBudget, DecoderContext, WorkspacePeak}; + use crate::j2c::tile; + use crate::reader::BitReader; + use crate::{encode, DecodeSettings, EncodeOptions, Image, DEFAULT_MAX_DECODE_BYTES}; + + fn budget_fixture(width: u32, height: u32) -> Vec { + let mut state = 0x2545_f491_u32; + let pixels: Vec = (0..width * height) + .map(|_| { + state ^= state << 13; + state ^= state >> 17; + state ^= state << 5; + state.to_le_bytes()[3] + }) + .collect(); + encode( + &pixels, + width, + height, + 1, + 8, + false, + &EncodeOptions { + reversible: true, + num_decomposition_levels: 4, + code_block_width_exp: 0, + code_block_height_exp: 0, + ..EncodeOptions::default() + }, + ) + .expect("encode budget fixture") + } + + /// The route's plan and clipped scratch stay live beside the executor. + /// Leave enough simulated headroom for either alone, but not both. No + /// large retained allocation is needed to exercise the cap boundary. + #[test] + fn pool_bands_respect_the_combined_plan_and_executor_budget() { + let (width, height) = (203, 157); + let bytes = budget_fixture(width, height); + let image = Image::new(&bytes, &DecodeSettings::default()).expect("parse budget fixture"); + rayon::ThreadPoolBuilder::new() + .num_threads(2) + .build() + .expect("build budget test pool") + .install(|| { + let mut context = DecoderContext::default(); + context.set_output_region(Some((0, 0, width, height))); + context.tile_decode_context.line_region_idwt = true; + let tiles = tile::parse(&mut BitReader::new(image.codestream), &image.header, 0) + .expect("parse tile"); + let baseline = context + .reset( + &image.header, + &tiles[0], + tiles.structural_workspace_bytes(), + 0, + false, + None, + ) + .expect("prepare output planes"); + let (tile_ctx, storage) = (&mut context.tile_decode_context, &mut context.storage); + // Stop at the tile boundary so its parsed block storage stays + // available. This also prepares the retained worker bank, so + // the admission check below needs no replacement bank. + super::super::decode_tile( + &tiles[0], + &image.header, + tile_ctx, + storage, + &mut None, + CpuDecodeParallelism::Auto, + false, + &mut DecodeProfileTimings::default(), + baseline, + super::super::TileOutput::Planes, + ) + .expect("decode with ample headroom"); + assert_eq!(tile_ctx.debug_counters.pool_band_components, 1); + let (route_bytes, executor) = { + let BandRouting::Route(route) = + plan_band_route(&tiles[0], 0, tile_ctx, storage, &image.header) + .expect("plan control route") + else { + panic!("control fixture must take the pool route"); + }; + // Reuse the existing peak accounting for the route's owned + // tables and clipped scratch, excluding the retained bank. + let retained_tier1 = tile_ctx.tier1_capacity_bytes().expect("bank bytes"); + tile_ctx.debug_counters.workspace_peak = WorkspacePeak::default(); + route + .record(tile_ctx, storage) + .expect("record route owners"); + let route_bytes = + tile_ctx.debug_counters.workspace_peak.bytes().tier1 - retained_tier1; + let (rings, bands, _) = + plan_rings(&route, storage, &image.header, 0).expect("plan control rings"); + ( + route_bytes, + executor_bytes(&rings, &bands).expect("executor bytes"), + ) + }; + let headroom = route_bytes + executor / 2; + assert!(headroom >= executor, "each owner must fit separately"); + let baseline = DecodeAllocationBudget::for_storage(storage) + .expect("control budget") + .live_bytes(); + storage.structural_workspace_bytes += + DEFAULT_MAX_DECODE_BYTES - baseline - headroom; + let route = match plan_band_route(&tiles[0], 0, tile_ctx, storage, &image.header) { + Ok(BandRouting::Route(route)) => route, + Ok(BandRouting::Serial | BandRouting::WholeWindows) => return, + Err(error) if super::super::reuse::is_capacity_error(&error) => return, + Err(error) => panic!("unexpected planning error: {error:?}"), + }; + let result = decode_route( + &route, + 0, + &image.header, + tile_ctx, + storage, + tiles[0].component_infos[0].wavelet_transform(), + false, + &mut 0, + ); + match result { + Err(error) if super::super::reuse::is_capacity_error(&error) => {} + Ok(false) => {} + other => panic!( + "pool route must decline or reject combined live owners: \ + headroom={headroom}, route={route_bytes}, executor={executor}, \ + excess={}, result={other:?}", + route_bytes + executor - headroom, + ), + } + }); + } +} diff --git a/crates/j2k-native/src/j2c/decode/subband.rs b/crates/j2k-native/src/j2c/decode/subband.rs index e3468d5e..a8405a23 100644 --- a/crates/j2k-native/src/j2c/decode/subband.rs +++ b/crates/j2k-native/src/j2c/decode/subband.rs @@ -40,6 +40,12 @@ use self::parallel::{ use self::pending::{collect_pending_classic_blocks, collect_pending_ht_blocks}; pub(in crate::j2c::decode) use self::pending::{count_classic_code_blocks, count_ht_code_blocks}; pub(crate) use self::placement::BlockPlacement; +#[cfg(all(test, feature = "parallel"))] +pub(crate) use self::tile_parallel::MIN_BAND_WORK; +#[cfg(feature = "parallel")] +pub(in crate::j2c::decode) use self::tile_parallel::{ + plan_band_route, routes_tile, BandRoute, BandRouting, +}; /// Decode the code-blocks of the tile's `components` into their sub-bands. #[expect( diff --git a/crates/j2k-native/src/j2c/decode/subband/tile_parallel.rs b/crates/j2k-native/src/j2c/decode/subband/tile_parallel.rs index 0051f645..cfb892bc 100644 --- a/crates/j2k-native/src/j2c/decode/subband/tile_parallel.rs +++ b/crates/j2k-native/src/j2c/decode/subband/tile_parallel.rs @@ -12,6 +12,7 @@ //! segments are gathered. Classic blocks take this route when each decodes as //! one arithmetic segment: no selective bypass or per-pass termination. +use alloc::boxed::Box; use alloc::vec::Vec; use core::ops::Range; use core::sync::atomic::{AtomicBool, Ordering}; @@ -52,6 +53,15 @@ const MIN_PARALLEL_CODED_BYTES: usize = 8 * 1024; #[cfg(test)] const MIN_PARALLEL_CODED_BYTES: usize = 0; +/// Below this average work per band, as the route weighs strips (payload +/// bytes plus an eighth of the samples), scheduling each band on the pool +/// costs more than overlapping Tier-1 with the IDWT saves, and a component's +/// bands decode over whole windows on the route instead. A tall region of +/// 16x4 HT blocks, at 88 per band, decoded 3.6 times slower in bands than +/// over whole windows at 4 threads; 1024 px regions of the benchmark images +/// average 2,000 to 15,000. +pub(crate) const MIN_BAND_WORK: usize = 1024; + /// Where a tile of `coding` blocks with `strips` rows of code blocks and /// `coded_bytes` of payload decodes on `threads` workers: on the route /// (`None`), serially, or per sub-band. A classic tile with fewer rows than @@ -148,6 +158,8 @@ struct TilePlan<'a> { /// Samples in the largest block its sub-band stores only part of. Each /// worker decodes such blocks into scratch of this size. max_clipped_samples: usize, + /// The sub-band of each band, in the order of `bands` or `classic_bands`. + band_sub_bands: Vec, } /// How [`plan_tile_route`] wants the tile's code blocks decoded. @@ -214,6 +226,7 @@ pub(in crate::j2c::decode) fn plan_tile_route<'a>( &mut budget, coding, required, + false, ) { Ok(Some(plan)) => plan, Ok(None) => return Ok(TileRoute::PerSubBand), @@ -372,6 +385,7 @@ fn record_route_peak( plan.classic_bands.capacity() * size_of::(), plan.jobs.capacity() * size_of::>(), plan.strips.capacity() * size_of::(), + plan.band_sub_bands.capacity() * size_of::(), plan.arena.capacity(), task_capacity * size_of::>(), workers @@ -424,6 +438,12 @@ fn decode_with_bank<'a, W: TaskScratch>( Ok(Some(growths)) } +/// Whether the route can decode the code-blocks of every component of +/// `tile`. +pub(in crate::j2c::decode) fn routes_tile(tile: &Tile<'_>) -> bool { + tile_route_coding(tile, 0..tile.component_infos.len()).is_some() +} + /// The block coding of every component in `components` when the route can /// decode it: HT only, or classic with one arithmetic segment per block. fn tile_route_coding(tile: &Tile<'_>, components: Range) -> Option { @@ -448,6 +468,7 @@ fn tile_route_coding(tile: &Tile<'_>, components: Range) -> Option( @@ -458,6 +479,7 @@ fn plan_tile<'a>( budget: &mut DecodeAllocationBudget, coding: CodeBlockCoding, required: usize, + banded: bool, ) -> Result>> { let mut plan = TilePlan { coding, @@ -473,10 +495,12 @@ fn plan_tile<'a>( max_width: 0, max_height: 0, max_clipped_samples: 0, + band_sub_bands: Vec::new(), }; // A region decode may need only a few of the tile's code blocks. budget.reserve_new(&mut plan.jobs, required)?; budget.reserve_new(&mut plan.strips, required)?; + budget.reserve_new(&mut plan.band_sub_bands, storage.sub_bands.len())?; // Classic job metadata must not enlarge the HT plan's band entries. if coding == CodeBlockCoding::Classic { budget.reserve_new(&mut plan.classic_bands, storage.sub_bands.len())?; @@ -537,6 +561,7 @@ fn plan_tile<'a>( band }; + plan.band_sub_bands.push(sub_band_idx); let first_job = plan.jobs.len(); for precinct in sub_band .precincts @@ -586,20 +611,27 @@ fn plan_tile<'a>( plan.jobs.push(job); } } - if !plan_band_strips( - &mut plan, - first_job, - sub_band.coefficients.clone(), + let (width, height) = ( sub_band.storage_rect.width(), sub_band.storage_rect.height(), - )? { + ); + // Banded strips address rows of the sub-band's stored + // rectangle; the executor gives each band its own slot. + let coefficients = if banded { + 0..(width as usize) + .checked_mul(height as usize) + .ok_or(ValidationError::ImageTooLarge)? + } else { + sub_band.coefficients.clone() + }; + if !plan_band_strips(&mut plan, first_job, coefficients, width, height)? { return Ok(None); } } } } - if !strips_are_disjoint(&mut plan.strips, storage.coefficients.len()) { + if !banded && !strips_are_disjoint(&mut plan.strips, storage.coefficients.len()) { return Ok(None); } plan.arena_len = arena_len; @@ -1184,6 +1216,243 @@ fn job_payload<'p>(plan: &'p TilePlan<'_>, job: &BlockJob<'p>) -> Result<&'p [u8 } } +/// One code-block row of a sub-band, as [`plan_band_route`] plans it: the +/// band's rows, counted from the top of the sub-band's stored rectangle. +#[derive(Clone, Copy)] +pub(in crate::j2c::decode) struct BandStrip { + pub(in crate::j2c::decode) sub_band: usize, + pub(in crate::j2c::decode) first_row: usize, + pub(in crate::j2c::decode) rows: usize, + /// The strip's estimated decode cost, as the route weighs it. + pub(in crate::j2c::decode) work: usize, +} + +/// A component's code-block rows planned for decoding band by band on the +/// worker pool, each into a slot of its own, with a workspace per worker in +/// the retained bank of the plan's block coding. +pub(in crate::j2c::decode) struct BandRoute<'a> { + plan: TilePlan<'a>, + workers: usize, + growths: usize, + /// The decode's budget with the plan, the payload arena and every + /// thread's clipped-block scratch charged, which stay live while the + /// bands decode. + budget: DecodeAllocationBudget, +} + +/// How [`plan_band_route`] wants a component's bands decoded. +pub(in crate::j2c::decode) enum BandRouting<'a> { + /// On the worker pool, band by band. + Route(Box>), + /// Too little work to fork: decode the bands serially. + Serial, + /// Bands the route can't decode, or too few rows for the workers: + /// decode the whole windows the old way. + WholeWindows, +} + +/// Plan component `component_idx` of a tile whose sub-bands hold rings of +/// bands, with the route's checks and cutoffs: its block coding, its coded +/// bytes and its rows of code-blocks. With a plan, the coding's retained bank +/// holds a workspace for each of `workers` pool threads and one more for the +/// calling thread, charged to the structural workspace. +pub(in crate::j2c::decode) fn plan_band_route<'a>( + tile: &Tile<'a>, + component_idx: usize, + tile_ctx: &mut TileDecodeContext, + storage: &mut DecompositionStorage<'a>, + header: &Header<'_>, +) -> Result> { + let components = component_idx..component_idx + 1; + let Some(coding) = tile_route_coding(tile, components.clone()) else { + return Ok(BandRouting::WholeWindows); + }; + let threads = rayon::current_num_threads(); + let declined = |route| match route { + TileRoute::Serial => BandRouting::Serial, + _ => BandRouting::WholeWindows, + }; + let (required, segment_bytes, rows) = + required_code_blocks(tile, components.clone(), storage, header, coding)?; + if let Some(route) = route_fallback( + coding, + rows, + segment_bytes, + threads, + MIN_PARALLEL_CODED_BYTES, + ) { + return Ok(declined(route)); + } + let mut budget = DecodeAllocationBudget::for_storage(storage)?; + let mut plan = match plan_tile( + tile, + components, + storage, + header, + &mut budget, + coding, + required, + true, + ) { + Ok(Some(plan)) => plan, + Ok(None) => return Ok(BandRouting::WholeWindows), + // The serial bands need less and report their own limits. + Err(error) if crate::j2c::decode::reuse::is_capacity_error(&error) => { + return Ok(BandRouting::Serial); + } + Err(error) => return Err(error), + }; + if let Some(route) = route_fallback( + coding, + plan.strips.len(), + plan.coded_bytes, + threads, + MIN_PARALLEL_CODED_BYTES, + ) { + return Ok(declined(route)); + } + let work = plan + .strips + .iter() + .fold(0usize, |work, strip| work.saturating_add(strip.work)); + // Unit tests use tiny fixtures and take the pool unless a test sets the + // cutoff on the decode's context. + #[cfg(test)] + let min_band_work = tile_ctx.band_work_cutoff; + #[cfg(not(test))] + let min_band_work = MIN_BAND_WORK; + if work / plan.strips.len().max(1) < min_band_work { + return Ok(BandRouting::WholeWindows); + } + let workers = threads.min(plan.strips.len()); + let prepared = (|| -> Result { + gather_payloads(&mut plan, storage, &mut budget)?; + // Each thread allocates its clipped-block scratch on first use. + budget.include_elements::( + (workers + 1) + .checked_mul(plan.max_clipped_samples) + .ok_or(ValidationError::ImageTooLarge)?, + )?; + let (width, height) = (plan.max_width, plan.max_height); + let structural = &mut storage.structural_workspace_bytes; + match coding { + CodeBlockCoding::Classic => prepare_uniform_task_workspaces( + &mut tile_ctx.classic_task_workspaces, + workers + 1, + width, + height, + structural, + &mut budget, + ), + CodeBlockCoding::HighThroughput => prepare_uniform_task_workspaces( + &mut tile_ctx.ht_task_workspaces, + workers + 1, + width, + height, + structural, + &mut budget, + ), + } + })(); + match prepared { + Ok(growths) => Ok(BandRouting::Route(Box::new(BandRoute { + plan, + workers, + growths, + budget, + }))), + Err(error) if crate::j2c::decode::reuse::is_capacity_error(&error) => { + Ok(BandRouting::Serial) + } + Err(error) => Err(error), + } +} + +impl BandRoute<'_> { + pub(in crate::j2c::decode) fn coding(&self) -> CodeBlockCoding { + self.plan.coding + } + + /// The decode's budget with the route's own allocations charged; the + /// executor charges what it adds to a copy. + pub(in crate::j2c::decode) fn budget(&self) -> DecodeAllocationBudget { + self.budget + } + + /// Pool threads to decode on, besides the calling thread. + pub(in crate::j2c::decode) fn workers(&self) -> usize { + self.workers + } + + pub(in crate::j2c::decode) fn strips(&self) -> impl Iterator + '_ { + self.plan.strips.iter().map(|strip| { + let band = self.plan.jobs[strip.jobs.start].band; + BandStrip { + sub_band: self.plan.band_sub_bands[band], + first_row: strip.start / strip.stride, + rows: strip.len / strip.stride, + work: strip.work, + } + }) + } + + /// Decode strip `index` of a classic plan into `output`, which holds the + /// band's rows from its first. + pub(in crate::j2c::decode) fn decode_classic( + &self, + index: usize, + output: &mut [f32], + workspace: &mut J2kCodeBlockDecodeWorkspace, + clipped_scratch: &mut Vec, + ) -> Result<()> { + let plan = &self.plan; + decode_strip( + plan, + &plan.strips[index], + output, + workspace, + clipped_scratch, + &|job, output, stride, workspace| { + decode_classic_strip_block(plan, job, output, stride, workspace) + }, + ) + } + + /// Decode strip `index` of an HT plan into `output`, which holds the + /// band's rows from its first. + pub(in crate::j2c::decode) fn decode_ht( + &self, + index: usize, + output: &mut [f32], + workspace: &mut HtCodeBlockDecodeWorkspace, + clipped_scratch: &mut Vec, + ) -> Result<()> { + let plan = &self.plan; + decode_strip( + plan, + &plan.strips[index], + output, + workspace, + clipped_scratch, + &|job, output, stride, workspace| { + decode_strip_block(plan, job, output, stride, workspace) + }, + ) + } + + /// Count the plan's blocks and record the decode's peak with the plan + /// and every thread's clipped-block scratch live. + pub(in crate::j2c::decode) fn record( + &self, + tile_ctx: &mut TileDecodeContext, + storage: &DecompositionStorage<'_>, + ) -> Result<()> { + record_route_peak(tile_ctx, storage, &self.plan, self.workers + 1, 0)?; + count_decoded_blocks(tile_ctx, &self.plan, self.workers + 1, self.growths); + Ok(()) + } +} + #[cfg(test)] mod tests { use super::{route_fallback, CodeBlockCoding, TileRoute}; diff --git a/crates/j2k-native/src/j2c/mod.rs b/crates/j2k-native/src/j2c/mod.rs index 50693890..5f8c71d8 100644 --- a/crates/j2k-native/src/j2c/mod.rs +++ b/crates/j2k-native/src/j2c/mod.rs @@ -8,6 +8,8 @@ pub(crate) mod codestream; pub(crate) mod codestream_write; pub(crate) mod coefficient_view; mod decode; +#[cfg(all(test, feature = "parallel"))] +pub(crate) use decode::MIN_BAND_WORK; pub(crate) mod encode; pub(crate) mod fdwt; pub(crate) mod forward_mct; diff --git a/crates/j2k-native/src/tests.rs b/crates/j2k-native/src/tests.rs index 6136f679..b1e7c0ee 100644 --- a/crates/j2k-native/src/tests.rs +++ b/crates/j2k-native/src/tests.rs @@ -12,6 +12,8 @@ mod idwt_window_storage; mod line_region_idwt; #[cfg(feature = "std")] mod packet_directory; +#[cfg(feature = "parallel")] +mod parallel_bands; mod progression_matrix; mod row_sink; mod workspace_peak; @@ -1122,15 +1124,29 @@ fn decode_region_budgets_coefficients_for_the_requested_region() { "tiny ROI must get past the coefficient budget: {tiny:?}" ); - // A region whose coefficients and output together exceed the cap is - // still rejected before any Tier-1 work. Its 324 MB output alone fits. - let Err(large) = image.decode_region((0, 0, 9_000, 9_000)) else { + // A region whose whole-window coefficients and output together exceed + // the cap is still rejected before any Tier-1 work. Its 324 MB output + // alone fits. + let large = (0, 0, 9_000, 9_000); + let mut context = DecoderContext::default(); + context.tile_decode_context.whole_window_tier1 = true; + let Err(whole) = image.decode_region_with_context(large, &mut context) else { panic!("a 9000x9000 region must exceed the coefficient budget"); }; assert_eq!( - large, + whole, DecodeError::Validation(ValidationError::ImageTooLarge) ); + + // Banded Tier-1, serial or on the pool, holds a few bands of each + // sub-band instead, so the same region gets as far as the packets. + let Err(banded) = image.decode_region(large) else { + panic!("rewritten fixture has no packet data for a 65537x65537 tile"); + }; + assert!( + matches!(banded, DecodeError::Decoding(_)), + "banded coefficients must fit the budget: {banded:?}" + ); } /// Declines every code-block, so the CPU decodes everything, but as a diff --git a/crates/j2k-native/src/tests/parallel_bands.rs b/crates/j2k-native/src/tests/parallel_bands.rs new file mode 100644 index 00000000..b74ce4e5 --- /dev/null +++ b/crates/j2k-native/src/tests/parallel_bands.rs @@ -0,0 +1,454 @@ +// SPDX-License-Identifier: MIT OR Apache-2.0 + +//! A parallel region decode on the streamed IDWT decodes its bands on the +//! worker pool ahead of the IDWT. It must match decoding Tier-1 over whole +//! windows bit for bit, decode and skip the same code-blocks, fail the same +//! way, store less, and finish when no pool thread is free to help. + +use alloc::vec::Vec; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::mpsc; + +use super::banded_tier1::{cases, noisy_pixels, HEIGHT, TILE, WIDTH}; +use crate::{encode, CpuDecodeParallelism, DecodeSettings, DecoderContext, EncodeOptions, Image}; + +struct RegionDecode { + native: Vec, + decoded_blocks: usize, + skipped_blocks: usize, + pool_components: usize, + peak_coefficients: usize, +} + +/// Decode `roi` with Auto parallelism on the current pool, in bands on the +/// pool unless `whole_windows`. +fn decode(image: &Image<'_>, roi: (u32, u32, u32, u32), whole_windows: bool) -> RegionDecode { + decode_with_cutoff(image, roi, whole_windows, 0) +} + +/// [`decode`] with bands averaging less than `band_work_cutoff` decoding over +/// whole windows instead of on the pool. +fn decode_with_cutoff( + image: &Image<'_>, + roi: (u32, u32, u32, u32), + whole_windows: bool, + band_work_cutoff: usize, +) -> RegionDecode { + let mut context = DecoderContext::default(); + context.tile_decode_context.band_work_cutoff = band_work_cutoff; + context.set_cpu_decode_parallelism(CpuDecodeParallelism::Auto); + context.tile_decode_context.whole_window_tier1 = whole_windows; + let native = image + .decode_native_region_with_context(roi, &mut context) + .expect("decode region") + .data; + let counters = context.tile_decode_context.debug_counters; + RegionDecode { + native, + decoded_blocks: counters.decoded_code_blocks, + skipped_blocks: counters.skipped_code_blocks, + pool_components: counters.pool_band_components, + peak_coefficients: counters.workspace_peak.bytes().coefficients, + } +} + +fn pool(threads: usize) -> rayon::ThreadPool { + rayon::ThreadPoolBuilder::new() + .num_threads(threads) + .build() + .expect("build pool") +} + +#[test] +fn pool_bands_match_whole_window_tier1_bit_for_bit() { + for threads in [2, 4, 12] { + let pool = pool(threads); + for case in cases() { + let pixels = noisy_pixels(case.components); + let bytes = encode( + &pixels, + WIDTH, + HEIGHT, + u16::from(case.components), + 8, + false, + &case.options, + ) + .expect("encode pool band fixture"); + let mut pool_components = 0; + for reduction in [0_u8, 1, 2] { + let image = + Image::new_with_reduction(&bytes, &DecodeSettings::default(), reduction) + .expect("parse pool band fixture"); + let (width, height) = (image.width(), image.height()); + let tile = TILE >> reduction; + let large = ( + width / 20, + height / 20, + width - width / 10, + height - height / 10, + ); + let regions = [ + (0, 0, width, height), + large, + (width / 3, height / 3, width / 7, height / 9), + (tile + 1, tile + 2, tile - 3, tile - 4), + (tile - 5, tile - 3, 11, 7), + (0, 0, width, 9), + (0, height - 9, width, 9), + (width - 1, height - 1, 1, 1), + ]; + for roi in regions { + let label = format!( + "{} threads={threads} reduction={reduction} roi={roi:?}", + case.label + ); + let (whole, banded) = + pool.install(|| (decode(&image, roi, true), decode(&image, roi, false))); + assert_eq!(whole.pool_components, 0, "{label}"); + assert_eq!(banded.native, whole.native, "{label}"); + assert_eq!(banded.decoded_blocks, whole.decoded_blocks, "{label}"); + assert_eq!(banded.skipped_blocks, whole.skipped_blocks, "{label}"); + pool_components += banded.pool_components; + // A band of short code-blocks is a small part of the + // window. + if banded.pool_components > 0 + && roi == large + && reduction == 0 + && case.options.tile_size.is_none() + && case.options.code_block_height_exp <= 1 + { + assert!( + banded.peak_coefficients < whole.peak_coefficients, + "{label}: coefficients {} vs {}", + banded.peak_coefficients, + whole.peak_coefficients + ); + } + } + } + // The route decodes HT blocks and classic blocks of one + // arithmetic segment. The encoder terminates every pass when it + // writes several layers, so those classic blocks keep whole + // windows. Classic tiles with fewer rows of code-blocks than + // threads do too, so only two threads reach every other case. + let routed = case.options.use_ht_block_coding || case.options.num_layers <= 1; + if !routed { + assert_eq!(pool_components, 0, "{}", case.label); + } else if threads == 2 { + assert!(pool_components > 0, "{} never used the pool", case.label); + } + } + } +} + +fn rgb_fixture() -> Vec { + let pixels = noisy_pixels(3); + encode( + &pixels, + WIDTH, + HEIGHT, + 3, + 8, + false, + &EncodeOptions { + reversible: false, + num_decomposition_levels: 4, + code_block_width_exp: 2, + code_block_height_exp: 2, + ..EncodeOptions::default() + }, + ) + .expect("encode rgb fixture") +} + +/// With the pool's other thread held by unrelated work, workers can't run +/// until the decode is over, so the IDWT must decode every band itself. +#[test] +fn pool_bands_finish_while_the_other_pool_thread_is_busy() { + let bytes = rgb_fixture(); + let image = Image::new(&bytes, &DecodeSettings::default()).expect("parse rgb fixture"); + let roi = (10, 12, 170, 130); + let expected = pool(2).install(|| decode(&image, roi, true)).native; + let pool = pool(2); + let done = AtomicBool::new(false); + let (started, running) = mpsc::channel(); + pool.in_place_scope(|scope| { + scope.spawn(|_| { + started.send(()).expect("signal start"); + while !done.load(Ordering::Acquire) { + std::thread::yield_now(); + } + }); + running.recv().expect("busy task started"); + let banded = pool.install(|| decode(&image, roi, false)); + done.store(true, Ordering::Release); + assert!(banded.pool_components > 0); + assert_eq!(banded.native, expected); + }); +} + +/// Region decodes that run inside the pool, several at a time, each wait +/// only for bands another thread is decoding. +#[test] +fn concurrent_pool_band_decodes_inside_the_pool_finish() { + use rayon::prelude::*; + let bytes = rgb_fixture(); + let image = Image::new(&bytes, &DecodeSettings::default()).expect("parse rgb fixture"); + let regions = [(0, 0, WIDTH, HEIGHT), (10, 12, 170, 130), (60, 40, 90, 70)]; + let expected: Vec> = regions + .iter() + .map(|roi| pool(2).install(|| decode(&image, *roi, true)).native) + .collect(); + for threads in [2, 3] { + let decoded: Vec> = pool(threads).install(|| { + (0..12) + .into_par_iter() + .map(|index| decode(&image, regions[index % regions.len()], false).native) + .collect() + }); + for (index, native) in decoded.iter().enumerate() { + assert_eq!( + native, + &expected[index % regions.len()], + "threads={threads} decode={index}" + ); + } + } +} + +/// Outcome of an Auto region decode on a two-thread pool: the pixels' hash, +/// or the error. +fn outcome(bytes: &[u8], strict: bool, roi: (u32, u32, u32, u32), whole_windows: bool) -> String { + let settings = DecodeSettings { + strict, + ..DecodeSettings::default() + }; + let image = match Image::new(bytes, &settings) { + Ok(image) => image, + Err(error) => return format!("parse {error:?}"), + }; + let mut context = DecoderContext::default(); + context.set_cpu_decode_parallelism(if whole_windows { + CpuDecodeParallelism::Serial + } else { + CpuDecodeParallelism::Auto + }); + context.tile_decode_context.whole_window_tier1 = whole_windows; + match image.decode_region_with_context(roi, &mut context) { + Ok(decoded) => { + let hash = decoded + .data + .iter() + .fold(0xcbf2_9ce4_8422_2325_u64, |hash, &byte| { + (hash ^ u64::from(byte)).wrapping_mul(0x0100_0000_01b3) + }); + format!("ok {hash:016x}") + } + Err(error) => format!("decode {error:?}"), + } +} + +#[test] +fn pool_bands_reject_corrupt_code_blocks_as_serial_whole_window_tier1_does() { + const SIZE: u32 = 48; + let mut state = 0x9e37_79b9_7f4a_7c15_u64; + let mut next = move || { + state ^= state << 13; + state ^= state >> 7; + state ^= state << 17; + state + }; + let pixels: Vec = (0..SIZE * SIZE * 3).map(|_| (next() >> 56) as u8).collect(); + let pool = pool(2); + for use_ht_block_coding in [false, true] { + let bytes = encode( + &pixels, + SIZE, + SIZE, + 3, + 8, + false, + &EncodeOptions { + reversible: true, + num_decomposition_levels: 2, + code_block_width_exp: 0, + code_block_height_exp: 0, + num_layers: 2, + use_ht_block_coding, + ..EncodeOptions::default() + }, + ) + .expect("encode corrupt-block fixture"); + let mut failures = 0; + for iteration in 0..300 { + let mut mutant = bytes.clone(); + for _ in 0..=next() % 4 { + let span = u64::try_from(mutant.len() - 120).expect("fixture length"); + let at = 120 + usize::try_from(next() % span).expect("offset in the fixture"); + mutant[at] ^= 1 << (next() % 8); + } + for strict in [false, true] { + let roi = [(0, 0, SIZE, SIZE), (30, 30, 18, 18), (5, 9, 13, 7)] + [usize::try_from(next() % 3).expect("region index")]; + let whole = outcome(&mutant, strict, roi, true); + let banded = pool.install(|| outcome(&mutant, strict, roi, false)); + assert_eq!( + banded, whole, + "ht={use_ht_block_coding} iteration={iteration} strict={strict} roi={roi:?}" + ); + failures += usize::from(whole.starts_with("decode")); + } + } + assert!( + failures > 20, + "ht={use_ht_block_coding}: {failures} failures" + ); + } +} + +/// A band left running with no worker to finish it, as an earlier build's +/// fuzz once found, must not hang the decode: the component gives up the +/// pool and decodes over whole windows, with the same output. +#[test] +fn a_stranded_band_falls_back_to_whole_windows() { + let bytes = rgb_fixture(); + let image = Image::new(&bytes, &DecodeSettings::default()).expect("parse rgb fixture"); + let roi = (10, 12, 170, 130); + let pool = pool(4); + let expected = pool.install(|| decode(&image, roi, true)).native; + let (native, counters) = pool.install(|| { + let mut context = DecoderContext::default(); + context.set_cpu_decode_parallelism(CpuDecodeParallelism::Auto); + context.tile_decode_context.strand_pool_band = true; + let native = image + .decode_native_region_with_context(roi, &mut context) + .expect("decode with a stranded band") + .data; + (native, context.tile_decode_context.debug_counters) + }); + assert_eq!(native, expected); + assert_eq!(counters.pool_band_fallbacks, 1); + assert_eq!(counters.pool_band_components, 0); +} + +/// Generate tall, narrow HT images to expose per-band scheduling costs. +/// Compare seven interleaved timings after warming both routes, with the +/// same four-thread pool and a fresh decoder context for every decode. +/// There is no timing assertion: the report is for optimized, quiet runs. +#[test] +#[ignore = "timing probe; run optimized with --ignored --nocapture --test-threads=1"] +fn pool_bands_tall_region_timing_report() { + let pool = pool(4); + for height in [8_192_u32, 32_768, 65_536] { + let width = 32; + let mut state = 0x2545_f491_u32; + let pixels: Vec = (0..width * height) + .map(|_| { + state ^= state << 13; + state ^= state >> 17; + state ^= state << 5; + state.to_le_bytes()[3] + }) + .collect(); + let bytes = encode( + &pixels, + width, + height, + 1, + 8, + false, + &EncodeOptions { + reversible: true, + use_ht_block_coding: true, + num_decomposition_levels: 1, + code_block_width_exp: 2, + code_block_height_exp: 0, + ..EncodeOptions::default() + }, + ) + .expect("encode 16x4-block timing fixture"); + let image = Image::new(&bytes, &DecodeSettings::default()).expect("parse timing fixture"); + let roi = (0, 0, width, height); + let mut timings = [Vec::new(), Vec::new()]; + let mut pool_components = 0; + for round in 0..8 { + let order = if round % 2 == 0 { + [true, false] + } else { + [false, true] + }; + for whole_windows in order { + let start = std::time::Instant::now(); + let decoded = pool.install(|| { + decode_with_cutoff(&image, roi, whole_windows, crate::j2c::MIN_BAND_WORK) + }); + let elapsed = start.elapsed(); + assert_eq!( + decoded.native, pixels, + "{width}x{height}, whole={whole_windows}" + ); + if !whole_windows { + pool_components = decoded.pool_components; + } + if round > 0 { + timings[usize::from(!whole_windows)].push(elapsed); + } + } + } + for samples in &mut timings { + samples.sort_unstable(); + } + let whole_ms = timings[0][3].as_secs_f64() * 1_000.0; + let bands_ms = timings[1][3].as_secs_f64() * 1_000.0; + println!( + "{width}x{height}, HT 16x4, 4 threads, median of 7: \ + whole={whole_ms:.3} ms, auto={bands_ms:.3} ms, \ + pool_components={pool_components}, ratio={:.2}", + bands_ms / whole_ms, + ); + } +} + +/// Bands too cheap to schedule one by one, such as rows of 16x4 HT blocks, +/// decode over whole windows on the route instead, with the same output. +#[test] +fn cheap_bands_decode_over_whole_windows() { + let (width, height) = (32, 4096); + let pixels = noisy_pixels(1); + let pixels: Vec = pixels + .iter() + .copied() + .cycle() + .take((width * height) as usize) + .collect(); + let bytes = encode( + &pixels, + width, + height, + 1, + 8, + false, + &EncodeOptions { + reversible: true, + use_ht_block_coding: true, + num_decomposition_levels: 1, + code_block_width_exp: 2, + code_block_height_exp: 0, + ..EncodeOptions::default() + }, + ) + .expect("encode 16x4-block fixture"); + let image = Image::new(&bytes, &DecodeSettings::default()).expect("parse 16x4-block fixture"); + let roi = (0, 0, width, height); + let pool = pool(4); + let unlimited = pool.install(|| decode(&image, roi, false)); + assert!( + unlimited.pool_components > 0, + "without the cutoff the bands take the pool" + ); + let cheap = pool.install(|| decode_with_cutoff(&image, roi, false, crate::j2c::MIN_BAND_WORK)); + assert_eq!(cheap.pool_components, 0); + assert_eq!(cheap.native, pixels); + assert_eq!(cheap.native, unlimited.native); +}