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