perf(j2k-native): decode region bands on the pool ahead of the streamed IDWT

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

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::<f32>(&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::<f32>(&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<usize> {
     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)?;

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<bool> {
+    // 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,

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<'_>,