Skip to main content

edgefirst_image/gl/
threaded.rs

1// SPDX-FileCopyrightText: Copyright 2025 Au-Zone Technologies
2// SPDX-License-Identifier: Apache-2.0
3
4use edgefirst_decoder::{DetectBox, ProtoData, Segmentation};
5use std::panic::AssertUnwindSafe;
6use std::ptr::NonNull;
7use std::thread::JoinHandle;
8use tokio::sync::mpsc::{Sender, WeakSender};
9
10use super::processor::GLProcessorST;
11use super::shaders::check_gl_error;
12use super::{EglDisplayKind, Int8InterpolationMode, TransferBackend};
13use crate::{Crop, Error, Flip, ImageProcessorTrait, Rotation};
14use edgefirst_tensor::TensorDyn;
15
16#[allow(clippy::type_complexity)]
17enum GLProcessorMessage {
18    ImageConvert(
19        SendablePtr<TensorDyn>,
20        SendablePtr<TensorDyn>,
21        Rotation,
22        Flip,
23        Crop,
24        bool, // defer: skip the per-tile glFinish (batch); flush syncs once
25        tokio::sync::oneshot::Sender<Result<(), Error>>,
26    ),
27    /// Convert without the terminal CPU sync, replying with a native
28    /// fence fd guarding the GPU work (the GL→NPU handoff). `Ok(None)` =
29    /// converted with blocking semantics (no fence on this platform).
30    ImageConvertFenced(
31        SendablePtr<TensorDyn>,
32        SendablePtr<TensorDyn>,
33        Rotation,
34        Flip,
35        Crop,
36        tokio::sync::oneshot::Sender<Result<Option<std::os::fd::OwnedFd>, Error>>,
37    ),
38    /// Complete any deferred batch render with a single GPU sync.
39    Flush(tokio::sync::oneshot::Sender<Result<(), Error>>),
40    SetColors(
41        Vec<[u8; 4]>,
42        tokio::sync::oneshot::Sender<Result<(), Error>>,
43    ),
44    DrawDecodedMasks(
45        SendablePtr<TensorDyn>,
46        SendablePtr<DetectBox>,
47        SendablePtr<Segmentation>,
48        f32,                            // opacity
49        Option<SendablePtr<TensorDyn>>, // background
50        Option<[f32; 4]>,               // letterbox
51        crate::ColorMode,
52        tokio::sync::oneshot::Sender<Result<(), Error>>,
53    ),
54    DrawProtoMasks(
55        SendablePtr<TensorDyn>,
56        SendablePtr<DetectBox>,
57        SendablePtr<ProtoData>,
58        f32,                            // opacity
59        Option<SendablePtr<TensorDyn>>, // background
60        Option<[f32; 4]>,               // letterbox
61        crate::ColorMode,
62        tokio::sync::oneshot::Sender<Result<(), Error>>,
63    ),
64    SetInt8Interpolation(
65        Int8InterpolationMode,
66        tokio::sync::oneshot::Sender<Result<(), Error>>,
67    ),
68    /// Set the colorimetry/performance trade-off (`ColorimetryMode`).
69    SetColorimetryMode(
70        crate::ColorimetryMode,
71        tokio::sync::oneshot::Sender<Result<(), Error>>,
72    ),
73    /// Snapshot the EGLImage cache counters (steady-state import assertions).
74    EglCacheStats(tokio::sync::oneshot::Sender<Result<super::cache::GlCacheStats, Error>>),
75    /// Snapshot the per-convert source-feed counters (zero-copy observability).
76    ConvertStats(tokio::sync::oneshot::Sender<Result<super::cache::ConvertStats, Error>>),
77    PboCreate(
78        usize, // buffer size in bytes
79        tokio::sync::oneshot::Sender<Result<u32, Error>>,
80    ),
81    PboMap(
82        u32,   // buffer_id
83        usize, // size
84        tokio::sync::oneshot::Sender<Result<edgefirst_tensor::PboMapping, Error>>,
85    ),
86    PboUnmap(
87        u32, // buffer_id
88        tokio::sync::oneshot::Sender<Result<(), Error>>,
89    ),
90    PboDelete(u32), // fire-and-forget, no reply
91    CudaRegisterBuffer(u32, tokio::sync::oneshot::Sender<Option<usize>>),
92    CudaMap(usize, tokio::sync::oneshot::Sender<Option<(usize, usize)>>),
93    CudaUnmap(usize),      // fire-and-forget
94    CudaUnregister(usize), // fire-and-forget
95}
96
97/// Compute the flat element count for a PBO image buffer of the given format.
98///
99/// NV12 and NV16 are semiplanar with non-trivial element counts; all other
100/// formats use `width * height * channels`.
101///
102/// Returns `None` on `usize` overflow. A wrapped element count would size
103/// the PBO too small and corrupt memory on readback, so callers must treat
104/// `None` as an invalid-dimensions error (the same way they already reject
105/// a zero count).
106fn pbo_elem_count(
107    width: usize,
108    height: usize,
109    format: edgefirst_tensor::PixelFormat,
110) -> Option<usize> {
111    // Element count == product of the canonical image shape (the PBO holds u8,
112    // one byte per element). Delegating to `image_shape` keeps the combined
113    // semi-planar height in lockstep with every other consumer (and is exact
114    // for odd heights — `height*3/2` truncation under-sized odd-H NV12 by a
115    // row, and the old NV24 arm fell through to `channels` = far too small).
116    // `checked_mul` so a wrapped count can't silently undersize the PBO.
117    format
118        .image_shape(width, height)?
119        .iter()
120        .try_fold(1usize, |acc, &d| acc.checked_mul(d))
121}
122
123/// Compute the tensor shape for a PBO image of the given format — the canonical
124/// [`PixelFormat::image_shape`] (Planar `[C,H,W]`, SemiPlanar `[total_h, W]`
125/// with the odd-height-exact combined-plane height, Packed `[H,W,C]`).
126fn pbo_shape(width: usize, height: usize, format: edgefirst_tensor::PixelFormat) -> Vec<usize> {
127    format
128        .image_shape(width, height)
129        .unwrap_or_else(|| vec![height, width, format.channels()])
130}
131
132/// Best-effort registration of a freshly-created PBO with CUDA GL interop.
133///
134/// When libcudart is present, asks the GL worker (where the GL context is
135/// current) to call `cudaGraphicsGLRegisterBuffer` for `buffer_id`, then
136/// attaches a [`CudaHandle`] so `cuda_map()` can later yield a linear,
137/// 256-byte-aligned device pointer for TensorRT. On any failure (no CUDA,
138/// channel closed, registration rejected) the handle is left unset and the
139/// PBO is still usable as a normal CPU/GL buffer.
140fn register_pbo_cuda<T>(
141    tensor: &mut edgefirst_tensor::Tensor<T>,
142    buffer_id: u32,
143    size: usize,
144    sender: &Sender<GLProcessorMessage>,
145) where
146    T: num_traits::Num + Clone + std::fmt::Debug + Send + Sync,
147{
148    if !edgefirst_tensor::is_cuda_available() {
149        return;
150    }
151    let (tx, rx) = tokio::sync::oneshot::channel();
152    if sender
153        .blocking_send(GLProcessorMessage::CudaRegisterBuffer(buffer_id, tx))
154        .is_ok()
155    {
156        if let Ok(Some(resource)) = rx.blocking_recv() {
157            let ops = std::sync::Arc::new(GlCudaOps {
158                sender: sender.downgrade(),
159            });
160            tensor.set_cuda_handle(edgefirst_tensor::CudaHandle::new_gl(
161                resource as *mut std::ffi::c_void,
162                size,
163                ops,
164            ));
165        }
166    }
167}
168
169/// Implements PboOps by sending commands to the GL thread.
170///
171/// Uses a `WeakSender` so that PBO images don't keep the GL thread's channel
172/// alive. When the `GLProcessorThreaded` is dropped, its `Sender` is the last
173/// strong reference — dropping it closes the channel and lets the GL thread
174/// exit. PBO operations after that return `PboDisconnected`.
175struct GlPboOps {
176    sender: WeakSender<GLProcessorMessage>,
177}
178
179// SAFETY: GlPboOps sends all GL operations to the dedicated GL thread via a
180// channel. `map_buffer` returns a CPU-visible pointer from `glMapBufferRange`
181// that remains valid until `unmap_buffer` calls `glUnmapBuffer` on the GL thread.
182// `delete_buffer` sends a fire-and-forget deletion command to the GL thread.
183unsafe impl edgefirst_tensor::PboOps for GlPboOps {
184    fn map_buffer(
185        &self,
186        buffer_id: u32,
187        size: usize,
188    ) -> edgefirst_tensor::Result<edgefirst_tensor::PboMapping> {
189        let sender = self
190            .sender
191            .upgrade()
192            .ok_or(edgefirst_tensor::Error::PboDisconnected)?;
193        let (tx, rx) = tokio::sync::oneshot::channel();
194        sender
195            .blocking_send(GLProcessorMessage::PboMap(buffer_id, size, tx))
196            .map_err(|_| edgefirst_tensor::Error::PboDisconnected)?;
197        rx.blocking_recv()
198            .map_err(|_| edgefirst_tensor::Error::PboDisconnected)?
199            .map_err(|e| {
200                edgefirst_tensor::Error::NotImplemented(format!("GL PBO map failed: {e:?}"))
201            })
202    }
203
204    fn unmap_buffer(&self, buffer_id: u32) -> edgefirst_tensor::Result<()> {
205        let sender = self
206            .sender
207            .upgrade()
208            .ok_or(edgefirst_tensor::Error::PboDisconnected)?;
209        let (tx, rx) = tokio::sync::oneshot::channel();
210        sender
211            .blocking_send(GLProcessorMessage::PboUnmap(buffer_id, tx))
212            .map_err(|_| edgefirst_tensor::Error::PboDisconnected)?;
213        rx.blocking_recv()
214            .map_err(|_| edgefirst_tensor::Error::PboDisconnected)?
215            .map_err(|e| {
216                edgefirst_tensor::Error::NotImplemented(format!("GL PBO unmap failed: {e:?}"))
217            })
218    }
219
220    fn delete_buffer(&self, buffer_id: u32) {
221        if let Some(sender) = self.sender.upgrade() {
222            let _ = sender.blocking_send(GLProcessorMessage::PboDelete(buffer_id));
223        }
224    }
225}
226
227/// Routes CUDA GL-interop map/unmap/unregister to the GL thread.
228///
229/// Mirrors [`GlPboOps`]: holds a [`WeakSender`] so a registered PBO doesn't
230/// keep the GL thread's channel alive. The CUDA primitives
231/// (`cudaGraphicsMapResources` etc.) require the GL context to be current,
232/// so they must execute on the dedicated GL worker thread.
233struct GlCudaOps {
234    sender: WeakSender<GLProcessorMessage>,
235}
236
237impl edgefirst_tensor::CudaGlOps for GlCudaOps {
238    fn map(&self, resource: *mut std::ffi::c_void) -> Option<(*mut std::ffi::c_void, usize)> {
239        let sender = self.sender.upgrade()?;
240        let (tx, rx) = tokio::sync::oneshot::channel();
241        sender
242            .blocking_send(GLProcessorMessage::CudaMap(resource as usize, tx))
243            .ok()?;
244        rx.blocking_recv()
245            .ok()?
246            .map(|(p, l)| (p as *mut std::ffi::c_void, l))
247    }
248
249    fn unmap(&self, resource: *mut std::ffi::c_void) {
250        if let Some(sender) = self.sender.upgrade() {
251            let _ = sender.blocking_send(GLProcessorMessage::CudaUnmap(resource as usize));
252        }
253    }
254
255    fn unregister(&self, resource: *mut std::ffi::c_void) {
256        if let Some(sender) = self.sender.upgrade() {
257            let _ = sender.blocking_send(GLProcessorMessage::CudaUnregister(resource as usize));
258        }
259    }
260}
261
262/// OpenGL multi-threaded image converter. The actual conversion is done in a
263/// separate rendering thread, as OpenGL contexts are not thread-safe. This can
264/// be safely sent between threads. The `convert()` call sends the conversion
265/// request to the rendering thread and waits for the result.
266#[derive(Debug)]
267pub struct GLProcessorThreaded {
268    // This is only None when the converter is being dropped.
269    handle: Option<JoinHandle<()>>,
270
271    // This is only None when the converter is being dropped.
272    sender: Option<Sender<GLProcessorMessage>>,
273    /// Immutable capability surface (transfer backend, float render
274    /// support, serialization policy), captured once from the worker at
275    /// construction. See `PlatformCaps` in `platform/mod.rs`.
276    caps: super::platform::PlatformCaps,
277}
278
279unsafe impl Send for GLProcessorThreaded {}
280unsafe impl Sync for GLProcessorThreaded {}
281
282struct SendablePtr<T: Send> {
283    ptr: NonNull<T>,
284    len: usize,
285}
286
287unsafe impl<T> Send for SendablePtr<T> where T: Send {}
288
289/// Extract a human-readable message from a `catch_unwind` panic payload.
290fn panic_message(info: &(dyn std::any::Any + Send)) -> String {
291    if let Some(s) = info.downcast_ref::<&str>() {
292        s.to_string()
293    } else if let Some(s) = info.downcast_ref::<String>() {
294        s.clone()
295    } else {
296        "unknown panic".to_string()
297    }
298}
299
300impl GLProcessorThreaded {
301    /// Creates a new OpenGL multi-threaded image converter.
302    pub fn new(kind: Option<EglDisplayKind>) -> Result<Self, Error> {
303        let (send, mut recv) = tokio::sync::mpsc::channel::<GLProcessorMessage>(1);
304
305        let (create_ctx_send, create_ctx_recv) = tokio::sync::oneshot::channel();
306
307        let func = move || {
308            // Creation runs under the global lifecycle lock on Linux
309            // (bring-up races on some embedded drivers); ANGLE serializes
310            // display-level entry points internally, so macOS needs none
311            // (validated by the A0 lifecycle-churn spike leg).
312            #[cfg(target_os = "linux")]
313            let init_result = {
314                let _guard = super::context::GL_MUTEX
315                    .lock()
316                    .unwrap_or_else(|e| e.into_inner());
317                GLProcessorST::new(kind)
318            };
319            // macOS: ANGLE serializes internally. Android: the system EGL
320            // is specification-conformant thread-safe. Neither needs the
321            // Linux lifecycle lock.
322            #[cfg(any(target_os = "macos", target_os = "ios", target_os = "android"))]
323            let init_result = GLProcessorST::new(kind);
324            let mut gl_converter = match init_result {
325                Ok(gl) => gl,
326                Err(e) => {
327                    let _ = create_ctx_send.send(Err(e));
328                    return;
329                }
330            };
331            // Capability surface captured ONCE before the message loop —
332            // immutable for the worker's life, never re-probed per message.
333            let caps = gl_converter.platform_caps();
334            let _ = create_ctx_send.send(Ok(caps));
335            let mut poisoned = false;
336            // Per-message serialization policy: Full where the platform
337            // demands it (`caps.serialize_gl` — Vivante/galcore is not
338            // thread-safe for concurrent GL across contexts), LifecycleOnly
339            // everywhere else — multiple processors then run GL work in
340            // parallel, each on its own thread + context. See the GL_MUTEX
341            // doc comment in context.rs. EDGEFIRST_GL_SERIALIZE overrides
342            // per-driver defaults (full = escape hatch for unknown-bad
343            // drivers; lifecycle = force-parallel).
344            let serialize_per_msg = match std::env::var("EDGEFIRST_GL_SERIALIZE").as_deref() {
345                Ok("full") => true,
346                Ok("lifecycle") => false,
347                _ => caps.serialize_gl,
348            };
349            log::debug!(
350                "GL serialization policy: {}",
351                if serialize_per_msg {
352                    "Full (per-message global lock)"
353                } else {
354                    "LifecycleOnly (parallel across processors)"
355                }
356            );
357            while let Some(msg) = recv.blocking_recv() {
358                // Full policy: one processor's message at a time, process-wide.
359                // Linux locks GL_MUTEX (messages must also exclude the locked
360                // lifecycle operations — on Vivante a convert racing a context
361                // bring-up is part of the driver bug class). macOS has no
362                // lifecycle lock (ANGLE serializes display entry points
363                // internally), so message-vs-message exclusion suffices —
364                // needed on virtualized GPUs, where concurrent GL across
365                // contexts mis-renders (paravirtual Metal, macOS CI runners).
366                #[cfg(target_os = "linux")]
367                let _guard = serialize_per_msg.then(|| {
368                    super::context::GL_MUTEX
369                        .lock()
370                        .unwrap_or_else(|e| e.into_inner())
371                });
372                #[cfg(any(target_os = "macos", target_os = "ios", target_os = "android"))]
373                let _guard = serialize_per_msg.then(|| {
374                    static MSG_MUTEX: std::sync::Mutex<()> = std::sync::Mutex::new(());
375                    MSG_MUTEX.lock().unwrap_or_else(|e| e.into_inner())
376                });
377
378                // After a panic, the GL context is in an undefined state. Reject
379                // all subsequent messages with an error rather than risking wrong
380                // output or a GPU hang from corrupted GL state. This follows the
381                // same pattern as std::sync::Mutex poisoning.
382                if poisoned {
383                    let poison_err = crate::Error::Internal(
384                        "GL context is poisoned after a prior panic".to_string(),
385                    );
386                    match msg {
387                        GLProcessorMessage::ImageConvertFenced(.., resp) => {
388                            let _ = resp.send(Err(poison_err));
389                        }
390                        GLProcessorMessage::ImageConvert(.., resp) => {
391                            let _ = resp.send(Err(poison_err));
392                        }
393                        GLProcessorMessage::Flush(resp) => {
394                            let _ = resp.send(Err(poison_err));
395                        }
396                        GLProcessorMessage::DrawDecodedMasks(.., resp) => {
397                            let _ = resp.send(Err(poison_err));
398                        }
399                        GLProcessorMessage::DrawProtoMasks(.., resp) => {
400                            let _ = resp.send(Err(poison_err));
401                        }
402                        GLProcessorMessage::SetColors(_, resp) => {
403                            let _ = resp.send(Err(poison_err));
404                        }
405                        GLProcessorMessage::SetInt8Interpolation(_, resp) => {
406                            let _ = resp.send(Err(poison_err));
407                        }
408                        GLProcessorMessage::SetColorimetryMode(_, resp) => {
409                            let _ = resp.send(Err(poison_err));
410                        }
411                        GLProcessorMessage::EglCacheStats(resp) => {
412                            let _ = resp.send(Err(poison_err));
413                        }
414                        GLProcessorMessage::ConvertStats(resp) => {
415                            let _ = resp.send(Err(poison_err));
416                        }
417                        GLProcessorMessage::PboCreate(_, resp) => {
418                            let _ = resp.send(Err(poison_err));
419                        }
420                        GLProcessorMessage::PboMap(_, _, resp) => {
421                            let _ = resp.send(Err(poison_err));
422                        }
423                        GLProcessorMessage::PboUnmap(_, resp) => {
424                            let _ = resp.send(Err(poison_err));
425                        }
426                        GLProcessorMessage::PboDelete(_) => {}
427                        GLProcessorMessage::CudaRegisterBuffer(_, resp) => {
428                            let _ = resp.send(None);
429                        }
430                        GLProcessorMessage::CudaMap(_, resp) => {
431                            let _ = resp.send(None);
432                        }
433                        // GL context is poisoned/dead — the GL buffer is unusable, so unmapping/
434                        // unregistering the CUDA interop is moot (the resource is reclaimed at
435                        // process exit). Same accepted "GL is gone" no-op as PboDelete above.
436                        GLProcessorMessage::CudaUnmap(_) => {}
437                        GLProcessorMessage::CudaUnregister(_) => {}
438                    }
439                    continue;
440                }
441
442                match msg {
443                    GLProcessorMessage::ImageConvert(
444                        src,
445                        mut dst,
446                        rotation,
447                        flip,
448                        crop,
449                        defer,
450                        resp,
451                    ) => {
452                        // SAFETY: This is safe because the convert() function waits for the resp to
453                        // be sent before dropping the borrow for src and dst
454                        let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
455                            let src = unsafe { src.ptr.as_ref() };
456                            let dst = unsafe { dst.ptr.as_mut() };
457                            // A deferred convert skips the per-tile glFinish and
458                            // leaves a single sync owed at Flush; an eager convert
459                            // finishes as usual.
460                            if defer {
461                                gl_converter.convert_deferred(src, dst, rotation, flip, crop)
462                            } else {
463                                gl_converter.convert(src, dst, rotation, flip, crop)
464                            }
465                        }));
466                        let _ = resp.send(match result {
467                            Ok(res) => res,
468                            Err(e) => {
469                                poisoned = true;
470                                Err(crate::Error::Internal(format!(
471                                    "GL thread panicked during ImageConvert: {}",
472                                    panic_message(e.as_ref()),
473                                )))
474                            }
475                        });
476                    }
477                    GLProcessorMessage::ImageConvertFenced(
478                        src,
479                        mut dst,
480                        rotation,
481                        flip,
482                        crop,
483                        resp,
484                    ) => {
485                        // SAFETY: as ImageConvert — convert_with_fence()
486                        // blocks on the reply before the borrows drop.
487                        let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
488                            let src = unsafe { src.ptr.as_ref() };
489                            let dst = unsafe { dst.ptr.as_mut() };
490                            gl_converter.convert_fenced(src, dst, rotation, flip, crop)
491                        }));
492                        let _ = resp.send(match result {
493                            Ok(res) => res,
494                            Err(e) => {
495                                poisoned = true;
496                                Err(crate::Error::Internal(format!(
497                                    "GL thread panicked during ImageConvertFenced: {}",
498                                    panic_message(e.as_ref()),
499                                )))
500                            }
501                        });
502                    }
503                    GLProcessorMessage::Flush(resp) => {
504                        let result =
505                            std::panic::catch_unwind(AssertUnwindSafe(|| gl_converter.flush()));
506                        let _ = resp.send(match result {
507                            Ok(res) => res,
508                            Err(e) => {
509                                poisoned = true;
510                                Err(crate::Error::Internal(format!(
511                                    "GL thread panicked during Flush: {}",
512                                    panic_message(e.as_ref()),
513                                )))
514                            }
515                        });
516                    }
517                    GLProcessorMessage::DrawDecodedMasks(
518                        mut dst,
519                        det,
520                        seg,
521                        opacity,
522                        bg,
523                        letterbox,
524                        color_mode,
525                        resp,
526                    ) => {
527                        // SAFETY: This is safe because the draw_decoded_masks() function waits for the
528                        // resp to be sent before dropping the borrow for dst, detect,
529                        // segmentation, and background
530                        let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
531                            let dst = unsafe { dst.ptr.as_mut() };
532                            let det =
533                                unsafe { std::slice::from_raw_parts(det.ptr.as_ptr(), det.len) };
534                            let seg =
535                                unsafe { std::slice::from_raw_parts(seg.ptr.as_ptr(), seg.len) };
536                            let bg_ref = bg.map(|p| unsafe { &*p.ptr.as_ptr() });
537                            gl_converter.draw_decoded_masks(
538                                dst,
539                                det,
540                                seg,
541                                crate::MaskOverlay {
542                                    background: bg_ref,
543                                    opacity,
544                                    letterbox,
545                                    color_mode,
546                                },
547                            )
548                        }));
549                        let _ = resp.send(match result {
550                            Ok(res) => res,
551                            Err(e) => {
552                                poisoned = true;
553                                Err(crate::Error::Internal(format!(
554                                    "GL thread panicked during DrawDecodedMasks: {}",
555                                    panic_message(e.as_ref()),
556                                )))
557                            }
558                        });
559                    }
560                    GLProcessorMessage::DrawProtoMasks(
561                        mut dst,
562                        det,
563                        proto_data,
564                        opacity,
565                        bg,
566                        letterbox,
567                        color_mode,
568                        resp,
569                    ) => {
570                        // SAFETY: Same safety invariant as DrawDecodedMasks — caller
571                        // blocks on resp before dropping borrows.
572                        let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
573                            let dst = unsafe { dst.ptr.as_mut() };
574                            let det =
575                                unsafe { std::slice::from_raw_parts(det.ptr.as_ptr(), det.len) };
576                            let bg_ref = bg.map(|p| unsafe { &*p.ptr.as_ptr() });
577                            let proto_data = unsafe { proto_data.ptr.as_ref() };
578                            gl_converter.draw_proto_masks(
579                                dst,
580                                det,
581                                proto_data,
582                                crate::MaskOverlay {
583                                    background: bg_ref,
584                                    opacity,
585                                    letterbox,
586                                    color_mode,
587                                },
588                            )
589                        }));
590                        let _ = resp.send(match result {
591                            Ok(res) => res,
592                            Err(e) => {
593                                poisoned = true;
594                                Err(crate::Error::Internal(format!(
595                                    "GL thread panicked during DrawProtoMasks: {}",
596                                    panic_message(e.as_ref()),
597                                )))
598                            }
599                        });
600                    }
601                    GLProcessorMessage::SetColors(colors, resp) => {
602                        let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
603                            gl_converter.set_class_colors(&colors)
604                        }));
605                        let _ = resp.send(match result {
606                            Ok(res) => res,
607                            Err(e) => {
608                                poisoned = true;
609                                Err(crate::Error::Internal(format!(
610                                    "GL thread panicked during SetColors: {}",
611                                    panic_message(e.as_ref()),
612                                )))
613                            }
614                        });
615                    }
616                    GLProcessorMessage::SetColorimetryMode(mode, resp) => {
617                        let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
618                            gl_converter.set_colorimetry_mode(mode);
619                            Ok(())
620                        }));
621                        let _ = resp.send(match result {
622                            Ok(res) => res,
623                            Err(e) => {
624                                poisoned = true;
625                                Err(crate::Error::Internal(format!(
626                                    "GL thread panicked during SetColorimetryMode: {}",
627                                    panic_message(e.as_ref()),
628                                )))
629                            }
630                        });
631                    }
632                    GLProcessorMessage::SetInt8Interpolation(mode, resp) => {
633                        let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
634                            gl_converter.set_int8_interpolation_mode(mode);
635                            Ok(())
636                        }));
637                        let _ = resp.send(match result {
638                            Ok(res) => res,
639                            Err(e) => {
640                                poisoned = true;
641                                Err(crate::Error::Internal(format!(
642                                    "GL thread panicked during SetInt8Interpolation: {}",
643                                    panic_message(e.as_ref()),
644                                )))
645                            }
646                        });
647                    }
648                    GLProcessorMessage::EglCacheStats(resp) => {
649                        let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
650                            Ok(gl_converter.egl_cache_stats())
651                        }));
652                        let _ = resp.send(match result {
653                            Ok(res) => res,
654                            Err(e) => {
655                                poisoned = true;
656                                Err(crate::Error::Internal(format!(
657                                    "GL thread panicked during EglCacheStats: {}",
658                                    panic_message(e.as_ref()),
659                                )))
660                            }
661                        });
662                    }
663                    GLProcessorMessage::ConvertStats(resp) => {
664                        let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
665                            Ok(gl_converter.convert_stats())
666                        }));
667                        let _ = resp.send(match result {
668                            Ok(res) => res,
669                            Err(e) => {
670                                poisoned = true;
671                                Err(crate::Error::Internal(format!(
672                                    "GL thread panicked during ConvertStats: {}",
673                                    panic_message(e.as_ref()),
674                                )))
675                            }
676                        });
677                    }
678                    GLProcessorMessage::PboCreate(size, resp) => {
679                        let result = std::panic::catch_unwind(AssertUnwindSafe(|| unsafe {
680                            let mut id: u32 = 0;
681                            edgefirst_gl::gl::GenBuffers(1, &mut id);
682                            edgefirst_gl::gl::BindBuffer(edgefirst_gl::gl::PIXEL_PACK_BUFFER, id);
683                            edgefirst_gl::gl::BufferData(
684                                edgefirst_gl::gl::PIXEL_PACK_BUFFER,
685                                size as isize,
686                                std::ptr::null(),
687                                edgefirst_gl::gl::STREAM_COPY,
688                            );
689                            edgefirst_gl::gl::BindBuffer(edgefirst_gl::gl::PIXEL_PACK_BUFFER, 0);
690                            match check_gl_error("PboCreate", 0) {
691                                Ok(()) => Ok(id),
692                                Err(e) => {
693                                    edgefirst_gl::gl::DeleteBuffers(1, &id);
694                                    Err(e)
695                                }
696                            }
697                        }));
698                        let _ = resp.send(match result {
699                            Ok(res) => res,
700                            Err(e) => {
701                                poisoned = true;
702                                Err(crate::Error::Internal(format!(
703                                    "GL thread panicked during PboCreate: {}",
704                                    panic_message(e.as_ref()),
705                                )))
706                            }
707                        });
708                    }
709                    GLProcessorMessage::PboMap(buffer_id, size, resp) => {
710                        let result = std::panic::catch_unwind(AssertUnwindSafe(|| unsafe {
711                            edgefirst_gl::gl::BindBuffer(
712                                edgefirst_gl::gl::PIXEL_PACK_BUFFER,
713                                buffer_id,
714                            );
715                            let ptr = edgefirst_gl::gl::MapBufferRange(
716                                edgefirst_gl::gl::PIXEL_PACK_BUFFER,
717                                0,
718                                size as isize,
719                                edgefirst_gl::gl::MAP_READ_BIT | edgefirst_gl::gl::MAP_WRITE_BIT,
720                            );
721                            edgefirst_gl::gl::BindBuffer(edgefirst_gl::gl::PIXEL_PACK_BUFFER, 0);
722                            if ptr.is_null() {
723                                Err(crate::Error::OpenGl(
724                                    "glMapBufferRange returned null".to_string(),
725                                ))
726                            } else {
727                                Ok(edgefirst_tensor::PboMapping {
728                                    ptr: ptr as *mut u8,
729                                    size,
730                                })
731                            }
732                        }));
733                        let _ = resp.send(match result {
734                            Ok(res) => res,
735                            Err(e) => {
736                                poisoned = true;
737                                Err(crate::Error::Internal(format!(
738                                    "GL thread panicked during PboMap: {}",
739                                    panic_message(e.as_ref()),
740                                )))
741                            }
742                        });
743                    }
744                    GLProcessorMessage::PboUnmap(buffer_id, resp) => {
745                        let result = std::panic::catch_unwind(AssertUnwindSafe(|| unsafe {
746                            edgefirst_gl::gl::BindBuffer(
747                                edgefirst_gl::gl::PIXEL_PACK_BUFFER,
748                                buffer_id,
749                            );
750                            let ok =
751                                edgefirst_gl::gl::UnmapBuffer(edgefirst_gl::gl::PIXEL_PACK_BUFFER);
752                            edgefirst_gl::gl::BindBuffer(edgefirst_gl::gl::PIXEL_PACK_BUFFER, 0);
753                            if ok == edgefirst_gl::gl::FALSE {
754                                Err(Error::OpenGl(
755                                    "PBO data was corrupted during mapping".into(),
756                                ))
757                            } else {
758                                check_gl_error("PboUnmap", 0)
759                            }
760                        }));
761                        let _ = resp.send(match result {
762                            Ok(res) => res,
763                            Err(e) => {
764                                poisoned = true;
765                                Err(crate::Error::Internal(format!(
766                                    "GL thread panicked during PboUnmap: {}",
767                                    panic_message(e.as_ref()),
768                                )))
769                            }
770                        });
771                    }
772                    GLProcessorMessage::PboDelete(buffer_id) => {
773                        if let Err(e) = std::panic::catch_unwind(AssertUnwindSafe(|| unsafe {
774                            edgefirst_gl::gl::DeleteBuffers(1, &buffer_id);
775                        })) {
776                            poisoned = true;
777                            log::error!(
778                                "GL thread panicked during PboDelete: {}",
779                                panic_message(e.as_ref()),
780                            );
781                        }
782                    }
783                    GLProcessorMessage::CudaRegisterBuffer(buffer_id, resp) => {
784                        // CUDA GL-interop must run on the GL-context thread.
785                        let _ = resp.send(edgefirst_tensor::gl_register_buffer(buffer_id));
786                    }
787                    GLProcessorMessage::CudaMap(resource, resp) => {
788                        // Auto-flush: if a batched `convert_deferred` is still
789                        // owed its single GPU sync, complete it before CUDA maps
790                        // the buffer — otherwise the device could read a render
791                        // still in flight. This makes `cuda_map()` correct on a
792                        // batched output with no API change (the caller pays the
793                        // sync it would have owed anyway).
794                        gl_converter.flush_pending();
795                        let _ = resp.send(edgefirst_tensor::gl_map_resource(resource));
796                    }
797                    GLProcessorMessage::CudaUnmap(resource) => {
798                        edgefirst_tensor::gl_unmap_resource(resource);
799                    }
800                    GLProcessorMessage::CudaUnregister(resource) => {
801                        edgefirst_tensor::gl_unregister_resource(resource);
802                    }
803                }
804                // Per-pass platform texture attachments (macOS pbuffer
805                // binds) are released once the message's GPU work has
806                // synced; deferred batches keep theirs until Flush.
807                gl_converter.end_gpu_pass_if_synced();
808            }
809            // Explicitly drop under the mutex so EGL teardown is serialized
810            // (Linux; ANGLE teardown is display-internal on macOS).
811            #[cfg(target_os = "linux")]
812            let _guard = super::context::GL_MUTEX
813                .lock()
814                .unwrap_or_else(|e| e.into_inner());
815            drop(gl_converter);
816        };
817
818        // let handle = tokio::task::spawn(func());
819        let handle = std::thread::spawn(func);
820
821        let caps = match create_ctx_recv.blocking_recv() {
822            Ok(Err(e)) => return Err(e),
823            Err(_) => {
824                return Err(Error::Internal(
825                    "GL converter error messaging closed without update".to_string(),
826                ));
827            }
828            Ok(Ok(caps)) => caps,
829        };
830
831        Ok(Self {
832            handle: Some(handle),
833            sender: Some(send),
834            caps,
835        })
836    }
837}
838
839impl ImageProcessorTrait for GLProcessorThreaded {
840    fn convert(
841        &mut self,
842        src: &TensorDyn,
843        dst: &mut TensorDyn,
844        rotation: crate::Rotation,
845        flip: Flip,
846        crop: Crop,
847    ) -> crate::Result<()> {
848        let (err_send, err_recv) = tokio::sync::oneshot::channel();
849        self.sender
850            .as_ref()
851            .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
852            .blocking_send(GLProcessorMessage::ImageConvert(
853                SendablePtr {
854                    ptr: NonNull::from(src),
855                    len: 1,
856                },
857                SendablePtr {
858                    ptr: NonNull::from(dst),
859                    len: 1,
860                },
861                rotation,
862                flip,
863                crop,
864                false, // eager: finish per convert
865                err_send,
866            ))
867            .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
868        err_recv.blocking_recv().map_err(|_| {
869            Error::Internal("GL converter error messaging closed without update".to_string())
870        })?
871    }
872
873    fn convert_deferred(
874        &mut self,
875        src: &TensorDyn,
876        dst: &mut TensorDyn,
877        rotation: crate::Rotation,
878        flip: Flip,
879        crop: Crop,
880    ) -> crate::Result<()> {
881        let (err_send, err_recv) = tokio::sync::oneshot::channel();
882        self.sender
883            .as_ref()
884            .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
885            .blocking_send(GLProcessorMessage::ImageConvert(
886                SendablePtr {
887                    ptr: NonNull::from(src),
888                    len: 1,
889                },
890                SendablePtr {
891                    ptr: NonNull::from(dst),
892                    len: 1,
893                },
894                rotation,
895                flip,
896                crop,
897                true, // defer: skip glFinish; flush() syncs the batch once
898                err_send,
899            ))
900            .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
901        err_recv.blocking_recv().map_err(|_| {
902            Error::Internal("GL converter error messaging closed without update".to_string())
903        })?
904    }
905
906    fn flush(&mut self) -> crate::Result<()> {
907        let (err_send, err_recv) = tokio::sync::oneshot::channel();
908        self.sender
909            .as_ref()
910            .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
911            .blocking_send(GLProcessorMessage::Flush(err_send))
912            .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
913        err_recv.blocking_recv().map_err(|_| {
914            Error::Internal("GL converter error messaging closed without update".to_string())
915        })?
916    }
917
918    fn draw_decoded_masks(
919        &mut self,
920        dst: &mut TensorDyn,
921        detect: &[crate::DetectBox],
922        segmentation: &[crate::Segmentation],
923        overlay: crate::MaskOverlay<'_>,
924    ) -> crate::Result<()> {
925        let (err_send, err_recv) = tokio::sync::oneshot::channel();
926        self.sender
927            .as_ref()
928            .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
929            .blocking_send(GLProcessorMessage::DrawDecodedMasks(
930                SendablePtr {
931                    ptr: NonNull::from(dst),
932                    len: 1,
933                },
934                SendablePtr {
935                    ptr: NonNull::new(detect.as_ptr() as *mut DetectBox).unwrap(),
936                    len: detect.len(),
937                },
938                SendablePtr {
939                    ptr: NonNull::new(segmentation.as_ptr() as *mut Segmentation).unwrap(),
940                    len: segmentation.len(),
941                },
942                overlay.opacity,
943                overlay.background.map(|bg| SendablePtr {
944                    ptr: NonNull::from(bg).cast::<TensorDyn>(),
945                    len: 1,
946                }),
947                overlay.letterbox,
948                overlay.color_mode,
949                err_send,
950            ))
951            .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
952        err_recv.blocking_recv().map_err(|_| {
953            Error::Internal("GL converter error messaging closed without update".to_string())
954        })?
955    }
956
957    fn draw_proto_masks(
958        &mut self,
959        dst: &mut TensorDyn,
960        detect: &[DetectBox],
961        proto_data: &ProtoData,
962        overlay: crate::MaskOverlay<'_>,
963    ) -> crate::Result<()> {
964        let (err_send, err_recv) = tokio::sync::oneshot::channel();
965        self.sender
966            .as_ref()
967            .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
968            .blocking_send(GLProcessorMessage::DrawProtoMasks(
969                SendablePtr {
970                    ptr: NonNull::from(dst),
971                    len: 1,
972                },
973                SendablePtr {
974                    ptr: NonNull::new(detect.as_ptr() as *mut DetectBox).unwrap(),
975                    len: detect.len(),
976                },
977                SendablePtr {
978                    ptr: NonNull::from(proto_data).cast::<ProtoData>(),
979                    len: 1,
980                },
981                overlay.opacity,
982                overlay.background.map(|bg| SendablePtr {
983                    ptr: NonNull::from(bg).cast::<TensorDyn>(),
984                    len: 1,
985                }),
986                overlay.letterbox,
987                overlay.color_mode,
988                err_send,
989            ))
990            .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
991        err_recv.blocking_recv().map_err(|_| {
992            Error::Internal("GL converter error messaging closed without update".to_string())
993        })?
994    }
995
996    fn set_class_colors(&mut self, colors: &[[u8; 4]]) -> Result<(), crate::Error> {
997        let (err_send, err_recv) = tokio::sync::oneshot::channel();
998        self.sender
999            .as_ref()
1000            .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1001            .blocking_send(GLProcessorMessage::SetColors(colors.to_vec(), err_send))
1002            .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1003        err_recv.blocking_recv().map_err(|_| {
1004            Error::Internal("GL converter error messaging closed without update".to_string())
1005        })?
1006    }
1007}
1008
1009impl GLProcessorThreaded {
1010    /// Sets the colorimetry/performance trade-off (see
1011    /// [`crate::ColorimetryMode`]). The `EDGEFIRST_COLORIMETRY` environment
1012    /// variable takes precedence — when set, this call logs and keeps the
1013    /// env-selected mode.
1014    pub fn set_colorimetry_mode(&mut self, mode: crate::ColorimetryMode) -> Result<(), Error> {
1015        let (err_send, err_recv) = tokio::sync::oneshot::channel();
1016        self.sender
1017            .as_ref()
1018            .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1019            .blocking_send(GLProcessorMessage::SetColorimetryMode(mode, err_send))
1020            .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1021        err_recv.blocking_recv().map_err(|_| {
1022            Error::Internal("GL converter error messaging closed without update".to_string())
1023        })?
1024    }
1025
1026    /// Sets the interpolation mode for int8 proto textures.
1027    pub fn set_int8_interpolation_mode(
1028        &mut self,
1029        mode: Int8InterpolationMode,
1030    ) -> Result<(), crate::Error> {
1031        let (err_send, err_recv) = tokio::sync::oneshot::channel();
1032        self.sender
1033            .as_ref()
1034            .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1035            .blocking_send(GLProcessorMessage::SetInt8Interpolation(mode, err_send))
1036            .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1037        err_recv.blocking_recv().map_err(|_| {
1038            Error::Internal("GL converter error messaging closed without update".to_string())
1039        })?
1040    }
1041
1042    /// Snapshot the EGLImage cache counters (src, dst, NV R8) from the GL
1043    /// thread. Steady-state tests capture this after warmup and after an
1044    /// N-frame loop over a fixed buffer pool and assert
1045    /// [`total_misses`](super::cache::GlCacheStats::total_misses) stays flat —
1046    /// any increase means a convert re-imported a buffer it should have found
1047    /// cached (the cache-behavior equality gate for GL refactors).
1048    pub fn egl_cache_stats(&self) -> Result<super::cache::GlCacheStats, Error> {
1049        let (send, recv) = tokio::sync::oneshot::channel();
1050        self.sender
1051            .as_ref()
1052            .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1053            .blocking_send(GLProcessorMessage::EglCacheStats(send))
1054            .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1055        recv.blocking_recv().map_err(|_| {
1056            Error::Internal("GL converter error messaging closed without update".to_string())
1057        })?
1058    }
1059
1060    /// Convert and return a native sync-fence fd that signals when the
1061    /// GPU work completes, instead of blocking the CPU on `glFinish` —
1062    /// the GL→NPU handoff (`EGL_ANDROID_native_fence_sync`).
1063    ///
1064    /// `Ok(None)` means the convert completed with the normal blocking
1065    /// semantics (platform/driver has no native fence, or a pure-CPU-
1066    /// readback path was taken) — the destination is already safe to
1067    /// read. `Ok(Some(fd))` means the caller (or the NPU runtime, via
1068    /// `ANeuralNetworksExecution_startComputeWithDependencies`) must wait
1069    /// on the fd before consuming the destination buffer.
1070    pub fn convert_with_fence(
1071        &mut self,
1072        src: &TensorDyn,
1073        dst: &mut TensorDyn,
1074        rotation: crate::Rotation,
1075        flip: Flip,
1076        crop: Crop,
1077    ) -> Result<Option<std::os::fd::OwnedFd>, Error> {
1078        use crate::ImageProcessorTrait as _;
1079        if !self.caps.native_fence_sync {
1080            // No fence on this display: the plain blocking convert IS the
1081            // fenced contract with `None` — skip the special round-trip.
1082            self.convert(src, dst, rotation, flip, crop)?;
1083            return Ok(None);
1084        }
1085        let (send, recv) = tokio::sync::oneshot::channel();
1086        self.sender
1087            .as_ref()
1088            .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1089            .blocking_send(GLProcessorMessage::ImageConvertFenced(
1090                SendablePtr {
1091                    ptr: NonNull::from(src),
1092                    len: 1,
1093                },
1094                SendablePtr {
1095                    ptr: NonNull::from(dst),
1096                    len: 1,
1097                },
1098                rotation,
1099                flip,
1100                crop,
1101                send,
1102            ))
1103            .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1104        recv.blocking_recv().map_err(|_| {
1105            Error::Internal("GL converter error messaging closed without update".to_string())
1106        })?
1107    }
1108
1109    /// Snapshot the per-convert source-feed counters from the GL thread.
1110    /// A pipeline that expects end-to-end zero-copy asserts
1111    /// [`src_uploads`](super::cache::ConvertStats::src_uploads) stays flat
1112    /// (every frame fed by import, never by a CPU map + upload).
1113    pub fn convert_stats(&self) -> Result<super::cache::ConvertStats, Error> {
1114        let (send, recv) = tokio::sync::oneshot::channel();
1115        self.sender
1116            .as_ref()
1117            .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1118            .blocking_send(GLProcessorMessage::ConvertStats(send))
1119            .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1120        recv.blocking_recv().map_err(|_| {
1121            Error::Internal("GL converter error messaging closed without update".to_string())
1122        })?
1123    }
1124
1125    /// Create a PBO-backed [`Tensor<u8>`] image on the GL thread.
1126    pub fn create_pbo_image(
1127        &self,
1128        width: usize,
1129        height: usize,
1130        format: edgefirst_tensor::PixelFormat,
1131    ) -> Result<edgefirst_tensor::Tensor<u8>, Error> {
1132        let sender = self
1133            .sender
1134            .as_ref()
1135            .ok_or(Error::OpenGl("GL processor is shutting down".to_string()))?;
1136
1137        let size = pbo_elem_count(width, height, format)
1138            .filter(|&s| s != 0)
1139            .ok_or_else(|| Error::OpenGl("Invalid image dimensions".to_string()))?;
1140
1141        // Allocate PBO on the GL thread
1142        let (tx, rx) = tokio::sync::oneshot::channel();
1143        sender
1144            .blocking_send(GLProcessorMessage::PboCreate(size, tx))
1145            .map_err(|_| Error::OpenGl("GL thread channel closed".to_string()))?;
1146        let buffer_id = rx
1147            .blocking_recv()
1148            .map_err(|_| Error::OpenGl("GL thread did not respond".to_string()))??;
1149
1150        let ops: std::sync::Arc<dyn edgefirst_tensor::PboOps> = std::sync::Arc::new(GlPboOps {
1151            sender: sender.downgrade(),
1152        });
1153
1154        let shape = pbo_shape(width, height, format);
1155
1156        let pbo_tensor =
1157            edgefirst_tensor::PboTensor::<u8>::from_pbo(buffer_id, size, &shape, None, ops)
1158                .map_err(|e| Error::OpenGl(format!("PBO tensor creation failed: {e:?}")))?;
1159        let mut tensor = edgefirst_tensor::Tensor::from_pbo(pbo_tensor);
1160        tensor
1161            .set_format(format)
1162            .map_err(|e| Error::OpenGl(format!("Failed to set format on PBO tensor: {e:?}")))?;
1163        // Register the PBO with CUDA (best-effort; no-op without libcudart) so
1164        // `cuda_map()` can yield a device pointer — matching the float PBO path
1165        // in `create_pbo_image_dtype`. Without this, u8 PBOs were never
1166        // CUDA-interop-capable, so the codec's nvJPEG backend (and any CUDA
1167        // consumer) could not use a `create_image`-allocated PBO destination.
1168        // The i8 transmute by the caller preserves the attached handle.
1169        register_pbo_cuda(&mut tensor, buffer_id, size, sender);
1170        Ok(tensor)
1171    }
1172
1173    /// Create a PBO-backed [`TensorDyn`] image on the GL thread with the given dtype.
1174    ///
1175    /// Sizes the underlying GL buffer by `elems * dtype.size()` and wraps it in
1176    /// the appropriately-typed [`PboTensor`]. Supports `DType::U8`, `DType::F16`,
1177    /// and `DType::F32`; returns an error for other dtypes.
1178    pub(crate) fn create_pbo_image_dtype(
1179        &self,
1180        width: usize,
1181        height: usize,
1182        format: edgefirst_tensor::PixelFormat,
1183        dtype: edgefirst_tensor::DType,
1184    ) -> Result<TensorDyn, Error> {
1185        let sender = self
1186            .sender
1187            .as_ref()
1188            .ok_or(Error::OpenGl("GL processor is shutting down".to_string()))?;
1189
1190        let elems = pbo_elem_count(width, height, format)
1191            .filter(|&e| e != 0)
1192            .ok_or_else(|| Error::OpenGl("Invalid image dimensions".to_string()))?;
1193
1194        let size = elems
1195            .checked_mul(dtype.size())
1196            .ok_or_else(|| Error::OpenGl("PBO size overflow".to_string()))?;
1197
1198        // Allocate PBO on the GL thread
1199        let (tx, rx) = tokio::sync::oneshot::channel();
1200        sender
1201            .blocking_send(GLProcessorMessage::PboCreate(size, tx))
1202            .map_err(|_| Error::OpenGl("GL thread channel closed".to_string()))?;
1203        let buffer_id = rx
1204            .blocking_recv()
1205            .map_err(|_| Error::OpenGl("GL thread did not respond".to_string()))??;
1206
1207        let ops: std::sync::Arc<dyn edgefirst_tensor::PboOps> = std::sync::Arc::new(GlPboOps {
1208            sender: sender.downgrade(),
1209        });
1210
1211        let shape = pbo_shape(width, height, format);
1212
1213        let map_err = |e: edgefirst_tensor::Error| {
1214            Error::OpenGl(format!("PBO tensor creation failed: {e:?}"))
1215        };
1216        let set_err = |e: edgefirst_tensor::Error| {
1217            Error::OpenGl(format!("Failed to set format on PBO tensor: {e:?}"))
1218        };
1219
1220        match dtype {
1221            edgefirst_tensor::DType::U8 => {
1222                let pbo =
1223                    edgefirst_tensor::PboTensor::<u8>::from_pbo(buffer_id, size, &shape, None, ops)
1224                        .map_err(map_err)?;
1225                let mut t = edgefirst_tensor::Tensor::from_pbo(pbo);
1226                t.set_format(format).map_err(set_err)?;
1227                register_pbo_cuda(&mut t, buffer_id, size, sender);
1228                Ok(TensorDyn::from(t))
1229            }
1230            edgefirst_tensor::DType::F16 => {
1231                let pbo = edgefirst_tensor::PboTensor::<edgefirst_tensor::f16>::from_pbo(
1232                    buffer_id, size, &shape, None, ops,
1233                )
1234                .map_err(map_err)?;
1235                let mut t = edgefirst_tensor::Tensor::from_pbo(pbo);
1236                t.set_format(format).map_err(set_err)?;
1237                register_pbo_cuda(&mut t, buffer_id, size, sender);
1238                Ok(TensorDyn::from(t))
1239            }
1240            edgefirst_tensor::DType::F32 => {
1241                let pbo = edgefirst_tensor::PboTensor::<f32>::from_pbo(
1242                    buffer_id, size, &shape, None, ops,
1243                )
1244                .map_err(map_err)?;
1245                let mut t = edgefirst_tensor::Tensor::from_pbo(pbo);
1246                t.set_format(format).map_err(set_err)?;
1247                register_pbo_cuda(&mut t, buffer_id, size, sender);
1248                Ok(TensorDyn::from(t))
1249            }
1250            other => Err(Error::OpenGl(format!("unsupported PBO dtype {other:?}"))),
1251        }
1252    }
1253
1254    /// Returns the active transfer backend.
1255    pub(crate) fn transfer_backend(&self) -> TransferBackend {
1256        self.caps.transfer_backend
1257    }
1258
1259    /// Report which float dtypes the GPU can render to.
1260    ///
1261    /// Values are probed once at construction time and adjusted for
1262    /// Vivante GC7000UL, whose float readback latency (170-320 ms) makes
1263    /// GL float destinations impractical; `ImageProcessor::convert()` falls
1264    /// back to CPU float output (normalized to `[0, 1]`) for these targets.
1265    pub(crate) fn supported_render_dtypes(&self) -> crate::RenderDtypeSupport {
1266        self.caps.render_dtypes
1267    }
1268}
1269
1270impl Drop for GLProcessorThreaded {
1271    fn drop(&mut self) {
1272        drop(self.sender.take());
1273        let _ = self.handle.take().and_then(|h| h.join().ok());
1274    }
1275}
1276
1277// `pbo_elem_count` / `pbo_shape` are pure (no GL), so they are unit-testable
1278// without a GPU. The overflow→None arm of `pbo_elem_count` guards against an
1279// undersized PBO allocation, so it is worth pinning explicitly.
1280#[cfg(test)]
1281#[cfg_attr(coverage_nightly, coverage(off))]
1282mod tests {
1283    use super::{pbo_elem_count, pbo_shape};
1284    use edgefirst_tensor::PixelFormat;
1285
1286    #[test]
1287    fn elem_count_per_format() {
1288        // Packed RGBA: w*h*4.
1289        assert_eq!(pbo_elem_count(8, 4, PixelFormat::Rgba), Some(8 * 4 * 4));
1290        // Packed RGB: w*h*3.
1291        assert_eq!(pbo_elem_count(8, 4, PixelFormat::Rgb), Some(8 * 4 * 3));
1292        // NV12 semiplanar: w*h*3/2.
1293        assert_eq!(pbo_elem_count(8, 4, PixelFormat::Nv12), Some(8 * 4 * 3 / 2));
1294        // NV16 semiplanar: w*h*2.
1295        assert_eq!(pbo_elem_count(8, 4, PixelFormat::Nv16), Some(8 * 4 * 2));
1296    }
1297
1298    #[test]
1299    fn elem_count_overflow_is_none() {
1300        // w*h already overflows usize → None (never a wrapped, undersized count).
1301        assert_eq!(pbo_elem_count(usize::MAX, 2, PixelFormat::Rgba), None);
1302        // w*h fits but *channels overflows → None.
1303        assert_eq!(pbo_elem_count(usize::MAX, 1, PixelFormat::Rgb), None);
1304    }
1305
1306    #[test]
1307    fn shape_per_format() {
1308        // Planar: [channels, height, width].
1309        assert_eq!(pbo_shape(8, 4, PixelFormat::PlanarRgb), vec![3, 4, 8]);
1310        // SemiPlanar NV12: [height*3/2, width].
1311        assert_eq!(pbo_shape(8, 4, PixelFormat::Nv12), vec![4 * 3 / 2, 8]);
1312        // SemiPlanar NV16: [height*2, width].
1313        assert_eq!(pbo_shape(8, 4, PixelFormat::Nv16), vec![4 * 2, 8]);
1314        // Packed: [height, width, channels].
1315        assert_eq!(pbo_shape(8, 4, PixelFormat::Rgba), vec![4, 8, 4]);
1316    }
1317}