j2k_core/
batch.rs

1// SPDX-License-Identifier: Apache-2.0
2
3use alloc::vec::Vec;
4use core::num::NonZeroUsize;
5
6use crate::{scale::Downscale, types::Rect};
7
8/// Worker configuration for CPU tile batches.
9#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
10pub struct TileBatchOptions {
11    /// Worker count. `None` asks the codec crate to use available parallelism.
12    pub workers: Option<NonZeroUsize>,
13}
14
15impl TileBatchOptions {
16    /// Construct tile-batch options with an optional fixed worker count.
17    pub const fn new(workers: Option<NonZeroUsize>) -> Self {
18        Self { workers }
19    }
20}
21
22/// Indexed result produced by one tile-batch worker.
23pub type IndexedBatchResult<T, E> = (usize, Result<T, E>);
24
25/// One full-tile decode request.
26pub struct TileDecodeJob<'i, 'o> {
27    /// Compressed tile bytes.
28    pub input: &'i [u8],
29    /// Caller-owned output buffer for this tile.
30    pub out: &'o mut [u8],
31    /// Distance in bytes between output rows.
32    pub stride: usize,
33}
34
35/// One region tile decode request.
36pub struct TileRegionDecodeJob<'i, 'o> {
37    /// Compressed tile bytes.
38    pub input: &'i [u8],
39    /// Caller-owned output buffer for this tile.
40    pub out: &'o mut [u8],
41    /// Distance in bytes between output rows.
42    pub stride: usize,
43    /// Region of interest in source-image coordinates.
44    pub roi: Rect,
45}
46
47/// One scaled tile decode request.
48pub struct TileScaledDecodeJob<'i, 'o> {
49    /// Compressed tile bytes.
50    pub input: &'i [u8],
51    /// Caller-owned output buffer for this tile.
52    pub out: &'o mut [u8],
53    /// Distance in bytes between output rows.
54    pub stride: usize,
55    /// Downscale factor applied to the full-tile decode.
56    pub scale: Downscale,
57}
58
59/// One region+scaled tile decode request.
60pub struct TileRegionScaledDecodeJob<'i, 'o> {
61    /// Compressed tile bytes.
62    pub input: &'i [u8],
63    /// Caller-owned output buffer for this tile.
64    pub out: &'o mut [u8],
65    /// Distance in bytes between output rows.
66    pub stride: usize,
67    /// Region of interest in source-image coordinates.
68    pub roi: Rect,
69    /// Downscale factor applied to the region decode.
70    pub scale: Downscale,
71}
72
73/// Error returned by tile batches, annotated with the failing input index.
74#[derive(Debug)]
75pub struct TileBatchError<E> {
76    /// Index of the first failing tile in input order.
77    pub index: usize,
78    /// Decode error reported for that tile.
79    pub source: E,
80}
81
82impl<E: core::fmt::Display> core::fmt::Display for TileBatchError<E> {
83    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
84        write!(f, "tile {} decode failed: {}", self.index, self.source)
85    }
86}
87
88impl<E: core::error::Error + 'static> core::error::Error for TileBatchError<E> {
89    fn source(&self) -> Option<&(dyn core::error::Error + 'static)> {
90        Some(&self.source)
91    }
92}
93
94/// Resolve the number of CPU workers for a tile batch.
95///
96/// `available_workers` should be the host's available parallelism. Passing
97/// `0` is accepted and treated as one available worker.
98pub fn tile_batch_worker_count(
99    batch_size: usize,
100    options: TileBatchOptions,
101    available_workers: usize,
102) -> usize {
103    if batch_size <= 1 {
104        return 1;
105    }
106    let workers = options.workers.map_or(available_workers, NonZeroUsize::get);
107    workers.max(1).min(batch_size)
108}
109
110/// Restore successful indexed worker results to caller input order.
111///
112/// If any worker result failed, returns the error produced by `make_error`
113/// for the lowest failing input index.
114///
115/// # Panics
116///
117/// Panics if a successful batch is missing an index or if a result index is
118/// outside `0..job_count`.
119pub fn collect_indexed_batch_results<T, E, B, F>(
120    job_count: usize,
121    results: Vec<IndexedBatchResult<T, E>>,
122    make_error: F,
123) -> Result<Vec<T>, B>
124where
125    F: FnOnce(usize, E) -> B,
126{
127    let mut outcomes = Vec::with_capacity(job_count);
128    outcomes.resize_with(job_count, || None);
129    let mut first_error = None::<(usize, E)>;
130    for (index, result) in results {
131        assert!(
132            index < job_count,
133            "indexed batch result index {index} outside job count {job_count}"
134        );
135        match result {
136            Ok(outcome) => outcomes[index] = Some(outcome),
137            Err(source) => {
138                if first_error
139                    .as_ref()
140                    .is_none_or(|(current, _)| index < *current)
141                {
142                    first_error = Some((index, source));
143                }
144            }
145        }
146    }
147
148    if let Some((index, source)) = first_error {
149        return Err(make_error(index, source));
150    }
151
152    Ok(outcomes
153        .into_iter()
154        .map(|outcome| outcome.expect("successful batch stores one outcome per tile"))
155        .collect())
156}
j2k_core/batch.rs

j2k_core/
batch.rs
