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::{ConvertOptions, Error, ImageProcessorTrait};
14use edgefirst_tensor::TensorDyn;
15
16#[allow(clippy::type_complexity)]
17enum GLProcessorMessage {
18 ImageConvert(
19 SendablePtr<TensorDyn>,
20 SendablePtr<TensorDyn>,
21 ConvertOptions,
22 bool, tokio::sync::oneshot::Sender<Result<(), Error>>,
24 ),
25 ImageConvertFenced(
29 SendablePtr<TensorDyn>,
30 SendablePtr<TensorDyn>,
31 ConvertOptions,
32 tokio::sync::oneshot::Sender<Result<Option<super::CompletionFence>, Error>>,
33 ),
34 Flush(tokio::sync::oneshot::Sender<Result<(), Error>>),
36 SetColors(
37 Vec<[u8; 4]>,
38 tokio::sync::oneshot::Sender<Result<(), Error>>,
39 ),
40 SetBoxLineWidth(u32, tokio::sync::oneshot::Sender<Result<(), Error>>),
41 DrawDecodedMasks(
42 SendablePtr<TensorDyn>,
43 SendablePtr<DetectBox>,
44 SendablePtr<Segmentation>,
45 f32, Option<SendablePtr<TensorDyn>>, Option<[f32; 4]>, crate::ColorMode,
49 tokio::sync::oneshot::Sender<Result<(), Error>>,
50 ),
51 DrawProtoMasks(
52 SendablePtr<TensorDyn>,
53 SendablePtr<DetectBox>,
54 SendablePtr<ProtoData>,
55 f32, Option<SendablePtr<TensorDyn>>, Option<[f32; 4]>, crate::ColorMode,
59 tokio::sync::oneshot::Sender<Result<(), Error>>,
60 ),
61 SetInt8Interpolation(
62 Int8InterpolationMode,
63 tokio::sync::oneshot::Sender<Result<(), Error>>,
64 ),
65 SetColorimetryMode(
67 crate::ColorimetryMode,
68 tokio::sync::oneshot::Sender<Result<(), Error>>,
69 ),
70 SetInterpolation(
72 crate::Interpolation,
73 tokio::sync::oneshot::Sender<Result<(), Error>>,
74 ),
75 #[cfg(test)]
77 SetFailDstImport(bool, tokio::sync::oneshot::Sender<Result<(), Error>>),
78 EglCacheStats(tokio::sync::oneshot::Sender<Result<super::cache::GlCacheStats, Error>>),
80 ConvertStats(tokio::sync::oneshot::Sender<Result<super::cache::ConvertStats, Error>>),
82 PboCreate(
83 usize, tokio::sync::oneshot::Sender<Result<u32, Error>>,
85 ),
86 PboMap(
87 u32, usize, tokio::sync::oneshot::Sender<Result<edgefirst_tensor::PboMapping, Error>>,
90 ),
91 PboUnmap(
92 u32, tokio::sync::oneshot::Sender<Result<(), Error>>,
94 ),
95 PboDelete(u32), CudaRegisterBuffer(u32, tokio::sync::oneshot::Sender<Option<usize>>),
97 CudaMap(usize, tokio::sync::oneshot::Sender<Option<(usize, usize)>>),
98 CudaUnmap(usize), CudaUnregister(usize), }
101
102fn pbo_elem_count(
112 width: usize,
113 height: usize,
114 format: edgefirst_tensor::PixelFormat,
115) -> Option<usize> {
116 format
123 .allocation_shape(width, height)?
124 .iter()
125 .try_fold(1usize, |acc, &d| acc.checked_mul(d))
126}
127
128fn pbo_shape(width: usize, height: usize, format: edgefirst_tensor::PixelFormat) -> Vec<usize> {
132 format
133 .allocation_shape(width, height)
134 .unwrap_or_else(|| vec![height, width, format.channels()])
135}
136
137fn register_pbo_cuda<T>(
146 tensor: &mut edgefirst_tensor::Tensor<T>,
147 buffer_id: u32,
148 size: usize,
149 sender: &Sender<GLProcessorMessage>,
150) where
151 T: num_traits::Num + Clone + std::fmt::Debug + Send + Sync + edgefirst_tensor::Element,
152{
153 if !edgefirst_tensor::is_cuda_available() {
154 return;
155 }
156 let (tx, rx) = tokio::sync::oneshot::channel();
157 if sender
158 .blocking_send(GLProcessorMessage::CudaRegisterBuffer(buffer_id, tx))
159 .is_ok()
160 {
161 if let Ok(Some(resource)) = rx.blocking_recv() {
162 let ops = std::sync::Arc::new(GlCudaOps {
163 sender: sender.downgrade(),
164 });
165 tensor.set_cuda_handle(edgefirst_tensor::CudaHandle::new_gl(
166 resource as *mut std::ffi::c_void,
167 size,
168 ops,
169 ));
170 }
171 }
172}
173
174struct GlPboOps {
181 sender: WeakSender<GLProcessorMessage>,
182 worker_id: u64,
187}
188
189static NEXT_GL_WORKER_ID: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
191
192struct GlWorkerState {
194 id: u64,
195 serializes: bool,
198 mapped: std::collections::HashSet<u32>,
200}
201
202thread_local! {
203 static GL_WORKER: std::cell::RefCell<Option<GlWorkerState>> =
219 const { std::cell::RefCell::new(None) };
220}
221
222fn with_worker_mapped<R>(f: impl FnOnce(&mut std::collections::HashSet<u32>) -> R) -> Option<R> {
226 GL_WORKER.with(|c| c.borrow_mut().as_mut().map(|s| f(&mut s.mapped)))
227}
228
229fn on_owning_gl_worker(worker_id: u64) -> bool {
231 GL_WORKER.with(|c| matches!(*c.borrow(), Some(ref s) if s.id == worker_id))
232}
233
234fn on_serializing_gl_worker() -> bool {
237 GL_WORKER.with(|c| matches!(*c.borrow(), Some(ref s) if s.serializes))
238}
239
240fn cross_context_pbo_error(op: &str, buffer_id: u32) -> edgefirst_tensor::Error {
244 edgefirst_tensor::Error::InvalidOperation(format!(
245 "PBO {buffer_id}: cannot {op} a buffer owned by another GL context while \
246 this one holds the serialization lock; convert it through the processor \
247 that created it"
248 ))
249}
250
251thread_local! {
252 static GL_WORKER_POISON: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
255}
256
257fn inline_gl_op<T>(what: &str, f: impl FnOnce() -> Result<T, Error>) -> Result<T, Error> {
265 match std::panic::catch_unwind(AssertUnwindSafe(f)) {
266 Ok(result) => result,
267 Err(e) => {
268 GL_WORKER_POISON.with(|p| p.set(true));
269 Err(Error::Internal(format!(
270 "GL thread panicked during {what}: {}",
271 panic_message(e.as_ref()),
272 )))
273 }
274 }
275}
276
277unsafe impl edgefirst_tensor::PboOps for GlPboOps {
283 fn map_buffer(
284 &self,
285 buffer_id: u32,
286 size: usize,
287 ) -> edgefirst_tensor::Result<edgefirst_tensor::PboMapping> {
288 if on_owning_gl_worker(self.worker_id) {
293 return inline_gl_op("PboMap", || pbo_map_on_gl_thread(buffer_id, size)).map_err(|e| {
294 edgefirst_tensor::Error::NotImplemented(format!("GL PBO map failed: {e:?}"))
295 });
296 }
297 if on_serializing_gl_worker() {
299 return Err(cross_context_pbo_error("map", buffer_id));
300 }
301 let sender = self
302 .sender
303 .upgrade()
304 .ok_or(edgefirst_tensor::Error::PboDisconnected)?;
305 let (tx, rx) = tokio::sync::oneshot::channel();
306 sender
307 .blocking_send(GLProcessorMessage::PboMap(buffer_id, size, tx))
308 .map_err(|_| edgefirst_tensor::Error::PboDisconnected)?;
309 rx.blocking_recv()
310 .map_err(|_| edgefirst_tensor::Error::PboDisconnected)?
311 .map_err(|e| {
312 edgefirst_tensor::Error::NotImplemented(format!("GL PBO map failed: {e:?}"))
313 })
314 }
315
316 fn unmap_buffer(&self, buffer_id: u32) -> edgefirst_tensor::Result<()> {
317 if on_owning_gl_worker(self.worker_id) {
319 return inline_gl_op("PboUnmap", || pbo_unmap_on_gl_thread(buffer_id)).map_err(|e| {
320 edgefirst_tensor::Error::NotImplemented(format!("GL PBO unmap failed: {e:?}"))
321 });
322 }
323 if on_serializing_gl_worker() {
325 return Err(cross_context_pbo_error("unmap", buffer_id));
326 }
327 let sender = self
328 .sender
329 .upgrade()
330 .ok_or(edgefirst_tensor::Error::PboDisconnected)?;
331 let (tx, rx) = tokio::sync::oneshot::channel();
332 sender
333 .blocking_send(GLProcessorMessage::PboUnmap(buffer_id, tx))
334 .map_err(|_| edgefirst_tensor::Error::PboDisconnected)?;
335 rx.blocking_recv()
336 .map_err(|_| edgefirst_tensor::Error::PboDisconnected)?
337 .map_err(|e| {
338 edgefirst_tensor::Error::NotImplemented(format!("GL PBO unmap failed: {e:?}"))
339 })
340 }
341
342 fn delete_buffer(&self, buffer_id: u32) {
343 if on_owning_gl_worker(self.worker_id) {
345 with_worker_mapped(|set| set.remove(&buffer_id));
346 let _ = inline_gl_op("PboDelete", || {
349 unsafe { edgefirst_gl::gl::DeleteBuffers(1, &buffer_id) };
350 Ok(())
351 });
352 return;
353 }
354 if let Some(sender) = self.sender.upgrade() {
355 if on_serializing_gl_worker() {
357 if sender
358 .try_send(GLProcessorMessage::PboDelete(buffer_id))
359 .is_err()
360 {
361 log::warn!(
362 "PBO {buffer_id}: cross-context delete skipped to avoid a deadlock; \
363 the buffer is freed when its context is destroyed"
364 );
365 }
366 return;
367 }
368 let _ = sender.blocking_send(GLProcessorMessage::PboDelete(buffer_id));
369 }
370 }
371}
372
373struct GlCudaOps {
380 sender: WeakSender<GLProcessorMessage>,
381}
382
383impl edgefirst_tensor::CudaGlOps for GlCudaOps {
384 fn map(&self, resource: *mut std::ffi::c_void) -> Option<(*mut std::ffi::c_void, usize)> {
385 let sender = self.sender.upgrade()?;
386 let (tx, rx) = tokio::sync::oneshot::channel();
387 sender
388 .blocking_send(GLProcessorMessage::CudaMap(resource as usize, tx))
389 .ok()?;
390 rx.blocking_recv()
391 .ok()?
392 .map(|(p, l)| (p as *mut std::ffi::c_void, l))
393 }
394
395 fn unmap(&self, resource: *mut std::ffi::c_void) {
396 if let Some(sender) = self.sender.upgrade() {
397 let _ = sender.blocking_send(GLProcessorMessage::CudaUnmap(resource as usize));
398 }
399 }
400
401 fn unregister(&self, resource: *mut std::ffi::c_void) {
402 if let Some(sender) = self.sender.upgrade() {
403 let _ = sender.blocking_send(GLProcessorMessage::CudaUnregister(resource as usize));
404 }
405 }
406}
407
408#[derive(Debug)]
413pub struct GLProcessorThreaded {
414 handle: Option<JoinHandle<()>>,
416
417 sender: Option<Sender<GLProcessorMessage>>,
419 worker_id: u64,
422 owned_pbos: std::sync::Arc<super::processor::OwnedPbos>,
425 caps: super::platform::PlatformCaps,
429}
430
431unsafe impl Send for GLProcessorThreaded {}
432unsafe impl Sync for GLProcessorThreaded {}
433
434struct SendablePtr<T: Send> {
435 ptr: NonNull<T>,
436 len: usize,
437}
438
439unsafe impl<T> Send for SendablePtr<T> where T: Send {}
440
441fn panic_message(info: &(dyn std::any::Any + Send)) -> String {
443 if let Some(s) = info.downcast_ref::<&str>() {
444 s.to_string()
445 } else if let Some(s) = info.downcast_ref::<String>() {
446 s.clone()
447 } else {
448 "unknown panic".to_string()
449 }
450}
451
452fn reply_caught<T>(
453 result: std::thread::Result<crate::Result<T>>,
454 resp: tokio::sync::oneshot::Sender<crate::Result<T>>,
455 poisoned: &mut bool,
456 what: &str,
457) {
458 let _ = resp.send(match result {
459 Ok(res) => res,
460 Err(e) => {
461 *poisoned = true;
462 Err(crate::Error::Internal(format!(
463 "GL thread panicked during {what}: {}",
464 panic_message(e.as_ref()),
465 )))
466 }
467 });
468}
469
470fn reject_poisoned_message(msg: GLProcessorMessage) {
471 let poison_err =
472 crate::Error::Internal("GL context is poisoned after a prior panic".to_string());
473 match msg {
474 GLProcessorMessage::ImageConvertFenced(.., resp) => {
475 let _ = resp.send(Err(poison_err));
476 }
477 GLProcessorMessage::ImageConvert(.., resp) => {
478 let _ = resp.send(Err(poison_err));
479 }
480 GLProcessorMessage::Flush(resp) => {
481 let _ = resp.send(Err(poison_err));
482 }
483 GLProcessorMessage::DrawDecodedMasks(.., resp) => {
484 let _ = resp.send(Err(poison_err));
485 }
486 GLProcessorMessage::DrawProtoMasks(.., resp) => {
487 let _ = resp.send(Err(poison_err));
488 }
489 GLProcessorMessage::SetColors(_, resp) => {
490 let _ = resp.send(Err(poison_err));
491 }
492 GLProcessorMessage::SetBoxLineWidth(_, resp) => {
493 let _ = resp.send(Err(poison_err));
494 }
495 GLProcessorMessage::SetInt8Interpolation(_, resp) => {
496 let _ = resp.send(Err(poison_err));
497 }
498 GLProcessorMessage::SetColorimetryMode(_, resp) => {
499 let _ = resp.send(Err(poison_err));
500 }
501 GLProcessorMessage::SetInterpolation(_, resp) => {
502 let _ = resp.send(Err(poison_err));
503 }
504 #[cfg(test)]
505 GLProcessorMessage::SetFailDstImport(_, resp) => {
506 let _ = resp.send(Err(poison_err));
507 }
508 GLProcessorMessage::EglCacheStats(resp) => {
509 let _ = resp.send(Err(poison_err));
510 }
511 GLProcessorMessage::ConvertStats(resp) => {
512 let _ = resp.send(Err(poison_err));
513 }
514 GLProcessorMessage::PboCreate(_, resp) => {
515 let _ = resp.send(Err(poison_err));
516 }
517 GLProcessorMessage::PboMap(_, _, resp) => {
518 let _ = resp.send(Err(poison_err));
519 }
520 GLProcessorMessage::PboUnmap(_, resp) => {
521 let _ = resp.send(Err(poison_err));
522 }
523 GLProcessorMessage::PboDelete(_) => {}
524 GLProcessorMessage::CudaRegisterBuffer(_, resp) => {
525 let _ = resp.send(None);
526 }
527 GLProcessorMessage::CudaMap(_, resp) => {
528 let _ = resp.send(None);
529 }
530 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, options, defer, resp) => {
749 handle_image_convert(src, dst, (options, defer), resp, gl_converter, poisoned);
750 }
751 GLProcessorMessage::ImageConvertFenced(src, dst, options, resp) => {
752 handle_image_convert_fenced(src, dst, options, resp, gl_converter, poisoned);
753 }
754 GLProcessorMessage::Flush(resp) => {
755 let result = std::panic::catch_unwind(AssertUnwindSafe(|| gl_converter.flush()));
756 reply_caught(result, resp, poisoned, "Flush");
757 }
758 GLProcessorMessage::DrawDecodedMasks(
759 dst,
760 det,
761 seg,
762 opacity,
763 bg,
764 content_region,
765 color_mode,
766 resp,
767 ) => {
768 handle_draw_decoded_masks(
769 dst,
770 det,
771 seg,
772 (opacity, bg, content_region, color_mode),
773 resp,
774 gl_converter,
775 poisoned,
776 );
777 }
778 GLProcessorMessage::DrawProtoMasks(
779 dst,
780 det,
781 proto_data,
782 opacity,
783 bg,
784 content_region,
785 color_mode,
786 resp,
787 ) => {
788 handle_draw_proto_masks(
789 dst,
790 det,
791 proto_data,
792 (opacity, bg, content_region, color_mode),
793 resp,
794 gl_converter,
795 poisoned,
796 );
797 }
798 GLProcessorMessage::SetColors(colors, resp) => {
799 let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
800 gl_converter.set_class_colors(&colors)
801 }));
802 reply_caught(result, resp, poisoned, "SetColors");
803 }
804 GLProcessorMessage::SetBoxLineWidth(width, resp) => {
805 let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
806 gl_converter.set_box_line_width(width)
807 }));
808 reply_caught(result, resp, poisoned, "SetBoxLineWidth");
809 }
810 GLProcessorMessage::SetColorimetryMode(mode, resp) => {
811 let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
812 gl_converter.set_colorimetry_mode(mode);
813 Ok(())
814 }));
815 reply_caught(result, resp, poisoned, "SetColorimetryMode");
816 }
817 GLProcessorMessage::SetInterpolation(mode, resp) => {
818 let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
819 gl_converter.set_interpolation(mode);
820 Ok(())
821 }));
822 reply_caught(result, resp, poisoned, "SetInterpolation");
823 }
824 GLProcessorMessage::SetInt8Interpolation(mode, resp) => {
825 let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
826 gl_converter.set_int8_interpolation_mode(mode);
827 Ok(())
828 }));
829 reply_caught(result, resp, poisoned, "SetInt8Interpolation");
830 }
831 #[cfg(test)]
832 GLProcessorMessage::SetFailDstImport(fail, resp) => {
833 let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
834 gl_converter.fail_dst_import = fail;
835 Ok(())
836 }));
837 reply_caught(result, resp, poisoned, "SetFailDstImport");
838 }
839 GLProcessorMessage::EglCacheStats(resp) => {
840 let result =
841 std::panic::catch_unwind(AssertUnwindSafe(|| Ok(gl_converter.egl_cache_stats())));
842 reply_caught(result, resp, poisoned, "EglCacheStats");
843 }
844 GLProcessorMessage::ConvertStats(resp) => {
845 let result =
846 std::panic::catch_unwind(AssertUnwindSafe(|| Ok(gl_converter.convert_stats())));
847 reply_caught(result, resp, poisoned, "ConvertStats");
848 }
849 GLProcessorMessage::PboCreate(size, resp) => {
850 handle_pbo_create(size, resp, poisoned);
851 }
852 GLProcessorMessage::PboMap(buffer_id, size, resp) => {
853 handle_pbo_map(buffer_id, size, resp, poisoned);
854 }
855 GLProcessorMessage::PboUnmap(buffer_id, resp) => {
856 handle_pbo_unmap(buffer_id, resp, poisoned);
857 }
858 GLProcessorMessage::PboDelete(buffer_id) => {
859 handle_pbo_delete(buffer_id, poisoned);
860 }
861 GLProcessorMessage::CudaRegisterBuffer(buffer_id, resp) => {
862 let _ = resp.send(edgefirst_tensor::gl_register_buffer(buffer_id));
864 }
865 GLProcessorMessage::CudaMap(resource, resp) => {
866 gl_converter.flush_pending();
873 let _ = resp.send(edgefirst_tensor::gl_map_resource(resource));
874 }
875 GLProcessorMessage::CudaUnmap(resource) => {
876 edgefirst_tensor::gl_unmap_resource(resource);
877 }
878 GLProcessorMessage::CudaUnregister(resource) => {
879 edgefirst_tensor::gl_unregister_resource(resource);
880 }
881 }
882}
883
884fn handle_image_convert(
885 src: SendablePtr<TensorDyn>,
886 mut dst: SendablePtr<TensorDyn>,
887 geom: (ConvertOptions, bool),
888 resp: tokio::sync::oneshot::Sender<Result<(), Error>>,
889 gl_converter: &mut GLProcessorST,
890 poisoned: &mut bool,
891) {
892 let (options, defer) = geom;
893 let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
896 let src = unsafe { src.ptr.as_ref() };
897 let dst = unsafe { dst.ptr.as_mut() };
898 if defer {
902 gl_converter.convert_deferred(src, dst, Some(options))
903 } else {
904 gl_converter.convert(src, dst, Some(options))
905 }
906 }));
907 reply_caught(result, resp, poisoned, "ImageConvert");
908}
909
910fn handle_image_convert_fenced(
911 src: SendablePtr<TensorDyn>,
912 mut dst: SendablePtr<TensorDyn>,
913 options: ConvertOptions,
914 resp: tokio::sync::oneshot::Sender<Result<Option<super::CompletionFence>, Error>>,
915 gl_converter: &mut GLProcessorST,
916 poisoned: &mut bool,
917) {
918 let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
921 let src = unsafe { src.ptr.as_ref() };
922 let dst = unsafe { dst.ptr.as_mut() };
923 gl_converter.convert_fenced(src, dst, options)
924 }));
925 reply_caught(result, resp, poisoned, "ImageConvertFenced");
926}
927
928fn handle_draw_decoded_masks(
929 mut dst: SendablePtr<TensorDyn>,
930 det: SendablePtr<DetectBox>,
931 seg: SendablePtr<Segmentation>,
932 overlay: (
933 f32,
934 Option<SendablePtr<TensorDyn>>,
935 Option<[f32; 4]>,
936 crate::ColorMode,
937 ),
938 resp: tokio::sync::oneshot::Sender<Result<(), Error>>,
939 gl_converter: &mut GLProcessorST,
940 poisoned: &mut bool,
941) {
942 let (opacity, bg, content_region, color_mode) = overlay;
943 let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
947 let dst = unsafe { dst.ptr.as_mut() };
948 let det = unsafe { std::slice::from_raw_parts(det.ptr.as_ptr(), det.len) };
949 let seg = unsafe { std::slice::from_raw_parts(seg.ptr.as_ptr(), seg.len) };
950 let bg_ref = bg.map(|p| unsafe { &*p.ptr.as_ptr() });
951 gl_converter.draw_decoded_masks(
952 dst,
953 det,
954 seg,
955 crate::MaskOverlay {
956 background: bg_ref,
957 opacity,
958 content_region,
959 color_mode,
960 },
961 )
962 }));
963 reply_caught(result, resp, poisoned, "DrawDecodedMasks");
964}
965
966fn handle_draw_proto_masks(
967 mut dst: SendablePtr<TensorDyn>,
968 det: SendablePtr<DetectBox>,
969 proto_data: SendablePtr<ProtoData>,
970 overlay: (
971 f32,
972 Option<SendablePtr<TensorDyn>>,
973 Option<[f32; 4]>,
974 crate::ColorMode,
975 ),
976 resp: tokio::sync::oneshot::Sender<Result<(), Error>>,
977 gl_converter: &mut GLProcessorST,
978 poisoned: &mut bool,
979) {
980 let (opacity, bg, content_region, color_mode) = overlay;
981 let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
984 let dst = unsafe { dst.ptr.as_mut() };
985 let det = unsafe { std::slice::from_raw_parts(det.ptr.as_ptr(), det.len) };
986 let bg_ref = bg.map(|p| unsafe { &*p.ptr.as_ptr() });
987 let proto_data = unsafe { proto_data.ptr.as_ref() };
988 gl_converter.draw_proto_masks(
989 dst,
990 det,
991 proto_data,
992 crate::MaskOverlay {
993 background: bg_ref,
994 opacity,
995 content_region,
996 color_mode,
997 },
998 )
999 }));
1000 reply_caught(result, resp, poisoned, "DrawProtoMasks");
1001}
1002
1003fn handle_pbo_create(
1004 size: usize,
1005 resp: tokio::sync::oneshot::Sender<Result<u32, Error>>,
1006 poisoned: &mut bool,
1007) {
1008 let result = std::panic::catch_unwind(AssertUnwindSafe(|| unsafe {
1009 let mut id: u32 = 0;
1010 edgefirst_gl::gl::GenBuffers(1, &mut id);
1011 edgefirst_gl::gl::BindBuffer(edgefirst_gl::gl::PIXEL_PACK_BUFFER, id);
1012 edgefirst_gl::gl::BufferData(
1013 edgefirst_gl::gl::PIXEL_PACK_BUFFER,
1014 size as isize,
1015 std::ptr::null(),
1016 edgefirst_gl::gl::STREAM_COPY,
1017 );
1018 edgefirst_gl::gl::BindBuffer(edgefirst_gl::gl::PIXEL_PACK_BUFFER, 0);
1019 match check_gl_error("PboCreate", 0) {
1020 Ok(()) => Ok(id),
1021 Err(e) => {
1022 edgefirst_gl::gl::DeleteBuffers(1, &id);
1023 Err(e)
1024 }
1025 }
1026 }));
1027 reply_caught(result, resp, poisoned, "PboCreate");
1028}
1029
1030fn pbo_map_on_gl_thread(
1034 buffer_id: u32,
1035 size: usize,
1036) -> Result<edgefirst_tensor::PboMapping, Error> {
1037 let already = with_worker_mapped(|set| !set.insert(buffer_id)).unwrap_or(false);
1038 if already {
1039 return Err(Error::OpenGl(format!(
1040 "PBO buffer {buffer_id} is already mapped in this GL context \
1041 (a producer and a reconstructed cross-package consumer may \
1042 both hold a live tensor over it)"
1043 )));
1044 }
1045 let result = std::panic::catch_unwind(AssertUnwindSafe(|| unsafe {
1046 edgefirst_gl::gl::BindBuffer(edgefirst_gl::gl::PIXEL_PACK_BUFFER, buffer_id);
1047 let ptr = edgefirst_gl::gl::MapBufferRange(
1048 edgefirst_gl::gl::PIXEL_PACK_BUFFER,
1049 0,
1050 size as isize,
1051 edgefirst_gl::gl::MAP_READ_BIT | edgefirst_gl::gl::MAP_WRITE_BIT,
1052 );
1053 edgefirst_gl::gl::BindBuffer(edgefirst_gl::gl::PIXEL_PACK_BUFFER, 0);
1054 if ptr.is_null() {
1055 Err(Error::OpenGl("glMapBufferRange returned null".to_string()))
1056 } else {
1057 Ok(edgefirst_tensor::PboMapping {
1058 ptr: ptr as *mut u8,
1059 size,
1060 })
1061 }
1062 }));
1063 if !matches!(result, Ok(Ok(_))) {
1064 with_worker_mapped(|set| set.remove(&buffer_id));
1067 }
1068 match result {
1069 Ok(result) => result,
1070 Err(panic) => std::panic::resume_unwind(panic),
1073 }
1074}
1075
1076fn pbo_unmap_on_gl_thread(buffer_id: u32) -> Result<(), Error> {
1078 with_worker_mapped(|set| set.remove(&buffer_id));
1081 unsafe {
1082 edgefirst_gl::gl::BindBuffer(edgefirst_gl::gl::PIXEL_PACK_BUFFER, buffer_id);
1083 let ok = edgefirst_gl::gl::UnmapBuffer(edgefirst_gl::gl::PIXEL_PACK_BUFFER);
1084 edgefirst_gl::gl::BindBuffer(edgefirst_gl::gl::PIXEL_PACK_BUFFER, 0);
1085 if ok == edgefirst_gl::gl::FALSE {
1086 Err(Error::OpenGl(
1087 "PBO data was corrupted during mapping".into(),
1088 ))
1089 } else {
1090 check_gl_error("PboUnmap", 0)
1091 }
1092 }
1093}
1094
1095fn handle_pbo_map(
1096 buffer_id: u32,
1097 size: usize,
1098 resp: tokio::sync::oneshot::Sender<Result<edgefirst_tensor::PboMapping, Error>>,
1099 poisoned: &mut bool,
1100) {
1101 let result =
1102 std::panic::catch_unwind(AssertUnwindSafe(|| pbo_map_on_gl_thread(buffer_id, size)));
1103 reply_caught(result, resp, poisoned, "PboMap");
1104}
1105
1106fn handle_pbo_unmap(
1107 buffer_id: u32,
1108 resp: tokio::sync::oneshot::Sender<Result<(), Error>>,
1109 poisoned: &mut bool,
1110) {
1111 let result = std::panic::catch_unwind(AssertUnwindSafe(|| pbo_unmap_on_gl_thread(buffer_id)));
1112 reply_caught(result, resp, poisoned, "PboUnmap");
1113}
1114
1115fn handle_pbo_delete(buffer_id: u32, poisoned: &mut bool) {
1116 with_worker_mapped(|set| set.remove(&buffer_id));
1123 if let Err(e) = std::panic::catch_unwind(AssertUnwindSafe(|| unsafe {
1124 edgefirst_gl::gl::DeleteBuffers(1, &buffer_id);
1125 })) {
1126 *poisoned = true;
1127 log::error!(
1128 "GL thread panicked during PboDelete: {}",
1129 panic_message(e.as_ref()),
1130 );
1131 }
1132}
1133
1134impl GLProcessorThreaded {
1135 pub fn new(kind: Option<EglDisplayKind>) -> Result<Self, Error> {
1138 Self::with_cache_capacity(kind, None)
1139 }
1140
1141 pub fn with_cache_capacity(
1147 kind: Option<EglDisplayKind>,
1148 capacity: Option<usize>,
1149 ) -> Result<Self, Error> {
1150 let worker_id = NEXT_GL_WORKER_ID.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
1151 let (send, recv) = tokio::sync::mpsc::channel::<GLProcessorMessage>(1);
1152
1153 let (create_ctx_send, create_ctx_recv) = tokio::sync::oneshot::channel();
1154
1155 let owned_pbos = std::sync::Arc::<super::processor::OwnedPbos>::default();
1156 let worker_owned = owned_pbos.clone();
1157 let handle = std::thread::spawn(move || {
1158 run_gl_worker(
1159 worker_id,
1160 worker_owned,
1161 kind,
1162 capacity,
1163 recv,
1164 create_ctx_send,
1165 );
1166 });
1167
1168 let caps = match create_ctx_recv.blocking_recv() {
1169 Ok(Err(e)) => return Err(e),
1170 Err(_) => {
1171 return Err(Error::Internal(
1172 "GL converter error messaging closed without update".to_string(),
1173 ));
1174 }
1175 Ok(Ok(caps)) => caps,
1176 };
1177
1178 Ok(Self {
1179 handle: Some(handle),
1180 sender: Some(send),
1181 worker_id,
1182 owned_pbos,
1183 caps,
1184 })
1185 }
1186}
1187
1188impl ImageProcessorTrait for GLProcessorThreaded {
1189 fn convert(
1190 &mut self,
1191 src: &TensorDyn,
1192 dst: &mut TensorDyn,
1193 options: Option<ConvertOptions>,
1194 ) -> crate::Result<()> {
1195 let options = options.unwrap_or_default();
1196 let (err_send, err_recv) = tokio::sync::oneshot::channel();
1197 self.sender
1198 .as_ref()
1199 .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1200 .blocking_send(GLProcessorMessage::ImageConvert(
1201 SendablePtr {
1202 ptr: NonNull::from(src),
1203 len: 1,
1204 },
1205 SendablePtr {
1206 ptr: NonNull::from(dst),
1207 len: 1,
1208 },
1209 options,
1210 false, err_send,
1212 ))
1213 .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1214 err_recv.blocking_recv().map_err(|_| {
1215 Error::Internal("GL converter error messaging closed without update".to_string())
1216 })?
1217 }
1218
1219 fn convert_deferred(
1220 &mut self,
1221 src: &TensorDyn,
1222 dst: &mut TensorDyn,
1223 options: Option<ConvertOptions>,
1224 ) -> crate::Result<()> {
1225 let options = options.unwrap_or_default();
1226 let (err_send, err_recv) = tokio::sync::oneshot::channel();
1227 self.sender
1228 .as_ref()
1229 .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1230 .blocking_send(GLProcessorMessage::ImageConvert(
1231 SendablePtr {
1232 ptr: NonNull::from(src),
1233 len: 1,
1234 },
1235 SendablePtr {
1236 ptr: NonNull::from(dst),
1237 len: 1,
1238 },
1239 options,
1240 true, err_send,
1242 ))
1243 .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1244 err_recv.blocking_recv().map_err(|_| {
1245 Error::Internal("GL converter error messaging closed without update".to_string())
1246 })?
1247 }
1248
1249 fn flush(&mut self) -> crate::Result<()> {
1250 let (err_send, err_recv) = tokio::sync::oneshot::channel();
1251 self.sender
1252 .as_ref()
1253 .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1254 .blocking_send(GLProcessorMessage::Flush(err_send))
1255 .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1256 err_recv.blocking_recv().map_err(|_| {
1257 Error::Internal("GL converter error messaging closed without update".to_string())
1258 })?
1259 }
1260
1261 fn draw_decoded_masks(
1262 &mut self,
1263 dst: &mut TensorDyn,
1264 detect: &[crate::DetectBox],
1265 segmentation: &[crate::Segmentation],
1266 overlay: crate::MaskOverlay<'_>,
1267 ) -> crate::Result<()> {
1268 let (err_send, err_recv) = tokio::sync::oneshot::channel();
1269 self.sender
1270 .as_ref()
1271 .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1272 .blocking_send(GLProcessorMessage::DrawDecodedMasks(
1273 SendablePtr {
1274 ptr: NonNull::from(dst),
1275 len: 1,
1276 },
1277 SendablePtr {
1278 ptr: NonNull::new(detect.as_ptr() as *mut DetectBox).unwrap(),
1279 len: detect.len(),
1280 },
1281 SendablePtr {
1282 ptr: NonNull::new(segmentation.as_ptr() as *mut Segmentation).unwrap(),
1283 len: segmentation.len(),
1284 },
1285 overlay.opacity,
1286 overlay.background.map(|bg| SendablePtr {
1287 ptr: NonNull::from(bg).cast::<TensorDyn>(),
1288 len: 1,
1289 }),
1290 overlay.content_region,
1291 overlay.color_mode,
1292 err_send,
1293 ))
1294 .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1295 err_recv.blocking_recv().map_err(|_| {
1296 Error::Internal("GL converter error messaging closed without update".to_string())
1297 })?
1298 }
1299
1300 fn draw_proto_masks(
1301 &mut self,
1302 dst: &mut TensorDyn,
1303 detect: &[DetectBox],
1304 proto_data: &ProtoData,
1305 overlay: crate::MaskOverlay<'_>,
1306 ) -> crate::Result<()> {
1307 let (err_send, err_recv) = tokio::sync::oneshot::channel();
1308 self.sender
1309 .as_ref()
1310 .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1311 .blocking_send(GLProcessorMessage::DrawProtoMasks(
1312 SendablePtr {
1313 ptr: NonNull::from(dst),
1314 len: 1,
1315 },
1316 SendablePtr {
1317 ptr: NonNull::new(detect.as_ptr() as *mut DetectBox).unwrap(),
1318 len: detect.len(),
1319 },
1320 SendablePtr {
1321 ptr: NonNull::from(proto_data).cast::<ProtoData>(),
1322 len: 1,
1323 },
1324 overlay.opacity,
1325 overlay.background.map(|bg| SendablePtr {
1326 ptr: NonNull::from(bg).cast::<TensorDyn>(),
1327 len: 1,
1328 }),
1329 overlay.content_region,
1330 overlay.color_mode,
1331 err_send,
1332 ))
1333 .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1334 err_recv.blocking_recv().map_err(|_| {
1335 Error::Internal("GL converter error messaging closed without update".to_string())
1336 })?
1337 }
1338
1339 fn set_class_colors(&mut self, colors: &[[u8; 4]]) -> Result<(), crate::Error> {
1340 let (err_send, err_recv) = tokio::sync::oneshot::channel();
1341 self.sender
1342 .as_ref()
1343 .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1344 .blocking_send(GLProcessorMessage::SetColors(colors.to_vec(), err_send))
1345 .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1346 err_recv.blocking_recv().map_err(|_| {
1347 Error::Internal("GL converter error messaging closed without update".to_string())
1348 })?
1349 }
1350
1351 fn set_box_line_width(&mut self, width: u32) -> Result<(), crate::Error> {
1352 crate::check_box_line_width(width)?;
1353 let (err_send, err_recv) = tokio::sync::oneshot::channel();
1354 self.sender
1355 .as_ref()
1356 .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1357 .blocking_send(GLProcessorMessage::SetBoxLineWidth(width, err_send))
1358 .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1359 err_recv.blocking_recv().map_err(|_| {
1360 Error::Internal("GL converter error messaging closed without update".to_string())
1361 })?
1362 }
1363}
1364
1365impl GLProcessorThreaded {
1366 pub fn set_colorimetry_mode(&mut self, mode: crate::ColorimetryMode) -> Result<(), Error> {
1371 let (err_send, err_recv) = tokio::sync::oneshot::channel();
1372 self.sender
1373 .as_ref()
1374 .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1375 .blocking_send(GLProcessorMessage::SetColorimetryMode(mode, err_send))
1376 .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1377 err_recv.blocking_recv().map_err(|_| {
1378 Error::Internal("GL converter error messaging closed without update".to_string())
1379 })?
1380 }
1381
1382 pub fn set_interpolation(&mut self, mode: crate::Interpolation) -> Result<(), Error> {
1384 let (err_send, err_recv) = tokio::sync::oneshot::channel();
1385 self.sender
1386 .as_ref()
1387 .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1388 .blocking_send(GLProcessorMessage::SetInterpolation(mode, err_send))
1389 .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1390 err_recv.blocking_recv().map_err(|_| {
1391 Error::Internal("GL converter error messaging closed without update".to_string())
1392 })?
1393 }
1394
1395 #[cfg(test)]
1401 pub(crate) fn set_fail_dst_import(&mut self, fail: bool) -> Result<(), Error> {
1402 let (err_send, err_recv) = tokio::sync::oneshot::channel();
1403 self.sender
1404 .as_ref()
1405 .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1406 .blocking_send(GLProcessorMessage::SetFailDstImport(fail, err_send))
1407 .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1408 err_recv.blocking_recv().map_err(|_| {
1409 Error::Internal("GL converter error messaging closed without update".to_string())
1410 })?
1411 }
1412
1413 pub fn set_int8_interpolation_mode(
1415 &mut self,
1416 mode: Int8InterpolationMode,
1417 ) -> Result<(), crate::Error> {
1418 let (err_send, err_recv) = tokio::sync::oneshot::channel();
1419 self.sender
1420 .as_ref()
1421 .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1422 .blocking_send(GLProcessorMessage::SetInt8Interpolation(mode, err_send))
1423 .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1424 err_recv.blocking_recv().map_err(|_| {
1425 Error::Internal("GL converter error messaging closed without update".to_string())
1426 })?
1427 }
1428
1429 pub fn egl_cache_stats(&self) -> Result<super::cache::GlCacheStats, Error> {
1436 let (send, recv) = tokio::sync::oneshot::channel();
1437 self.sender
1438 .as_ref()
1439 .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1440 .blocking_send(GLProcessorMessage::EglCacheStats(send))
1441 .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1442 recv.blocking_recv().map_err(|_| {
1443 Error::Internal("GL converter error messaging closed without update".to_string())
1444 })?
1445 }
1446
1447 pub fn convert_with_fence(
1458 &mut self,
1459 src: &TensorDyn,
1460 dst: &mut TensorDyn,
1461 options: Option<ConvertOptions>,
1462 ) -> Result<Option<super::CompletionFence>, Error> {
1463 let options = options.unwrap_or_default();
1464 use crate::ImageProcessorTrait as _;
1465 if !self.caps.native_fence_sync {
1466 self.convert(src, dst, Some(options))?;
1469 return Ok(None);
1470 }
1471 let (send, recv) = tokio::sync::oneshot::channel();
1472 self.sender
1473 .as_ref()
1474 .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1475 .blocking_send(GLProcessorMessage::ImageConvertFenced(
1476 SendablePtr {
1477 ptr: NonNull::from(src),
1478 len: 1,
1479 },
1480 SendablePtr {
1481 ptr: NonNull::from(dst),
1482 len: 1,
1483 },
1484 options,
1485 send,
1486 ))
1487 .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1488 recv.blocking_recv().map_err(|_| {
1489 Error::Internal("GL converter error messaging closed without update".to_string())
1490 })?
1491 }
1492
1493 pub fn convert_stats(&self) -> Result<super::cache::ConvertStats, Error> {
1498 let (send, recv) = tokio::sync::oneshot::channel();
1499 self.sender
1500 .as_ref()
1501 .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1502 .blocking_send(GLProcessorMessage::ConvertStats(send))
1503 .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1504 recv.blocking_recv().map_err(|_| {
1505 Error::Internal("GL converter error messaging closed without update".to_string())
1506 })?
1507 }
1508
1509 pub fn create_pbo_image(
1515 &self,
1516 width: usize,
1517 height: usize,
1518 format: edgefirst_tensor::PixelFormat,
1519 ) -> Result<edgefirst_tensor::Tensor<u8>, Error> {
1520 let sender = self
1521 .sender
1522 .as_ref()
1523 .ok_or(Error::OpenGl("GL processor is shutting down".to_string()))?;
1524
1525 let size = pbo_elem_count(width, height, format)
1526 .filter(|&s| s != 0)
1527 .ok_or_else(|| Error::OpenGl("Invalid image dimensions".to_string()))?;
1528
1529 let (tx, rx) = tokio::sync::oneshot::channel();
1531 sender
1532 .blocking_send(GLProcessorMessage::PboCreate(size, tx))
1533 .map_err(|_| Error::OpenGl("GL thread channel closed".to_string()))?;
1534 let buffer_id = rx
1535 .blocking_recv()
1536 .map_err(|_| Error::OpenGl("GL thread did not respond".to_string()))??;
1537
1538 let ops: std::sync::Arc<dyn edgefirst_tensor::PboOps> = std::sync::Arc::new(GlPboOps {
1539 sender: sender.downgrade(),
1540 worker_id: self.worker_id,
1541 });
1542 let owned_ops = ops.clone();
1543
1544 let shape = pbo_shape(width, height, format);
1545
1546 let pbo_tensor =
1547 edgefirst_tensor::PboTensor::<u8>::from_pbo(buffer_id, size, &shape, None, ops)
1548 .map_err(|e| Error::OpenGl(format!("PBO tensor creation failed: {e:?}")))?;
1549 let mut tensor = edgefirst_tensor::Tensor::from_pbo(pbo_tensor)
1550 .map_err(|e| Error::OpenGl(format!("PBO tensor wrap failed: {e:?}")))?;
1551 tensor
1552 .set_format(format)
1553 .map_err(|e| Error::OpenGl(format!("Failed to set format on PBO tensor: {e:?}")))?;
1554 register_pbo_cuda(&mut tensor, buffer_id, size, sender);
1561 self.owned_pbos.insert(tensor.pbo_vtable_ptr(), &owned_ops);
1562 Ok(tensor)
1563 }
1564
1565 pub(crate) fn create_pbo_image_dtype(
1571 &self,
1572 width: usize,
1573 height: usize,
1574 format: edgefirst_tensor::PixelFormat,
1575 dtype: edgefirst_tensor::DType,
1576 ) -> Result<TensorDyn, Error> {
1577 let sender = self
1578 .sender
1579 .as_ref()
1580 .ok_or(Error::OpenGl("GL processor is shutting down".to_string()))?;
1581
1582 let elems = pbo_elem_count(width, height, format)
1583 .filter(|&e| e != 0)
1584 .ok_or_else(|| Error::OpenGl("Invalid image dimensions".to_string()))?;
1585
1586 let size = elems
1587 .checked_mul(dtype.size())
1588 .ok_or_else(|| Error::OpenGl("PBO size overflow".to_string()))?;
1589
1590 let (tx, rx) = tokio::sync::oneshot::channel();
1592 sender
1593 .blocking_send(GLProcessorMessage::PboCreate(size, tx))
1594 .map_err(|_| Error::OpenGl("GL thread channel closed".to_string()))?;
1595 let buffer_id = rx
1596 .blocking_recv()
1597 .map_err(|_| Error::OpenGl("GL thread did not respond".to_string()))??;
1598
1599 let ops: std::sync::Arc<dyn edgefirst_tensor::PboOps> = std::sync::Arc::new(GlPboOps {
1600 sender: sender.downgrade(),
1601 worker_id: self.worker_id,
1602 });
1603 let owned_ops = ops.clone();
1604
1605 let shape = pbo_shape(width, height, format);
1606
1607 let map_err = |e: edgefirst_tensor::Error| {
1608 Error::OpenGl(format!("PBO tensor creation failed: {e:?}"))
1609 };
1610 let set_err = |e: edgefirst_tensor::Error| {
1611 Error::OpenGl(format!("Failed to set format on PBO tensor: {e:?}"))
1612 };
1613
1614 let tensor = match dtype {
1615 edgefirst_tensor::DType::U8 => {
1616 let pbo =
1617 edgefirst_tensor::PboTensor::<u8>::from_pbo(buffer_id, size, &shape, None, ops)
1618 .map_err(map_err)?;
1619 let mut t = edgefirst_tensor::Tensor::from_pbo(pbo).map_err(map_err)?;
1620 t.set_format(format).map_err(set_err)?;
1621 register_pbo_cuda(&mut t, buffer_id, size, sender);
1622 TensorDyn::from(t)
1623 }
1624 edgefirst_tensor::DType::F16 => {
1625 let pbo = edgefirst_tensor::PboTensor::<edgefirst_tensor::f16>::from_pbo(
1626 buffer_id, size, &shape, None, ops,
1627 )
1628 .map_err(map_err)?;
1629 let mut t = edgefirst_tensor::Tensor::from_pbo(pbo).map_err(map_err)?;
1630 t.set_format(format).map_err(set_err)?;
1631 register_pbo_cuda(&mut t, buffer_id, size, sender);
1632 TensorDyn::from(t)
1633 }
1634 edgefirst_tensor::DType::F32 => {
1635 let pbo = edgefirst_tensor::PboTensor::<f32>::from_pbo(
1636 buffer_id, size, &shape, None, ops,
1637 )
1638 .map_err(map_err)?;
1639 let mut t = edgefirst_tensor::Tensor::from_pbo(pbo).map_err(map_err)?;
1640 t.set_format(format).map_err(set_err)?;
1641 register_pbo_cuda(&mut t, buffer_id, size, sender);
1642 TensorDyn::from(t)
1643 }
1644 other => return Err(Error::OpenGl(format!("unsupported PBO dtype {other:?}"))),
1645 };
1646 self.owned_pbos.insert(tensor.pbo_vtable_ptr(), &owned_ops);
1647 Ok(tensor)
1648 }
1649
1650 pub(crate) fn transfer_backend(&self) -> TransferBackend {
1652 self.caps.transfer_backend
1653 }
1654
1655 pub(crate) fn supported_render_dtypes(&self) -> crate::RenderDtypeSupport {
1662 self.caps.render_dtypes
1663 }
1664
1665 pub(crate) fn dst_cpu_access(
1670 &self,
1671 memory: edgefirst_tensor::TensorMemory,
1672 format: edgefirst_tensor::PixelFormat,
1673 dtype: edgefirst_tensor::DType,
1674 ) -> Option<edgefirst_tensor::CpuAccess> {
1675 use edgefirst_tensor::{CpuAccess, DType, PixelFormat, TensorMemory};
1676 let renders = match dtype {
1677 DType::U8 | DType::I8 => GLProcessorST::dst_format_renderable(
1678 self.caps.transfer_backend,
1679 memory,
1680 format,
1681 self.caps.has_bgra,
1682 ),
1683 DType::F16 | DType::F32 => {
1684 super::float_dispatch::classify_float_render(
1685 PixelFormat::Rgba,
1686 format,
1687 dtype,
1688 memory,
1689 self.caps.render_dtypes,
1690 <super::platform::Platform as super::platform::GlPlatform>::ZERO_COPY_FLOAT,
1691 ) != super::float_dispatch::FloatRenderPath::None
1692 }
1693 _ => false,
1694 };
1695 let readback = matches!(memory, TensorMemory::Mem | TensorMemory::Shm);
1696 renders.then_some(if readback {
1697 CpuAccess::Write
1698 } else {
1699 CpuAccess::None
1700 })
1701 }
1702}
1703
1704impl Drop for GLProcessorThreaded {
1705 fn drop(&mut self) {
1706 drop(self.sender.take());
1712 let _ = self.handle.take().and_then(|h| h.join().ok());
1713 }
1714}
1715
1716#[cfg(test)]
1720#[cfg_attr(coverage_nightly, coverage(off))]
1721mod tests {
1722 use super::{pbo_elem_count, pbo_shape, GLProcessorThreaded, GlPboOps};
1723 use edgefirst_tensor::{PixelFormat, TensorTrait};
1724
1725 #[test]
1726 fn elem_count_per_format() {
1727 assert_eq!(pbo_elem_count(8, 4, PixelFormat::Rgba), Some(8 * 4 * 4));
1729 assert_eq!(pbo_elem_count(8, 4, PixelFormat::Rgb), Some(8 * 4 * 3));
1731 assert_eq!(pbo_elem_count(8, 4, PixelFormat::Nv12), Some(8 * 4 * 3 / 2));
1733 assert_eq!(pbo_elem_count(8, 4, PixelFormat::Nv16), Some(8 * 4 * 2));
1735 }
1736
1737 #[test]
1738 fn elem_count_overflow_is_none() {
1739 assert_eq!(pbo_elem_count(usize::MAX, 2, PixelFormat::Rgba), None);
1741 assert_eq!(pbo_elem_count(usize::MAX, 1, PixelFormat::Rgb), None);
1743 }
1744
1745 #[test]
1746 fn shape_per_format() {
1747 assert_eq!(pbo_shape(8, 4, PixelFormat::PlanarRgb), vec![3, 4, 8]);
1749 assert_eq!(pbo_shape(8, 4, PixelFormat::Nv12), vec![4 * 3 / 2, 8]);
1751 assert_eq!(pbo_shape(8, 4, PixelFormat::Nv16), vec![4 * 2, 8]);
1753 assert_eq!(pbo_shape(8, 4, PixelFormat::Rgba), vec![4, 8, 4]);
1755 }
1756
1757 #[test]
1775 #[cfg(any(target_os = "linux", target_os = "windows"))] fn concurrent_maps_from_two_independent_handles_on_one_buffer_do_not_corrupt_gl_state() {
1777 let gl = match GLProcessorThreaded::new(None) {
1778 Ok(gl) => gl,
1779 Err(e) => {
1780 crate::test_support::report_skip(&format!(
1781 "concurrent_maps_from_two_independent_handles... - \
1782 OpenGL not available: {e:?}"
1783 ));
1784 return;
1785 }
1786 };
1787 let tensor_a = gl
1788 .create_pbo_image(64, 64, PixelFormat::Rgba)
1789 .expect("a PBO on a live GL context");
1790 let buffer_id = tensor_a.pbo_id().expect("just created a PBO tensor");
1791 let size = tensor_a.capacity_bytes();
1792 let shape = tensor_a.shape().to_vec();
1793
1794 let ops_b: std::sync::Arc<dyn edgefirst_tensor::PboOps> = std::sync::Arc::new(GlPboOps {
1798 sender: gl.sender.as_ref().unwrap().downgrade(),
1799 worker_id: gl.worker_id,
1800 });
1801 let tensor_b =
1802 edgefirst_tensor::PboTensor::<u8>::from_pbo(buffer_id, size, &shape, None, ops_b)
1803 .expect("from_pbo against a real, already-created buffer_id must succeed");
1804
1805 let map_a = tensor_a.map().expect("first map must succeed");
1807
1808 let map_b_result = tensor_b.map();
1812 let Err(map_b_err) = map_b_result else {
1820 panic!(
1821 "a second, independent handle mapping the SAME real buffer while the \
1822 first mapping is still open must be refused, not silently succeed"
1823 );
1824 };
1825 let msg = format!("{map_b_err:?}");
1826 assert!(
1827 msg.contains("already mapped"),
1828 "must be refused specifically by the GL-thread-level reservation check \
1829 (mapped_pbo_buffers), not some other error path: {msg}"
1830 );
1831
1832 drop(map_a);
1833
1834 let map_b_after = tensor_b.map();
1838 assert!(
1839 map_b_after.is_ok(),
1840 "after the first handle's mapping releases, the second handle must be able to \
1841 map the same buffer: {map_b_after:?}"
1842 );
1843 }
1844}