1use edgefirst_decoder::{DetectBox, ProtoData, Segmentation};
5use std::panic::AssertUnwindSafe;
6use std::ptr::NonNull;
7use std::thread::JoinHandle;
8use tokio::sync::mpsc::{Sender, WeakSender};
9
10use super::processor::GLProcessorST;
11use super::shaders::check_gl_error;
12use super::{EglDisplayKind, Int8InterpolationMode, TransferBackend};
13use crate::{Crop, Error, Flip, ImageProcessorTrait, Rotation};
14use edgefirst_tensor::TensorDyn;
15
16#[allow(clippy::type_complexity)]
17enum GLProcessorMessage {
18 ImageConvert(
19 SendablePtr<TensorDyn>,
20 SendablePtr<TensorDyn>,
21 Rotation,
22 Flip,
23 Crop,
24 bool, 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<std::os::fd::OwnedFd>, 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 EglCacheStats(tokio::sync::oneshot::Sender<Result<super::cache::GlCacheStats, Error>>),
75 ConvertStats(tokio::sync::oneshot::Sender<Result<super::cache::ConvertStats, Error>>),
77 PboCreate(
78 usize, tokio::sync::oneshot::Sender<Result<u32, Error>>,
80 ),
81 PboMap(
82 u32, usize, tokio::sync::oneshot::Sender<Result<edgefirst_tensor::PboMapping, Error>>,
85 ),
86 PboUnmap(
87 u32, tokio::sync::oneshot::Sender<Result<(), Error>>,
89 ),
90 PboDelete(u32), CudaRegisterBuffer(u32, tokio::sync::oneshot::Sender<Option<usize>>),
92 CudaMap(usize, tokio::sync::oneshot::Sender<Option<(usize, usize)>>),
93 CudaUnmap(usize), CudaUnregister(usize), }
96
97fn pbo_elem_count(
107 width: usize,
108 height: usize,
109 format: edgefirst_tensor::PixelFormat,
110) -> Option<usize> {
111 format
118 .image_shape(width, height)?
119 .iter()
120 .try_fold(1usize, |acc, &d| acc.checked_mul(d))
121}
122
123fn pbo_shape(width: usize, height: usize, format: edgefirst_tensor::PixelFormat) -> Vec<usize> {
127 format
128 .image_shape(width, height)
129 .unwrap_or_else(|| vec![height, width, format.channels()])
130}
131
132fn register_pbo_cuda<T>(
141 tensor: &mut edgefirst_tensor::Tensor<T>,
142 buffer_id: u32,
143 size: usize,
144 sender: &Sender<GLProcessorMessage>,
145) where
146 T: num_traits::Num + Clone + std::fmt::Debug + Send + Sync,
147{
148 if !edgefirst_tensor::is_cuda_available() {
149 return;
150 }
151 let (tx, rx) = tokio::sync::oneshot::channel();
152 if sender
153 .blocking_send(GLProcessorMessage::CudaRegisterBuffer(buffer_id, tx))
154 .is_ok()
155 {
156 if let Ok(Some(resource)) = rx.blocking_recv() {
157 let ops = std::sync::Arc::new(GlCudaOps {
158 sender: sender.downgrade(),
159 });
160 tensor.set_cuda_handle(edgefirst_tensor::CudaHandle::new_gl(
161 resource as *mut std::ffi::c_void,
162 size,
163 ops,
164 ));
165 }
166 }
167}
168
169struct GlPboOps {
176 sender: WeakSender<GLProcessorMessage>,
177}
178
179unsafe impl edgefirst_tensor::PboOps for GlPboOps {
184 fn map_buffer(
185 &self,
186 buffer_id: u32,
187 size: usize,
188 ) -> edgefirst_tensor::Result<edgefirst_tensor::PboMapping> {
189 let sender = self
190 .sender
191 .upgrade()
192 .ok_or(edgefirst_tensor::Error::PboDisconnected)?;
193 let (tx, rx) = tokio::sync::oneshot::channel();
194 sender
195 .blocking_send(GLProcessorMessage::PboMap(buffer_id, size, tx))
196 .map_err(|_| edgefirst_tensor::Error::PboDisconnected)?;
197 rx.blocking_recv()
198 .map_err(|_| edgefirst_tensor::Error::PboDisconnected)?
199 .map_err(|e| {
200 edgefirst_tensor::Error::NotImplemented(format!("GL PBO map failed: {e:?}"))
201 })
202 }
203
204 fn unmap_buffer(&self, buffer_id: u32) -> edgefirst_tensor::Result<()> {
205 let sender = self
206 .sender
207 .upgrade()
208 .ok_or(edgefirst_tensor::Error::PboDisconnected)?;
209 let (tx, rx) = tokio::sync::oneshot::channel();
210 sender
211 .blocking_send(GLProcessorMessage::PboUnmap(buffer_id, tx))
212 .map_err(|_| edgefirst_tensor::Error::PboDisconnected)?;
213 rx.blocking_recv()
214 .map_err(|_| edgefirst_tensor::Error::PboDisconnected)?
215 .map_err(|e| {
216 edgefirst_tensor::Error::NotImplemented(format!("GL PBO unmap failed: {e:?}"))
217 })
218 }
219
220 fn delete_buffer(&self, buffer_id: u32) {
221 if let Some(sender) = self.sender.upgrade() {
222 let _ = sender.blocking_send(GLProcessorMessage::PboDelete(buffer_id));
223 }
224 }
225}
226
227struct GlCudaOps {
234 sender: WeakSender<GLProcessorMessage>,
235}
236
237impl edgefirst_tensor::CudaGlOps for GlCudaOps {
238 fn map(&self, resource: *mut std::ffi::c_void) -> Option<(*mut std::ffi::c_void, usize)> {
239 let sender = self.sender.upgrade()?;
240 let (tx, rx) = tokio::sync::oneshot::channel();
241 sender
242 .blocking_send(GLProcessorMessage::CudaMap(resource as usize, tx))
243 .ok()?;
244 rx.blocking_recv()
245 .ok()?
246 .map(|(p, l)| (p as *mut std::ffi::c_void, l))
247 }
248
249 fn unmap(&self, resource: *mut std::ffi::c_void) {
250 if let Some(sender) = self.sender.upgrade() {
251 let _ = sender.blocking_send(GLProcessorMessage::CudaUnmap(resource as usize));
252 }
253 }
254
255 fn unregister(&self, resource: *mut std::ffi::c_void) {
256 if let Some(sender) = self.sender.upgrade() {
257 let _ = sender.blocking_send(GLProcessorMessage::CudaUnregister(resource as usize));
258 }
259 }
260}
261
262#[derive(Debug)]
267pub struct GLProcessorThreaded {
268 handle: Option<JoinHandle<()>>,
270
271 sender: Option<Sender<GLProcessorMessage>>,
273 caps: super::platform::PlatformCaps,
277}
278
279unsafe impl Send for GLProcessorThreaded {}
280unsafe impl Sync for GLProcessorThreaded {}
281
282struct SendablePtr<T: Send> {
283 ptr: NonNull<T>,
284 len: usize,
285}
286
287unsafe impl<T> Send for SendablePtr<T> where T: Send {}
288
289fn panic_message(info: &(dyn std::any::Any + Send)) -> String {
291 if let Some(s) = info.downcast_ref::<&str>() {
292 s.to_string()
293 } else if let Some(s) = info.downcast_ref::<String>() {
294 s.clone()
295 } else {
296 "unknown panic".to_string()
297 }
298}
299
300impl GLProcessorThreaded {
301 pub fn new(kind: Option<EglDisplayKind>) -> Result<Self, Error> {
303 let (send, mut recv) = tokio::sync::mpsc::channel::<GLProcessorMessage>(1);
304
305 let (create_ctx_send, create_ctx_recv) = tokio::sync::oneshot::channel();
306
307 let func = move || {
308 #[cfg(target_os = "linux")]
313 let init_result = {
314 let _guard = super::context::GL_MUTEX
315 .lock()
316 .unwrap_or_else(|e| e.into_inner());
317 GLProcessorST::new(kind)
318 };
319 #[cfg(any(target_os = "macos", target_os = "ios", target_os = "android"))]
323 let init_result = GLProcessorST::new(kind);
324 let mut gl_converter = match init_result {
325 Ok(gl) => gl,
326 Err(e) => {
327 let _ = create_ctx_send.send(Err(e));
328 return;
329 }
330 };
331 let caps = gl_converter.platform_caps();
334 let _ = create_ctx_send.send(Ok(caps));
335 let mut poisoned = false;
336 let serialize_per_msg = match std::env::var("EDGEFIRST_GL_SERIALIZE").as_deref() {
345 Ok("full") => true,
346 Ok("lifecycle") => false,
347 _ => caps.serialize_gl,
348 };
349 log::debug!(
350 "GL serialization policy: {}",
351 if serialize_per_msg {
352 "Full (per-message global lock)"
353 } else {
354 "LifecycleOnly (parallel across processors)"
355 }
356 );
357 while let Some(msg) = recv.blocking_recv() {
358 #[cfg(target_os = "linux")]
367 let _guard = serialize_per_msg.then(|| {
368 super::context::GL_MUTEX
369 .lock()
370 .unwrap_or_else(|e| e.into_inner())
371 });
372 #[cfg(any(target_os = "macos", target_os = "ios", target_os = "android"))]
373 let _guard = serialize_per_msg.then(|| {
374 static MSG_MUTEX: std::sync::Mutex<()> = std::sync::Mutex::new(());
375 MSG_MUTEX.lock().unwrap_or_else(|e| e.into_inner())
376 });
377
378 if poisoned {
383 let poison_err = crate::Error::Internal(
384 "GL context is poisoned after a prior panic".to_string(),
385 );
386 match msg {
387 GLProcessorMessage::ImageConvertFenced(.., resp) => {
388 let _ = resp.send(Err(poison_err));
389 }
390 GLProcessorMessage::ImageConvert(.., resp) => {
391 let _ = resp.send(Err(poison_err));
392 }
393 GLProcessorMessage::Flush(resp) => {
394 let _ = resp.send(Err(poison_err));
395 }
396 GLProcessorMessage::DrawDecodedMasks(.., resp) => {
397 let _ = resp.send(Err(poison_err));
398 }
399 GLProcessorMessage::DrawProtoMasks(.., resp) => {
400 let _ = resp.send(Err(poison_err));
401 }
402 GLProcessorMessage::SetColors(_, resp) => {
403 let _ = resp.send(Err(poison_err));
404 }
405 GLProcessorMessage::SetInt8Interpolation(_, resp) => {
406 let _ = resp.send(Err(poison_err));
407 }
408 GLProcessorMessage::SetColorimetryMode(_, resp) => {
409 let _ = resp.send(Err(poison_err));
410 }
411 GLProcessorMessage::EglCacheStats(resp) => {
412 let _ = resp.send(Err(poison_err));
413 }
414 GLProcessorMessage::ConvertStats(resp) => {
415 let _ = resp.send(Err(poison_err));
416 }
417 GLProcessorMessage::PboCreate(_, resp) => {
418 let _ = resp.send(Err(poison_err));
419 }
420 GLProcessorMessage::PboMap(_, _, resp) => {
421 let _ = resp.send(Err(poison_err));
422 }
423 GLProcessorMessage::PboUnmap(_, resp) => {
424 let _ = resp.send(Err(poison_err));
425 }
426 GLProcessorMessage::PboDelete(_) => {}
427 GLProcessorMessage::CudaRegisterBuffer(_, resp) => {
428 let _ = resp.send(None);
429 }
430 GLProcessorMessage::CudaMap(_, resp) => {
431 let _ = resp.send(None);
432 }
433 GLProcessorMessage::CudaUnmap(_) => {}
437 GLProcessorMessage::CudaUnregister(_) => {}
438 }
439 continue;
440 }
441
442 match msg {
443 GLProcessorMessage::ImageConvert(
444 src,
445 mut dst,
446 rotation,
447 flip,
448 crop,
449 defer,
450 resp,
451 ) => {
452 let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
455 let src = unsafe { src.ptr.as_ref() };
456 let dst = unsafe { dst.ptr.as_mut() };
457 if defer {
461 gl_converter.convert_deferred(src, dst, rotation, flip, crop)
462 } else {
463 gl_converter.convert(src, dst, rotation, flip, crop)
464 }
465 }));
466 let _ = resp.send(match result {
467 Ok(res) => res,
468 Err(e) => {
469 poisoned = true;
470 Err(crate::Error::Internal(format!(
471 "GL thread panicked during ImageConvert: {}",
472 panic_message(e.as_ref()),
473 )))
474 }
475 });
476 }
477 GLProcessorMessage::ImageConvertFenced(
478 src,
479 mut dst,
480 rotation,
481 flip,
482 crop,
483 resp,
484 ) => {
485 let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
488 let src = unsafe { src.ptr.as_ref() };
489 let dst = unsafe { dst.ptr.as_mut() };
490 gl_converter.convert_fenced(src, dst, rotation, flip, crop)
491 }));
492 let _ = resp.send(match result {
493 Ok(res) => res,
494 Err(e) => {
495 poisoned = true;
496 Err(crate::Error::Internal(format!(
497 "GL thread panicked during ImageConvertFenced: {}",
498 panic_message(e.as_ref()),
499 )))
500 }
501 });
502 }
503 GLProcessorMessage::Flush(resp) => {
504 let result =
505 std::panic::catch_unwind(AssertUnwindSafe(|| gl_converter.flush()));
506 let _ = resp.send(match result {
507 Ok(res) => res,
508 Err(e) => {
509 poisoned = true;
510 Err(crate::Error::Internal(format!(
511 "GL thread panicked during Flush: {}",
512 panic_message(e.as_ref()),
513 )))
514 }
515 });
516 }
517 GLProcessorMessage::DrawDecodedMasks(
518 mut dst,
519 det,
520 seg,
521 opacity,
522 bg,
523 letterbox,
524 color_mode,
525 resp,
526 ) => {
527 let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
531 let dst = unsafe { dst.ptr.as_mut() };
532 let det =
533 unsafe { std::slice::from_raw_parts(det.ptr.as_ptr(), det.len) };
534 let seg =
535 unsafe { std::slice::from_raw_parts(seg.ptr.as_ptr(), seg.len) };
536 let bg_ref = bg.map(|p| unsafe { &*p.ptr.as_ptr() });
537 gl_converter.draw_decoded_masks(
538 dst,
539 det,
540 seg,
541 crate::MaskOverlay {
542 background: bg_ref,
543 opacity,
544 letterbox,
545 color_mode,
546 },
547 )
548 }));
549 let _ = resp.send(match result {
550 Ok(res) => res,
551 Err(e) => {
552 poisoned = true;
553 Err(crate::Error::Internal(format!(
554 "GL thread panicked during DrawDecodedMasks: {}",
555 panic_message(e.as_ref()),
556 )))
557 }
558 });
559 }
560 GLProcessorMessage::DrawProtoMasks(
561 mut dst,
562 det,
563 proto_data,
564 opacity,
565 bg,
566 letterbox,
567 color_mode,
568 resp,
569 ) => {
570 let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
573 let dst = unsafe { dst.ptr.as_mut() };
574 let det =
575 unsafe { std::slice::from_raw_parts(det.ptr.as_ptr(), det.len) };
576 let bg_ref = bg.map(|p| unsafe { &*p.ptr.as_ptr() });
577 let proto_data = unsafe { proto_data.ptr.as_ref() };
578 gl_converter.draw_proto_masks(
579 dst,
580 det,
581 proto_data,
582 crate::MaskOverlay {
583 background: bg_ref,
584 opacity,
585 letterbox,
586 color_mode,
587 },
588 )
589 }));
590 let _ = resp.send(match result {
591 Ok(res) => res,
592 Err(e) => {
593 poisoned = true;
594 Err(crate::Error::Internal(format!(
595 "GL thread panicked during DrawProtoMasks: {}",
596 panic_message(e.as_ref()),
597 )))
598 }
599 });
600 }
601 GLProcessorMessage::SetColors(colors, resp) => {
602 let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
603 gl_converter.set_class_colors(&colors)
604 }));
605 let _ = resp.send(match result {
606 Ok(res) => res,
607 Err(e) => {
608 poisoned = true;
609 Err(crate::Error::Internal(format!(
610 "GL thread panicked during SetColors: {}",
611 panic_message(e.as_ref()),
612 )))
613 }
614 });
615 }
616 GLProcessorMessage::SetColorimetryMode(mode, resp) => {
617 let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
618 gl_converter.set_colorimetry_mode(mode);
619 Ok(())
620 }));
621 let _ = resp.send(match result {
622 Ok(res) => res,
623 Err(e) => {
624 poisoned = true;
625 Err(crate::Error::Internal(format!(
626 "GL thread panicked during SetColorimetryMode: {}",
627 panic_message(e.as_ref()),
628 )))
629 }
630 });
631 }
632 GLProcessorMessage::SetInt8Interpolation(mode, resp) => {
633 let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
634 gl_converter.set_int8_interpolation_mode(mode);
635 Ok(())
636 }));
637 let _ = resp.send(match result {
638 Ok(res) => res,
639 Err(e) => {
640 poisoned = true;
641 Err(crate::Error::Internal(format!(
642 "GL thread panicked during SetInt8Interpolation: {}",
643 panic_message(e.as_ref()),
644 )))
645 }
646 });
647 }
648 GLProcessorMessage::EglCacheStats(resp) => {
649 let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
650 Ok(gl_converter.egl_cache_stats())
651 }));
652 let _ = resp.send(match result {
653 Ok(res) => res,
654 Err(e) => {
655 poisoned = true;
656 Err(crate::Error::Internal(format!(
657 "GL thread panicked during EglCacheStats: {}",
658 panic_message(e.as_ref()),
659 )))
660 }
661 });
662 }
663 GLProcessorMessage::ConvertStats(resp) => {
664 let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
665 Ok(gl_converter.convert_stats())
666 }));
667 let _ = resp.send(match result {
668 Ok(res) => res,
669 Err(e) => {
670 poisoned = true;
671 Err(crate::Error::Internal(format!(
672 "GL thread panicked during ConvertStats: {}",
673 panic_message(e.as_ref()),
674 )))
675 }
676 });
677 }
678 GLProcessorMessage::PboCreate(size, resp) => {
679 let result = std::panic::catch_unwind(AssertUnwindSafe(|| unsafe {
680 let mut id: u32 = 0;
681 edgefirst_gl::gl::GenBuffers(1, &mut id);
682 edgefirst_gl::gl::BindBuffer(edgefirst_gl::gl::PIXEL_PACK_BUFFER, id);
683 edgefirst_gl::gl::BufferData(
684 edgefirst_gl::gl::PIXEL_PACK_BUFFER,
685 size as isize,
686 std::ptr::null(),
687 edgefirst_gl::gl::STREAM_COPY,
688 );
689 edgefirst_gl::gl::BindBuffer(edgefirst_gl::gl::PIXEL_PACK_BUFFER, 0);
690 match check_gl_error("PboCreate", 0) {
691 Ok(()) => Ok(id),
692 Err(e) => {
693 edgefirst_gl::gl::DeleteBuffers(1, &id);
694 Err(e)
695 }
696 }
697 }));
698 let _ = resp.send(match result {
699 Ok(res) => res,
700 Err(e) => {
701 poisoned = true;
702 Err(crate::Error::Internal(format!(
703 "GL thread panicked during PboCreate: {}",
704 panic_message(e.as_ref()),
705 )))
706 }
707 });
708 }
709 GLProcessorMessage::PboMap(buffer_id, size, resp) => {
710 let result = std::panic::catch_unwind(AssertUnwindSafe(|| unsafe {
711 edgefirst_gl::gl::BindBuffer(
712 edgefirst_gl::gl::PIXEL_PACK_BUFFER,
713 buffer_id,
714 );
715 let ptr = edgefirst_gl::gl::MapBufferRange(
716 edgefirst_gl::gl::PIXEL_PACK_BUFFER,
717 0,
718 size as isize,
719 edgefirst_gl::gl::MAP_READ_BIT | edgefirst_gl::gl::MAP_WRITE_BIT,
720 );
721 edgefirst_gl::gl::BindBuffer(edgefirst_gl::gl::PIXEL_PACK_BUFFER, 0);
722 if ptr.is_null() {
723 Err(crate::Error::OpenGl(
724 "glMapBufferRange returned null".to_string(),
725 ))
726 } else {
727 Ok(edgefirst_tensor::PboMapping {
728 ptr: ptr as *mut u8,
729 size,
730 })
731 }
732 }));
733 let _ = resp.send(match result {
734 Ok(res) => res,
735 Err(e) => {
736 poisoned = true;
737 Err(crate::Error::Internal(format!(
738 "GL thread panicked during PboMap: {}",
739 panic_message(e.as_ref()),
740 )))
741 }
742 });
743 }
744 GLProcessorMessage::PboUnmap(buffer_id, resp) => {
745 let result = std::panic::catch_unwind(AssertUnwindSafe(|| unsafe {
746 edgefirst_gl::gl::BindBuffer(
747 edgefirst_gl::gl::PIXEL_PACK_BUFFER,
748 buffer_id,
749 );
750 let ok =
751 edgefirst_gl::gl::UnmapBuffer(edgefirst_gl::gl::PIXEL_PACK_BUFFER);
752 edgefirst_gl::gl::BindBuffer(edgefirst_gl::gl::PIXEL_PACK_BUFFER, 0);
753 if ok == edgefirst_gl::gl::FALSE {
754 Err(Error::OpenGl(
755 "PBO data was corrupted during mapping".into(),
756 ))
757 } else {
758 check_gl_error("PboUnmap", 0)
759 }
760 }));
761 let _ = resp.send(match result {
762 Ok(res) => res,
763 Err(e) => {
764 poisoned = true;
765 Err(crate::Error::Internal(format!(
766 "GL thread panicked during PboUnmap: {}",
767 panic_message(e.as_ref()),
768 )))
769 }
770 });
771 }
772 GLProcessorMessage::PboDelete(buffer_id) => {
773 if let Err(e) = std::panic::catch_unwind(AssertUnwindSafe(|| unsafe {
774 edgefirst_gl::gl::DeleteBuffers(1, &buffer_id);
775 })) {
776 poisoned = true;
777 log::error!(
778 "GL thread panicked during PboDelete: {}",
779 panic_message(e.as_ref()),
780 );
781 }
782 }
783 GLProcessorMessage::CudaRegisterBuffer(buffer_id, resp) => {
784 let _ = resp.send(edgefirst_tensor::gl_register_buffer(buffer_id));
786 }
787 GLProcessorMessage::CudaMap(resource, resp) => {
788 gl_converter.flush_pending();
795 let _ = resp.send(edgefirst_tensor::gl_map_resource(resource));
796 }
797 GLProcessorMessage::CudaUnmap(resource) => {
798 edgefirst_tensor::gl_unmap_resource(resource);
799 }
800 GLProcessorMessage::CudaUnregister(resource) => {
801 edgefirst_tensor::gl_unregister_resource(resource);
802 }
803 }
804 gl_converter.end_gpu_pass_if_synced();
808 }
809 #[cfg(target_os = "linux")]
812 let _guard = super::context::GL_MUTEX
813 .lock()
814 .unwrap_or_else(|e| e.into_inner());
815 drop(gl_converter);
816 };
817
818 let handle = std::thread::spawn(func);
820
821 let caps = match create_ctx_recv.blocking_recv() {
822 Ok(Err(e)) => return Err(e),
823 Err(_) => {
824 return Err(Error::Internal(
825 "GL converter error messaging closed without update".to_string(),
826 ));
827 }
828 Ok(Ok(caps)) => caps,
829 };
830
831 Ok(Self {
832 handle: Some(handle),
833 sender: Some(send),
834 caps,
835 })
836 }
837}
838
839impl ImageProcessorTrait for GLProcessorThreaded {
840 fn convert(
841 &mut self,
842 src: &TensorDyn,
843 dst: &mut TensorDyn,
844 rotation: crate::Rotation,
845 flip: Flip,
846 crop: Crop,
847 ) -> crate::Result<()> {
848 let (err_send, err_recv) = tokio::sync::oneshot::channel();
849 self.sender
850 .as_ref()
851 .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
852 .blocking_send(GLProcessorMessage::ImageConvert(
853 SendablePtr {
854 ptr: NonNull::from(src),
855 len: 1,
856 },
857 SendablePtr {
858 ptr: NonNull::from(dst),
859 len: 1,
860 },
861 rotation,
862 flip,
863 crop,
864 false, err_send,
866 ))
867 .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
868 err_recv.blocking_recv().map_err(|_| {
869 Error::Internal("GL converter error messaging closed without update".to_string())
870 })?
871 }
872
873 fn convert_deferred(
874 &mut self,
875 src: &TensorDyn,
876 dst: &mut TensorDyn,
877 rotation: crate::Rotation,
878 flip: Flip,
879 crop: Crop,
880 ) -> crate::Result<()> {
881 let (err_send, err_recv) = tokio::sync::oneshot::channel();
882 self.sender
883 .as_ref()
884 .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
885 .blocking_send(GLProcessorMessage::ImageConvert(
886 SendablePtr {
887 ptr: NonNull::from(src),
888 len: 1,
889 },
890 SendablePtr {
891 ptr: NonNull::from(dst),
892 len: 1,
893 },
894 rotation,
895 flip,
896 crop,
897 true, err_send,
899 ))
900 .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
901 err_recv.blocking_recv().map_err(|_| {
902 Error::Internal("GL converter error messaging closed without update".to_string())
903 })?
904 }
905
906 fn flush(&mut self) -> crate::Result<()> {
907 let (err_send, err_recv) = tokio::sync::oneshot::channel();
908 self.sender
909 .as_ref()
910 .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
911 .blocking_send(GLProcessorMessage::Flush(err_send))
912 .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
913 err_recv.blocking_recv().map_err(|_| {
914 Error::Internal("GL converter error messaging closed without update".to_string())
915 })?
916 }
917
918 fn draw_decoded_masks(
919 &mut self,
920 dst: &mut TensorDyn,
921 detect: &[crate::DetectBox],
922 segmentation: &[crate::Segmentation],
923 overlay: crate::MaskOverlay<'_>,
924 ) -> crate::Result<()> {
925 let (err_send, err_recv) = tokio::sync::oneshot::channel();
926 self.sender
927 .as_ref()
928 .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
929 .blocking_send(GLProcessorMessage::DrawDecodedMasks(
930 SendablePtr {
931 ptr: NonNull::from(dst),
932 len: 1,
933 },
934 SendablePtr {
935 ptr: NonNull::new(detect.as_ptr() as *mut DetectBox).unwrap(),
936 len: detect.len(),
937 },
938 SendablePtr {
939 ptr: NonNull::new(segmentation.as_ptr() as *mut Segmentation).unwrap(),
940 len: segmentation.len(),
941 },
942 overlay.opacity,
943 overlay.background.map(|bg| SendablePtr {
944 ptr: NonNull::from(bg).cast::<TensorDyn>(),
945 len: 1,
946 }),
947 overlay.letterbox,
948 overlay.color_mode,
949 err_send,
950 ))
951 .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
952 err_recv.blocking_recv().map_err(|_| {
953 Error::Internal("GL converter error messaging closed without update".to_string())
954 })?
955 }
956
957 fn draw_proto_masks(
958 &mut self,
959 dst: &mut TensorDyn,
960 detect: &[DetectBox],
961 proto_data: &ProtoData,
962 overlay: crate::MaskOverlay<'_>,
963 ) -> crate::Result<()> {
964 let (err_send, err_recv) = tokio::sync::oneshot::channel();
965 self.sender
966 .as_ref()
967 .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
968 .blocking_send(GLProcessorMessage::DrawProtoMasks(
969 SendablePtr {
970 ptr: NonNull::from(dst),
971 len: 1,
972 },
973 SendablePtr {
974 ptr: NonNull::new(detect.as_ptr() as *mut DetectBox).unwrap(),
975 len: detect.len(),
976 },
977 SendablePtr {
978 ptr: NonNull::from(proto_data).cast::<ProtoData>(),
979 len: 1,
980 },
981 overlay.opacity,
982 overlay.background.map(|bg| SendablePtr {
983 ptr: NonNull::from(bg).cast::<TensorDyn>(),
984 len: 1,
985 }),
986 overlay.letterbox,
987 overlay.color_mode,
988 err_send,
989 ))
990 .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
991 err_recv.blocking_recv().map_err(|_| {
992 Error::Internal("GL converter error messaging closed without update".to_string())
993 })?
994 }
995
996 fn set_class_colors(&mut self, colors: &[[u8; 4]]) -> Result<(), crate::Error> {
997 let (err_send, err_recv) = tokio::sync::oneshot::channel();
998 self.sender
999 .as_ref()
1000 .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1001 .blocking_send(GLProcessorMessage::SetColors(colors.to_vec(), err_send))
1002 .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1003 err_recv.blocking_recv().map_err(|_| {
1004 Error::Internal("GL converter error messaging closed without update".to_string())
1005 })?
1006 }
1007}
1008
1009impl GLProcessorThreaded {
1010 pub fn set_colorimetry_mode(&mut self, mode: crate::ColorimetryMode) -> Result<(), Error> {
1015 let (err_send, err_recv) = tokio::sync::oneshot::channel();
1016 self.sender
1017 .as_ref()
1018 .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1019 .blocking_send(GLProcessorMessage::SetColorimetryMode(mode, err_send))
1020 .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1021 err_recv.blocking_recv().map_err(|_| {
1022 Error::Internal("GL converter error messaging closed without update".to_string())
1023 })?
1024 }
1025
1026 pub fn set_int8_interpolation_mode(
1028 &mut self,
1029 mode: Int8InterpolationMode,
1030 ) -> Result<(), crate::Error> {
1031 let (err_send, err_recv) = tokio::sync::oneshot::channel();
1032 self.sender
1033 .as_ref()
1034 .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1035 .blocking_send(GLProcessorMessage::SetInt8Interpolation(mode, err_send))
1036 .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1037 err_recv.blocking_recv().map_err(|_| {
1038 Error::Internal("GL converter error messaging closed without update".to_string())
1039 })?
1040 }
1041
1042 pub fn egl_cache_stats(&self) -> Result<super::cache::GlCacheStats, Error> {
1049 let (send, recv) = tokio::sync::oneshot::channel();
1050 self.sender
1051 .as_ref()
1052 .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1053 .blocking_send(GLProcessorMessage::EglCacheStats(send))
1054 .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1055 recv.blocking_recv().map_err(|_| {
1056 Error::Internal("GL converter error messaging closed without update".to_string())
1057 })?
1058 }
1059
1060 pub fn convert_with_fence(
1071 &mut self,
1072 src: &TensorDyn,
1073 dst: &mut TensorDyn,
1074 rotation: crate::Rotation,
1075 flip: Flip,
1076 crop: Crop,
1077 ) -> Result<Option<std::os::fd::OwnedFd>, Error> {
1078 use crate::ImageProcessorTrait as _;
1079 if !self.caps.native_fence_sync {
1080 self.convert(src, dst, rotation, flip, crop)?;
1083 return Ok(None);
1084 }
1085 let (send, recv) = tokio::sync::oneshot::channel();
1086 self.sender
1087 .as_ref()
1088 .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1089 .blocking_send(GLProcessorMessage::ImageConvertFenced(
1090 SendablePtr {
1091 ptr: NonNull::from(src),
1092 len: 1,
1093 },
1094 SendablePtr {
1095 ptr: NonNull::from(dst),
1096 len: 1,
1097 },
1098 rotation,
1099 flip,
1100 crop,
1101 send,
1102 ))
1103 .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1104 recv.blocking_recv().map_err(|_| {
1105 Error::Internal("GL converter error messaging closed without update".to_string())
1106 })?
1107 }
1108
1109 pub fn convert_stats(&self) -> Result<super::cache::ConvertStats, Error> {
1114 let (send, recv) = tokio::sync::oneshot::channel();
1115 self.sender
1116 .as_ref()
1117 .ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
1118 .blocking_send(GLProcessorMessage::ConvertStats(send))
1119 .map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
1120 recv.blocking_recv().map_err(|_| {
1121 Error::Internal("GL converter error messaging closed without update".to_string())
1122 })?
1123 }
1124
1125 pub fn create_pbo_image(
1127 &self,
1128 width: usize,
1129 height: usize,
1130 format: edgefirst_tensor::PixelFormat,
1131 ) -> Result<edgefirst_tensor::Tensor<u8>, Error> {
1132 let sender = self
1133 .sender
1134 .as_ref()
1135 .ok_or(Error::OpenGl("GL processor is shutting down".to_string()))?;
1136
1137 let size = pbo_elem_count(width, height, format)
1138 .filter(|&s| s != 0)
1139 .ok_or_else(|| Error::OpenGl("Invalid image dimensions".to_string()))?;
1140
1141 let (tx, rx) = tokio::sync::oneshot::channel();
1143 sender
1144 .blocking_send(GLProcessorMessage::PboCreate(size, tx))
1145 .map_err(|_| Error::OpenGl("GL thread channel closed".to_string()))?;
1146 let buffer_id = rx
1147 .blocking_recv()
1148 .map_err(|_| Error::OpenGl("GL thread did not respond".to_string()))??;
1149
1150 let ops: std::sync::Arc<dyn edgefirst_tensor::PboOps> = std::sync::Arc::new(GlPboOps {
1151 sender: sender.downgrade(),
1152 });
1153
1154 let shape = pbo_shape(width, height, format);
1155
1156 let pbo_tensor =
1157 edgefirst_tensor::PboTensor::<u8>::from_pbo(buffer_id, size, &shape, None, ops)
1158 .map_err(|e| Error::OpenGl(format!("PBO tensor creation failed: {e:?}")))?;
1159 let mut tensor = edgefirst_tensor::Tensor::from_pbo(pbo_tensor);
1160 tensor
1161 .set_format(format)
1162 .map_err(|e| Error::OpenGl(format!("Failed to set format on PBO tensor: {e:?}")))?;
1163 register_pbo_cuda(&mut tensor, buffer_id, size, sender);
1170 Ok(tensor)
1171 }
1172
1173 pub(crate) fn create_pbo_image_dtype(
1179 &self,
1180 width: usize,
1181 height: usize,
1182 format: edgefirst_tensor::PixelFormat,
1183 dtype: edgefirst_tensor::DType,
1184 ) -> Result<TensorDyn, Error> {
1185 let sender = self
1186 .sender
1187 .as_ref()
1188 .ok_or(Error::OpenGl("GL processor is shutting down".to_string()))?;
1189
1190 let elems = pbo_elem_count(width, height, format)
1191 .filter(|&e| e != 0)
1192 .ok_or_else(|| Error::OpenGl("Invalid image dimensions".to_string()))?;
1193
1194 let size = elems
1195 .checked_mul(dtype.size())
1196 .ok_or_else(|| Error::OpenGl("PBO size overflow".to_string()))?;
1197
1198 let (tx, rx) = tokio::sync::oneshot::channel();
1200 sender
1201 .blocking_send(GLProcessorMessage::PboCreate(size, tx))
1202 .map_err(|_| Error::OpenGl("GL thread channel closed".to_string()))?;
1203 let buffer_id = rx
1204 .blocking_recv()
1205 .map_err(|_| Error::OpenGl("GL thread did not respond".to_string()))??;
1206
1207 let ops: std::sync::Arc<dyn edgefirst_tensor::PboOps> = std::sync::Arc::new(GlPboOps {
1208 sender: sender.downgrade(),
1209 });
1210
1211 let shape = pbo_shape(width, height, format);
1212
1213 let map_err = |e: edgefirst_tensor::Error| {
1214 Error::OpenGl(format!("PBO tensor creation failed: {e:?}"))
1215 };
1216 let set_err = |e: edgefirst_tensor::Error| {
1217 Error::OpenGl(format!("Failed to set format on PBO tensor: {e:?}"))
1218 };
1219
1220 match dtype {
1221 edgefirst_tensor::DType::U8 => {
1222 let pbo =
1223 edgefirst_tensor::PboTensor::<u8>::from_pbo(buffer_id, size, &shape, None, ops)
1224 .map_err(map_err)?;
1225 let mut t = edgefirst_tensor::Tensor::from_pbo(pbo);
1226 t.set_format(format).map_err(set_err)?;
1227 register_pbo_cuda(&mut t, buffer_id, size, sender);
1228 Ok(TensorDyn::from(t))
1229 }
1230 edgefirst_tensor::DType::F16 => {
1231 let pbo = edgefirst_tensor::PboTensor::<edgefirst_tensor::f16>::from_pbo(
1232 buffer_id, size, &shape, None, ops,
1233 )
1234 .map_err(map_err)?;
1235 let mut t = edgefirst_tensor::Tensor::from_pbo(pbo);
1236 t.set_format(format).map_err(set_err)?;
1237 register_pbo_cuda(&mut t, buffer_id, size, sender);
1238 Ok(TensorDyn::from(t))
1239 }
1240 edgefirst_tensor::DType::F32 => {
1241 let pbo = edgefirst_tensor::PboTensor::<f32>::from_pbo(
1242 buffer_id, size, &shape, None, ops,
1243 )
1244 .map_err(map_err)?;
1245 let mut t = edgefirst_tensor::Tensor::from_pbo(pbo);
1246 t.set_format(format).map_err(set_err)?;
1247 register_pbo_cuda(&mut t, buffer_id, size, sender);
1248 Ok(TensorDyn::from(t))
1249 }
1250 other => Err(Error::OpenGl(format!("unsupported PBO dtype {other:?}"))),
1251 }
1252 }
1253
1254 pub(crate) fn transfer_backend(&self) -> TransferBackend {
1256 self.caps.transfer_backend
1257 }
1258
1259 pub(crate) fn supported_render_dtypes(&self) -> crate::RenderDtypeSupport {
1266 self.caps.render_dtypes
1267 }
1268}
1269
1270impl Drop for GLProcessorThreaded {
1271 fn drop(&mut self) {
1272 drop(self.sender.take());
1273 let _ = self.handle.take().and_then(|h| h.join().ok());
1274 }
1275}
1276
1277#[cfg(test)]
1281#[cfg_attr(coverage_nightly, coverage(off))]
1282mod tests {
1283 use super::{pbo_elem_count, pbo_shape};
1284 use edgefirst_tensor::PixelFormat;
1285
1286 #[test]
1287 fn elem_count_per_format() {
1288 assert_eq!(pbo_elem_count(8, 4, PixelFormat::Rgba), Some(8 * 4 * 4));
1290 assert_eq!(pbo_elem_count(8, 4, PixelFormat::Rgb), Some(8 * 4 * 3));
1292 assert_eq!(pbo_elem_count(8, 4, PixelFormat::Nv12), Some(8 * 4 * 3 / 2));
1294 assert_eq!(pbo_elem_count(8, 4, PixelFormat::Nv16), Some(8 * 4 * 2));
1296 }
1297
1298 #[test]
1299 fn elem_count_overflow_is_none() {
1300 assert_eq!(pbo_elem_count(usize::MAX, 2, PixelFormat::Rgba), None);
1302 assert_eq!(pbo_elem_count(usize::MAX, 1, PixelFormat::Rgb), None);
1304 }
1305
1306 #[test]
1307 fn shape_per_format() {
1308 assert_eq!(pbo_shape(8, 4, PixelFormat::PlanarRgb), vec![3, 4, 8]);
1310 assert_eq!(pbo_shape(8, 4, PixelFormat::Nv12), vec![4 * 3 / 2, 8]);
1312 assert_eq!(pbo_shape(8, 4, PixelFormat::Nv16), vec![4 * 2, 8]);
1314 assert_eq!(pbo_shape(8, 4, PixelFormat::Rgba), vec![4, 8, 4]);
1316 }
1317}