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    #[allow(
130        clippy::disallowed_macros,
131        reason = "the infallible diagnostic report API returns a codec-fixed six-stage Vec"
132    )]
133    pub fn from_timings(timings: &TranscodeTimingReport) -> Self {
134        Self {
135            stages: vec![
136                entropy_decode_stage(timings),
137                coefficient_prep_stage(timings),
138                transform_stage(timings),
139                quantization_code_block_stage(timings),
140                packetization_stage(timings),
141                codestream_assembly_stage(timings),
142            ],
143            recommendation: recommend_next_resident_stage(timings),
144        }
145    }
146}
147
148impl TranscodeTimingReport {
149    /// Convert this timing report into a CPU/Metal transcode pipeline map.
150    #[must_use]
151    pub fn pipeline_map(&self) -> TranscodePipelineMap {
152        TranscodePipelineMap::from_timings(self)
153    }
154}
155
156impl TranscodeReport {
157    /// Convert this transcode report into a CPU/Metal pipeline map.
158    #[must_use]
159    pub fn pipeline_map(&self) -> TranscodePipelineMap {
160        self.timings.pipeline_map()
161    }
162}
163
164impl BatchTranscodeReport {
165    /// Convert this batch transcode report into a CPU/Metal pipeline map.
166    #[must_use]
167    pub fn pipeline_map(&self) -> TranscodePipelineMap {
168        self.timings.pipeline_map()
169    }
170}
171
172fn entropy_decode_stage(timings: &TranscodeTimingReport) -> TranscodePipelineStageReport {
173    TranscodePipelineStageReport {
174        stage: TranscodePipelineStageKind::EntropyDecode,
175        processor: TranscodeStageProcessor::Cpu,
176        cpu_us: timings.jpeg_dct_extract_us,
177        metal_us: 0,
178        transfer_us: 0,
179        transfer_count: 0,
180        transfer_bytes: 0,
181        resident_handoff_count: 0,
182        dispatches: 0,
183        fallback_jobs: 0,
184        note: "JPEG marker parsing, entropy decode, dequantization, and DCT coefficient extraction stay on CPU",
185    }
186}
187
188fn coefficient_prep_stage(timings: &TranscodeTimingReport) -> TranscodePipelineStageReport {
189    let transfer_us = timings.dwt97_batch_pack_upload_us;
190    let transfer_count = transfer_count_or_timing(
191        timings.dwt97_batch_pack_upload_transfers,
192        transfer_us,
193        timings.dwt97_batch_pack_upload_bytes,
194    );
195    let transfer_dispatch = usize::from(
196        transfer_us > 0 || transfer_count > 0 || timings.dwt97_batch_pack_upload_bytes > 0,
197    );
198    TranscodePipelineStageReport {
199        stage: TranscodePipelineStageKind::CoefficientPrep,
200        processor: processor_for(
201            timings.jpeg_dct_repack_us,
202            0,
203            transfer_us,
204            transfer_dispatch,
205            0,
206        ),
207        cpu_us: timings.jpeg_dct_repack_us,
208        metal_us: 0,
209        transfer_us,
210        transfer_count,
211        transfer_bytes: timings.dwt97_batch_pack_upload_bytes,
212        resident_handoff_count: timings.dwt97_batch_resident_dct_handoff_count,
213        dispatches: transfer_dispatch,
214        fallback_jobs: 0,
215        note: "DCT coefficient repack and Metal buffer pack/upload are visible before transform dispatch",
216    }
217}
218
219fn transform_stage(timings: &TranscodeTimingReport) -> TranscodePipelineStageReport {
220    let dwt97_kernel_us = timings
221        .dwt97_batch_idct_row_lift_us
222        .saturating_add(timings.dwt97_batch_column_lift_us);
223    let metal_us = if dwt97_kernel_us > 0 {
224        dwt97_kernel_us
225    } else if timings.accelerator_dispatches > 0 {
226        timings.dct_to_wavelet_accelerator_us
227    } else {
228        0
229    };
230    let cpu_us = timings
231        .dct_to_wavelet_cpu_fallback_us
232        .saturating_add(timings.dwt_decompose_us);
233    TranscodePipelineStageReport {
234        stage: TranscodePipelineStageKind::Transform,
235        processor: processor_for(
236            cpu_us,
237            metal_us,
238            timings.dwt97_batch_readback_us,
239            timings.accelerator_dispatches,
240            timings.cpu_fallback_jobs,
241        ),
242        cpu_us,
243        metal_us,
244        transfer_us: timings.dwt97_batch_readback_us,
245        transfer_count: transfer_count_or_timing(
246            timings.dwt97_batch_readback_transfers,
247            timings.dwt97_batch_readback_us,
248            timings.dwt97_batch_readback_bytes,
249        ),
250        transfer_bytes: timings.dwt97_batch_readback_bytes,
251        resident_handoff_count: timings.dwt97_batch_resident_dwt_handoff_count,
252        dispatches: timings.accelerator_dispatches,
253        fallback_jobs: timings.cpu_fallback_jobs,
254        note: "DCT-grid to DWT projection uses accelerator dispatches when available; Ok(None) jobs remain caller CPU fallback",
255    }
256}
257
258fn quantization_code_block_stage(timings: &TranscodeTimingReport) -> TranscodePipelineStageReport {
259    let metal_us = timings
260        .dwt97_batch_quantize_codeblock_us
261        .saturating_add(timings.dwt97_batch_ht_encode_us)
262        .saturating_add(timings.dwt97_batch_ht_kernel_us)
263        .saturating_add(timings.dwt97_batch_ht_compact_us);
264    let transfer_us = timings
265        .dwt97_batch_ht_status_readback_us
266        .saturating_add(timings.dwt97_batch_ht_output_readback_us);
267    let transfer_count = timings
268        .dwt97_batch_ht_status_readback_transfers
269        .saturating_add(timings.dwt97_batch_ht_output_readback_transfers);
270    let transfer_bytes = timings
271        .dwt97_batch_ht_status_readback_bytes
272        .saturating_add(timings.dwt97_batch_ht_output_readback_bytes);
273    let transfer_count = transfer_count_or_timing(transfer_count, transfer_us, transfer_bytes);
274    let dispatches = timings
275        .dwt97_batch_ht_codeblock_dispatches
276        .saturating_add(timings.htj2k_encode_ht_code_block_dispatches);
277    TranscodePipelineStageReport {
278        stage: TranscodePipelineStageKind::QuantizationCodeBlockPrep,
279        processor: processor_for(0, metal_us, transfer_us, dispatches, 0),
280        cpu_us: 0,
281        metal_us,
282        transfer_us,
283        transfer_count,
284        transfer_bytes,
285        resident_handoff_count: 0,
286        dispatches,
287        fallback_jobs: 0,
288        note: "9/7 quantization/code-block layout is only isolated when a backend reports resident stage timings; otherwise it is inside native encode time",
289    }
290}
291
292fn packetization_stage(timings: &TranscodeTimingReport) -> TranscodePipelineStageReport {
293    let dispatches = timings.htj2k_encode_packetization_dispatches;
294    TranscodePipelineStageReport {
295        stage: TranscodePipelineStageKind::Packetization,
296        processor: processor_for(0, 0, 0, dispatches, 0),
297        cpu_us: 0,
298        metal_us: 0,
299        transfer_us: 0,
300        transfer_count: 0,
301        transfer_bytes: 0,
302        resident_handoff_count: 0,
303        dispatches,
304        fallback_jobs: 0,
305        note: "Packetization dispatches are counted separately when an encode-stage accelerator handles them; CPU time is otherwise inside native encode time",
306    }
307}
308
309fn codestream_assembly_stage(timings: &TranscodeTimingReport) -> TranscodePipelineStageReport {
310    TranscodePipelineStageReport {
311        stage: TranscodePipelineStageKind::CodestreamAssembly,
312        processor: TranscodeStageProcessor::Cpu,
313        cpu_us: timings.htj2k_encode_us,
314        metal_us: 0,
315        transfer_us: 0,
316        transfer_count: 0,
317        transfer_bytes: 0,
318        resident_handoff_count: 0,
319        dispatches: 0,
320        fallback_jobs: 0,
321        note: "Final marker, tile-part, packet byte ordering, and codestream assembly remain at the CPU/native encoder boundary",
322    }
323}
324
325fn recommend_next_resident_stage(
326    timings: &TranscodeTimingReport,
327) -> TranscodeResidentStageRecommendation {
328    let transform_readback_us = timings.dwt97_batch_readback_us;
329    let code_block_readback_us = timings
330        .dwt97_batch_ht_status_readback_us
331        .saturating_add(timings.dwt97_batch_ht_output_readback_us);
332    let has_resident_transform = timings.accelerator_dispatches > 0
333        && (timings.dwt97_batch_idct_row_lift_us > 0
334            || timings.dwt97_batch_column_lift_us > 0
335            || timings.dct_to_wavelet_accelerator_us > 0);
336
337    if timings.cpu_fallback_jobs > 0 && timings.accelerator_dispatches == 0 {
338        return TranscodeResidentStageRecommendation {
339            stage: TranscodePipelineStageKind::Transform,
340            evidence_us: timings.dct_to_wavelet_cpu_fallback_us,
341            evidence_dispatches: timings.accelerator_attempts,
342            reason: "transform jobs are still completing through caller CPU fallback",
343        };
344    }
345
346    if has_resident_transform && timings.dwt97_batch_quantize_codeblock_us == 0 {
347        return TranscodeResidentStageRecommendation {
348            stage: TranscodePipelineStageKind::QuantizationCodeBlockPrep,
349            evidence_us: transform_readback_us.saturating_add(timings.htj2k_encode_us),
350            evidence_dispatches: timings.accelerator_dispatches,
351            reason: "resident transform output is read back before quantization/code-block prep and native encode",
352        };
353    }
354
355    if timings.dwt97_batch_quantize_codeblock_us > 0
356        && timings.dwt97_batch_ht_codeblock_dispatches == 0
357    {
358        return TranscodeResidentStageRecommendation {
359            stage: TranscodePipelineStageKind::QuantizationCodeBlockPrep,
360            evidence_us: transform_readback_us
361                .saturating_add(code_block_readback_us)
362                .saturating_add(timings.htj2k_encode_us),
363            evidence_dispatches: timings.accelerator_dispatches,
364            reason: "Metal already reaches 9/7 code-block prep; extend residency through HT code-block encode before packetization",
365        };
366    }
367
368    if timings.cpu_fallback_jobs > 0 {
369        return TranscodeResidentStageRecommendation {
370            stage: TranscodePipelineStageKind::Transform,
371            evidence_us: timings.dct_to_wavelet_cpu_fallback_us,
372            evidence_dispatches: timings.accelerator_attempts,
373            reason: "some transform jobs still use CPU fallback after accelerator attempts",
374        };
375    }
376
377    let coefficient_prep_us = timings
378        .jpeg_dct_repack_us
379        .saturating_add(timings.dwt97_batch_pack_upload_us);
380    if coefficient_prep_us > 0 {
381        return TranscodeResidentStageRecommendation {
382            stage: TranscodePipelineStageKind::CoefficientPrep,
383            evidence_us: coefficient_prep_us,
384            evidence_dispatches: timings.accelerator_dispatches,
385            reason:
386                "coefficient repack and host-to-device upload are visible before Metal work starts",
387        };
388    }
389
390    if timings.htj2k_encode_packetization_dispatches == 0 && timings.htj2k_encode_us > 0 {
391        return TranscodeResidentStageRecommendation {
392            stage: TranscodePipelineStageKind::Packetization,
393            evidence_us: timings.htj2k_encode_us,
394            evidence_dispatches: timings.htj2k_encode_accelerator_dispatches,
395            reason:
396                "packetization and codestream assembly remain inside the CPU/native encode boundary",
397        };
398    }
399
400    TranscodeResidentStageRecommendation {
401        stage: TranscodePipelineStageKind::CodestreamAssembly,
402        evidence_us: timings.htj2k_encode_us,
403        evidence_dispatches: timings.htj2k_encode_accelerator_dispatches,
404        reason: "no stronger resident-stage candidate is visible from the current counters",
405    }
406}
407
408fn processor_for(
409    cpu_us: u128,
410    metal_us: u128,
411    transfer_us: u128,
412    dispatches: usize,
413    fallback_jobs: usize,
414) -> TranscodeStageProcessor {
415    let has_cpu = cpu_us > 0 || fallback_jobs > 0;
416    let has_metal = metal_us > 0 || transfer_us > 0 || dispatches > 0;
417    match (has_cpu, has_metal) {
418        (true, true) => TranscodeStageProcessor::Hybrid,
419        (false, true) => TranscodeStageProcessor::Metal,
420        (_, false) => TranscodeStageProcessor::Cpu,
421    }
422}
423
424fn transfer_count_or_timing(
425    explicit_count: usize,
426    transfer_us: u128,
427    transfer_bytes: u64,
428) -> usize {
429    if explicit_count > 0 {
430        explicit_count
431    } else {
432        usize::from(transfer_us > 0 || transfer_bytes > 0)
433    }
434}
j2k_transcode/pipeline_map.rs

j2k_transcode/
pipeline_map.rs
