use edgefirst_tensor::{DetectBox, ProtoData, Segmentation};
use std::panic::AssertUnwindSafe;
use std::ptr::NonNull;
use std::thread::JoinHandle;
use tokio::sync::mpsc::{Sender, WeakSender};
use super::processor::GLProcessorST;
use super::shaders::check_gl_error;
use super::{EglDisplayKind, Int8InterpolationMode, TransferBackend};
use crate::{ConvertOptions, Error, ImageProcessorTrait};
use edgefirst_tensor::TensorDyn;
#[allow(clippy::type_complexity)]
enum GLProcessorMessage {
ImageConvert(
SendablePtr<TensorDyn>,
SendablePtr<TensorDyn>,
ConvertOptions,
bool, tokio::sync::oneshot::Sender<Result<(), Error>>,
),
ImageConvertFenced(
SendablePtr<TensorDyn>,
SendablePtr<TensorDyn>,
ConvertOptions,
tokio::sync::oneshot::Sender<Result<Option<super::CompletionFence>, Error>>,
),
Flush(tokio::sync::oneshot::Sender<Result<(), Error>>),
SetColors(
Vec<[u8; 4]>,
tokio::sync::oneshot::Sender<Result<(), Error>>,
),
SetBoxLineWidth(u32, tokio::sync::oneshot::Sender<Result<(), Error>>),
DrawDecodedMasks(
SendablePtr<TensorDyn>,
SendablePtr<DetectBox>,
SendablePtr<Segmentation>,
f32, Option<SendablePtr<TensorDyn>>, Option<[f32; 4]>, crate::ColorMode,
tokio::sync::oneshot::Sender<Result<(), Error>>,
),
DrawProtoMasks(
SendablePtr<TensorDyn>,
SendablePtr<DetectBox>,
SendablePtr<ProtoData>,
f32, Option<SendablePtr<TensorDyn>>, Option<[f32; 4]>, crate::ColorMode,
tokio::sync::oneshot::Sender<Result<(), Error>>,
),
SetInt8Interpolation(
Int8InterpolationMode,
tokio::sync::oneshot::Sender<Result<(), Error>>,
),
SetColorimetryMode(
crate::ColorimetryMode,
tokio::sync::oneshot::Sender<Result<(), Error>>,
),
SetInterpolation(
crate::Interpolation,
tokio::sync::oneshot::Sender<Result<(), Error>>,
),
#[cfg(test)]
SetFailDstImport(bool, tokio::sync::oneshot::Sender<Result<(), Error>>),
EglCacheStats(tokio::sync::oneshot::Sender<Result<super::cache::GlCacheStats, Error>>),
ConvertStats(tokio::sync::oneshot::Sender<Result<super::cache::ConvertStats, Error>>),
PboCreate(
usize, tokio::sync::oneshot::Sender<Result<u32, Error>>,
),
PboMap(
u32, usize, tokio::sync::oneshot::Sender<Result<edgefirst_tensor::PboMapping, Error>>,
),
PboUnmap(
u32, tokio::sync::oneshot::Sender<Result<(), Error>>,
),
PboDelete(u32), CudaRegisterBuffer(u32, tokio::sync::oneshot::Sender<Option<usize>>),
CudaMap(usize, tokio::sync::oneshot::Sender<Option<(usize, usize)>>),
CudaUnmap(usize), CudaUnregister(usize), }
fn pbo_elem_count(
width: usize,
height: usize,
format: edgefirst_tensor::PixelFormat,
) -> Option<usize> {
format
.allocation_shape(width, height)?
.iter()
.try_fold(1usize, |acc, &d| acc.checked_mul(d))
}
fn pbo_shape(width: usize, height: usize, format: edgefirst_tensor::PixelFormat) -> Vec<usize> {
format
.allocation_shape(width, height)
.unwrap_or_else(|| vec![height, width, format.channels()])
}
fn register_pbo_cuda<T>(
tensor: &mut edgefirst_tensor::Tensor<T>,
buffer_id: u32,
size: usize,
sender: &Sender<GLProcessorMessage>,
) where
T: num_traits::Num + Clone + std::fmt::Debug + Send + Sync + edgefirst_tensor::Element,
{
if !edgefirst_tensor::is_cuda_available() {
return;
}
let (tx, rx) = tokio::sync::oneshot::channel();
if sender
.blocking_send(GLProcessorMessage::CudaRegisterBuffer(buffer_id, tx))
.is_ok()
{
if let Ok(Some(resource)) = rx.blocking_recv() {
let ops = std::sync::Arc::new(GlCudaOps {
sender: sender.downgrade(),
});
tensor.set_cuda_handle(edgefirst_tensor::CudaHandle::new_gl(
resource as *mut std::ffi::c_void,
size,
ops,
));
}
}
}
struct GlPboOps {
sender: WeakSender<GLProcessorMessage>,
worker_id: u64,
}
static NEXT_GL_WORKER_ID: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
struct GlWorkerState {
id: u64,
serializes: bool,
mapped: std::collections::HashSet<u32>,
}
thread_local! {
static GL_WORKER: std::cell::RefCell<Option<GlWorkerState>> =
const { std::cell::RefCell::new(None) };
}
fn with_worker_mapped<R>(f: impl FnOnce(&mut std::collections::HashSet<u32>) -> R) -> Option<R> {
GL_WORKER.with(|c| c.borrow_mut().as_mut().map(|s| f(&mut s.mapped)))
}
fn on_owning_gl_worker(worker_id: u64) -> bool {
GL_WORKER.with(|c| matches!(*c.borrow(), Some(ref s) if s.id == worker_id))
}
fn on_serializing_gl_worker() -> bool {
GL_WORKER.with(|c| matches!(*c.borrow(), Some(ref s) if s.serializes))
}
fn cross_context_pbo_error(op: &str, buffer_id: u32) -> edgefirst_tensor::Error {
edgefirst_tensor::Error::InvalidOperation(format!(
"PBO {buffer_id}: cannot {op} a buffer owned by another GL context while \
this one holds the serialization lock; convert it through the processor \
that created it"
))
}
thread_local! {
static GL_WORKER_POISON: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
}
fn inline_gl_op<T>(what: &str, f: impl FnOnce() -> Result<T, Error>) -> Result<T, Error> {
match std::panic::catch_unwind(AssertUnwindSafe(f)) {
Ok(result) => result,
Err(e) => {
GL_WORKER_POISON.with(|p| p.set(true));
Err(Error::Internal(format!(
"GL thread panicked during {what}: {}",
panic_message(e.as_ref()),
)))
}
}
}
unsafe impl edgefirst_tensor::PboOps for GlPboOps {
fn map_buffer(
&self,
buffer_id: u32,
size: usize,
) -> edgefirst_tensor::Result<edgefirst_tensor::PboMapping> {
if on_owning_gl_worker(self.worker_id) {
return inline_gl_op("PboMap", || pbo_map_on_gl_thread(buffer_id, size)).map_err(|e| {
edgefirst_tensor::Error::NotImplemented(format!("GL PBO map failed: {e:?}"))
});
}
if on_serializing_gl_worker() {
return Err(cross_context_pbo_error("map", buffer_id));
}
let sender = self
.sender
.upgrade()
.ok_or(edgefirst_tensor::Error::PboDisconnected)?;
let (tx, rx) = tokio::sync::oneshot::channel();
sender
.blocking_send(GLProcessorMessage::PboMap(buffer_id, size, tx))
.map_err(|_| edgefirst_tensor::Error::PboDisconnected)?;
rx.blocking_recv()
.map_err(|_| edgefirst_tensor::Error::PboDisconnected)?
.map_err(|e| {
edgefirst_tensor::Error::NotImplemented(format!("GL PBO map failed: {e:?}"))
})
}
fn unmap_buffer(&self, buffer_id: u32) -> edgefirst_tensor::Result<()> {
if on_owning_gl_worker(self.worker_id) {
return inline_gl_op("PboUnmap", || pbo_unmap_on_gl_thread(buffer_id)).map_err(|e| {
edgefirst_tensor::Error::NotImplemented(format!("GL PBO unmap failed: {e:?}"))
});
}
if on_serializing_gl_worker() {
return Err(cross_context_pbo_error("unmap", buffer_id));
}
let sender = self
.sender
.upgrade()
.ok_or(edgefirst_tensor::Error::PboDisconnected)?;
let (tx, rx) = tokio::sync::oneshot::channel();
sender
.blocking_send(GLProcessorMessage::PboUnmap(buffer_id, tx))
.map_err(|_| edgefirst_tensor::Error::PboDisconnected)?;
rx.blocking_recv()
.map_err(|_| edgefirst_tensor::Error::PboDisconnected)?
.map_err(|e| {
edgefirst_tensor::Error::NotImplemented(format!("GL PBO unmap failed: {e:?}"))
})
}
fn delete_buffer(&self, buffer_id: u32) {
if on_owning_gl_worker(self.worker_id) {
with_worker_mapped(|set| set.remove(&buffer_id));
let _ = inline_gl_op("PboDelete", || {
unsafe { edgefirst_gl::gl::DeleteBuffers(1, &buffer_id) };
Ok(())
});
return;
}
if let Some(sender) = self.sender.upgrade() {
if on_serializing_gl_worker() {
if sender
.try_send(GLProcessorMessage::PboDelete(buffer_id))
.is_err()
{
log::warn!(
"PBO {buffer_id}: cross-context delete skipped to avoid a deadlock; \
the buffer is freed when its context is destroyed"
);
}
return;
}
let _ = sender.blocking_send(GLProcessorMessage::PboDelete(buffer_id));
}
}
}
struct GlCudaOps {
sender: WeakSender<GLProcessorMessage>,
}
impl edgefirst_tensor::CudaGlOps for GlCudaOps {
fn map(&self, resource: *mut std::ffi::c_void) -> Option<(*mut std::ffi::c_void, usize)> {
let sender = self.sender.upgrade()?;
let (tx, rx) = tokio::sync::oneshot::channel();
sender
.blocking_send(GLProcessorMessage::CudaMap(resource as usize, tx))
.ok()?;
rx.blocking_recv()
.ok()?
.map(|(p, l)| (p as *mut std::ffi::c_void, l))
}
fn unmap(&self, resource: *mut std::ffi::c_void) {
if let Some(sender) = self.sender.upgrade() {
let _ = sender.blocking_send(GLProcessorMessage::CudaUnmap(resource as usize));
}
}
fn unregister(&self, resource: *mut std::ffi::c_void) {
if let Some(sender) = self.sender.upgrade() {
let _ = sender.blocking_send(GLProcessorMessage::CudaUnregister(resource as usize));
}
}
}
#[derive(Debug)]
pub struct GLProcessorThreaded {
handle: Option<JoinHandle<()>>,
sender: Option<Sender<GLProcessorMessage>>,
worker_id: u64,
owned_pbos: std::sync::Arc<super::processor::OwnedPbos>,
caps: super::platform::PlatformCaps,
}
unsafe impl Send for GLProcessorThreaded {}
unsafe impl Sync for GLProcessorThreaded {}
struct SendablePtr<T: Send> {
ptr: NonNull<T>,
len: usize,
}
unsafe impl<T> Send for SendablePtr<T> where T: Send {}
fn panic_message(info: &(dyn std::any::Any + Send)) -> String {
if let Some(s) = info.downcast_ref::<&str>() {
s.to_string()
} else if let Some(s) = info.downcast_ref::<String>() {
s.clone()
} else {
"unknown panic".to_string()
}
}
fn reply_caught<T>(
result: std::thread::Result<crate::Result<T>>,
resp: tokio::sync::oneshot::Sender<crate::Result<T>>,
poisoned: &mut bool,
what: &str,
) {
let _ = resp.send(match result {
Ok(res) => res,
Err(e) => {
*poisoned = true;
Err(crate::Error::Internal(format!(
"GL thread panicked during {what}: {}",
panic_message(e.as_ref()),
)))
}
});
}
fn reject_poisoned_message(msg: GLProcessorMessage) {
let poison_err =
crate::Error::Internal("GL context is poisoned after a prior panic".to_string());
match msg {
GLProcessorMessage::ImageConvertFenced(.., resp) => {
let _ = resp.send(Err(poison_err));
}
GLProcessorMessage::ImageConvert(.., resp) => {
let _ = resp.send(Err(poison_err));
}
GLProcessorMessage::Flush(resp) => {
let _ = resp.send(Err(poison_err));
}
GLProcessorMessage::DrawDecodedMasks(.., resp) => {
let _ = resp.send(Err(poison_err));
}
GLProcessorMessage::DrawProtoMasks(.., resp) => {
let _ = resp.send(Err(poison_err));
}
GLProcessorMessage::SetColors(_, resp) => {
let _ = resp.send(Err(poison_err));
}
GLProcessorMessage::SetBoxLineWidth(_, resp) => {
let _ = resp.send(Err(poison_err));
}
GLProcessorMessage::SetInt8Interpolation(_, resp) => {
let _ = resp.send(Err(poison_err));
}
GLProcessorMessage::SetColorimetryMode(_, resp) => {
let _ = resp.send(Err(poison_err));
}
GLProcessorMessage::SetInterpolation(_, resp) => {
let _ = resp.send(Err(poison_err));
}
#[cfg(test)]
GLProcessorMessage::SetFailDstImport(_, resp) => {
let _ = resp.send(Err(poison_err));
}
GLProcessorMessage::EglCacheStats(resp) => {
let _ = resp.send(Err(poison_err));
}
GLProcessorMessage::ConvertStats(resp) => {
let _ = resp.send(Err(poison_err));
}
GLProcessorMessage::PboCreate(_, resp) => {
let _ = resp.send(Err(poison_err));
}
GLProcessorMessage::PboMap(_, _, resp) => {
let _ = resp.send(Err(poison_err));
}
GLProcessorMessage::PboUnmap(_, resp) => {
let _ = resp.send(Err(poison_err));
}
GLProcessorMessage::PboDelete(_) => {}
GLProcessorMessage::CudaRegisterBuffer(_, resp) => {
let _ = resp.send(None);
}
GLProcessorMessage::CudaMap(_, resp) => {
let _ = resp.send(None);
}
GLProcessorMessage::CudaUnmap(_) => {}
GLProcessorMessage::CudaUnregister(_) => {}
}
}
fn run_gl_worker(
worker_id: u64,
owned_pbos: std::sync::Arc<super::processor::OwnedPbos>,
kind: Option<EglDisplayKind>,
capacity: Option<usize>,
mut recv: tokio::sync::mpsc::Receiver<GLProcessorMessage>,
create_ctx_send: tokio::sync::oneshot::Sender<Result<super::platform::PlatformCaps, Error>>,
) {
#[cfg(target_os = "linux")]
let init_result = {
let _guard = super::context::GL_MUTEX
.lock()
.unwrap_or_else(|e| e.into_inner());
GLProcessorST::new(kind, capacity)
};
#[cfg(any(target_os = "macos", target_os = "ios", target_os = "android"))]
let init_result = GLProcessorST::new(kind, capacity);
#[cfg(target_os = "windows")]
let init_result = {
let _guard = NON_LINUX_GL_MUTEX.lock().unwrap_or_else(|e| e.into_inner());
GLProcessorST::new(kind, capacity)
};
let mut gl_converter = match init_result {
Ok(gl) => gl,
Err(e) => {
let _ = create_ctx_send.send(Err(e));
return;
}
};
gl_converter.owned_pbos = owned_pbos;
let caps = gl_converter.platform_caps();
let _ = create_ctx_send.send(Ok(caps));
let mut poisoned = false;
GL_WORKER.with(|c| {
*c.borrow_mut() = Some(GlWorkerState {
id: worker_id,
serializes: false,
mapped: std::collections::HashSet::new(),
});
});
struct GlWorkerSlot;
impl Drop for GlWorkerSlot {
fn drop(&mut self) {
GL_WORKER.with(|c| *c.borrow_mut() = None);
GL_WORKER_POISON.with(|p| p.set(false));
}
}
let _gl_worker_slot = GlWorkerSlot;
let serialize_per_msg = match std::env::var("EDGEFIRST_GL_SERIALIZE").as_deref() {
Ok("full") => true,
Ok("lifecycle") => {
#[cfg(target_os = "windows")]
log::warn!(
"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"
);
false
}
_ => caps.serialize_gl,
};
log::debug!(
"GL serialization policy: {}",
if serialize_per_msg {
"Full (per-message global lock)"
} else {
"LifecycleOnly (parallel across processors)"
}
);
GL_WORKER.with(|c| {
if let Some(state) = c.borrow_mut().as_mut() {
state.serializes = serialize_per_msg;
}
});
while let Some(msg) = recv.blocking_recv() {
#[cfg(target_os = "linux")]
let _guard = serialize_per_msg.then(|| {
super::context::GL_MUTEX
.lock()
.unwrap_or_else(|e| e.into_inner())
});
#[cfg(any(
target_os = "macos",
target_os = "ios",
target_os = "android",
target_os = "windows"
))]
let _guard =
serialize_per_msg.then(|| NON_LINUX_GL_MUTEX.lock().unwrap_or_else(|e| e.into_inner()));
if poisoned {
reject_poisoned_message(msg);
continue;
}
gl_converter.begin_gpu_pass();
handle_gl_message(msg, &mut gl_converter, &mut poisoned);
poisoned |= GL_WORKER_POISON.with(|p| p.replace(false));
gl_converter.end_gpu_pass_if_synced();
}
#[cfg(target_os = "linux")]
let _guard = super::context::GL_MUTEX
.lock()
.unwrap_or_else(|e| e.into_inner());
#[cfg(target_os = "windows")]
let _guard = NON_LINUX_GL_MUTEX.lock().unwrap_or_else(|e| e.into_inner());
drop(gl_converter);
}
#[cfg(not(target_os = "linux"))]
static NON_LINUX_GL_MUTEX: std::sync::Mutex<()> = std::sync::Mutex::new(());
#[cfg(target_os = "windows")]
pub(super) fn lifecycle_guard() -> std::sync::MutexGuard<'static, ()> {
NON_LINUX_GL_MUTEX.lock().unwrap_or_else(|e| e.into_inner())
}
fn handle_gl_message(
msg: GLProcessorMessage,
gl_converter: &mut GLProcessorST,
poisoned: &mut bool,
) {
match msg {
GLProcessorMessage::ImageConvert(src, dst, options, defer, resp) => {
handle_image_convert(src, dst, (options, defer), resp, gl_converter, poisoned);
}
GLProcessorMessage::ImageConvertFenced(src, dst, options, resp) => {
handle_image_convert_fenced(src, dst, options, resp, gl_converter, poisoned);
}
GLProcessorMessage::Flush(resp) => {
let result = std::panic::catch_unwind(AssertUnwindSafe(|| gl_converter.flush()));
reply_caught(result, resp, poisoned, "Flush");
}
GLProcessorMessage::DrawDecodedMasks(
dst,
det,
seg,
opacity,
bg,
content_region,
color_mode,
resp,
) => {
handle_draw_decoded_masks(
dst,
det,
seg,
(opacity, bg, content_region, color_mode),
resp,
gl_converter,
poisoned,
);
}
GLProcessorMessage::DrawProtoMasks(
dst,
det,
proto_data,
opacity,
bg,
content_region,
color_mode,
resp,
) => {
handle_draw_proto_masks(
dst,
det,
proto_data,
(opacity, bg, content_region, color_mode),
resp,
gl_converter,
poisoned,
);
}
GLProcessorMessage::SetColors(colors, resp) => {
let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
gl_converter.set_class_colors(&colors)
}));
reply_caught(result, resp, poisoned, "SetColors");
}
GLProcessorMessage::SetBoxLineWidth(width, resp) => {
let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
gl_converter.set_box_line_width(width)
}));
reply_caught(result, resp, poisoned, "SetBoxLineWidth");
}
GLProcessorMessage::SetColorimetryMode(mode, resp) => {
let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
gl_converter.set_colorimetry_mode(mode);
Ok(())
}));
reply_caught(result, resp, poisoned, "SetColorimetryMode");
}
GLProcessorMessage::SetInterpolation(mode, resp) => {
let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
gl_converter.set_interpolation(mode);
Ok(())
}));
reply_caught(result, resp, poisoned, "SetInterpolation");
}
GLProcessorMessage::SetInt8Interpolation(mode, resp) => {
let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
gl_converter.set_int8_interpolation_mode(mode);
Ok(())
}));
reply_caught(result, resp, poisoned, "SetInt8Interpolation");
}
#[cfg(test)]
GLProcessorMessage::SetFailDstImport(fail, resp) => {
let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
gl_converter.fail_dst_import = fail;
Ok(())
}));
reply_caught(result, resp, poisoned, "SetFailDstImport");
}
GLProcessorMessage::EglCacheStats(resp) => {
let result =
std::panic::catch_unwind(AssertUnwindSafe(|| Ok(gl_converter.egl_cache_stats())));
reply_caught(result, resp, poisoned, "EglCacheStats");
}
GLProcessorMessage::ConvertStats(resp) => {
let result =
std::panic::catch_unwind(AssertUnwindSafe(|| Ok(gl_converter.convert_stats())));
reply_caught(result, resp, poisoned, "ConvertStats");
}
GLProcessorMessage::PboCreate(size, resp) => {
handle_pbo_create(size, resp, poisoned);
}
GLProcessorMessage::PboMap(buffer_id, size, resp) => {
handle_pbo_map(buffer_id, size, resp, poisoned);
}
GLProcessorMessage::PboUnmap(buffer_id, resp) => {
handle_pbo_unmap(buffer_id, resp, poisoned);
}
GLProcessorMessage::PboDelete(buffer_id) => {
handle_pbo_delete(buffer_id, poisoned);
}
GLProcessorMessage::CudaRegisterBuffer(buffer_id, resp) => {
let _ = resp.send(edgefirst_tensor::gl_register_buffer(buffer_id));
}
GLProcessorMessage::CudaMap(resource, resp) => {
gl_converter.flush_pending();
let _ = resp.send(edgefirst_tensor::gl_map_resource(resource));
}
GLProcessorMessage::CudaUnmap(resource) => {
edgefirst_tensor::gl_unmap_resource(resource);
}
GLProcessorMessage::CudaUnregister(resource) => {
edgefirst_tensor::gl_unregister_resource(resource);
}
}
}
fn handle_image_convert(
src: SendablePtr<TensorDyn>,
mut dst: SendablePtr<TensorDyn>,
geom: (ConvertOptions, bool),
resp: tokio::sync::oneshot::Sender<Result<(), Error>>,
gl_converter: &mut GLProcessorST,
poisoned: &mut bool,
) {
let (options, defer) = geom;
let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
let src = unsafe { src.ptr.as_ref() };
let dst = unsafe { dst.ptr.as_mut() };
if defer {
gl_converter.convert_deferred(src, dst, Some(options))
} else {
gl_converter.convert(src, dst, Some(options))
}
}));
reply_caught(result, resp, poisoned, "ImageConvert");
}
fn handle_image_convert_fenced(
src: SendablePtr<TensorDyn>,
mut dst: SendablePtr<TensorDyn>,
options: ConvertOptions,
resp: tokio::sync::oneshot::Sender<Result<Option<super::CompletionFence>, Error>>,
gl_converter: &mut GLProcessorST,
poisoned: &mut bool,
) {
let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
let src = unsafe { src.ptr.as_ref() };
let dst = unsafe { dst.ptr.as_mut() };
gl_converter.convert_fenced(src, dst, options)
}));
reply_caught(result, resp, poisoned, "ImageConvertFenced");
}
fn handle_draw_decoded_masks(
mut dst: SendablePtr<TensorDyn>,
det: SendablePtr<DetectBox>,
seg: SendablePtr<Segmentation>,
overlay: (
f32,
Option<SendablePtr<TensorDyn>>,
Option<[f32; 4]>,
crate::ColorMode,
),
resp: tokio::sync::oneshot::Sender<Result<(), Error>>,
gl_converter: &mut GLProcessorST,
poisoned: &mut bool,
) {
let (opacity, bg, content_region, color_mode) = overlay;
let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
let dst = unsafe { dst.ptr.as_mut() };
let det = unsafe { std::slice::from_raw_parts(det.ptr.as_ptr(), det.len) };
let seg = unsafe { std::slice::from_raw_parts(seg.ptr.as_ptr(), seg.len) };
let bg_ref = bg.map(|p| unsafe { &*p.ptr.as_ptr() });
gl_converter.draw_decoded_masks(
dst,
det,
seg,
crate::MaskOverlay {
background: bg_ref,
opacity,
content_region,
color_mode,
},
)
}));
reply_caught(result, resp, poisoned, "DrawDecodedMasks");
}
fn handle_draw_proto_masks(
mut dst: SendablePtr<TensorDyn>,
det: SendablePtr<DetectBox>,
proto_data: SendablePtr<ProtoData>,
overlay: (
f32,
Option<SendablePtr<TensorDyn>>,
Option<[f32; 4]>,
crate::ColorMode,
),
resp: tokio::sync::oneshot::Sender<Result<(), Error>>,
gl_converter: &mut GLProcessorST,
poisoned: &mut bool,
) {
let (opacity, bg, content_region, color_mode) = overlay;
let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
let dst = unsafe { dst.ptr.as_mut() };
let det = unsafe { std::slice::from_raw_parts(det.ptr.as_ptr(), det.len) };
let bg_ref = bg.map(|p| unsafe { &*p.ptr.as_ptr() });
let proto_data = unsafe { proto_data.ptr.as_ref() };
gl_converter.draw_proto_masks(
dst,
det,
proto_data,
crate::MaskOverlay {
background: bg_ref,
opacity,
content_region,
color_mode,
},
)
}));
reply_caught(result, resp, poisoned, "DrawProtoMasks");
}
fn handle_pbo_create(
size: usize,
resp: tokio::sync::oneshot::Sender<Result<u32, Error>>,
poisoned: &mut bool,
) {
let result = std::panic::catch_unwind(AssertUnwindSafe(|| unsafe {
let mut id: u32 = 0;
edgefirst_gl::gl::GenBuffers(1, &mut id);
edgefirst_gl::gl::BindBuffer(edgefirst_gl::gl::PIXEL_PACK_BUFFER, id);
edgefirst_gl::gl::BufferData(
edgefirst_gl::gl::PIXEL_PACK_BUFFER,
size as isize,
std::ptr::null(),
edgefirst_gl::gl::STREAM_COPY,
);
edgefirst_gl::gl::BindBuffer(edgefirst_gl::gl::PIXEL_PACK_BUFFER, 0);
match check_gl_error("PboCreate", 0) {
Ok(()) => Ok(id),
Err(e) => {
edgefirst_gl::gl::DeleteBuffers(1, &id);
Err(e)
}
}
}));
reply_caught(result, resp, poisoned, "PboCreate");
}
fn pbo_map_on_gl_thread(
buffer_id: u32,
size: usize,
) -> Result<edgefirst_tensor::PboMapping, Error> {
let already = with_worker_mapped(|set| !set.insert(buffer_id)).unwrap_or(false);
if already {
return Err(Error::OpenGl(format!(
"PBO buffer {buffer_id} is already mapped in this GL context \
(a producer and a reconstructed cross-package consumer may \
both hold a live tensor over it)"
)));
}
let result = std::panic::catch_unwind(AssertUnwindSafe(|| unsafe {
edgefirst_gl::gl::BindBuffer(edgefirst_gl::gl::PIXEL_PACK_BUFFER, buffer_id);
let ptr = edgefirst_gl::gl::MapBufferRange(
edgefirst_gl::gl::PIXEL_PACK_BUFFER,
0,
size as isize,
edgefirst_gl::gl::MAP_READ_BIT | edgefirst_gl::gl::MAP_WRITE_BIT,
);
edgefirst_gl::gl::BindBuffer(edgefirst_gl::gl::PIXEL_PACK_BUFFER, 0);
if ptr.is_null() {
Err(Error::OpenGl("glMapBufferRange returned null".to_string()))
} else {
Ok(edgefirst_tensor::PboMapping {
ptr: ptr as *mut u8,
size,
})
}
}));
if !matches!(result, Ok(Ok(_))) {
with_worker_mapped(|set| set.remove(&buffer_id));
}
match result {
Ok(result) => result,
Err(panic) => std::panic::resume_unwind(panic),
}
}
fn pbo_unmap_on_gl_thread(buffer_id: u32) -> Result<(), Error> {
with_worker_mapped(|set| set.remove(&buffer_id));
unsafe {
edgefirst_gl::gl::BindBuffer(edgefirst_gl::gl::PIXEL_PACK_BUFFER, buffer_id);
let ok = edgefirst_gl::gl::UnmapBuffer(edgefirst_gl::gl::PIXEL_PACK_BUFFER);
edgefirst_gl::gl::BindBuffer(edgefirst_gl::gl::PIXEL_PACK_BUFFER, 0);
if ok == edgefirst_gl::gl::FALSE {
Err(Error::OpenGl(
"PBO data was corrupted during mapping".into(),
))
} else {
check_gl_error("PboUnmap", 0)
}
}
}
fn handle_pbo_map(
buffer_id: u32,
size: usize,
resp: tokio::sync::oneshot::Sender<Result<edgefirst_tensor::PboMapping, Error>>,
poisoned: &mut bool,
) {
let result =
std::panic::catch_unwind(AssertUnwindSafe(|| pbo_map_on_gl_thread(buffer_id, size)));
reply_caught(result, resp, poisoned, "PboMap");
}
fn handle_pbo_unmap(
buffer_id: u32,
resp: tokio::sync::oneshot::Sender<Result<(), Error>>,
poisoned: &mut bool,
) {
let result = std::panic::catch_unwind(AssertUnwindSafe(|| pbo_unmap_on_gl_thread(buffer_id)));
reply_caught(result, resp, poisoned, "PboUnmap");
}
fn handle_pbo_delete(buffer_id: u32, poisoned: &mut bool) {
with_worker_mapped(|set| set.remove(&buffer_id));
if let Err(e) = std::panic::catch_unwind(AssertUnwindSafe(|| unsafe {
edgefirst_gl::gl::DeleteBuffers(1, &buffer_id);
})) {
*poisoned = true;
log::error!(
"GL thread panicked during PboDelete: {}",
panic_message(e.as_ref()),
);
}
}
impl GLProcessorThreaded {
pub fn new(kind: Option<EglDisplayKind>) -> Result<Self, Error> {
Self::with_cache_capacity(kind, None)
}
pub fn with_cache_capacity(
kind: Option<EglDisplayKind>,
capacity: Option<usize>,
) -> Result<Self, Error> {
let worker_id = NEXT_GL_WORKER_ID.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let (send, recv) = tokio::sync::mpsc::channel::<GLProcessorMessage>(1);
let (create_ctx_send, create_ctx_recv) = tokio::sync::oneshot::channel();
let owned_pbos = std::sync::Arc::<super::processor::OwnedPbos>::default();
let worker_owned = owned_pbos.clone();
let handle = std::thread::spawn(move || {
run_gl_worker(
worker_id,
worker_owned,
kind,
capacity,
recv,
create_ctx_send,
);
});
let caps = match create_ctx_recv.blocking_recv() {
Ok(Err(e)) => return Err(e),
Err(_) => {
return Err(Error::Internal(
"GL converter error messaging closed without update".to_string(),
));
}
Ok(Ok(caps)) => caps,
};
Ok(Self {
handle: Some(handle),
sender: Some(send),
worker_id,
owned_pbos,
caps,
})
}
}
impl ImageProcessorTrait for GLProcessorThreaded {
fn convert(
&mut self,
src: &TensorDyn,
dst: &mut TensorDyn,
options: Option<ConvertOptions>,
) -> crate::Result<()> {
let options = options.unwrap_or_default();
let (err_send, err_recv) = tokio::sync::oneshot::channel();
self.sender
.as_ref()
.ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
.blocking_send(GLProcessorMessage::ImageConvert(
SendablePtr {
ptr: NonNull::from(src),
len: 1,
},
SendablePtr {
ptr: NonNull::from(dst),
len: 1,
},
options,
false, err_send,
))
.map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
err_recv.blocking_recv().map_err(|_| {
Error::Internal("GL converter error messaging closed without update".to_string())
})?
}
fn convert_deferred(
&mut self,
src: &TensorDyn,
dst: &mut TensorDyn,
options: Option<ConvertOptions>,
) -> crate::Result<()> {
let options = options.unwrap_or_default();
let (err_send, err_recv) = tokio::sync::oneshot::channel();
self.sender
.as_ref()
.ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
.blocking_send(GLProcessorMessage::ImageConvert(
SendablePtr {
ptr: NonNull::from(src),
len: 1,
},
SendablePtr {
ptr: NonNull::from(dst),
len: 1,
},
options,
true, err_send,
))
.map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
err_recv.blocking_recv().map_err(|_| {
Error::Internal("GL converter error messaging closed without update".to_string())
})?
}
fn flush(&mut self) -> crate::Result<()> {
let (err_send, err_recv) = tokio::sync::oneshot::channel();
self.sender
.as_ref()
.ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
.blocking_send(GLProcessorMessage::Flush(err_send))
.map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
err_recv.blocking_recv().map_err(|_| {
Error::Internal("GL converter error messaging closed without update".to_string())
})?
}
fn draw_decoded_masks(
&mut self,
dst: &mut TensorDyn,
detect: &[crate::DetectBox],
segmentation: &[crate::Segmentation],
overlay: crate::MaskOverlay<'_>,
) -> crate::Result<()> {
let (err_send, err_recv) = tokio::sync::oneshot::channel();
self.sender
.as_ref()
.ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
.blocking_send(GLProcessorMessage::DrawDecodedMasks(
SendablePtr {
ptr: NonNull::from(dst),
len: 1,
},
SendablePtr {
ptr: NonNull::new(detect.as_ptr() as *mut DetectBox).unwrap(),
len: detect.len(),
},
SendablePtr {
ptr: NonNull::new(segmentation.as_ptr() as *mut Segmentation).unwrap(),
len: segmentation.len(),
},
overlay.opacity,
overlay.background.map(|bg| SendablePtr {
ptr: NonNull::from(bg).cast::<TensorDyn>(),
len: 1,
}),
overlay.content_region,
overlay.color_mode,
err_send,
))
.map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
err_recv.blocking_recv().map_err(|_| {
Error::Internal("GL converter error messaging closed without update".to_string())
})?
}
fn draw_proto_masks(
&mut self,
dst: &mut TensorDyn,
detect: &[DetectBox],
proto_data: &ProtoData,
overlay: crate::MaskOverlay<'_>,
) -> crate::Result<()> {
let (err_send, err_recv) = tokio::sync::oneshot::channel();
self.sender
.as_ref()
.ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
.blocking_send(GLProcessorMessage::DrawProtoMasks(
SendablePtr {
ptr: NonNull::from(dst),
len: 1,
},
SendablePtr {
ptr: NonNull::new(detect.as_ptr() as *mut DetectBox).unwrap(),
len: detect.len(),
},
SendablePtr {
ptr: NonNull::from(proto_data).cast::<ProtoData>(),
len: 1,
},
overlay.opacity,
overlay.background.map(|bg| SendablePtr {
ptr: NonNull::from(bg).cast::<TensorDyn>(),
len: 1,
}),
overlay.content_region,
overlay.color_mode,
err_send,
))
.map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
err_recv.blocking_recv().map_err(|_| {
Error::Internal("GL converter error messaging closed without update".to_string())
})?
}
fn set_class_colors(&mut self, colors: &[[u8; 4]]) -> Result<(), crate::Error> {
let (err_send, err_recv) = tokio::sync::oneshot::channel();
self.sender
.as_ref()
.ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
.blocking_send(GLProcessorMessage::SetColors(colors.to_vec(), err_send))
.map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
err_recv.blocking_recv().map_err(|_| {
Error::Internal("GL converter error messaging closed without update".to_string())
})?
}
fn set_box_line_width(&mut self, width: u32) -> Result<(), crate::Error> {
crate::check_box_line_width(width)?;
let (err_send, err_recv) = tokio::sync::oneshot::channel();
self.sender
.as_ref()
.ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
.blocking_send(GLProcessorMessage::SetBoxLineWidth(width, err_send))
.map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
err_recv.blocking_recv().map_err(|_| {
Error::Internal("GL converter error messaging closed without update".to_string())
})?
}
}
impl GLProcessorThreaded {
pub fn set_colorimetry_mode(&mut self, mode: crate::ColorimetryMode) -> Result<(), Error> {
let (err_send, err_recv) = tokio::sync::oneshot::channel();
self.sender
.as_ref()
.ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
.blocking_send(GLProcessorMessage::SetColorimetryMode(mode, err_send))
.map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
err_recv.blocking_recv().map_err(|_| {
Error::Internal("GL converter error messaging closed without update".to_string())
})?
}
pub fn set_interpolation(&mut self, mode: crate::Interpolation) -> Result<(), Error> {
let (err_send, err_recv) = tokio::sync::oneshot::channel();
self.sender
.as_ref()
.ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
.blocking_send(GLProcessorMessage::SetInterpolation(mode, err_send))
.map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
err_recv.blocking_recv().map_err(|_| {
Error::Internal("GL converter error messaging closed without update".to_string())
})?
}
#[cfg(test)]
pub(crate) fn set_fail_dst_import(&mut self, fail: bool) -> Result<(), Error> {
let (err_send, err_recv) = tokio::sync::oneshot::channel();
self.sender
.as_ref()
.ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
.blocking_send(GLProcessorMessage::SetFailDstImport(fail, err_send))
.map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
err_recv.blocking_recv().map_err(|_| {
Error::Internal("GL converter error messaging closed without update".to_string())
})?
}
pub fn set_int8_interpolation_mode(
&mut self,
mode: Int8InterpolationMode,
) -> Result<(), crate::Error> {
let (err_send, err_recv) = tokio::sync::oneshot::channel();
self.sender
.as_ref()
.ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
.blocking_send(GLProcessorMessage::SetInt8Interpolation(mode, err_send))
.map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
err_recv.blocking_recv().map_err(|_| {
Error::Internal("GL converter error messaging closed without update".to_string())
})?
}
pub fn egl_cache_stats(&self) -> Result<super::cache::GlCacheStats, Error> {
let (send, recv) = tokio::sync::oneshot::channel();
self.sender
.as_ref()
.ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
.blocking_send(GLProcessorMessage::EglCacheStats(send))
.map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
recv.blocking_recv().map_err(|_| {
Error::Internal("GL converter error messaging closed without update".to_string())
})?
}
pub fn convert_with_fence(
&mut self,
src: &TensorDyn,
dst: &mut TensorDyn,
options: Option<ConvertOptions>,
) -> Result<Option<super::CompletionFence>, Error> {
let options = options.unwrap_or_default();
use crate::ImageProcessorTrait as _;
if !self.caps.native_fence_sync {
self.convert(src, dst, Some(options))?;
return Ok(None);
}
let (send, recv) = tokio::sync::oneshot::channel();
self.sender
.as_ref()
.ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
.blocking_send(GLProcessorMessage::ImageConvertFenced(
SendablePtr {
ptr: NonNull::from(src),
len: 1,
},
SendablePtr {
ptr: NonNull::from(dst),
len: 1,
},
options,
send,
))
.map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
recv.blocking_recv().map_err(|_| {
Error::Internal("GL converter error messaging closed without update".to_string())
})?
}
pub fn convert_stats(&self) -> Result<super::cache::ConvertStats, Error> {
let (send, recv) = tokio::sync::oneshot::channel();
self.sender
.as_ref()
.ok_or_else(|| Error::Internal("GL processor is shutting down".to_string()))?
.blocking_send(GLProcessorMessage::ConvertStats(send))
.map_err(|_| Error::Internal("GL converter thread exited".to_string()))?;
recv.blocking_recv().map_err(|_| {
Error::Internal("GL converter error messaging closed without update".to_string())
})?
}
pub fn create_pbo_image(
&self,
width: usize,
height: usize,
format: edgefirst_tensor::PixelFormat,
) -> Result<edgefirst_tensor::Tensor<u8>, Error> {
let sender = self
.sender
.as_ref()
.ok_or(Error::OpenGl("GL processor is shutting down".to_string()))?;
let size = pbo_elem_count(width, height, format)
.filter(|&s| s != 0)
.ok_or_else(|| Error::OpenGl("Invalid image dimensions".to_string()))?;
let (tx, rx) = tokio::sync::oneshot::channel();
sender
.blocking_send(GLProcessorMessage::PboCreate(size, tx))
.map_err(|_| Error::OpenGl("GL thread channel closed".to_string()))?;
let buffer_id = rx
.blocking_recv()
.map_err(|_| Error::OpenGl("GL thread did not respond".to_string()))??;
let ops: std::sync::Arc<dyn edgefirst_tensor::PboOps> = std::sync::Arc::new(GlPboOps {
sender: sender.downgrade(),
worker_id: self.worker_id,
});
let owned_ops = ops.clone();
let shape = pbo_shape(width, height, format);
let pbo_tensor =
edgefirst_tensor::PboTensor::<u8>::from_pbo(buffer_id, size, &shape, None, ops)
.map_err(|e| Error::OpenGl(format!("PBO tensor creation failed: {e:?}")))?;
let mut tensor = edgefirst_tensor::Tensor::from_pbo(pbo_tensor)
.map_err(|e| Error::OpenGl(format!("PBO tensor wrap failed: {e:?}")))?;
tensor
.set_format(format)
.map_err(|e| Error::OpenGl(format!("Failed to set format on PBO tensor: {e:?}")))?;
register_pbo_cuda(&mut tensor, buffer_id, size, sender);
self.owned_pbos.insert(tensor.pbo_vtable_ptr(), &owned_ops);
Ok(tensor)
}
pub(crate) fn create_pbo_image_dtype(
&self,
width: usize,
height: usize,
format: edgefirst_tensor::PixelFormat,
dtype: edgefirst_tensor::DType,
) -> Result<TensorDyn, Error> {
let sender = self
.sender
.as_ref()
.ok_or(Error::OpenGl("GL processor is shutting down".to_string()))?;
let elems = pbo_elem_count(width, height, format)
.filter(|&e| e != 0)
.ok_or_else(|| Error::OpenGl("Invalid image dimensions".to_string()))?;
let size = elems
.checked_mul(dtype.size())
.ok_or_else(|| Error::OpenGl("PBO size overflow".to_string()))?;
let (tx, rx) = tokio::sync::oneshot::channel();
sender
.blocking_send(GLProcessorMessage::PboCreate(size, tx))
.map_err(|_| Error::OpenGl("GL thread channel closed".to_string()))?;
let buffer_id = rx
.blocking_recv()
.map_err(|_| Error::OpenGl("GL thread did not respond".to_string()))??;
let ops: std::sync::Arc<dyn edgefirst_tensor::PboOps> = std::sync::Arc::new(GlPboOps {
sender: sender.downgrade(),
worker_id: self.worker_id,
});
let owned_ops = ops.clone();
let shape = pbo_shape(width, height, format);
let map_err = |e: edgefirst_tensor::Error| {
Error::OpenGl(format!("PBO tensor creation failed: {e:?}"))
};
let set_err = |e: edgefirst_tensor::Error| {
Error::OpenGl(format!("Failed to set format on PBO tensor: {e:?}"))
};
let tensor = match dtype {
edgefirst_tensor::DType::U8 => {
let pbo =
edgefirst_tensor::PboTensor::<u8>::from_pbo(buffer_id, size, &shape, None, ops)
.map_err(map_err)?;
let mut t = edgefirst_tensor::Tensor::from_pbo(pbo).map_err(map_err)?;
t.set_format(format).map_err(set_err)?;
register_pbo_cuda(&mut t, buffer_id, size, sender);
TensorDyn::from(t)
}
edgefirst_tensor::DType::F16 => {
let pbo = edgefirst_tensor::PboTensor::<edgefirst_tensor::f16>::from_pbo(
buffer_id, size, &shape, None, ops,
)
.map_err(map_err)?;
let mut t = edgefirst_tensor::Tensor::from_pbo(pbo).map_err(map_err)?;
t.set_format(format).map_err(set_err)?;
register_pbo_cuda(&mut t, buffer_id, size, sender);
TensorDyn::from(t)
}
edgefirst_tensor::DType::F32 => {
let pbo = edgefirst_tensor::PboTensor::<f32>::from_pbo(
buffer_id, size, &shape, None, ops,
)
.map_err(map_err)?;
let mut t = edgefirst_tensor::Tensor::from_pbo(pbo).map_err(map_err)?;
t.set_format(format).map_err(set_err)?;
register_pbo_cuda(&mut t, buffer_id, size, sender);
TensorDyn::from(t)
}
other => return Err(Error::OpenGl(format!("unsupported PBO dtype {other:?}"))),
};
self.owned_pbos.insert(tensor.pbo_vtable_ptr(), &owned_ops);
Ok(tensor)
}
pub(crate) fn transfer_backend(&self) -> TransferBackend {
self.caps.transfer_backend
}
pub(crate) fn supported_render_dtypes(&self) -> crate::RenderDtypeSupport {
self.caps.render_dtypes
}
pub(crate) fn dst_cpu_access(
&self,
memory: edgefirst_tensor::TensorMemory,
format: edgefirst_tensor::PixelFormat,
dtype: edgefirst_tensor::DType,
) -> Option<edgefirst_tensor::CpuAccess> {
use edgefirst_tensor::{CpuAccess, DType, PixelFormat, TensorMemory};
let renders = match dtype {
DType::U8 | DType::I8 => GLProcessorST::dst_format_renderable(
self.caps.transfer_backend,
memory,
format,
self.caps.has_bgra,
),
DType::F16 | DType::F32 => {
super::float_dispatch::classify_float_render(
PixelFormat::Rgba,
format,
dtype,
memory,
self.caps.render_dtypes,
<super::platform::Platform as super::platform::GlPlatform>::ZERO_COPY_FLOAT,
) != super::float_dispatch::FloatRenderPath::None
}
_ => false,
};
let readback = matches!(memory, TensorMemory::Mem | TensorMemory::Shm);
renders.then_some(if readback {
CpuAccess::Write
} else {
CpuAccess::None
})
}
}
impl Drop for GLProcessorThreaded {
fn drop(&mut self) {
drop(self.sender.take());
let _ = self.handle.take().and_then(|h| h.join().ok());
}
}
#[cfg(test)]
#[cfg_attr(coverage_nightly, coverage(off))]
mod tests {
use super::{pbo_elem_count, pbo_shape, GLProcessorThreaded, GlPboOps};
use edgefirst_tensor::{PixelFormat, TensorTrait};
#[test]
fn elem_count_per_format() {
assert_eq!(pbo_elem_count(8, 4, PixelFormat::Rgba), Some(8 * 4 * 4));
assert_eq!(pbo_elem_count(8, 4, PixelFormat::Rgb), Some(8 * 4 * 3));
assert_eq!(pbo_elem_count(8, 4, PixelFormat::Nv12), Some(8 * 4 * 3 / 2));
assert_eq!(pbo_elem_count(8, 4, PixelFormat::Nv16), Some(8 * 4 * 2));
}
#[test]
fn elem_count_overflow_is_none() {
assert_eq!(pbo_elem_count(usize::MAX, 2, PixelFormat::Rgba), None);
assert_eq!(pbo_elem_count(usize::MAX, 1, PixelFormat::Rgb), None);
}
#[test]
fn shape_per_format() {
assert_eq!(pbo_shape(8, 4, PixelFormat::PlanarRgb), vec![3, 4, 8]);
assert_eq!(pbo_shape(8, 4, PixelFormat::Nv12), vec![4 * 3 / 2, 8]);
assert_eq!(pbo_shape(8, 4, PixelFormat::Nv16), vec![4 * 2, 8]);
assert_eq!(pbo_shape(8, 4, PixelFormat::Rgba), vec![4, 8, 4]);
}
#[test]
#[cfg(any(target_os = "linux", target_os = "windows"))] fn concurrent_maps_from_two_independent_handles_on_one_buffer_do_not_corrupt_gl_state() {
let gl = match GLProcessorThreaded::new(None) {
Ok(gl) => gl,
Err(e) => {
crate::test_support::report_skip(&format!(
"concurrent_maps_from_two_independent_handles... - \
OpenGL not available: {e:?}"
));
return;
}
};
let tensor_a = gl
.create_pbo_image(64, 64, PixelFormat::Rgba)
.expect("a PBO on a live GL context");
let buffer_id = tensor_a.pbo_id().expect("just created a PBO tensor");
let size = tensor_a.capacity_bytes();
let shape = tensor_a.shape().to_vec();
let ops_b: std::sync::Arc<dyn edgefirst_tensor::PboOps> = std::sync::Arc::new(GlPboOps {
sender: gl.sender.as_ref().unwrap().downgrade(),
worker_id: gl.worker_id,
});
let tensor_b =
edgefirst_tensor::PboTensor::<u8>::from_pbo(buffer_id, size, &shape, None, ops_b)
.expect("from_pbo against a real, already-created buffer_id must succeed");
let map_a = tensor_a.map().expect("first map must succeed");
let map_b_result = tensor_b.map();
let Err(map_b_err) = map_b_result else {
panic!(
"a second, independent handle mapping the SAME real buffer while the \
first mapping is still open must be refused, not silently succeed"
);
};
let msg = format!("{map_b_err:?}");
assert!(
msg.contains("already mapped"),
"must be refused specifically by the GL-thread-level reservation check \
(mapped_pbo_buffers), not some other error path: {msg}"
);
drop(map_a);
let map_b_after = tensor_b.map();
assert!(
map_b_after.is_ok(),
"after the first handle's mapping releases, the second handle must be able to \
map the same buffer: {map_b_after:?}"
);
}
}