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_tensor::{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::{ConvertOptions, Error, ImageProcessorTrait};
14use edgefirst_tensor::TensorDyn;
15
16#[allow(clippy::type_complexity)]
17enum GLProcessorMessage {
18    ImageConvert(
19        SendablePtr<TensorDyn>,
20        SendablePtr<TensorDyn>,
21        ConvertOptions,
22        bool, // defer: skip the per-tile glFinish (batch); flush syncs once
23        tokio::sync::oneshot::Sender<Result<(), Error>>,
24    ),
25    /// Convert without the terminal CPU sync, replying with a native
26    /// fence fd guarding the GPU work (the GL→NPU handoff). `Ok(None)` =
27    /// converted with blocking semantics (no fence on this platform).
28    ImageConvertFenced(
29        SendablePtr<TensorDyn>,
30        SendablePtr<TensorDyn>,
31        ConvertOptions,
32        tokio::sync::oneshot::Sender<Result<Option<super::CompletionFence>, Error>>,
33    ),
34    /// Complete any deferred batch render with a single GPU sync.
35    Flush(tokio::sync::oneshot::Sender<Result<(), Error>>),
36    SetColors(
37        Vec<[u8; 4]>,
38        tokio::sync::oneshot::Sender<Result<(), Error>>,
39    ),
40    SetBoxLineWidth(u32, tokio::sync::oneshot::Sender<Result<(), Error>>),
41    DrawDecodedMasks(
42        SendablePtr<TensorDyn>,
43        SendablePtr<DetectBox>,
44        SendablePtr<Segmentation>,
45        f32,                            // opacity
46        Option<SendablePtr<TensorDyn>>, // background
47        Option<[f32; 4]>,               // content region
48        crate::ColorMode,
49        tokio::sync::oneshot::Sender<Result<(), Error>>,
50    ),
51    DrawProtoMasks(
52        SendablePtr<TensorDyn>,
53        SendablePtr<DetectBox>,
54        SendablePtr<ProtoData>,
55        f32,                            // opacity
56        Option<SendablePtr<TensorDyn>>, // background
57        Option<[f32; 4]>,               // content region
58        crate::ColorMode,
59        tokio::sync::oneshot::Sender<Result<(), Error>>,
60    ),
61    SetInt8Interpolation(
62        Int8InterpolationMode,
63        tokio::sync::oneshot::Sender<Result<(), Error>>,
64    ),
65    /// Set the colorimetry/performance trade-off (`ColorimetryMode`).
66    SetColorimetryMode(
67        crate::ColorimetryMode,
68        tokio::sync::oneshot::Sender<Result<(), Error>>,
69    ),
70    /// Set how resizing converts resample (`Interpolation`).
71    SetInterpolation(
72        crate::Interpolation,
73        tokio::sync::oneshot::Sender<Result<(), Error>>,
74    ),
75    /// Test hook (issue #175): make every zero-copy destination import fail.
76    #[cfg(test)]
77    SetFailDstImport(bool, tokio::sync::oneshot::Sender<Result<(), Error>>),
78    /// Snapshot the EGLImage cache counters (steady-state import assertions).
79    EglCacheStats(tokio::sync::oneshot::Sender<Result<super::cache::GlCacheStats, Error>>),
80    /// Snapshot the per-convert source-feed counters (zero-copy observability).
81    ConvertStats(tokio::sync::oneshot::Sender<Result<super::cache::ConvertStats, Error>>),
82    PboCreate(
83        usize, // buffer size in bytes
84        tokio::sync::oneshot::Sender<Result<u32, Error>>,
85    ),
86    PboMap(
87        u32,   // buffer_id
88        usize, // size
89        tokio::sync::oneshot::Sender<Result<edgefirst_tensor::PboMapping, Error>>,
90    ),
91    PboUnmap(
92        u32, // buffer_id
93        tokio::sync::oneshot::Sender<Result<(), Error>>,
94    ),
95    PboDelete(u32), // fire-and-forget, no reply
96    CudaRegisterBuffer(u32, tokio::sync::oneshot::Sender<Option<usize>>),
97    CudaMap(usize, tokio::sync::oneshot::Sender<Option<(usize, usize)>>),
98    CudaUnmap(usize),      // fire-and-forget
99    CudaUnregister(usize), // fire-and-forget
100}
101
102/// Compute the flat element count for a PBO image buffer of the given format.
103///
104/// NV12 and NV16 are semiplanar with non-trivial element counts; all other
105/// formats use `width * height * channels`.
106///
107/// Returns `None` on `usize` overflow. A wrapped element count would size
108/// the PBO too small and corrupt memory on readback, so callers must treat
109/// `None` as an invalid-dimensions error (the same way they already reject
110/// a zero count).
111fn pbo_elem_count(
112    width: usize,
113    height: usize,
114    format: edgefirst_tensor::PixelFormat,
115) -> Option<usize> {
116    // Element count == product of the canonical image shape (the PBO holds u8,
117    // one byte per element). Delegating to `allocation_shape` keeps the combined
118    // semi-planar height in lockstep with every other consumer (and is exact
119    // for odd heights — `height*3/2` truncation under-sized odd-H NV12 by a
120    // row, and the old NV24 arm fell through to `channels` = far too small).
121    // `checked_mul` so a wrapped count can't silently undersize the PBO.
122    format
123        .allocation_shape(width, height)?
124        .iter()
125        .try_fold(1usize, |acc, &d| acc.checked_mul(d))
126}
127
128/// Compute the tensor shape for a PBO image of the given format — the canonical
129/// [`PixelFormat::allocation_shape`] (Planar `[C,H,W]`, SemiPlanar `[total_h, W]`
130/// with the odd-height-exact combined-plane height, Packed `[H,W,C]`).
131fn pbo_shape(width: usize, height: usize, format: edgefirst_tensor::PixelFormat) -> Vec<usize> {
132    format
133        .allocation_shape(width, height)
134        .unwrap_or_else(|| vec![height, width, format.channels()])
135}
136
137/// Best-effort registration of a freshly-created PBO with CUDA GL interop.
138///
139/// When libcudart is present, asks the GL worker (where the GL context is
140/// current) to call `cudaGraphicsGLRegisterBuffer` for `buffer_id`, then
141/// attaches a [`CudaHandle`] so `cuda_map()` can later yield a linear,
142/// 256-byte-aligned device pointer for TensorRT. On any failure (no CUDA,
143/// channel closed, registration rejected) the handle is left unset and the
144/// PBO is still usable as a normal CPU/GL buffer.
145fn register_pbo_cuda<T>(
146    tensor: &mut edgefirst_tensor::Tensor<T>,
147    buffer_id: u32,
148    size: usize,
149    sender: &Sender<GLProcessorMessage>,
150) where
151    T: num_traits::Num + Clone + std::fmt::Debug + Send + Sync + edgefirst_tensor::Element,
152{
153    if !edgefirst_tensor::is_cuda_available() {
154        return;
155    }
156    let (tx, rx) = tokio::sync::oneshot::channel();
157    if sender
158        .blocking_send(GLProcessorMessage::CudaRegisterBuffer(buffer_id, tx))
159        .is_ok()
160    {
161        if let Ok(Some(resource)) = rx.blocking_recv() {
162            let ops = std::sync::Arc::new(GlCudaOps {
163                sender: sender.downgrade(),
164            });
165            tensor.set_cuda_handle(edgefirst_tensor::CudaHandle::new_gl(
166                resource as *mut std::ffi::c_void,
167                size,
168                ops,
169            ));
170        }
171    }
172}
173
174/// Implements PboOps by sending commands to the GL thread.
175///
176/// Uses a `WeakSender` so that PBO images don't keep the GL thread's channel
177/// alive. When the `GLProcessorThreaded` is dropped, its `Sender` is the last
178/// strong reference — dropping it closes the channel and lets the GL thread
179/// exit. PBO operations after that return `PboDisconnected`.
180struct GlPboOps {
181    sender: WeakSender<GLProcessorMessage>,
182    /// Which GL worker owns the context these buffers live in. Compared
183    /// against [`GL_WORKER`] so a re-entrant map is serviced inline ONLY by
184    /// the worker whose context actually holds the buffer — a different
185    /// processor's worker must still go through the channel.
186    worker_id: u64,
187}
188
189/// Next id handed to a GL worker. Ids are per-process and never reused.
190static NEXT_GL_WORKER_ID: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
191
192/// Identity and PBO bookkeeping of the GL worker running on this thread.
193struct GlWorkerState {
194    id: u64,
195    /// Full per-message serialization: this thread holds the process-wide GL
196    /// lock for the whole message, so no other worker can run until it returns.
197    serializes: bool,
198    /// Buffers currently `glMapBufferRange`-ed against this context.
199    mapped: std::collections::HashSet<u32>,
200}
201
202thread_local! {
203    /// `Some(state)` on a GL worker thread, `None` everywhere else.
204    ///
205    /// This exists because a GL message handler can itself need a PBO map:
206    /// a mask draw into a PBO destination the caller holds mapped reads it
207    /// back through `TensorTrait::map` -> `PboHandle::acquire_map` ->
208    /// `GlPboOps::map_buffer`. Routing that through the channel would have the
209    /// worker `blocking_send` to the queue it is itself the only consumer of —
210    /// a self-deadlock (observed: a 1-in-5 hang on the desktop NVIDIA PBO path).
211    /// The worker already holds the context current on this thread, so it
212    /// performs the map directly instead.
213    ///
214    /// The mapped-buffer set lives here rather than on the worker's stack so the
215    /// inline path and the message path share one view of which buffers are
216    /// mapped — `glMapBufferRange` on an already-mapped buffer is undefined
217    /// behaviour, and the two paths must not be able to double-map.
218    static GL_WORKER: std::cell::RefCell<Option<GlWorkerState>> =
219        const { std::cell::RefCell::new(None) };
220}
221
222/// The mapped-PBO set for the current GL worker thread, or `None` off-worker.
223/// The closure must not re-enter `GL_WORKER` (keep it to set bookkeeping — no
224/// GL calls inside).
225fn with_worker_mapped<R>(f: impl FnOnce(&mut std::collections::HashSet<u32>) -> R) -> Option<R> {
226    GL_WORKER.with(|c| c.borrow_mut().as_mut().map(|s| f(&mut s.mapped)))
227}
228
229/// True when the calling thread is the GL worker that owns `worker_id`.
230fn on_owning_gl_worker(worker_id: u64) -> bool {
231    GL_WORKER.with(|c| matches!(*c.borrow(), Some(ref s) if s.id == worker_id))
232}
233
234/// True when this thread is a GL worker holding the process-wide per-message
235/// lock, so no other worker can make progress until this message returns.
236fn on_serializing_gl_worker() -> bool {
237    GL_WORKER.with(|c| matches!(*c.borrow(), Some(ref s) if s.serializes))
238}
239
240/// A PBO owned by another worker's context, reached from a worker holding the
241/// process-wide per-message lock. The owning worker cannot take that lock until
242/// this message returns, so sending to it would never be answered.
243fn cross_context_pbo_error(op: &str, buffer_id: u32) -> edgefirst_tensor::Error {
244    edgefirst_tensor::Error::InvalidOperation(format!(
245        "PBO {buffer_id}: cannot {op} a buffer owned by another GL context while \
246         this one holds the serialization lock; convert it through the processor \
247         that created it"
248    ))
249}
250
251thread_local! {
252    /// Set when a re-entrant inline PBO op panicked; the worker loop folds it
253    /// into `poisoned` after the message that raised it.
254    static GL_WORKER_POISON: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
255}
256
257/// Run a re-entrant PBO op inline on the owning worker, turning a panic into
258/// an error exactly as the channel path's `reply_caught` does.
259///
260/// The unwind MUST stop here. `PboHandle::acquire_map` publishes
261/// `MapState::Mapping` before calling `map_buffer` and clears it only on
262/// return, so a panic crossing it strands that handle in `Mapping` and every
263/// later map on it waits on the condvar forever.
264fn inline_gl_op<T>(what: &str, f: impl FnOnce() -> Result<T, Error>) -> Result<T, Error> {
265    match std::panic::catch_unwind(AssertUnwindSafe(f)) {
266        Ok(result) => result,
267        Err(e) => {
268            GL_WORKER_POISON.with(|p| p.set(true));
269            Err(Error::Internal(format!(
270                "GL thread panicked during {what}: {}",
271                panic_message(e.as_ref()),
272            )))
273        }
274    }
275}
276
277// SAFETY: GlPboOps runs every GL operation on the thread owning the context --
278// via the channel, or inline when that thread is already this one (see
279// `GL_WORKER`). `map_buffer` returns a CPU-visible pointer from `glMapBufferRange`
280// that remains valid until `unmap_buffer` calls `glUnmapBuffer` on the GL thread.
281// `delete_buffer` sends a fire-and-forget deletion command to the GL thread.
282unsafe impl edgefirst_tensor::PboOps for GlPboOps {
283    fn map_buffer(
284        &self,
285        buffer_id: u32,
286        size: usize,
287    ) -> edgefirst_tensor::Result<edgefirst_tensor::PboMapping> {
288        // Re-entrant call from inside a GL message handler on the worker that
289        // owns this context (e.g. a mask draw reading back into a mapped PBO
290        // destination). The context is already current on this thread,
291        // and sending would block on the queue this thread alone drains.
292        if on_owning_gl_worker(self.worker_id) {
293            return inline_gl_op("PboMap", || pbo_map_on_gl_thread(buffer_id, size)).map_err(|e| {
294                edgefirst_tensor::Error::NotImplemented(format!("GL PBO map failed: {e:?}"))
295            });
296        }
297        // Foreign context, and this thread holds the lock the owner needs.
298        if on_serializing_gl_worker() {
299            return Err(cross_context_pbo_error("map", buffer_id));
300        }
301        let sender = self
302            .sender
303            .upgrade()
304            .ok_or(edgefirst_tensor::Error::PboDisconnected)?;
305        let (tx, rx) = tokio::sync::oneshot::channel();
306        sender
307            .blocking_send(GLProcessorMessage::PboMap(buffer_id, size, tx))
308            .map_err(|_| edgefirst_tensor::Error::PboDisconnected)?;
309        rx.blocking_recv()
310            .map_err(|_| edgefirst_tensor::Error::PboDisconnected)?
311            .map_err(|e| {
312                edgefirst_tensor::Error::NotImplemented(format!("GL PBO map failed: {e:?}"))
313            })
314    }
315
316    fn unmap_buffer(&self, buffer_id: u32) -> edgefirst_tensor::Result<()> {
317        // See `map_buffer` — same re-entrancy, same reasoning.
318        if on_owning_gl_worker(self.worker_id) {
319            return inline_gl_op("PboUnmap", || pbo_unmap_on_gl_thread(buffer_id)).map_err(|e| {
320                edgefirst_tensor::Error::NotImplemented(format!("GL PBO unmap failed: {e:?}"))
321            });
322        }
323        // See `map_buffer`.
324        if on_serializing_gl_worker() {
325            return Err(cross_context_pbo_error("unmap", buffer_id));
326        }
327        let sender = self
328            .sender
329            .upgrade()
330            .ok_or(edgefirst_tensor::Error::PboDisconnected)?;
331        let (tx, rx) = tokio::sync::oneshot::channel();
332        sender
333            .blocking_send(GLProcessorMessage::PboUnmap(buffer_id, tx))
334            .map_err(|_| edgefirst_tensor::Error::PboDisconnected)?;
335        rx.blocking_recv()
336            .map_err(|_| edgefirst_tensor::Error::PboDisconnected)?
337            .map_err(|e| {
338                edgefirst_tensor::Error::NotImplemented(format!("GL PBO unmap failed: {e:?}"))
339            })
340    }
341
342    fn delete_buffer(&self, buffer_id: u32) {
343        // See `map_buffer` — a drop mid-convert must not queue to itself.
344        if on_owning_gl_worker(self.worker_id) {
345            with_worker_mapped(|set| set.remove(&buffer_id));
346            // Reached from `PboHandle::Drop`; an unwind out of a drop that is
347            // itself unwinding aborts the process.
348            let _ = inline_gl_op("PboDelete", || {
349                unsafe { edgefirst_gl::gl::DeleteBuffers(1, &buffer_id) };
350                Ok(())
351            });
352            return;
353        }
354        if let Some(sender) = self.sender.upgrade() {
355            // See `map_buffer`: blocking here would deadlock the same way.
356            if on_serializing_gl_worker() {
357                if sender
358                    .try_send(GLProcessorMessage::PboDelete(buffer_id))
359                    .is_err()
360                {
361                    log::warn!(
362                        "PBO {buffer_id}: cross-context delete skipped to avoid a deadlock; \
363                         the buffer is freed when its context is destroyed"
364                    );
365                }
366                return;
367            }
368            let _ = sender.blocking_send(GLProcessorMessage::PboDelete(buffer_id));
369        }
370    }
371}
372
373/// Routes CUDA GL-interop map/unmap/unregister to the GL thread.
374///
375/// Mirrors [`GlPboOps`]: holds a [`WeakSender`] so a registered PBO doesn't
376/// keep the GL thread's channel alive. The CUDA primitives
377/// (`cudaGraphicsMapResources` etc.) require the GL context to be current,
378/// so they must execute on the dedicated GL worker thread.
379struct GlCudaOps {
380    sender: WeakSender<GLProcessorMessage>,
381}
382
383impl edgefirst_tensor::CudaGlOps for GlCudaOps {
384    fn map(&self, resource: *mut std::ffi::c_void) -> Option<(*mut std::ffi::c_void, usize)> {
385        let sender = self.sender.upgrade()?;
386        let (tx, rx) = tokio::sync::oneshot::channel();
387        sender
388            .blocking_send(GLProcessorMessage::CudaMap(resource as usize, tx))
389            .ok()?;
390        rx.blocking_recv()
391            .ok()?
392            .map(|(p, l)| (p as *mut std::ffi::c_void, l))
393    }
394
395    fn unmap(&self, resource: *mut std::ffi::c_void) {
396        if let Some(sender) = self.sender.upgrade() {
397            let _ = sender.blocking_send(GLProcessorMessage::CudaUnmap(resource as usize));
398        }
399    }
400
401    fn unregister(&self, resource: *mut std::ffi::c_void) {
402        if let Some(sender) = self.sender.upgrade() {
403            let _ = sender.blocking_send(GLProcessorMessage::CudaUnregister(resource as usize));
404        }
405    }
406}
407
408/// OpenGL multi-threaded image converter. The actual conversion is done in a
409/// separate rendering thread, as OpenGL contexts are not thread-safe. This can
410/// be safely sent between threads. The `convert()` call sends the conversion
411/// request to the rendering thread and waits for the result.
412#[derive(Debug)]
413pub struct GLProcessorThreaded {
414    // This is only None when the converter is being dropped.
415    handle: Option<JoinHandle<()>>,
416
417    // This is only None when the converter is being dropped.
418    sender: Option<Sender<GLProcessorMessage>>,
419    /// Identity of this processor's worker thread, handed to every `GlPboOps`
420    /// it creates so a re-entrant map can be matched to the owning context.
421    worker_id: u64,
422    /// The PBOs this processor handed out, shared with its worker's
423    /// `GLProcessorST`: only these may be bound in the worker's context.
424    owned_pbos: std::sync::Arc<super::processor::OwnedPbos>,
425    /// Immutable capability surface (transfer backend, float render
426    /// support, serialization policy), captured once from the worker at
427    /// construction. See `PlatformCaps` in `platform/mod.rs`.
428    caps: super::platform::PlatformCaps,
429}
430
431unsafe impl Send for GLProcessorThreaded {}
432unsafe impl Sync for GLProcessorThreaded {}
433
434struct SendablePtr<T: Send> {
435    ptr: NonNull<T>,
436    len: usize,
437}
438
439unsafe impl<T> Send for SendablePtr<T> where T: Send {}
440
441/// Extract a human-readable message from a `catch_unwind` panic payload.
442fn panic_message(info: &(dyn std::any::Any + Send)) -> String {
443    if let Some(s) = info.downcast_ref::<&str>() {
444        s.to_string()
445    } else if let Some(s) = info.downcast_ref::<String>() {
446        s.clone()
447    } else {
448        "unknown panic".to_string()
449    }
450}
451
452fn reply_caught<T>(
453    result: std::thread::Result<crate::Result<T>>,
454    resp: tokio::sync::oneshot::Sender<crate::Result<T>>,
455    poisoned: &mut bool,
456    what: &str,
457) {
458    let _ = resp.send(match result {
459        Ok(res) => res,
460        Err(e) => {
461            *poisoned = true;
462            Err(crate::Error::Internal(format!(
463                "GL thread panicked during {what}: {}",
464                panic_message(e.as_ref()),
465            )))
466        }
467    });
468}
469
470fn reject_poisoned_message(msg: GLProcessorMessage) {
471    let poison_err =
472        crate::Error::Internal("GL context is poisoned after a prior panic".to_string());
473    match msg {
474        GLProcessorMessage::ImageConvertFenced(.., resp) => {
475            let _ = resp.send(Err(poison_err));
476        }
477        GLProcessorMessage::ImageConvert(.., resp) => {
478            let _ = resp.send(Err(poison_err));
479        }
480        GLProcessorMessage::Flush(resp) => {
481            let _ = resp.send(Err(poison_err));
482        }
483        GLProcessorMessage::DrawDecodedMasks(.., resp) => {
484            let _ = resp.send(Err(poison_err));
485        }
486        GLProcessorMessage::DrawProtoMasks(.., resp) => {
487            let _ = resp.send(Err(poison_err));
488        }
489        GLProcessorMessage::SetColors(_, resp) => {
490            let _ = resp.send(Err(poison_err));
491        }
492        GLProcessorMessage::SetBoxLineWidth(_, resp) => {
493            let _ = resp.send(Err(poison_err));
494        }
495        GLProcessorMessage::SetInt8Interpolation(_, resp) => {
496            let _ = resp.send(Err(poison_err));
497        }
498        GLProcessorMessage::SetColorimetryMode(_, resp) => {
499            let _ = resp.send(Err(poison_err));
500        }
501        GLProcessorMessage::SetInterpolation(_, resp) => {
502            let _ = resp.send(Err(poison_err));
503        }
504        #[cfg(test)]
505        GLProcessorMessage::SetFailDstImport(_, resp) => {
506            let _ = resp.send(Err(poison_err));
507        }
508        GLProcessorMessage::EglCacheStats(resp) => {
509            let _ = resp.send(Err(poison_err));
510        }
511        GLProcessorMessage::ConvertStats(resp) => {
512            let _ = resp.send(Err(poison_err));
513        }
514        GLProcessorMessage::PboCreate(_, resp) => {
515            let _ = resp.send(Err(poison_err));
516        }
517        GLProcessorMessage::PboMap(_, _, resp) => {
518            let _ = resp.send(Err(poison_err));
519        }
520        GLProcessorMessage::PboUnmap(_, resp) => {
521            let _ = resp.send(Err(poison_err));
522        }
523        GLProcessorMessage::PboDelete(_) => {}
524        GLProcessorMessage::CudaRegisterBuffer(_, resp) => {
525            let _ = resp.send(None);
526        }
527        GLProcessorMessage::CudaMap(_, resp) => {
528            let _ = resp.send(None);
529        }
530        // GL context is poisoned/dead — the GL buffer is unusable, so unmapping/
531        // unregistering the CUDA interop is moot (the resource is reclaimed at
532        // process exit). Same accepted "GL is gone" no-op as PboDelete above.
533        GLProcessorMessage::CudaUnmap(_) => {}
534        GLProcessorMessage::CudaUnregister(_) => {}
535    }
536}
537
538fn run_gl_worker(
539    worker_id: u64,
540    owned_pbos: std::sync::Arc<super::processor::OwnedPbos>,
541    kind: Option<EglDisplayKind>,
542    capacity: Option<usize>,
543    mut recv: tokio::sync::mpsc::Receiver<GLProcessorMessage>,
544    create_ctx_send: tokio::sync::oneshot::Sender<Result<super::platform::PlatformCaps, Error>>,
545) {
546    // Creation runs under the global lifecycle lock on Linux
547    // (bring-up races on some embedded drivers); ANGLE serializes
548    // display-level entry points internally, so macOS needs none
549    // (validated by the A0 lifecycle-churn spike leg).
550    #[cfg(target_os = "linux")]
551    let init_result = {
552        let _guard = super::context::GL_MUTEX
553            .lock()
554            .unwrap_or_else(|e| e.into_inner());
555        GLProcessorST::new(kind, capacity)
556    };
557    // macOS: ANGLE/Metal serializes display-level entry points
558    // internally. Android: the system EGL is specification-conformant
559    // thread-safe. Neither needs the Linux lifecycle lock.
560    #[cfg(any(target_os = "macos", target_os = "ios", target_os = "android"))]
561    let init_result = GLProcessorST::new(kind, capacity);
562    // Windows: ANGLE's D3D11 backend drives one immediate device context
563    // (and one state manager) per display from every GL context and is not
564    // safe to use from two threads at once: a processor bringing up its
565    // context (shader compiles, texture allocation) while another converted
566    // crashed with an access violation. Lifecycle and messages therefore
567    // share NON_LINUX_GL_MUTEX there; per-message locking is the ANGLE Full
568    // policy anyway.
569    #[cfg(target_os = "windows")]
570    let init_result = {
571        let _guard = NON_LINUX_GL_MUTEX.lock().unwrap_or_else(|e| e.into_inner());
572        GLProcessorST::new(kind, capacity)
573    };
574    let mut gl_converter = match init_result {
575        Ok(gl) => gl,
576        Err(e) => {
577            let _ = create_ctx_send.send(Err(e));
578            return;
579        }
580    };
581    gl_converter.owned_pbos = owned_pbos;
582    // Capability surface captured ONCE before the message loop —
583    // immutable for the worker's life, never re-probed per message.
584    let caps = gl_converter.platform_caps();
585    let _ = create_ctx_send.send(Ok(caps));
586    let mut poisoned = false;
587    // Tracks which PBO buffer_ids currently have an outstanding GL
588    // mapping, scoped to this one GL context/thread -- the ONE
589    // place every PboOps call against this context's buffers
590    // actually serializes, regardless of which `PboHandle`
591    // (edgefirst-tensor) issued the `PboMap`/`PboUnmap` message.
592    // `PboHandle::acquire_map`'s own per-handle `Mutex<MapState>`
593    // enforces "at most one mapping" WITHIN one handle, including
594    // read-sharing (several readers reuse ONE mapping without ever
595    // re-sending `PboMap` here) -- but it has no visibility into a
596    // SEPARATE `PboHandle` for the same real `buffer_id`, which
597    // task 11's cross-cdylib PBO import made newly reachable: a
598    // producer's own tensor and a reconstructed cross-package
599    // consumer tensor for the same `buffer_id` are two independent
600    // `PboHandle`s, each believing it is the only mapper.
601    // `glMapBufferRange` on an already-mapped buffer is undefined
602    // behaviour per the GL spec, so THIS set -- not the per-handle
603    // mutex, which never sees another handle's calls -- is what
604    // actually prevents that.
605    // Install this thread's GL-worker identity and its mapped-PBO set. Both
606    // the message path and the re-entrant inline path read them from here —
607    // see [`GL_WORKER`] for why the inline path has to exist at all.
608    GL_WORKER.with(|c| {
609        *c.borrow_mut() = Some(GlWorkerState {
610            id: worker_id,
611            // Set once the policy below is resolved.
612            serializes: false,
613            mapped: std::collections::HashSet::new(),
614        });
615    });
616    // Clear on every exit path, including unwind, so a recycled thread id can
617    // never inherit a dead worker's identity.
618    struct GlWorkerSlot;
619    impl Drop for GlWorkerSlot {
620        fn drop(&mut self) {
621            GL_WORKER.with(|c| *c.borrow_mut() = None);
622            GL_WORKER_POISON.with(|p| p.set(false));
623        }
624    }
625    let _gl_worker_slot = GlWorkerSlot;
626    // Per-message serialization policy: Full where the platform
627    // demands it (`caps.serialize_gl` — Vivante/galcore is not
628    // thread-safe for concurrent GL across contexts), LifecycleOnly
629    // everywhere else — multiple processors then run GL work in
630    // parallel, each on its own thread + context. See the GL_MUTEX
631    // doc comment in context.rs. EDGEFIRST_GL_SERIALIZE overrides
632    // per-driver defaults (full = escape hatch for unknown-bad
633    // drivers; lifecycle = force-parallel).
634    let serialize_per_msg = match std::env::var("EDGEFIRST_GL_SERIALIZE").as_deref() {
635        Ok("full") => true,
636        Ok("lifecycle") => {
637            // The per-message lock is what makes ANGLE's per-display state
638            // manager safe here: `LAST_ACTIVE_CONTEXT` and the per-message
639            // context re-sync it drives are documented as touched only under
640            // it, and concurrent GL on this backend has crashed this box.
641            #[cfg(target_os = "windows")]
642            log::warn!(
643                "EDGEFIRST_GL_SERIALIZE=lifecycle drops the per-message lock the ANGLE/D3D11 context re-sync relies on; concurrent GL across processors is not supported on this backend"
644            );
645            false
646        }
647        _ => caps.serialize_gl,
648    };
649    log::debug!(
650        "GL serialization policy: {}",
651        if serialize_per_msg {
652            "Full (per-message global lock)"
653        } else {
654            "LifecycleOnly (parallel across processors)"
655        }
656    );
657    GL_WORKER.with(|c| {
658        if let Some(state) = c.borrow_mut().as_mut() {
659            state.serializes = serialize_per_msg;
660        }
661    });
662    while let Some(msg) = recv.blocking_recv() {
663        // Full policy: one processor's message at a time, process-wide.
664        // Linux locks GL_MUTEX (messages must also exclude the locked
665        // lifecycle operations — on Vivante a convert racing a context
666        // bring-up is part of the driver bug class). macOS has no
667        // lifecycle lock (ANGLE serializes display entry points
668        // internally), so message-vs-message exclusion suffices —
669        // needed on virtualized GPUs, where concurrent GL across
670        // contexts mis-renders (paravirtual Metal, macOS CI runners).
671        #[cfg(target_os = "linux")]
672        let _guard = serialize_per_msg.then(|| {
673            super::context::GL_MUTEX
674                .lock()
675                .unwrap_or_else(|e| e.into_inner())
676        });
677        #[cfg(any(
678            target_os = "macos",
679            target_os = "ios",
680            target_os = "android",
681            target_os = "windows"
682        ))]
683        let _guard =
684            serialize_per_msg.then(|| NON_LINUX_GL_MUTEX.lock().unwrap_or_else(|e| e.into_inner()));
685
686        // After a panic, the GL context is in an undefined state. Reject
687        // all subsequent messages with an error rather than risking wrong
688        // output or a GPU hang from corrupted GL state. This follows the
689        // same pattern as std::sync::Mutex poisoning.
690        if poisoned {
691            reject_poisoned_message(msg);
692            continue;
693        }
694
695        // Platform pre-message hook (ANGLE/D3D11 re-syncs this context's
696        // state when another processor's context issued the last commands;
697        // a no-op elsewhere). Runs under the lock taken above.
698        gl_converter.begin_gpu_pass();
699        handle_gl_message(msg, &mut gl_converter, &mut poisoned);
700        // A re-entrant inline PBO op reports its panic here: `inline_gl_op`
701        // converted it to an error so it could not unwind into the tensor
702        // crate, which means `reply_caught` never saw it.
703        poisoned |= GL_WORKER_POISON.with(|p| p.replace(false));
704        // Per-pass platform texture attachments (macOS pbuffer
705        // binds) are released once the message's GPU work has
706        // synced; deferred batches keep theirs until Flush.
707        gl_converter.end_gpu_pass_if_synced();
708    }
709    // Explicitly drop under the mutex so EGL teardown is serialized
710    // (Linux, and Windows for the D3D11 reason above; ANGLE teardown is
711    // display-internal on macOS).
712    #[cfg(target_os = "linux")]
713    let _guard = super::context::GL_MUTEX
714        .lock()
715        .unwrap_or_else(|e| e.into_inner());
716    #[cfg(target_os = "windows")]
717    let _guard = NON_LINUX_GL_MUTEX.lock().unwrap_or_else(|e| e.into_inner());
718    drop(gl_converter);
719}
720
721/// Per-message serialization lock for the platforms without a Linux-style
722/// `GL_MUTEX` (macOS/iOS, Android, Windows) — taken per message under the
723/// Full policy, and on Windows also around processor bring-up/teardown.
724#[cfg(not(target_os = "linux"))]
725static NON_LINUX_GL_MUTEX: std::sync::Mutex<()> = std::sync::Mutex::new(());
726
727/// Take [`NON_LINUX_GL_MUTEX`] on behalf of a caller outside the GL worker
728/// threads. Poison-tolerant, like every other user of it.
729///
730/// Windows leaf tests MUST hold this guard for their whole body: any test
731/// that brings up an ANGLE display or a GL context does so on the test's own
732/// thread, and the lib test binary runs other tests whose GL workers are
733/// alive at the same time. ANGLE's D3D11 backend drives one immediate device
734/// context per display from every GL context and is not safe to use from two
735/// threads at once — the access-violation pattern documented on
736/// [`run_gl_worker`], which takes the same lock around bring-up and teardown.
737#[cfg(target_os = "windows")]
738pub(super) fn lifecycle_guard() -> std::sync::MutexGuard<'static, ()> {
739    NON_LINUX_GL_MUTEX.lock().unwrap_or_else(|e| e.into_inner())
740}
741
742fn handle_gl_message(
743    msg: GLProcessorMessage,
744    gl_converter: &mut GLProcessorST,
745    poisoned: &mut bool,
746) {
747    match msg {
748        GLProcessorMessage::ImageConvert(src, dst, options, defer, resp) => {
749            handle_image_convert(src, dst, (options, defer), resp, gl_converter, poisoned);
750        }
751        GLProcessorMessage::ImageConvertFenced(src, dst, options, resp) => {
752            handle_image_convert_fenced(src, dst, options, resp, gl_converter, poisoned);
753        }
754        GLProcessorMessage::Flush(resp) => {
755            let result = std::panic::catch_unwind(AssertUnwindSafe(|| gl_converter.flush()));
756            reply_caught(result, resp, poisoned, "Flush");
757        }
758        GLProcessorMessage::DrawDecodedMasks(
759            dst,
760            det,
761            seg,
762            opacity,
763            bg,
764            content_region,
765            color_mode,
766            resp,
767        ) => {
768            handle_draw_decoded_masks(
769                dst,
770                det,
771                seg,
772                (opacity, bg, content_region, color_mode),
773                resp,
774                gl_converter,
775                poisoned,
776            );
777        }
778        GLProcessorMessage::DrawProtoMasks(
779            dst,
780            det,
781            proto_data,
782            opacity,
783            bg,
784            content_region,
785            color_mode,
786            resp,
787        ) => {
788            handle_draw_proto_masks(
789                dst,
790                det,
791                proto_data,
792                (opacity, bg, content_region, color_mode),
793                resp,
794                gl_converter,
795                poisoned,
796            );
797        }
798        GLProcessorMessage::SetColors(colors, resp) => {
799            let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
800                gl_converter.set_class_colors(&colors)
801            }));
802            reply_caught(result, resp, poisoned, "SetColors");
803        }
804        GLProcessorMessage::SetBoxLineWidth(width, resp) => {
805            let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
806                gl_converter.set_box_line_width(width)
807            }));
808            reply_caught(result, resp, poisoned, "SetBoxLineWidth");
809        }
810        GLProcessorMessage::SetColorimetryMode(mode, resp) => {
811            let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
812                gl_converter.set_colorimetry_mode(mode);
813                Ok(())
814            }));
815            reply_caught(result, resp, poisoned, "SetColorimetryMode");
816        }
817        GLProcessorMessage::SetInterpolation(mode, resp) => {
818            let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
819                gl_converter.set_interpolation(mode);
820                Ok(())
821            }));
822            reply_caught(result, resp, poisoned, "SetInterpolation");
823        }
824        GLProcessorMessage::SetInt8Interpolation(mode, resp) => {
825            let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
826                gl_converter.set_int8_interpolation_mode(mode);
827                Ok(())
828            }));
829            reply_caught(result, resp, poisoned, "SetInt8Interpolation");
830        }
831        #[cfg(test)]
832        GLProcessorMessage::SetFailDstImport(fail, resp) => {
833            let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
834                gl_converter.fail_dst_import = fail;
835                Ok(())
836            }));
837            reply_caught(result, resp, poisoned, "SetFailDstImport");
838        }
839        GLProcessorMessage::EglCacheStats(resp) => {
840            let result =
841                std::panic::catch_unwind(AssertUnwindSafe(|| Ok(gl_converter.egl_cache_stats())));
842            reply_caught(result, resp, poisoned, "EglCacheStats");
843        }
844        GLProcessorMessage::ConvertStats(resp) => {
845            let result =
846                std::panic::catch_unwind(AssertUnwindSafe(|| Ok(gl_converter.convert_stats())));
847            reply_caught(result, resp, poisoned, "ConvertStats");
848        }
849        GLProcessorMessage::PboCreate(size, resp) => {
850            handle_pbo_create(size, resp, poisoned);
851        }
852        GLProcessorMessage::PboMap(buffer_id, size, resp) => {
853            handle_pbo_map(buffer_id, size, resp, poisoned);
854        }
855        GLProcessorMessage::PboUnmap(buffer_id, resp) => {
856            handle_pbo_unmap(buffer_id, resp, poisoned);
857        }
858        GLProcessorMessage::PboDelete(buffer_id) => {
859            handle_pbo_delete(buffer_id, poisoned);
860        }
861        GLProcessorMessage::CudaRegisterBuffer(buffer_id, resp) => {
862            // CUDA GL-interop must run on the GL-context thread.
863            let _ = resp.send(edgefirst_tensor::gl_register_buffer(buffer_id));
864        }
865        GLProcessorMessage::CudaMap(resource, resp) => {
866            // Auto-flush: if a batched `convert_deferred` is still
867            // owed its single GPU sync, complete it before CUDA maps
868            // the buffer — otherwise the device could read a render
869            // still in flight. This makes `cuda_map()` correct on a
870            // batched output with no API change (the caller pays the
871            // sync it would have owed anyway).
872            gl_converter.flush_pending();
873            let _ = resp.send(edgefirst_tensor::gl_map_resource(resource));
874        }
875        GLProcessorMessage::CudaUnmap(resource) => {
876            edgefirst_tensor::gl_unmap_resource(resource);
877        }
878        GLProcessorMessage::CudaUnregister(resource) => {
879            edgefirst_tensor::gl_unregister_resource(resource);
880        }
881    }
882}
883
884fn handle_image_convert(
885    src: SendablePtr<TensorDyn>,
886    mut dst: SendablePtr<TensorDyn>,
887    geom: (ConvertOptions, bool),
888    resp: tokio::sync::oneshot::Sender<Result<(), Error>>,
889    gl_converter: &mut GLProcessorST,
890    poisoned: &mut bool,
891) {
892    let (options, defer) = geom;
893    // SAFETY: This is safe because the convert() function waits for the resp to
894    // be sent before dropping the borrow for src and dst
895    let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
896        let src = unsafe { src.ptr.as_ref() };
897        let dst = unsafe { dst.ptr.as_mut() };
898        // A deferred convert skips the per-tile glFinish and
899        // leaves a single sync owed at Flush; an eager convert
900        // finishes as usual.
901        if defer {
902            gl_converter.convert_deferred(src, dst, Some(options))
903        } else {
904            gl_converter.convert(src, dst, Some(options))
905        }
906    }));
907    reply_caught(result, resp, poisoned, "ImageConvert");
908}
909
910fn handle_image_convert_fenced(
911    src: SendablePtr<TensorDyn>,
912    mut dst: SendablePtr<TensorDyn>,
913    options: ConvertOptions,
914    resp: tokio::sync::oneshot::Sender<Result<Option<super::CompletionFence>, Error>>,
915    gl_converter: &mut GLProcessorST,
916    poisoned: &mut bool,
917) {
918    // SAFETY: as ImageConvert — convert_with_fence()
919    // blocks on the reply before the borrows drop.
920    let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
921        let src = unsafe { src.ptr.as_ref() };
922        let dst = unsafe { dst.ptr.as_mut() };
923        gl_converter.convert_fenced(src, dst, options)
924    }));
925    reply_caught(result, resp, poisoned, "ImageConvertFenced");
926}
927
928fn handle_draw_decoded_masks(
929    mut dst: SendablePtr<TensorDyn>,
930    det: SendablePtr<DetectBox>,
931    seg: SendablePtr<Segmentation>,
932    overlay: (
933        f32,
934        Option<SendablePtr<TensorDyn>>,
935        Option<[f32; 4]>,
936        crate::ColorMode,
937    ),
938    resp: tokio::sync::oneshot::Sender<Result<(), Error>>,
939    gl_converter: &mut GLProcessorST,
940    poisoned: &mut bool,
941) {
942    let (opacity, bg, content_region, color_mode) = overlay;
943    // SAFETY: This is safe because the draw_decoded_masks() function waits for the
944    // resp to be sent before dropping the borrow for dst, detect,
945    // segmentation, and background
946    let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
947        let dst = unsafe { dst.ptr.as_mut() };
948        let det = unsafe { std::slice::from_raw_parts(det.ptr.as_ptr(), det.len) };
949        let seg = unsafe { std::slice::from_raw_parts(seg.ptr.as_ptr(), seg.len) };
950        let bg_ref = bg.map(|p| unsafe { &*p.ptr.as_ptr() });
951        gl_converter.draw_decoded_masks(
952            dst,
953            det,
954            seg,
955            crate::MaskOverlay {
956                background: bg_ref,
957                opacity,
958                content_region,
959                color_mode,
960            },
961        )
962    }));
963    reply_caught(result, resp, poisoned, "DrawDecodedMasks");
964}
965
966fn handle_draw_proto_masks(
967    mut dst: SendablePtr<TensorDyn>,
968    det: SendablePtr<DetectBox>,
969    proto_data: SendablePtr<ProtoData>,
970    overlay: (
971        f32,
972        Option<SendablePtr<TensorDyn>>,
973        Option<[f32; 4]>,
974        crate::ColorMode,
975    ),
976    resp: tokio::sync::oneshot::Sender<Result<(), Error>>,
977    gl_converter: &mut GLProcessorST,
978    poisoned: &mut bool,
979) {
980    let (opacity, bg, content_region, color_mode) = overlay;
981    // SAFETY: Same safety invariant as DrawDecodedMasks — caller
982    // blocks on resp before dropping borrows.
983    let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
984        let dst = unsafe { dst.ptr.as_mut() };
985        let det = unsafe { std::slice::from_raw_parts(det.ptr.as_ptr(), det.len) };
986        let bg_ref = bg.map(|p| unsafe { &*p.ptr.as_ptr() });
987        let proto_data = unsafe { proto_data.ptr.as_ref() };
988        gl_converter.draw_proto_masks(
989            dst,
990            det,
991            proto_data,
992            crate::MaskOverlay {
993                background: bg_ref,
994                opacity,
995                content_region,
996                color_mode,
997            },
998        )
999    }));
1000    reply_caught(result, resp, poisoned, "DrawProtoMasks");
1001}
1002
1003fn handle_pbo_create(
1004    size: usize,
1005    resp: tokio::sync::oneshot::Sender<Result<u32, Error>>,
1006    poisoned: &mut bool,
1007) {
1008    let result = std::panic::catch_unwind(AssertUnwindSafe(|| unsafe {
1009        let mut id: u32 = 0;
1010        edgefirst_gl::gl::GenBuffers(1, &mut id);
1011        edgefirst_gl::gl::BindBuffer(edgefirst_gl::gl::PIXEL_PACK_BUFFER, id);
1012        edgefirst_gl::gl::BufferData(
1013            edgefirst_gl::gl::PIXEL_PACK_BUFFER,
1014            size as isize,
1015            std::ptr::null(),
1016            edgefirst_gl::gl::STREAM_COPY,
1017        );
1018        edgefirst_gl::gl::BindBuffer(edgefirst_gl::gl::PIXEL_PACK_BUFFER, 0);
1019        match check_gl_error("PboCreate", 0) {
1020            Ok(()) => Ok(id),
1021            Err(e) => {
1022                edgefirst_gl::gl::DeleteBuffers(1, &id);
1023                Err(e)
1024            }
1025        }
1026    }));
1027    reply_caught(result, resp, poisoned, "PboCreate");
1028}
1029
1030/// Perform a PBO map on the calling thread, which MUST be the GL worker that
1031/// owns the buffer (context already current). Shared by the message handler and
1032/// the re-entrant inline path so the dedup rule is enforced identically.
1033fn pbo_map_on_gl_thread(
1034    buffer_id: u32,
1035    size: usize,
1036) -> Result<edgefirst_tensor::PboMapping, Error> {
1037    let already = with_worker_mapped(|set| !set.insert(buffer_id)).unwrap_or(false);
1038    if already {
1039        return Err(Error::OpenGl(format!(
1040            "PBO buffer {buffer_id} is already mapped in this GL context \
1041             (a producer and a reconstructed cross-package consumer may \
1042             both hold a live tensor over it)"
1043        )));
1044    }
1045    let result = std::panic::catch_unwind(AssertUnwindSafe(|| unsafe {
1046        edgefirst_gl::gl::BindBuffer(edgefirst_gl::gl::PIXEL_PACK_BUFFER, buffer_id);
1047        let ptr = edgefirst_gl::gl::MapBufferRange(
1048            edgefirst_gl::gl::PIXEL_PACK_BUFFER,
1049            0,
1050            size as isize,
1051            edgefirst_gl::gl::MAP_READ_BIT | edgefirst_gl::gl::MAP_WRITE_BIT,
1052        );
1053        edgefirst_gl::gl::BindBuffer(edgefirst_gl::gl::PIXEL_PACK_BUFFER, 0);
1054        if ptr.is_null() {
1055            Err(Error::OpenGl("glMapBufferRange returned null".to_string()))
1056        } else {
1057            Ok(edgefirst_tensor::PboMapping {
1058                ptr: ptr as *mut u8,
1059                size,
1060            })
1061        }
1062    }));
1063    if !matches!(result, Ok(Ok(_))) {
1064        // The map did not take — release the reservation so a retry is not
1065        // permanently blocked by a mapping that never happened.
1066        with_worker_mapped(|set| set.remove(&buffer_id));
1067    }
1068    match result {
1069        Ok(result) => result,
1070        // Reaches whichever boundary the caller has: `handle_pbo_map` on the
1071        // message path, `inline_gl_op` on the re-entrant one.
1072        Err(panic) => std::panic::resume_unwind(panic),
1073    }
1074}
1075
1076/// Unmap counterpart of [`pbo_map_on_gl_thread`]; same thread requirement.
1077fn pbo_unmap_on_gl_thread(buffer_id: u32) -> Result<(), Error> {
1078    // Always release the reservation, matching `PboHandle::release_map`'s
1079    // "state returns to Unmapped either way" contract.
1080    with_worker_mapped(|set| set.remove(&buffer_id));
1081    unsafe {
1082        edgefirst_gl::gl::BindBuffer(edgefirst_gl::gl::PIXEL_PACK_BUFFER, buffer_id);
1083        let ok = edgefirst_gl::gl::UnmapBuffer(edgefirst_gl::gl::PIXEL_PACK_BUFFER);
1084        edgefirst_gl::gl::BindBuffer(edgefirst_gl::gl::PIXEL_PACK_BUFFER, 0);
1085        if ok == edgefirst_gl::gl::FALSE {
1086            Err(Error::OpenGl(
1087                "PBO data was corrupted during mapping".into(),
1088            ))
1089        } else {
1090            check_gl_error("PboUnmap", 0)
1091        }
1092    }
1093}
1094
1095fn handle_pbo_map(
1096    buffer_id: u32,
1097    size: usize,
1098    resp: tokio::sync::oneshot::Sender<Result<edgefirst_tensor::PboMapping, Error>>,
1099    poisoned: &mut bool,
1100) {
1101    let result =
1102        std::panic::catch_unwind(AssertUnwindSafe(|| pbo_map_on_gl_thread(buffer_id, size)));
1103    reply_caught(result, resp, poisoned, "PboMap");
1104}
1105
1106fn handle_pbo_unmap(
1107    buffer_id: u32,
1108    resp: tokio::sync::oneshot::Sender<Result<(), Error>>,
1109    poisoned: &mut bool,
1110) {
1111    let result = std::panic::catch_unwind(AssertUnwindSafe(|| pbo_unmap_on_gl_thread(buffer_id)));
1112    reply_caught(result, resp, poisoned, "PboUnmap");
1113}
1114
1115fn handle_pbo_delete(buffer_id: u32, poisoned: &mut bool) {
1116    // Defensive: a live mapping should never reach here
1117    // (PboHandle::Drop unmaps before deleting), but a
1118    // stale reservation for a since-deleted (and
1119    // possibly buffer-id-recycled) buffer would
1120    // otherwise wrongly block every future map of a NEW
1121    // buffer that reuses the same id.
1122    with_worker_mapped(|set| set.remove(&buffer_id));
1123    if let Err(e) = std::panic::catch_unwind(AssertUnwindSafe(|| unsafe {
1124        edgefirst_gl::gl::DeleteBuffers(1, &buffer_id);
1125    })) {
1126        *poisoned = true;
1127        log::error!(
1128            "GL thread panicked during PboDelete: {}",
1129            panic_message(e.as_ref()),
1130        );
1131    }
1132}
1133
1134impl GLProcessorThreaded {
1135    /// Creates a new OpenGL multi-threaded image converter whose import
1136    /// caches use the environment/default capacity.
1137    pub fn new(kind: Option<EglDisplayKind>) -> Result<Self, Error> {
1138        Self::with_cache_capacity(kind, None)
1139    }
1140
1141    /// As [`new`](Self::new), with an explicit per-cache import capacity —
1142    /// [`ImageProcessorConfig::egl_cache_capacity`], which documents what the
1143    /// number bounds and takes precedence over the environment.
1144    ///
1145    /// [`ImageProcessorConfig::egl_cache_capacity`]: crate::ImageProcessorConfig::egl_cache_capacity
1146    pub fn with_cache_capacity(
1147        kind: Option<EglDisplayKind>,
1148        capacity: Option<usize>,
1149    ) -> Result<Self, Error> {
1150        let worker_id = NEXT_GL_WORKER_ID.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
1151        let (send, recv) = tokio::sync::mpsc::channel::<GLProcessorMessage>(1);
1152
1153        let (create_ctx_send, create_ctx_recv) = tokio::sync::oneshot::channel();
1154
1155        let owned_pbos = std::sync::Arc::<super::processor::OwnedPbos>::default();
1156        let worker_owned = owned_pbos.clone();
1157        let handle = std::thread::spawn(move || {
1158            run_gl_worker(
1159                worker_id,
1160                worker_owned,
1161                kind,
1162                capacity,
1163                recv,
1164                create_ctx_send,
1165            );
1166        });
1167
1168        let caps = match create_ctx_recv.blocking_recv() {
1169            Ok(Err(e)) => return Err(e),
1170            Err(_) => {
1171                return Err(Error::Internal(
1172                    "GL converter error messaging closed without update".to_string(),
1173                ));
1174            }
1175            Ok(Ok(caps)) => caps,
1176        };
1177
1178        Ok(Self {
1179            handle: Some(handle),
1180            sender: Some(send),
1181            worker_id,
1182            owned_pbos,
1183            caps,
1184        })
1185    }
1186}
1187
1188impl ImageProcessorTrait for GLProcessorThreaded {
1189    fn convert(
1190        &mut self,
1191        src: &TensorDyn,
1192        dst: &mut TensorDyn,
1193        options: Option<ConvertOptions>,
1194    ) -> crate::Result<()> {
1195        let options = options.unwrap_or_default();
1196        let (err_send, err_recv) = tokio::sync::oneshot::channel();
1197        self.sender
1198            .as_ref()
1199            .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1200            .blocking_send(GLProcessorMessage::ImageConvert(
1201                SendablePtr {
1202                    ptr: NonNull::from(src),
1203                    len: 1,
1204                },
1205                SendablePtr {
1206                    ptr: NonNull::from(dst),
1207                    len: 1,
1208                },
1209                options,
1210                false, // eager: finish per convert
1211                err_send,
1212            ))
1213            .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1214        err_recv.blocking_recv().map_err(|_| {
1215            Error::Internal("GL converter error messaging closed without update".to_string())
1216        })?
1217    }
1218
1219    fn convert_deferred(
1220        &mut self,
1221        src: &TensorDyn,
1222        dst: &mut TensorDyn,
1223        options: Option<ConvertOptions>,
1224    ) -> crate::Result<()> {
1225        let options = options.unwrap_or_default();
1226        let (err_send, err_recv) = tokio::sync::oneshot::channel();
1227        self.sender
1228            .as_ref()
1229            .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1230            .blocking_send(GLProcessorMessage::ImageConvert(
1231                SendablePtr {
1232                    ptr: NonNull::from(src),
1233                    len: 1,
1234                },
1235                SendablePtr {
1236                    ptr: NonNull::from(dst),
1237                    len: 1,
1238                },
1239                options,
1240                true, // defer: skip glFinish; flush() syncs the batch once
1241                err_send,
1242            ))
1243            .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1244        err_recv.blocking_recv().map_err(|_| {
1245            Error::Internal("GL converter error messaging closed without update".to_string())
1246        })?
1247    }
1248
1249    fn flush(&mut self) -> crate::Result<()> {
1250        let (err_send, err_recv) = tokio::sync::oneshot::channel();
1251        self.sender
1252            .as_ref()
1253            .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1254            .blocking_send(GLProcessorMessage::Flush(err_send))
1255            .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1256        err_recv.blocking_recv().map_err(|_| {
1257            Error::Internal("GL converter error messaging closed without update".to_string())
1258        })?
1259    }
1260
1261    fn draw_decoded_masks(
1262        &mut self,
1263        dst: &mut TensorDyn,
1264        detect: &[crate::DetectBox],
1265        segmentation: &[crate::Segmentation],
1266        overlay: crate::MaskOverlay<'_>,
1267    ) -> crate::Result<()> {
1268        let (err_send, err_recv) = tokio::sync::oneshot::channel();
1269        self.sender
1270            .as_ref()
1271            .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1272            .blocking_send(GLProcessorMessage::DrawDecodedMasks(
1273                SendablePtr {
1274                    ptr: NonNull::from(dst),
1275                    len: 1,
1276                },
1277                SendablePtr {
1278                    ptr: NonNull::new(detect.as_ptr() as *mut DetectBox).unwrap(),
1279                    len: detect.len(),
1280                },
1281                SendablePtr {
1282                    ptr: NonNull::new(segmentation.as_ptr() as *mut Segmentation).unwrap(),
1283                    len: segmentation.len(),
1284                },
1285                overlay.opacity,
1286                overlay.background.map(|bg| SendablePtr {
1287                    ptr: NonNull::from(bg).cast::<TensorDyn>(),
1288                    len: 1,
1289                }),
1290                overlay.content_region,
1291                overlay.color_mode,
1292                err_send,
1293            ))
1294            .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1295        err_recv.blocking_recv().map_err(|_| {
1296            Error::Internal("GL converter error messaging closed without update".to_string())
1297        })?
1298    }
1299
1300    fn draw_proto_masks(
1301        &mut self,
1302        dst: &mut TensorDyn,
1303        detect: &[DetectBox],
1304        proto_data: &ProtoData,
1305        overlay: crate::MaskOverlay<'_>,
1306    ) -> crate::Result<()> {
1307        let (err_send, err_recv) = tokio::sync::oneshot::channel();
1308        self.sender
1309            .as_ref()
1310            .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1311            .blocking_send(GLProcessorMessage::DrawProtoMasks(
1312                SendablePtr {
1313                    ptr: NonNull::from(dst),
1314                    len: 1,
1315                },
1316                SendablePtr {
1317                    ptr: NonNull::new(detect.as_ptr() as *mut DetectBox).unwrap(),
1318                    len: detect.len(),
1319                },
1320                SendablePtr {
1321                    ptr: NonNull::from(proto_data).cast::<ProtoData>(),
1322                    len: 1,
1323                },
1324                overlay.opacity,
1325                overlay.background.map(|bg| SendablePtr {
1326                    ptr: NonNull::from(bg).cast::<TensorDyn>(),
1327                    len: 1,
1328                }),
1329                overlay.content_region,
1330                overlay.color_mode,
1331                err_send,
1332            ))
1333            .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1334        err_recv.blocking_recv().map_err(|_| {
1335            Error::Internal("GL converter error messaging closed without update".to_string())
1336        })?
1337    }
1338
1339    fn set_class_colors(&mut self, colors: &[[u8; 4]]) -> Result<(), crate::Error> {
1340        let (err_send, err_recv) = tokio::sync::oneshot::channel();
1341        self.sender
1342            .as_ref()
1343            .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1344            .blocking_send(GLProcessorMessage::SetColors(colors.to_vec(), err_send))
1345            .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1346        err_recv.blocking_recv().map_err(|_| {
1347            Error::Internal("GL converter error messaging closed without update".to_string())
1348        })?
1349    }
1350
1351    fn set_box_line_width(&mut self, width: u32) -> Result<(), crate::Error> {
1352        crate::check_box_line_width(width)?;
1353        let (err_send, err_recv) = tokio::sync::oneshot::channel();
1354        self.sender
1355            .as_ref()
1356            .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1357            .blocking_send(GLProcessorMessage::SetBoxLineWidth(width, err_send))
1358            .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1359        err_recv.blocking_recv().map_err(|_| {
1360            Error::Internal("GL converter error messaging closed without update".to_string())
1361        })?
1362    }
1363}
1364
1365impl GLProcessorThreaded {
1366    /// Sets the colorimetry/performance trade-off (see
1367    /// [`crate::ColorimetryMode`]). The `EDGEFIRST_COLORIMETRY` environment
1368    /// variable takes precedence — when set, this call logs and keeps the
1369    /// env-selected mode.
1370    pub fn set_colorimetry_mode(&mut self, mode: crate::ColorimetryMode) -> Result<(), Error> {
1371        let (err_send, err_recv) = tokio::sync::oneshot::channel();
1372        self.sender
1373            .as_ref()
1374            .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1375            .blocking_send(GLProcessorMessage::SetColorimetryMode(mode, err_send))
1376            .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1377        err_recv.blocking_recv().map_err(|_| {
1378            Error::Internal("GL converter error messaging closed without update".to_string())
1379        })?
1380    }
1381
1382    /// Sets how resizing converts resample (see [`crate::Interpolation`]).
1383    pub fn set_interpolation(&mut self, mode: crate::Interpolation) -> Result<(), Error> {
1384        let (err_send, err_recv) = tokio::sync::oneshot::channel();
1385        self.sender
1386            .as_ref()
1387            .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1388            .blocking_send(GLProcessorMessage::SetInterpolation(mode, err_send))
1389            .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1390        err_recv.blocking_recv().map_err(|_| {
1391            Error::Internal("GL converter error messaging closed without update".to_string())
1392        })?
1393    }
1394
1395    /// Test hook (issue #175): make every zero-copy DESTINATION import fail,
1396    /// the way a driver that refuses the buffer does, so the fallback to the
1397    /// mapped-texture readback is exercised on a host whose driver accepts
1398    /// every destination. The failure is injected at the import itself, so
1399    /// what a test drives is the production trigger.
1400    #[cfg(test)]
1401    pub(crate) fn set_fail_dst_import(&mut self, fail: bool) -> Result<(), Error> {
1402        let (err_send, err_recv) = tokio::sync::oneshot::channel();
1403        self.sender
1404            .as_ref()
1405            .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1406            .blocking_send(GLProcessorMessage::SetFailDstImport(fail, err_send))
1407            .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1408        err_recv.blocking_recv().map_err(|_| {
1409            Error::Internal("GL converter error messaging closed without update".to_string())
1410        })?
1411    }
1412
1413    /// Sets the interpolation mode for int8 proto textures.
1414    pub fn set_int8_interpolation_mode(
1415        &mut self,
1416        mode: Int8InterpolationMode,
1417    ) -> Result<(), crate::Error> {
1418        let (err_send, err_recv) = tokio::sync::oneshot::channel();
1419        self.sender
1420            .as_ref()
1421            .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1422            .blocking_send(GLProcessorMessage::SetInt8Interpolation(mode, err_send))
1423            .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1424        err_recv.blocking_recv().map_err(|_| {
1425            Error::Internal("GL converter error messaging closed without update".to_string())
1426        })?
1427    }
1428
1429    /// Snapshot the EGLImage cache counters (src, dst, NV R8) from the GL
1430    /// thread. Steady-state tests capture this after warmup and after an
1431    /// N-frame loop over a fixed buffer pool and assert
1432    /// [`total_misses`](super::cache::GlCacheStats::total_misses) stays flat —
1433    /// any increase means a convert re-imported a buffer it should have found
1434    /// cached (the cache-behavior equality gate for GL refactors).
1435    pub fn egl_cache_stats(&self) -> Result<super::cache::GlCacheStats, Error> {
1436        let (send, recv) = tokio::sync::oneshot::channel();
1437        self.sender
1438            .as_ref()
1439            .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1440            .blocking_send(GLProcessorMessage::EglCacheStats(send))
1441            .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1442        recv.blocking_recv().map_err(|_| {
1443            Error::Internal("GL converter error messaging closed without update".to_string())
1444        })?
1445    }
1446
1447    /// Convert and return a native sync-fence fd that signals when the
1448    /// GPU work completes, instead of blocking the CPU on `glFinish` —
1449    /// the GL→NPU handoff (`EGL_ANDROID_native_fence_sync`).
1450    ///
1451    /// `Ok(None)` means the convert completed with the normal blocking
1452    /// semantics (platform/driver has no native fence, or a pure-CPU-
1453    /// readback path was taken) — the destination is already safe to
1454    /// read. `Ok(Some(fd))` means the caller (or the NPU runtime, via
1455    /// `ANeuralNetworksExecution_startComputeWithDependencies`) must wait
1456    /// on the fd before consuming the destination buffer.
1457    pub fn convert_with_fence(
1458        &mut self,
1459        src: &TensorDyn,
1460        dst: &mut TensorDyn,
1461        options: Option<ConvertOptions>,
1462    ) -> Result<Option<super::CompletionFence>, Error> {
1463        let options = options.unwrap_or_default();
1464        use crate::ImageProcessorTrait as _;
1465        if !self.caps.native_fence_sync {
1466            // No fence on this display: the plain blocking convert IS the
1467            // fenced contract with `None` — skip the special round-trip.
1468            self.convert(src, dst, Some(options))?;
1469            return Ok(None);
1470        }
1471        let (send, recv) = tokio::sync::oneshot::channel();
1472        self.sender
1473            .as_ref()
1474            .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1475            .blocking_send(GLProcessorMessage::ImageConvertFenced(
1476                SendablePtr {
1477                    ptr: NonNull::from(src),
1478                    len: 1,
1479                },
1480                SendablePtr {
1481                    ptr: NonNull::from(dst),
1482                    len: 1,
1483                },
1484                options,
1485                send,
1486            ))
1487            .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1488        recv.blocking_recv().map_err(|_| {
1489            Error::Internal("GL converter error messaging closed without update".to_string())
1490        })?
1491    }
1492
1493    /// Snapshot the per-convert source-feed counters from the GL thread.
1494    /// A pipeline that expects end-to-end zero-copy asserts
1495    /// [`src_uploads`](super::cache::ConvertStats::src_uploads) stays flat
1496    /// (every frame fed by import, never by a CPU map + upload).
1497    pub fn convert_stats(&self) -> Result<super::cache::ConvertStats, Error> {
1498        let (send, recv) = tokio::sync::oneshot::channel();
1499        self.sender
1500            .as_ref()
1501            .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1502            .blocking_send(GLProcessorMessage::ConvertStats(send))
1503            .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1504        recv.blocking_recv().map_err(|_| {
1505            Error::Internal("GL converter error messaging closed without update".to_string())
1506        })?
1507    }
1508
1509    /// Create a PBO-backed [`Tensor<u8>`](edgefirst_tensor::Tensor) image on the GL thread.
1510    ///
1511    /// Declared [`CpuAccess::ReadWrite`](edgefirst_tensor::CpuAccess): the GL
1512    /// readback maps the buffer on the CPU itself (repack, BGRA swap, row
1513    /// spread), and a narrower declaration buys no cheaper PBO mapping.
1514    pub fn create_pbo_image(
1515        &self,
1516        width: usize,
1517        height: usize,
1518        format: edgefirst_tensor::PixelFormat,
1519    ) -> Result<edgefirst_tensor::Tensor<u8>, Error> {
1520        let sender = self
1521            .sender
1522            .as_ref()
1523            .ok_or(Error::OpenGl("GL processor is shutting down".to_string()))?;
1524
1525        let size = pbo_elem_count(width, height, format)
1526            .filter(|&s| s != 0)
1527            .ok_or_else(|| Error::OpenGl("Invalid image dimensions".to_string()))?;
1528
1529        // Allocate PBO on the GL thread
1530        let (tx, rx) = tokio::sync::oneshot::channel();
1531        sender
1532            .blocking_send(GLProcessorMessage::PboCreate(size, tx))
1533            .map_err(|_| Error::OpenGl("GL thread channel closed".to_string()))?;
1534        let buffer_id = rx
1535            .blocking_recv()
1536            .map_err(|_| Error::OpenGl("GL thread did not respond".to_string()))??;
1537
1538        let ops: std::sync::Arc<dyn edgefirst_tensor::PboOps> = std::sync::Arc::new(GlPboOps {
1539            sender: sender.downgrade(),
1540            worker_id: self.worker_id,
1541        });
1542        let owned_ops = ops.clone();
1543
1544        let shape = pbo_shape(width, height, format);
1545
1546        let pbo_tensor =
1547            edgefirst_tensor::PboTensor::<u8>::from_pbo(buffer_id, size, &shape, None, ops)
1548                .map_err(|e| Error::OpenGl(format!("PBO tensor creation failed: {e:?}")))?;
1549        let mut tensor = edgefirst_tensor::Tensor::from_pbo(pbo_tensor)
1550            .map_err(|e| Error::OpenGl(format!("PBO tensor wrap failed: {e:?}")))?;
1551        tensor
1552            .set_format(format)
1553            .map_err(|e| Error::OpenGl(format!("Failed to set format on PBO tensor: {e:?}")))?;
1554        // Register the PBO with CUDA (best-effort; no-op without libcudart) so
1555        // `cuda_map()` can yield a device pointer — matching the float PBO path
1556        // in `create_pbo_image_dtype`. Without this, u8 PBOs were never
1557        // CUDA-interop-capable, so the codec's nvJPEG backend (and any CUDA
1558        // consumer) could not use a `create_image`-allocated PBO destination.
1559        // The i8 transmute by the caller preserves the attached handle.
1560        register_pbo_cuda(&mut tensor, buffer_id, size, sender);
1561        self.owned_pbos.insert(tensor.pbo_vtable_ptr(), &owned_ops);
1562        Ok(tensor)
1563    }
1564
1565    /// Create a PBO-backed [`TensorDyn`] image on the GL thread with the given dtype.
1566    ///
1567    /// Sizes the underlying GL buffer by `elems * dtype.size()` and wraps it in
1568    /// the appropriately-typed [`PboTensor`]. Supports `DType::U8`, `DType::F16`,
1569    /// and `DType::F32`; returns an error for other dtypes.
1570    pub(crate) fn create_pbo_image_dtype(
1571        &self,
1572        width: usize,
1573        height: usize,
1574        format: edgefirst_tensor::PixelFormat,
1575        dtype: edgefirst_tensor::DType,
1576    ) -> Result<TensorDyn, Error> {
1577        let sender = self
1578            .sender
1579            .as_ref()
1580            .ok_or(Error::OpenGl("GL processor is shutting down".to_string()))?;
1581
1582        let elems = pbo_elem_count(width, height, format)
1583            .filter(|&e| e != 0)
1584            .ok_or_else(|| Error::OpenGl("Invalid image dimensions".to_string()))?;
1585
1586        let size = elems
1587            .checked_mul(dtype.size())
1588            .ok_or_else(|| Error::OpenGl("PBO size overflow".to_string()))?;
1589
1590        // Allocate PBO on the GL thread
1591        let (tx, rx) = tokio::sync::oneshot::channel();
1592        sender
1593            .blocking_send(GLProcessorMessage::PboCreate(size, tx))
1594            .map_err(|_| Error::OpenGl("GL thread channel closed".to_string()))?;
1595        let buffer_id = rx
1596            .blocking_recv()
1597            .map_err(|_| Error::OpenGl("GL thread did not respond".to_string()))??;
1598
1599        let ops: std::sync::Arc<dyn edgefirst_tensor::PboOps> = std::sync::Arc::new(GlPboOps {
1600            sender: sender.downgrade(),
1601            worker_id: self.worker_id,
1602        });
1603        let owned_ops = ops.clone();
1604
1605        let shape = pbo_shape(width, height, format);
1606
1607        let map_err = |e: edgefirst_tensor::Error| {
1608            Error::OpenGl(format!("PBO tensor creation failed: {e:?}"))
1609        };
1610        let set_err = |e: edgefirst_tensor::Error| {
1611            Error::OpenGl(format!("Failed to set format on PBO tensor: {e:?}"))
1612        };
1613
1614        let tensor = match dtype {
1615            edgefirst_tensor::DType::U8 => {
1616                let pbo =
1617                    edgefirst_tensor::PboTensor::<u8>::from_pbo(buffer_id, size, &shape, None, ops)
1618                        .map_err(map_err)?;
1619                let mut t = edgefirst_tensor::Tensor::from_pbo(pbo).map_err(map_err)?;
1620                t.set_format(format).map_err(set_err)?;
1621                register_pbo_cuda(&mut t, buffer_id, size, sender);
1622                TensorDyn::from(t)
1623            }
1624            edgefirst_tensor::DType::F16 => {
1625                let pbo = edgefirst_tensor::PboTensor::<edgefirst_tensor::f16>::from_pbo(
1626                    buffer_id, size, &shape, None, ops,
1627                )
1628                .map_err(map_err)?;
1629                let mut t = edgefirst_tensor::Tensor::from_pbo(pbo).map_err(map_err)?;
1630                t.set_format(format).map_err(set_err)?;
1631                register_pbo_cuda(&mut t, buffer_id, size, sender);
1632                TensorDyn::from(t)
1633            }
1634            edgefirst_tensor::DType::F32 => {
1635                let pbo = edgefirst_tensor::PboTensor::<f32>::from_pbo(
1636                    buffer_id, size, &shape, None, ops,
1637                )
1638                .map_err(map_err)?;
1639                let mut t = edgefirst_tensor::Tensor::from_pbo(pbo).map_err(map_err)?;
1640                t.set_format(format).map_err(set_err)?;
1641                register_pbo_cuda(&mut t, buffer_id, size, sender);
1642                TensorDyn::from(t)
1643            }
1644            other => return Err(Error::OpenGl(format!("unsupported PBO dtype {other:?}"))),
1645        };
1646        self.owned_pbos.insert(tensor.pbo_vtable_ptr(), &owned_ops);
1647        Ok(tensor)
1648    }
1649
1650    /// Returns the active transfer backend.
1651    pub(crate) fn transfer_backend(&self) -> TransferBackend {
1652        self.caps.transfer_backend
1653    }
1654
1655    /// Report which float dtypes the GPU can render to.
1656    ///
1657    /// Values are probed once at construction time and adjusted for
1658    /// Vivante GC7000UL, whose float readback latency (170-320 ms) makes
1659    /// GL float destinations impractical; `ImageProcessor::convert()` falls
1660    /// back to CPU float output (normalized to `[0, 1]`) for these targets.
1661    pub(crate) fn supported_render_dtypes(&self) -> crate::RenderDtypeSupport {
1662        self.caps.render_dtypes
1663    }
1664
1665    /// The CPU access `convert` needs on a `format`/`dtype` destination held
1666    /// in `memory` when it renders it from an RGBA or NV source (`Write` for
1667    /// a host-memory readback, else `None`), or `None` when it declines it to
1668    /// the CPU fallback.
1669    pub(crate) fn dst_cpu_access(
1670        &self,
1671        memory: edgefirst_tensor::TensorMemory,
1672        format: edgefirst_tensor::PixelFormat,
1673        dtype: edgefirst_tensor::DType,
1674    ) -> Option<edgefirst_tensor::CpuAccess> {
1675        use edgefirst_tensor::{CpuAccess, DType, PixelFormat, TensorMemory};
1676        let renders = match dtype {
1677            DType::U8 | DType::I8 => GLProcessorST::dst_format_renderable(
1678                self.caps.transfer_backend,
1679                memory,
1680                format,
1681                self.caps.has_bgra,
1682            ),
1683            DType::F16 | DType::F32 => {
1684                super::float_dispatch::classify_float_render(
1685                    PixelFormat::Rgba,
1686                    format,
1687                    dtype,
1688                    memory,
1689                    self.caps.render_dtypes,
1690                    <super::platform::Platform as super::platform::GlPlatform>::ZERO_COPY_FLOAT,
1691                ) != super::float_dispatch::FloatRenderPath::None
1692            }
1693            _ => false,
1694        };
1695        let readback = matches!(memory, TensorMemory::Mem | TensorMemory::Shm);
1696        renders.then_some(if readback {
1697            CpuAccess::Write
1698        } else {
1699            CpuAccess::None
1700        })
1701    }
1702}
1703
1704impl Drop for GLProcessorThreaded {
1705    fn drop(&mut self) {
1706        // Closing the channel ends the worker loop, which drops the
1707        // `GLProcessorST` it owns; that `Drop` pays any device submission an
1708        // open batch still owes. The join is what makes it ordered before this
1709        // returns -- and it has to happen on the worker, which is the thread
1710        // holding the context.
1711        drop(self.sender.take());
1712        let _ = self.handle.take().and_then(|h| h.join().ok());
1713    }
1714}
1715
1716// `pbo_elem_count` / `pbo_shape` are pure (no GL), so they are unit-testable
1717// without a GPU. The overflow→None arm of `pbo_elem_count` guards against an
1718// undersized PBO allocation, so it is worth pinning explicitly.
1719#[cfg(test)]
1720#[cfg_attr(coverage_nightly, coverage(off))]
1721mod tests {
1722    use super::{pbo_elem_count, pbo_shape, GLProcessorThreaded, GlPboOps};
1723    use edgefirst_tensor::{PixelFormat, TensorTrait};
1724
1725    #[test]
1726    fn elem_count_per_format() {
1727        // Packed RGBA: w*h*4.
1728        assert_eq!(pbo_elem_count(8, 4, PixelFormat::Rgba), Some(8 * 4 * 4));
1729        // Packed RGB: w*h*3.
1730        assert_eq!(pbo_elem_count(8, 4, PixelFormat::Rgb), Some(8 * 4 * 3));
1731        // NV12 semiplanar: w*h*3/2.
1732        assert_eq!(pbo_elem_count(8, 4, PixelFormat::Nv12), Some(8 * 4 * 3 / 2));
1733        // NV16 semiplanar: w*h*2.
1734        assert_eq!(pbo_elem_count(8, 4, PixelFormat::Nv16), Some(8 * 4 * 2));
1735    }
1736
1737    #[test]
1738    fn elem_count_overflow_is_none() {
1739        // w*h already overflows usize → None (never a wrapped, undersized count).
1740        assert_eq!(pbo_elem_count(usize::MAX, 2, PixelFormat::Rgba), None);
1741        // w*h fits but *channels overflows → None.
1742        assert_eq!(pbo_elem_count(usize::MAX, 1, PixelFormat::Rgb), None);
1743    }
1744
1745    #[test]
1746    fn shape_per_format() {
1747        // Planar: [channels, height, width].
1748        assert_eq!(pbo_shape(8, 4, PixelFormat::PlanarRgb), vec![3, 4, 8]);
1749        // SemiPlanar NV12: [height*3/2, width].
1750        assert_eq!(pbo_shape(8, 4, PixelFormat::Nv12), vec![4 * 3 / 2, 8]);
1751        // SemiPlanar NV16: [height*2, width].
1752        assert_eq!(pbo_shape(8, 4, PixelFormat::Nv16), vec![4 * 2, 8]);
1753        // Packed: [height, width, channels].
1754        assert_eq!(pbo_shape(8, 4, PixelFormat::Rgba), vec![4, 8, 4]);
1755    }
1756
1757    /// Regression test for the cross-handle PBO map race task 11's review
1758    /// found: `PboHandle::acquire_map`'s per-handle `Mutex<MapState>`
1759    /// (`edgefirst-tensor`) has no visibility into a SEPARATE `PboHandle`
1760    /// for the same real `buffer_id` -- exactly what a cross-cdylib
1761    /// reconstructed consumer tensor now is, alongside the producer's own
1762    /// tensor, since task 11 wired a real `kind::PBO` `import_descriptor`
1763    /// arm. Simulates that with two independent `PboTensor`s built
1764    /// directly (rather than through the full descriptor/vtable round
1765    /// trip, which needs a second compiled `.so` to be meaningful) that
1766    /// both point at the SAME real GL buffer via a fresh [`GlPboOps`]
1767    /// wrapping the SAME sender -- exactly the relationship `import_descriptor`'s
1768    /// PBO arm produces between a producer's handle and a reconstructed
1769    /// one. Without `mapped_pbo_buffers`, the second map would reach
1770    /// `glMapBufferRange` on an already-mapped buffer (undefined
1771    /// behaviour); this asserts it is refused instead, and that the
1772    /// refusal is a real, releasable reservation rather than a permanently
1773    /// stuck `buffer_id`.
1774    #[test]
1775    #[cfg(any(target_os = "linux", target_os = "windows"))] // PBO destinations: Linux + Windows
1776    fn concurrent_maps_from_two_independent_handles_on_one_buffer_do_not_corrupt_gl_state() {
1777        let gl = match GLProcessorThreaded::new(None) {
1778            Ok(gl) => gl,
1779            Err(e) => {
1780                crate::test_support::report_skip(&format!(
1781                    "concurrent_maps_from_two_independent_handles... - \
1782                     OpenGL not available: {e:?}"
1783                ));
1784                return;
1785            }
1786        };
1787        let tensor_a = gl
1788            .create_pbo_image(64, 64, PixelFormat::Rgba)
1789            .expect("a PBO on a live GL context");
1790        let buffer_id = tensor_a.pbo_id().expect("just created a PBO tensor");
1791        let size = tensor_a.capacity_bytes();
1792        let shape = tensor_a.shape().to_vec();
1793
1794        // A second, independent PboHandle for the IDENTICAL real buffer --
1795        // same relationship a producer's handle and a reconstructed
1796        // consumer's handle have to the same GL context.
1797        let ops_b: std::sync::Arc<dyn edgefirst_tensor::PboOps> = std::sync::Arc::new(GlPboOps {
1798            sender: gl.sender.as_ref().unwrap().downgrade(),
1799            worker_id: gl.worker_id,
1800        });
1801        let tensor_b =
1802            edgefirst_tensor::PboTensor::<u8>::from_pbo(buffer_id, size, &shape, None, ops_b)
1803                .expect("from_pbo against a real, already-created buffer_id must succeed");
1804
1805        // tensor_a maps first and holds the mapping open.
1806        let map_a = tensor_a.map().expect("first map must succeed");
1807
1808        // tensor_b's own Mutex<MapState> starts at Unmapped and knows
1809        // nothing about tensor_a's outstanding mapping -- without the
1810        // GL-thread-level fix this reaches glMapBufferRange a second time.
1811        let map_b_result = tensor_b.map();
1812        // `glMapBufferRange` on an already-mapped buffer is UNDEFINED
1813        // BEHAVIOUR per the GL spec, not guaranteed to fail cleanly on
1814        // every driver -- so this asserts the SPECIFIC refusal
1815        // `mapped_pbo_buffers` produces (matching this file's own error
1816        // text), not merely "some error occurred", which a
1817        // driver-dependent null return could also produce and would make
1818        // this test pass even with the fix reverted.
1819        let Err(map_b_err) = map_b_result else {
1820            panic!(
1821                "a second, independent handle mapping the SAME real buffer while the \
1822                     first mapping is still open must be refused, not silently succeed"
1823            );
1824        };
1825        let msg = format!("{map_b_err:?}");
1826        assert!(
1827            msg.contains("already mapped"),
1828            "must be refused specifically by the GL-thread-level reservation check \
1829             (mapped_pbo_buffers), not some other error path: {msg}"
1830        );
1831
1832        drop(map_a);
1833
1834        // Once the first mapping releases, the SAME second handle must be
1835        // able to map successfully -- proves the refusal above was a real,
1836        // releasable reservation, not a permanently stuck buffer_id.
1837        let map_b_after = tensor_b.map();
1838        assert!(
1839            map_b_after.is_ok(),
1840            "after the first handle's mapping releases, the second handle must be able to \
1841             map the same buffer: {map_b_after:?}"
1842        );
1843    }
1844}