From dfea2e1a84b7c563676ee0f0a5e4b76437c445db Mon Sep 17 00:00:00 2001 From: Osei Fortune Date: Mon, 9 Nov 2026 13:01:34 -0400 Subject: [PATCH] perf(webgpu): run queue work on a per-device thread finish, submit, writeBuffer/writeTexture and present now run in order on one thread per device instead of inside the JS call: - encoder.finish() returns at once; the command buffer is encoded on the queue thread just before the submit that uses it. - writeBuffer/writeTexture copy their data and are recorded without waking the thread; recorded work is sent with the next submit, present or drain. - The page draws into a drawing-buffer texture. Present acquires the swapchain image, copies the frame into it and presents on the queue thread, so getCurrentTexture never waits on the swapchain. The copy replaces the readback copy present already made; readbacks read the drawing buffer. Each frame starts cleared through Texture::discard_contents (wgpu bump to 04e9fef72). The page runs at most one frame ahead. - Error scopes are mirrored onto the queue thread, so errors from queued work reach the scope open when the call was made. Uncaptured errors go to the JS thread: through its loop on Android and Apple, otherwise on its next WebGPU call. - getCurrentTexture (non-drawing-buffer paths), configure/resize, readbacks, mapAsync, destroy and the copy helpers wait for queued work first. --- Cargo.toml | 6 +- crates/canvas-c/src/lib.rs | 1 + crates/canvas-c/src/thread_poster.rs | 297 +++++++++++++++ crates/canvas-c/src/webgpu/gpu_adapter.rs | 7 + crates/canvas-c/src/webgpu/gpu_buffer.rs | 4 + .../canvas-c/src/webgpu/gpu_canvas_context.rs | 356 ++++++++++++++---- .../canvas-c/src/webgpu/gpu_command_buffer.rs | 29 +- .../src/webgpu/gpu_command_encoder.rs | 15 +- crates/canvas-c/src/webgpu/gpu_device.rs | 88 ++++- .../canvas-c/src/webgpu/gpu_native_texture.rs | 2 + crates/canvas-c/src/webgpu/gpu_queue.rs | 79 ++-- crates/canvas-c/src/webgpu/gpu_texture.rs | 2 + crates/canvas-c/src/webgpu/mod.rs | 1 + crates/canvas-c/src/webgpu/queue_thread.rs | 245 ++++++++++++ 14 files changed, 997 insertions(+), 135 deletions(-) create mode 100644 crates/canvas-c/src/thread_poster.rs create mode 100644 crates/canvas-c/src/webgpu/queue_thread.rs diff --git a/Cargo.toml b/Cargo.toml index e8056ce3c..1e718f6ff 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -81,7 +81,7 @@ cpal = { path = "crates/vendor/cpal" } [workspace.dependencies.wgt] package = "wgpu-types" git = "https://github.com/triniwiz/wgpu" -rev = "399344e78" +rev = "04e9fef72" @@ -103,8 +103,8 @@ quick-xml = "0.34" skia-safe = { git = "https://github.com/triniwiz/rust-skia", rev = "b9d29071cd8876b2447d58a14346b9fe6a5d21e2", features = ["textlayout"] } itertools = "0.14.0" ustr = "1.1.0" -wgpu-core = { git = "https://github.com/triniwiz/wgpu", rev = "399344e78", features = ["wgsl"] } -wgpu-hal = { git = "https://github.com/triniwiz/wgpu", rev = "399344e78" } +wgpu-core = { git = "https://github.com/triniwiz/wgpu", rev = "04e9fef72", features = ["wgsl"] } +wgpu-hal = { git = "https://github.com/triniwiz/wgpu", rev = "04e9fef72" } ureq = "2.10.1" jni = "0.21.1" regex-lite = "0.1.9" diff --git a/crates/canvas-c/src/lib.rs b/crates/canvas-c/src/lib.rs index b3558e2af..d86b183ab 100644 --- a/crates/canvas-c/src/lib.rs +++ b/crates/canvas-c/src/lib.rs @@ -31,6 +31,7 @@ pub use image_asset::*; #[cfg(any(target_os = "android", target_os = "ios", target_os = "visionos", target_os = "tvos"))] mod raf; mod text_decoder; +mod thread_poster; pub use text_decoder::*; mod text_encoder; pub use text_encoder::*; diff --git a/crates/canvas-c/src/thread_poster.rs b/crates/canvas-c/src/thread_poster.rs new file mode 100644 index 000000000..0ba185611 --- /dev/null +++ b/crates/canvas-c/src/thread_poster.rs @@ -0,0 +1,297 @@ +//! Runs tasks on the thread that created the poster, posted from any thread: the +//! thread's ALooper (Android) or run loop (Apple). Used to call back into JS from +//! the WebGPU queue thread. + +use std::sync::Arc; + +type Task = Box; + +pub(crate) struct ThreadPoster { + tasks: parking_lot::Mutex>, + waker: platform::Waker, + thread: std::thread::ThreadId, +} + +impl ThreadPoster { + /// A poster for the current thread, or `None` if it has no loop. + pub(crate) fn for_current_thread() -> Option> { + platform::Waker::create().map(|waker| { + let poster = Arc::new(ThreadPoster { + tasks: parking_lot::Mutex::new(Vec::new()), + waker, + thread: std::thread::current().id(), + }); + poster.waker.attach(Arc::downgrade(&poster)); + poster + }) + } + + pub(crate) fn is_current(&self) -> bool { + std::thread::current().id() == self.thread + } + + /// Runs `task` on the poster's thread: now if called there, else on its loop. + pub(crate) fn post(&self, task: Task) { + if self.is_current() { + task(); + return; + } + self.tasks.lock().push(task); + self.waker.wake(); + } + + fn run(&self) { + loop { + let tasks = std::mem::take(&mut *self.tasks.lock()); + if tasks.is_empty() { + return; + } + for task in tasks { + task(); + } + } + } +} + +#[cfg(target_os = "android")] +mod platform { + use super::ThreadPoster; + use std::os::fd::RawFd; + use std::sync::Weak; + + extern "C" { + fn ALooper_forThread() -> *mut std::ffi::c_void; + fn ALooper_addFd( + looper: *mut std::ffi::c_void, + fd: RawFd, + ident: i32, + events: i32, + callback: Option i32>, + data: *mut std::ffi::c_void, + ) -> i32; + fn ALooper_removeFd(looper: *mut std::ffi::c_void, fd: RawFd) -> i32; + fn ALooper_acquire(looper: *mut std::ffi::c_void); + fn ALooper_release(looper: *mut std::ffi::c_void); + } + + const ALOOPER_POLL_CALLBACK: i32 = -2; + const ALOOPER_EVENT_INPUT: i32 = 1; + + pub(super) struct Waker { + looper: *mut std::ffi::c_void, + fd: RawFd, + poster: parking_lot::Mutex<*mut Weak>, + } + + // SAFETY: ALooper and eventfd writes are thread-safe; `poster` is only set once. + unsafe impl Send for Waker {} + unsafe impl Sync for Waker {} + + extern "C" fn on_wake(fd: RawFd, _events: i32, data: *mut std::ffi::c_void) -> i32 { + let mut count = 0u64; + unsafe { libc::read(fd, &mut count as *mut u64 as *mut _, 8) }; + let poster = unsafe { &*(data as *const Weak) }; + if let Some(poster) = poster.upgrade() { + poster.run(); + } + 1 + } + + impl Waker { + pub(super) fn create() -> Option { + let looper = unsafe { ALooper_forThread() }; + if looper.is_null() { + return None; + } + let fd = unsafe { libc::eventfd(0, libc::EFD_NONBLOCK | libc::EFD_CLOEXEC) }; + if fd < 0 { + return None; + } + unsafe { ALooper_acquire(looper) }; + Some(Waker { + looper, + fd, + poster: parking_lot::Mutex::new(std::ptr::null_mut()), + }) + } + + pub(super) fn attach(&self, poster: Weak) { + let data = Box::into_raw(Box::new(poster)); + *self.poster.lock() = data; + unsafe { + ALooper_addFd( + self.looper, + self.fd, + ALOOPER_POLL_CALLBACK, + ALOOPER_EVENT_INPUT, + Some(on_wake), + data as *mut _, + ) + }; + } + + pub(super) fn wake(&self) { + let one = 1u64; + unsafe { libc::write(self.fd, &one as *const u64 as *const _, 8) }; + } + } + + impl Drop for Waker { + fn drop(&mut self) { + unsafe { + ALooper_removeFd(self.looper, self.fd); + libc::close(self.fd); + ALooper_release(self.looper); + let data = *self.poster.lock(); + if !data.is_null() { + drop(Box::from_raw(data)); + } + } + } + } +} + +#[cfg(any(target_os = "ios", target_os = "visionos", target_os = "tvos"))] +mod platform { + use super::ThreadPoster; + use std::ffi::c_void; + use std::sync::Weak; + + type CFTypeRef = *const c_void; + + #[repr(C)] + struct CFRunLoopSourceContext { + version: isize, + info: *mut c_void, + retain: Option *const c_void>, + release: Option, + copy_description: Option CFTypeRef>, + equal: Option u8>, + hash: Option usize>, + schedule: Option, + cancel: Option, + perform: Option, + } + + #[link(name = "CoreFoundation", kind = "framework")] + extern "C" { + static kCFRunLoopCommonModes: CFTypeRef; + fn CFRunLoopGetCurrent() -> CFTypeRef; + fn CFRunLoopSourceCreate( + allocator: CFTypeRef, + order: isize, + context: *mut CFRunLoopSourceContext, + ) -> CFTypeRef; + fn CFRunLoopAddSource(run_loop: CFTypeRef, source: CFTypeRef, mode: CFTypeRef); + fn CFRunLoopSourceInvalidate(source: CFTypeRef); + fn CFRunLoopSourceSignal(source: CFTypeRef); + fn CFRunLoopWakeUp(run_loop: CFTypeRef); + fn CFRetain(value: CFTypeRef) -> CFTypeRef; + fn CFRelease(value: CFTypeRef); + } + + pub(super) struct Waker { + run_loop: CFTypeRef, + source: parking_lot::Mutex, + poster: parking_lot::Mutex<*mut Weak>, + } + + // SAFETY: CFRunLoopSourceSignal and CFRunLoopWakeUp are thread-safe. + unsafe impl Send for Waker {} + unsafe impl Sync for Waker {} + + extern "C" fn perform(info: *mut c_void) { + let poster = unsafe { &*(info as *const Weak) }; + if let Some(poster) = poster.upgrade() { + poster.run(); + } + } + + impl Waker { + pub(super) fn create() -> Option { + let run_loop = unsafe { CFRunLoopGetCurrent() }; + if run_loop.is_null() { + return None; + } + unsafe { CFRetain(run_loop) }; + Some(Waker { + run_loop, + source: parking_lot::Mutex::new(std::ptr::null()), + poster: parking_lot::Mutex::new(std::ptr::null_mut()), + }) + } + + pub(super) fn attach(&self, poster: Weak) { + let data = Box::into_raw(Box::new(poster)); + *self.poster.lock() = data; + let mut context = CFRunLoopSourceContext { + version: 0, + info: data as *mut c_void, + retain: None, + release: None, + copy_description: None, + equal: None, + hash: None, + schedule: None, + cancel: None, + perform: Some(perform), + }; + unsafe { + let source = CFRunLoopSourceCreate(std::ptr::null(), 0, &mut context); + if !source.is_null() { + CFRunLoopAddSource(self.run_loop, source, kCFRunLoopCommonModes); + *self.source.lock() = source; + } + } + } + + pub(super) fn wake(&self) { + let source = *self.source.lock(); + if source.is_null() { + return; + } + unsafe { + CFRunLoopSourceSignal(source); + CFRunLoopWakeUp(self.run_loop); + } + } + } + + impl Drop for Waker { + fn drop(&mut self) { + unsafe { + let source = *self.source.lock(); + if !source.is_null() { + CFRunLoopSourceInvalidate(source); + CFRelease(source); + } + CFRelease(self.run_loop); + let data = *self.poster.lock(); + if !data.is_null() { + drop(Box::from_raw(data)); + } + } + } + } +} + +#[cfg(not(any( + target_os = "android", + target_os = "ios", + target_os = "visionos", + target_os = "tvos" +)))] +mod platform { + use super::ThreadPoster; + use std::sync::Weak; + + pub(super) struct Waker; + + impl Waker { + pub(super) fn create() -> Option { + None + } + pub(super) fn attach(&self, _poster: Weak) {} + pub(super) fn wake(&self) {} + } +} diff --git a/crates/canvas-c/src/webgpu/gpu_adapter.rs b/crates/canvas-c/src/webgpu/gpu_adapter.rs index 6207c78ea..04ece2db2 100644 --- a/crates/canvas-c/src/webgpu/gpu_adapter.rs +++ b/crates/canvas-c/src/webgpu/gpu_adapter.rs @@ -196,6 +196,13 @@ pub extern "C" fn canvas_native_webgpu_adapter_request_device( let error_sink = Arc::new(parking_lot::Mutex::new(ErrorSinkRaw::new( DEFAULT_DEVICE_LOST_HANDLER, ))); + // wgpu's own default panics; errors from the queue thread must be routed. + super::gpu_device::install_uncaptured_handler( + &device, + &error_sink, + None, + std::ptr::null_mut(), + ); let queue = Arc::new(CanvasGPUQueue { label: descriptor.label, diff --git a/crates/canvas-c/src/webgpu/gpu_buffer.rs b/crates/canvas-c/src/webgpu/gpu_buffer.rs index e1f3c2fc1..aefa71c46 100644 --- a/crates/canvas-c/src/webgpu/gpu_buffer.rs +++ b/crates/canvas-c/src/webgpu/gpu_buffer.rs @@ -78,6 +78,8 @@ impl CanvasGPUBuffer { } pub fn destroy(&self) { + // After the queued work that may still use it. + super::gpu_device::timeline_of(&self.error_sink).drain(); self.buffer.destroy(); } @@ -211,6 +213,8 @@ pub extern "C" fn canvas_native_webgpu_buffer_map_async( return; } let buffer = unsafe { &*buffer }; + // Mapping must follow the queued submits that use the buffer. + super::gpu_device::timeline_of(&buffer.error_sink).drain(); let offset: u64 = offset.try_into().ok().unwrap_or_default(); let size: Option = size.try_into().ok(); diff --git a/crates/canvas-c/src/webgpu/gpu_canvas_context.rs b/crates/canvas-c/src/webgpu/gpu_canvas_context.rs index 08d936597..cf9469104 100644 --- a/crates/canvas-c/src/webgpu/gpu_canvas_context.rs +++ b/crates/canvas-c/src/webgpu/gpu_canvas_context.rs @@ -62,6 +62,11 @@ pub struct CanvasGPUCanvasContext { pub(crate) view_data: parking_lot::Mutex, pub(crate) current_texture: parking_lot::Mutex>>, pub(crate) offscreen_texture: parking_lot::Mutex>, + /// The texture the page draws into. Present copies it to the swapchain on the + /// queue thread, so the page never waits on the swapchain. + pub(crate) drawing: parking_lot::Mutex>, + /// Presents queued and not yet done; the page may run at most one frame ahead. + pub(crate) presents_queued: Arc, pub(crate) last_capabilities: parking_lot::Mutex>, /// Windows: the SwapChainPanel stand-in wgpu binds its swapchain through (keeps the /// swapchain's DPI / fit transform ours). @@ -80,6 +85,18 @@ impl Drop for CanvasGPUCanvasContext { } } +/// Waits for the device's queued work; surface and readback calls must follow it. +fn drain_queue(context: &CanvasGPUCanvasContext) { + let timeline = context + .data + .lock() + .as_ref() + .map(|data| super::gpu_device::timeline_of(&data.error_sink)); + if let Some(timeline) = timeline { + timeline.drain(); + } +} + fn discard_current_texture(context: &CanvasGPUCanvasContext, operation: &'static str) { let current = context.current_texture.lock().take(); if let Some(current) = current { @@ -348,6 +365,7 @@ pub unsafe extern "C" fn canvas_native_webgpu_context_detach_surface( return; } let context = &*context; + drain_queue(context); let mut surface = context.surface.lock(); discard_current_texture(context, "canvas_native_webgpu_context_detach_surface"); *surface = None; @@ -362,6 +380,7 @@ pub unsafe extern "C" fn canvas_native_webgpu_to_data_url_with_fallback( if context.is_null() { return CString::new("data:,").unwrap().into_raw(); } + drain_queue(&*context); let format = if format.is_null() { unsafe { CStr::from_bytes_with_nul_unchecked(b"image/png\0") } @@ -485,8 +504,9 @@ fn to_data_url( is_bgra: bool, ) -> Option { let (texture_id, width, height) = { + let drawing = context.drawing.lock(); let read_back_texture_lock = context.read_back_texture.lock(); - let texture = read_back_texture_lock.as_ref()?; + let texture = drawing.as_ref().or(read_back_texture_lock.as_ref())?; (Arc::clone(&texture.texture), texture.data.size.width, texture.data.size.height) }; to_data_url_with_texture(context, device, texture_id, width, height, format, quality, is_bgra) @@ -657,6 +677,7 @@ fn with_mapped_texture( /// The canvas's frame as tightly packed, unpremultiplied RGBA8: the current texture if one is /// out, else the last presented frame. `None` before `configure` or for a non-8-bit format. pub(crate) fn read_rgba8(context: &CanvasGPUCanvasContext) -> Option<(u32, u32, Vec)> { + drain_queue(context); let (device, format, premultiplied) = { let data = context.data.lock(); let data = data.as_ref()?; @@ -676,8 +697,9 @@ pub(crate) fn read_rgba8(context: &CanvasGPUCanvasContext) -> Option<(u32, u32, let (texture, width, height) = match current { Some(current) => (Arc::clone(¤t.texture), current.width, current.height), None => { + let drawing = context.drawing.lock(); let read_back = context.read_back_texture.lock(); - let read_back = read_back.as_ref()?; + let read_back = drawing.as_ref().or(read_back.as_ref())?; (Arc::clone(&read_back.texture), read_back.data.size.width, read_back.data.size.height) } }; @@ -723,6 +745,8 @@ pub unsafe extern "C" fn canvas_native_webgpu_context_create_offscreen( data: Mutex::default(), view_data: Mutex::new(ViewData { width, height }), read_back_texture: Mutex::default(), + drawing: Default::default(), + presents_queued: Default::default(), current_texture: Mutex::default(), offscreen_texture: Mutex::default(), last_capabilities: Mutex::default(), @@ -743,6 +767,7 @@ pub unsafe extern "C" fn canvas_native_webgpu_context_resize_offscreen( return; } let context = &*context; + drain_queue(context); discard_current_texture(context, "canvas_native_webgpu_context_resize_offscreen"); { let mut view_data = context.view_data.lock(); @@ -829,6 +854,8 @@ pub unsafe extern "C" fn canvas_native_webgpu_context_create( data: Default::default(), view_data: Mutex::new(ViewData { width, height }), read_back_texture: Mutex::default(), + drawing: Default::default(), + presents_queued: Default::default(), current_texture: Mutex::default(), offscreen_texture: Mutex::default(), last_capabilities: Mutex::default(), @@ -859,6 +886,8 @@ pub unsafe extern "C" fn canvas_native_webgpu_context_resize( let context = &*context; + drain_queue(context); + let display_handle = RawDisplayHandle::Android(raw_window_handle::AndroidDisplayHandle::new()); let Some(window_handle_ptr) = std::ptr::NonNull::new(window) else { @@ -997,6 +1026,8 @@ pub unsafe extern "C" fn canvas_native_webgpu_context_create( has_surface_presented: Arc::default(), view_data: Mutex::new(ViewData { width, height }), read_back_texture: Mutex::default(), + drawing: Default::default(), + presents_queued: Default::default(), current_texture: Mutex::default(), offscreen_texture: Mutex::default(), last_capabilities: Mutex::default(), @@ -1051,6 +1082,8 @@ pub(crate) unsafe fn create_swap_chain_panel( instance, surface: Mutex::new(Some(surface)), read_back_texture: Default::default(), + drawing: Default::default(), + presents_queued: Default::default(), has_surface_presented: Arc::default(), data: Mutex::default(), view_data: Mutex::new(ViewData { width, height }), @@ -1101,6 +1134,7 @@ pub unsafe extern "C" fn canvas_native_webgpu_context_resize_swap_chain_panel( return; } let context = &*context; + drain_queue(context); let surface = context.surface.lock(); discard_current_texture(context, "canvas_native_webgpu_context_resize_swap_chain_panel"); { @@ -1178,6 +1212,8 @@ pub unsafe extern "C" fn canvas_native_webgpu_context_create_uiview( instance, surface: Mutex::new(Some(surface_id)), read_back_texture: Default::default(), + drawing: Default::default(), + presents_queued: Default::default(), has_surface_presented: Arc::default(), data: Mutex::default(), view_data: Mutex::new(ViewData { width, height }), @@ -1209,6 +1245,7 @@ pub unsafe extern "C" fn canvas_native_webgpu_context_resize_uiview( return; } let context = &*context; + drain_queue(context); let mut surface = context.surface.lock(); @@ -1336,6 +1373,8 @@ pub unsafe extern "C" fn canvas_native_webgpu_context_create_nsview( data: Mutex::default(), view_data: Mutex::new(ViewData { width, height }), read_back_texture: Mutex::default(), + drawing: Default::default(), + presents_queued: Default::default(), current_texture: Mutex::default(), offscreen_texture: Mutex::default(), last_capabilities: Mutex::default(), @@ -1364,6 +1403,7 @@ pub unsafe extern "C" fn canvas_native_webgpu_context_resize_nsview( return; } let context = &*context; + drain_queue(context); let display_handle = RawDisplayHandle::AppKit(AppKitDisplayHandle::new()); @@ -1473,6 +1513,7 @@ pub unsafe extern "C" fn canvas_native_webgpu_context_resize_layer( return; } let context = &*context; + drain_queue(context); let mut surface = context.surface.lock(); @@ -1703,6 +1744,8 @@ pub unsafe extern "C" fn canvas_native_webgpu_context_configure( } Arc::increment_strong_count(context); let context = Arc::from_raw(context); + drain_queue(&context); + super::gpu_device::timeline_of(&(*device).error_sink).drain(); let surface_id = context.surface.lock(); let device_ref = unsafe { &*device }; let device_id = Arc::clone(&device_ref.device); @@ -1882,6 +1925,7 @@ pub unsafe extern "C" fn canvas_native_webgpu_context_unconfigure( return; } let context = &*context; + drain_queue(context); let mut lock = context.data.lock(); *lock = None; @@ -1933,6 +1977,14 @@ pub extern "C" fn canvas_native_webgpu_context_get_current_texture( } } + #[cfg(not(target_os = "linux"))] + if let Some(texture) = drawing_current_texture(context) { + return texture; + } + + // One image can be acquired at a time: the queued present of the last one goes first. + drain_queue(context); + let surface_guard = context.surface.lock(); let Some(surface_id) = surface_guard.as_ref() else { drop(surface_guard); @@ -2033,68 +2085,239 @@ pub unsafe extern "C" fn canvas_native_webgpu_context_present_surface( return; } - let context = unsafe { &*context }; - - let surface_id = context.surface.lock(); - let texture = unsafe { &*texture }; + Arc::increment_strong_count(context); + let context = Arc::from_raw(context); + // Takes over the caller's reference. + let texture = Arc::from_raw(texture); + // Presented as far as the page can tell; the GPU work follows the frame's submits. + context + .has_surface_presented + .store(true, std::sync::atomic::Ordering::SeqCst); { - let lock = context.read_back_texture.lock(); - let mut surface_lock = context.data.lock(); - if let (Some(dst_texture), Some(data)) = (lock.as_ref(), surface_lock.as_mut()) { - unsafe { - let texture_extent = wgt::Extent3d { - width: dst_texture.data.size.width, - height: dst_texture.data.size.height, - depth_or_array_layers: 1, - }; + let mut current_texture = context.current_texture.lock(); + if current_texture + .as_ref() + .is_some_and(|current| Arc::ptr_eq(¤t.texture, &texture.texture)) + { + *current_texture = None; + } + } - let texture_src_copy = wgt::TexelCopyTextureInfo { - texture: Arc::clone(&texture.texture), - mip_level: 0, - origin: wgt::Origin3d::ZERO, - aspect: wgt::TextureAspect::All, - }; + let timeline = context + .data + .lock() + .as_ref() + .map(|data| super::gpu_device::timeline_of(&data.error_sink)); + let drawing = is_drawing_buffer(&context, &texture); + if drawing { + context + .presents_queued + .fetch_add(1, std::sync::atomic::Ordering::AcqRel); + } + let present = move || { + if drawing { + present_drawing_buffer(&context, &texture); + context + .presents_queued + .fetch_sub(1, std::sync::atomic::Ordering::AcqRel); + } else { + present_now(&context, &texture); + } + }; + match timeline { + Some(timeline) => timeline.get().enqueue(Box::new(present)), + None => present(), + } +} - let texture_dst_copy = wgt::TexelCopyTextureInfo { - texture: Arc::clone(&dst_texture.texture), - mip_level: 0, - origin: wgt::Origin3d::ZERO, - aspect: wgt::TextureAspect::All, +/// The drawing buffer as the frame's texture, when a surface is configured. +#[cfg(not(target_os = "linux"))] +fn drawing_current_texture(context: &CanvasGPUCanvasContext) -> Option<*const CanvasGPUTexture> { + if context.surface.lock().is_none() { + return None; + } + let (texture, data, error_sink) = { + let data_guard = context.data.lock(); + let surface_data = data_guard.as_ref()?; + let data = surface_data.texture_data; + let usage = data.usage | wgt::TextureUsages::COPY_SRC; + let mut drawing = context.drawing.lock(); + let texture = match drawing.as_ref() { + Some(drawing) + if drawing.data.size == data.size + && drawing.data.format == data.format + && drawing.data.usage == usage => + { + Arc::clone(&drawing.texture) + } + _ => { + let desc = wgt::TextureDescriptor { + label: Some(Cow::Borrowed("CanvasDrawingBuffer")), + size: data.size, + mip_level_count: 1, + sample_count: 1, + dimension: wgt::TextureDimension::D2, + format: data.format, + usage, + view_formats: surface_data.previous_configuration.view_formats.clone(), }; + let texture = surface_data.device.device.create_texture(&desc); + *drawing = Some(ReadBackTexture { + texture: Arc::clone(&texture), + data: TextureData { usage, ..data }, + }); + texture + } + }; + (texture, TextureData { usage, ..data }, surface_data.error_sink.clone()) + }; - let device = &data.device.device; - - let label = Cow::Borrowed("PresentSurface:Encoder"); - - let encoder = device.create_command_encoder( - &wgt::CommandEncoderDescriptor { - label: wgpu_core::Label::from(label), - }, - ); - - { - { - { - encoder.copy_texture_to_texture( - &texture_src_copy, - &texture_dst_copy, - &texture_extent, - ); + let timeline = super::gpu_device::timeline_of(&error_sink); + // At most one frame ahead of the queue thread. + if context + .presents_queued + .load(std::sync::atomic::Ordering::Acquire) + > 1 + { + timeline.drain(); + } + // A new frame starts transparent black: after the last frame's work, in order. + let discard = Arc::clone(&texture); + timeline.get().record(Box::new(move || discard.discard_contents())); - let desc = wgt::CommandBufferDescriptor { label: None }; + context + .has_surface_presented + .store(false, std::sync::atomic::Ordering::SeqCst); + let texture = Arc::new(CanvasGPUTexture { + label: None, + instance: context.instance.clone(), + texture, + surface_id: None, + owned: false, + depth_or_array_layers: 1, + dimension: super::enums::CanvasTextureDimension::D2, + format: data.format.into(), + mipLevelCount: 1, + sampleCount: 1, + width: data.size.width, + height: data.size.height, + usage: data.usage.bits(), + error_sink, + suboptimal: false, + status: SurfaceGetCurrentTextureStatus::Success, + has_surface_presented: context.has_surface_presented.clone(), + }); + let ret = Arc::into_raw(Arc::clone(&texture)); + *context.current_texture.lock() = Some(texture); + Some(ret) +} - let id = encoder.finish(&desc); +fn is_drawing_buffer(context: &CanvasGPUCanvasContext, texture: &CanvasGPUTexture) -> bool { + context + .drawing + .lock() + .as_ref() + .is_some_and(|drawing| Arc::ptr_eq(&drawing.texture, &texture.texture)) +} - let index = data.device.queue.queue.id.submit(&[id]); - if texture.surface_id.is_some() { - data.presented_submission = Some(index); - } - } +/// Queue thread: copies the drawing buffer into the next swapchain image and presents it. +fn present_drawing_buffer(context: &CanvasGPUCanvasContext, texture: &CanvasGPUTexture) { + let surface_guard = context.surface.lock(); + let Some(surface) = surface_guard.as_ref() else { + return; + }; + if !apply_pending_configuration(context, surface, "canvas_native_webgpu_context_present_surface") { + return; + } + let output = match surface.get_current_texture() { + Ok(output) => output, + Err(cause) => { + handle_error_fatal(cause, "canvas_native_webgpu_context_present_surface"); + return; + } + }; + let target = match (output.status, output.texture) { + (SurfaceStatus::Good | SurfaceStatus::Suboptimal, Some(target)) => target, + (status, target) => { + if target.is_some() { + let _ = surface.discard(); + } + if matches!(status, SurfaceStatus::Lost | SurfaceStatus::Outdated) { + if let Some(data) = context.data.lock().as_mut() { + if data.pending_configuration.is_none() { + data.pending_configuration = Some(data.previous_configuration.clone()); } } } + return; + } + }; + + { + let mut data_guard = context.data.lock(); + let Some(data) = data_guard.as_mut() else { + let _ = surface.discard(); + return; }; + let copy = |texture| wgt::TexelCopyTextureInfo { + texture, + mip_level: 0, + origin: wgt::Origin3d::ZERO, + aspect: wgt::TextureAspect::All, + }; + let size = wgt::Extent3d { + width: texture.width.min(data.texture_data.size.width), + height: texture.height.min(data.texture_data.size.height), + depth_or_array_layers: 1, + }; + let device = &data.device.device; + let encoder = device.create_command_encoder(&wgt::CommandEncoderDescriptor { + label: wgpu_core::Label::from(Cow::Borrowed("PresentDrawingBuffer")), + }); + encoder.copy_texture_to_texture(©(Arc::clone(&texture.texture)), ©(target), &size); + let id = encoder.finish(&wgt::CommandBufferDescriptor { label: None }); + data.presented_submission = Some(data.device.queue.queue.id.submit(&[id])); + } + + if let Err(cause) = surface.present() { + handle_error_fatal(cause, "canvas_native_webgpu_context_present_surface"); + } +} + +fn present_now(context: &CanvasGPUCanvasContext, texture: &CanvasGPUTexture) { + let surface_id = context.surface.lock(); + + { + let lock = context.read_back_texture.lock(); + let mut surface_lock = context.data.lock(); + if let (Some(dst_texture), Some(data)) = (lock.as_ref(), surface_lock.as_mut()) { + let texture_extent = wgt::Extent3d { + width: dst_texture.data.size.width, + height: dst_texture.data.size.height, + depth_or_array_layers: 1, + }; + let copy = |texture| wgt::TexelCopyTextureInfo { + texture, + mip_level: 0, + origin: wgt::Origin3d::ZERO, + aspect: wgt::TextureAspect::All, + }; + let device = &data.device.device; + let encoder = device.create_command_encoder(&wgt::CommandEncoderDescriptor { + label: wgpu_core::Label::from(Cow::Borrowed("PresentSurface:Encoder")), + }); + encoder.copy_texture_to_texture( + ©(Arc::clone(&texture.texture)), + ©(Arc::clone(&dst_texture.texture)), + &texture_extent, + ); + let id = encoder.finish(&wgt::CommandBufferDescriptor { label: None }); + let index = data.device.queue.queue.id.submit(&[id]); + if texture.surface_id.is_some() { + data.presented_submission = Some(index); + } + } } #[cfg(target_os = "linux")] @@ -2111,43 +2334,14 @@ pub unsafe extern "C" fn canvas_native_webgpu_context_present_surface( }; if let Err(cause) = presented { - context - .has_surface_presented - .store(true, std::sync::atomic::Ordering::SeqCst); - { - let mut current_texture = context.current_texture.lock(); - if let Some(current_texture_ref) = current_texture.as_ref() { - if Arc::ptr_eq(¤t_texture_ref.texture, &texture.texture) { - *current_texture = None; - } - } - } - unsafe { Arc::decrement_strong_count(texture) }; match cause { SurfaceError::AlreadyAcquired => { log::warn!("present_surface: no acquired texture in wgpu (cache was stale), skipping"); } _ => { - handle_error_fatal(cause, - "canvas_native_webgpu_context_present_surface", - ); + handle_error_fatal(cause, "canvas_native_webgpu_context_present_surface"); } } - } else { - context - .has_surface_presented - .store(true, std::sync::atomic::Ordering::SeqCst); - - { - let mut current_texture = context.current_texture.lock(); - if let Some(current_texture_ref) = current_texture.as_ref() { - if Arc::ptr_eq(¤t_texture_ref.texture, &texture.texture) { - *current_texture = None; - } - } - } - - unsafe { Arc::decrement_strong_count(texture) }; } } diff --git a/crates/canvas-c/src/webgpu/gpu_command_buffer.rs b/crates/canvas-c/src/webgpu/gpu_command_buffer.rs index 851a2a189..a052c9b2e 100644 --- a/crates/canvas-c/src/webgpu/gpu_command_buffer.rs +++ b/crates/canvas-c/src/webgpu/gpu_command_buffer.rs @@ -7,13 +7,40 @@ use crate::webgpu::prelude::label_to_ptr; //use wgpu_core::gfx_select; use super::gpu::CanvasWebGPUInstance; +/// An encoder whose `finish` runs when the command buffer is first needed. +pub(crate) struct PendingFinish { + pub(crate) encoder: Arc, + pub(crate) label: Option>, + pub(crate) scopes: super::queue_thread::ScopeSnapshot, +} + pub struct CanvasGPUCommandBuffer { pub(crate) label: Option>, pub(crate) instance: Arc, - pub(crate) command_buffer: Arc, + pub(crate) pending: parking_lot::Mutex>, + pub(crate) finished: std::sync::OnceLock>, pub(crate) open: std::sync::atomic::AtomicBool, } +impl CanvasGPUCommandBuffer { + /// Encoding the commands is the expensive part of `finish`; it runs here, on the + /// queue thread for a submit, reporting into the scopes open at `finish()`. + pub(crate) fn command_buffer(&self) -> Arc { + Arc::clone(self.finished.get_or_init(|| { + let pending = self + .pending + .lock() + .take() + .expect("a command buffer is finished once"); + super::queue_thread::with_scopes(&pending.scopes, || { + pending.encoder.finish(&wgt::CommandBufferDescriptor { + label: pending.label.clone(), + }) + }) + })) + } +} + #[no_mangle] pub unsafe extern "C" fn canvas_native_webgpu_command_buffer_get_label( command_buffer: *const CanvasGPUCommandBuffer, diff --git a/crates/canvas-c/src/webgpu/gpu_command_encoder.rs b/crates/canvas-c/src/webgpu/gpu_command_encoder.rs index 6b1ba5c78..e78945971 100644 --- a/crates/canvas-c/src/webgpu/gpu_command_encoder.rs +++ b/crates/canvas-c/src/webgpu/gpu_command_encoder.rs @@ -522,16 +522,15 @@ pub unsafe extern "C" fn canvas_native_webgpu_command_encoder_finish( let label = ptr_into_label(label); - let desc = wgt::CommandBufferDescriptor { - label: label.clone(), - }; - - let id = command_encoder.encoder.finish(&desc); - Arc::into_raw(Arc::new(CanvasGPUCommandBuffer { - label, + label: label.clone(), instance: command_encoder.instance.clone(), - command_buffer: id, + pending: parking_lot::Mutex::new(Some(super::gpu_command_buffer::PendingFinish { + encoder: Arc::clone(&command_encoder.encoder), + label, + scopes: super::gpu_device::scopes_of(&command_encoder.error_sink), + })), + finished: std::sync::OnceLock::new(), open: std::sync::atomic::AtomicBool::new(true), })) } diff --git a/crates/canvas-c/src/webgpu/gpu_device.rs b/crates/canvas-c/src/webgpu/gpu_device.rs index e284e32bb..5bd2378af 100644 --- a/crates/canvas-c/src/webgpu/gpu_device.rs +++ b/crates/canvas-c/src/webgpu/gpu_device.rs @@ -149,6 +149,51 @@ pub struct ErrorSinkRaw { scopes: Vec, uncaptured_handler: UncapturedErrorCallback, device_lost_handler: DeviceLostCallback, + pub(crate) timeline: Arc, + /// Mirrors the JS thread's error scopes for work that runs on the queue thread. + open_scopes: Vec>, +} + +/// The device timeline every object of a device queues work on. +pub(crate) fn timeline_of(sink: &ErrorSink) -> Arc { + Arc::clone(&sink.lock().timeline) +} + +/// The error scopes open now, for work queued now. +pub(crate) fn scopes_of(sink: &ErrorSink) -> super::queue_thread::ScopeSnapshot { + super::queue_thread::ScopeSnapshot(sink.lock().open_scopes.clone()) +} + +/// Routes wgpu errors raised on the queue thread to the scope that was open when +/// the work was queued, or to `handler` on the JS thread. +pub(crate) fn install_uncaptured_handler( + device: &Arc, + sink: &ErrorSink, + handler: GPUErrorCallback, + userdata: *mut c_void, +) { + let timeline = timeline_of(sink); + let userdata = userdata as usize; + device.on_uncaptured_error(Arc::new(move |error: wgt::error::Error| { + let Err(error) = super::queue_thread::capture(error) else { + return; + }; + let typ = match error { + wgt::error::Error::OutOfMemory { .. } => CanvasGPUErrorType::OutOfMemory, + wgt::error::Error::Validation { .. } => CanvasGPUErrorType::Validation, + _ => CanvasGPUErrorType::Internal, + }; + let msg = error.to_string().replace('\0', ""); + timeline.get().post_to_js(Box::new(move || { + let msg = CString::new(msg).unwrap_or_else(|_| CString::new("webgpu error").unwrap()); + match handler { + Some(callback) => unsafe { callback(typ, msg.into_raw(), userdata as *mut c_void) }, + None => unsafe { + default_uncaptured_error_handler(typ, msg.into_raw(), std::ptr::null_mut()) + }, + } + })); + })); } impl PartialEq for CanvasGPUError { @@ -169,6 +214,8 @@ impl ErrorSinkRaw { scopes: Vec::new(), uncaptured_handler: DEFAULT_UNCAPTURED_ERROR_HANDLER, device_lost_handler, + timeline: Arc::default(), + open_scopes: Vec::new(), } } @@ -225,11 +272,13 @@ impl ErrorSinkRaw { } None => { if let Some(callback) = self.uncaptured_handler.callback { - let userdata = self.uncaptured_handler.userdata; + let userdata = self.uncaptured_handler.userdata as usize; let err_str = err.to_string().replace('\0', ""); - let msg = CString::new(err_str) - .unwrap_or_else(|_| CString::new("uncaptured error").unwrap()); - unsafe { callback(typ, msg.into_raw(), userdata) }; + self.timeline.get().post_to_js(Box::new(move || { + let msg = CString::new(err_str) + .unwrap_or_else(|_| CString::new("uncaptured error").unwrap()); + unsafe { callback(typ, msg.into_raw(), userdata as *mut c_void) }; + })); } } } @@ -274,6 +323,7 @@ impl CanvasGPUDevice { } pub fn destroy(&self) { + timeline_of(&self.error_sink).drain(); self.device.destroy(); } @@ -516,9 +566,14 @@ pub unsafe extern "C" fn canvas_native_webgpu_device_pop_error_scope( } }; + // Errors from work queued inside the scope have to have been reported. + let timeline = timeline_of(&device.error_sink); + timeline.drain(); + let slot = device.error_sink.lock().open_scopes.pop(); + // An empty stack is a JS-visible OperationError; wgpu signals it as Err. let popped = match device.device.pop_error_scope() { - Ok(scope) => scope, + Ok(scope) => scope.or_else(|| slot.and_then(|slot| slot.take())), Err(_) => return report(CanvasGPUErrorType::None, std::ptr::null_mut()), }; @@ -544,10 +599,14 @@ pub unsafe extern "C" fn canvas_native_webgpu_device_push_error_scope( filter: CanvasGPUErrorFilter, ) { let device = &*device; - // wgpu-core owns the WebGPU error-scope stack now, and every infallible - // create_* reports into it, so this delegates rather than keeping a second - // stack that those calls would never reach. + // wgpu-core keeps a stack per thread: this one catches calls made on the JS + // thread, the mirror catches work run on the queue thread. device.device.push_error_scope(filter.into()); + device + .error_sink + .lock() + .open_scopes + .push(super::queue_thread::ScopeSlot::new(filter.into())); } #[no_mangle] @@ -565,18 +624,7 @@ pub unsafe extern "C" fn canvas_native_webgpu_device_set_uncaptured_error_callba error_sink.uncaptured_handler = UncapturedErrorCallback { callback, userdata }; } - let userdata = userdata as usize; - device.device.on_uncaptured_error(Arc::new(move |error: wgt::error::Error| { - let Some(callback) = callback else { return }; - let typ = match error { - wgt::error::Error::OutOfMemory { .. } => CanvasGPUErrorType::OutOfMemory, - wgt::error::Error::Validation { .. } => CanvasGPUErrorType::Validation, - _ => CanvasGPUErrorType::Internal, - }; - let msg = error.to_string().replace('\0', ""); - let msg = CString::new(msg).unwrap_or_else(|_| CString::new("webgpu error").unwrap()); - unsafe { callback(typ, msg.into_raw(), userdata as *mut c_void) }; - })); + install_uncaptured_handler(&device.device, &device.error_sink, callback, userdata); } #[no_mangle] diff --git a/crates/canvas-c/src/webgpu/gpu_native_texture.rs b/crates/canvas-c/src/webgpu/gpu_native_texture.rs index 692b16c1f..a47b2944f 100644 --- a/crates/canvas-c/src/webgpu/gpu_native_texture.rs +++ b/crates/canvas-c/src/webgpu/gpu_native_texture.rs @@ -420,6 +420,8 @@ pub(crate) unsafe fn blit_texture( destination_origin: (u32, u32), size: wgt::Extent3d, ) -> bool { + // Runs here, not queued: wait for the queued work it must follow. + super::gpu_device::timeline_of(&queue.error_sink).drain(); let device = &queue.device_id; let destination_descriptor = destination_texture.descriptor(); let destination_format = destination_descriptor.format; diff --git a/crates/canvas-c/src/webgpu/gpu_queue.rs b/crates/canvas-c/src/webgpu/gpu_queue.rs index 2770ae824..e0cfdd9ef 100644 --- a/crates/canvas-c/src/webgpu/gpu_queue.rs +++ b/crates/canvas-c/src/webgpu/gpu_queue.rs @@ -434,6 +434,8 @@ pub unsafe extern "C" fn canvas_native_webgpu_queue_copy_external_image_to_textu } let queue = &*queue; + // Runs here, not queued: wait for the queued work it must follow. + super::gpu_device::timeline_of(&queue.error_sink).drain(); let source = &*source; @@ -541,12 +543,16 @@ pub unsafe extern "C" fn canvas_native_webgpu_queue_copy_gpu_context_to_texture( let mut texture = None; if let Some(current_texture) = context.current_texture.lock().as_ref() { texture = Some(Arc::clone(¤t_texture.texture)); + } else if let Some(drawing) = context.drawing.lock().as_ref() { + texture = Some(Arc::clone(&drawing.texture)); } else if let Some(read_back_texture) = context.read_back_texture.lock().as_ref() { texture = Some(Arc::clone(&read_back_texture.texture)); } if let Some(texture) = texture { let queue = &*queue; + // Runs here, not queued: wait for the queued work it must follow. + super::gpu_device::timeline_of(&queue.error_sink).drain(); if let Some(data) = context.data.lock().as_ref() { let label = Cow::Borrowed("copyExternalImageToTexture:Encoder"); @@ -610,10 +616,16 @@ pub unsafe extern "C" fn canvas_native_webgpu_queue_on_submitted_work_done( callback(std::ptr::null_mut(), callback_data); }); - queue.queue.id.on_submitted_work_done(done); - // The closure only runs when the device is polled; without a frame loop (compute-only, - // headless) nothing else would. - super::gpu_buffer::poll_mappings(Arc::clone(&queue.queue.instance)); + // Registered on the queue thread, after the submits queued before it. + let target = Arc::clone(&queue.queue); + super::gpu_device::timeline_of(&queue.error_sink) + .get() + .enqueue(Box::new(move || { + target.id.on_submitted_work_done(done); + // The closure only runs when the device is polled; without a frame loop + // (compute-only, headless) nothing else would. + super::gpu_buffer::poll_mappings(Arc::clone(&target.instance)); + })); } #[no_mangle] @@ -628,22 +640,28 @@ pub unsafe extern "C" fn canvas_native_webgpu_queue_submit( let queue = &*queue; - let command_buffer_ids = std::slice::from_raw_parts(command_buffers, command_buffers_size) + let buffers = std::slice::from_raw_parts(command_buffers, command_buffers_size) .iter() - .map(|buffer| { - let buffer = &**buffer; - buffer + .map(|&buffer| { + (*buffer) .open .store(false, std::sync::atomic::Ordering::SeqCst); - // let mut id = buffer.command_buffer.lock(); - // id.take() - Arc::clone(&buffer.command_buffer) + Arc::increment_strong_count(buffer); + Arc::from_raw(buffer) }) .collect::>(); - queue.queue.id.submit(&command_buffer_ids); - - // command buffers are released by dropping the Vec + let target = Arc::clone(&queue.queue); + let scopes = super::gpu_device::scopes_of(&queue.error_sink); + super::gpu_device::timeline_of(&queue.error_sink) + .get() + .enqueue(Box::new(move || { + let command_buffers = buffers + .iter() + .map(|buffer| buffer.command_buffer()) + .collect::>(); + super::queue_thread::with_scopes(&scopes, || target.id.submit(&command_buffers)); + })); } unsafe fn write_buffer_size( @@ -671,15 +689,22 @@ unsafe fn write_buffer_size( None => &data[data_offset..], }; + // The page may reuse `data` once this returns, so the queue thread gets a copy. const ALIGNMENT: usize = wgt::COPY_BUFFER_ALIGNMENT as usize; let aligned_len = (data.len() + ALIGNMENT - 1) & !(ALIGNMENT - 1); - if aligned_len != data.len() { - let mut buf = vec![0u8; aligned_len]; - buf[..data.len()].copy_from_slice(data); - queue.queue.id.write_buffer(buffer_id, buffer_offset, &buf); - } else { - queue.queue.id.write_buffer(buffer_id, buffer_offset, data); - } + let mut bytes = Vec::with_capacity(aligned_len); + bytes.extend_from_slice(data); + bytes.resize(aligned_len, 0); + + let target = Arc::clone(&queue.queue); + let scopes = super::gpu_device::scopes_of(&queue.error_sink); + super::gpu_device::timeline_of(&queue.error_sink) + .get() + .record(Box::new(move || { + super::queue_thread::with_scopes(&scopes, || { + target.id.write_buffer(buffer_id, buffer_offset, &bytes) + }); + })); } #[no_mangle] @@ -760,7 +785,17 @@ pub unsafe extern "C" fn canvas_native_webgpu_queue_write_texture( let size: wgt::Extent3d = size.into(); - queue.queue.id.write_texture(destination, data, &data_layout, &size);} + let bytes = data.to_vec(); + let target = Arc::clone(&queue.queue); + let scopes = super::gpu_device::scopes_of(&queue.error_sink); + super::gpu_device::timeline_of(&queue.error_sink) + .get() + .record(Box::new(move || { + super::queue_thread::with_scopes(&scopes, || { + target.id.write_texture(destination, &bytes, &data_layout, &size) + }); + })); +} #[cfg(test)] mod tests { diff --git a/crates/canvas-c/src/webgpu/gpu_texture.rs b/crates/canvas-c/src/webgpu/gpu_texture.rs index 22bfa2bfd..4c9227d08 100644 --- a/crates/canvas-c/src/webgpu/gpu_texture.rs +++ b/crates/canvas-c/src/webgpu/gpu_texture.rs @@ -246,5 +246,7 @@ pub extern "C" fn canvas_native_webgpu_texture_destroy(texture: *const CanvasGPU } let texture = unsafe { &*texture }; + // After the queued work that may still use it. + super::gpu_device::timeline_of(&texture.error_sink).drain(); texture.texture.destroy(); } diff --git a/crates/canvas-c/src/webgpu/mod.rs b/crates/canvas-c/src/webgpu/mod.rs index 6a5edc67b..c057c02b8 100644 --- a/crates/canvas-c/src/webgpu/mod.rs +++ b/crates/canvas-c/src/webgpu/mod.rs @@ -75,4 +75,5 @@ pub mod linux_frames; #[cfg(target_os = "linux")] pub mod linux_video_frame; pub mod prelude; +pub(crate) mod queue_thread; pub mod structs; diff --git a/crates/canvas-c/src/webgpu/queue_thread.rs b/crates/canvas-c/src/webgpu/queue_thread.rs new file mode 100644 index 000000000..64e48e671 --- /dev/null +++ b/crates/canvas-c/src/webgpu/queue_thread.rs @@ -0,0 +1,245 @@ +//! The device timeline: queue work (finish, submit, uploads, present) runs in +//! order on one thread per device instead of in the JS call, as a browser's GPU +//! process does. Calls that need the queue's results first wait for it to drain. +//! +//! Callbacks into JS go through the JS thread's loop where there is one (Android, +//! Apple), else wait for the JS thread's next call into WebGPU. + +use std::cell::RefCell; +use std::sync::{mpsc, Arc, OnceLock}; + +use crate::thread_poster::ThreadPoster; + +type Job = Box; + +/// An error scope open on the JS thread. Work queued while it was open reports +/// into it from the queue thread. +#[derive(Debug)] +pub(crate) struct ScopeSlot { + filter: wgt::error::ErrorFilter, + error: parking_lot::Mutex>, +} + +impl ScopeSlot { + pub(crate) fn new(filter: wgt::error::ErrorFilter) -> Arc { + Arc::new(ScopeSlot { + filter, + error: parking_lot::Mutex::new(None), + }) + } + + pub(crate) fn take(&self) -> Option { + self.error.lock().take() + } +} + +/// The scopes open when a call was made. +#[derive(Clone, Default)] +pub(crate) struct ScopeSnapshot(pub(crate) Vec>); + +thread_local! { + static CURRENT_SCOPES: RefCell> = const { RefCell::new(None) }; +} + +fn filter_of(error: &wgt::error::Error) -> wgt::error::ErrorFilter { + match error { + wgt::error::Error::OutOfMemory { .. } => wgt::error::ErrorFilter::OutOfMemory, + wgt::error::Error::Validation { .. } => wgt::error::ErrorFilter::Validation, + _ => wgt::error::ErrorFilter::Internal, + } +} + +/// Runs `f` with errors reported into `scopes`. +pub(crate) fn with_scopes(scopes: &ScopeSnapshot, f: impl FnOnce() -> R) -> R { + let previous = CURRENT_SCOPES.with(|current| current.replace(Some(scopes.clone()))); + let result = f(); + CURRENT_SCOPES.with(|current| *current.borrow_mut() = previous); + result +} + +/// Stores `error` in the innermost matching scope of the running job. Hands it +/// back if there is none: it is uncaptured. +pub(crate) fn capture(error: wgt::error::Error) -> Result<(), wgt::error::Error> { + CURRENT_SCOPES.with(|current| { + let current = current.borrow(); + let Some(scopes) = current.as_ref() else { + return Err(error); + }; + let filter = filter_of(&error); + match scopes.0.iter().rev().find(|slot| slot.filter == filter) { + Some(slot) => { + let mut slot = slot.error.lock(); + if slot.is_none() { + *slot = Some(error); + } + Ok(()) + } + None => Err(error), + } + }) +} + +/// Display priority, like a browser's GPU thread: frames wait on this thread. +fn raise_priority() { + #[cfg(target_os = "android")] + unsafe { + // THREAD_PRIORITY_DISPLAY + libc::setpriority(libc::PRIO_PROCESS, libc::gettid() as libc::id_t, -4); + } + #[cfg(any(target_os = "ios", target_os = "visionos", target_os = "tvos"))] + unsafe { + // QOS_CLASS_USER_INTERACTIVE, as the main thread. + extern "C" { + fn pthread_set_qos_class_self_np(qos_class: u32, relative_priority: i32) -> i32; + } + pthread_set_qos_class_self_np(0x21, 0); + } +} + +pub(crate) struct QueueThread { + /// Work recorded since the last send: sent together, one wakeup per batch. + recorded: parking_lot::Mutex>, + sender: parking_lot::Mutex>>, + pending: Arc<(parking_lot::Mutex, parking_lot::Condvar)>, + poster: Option>, + js_thread: std::thread::ThreadId, + /// Callbacks for the JS thread when it has no loop to post to. + for_js: parking_lot::Mutex>, + worker: OnceLock, +} + +impl QueueThread { + fn new() -> Self { + QueueThread { + recorded: parking_lot::Mutex::new(Vec::new()), + sender: parking_lot::Mutex::new(None), + pending: Arc::default(), + poster: ThreadPoster::for_current_thread(), + js_thread: std::thread::current().id(), + for_js: parking_lot::Mutex::new(Vec::new()), + worker: OnceLock::new(), + } + } + + fn on_worker(&self) -> bool { + self.worker.get() == Some(&std::thread::current().id()) + } + + /// Runs `job` after everything queued before it, without waking the queue + /// thread: it goes with the next `enqueue`, `flush` or `drain`. + pub(crate) fn record(&self, job: Job) { + if self.on_worker() { + job(); + return; + } + self.deliver_to_js(); + self.recorded.lock().push(job); + } + + /// Runs `job` after everything queued before it. + pub(crate) fn enqueue(&self, job: Job) { + self.record(job); + self.flush(); + } + + /// Sends what was recorded. + pub(crate) fn flush(&self) { + let jobs = std::mem::take(&mut *self.recorded.lock()); + if jobs.is_empty() { + return; + } + self.send(Box::new(move || { + for job in jobs { + job(); + } + })); + } + + fn send(&self, job: Job) { + *self.pending.0.lock() += 1; + let mut sender = self.sender.lock(); + let sender = sender.get_or_insert_with(|| { + let (tx, rx) = mpsc::channel::(); + let pending = Arc::clone(&self.pending); + let handle = std::thread::Builder::new() + .name("webgpu-queue".into()) + .spawn(move || { + raise_priority(); + while let Ok(job) = rx.recv() { + job(); + let mut count = pending.0.lock(); + *count -= 1; + if *count == 0 { + pending.1.notify_all(); + } + } + }) + .expect("webgpu queue thread"); + let _ = self.worker.set(handle.thread().id()); + tx + }); + if let Err(mpsc::SendError(job)) = sender.send(job) { + *self.pending.0.lock() -= 1; + job(); + } + } + + /// Blocks until everything queued so far has run. + pub(crate) fn drain(&self) { + if self.on_worker() { + return; + } + self.flush(); + let mut count = self.pending.0.lock(); + while *count > 0 { + self.pending.1.wait(&mut count); + } + drop(count); + self.deliver_to_js(); + } + + /// Runs `task` on the JS thread. + pub(crate) fn post_to_js(&self, task: Job) { + if std::thread::current().id() == self.js_thread { + task(); + } else if let Some(poster) = self.poster.as_ref() { + poster.post(task); + } else { + self.for_js.lock().push(task); + } + } + + fn deliver_to_js(&self) { + if std::thread::current().id() != self.js_thread { + return; + } + let tasks = std::mem::take(&mut *self.for_js.lock()); + for task in tasks { + task(); + } + } +} + +/// Shared by every object of a device; the thread starts on first use, from the +/// JS thread whose loop callbacks are posted back to. +#[derive(Default)] +pub(crate) struct Timeline(OnceLock); + +impl std::fmt::Debug for Timeline { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("Timeline") + } +} + +impl Timeline { + pub(crate) fn get(&self) -> &QueueThread { + self.0.get_or_init(QueueThread::new) + } + + /// Waits for queued work, if the thread was ever started. + pub(crate) fn drain(&self) { + if let Some(thread) = self.0.get() { + thread.drain(); + } + } +}