1use 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, tokio::sync::oneshot::Sender<Result<(), Error>>,
26 ),
27 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 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, Option<SendablePtr<TensorDyn>>, Option<[f32; 4]>, crate::ColorMode,
52 tokio::sync::oneshot::Sender<Result<(), Error>>,
53 ),
54 DrawProtoMasks(
55 SendablePtr<TensorDyn>,
56 SendablePtr<DetectBox>,
57 SendablePtr<ProtoData>,
58 f32, Option<SendablePtr<TensorDyn>>, Option<[f32; 4]>, crate::ColorMode,
62 tokio::sync::oneshot::Sender<Result<(), Error>>,
63 ),
64 SetInt8Interpolation(
65 Int8InterpolationMode,
66 tokio::sync::oneshot::Sender<Result<(), Error>>,
67 ),
68 SetColorimetryMode(
70 crate::ColorimetryMode,
71 tokio::sync::oneshot::Sender<Result<(), Error>>,
72 ),
73 SetInterpolation(
75 crate::Interpolation,
76 tokio::sync::oneshot::Sender<Result<(), Error>>,
77 ),
78 #[cfg(test)]
80 SetFailDstImport(bool, tokio::sync::oneshot::Sender<Result<(), Error>>),
81 EglCacheStats(tokio::sync::oneshot::Sender<Result<super::cache::GlCacheStats, Error>>),
83 ConvertStats(tokio::sync::oneshot::Sender<Result<super::cache::ConvertStats, Error>>),
85 PboCreate(
86 usize, tokio::sync::oneshot::Sender<Result<u32, Error>>,
88 ),
89 PboMap(
90 u32, usize, tokio::sync::oneshot::Sender<Result<edgefirst_tensor::PboMapping, Error>>,
93 ),
94 PboUnmap(
95 u32, tokio::sync::oneshot::Sender<Result<(), Error>>,
97 ),
98 PboDelete(u32), CudaRegisterBuffer(u32, tokio::sync::oneshot::Sender<Option<usize>>),
100 CudaMap(usize, tokio::sync::oneshot::Sender<Option<(usize, usize)>>),
101 CudaUnmap(usize), CudaUnregister(usize), }
104
105fn pbo_elem_count(
115 width: usize,
116 height: usize,
117 format: edgefirst_tensor::PixelFormat,
118) -> Option<usize> {
119 format
126 .allocation_shape(width, height)?
127 .iter()
128 .try_fold(1usize, |acc, &d| acc.checked_mul(d))
129}
130
131fn 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
140fn 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
177struct GlPboOps {
184 sender: WeakSender<GLProcessorMessage>,
185 worker_id: u64,
190}
191
192static NEXT_GL_WORKER_ID: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
194
195struct GlWorkerState {
197 id: u64,
198 serializes: bool,
201 mapped: std::collections::HashSet<u32>,
203}
204
205thread_local! {
206 static GL_WORKER: std::cell::RefCell<Option<GlWorkerState>> =
222 const { std::cell::RefCell::new(None) };
223}
224
225fn 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
232fn 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
237fn on_serializing_gl_worker() -> bool {
240 GL_WORKER.with(|c| matches!(*c.borrow(), Some(ref s) if s.serializes))
241}
242
243fn 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 static GL_WORKER_POISON: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
258}
259
260fn 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
280unsafe 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 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 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 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 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 if on_owning_gl_worker(self.worker_id) {
348 with_worker_mapped(|set| set.remove(&buffer_id));
349 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 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
376struct 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#[derive(Debug)]
416pub struct GLProcessorThreaded {
417 handle: Option<JoinHandle<()>>,
419
420 sender: Option<Sender<GLProcessorMessage>>,
422 worker_id: u64,
425 owned_pbos: std::sync::Arc<super::processor::OwnedPbos>,
428 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
444fn 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 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 #[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 #[cfg(any(target_os = "macos", target_os = "ios", target_os = "android"))]
561 let init_result = GLProcessorST::new(kind, capacity);
562 #[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 let caps = gl_converter.platform_caps();
585 let _ = create_ctx_send.send(Ok(caps));
586 let mut poisoned = false;
587 GL_WORKER.with(|c| {
609 *c.borrow_mut() = Some(GlWorkerState {
610 id: worker_id,
611 serializes: false,
613 mapped: std::collections::HashSet::new(),
614 });
615 });
616 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 let serialize_per_msg = match std::env::var("EDGEFIRST_GL_SERIALIZE").as_deref() {
635 Ok("full") => true,
636 Ok("lifecycle") => {
637 #[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 #[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 if poisoned {
691 reject_poisoned_message(msg);
692 continue;
693 }
694
695 gl_converter.begin_gpu_pass();
699 handle_gl_message(msg, &mut gl_converter, &mut poisoned);
700 poisoned |= GL_WORKER_POISON.with(|p| p.replace(false));
704 gl_converter.end_gpu_pass_if_synced();
708 }
709 #[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#[cfg(not(target_os = "linux"))]
725static NON_LINUX_GL_MUTEX: std::sync::Mutex<()> = std::sync::Mutex::new(());
726
727#[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 let _ = resp.send(edgefirst_tensor::gl_register_buffer(buffer_id));
872 }
873 GLProcessorMessage::CudaMap(resource, resp) => {
874 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 let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
904 let src = unsafe { src.ptr.as_ref() };
905 let dst = unsafe { dst.ptr.as_mut() };
906 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 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 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 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
1039fn 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 with_worker_mapped(|set| set.remove(&buffer_id));
1076 }
1077 match result {
1078 Ok(result) => result,
1079 Err(panic) => std::panic::resume_unwind(panic),
1082 }
1083}
1084
1085fn pbo_unmap_on_gl_thread(buffer_id: u32) -> Result<(), Error> {
1087 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 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 pub fn new(kind: Option<EglDisplayKind>) -> Result<Self, Error> {
1147 Self::with_cache_capacity(kind, None)
1148 }
1149
1150 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, 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, 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 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 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 #[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 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 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 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 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 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 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 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_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 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 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 pub(crate) fn transfer_backend(&self) -> TransferBackend {
1653 self.caps.transfer_backend
1654 }
1655
1656 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 drop(self.sender.take());
1675 let _ = self.handle.take().and_then(|h| h.join().ok());
1676 }
1677}
1678
1679#[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 assert_eq!(pbo_elem_count(8, 4, PixelFormat::Rgba), Some(8 * 4 * 4));
1692 assert_eq!(pbo_elem_count(8, 4, PixelFormat::Rgb), Some(8 * 4 * 3));
1694 assert_eq!(pbo_elem_count(8, 4, PixelFormat::Nv12), Some(8 * 4 * 3 / 2));
1696 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 assert_eq!(pbo_elem_count(usize::MAX, 2, PixelFormat::Rgba), None);
1704 assert_eq!(pbo_elem_count(usize::MAX, 1, PixelFormat::Rgb), None);
1706 }
1707
1708 #[test]
1709 fn shape_per_format() {
1710 assert_eq!(pbo_shape(8, 4, PixelFormat::PlanarRgb), vec![3, 4, 8]);
1712 assert_eq!(pbo_shape(8, 4, PixelFormat::Nv12), vec![4 * 3 / 2, 8]);
1714 assert_eq!(pbo_shape(8, 4, PixelFormat::Nv16), vec![4 * 2, 8]);
1716 assert_eq!(pbo_shape(8, 4, PixelFormat::Rgba), vec![4, 8, 4]);
1718 }
1719
1720 #[test]
1738 #[cfg(any(target_os = "linux", target_os = "windows"))] 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 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 let map_a = tensor_a.map().expect("first map must succeed");
1777
1778 let map_b_result = tensor_b.map();
1782 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 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}