j2k_transcode/
pipeline_map.rs

1// SPDX-License-Identifier: MIT OR Apache-2.0
2// j2k-coverage: shared-accelerator-host
3
4//! Stage-level residency report for JPEG-to-HTJ2K transcode timings.
5
6use core::fmt;
7
8use crate::{BatchTranscodeReport, TranscodeReport, TranscodeTimingReport};
9
10/// Logical stages in the JPEG-to-J2K/HTJ2K transcode pipeline.
11#[derive(Debug, Clone, Copy, PartialEq, Eq)]
12pub enum TranscodePipelineStageKind {
13    /// JPEG marker parsing, entropy decode, and DCT coefficient extraction.
14    EntropyDecode,
15    /// Coefficient repacking and host/device input preparation.
16    CoefficientPrep,
17    /// DCT-grid to wavelet-domain transform.
18    Transform,
19    /// Quantization, code-block layout, and pre-packet code-block work.
20    QuantizationCodeBlockPrep,
21    /// Packet header and packet body formation.
22    Packetization,
23    /// Final marker, tile-part, and codestream byte assembly.
24    CodestreamAssembly,
25}
26
27impl TranscodePipelineStageKind {
28    /// Stable snake-case label used by debug reports and logs.
29    #[must_use]
30    pub const fn as_str(self) -> &'static str {
31        match self {
32            Self::EntropyDecode => "entropy_decode",
33            Self::CoefficientPrep => "coefficient_prep",
34            Self::Transform => "transform",
35            Self::QuantizationCodeBlockPrep => "quantization_code_block_prep",
36            Self::Packetization => "packetization",
37            Self::CodestreamAssembly => "codestream_assembly",
38        }
39    }
40}
41
42impl fmt::Display for TranscodePipelineStageKind {
43    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
44        f.write_str(self.as_str())
45    }
46}
47
48/// Observed residency for a transcode stage.
49#[derive(Debug, Clone, Copy, PartialEq, Eq)]
50pub enum TranscodeStageProcessor {
51    /// Work is currently observed at CPU/native Rust or native encoder boundaries.
52    Cpu,
53    /// Work is observed through the Metal/accelerator stage counters.
54    Metal,
55    /// Existing counters show both CPU and Metal/accelerator work for this stage.
56    Hybrid,
57}
58
59impl TranscodeStageProcessor {
60    /// Stable label used by debug reports and logs.
61    #[must_use]
62    pub const fn as_str(self) -> &'static str {
63        match self {
64            Self::Cpu => "Cpu",
65            Self::Metal => "Metal",
66            Self::Hybrid => "Hybrid",
67        }
68    }
69}
70
71impl fmt::Display for TranscodeStageProcessor {
72    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
73        f.write_str(self.as_str())
74    }
75}
76
77/// One stage in a transcode pipeline residency map.
78#[derive(Debug, Clone, Copy, PartialEq, Eq)]
79pub struct TranscodePipelineStageReport {
80    /// Logical transcode stage.
81    pub stage: TranscodePipelineStageKind,
82    /// Observed CPU/Metal residency for this stage.
83    pub processor: TranscodeStageProcessor,
84    /// CPU/native time currently visible for this stage, in microseconds.
85    pub cpu_us: u128,
86    /// Metal/accelerator time currently visible for this stage, in microseconds.
87    pub metal_us: u128,
88    /// Host/device transfer time visible for this stage, in microseconds.
89    pub transfer_us: u128,
90    /// Logical host/device transfer operations visible for this stage.
91    pub transfer_count: usize,
92    /// Host/device transfer bytes visible for this stage.
93    pub transfer_bytes: u64,
94    /// Validated resident handoff descriptors visible for this stage.
95    pub resident_handoff_count: usize,
96    /// Dispatches observed for this stage.
97    pub dispatches: usize,
98    /// Component jobs that used CPU fallback at this stage.
99    pub fallback_jobs: usize,
100    /// Short interpretation of the counters behind this stage.
101    pub note: &'static str,
102}
103
104/// Recommended next stage to evaluate for Metal residency.
105#[derive(Debug, Clone, Copy, PartialEq, Eq)]
106pub struct TranscodeResidentStageRecommendation {
107    /// Stage that should be evaluated next.
108    pub stage: TranscodePipelineStageKind,
109    /// Existing measured time supporting the recommendation, in microseconds.
110    pub evidence_us: u128,
111    /// Existing dispatch count supporting the recommendation.
112    pub evidence_dispatches: usize,
113    /// Why this stage is the next candidate.
114    pub reason: &'static str,
115}
116
117/// Stage-by-stage transcode residency map derived from existing timings.
118#[derive(Debug, Clone, PartialEq, Eq)]
119pub struct TranscodePipelineMap {
120    /// Ordered stage reports from JPEG input through codestream output.
121    pub stages: Vec<TranscodePipelineStageReport>,
122    /// Next resident-stage candidate derived from the observed counters.
123    pub recommendation: TranscodeResidentStageRecommendation,
124}
125
126impl TranscodePipelineMap {
127    /// Build a pipeline map from an existing timing report.
128    #[must_use]
129    pub fn from_timings(timings: &TranscodeTimingReport) -> Self {
130        Self {
131            stages: vec![
132                entropy_decode_stage(timings),
133                coefficient_prep_stage(timings),
134                transform_stage(timings),
135                quantization_code_block_stage(timings),
136                packetization_stage(timings),
137                codestream_assembly_stage(timings),
138            ],
139            recommendation: recommend_next_resident_stage(timings),
140        }
141    }
142}
143
144impl TranscodeTimingReport {
145    /// Convert this timing report into a CPU/Metal transcode pipeline map.
146    #[must_use]
147    pub fn pipeline_map(&self) -> TranscodePipelineMap {
148        TranscodePipelineMap::from_timings(self)
149    }
150}
151
152impl TranscodeReport {
153    /// Convert this transcode report into a CPU/Metal pipeline map.
154    #[must_use]
155    pub fn pipeline_map(&self) -> TranscodePipelineMap {
156        self.timings.pipeline_map()
157    }
158}
159
160impl BatchTranscodeReport {
161    /// Convert this batch transcode report into a CPU/Metal pipeline map.
162    #[must_use]
163    pub fn pipeline_map(&self) -> TranscodePipelineMap {
164        self.timings.pipeline_map()
165    }
166}
167
168fn entropy_decode_stage(timings: &TranscodeTimingReport) -> TranscodePipelineStageReport {
169    TranscodePipelineStageReport {
170        stage: TranscodePipelineStageKind::EntropyDecode,
171        processor: TranscodeStageProcessor::Cpu,
172        cpu_us: timings.jpeg_dct_extract_us,
173        metal_us: 0,
174        transfer_us: 0,
175        transfer_count: 0,
176        transfer_bytes: 0,
177        resident_handoff_count: 0,
178        dispatches: 0,
179        fallback_jobs: 0,
180        note: "JPEG marker parsing, entropy decode, dequantization, and DCT coefficient extraction stay on CPU",
181    }
182}
183
184fn coefficient_prep_stage(timings: &TranscodeTimingReport) -> TranscodePipelineStageReport {
185    let transfer_us = timings.dwt97_batch_pack_upload_us;
186    let transfer_count = transfer_count_or_timing(
187        timings.dwt97_batch_pack_upload_transfers,
188        transfer_us,
189        timings.dwt97_batch_pack_upload_bytes,
190    );
191    let transfer_dispatch = usize::from(
192        transfer_us > 0 || transfer_count > 0 || timings.dwt97_batch_pack_upload_bytes > 0,
193    );
194    TranscodePipelineStageReport {
195        stage: TranscodePipelineStageKind::CoefficientPrep,
196        processor: processor_for(
197            timings.jpeg_dct_repack_us,
198            0,
199            transfer_us,
200            transfer_dispatch,
201            0,
202        ),
203        cpu_us: timings.jpeg_dct_repack_us,
204        metal_us: 0,
205        transfer_us,
206        transfer_count,
207        transfer_bytes: timings.dwt97_batch_pack_upload_bytes,
208        resident_handoff_count: timings.dwt97_batch_resident_dct_handoff_count,
209        dispatches: transfer_dispatch,
210        fallback_jobs: 0,
211        note: "DCT coefficient repack and Metal buffer pack/upload are visible before transform dispatch",
212    }
213}
214
215fn transform_stage(timings: &TranscodeTimingReport) -> TranscodePipelineStageReport {
216    let dwt97_kernel_us = timings
217        .dwt97_batch_idct_row_lift_us
218        .saturating_add(timings.dwt97_batch_column_lift_us);
219    let metal_us = if dwt97_kernel_us > 0 {
220        dwt97_kernel_us
221    } else if timings.accelerator_dispatches > 0 {
222        timings.dct_to_wavelet_accelerator_us
223    } else {
224        0
225    };
226    let cpu_us = timings
227        .dct_to_wavelet_cpu_fallback_us
228        .saturating_add(timings.dwt_decompose_us);
229    TranscodePipelineStageReport {
230        stage: TranscodePipelineStageKind::Transform,
231        processor: processor_for(
232            cpu_us,
233            metal_us,
234            timings.dwt97_batch_readback_us,
235            timings.accelerator_dispatches,
236            timings.cpu_fallback_jobs,
237        ),
238        cpu_us,
239        metal_us,
240        transfer_us: timings.dwt97_batch_readback_us,
241        transfer_count: transfer_count_or_timing(
242            timings.dwt97_batch_readback_transfers,
243            timings.dwt97_batch_readback_us,
244            timings.dwt97_batch_readback_bytes,
245        ),
246        transfer_bytes: timings.dwt97_batch_readback_bytes,
247        resident_handoff_count: timings.dwt97_batch_resident_dwt_handoff_count,
248        dispatches: timings.accelerator_dispatches,
249        fallback_jobs: timings.cpu_fallback_jobs,
250        note: "DCT-grid to DWT projection uses accelerator dispatches when available; Ok(None) jobs remain caller CPU fallback",
251    }
252}
253
254fn quantization_code_block_stage(timings: &TranscodeTimingReport) -> TranscodePipelineStageReport {
255    let metal_us = timings
256        .dwt97_batch_quantize_codeblock_us
257        .saturating_add(timings.dwt97_batch_ht_encode_us)
258        .saturating_add(timings.dwt97_batch_ht_kernel_us)
259        .saturating_add(timings.dwt97_batch_ht_compact_us);
260    let transfer_us = timings
261        .dwt97_batch_ht_status_readback_us
262        .saturating_add(timings.dwt97_batch_ht_output_readback_us);
263    let transfer_count = timings
264        .dwt97_batch_ht_status_readback_transfers
265        .saturating_add(timings.dwt97_batch_ht_output_readback_transfers);
266    let transfer_bytes = timings
267        .dwt97_batch_ht_status_readback_bytes
268        .saturating_add(timings.dwt97_batch_ht_output_readback_bytes);
269    let transfer_count = transfer_count_or_timing(transfer_count, transfer_us, transfer_bytes);
270    let dispatches = timings
271        .dwt97_batch_ht_codeblock_dispatches
272        .saturating_add(timings.htj2k_encode_ht_code_block_dispatches);
273    TranscodePipelineStageReport {
274        stage: TranscodePipelineStageKind::QuantizationCodeBlockPrep,
275        processor: processor_for(0, metal_us, transfer_us, dispatches, 0),
276        cpu_us: 0,
277        metal_us,
278        transfer_us,
279        transfer_count,
280        transfer_bytes,
281        resident_handoff_count: 0,
282        dispatches,
283        fallback_jobs: 0,
284        note: "9/7 quantization/code-block layout is only isolated when a backend reports resident stage timings; otherwise it is inside native encode time",
285    }
286}
287
288fn packetization_stage(timings: &TranscodeTimingReport) -> TranscodePipelineStageReport {
289    let dispatches = timings.htj2k_encode_packetization_dispatches;
290    TranscodePipelineStageReport {
291        stage: TranscodePipelineStageKind::Packetization,
292        processor: processor_for(0, 0, 0, dispatches, 0),
293        cpu_us: 0,
294        metal_us: 0,
295        transfer_us: 0,
296        transfer_count: 0,
297        transfer_bytes: 0,
298        resident_handoff_count: 0,
299        dispatches,
300        fallback_jobs: 0,
301        note: "Packetization dispatches are counted separately when an encode-stage accelerator handles them; CPU time is otherwise inside native encode time",
302    }
303}
304
305fn codestream_assembly_stage(timings: &TranscodeTimingReport) -> TranscodePipelineStageReport {
306    TranscodePipelineStageReport {
307        stage: TranscodePipelineStageKind::CodestreamAssembly,
308        processor: TranscodeStageProcessor::Cpu,
309        cpu_us: timings.htj2k_encode_us,
310        metal_us: 0,
311        transfer_us: 0,
312        transfer_count: 0,
313        transfer_bytes: 0,
314        resident_handoff_count: 0,
315        dispatches: 0,
316        fallback_jobs: 0,
317        note: "Final marker, tile-part, packet byte ordering, and codestream assembly remain at the CPU/native encoder boundary",
318    }
319}
320
321fn recommend_next_resident_stage(
322    timings: &TranscodeTimingReport,
323) -> TranscodeResidentStageRecommendation {
324    let transform_readback_us = timings.dwt97_batch_readback_us;
325    let code_block_readback_us = timings
326        .dwt97_batch_ht_status_readback_us
327        .saturating_add(timings.dwt97_batch_ht_output_readback_us);
328    let has_resident_transform = timings.accelerator_dispatches > 0
329        && (timings.dwt97_batch_idct_row_lift_us > 0
330            || timings.dwt97_batch_column_lift_us > 0
331            || timings.dct_to_wavelet_accelerator_us > 0);
332
333    if timings.cpu_fallback_jobs > 0 && timings.accelerator_dispatches == 0 {
334        return TranscodeResidentStageRecommendation {
335            stage: TranscodePipelineStageKind::Transform,
336            evidence_us: timings.dct_to_wavelet_cpu_fallback_us,
337            evidence_dispatches: timings.accelerator_attempts,
338            reason: "transform jobs are still completing through caller CPU fallback",
339        };
340    }
341
342    if has_resident_transform && timings.dwt97_batch_quantize_codeblock_us == 0 {
343        return TranscodeResidentStageRecommendation {
344            stage: TranscodePipelineStageKind::QuantizationCodeBlockPrep,
345            evidence_us: transform_readback_us.saturating_add(timings.htj2k_encode_us),
346            evidence_dispatches: timings.accelerator_dispatches,
347            reason: "resident transform output is read back before quantization/code-block prep and native encode",
348        };
349    }
350
351    if timings.dwt97_batch_quantize_codeblock_us > 0
352        && timings.dwt97_batch_ht_codeblock_dispatches == 0
353    {
354        return TranscodeResidentStageRecommendation {
355            stage: TranscodePipelineStageKind::QuantizationCodeBlockPrep,
356            evidence_us: transform_readback_us
357                .saturating_add(code_block_readback_us)
358                .saturating_add(timings.htj2k_encode_us),
359            evidence_dispatches: timings.accelerator_dispatches,
360            reason: "Metal already reaches 9/7 code-block prep; extend residency through HT code-block encode before packetization",
361        };
362    }
363
364    if timings.cpu_fallback_jobs > 0 {
365        return TranscodeResidentStageRecommendation {
366            stage: TranscodePipelineStageKind::Transform,
367            evidence_us: timings.dct_to_wavelet_cpu_fallback_us,
368            evidence_dispatches: timings.accelerator_attempts,
369            reason: "some transform jobs still use CPU fallback after accelerator attempts",
370        };
371    }
372
373    let coefficient_prep_us = timings
374        .jpeg_dct_repack_us
375        .saturating_add(timings.dwt97_batch_pack_upload_us);
376    if coefficient_prep_us > 0 {
377        return TranscodeResidentStageRecommendation {
378            stage: TranscodePipelineStageKind::CoefficientPrep,
379            evidence_us: coefficient_prep_us,
380            evidence_dispatches: timings.accelerator_dispatches,
381            reason:
382                "coefficient repack and host-to-device upload are visible before Metal work starts",
383        };
384    }
385
386    if timings.htj2k_encode_packetization_dispatches == 0 && timings.htj2k_encode_us > 0 {
387        return TranscodeResidentStageRecommendation {
388            stage: TranscodePipelineStageKind::Packetization,
389            evidence_us: timings.htj2k_encode_us,
390            evidence_dispatches: timings.htj2k_encode_accelerator_dispatches,
391            reason:
392                "packetization and codestream assembly remain inside the CPU/native encode boundary",
393        };
394    }
395
396    TranscodeResidentStageRecommendation {
397        stage: TranscodePipelineStageKind::CodestreamAssembly,
398        evidence_us: timings.htj2k_encode_us,
399        evidence_dispatches: timings.htj2k_encode_accelerator_dispatches,
400        reason: "no stronger resident-stage candidate is visible from the current counters",
401    }
402}
403
404fn processor_for(
405    cpu_us: u128,
406    metal_us: u128,
407    transfer_us: u128,
408    dispatches: usize,
409    fallback_jobs: usize,
410) -> TranscodeStageProcessor {
411    let has_cpu = cpu_us > 0 || fallback_jobs > 0;
412    let has_metal = metal_us > 0 || transfer_us > 0 || dispatches > 0;
413    match (has_cpu, has_metal) {
414        (true, true) => TranscodeStageProcessor::Hybrid,
415        (false, true) => TranscodeStageProcessor::Metal,
416        (_, false) => TranscodeStageProcessor::Cpu,
417    }
418}
419
420fn transfer_count_or_timing(
421    explicit_count: usize,
422    transfer_us: u128,
423    transfer_bytes: u64,
424) -> usize {
425    if explicit_count > 0 {
426        explicit_count
427    } else {
428        usize::from(transfer_us > 0 || transfer_bytes > 0)
429    }
430}
j2k_transcode/pipeline_map.rs

j2k_transcode/
pipeline_map.rs
