diff --git a/crates/export/src/pipeline.rs b/crates/export/src/pipeline.rs index 8e8c90561..b0d1d96c9 100644 --- a/crates/export/src/pipeline.rs +++ b/crates/export/src/pipeline.rs @@ -391,11 +391,14 @@ impl<'a> Cx<'a> { } let t0 = Instant::now(); let r = self.renderer(); - let planes = r.active_accel()?.comp_frame_yuv420(&r, self.job.comp, t, comp.background)?; + let a = r.active_accel()?; + let planes = a.comp_frame_yuv420(&r, self.job.comp, t, comp.background)?; let (cw, ch) = (w.div_ceil(2) as usize, h.div_ceil(2) as usize); if planes.len() != w as usize * h as usize + 2 * cw * ch { return None; } + // A frame rendered off `passes::comp_frame` ends here; on a `None` the fallback renders and ends it. + a.frame_end(); if let Some(ft) = &self.frame_times { ft.lock().unwrap_or_else(|e| e.into_inner()).push((i, t0.elapsed().as_secs_f64())); } diff --git a/crates/export/tests/it/h264_gpu_yuv.rs b/crates/export/tests/it/h264_gpu_yuv.rs index 0f99a7b54..02963ef46 100644 --- a/crates/export/tests/it/h264_gpu_yuv.rs +++ b/crates/export/tests/it/h264_gpu_yuv.rs @@ -24,11 +24,21 @@ struct Fake { /// Frames asking for planes right now, and the most at once after the stop. asking: AtomicUsize, most_after_stop: AtomicUsize, + /// `Accelerator::frame_end` calls: one per rendered comp frame, planes or not. + frame_ends: AtomicUsize, } impl Fake { fn new(yuv: bool, deferred: bool, stop_after: usize) -> Self { - Fake { yuv, yuv_frames: AtomicUsize::new(0), deferred, stop_after, asking: AtomicUsize::new(0), most_after_stop: AtomicUsize::new(0) } + Fake { + yuv, + yuv_frames: AtomicUsize::new(0), + deferred, + stop_after, + asking: AtomicUsize::new(0), + most_after_stop: AtomicUsize::new(0), + frame_ends: AtomicUsize::new(0), + } } } @@ -54,6 +64,9 @@ impl Accelerator for Fake { } Some(effectcraft_raster::yuv::yuv420_over(&r.comp_frame_cpu(comp, t), bg)) } + fn frame_end(&self) { + self.frame_ends.fetch_add(1, Ordering::SeqCst); + } fn available(&self) -> bool { self.yuv_frames.load(Ordering::SeqCst) < self.stop_after } @@ -122,6 +135,8 @@ fn accelerator_planes_encode_like_the_cpu_conversion() { let b = run(&p, cid, Some(&rgba), &s, &d.join("rgba.mp4")); let c = run(&p, cid, None, &s, &d.join("cpu.mp4")); assert_eq!(yuv.yuv_frames.load(Ordering::SeqCst), 20, "every frame came from the accelerator"); + assert_eq!(yuv.frame_ends.load(Ordering::SeqCst), 20, "every planes frame ends once"); + assert_eq!(rgba.frame_ends.load(Ordering::SeqCst), 20, "every RGBA frame ends once"); assert!(a == b && b == c, "the three files differ"); let f = effectcraft_media::probe(&yuv_mp4).expect("probe yuv.mp4"); assert_eq!((f.width, f.height), (64, 48)); @@ -144,6 +159,7 @@ fn frames_the_accelerator_cannot_take_fall_back_to_rgba() { let a = run(&p, cid, Some(&yuv), &odd, &d.join("yuv.mp4")); let c = run(&p, cid, None, &odd, &d.join("cpu.mp4")); assert_eq!(yuv.yuv_frames.load(Ordering::SeqCst), 0); + assert_eq!(yuv.frame_ends.load(Ordering::SeqCst), 6, "every fallback frame ends once"); assert!(a == c); } diff --git a/crates/gpu/src/context.rs b/crates/gpu/src/context.rs index 2c9a40735..543c79283 100644 --- a/crates/gpu/src/context.rs +++ b/crates/gpu/src/context.rs @@ -117,6 +117,9 @@ impl Drop for Lease { if let Some(pool) = self.pool.upgrade() && let Ok(mut p) = pool.lock() { + // Always undo `pooled`'s leased count first: `Pool::discard` does not reset + // `leased`, and a stale-generation lease must still clear its contribution. + p.leased = p.leased.saturating_sub(tex_bytes(&self.texture)); if !self.escaped.load(std::sync::atomic::Ordering::Relaxed) { p.release_owned(self.texture.clone(), self.owner, self.generation); } else if self.generation == p.generation { @@ -148,6 +151,8 @@ pub(crate) struct Pool { open: BTreeSet, next: u64, bytes: usize, + /// Bytes currently held by live [`Lease`]s. + leased: usize, /// Working textures reused / newly allocated (diagnostics, tests). hits: u64, misses: u64, @@ -462,6 +467,44 @@ impl Pool { } } + /// Moves `owner`'s returned working textures onto the shared free list (dropping any over + /// budget) right after `owner`'s encoder has been submitted, while keeping `owner` registered. + fn flush_owned(&mut self, owner: u64) { + let Some(m) = self.owned.get_mut(&owner).map(std::mem::take) else { return }; + for t in m.into_values().flatten() { + self.make_free(t); + } + } + + /// Bytes the pool will still keep alive (live `Lease`s, other owners' in-flight working sets, + /// released textures still guarded by other open encoders, and up to `budget` on the shared + /// free list) once `owner` flushes its returned textures and `closing_enc` submits. + fn kept_after_flush(&self, owner: u64, closing_enc: Option) -> usize { + let oldest = self.open.iter().copied().find(|&o| Some(o) != closing_enc).unwrap_or(u64::MAX); + let mut ready = 0usize; + let mut still_released = 0usize; + for (t, w) in &self.released { + let b = tex_bytes(t); + if *w < oldest { + ready += b; + } else { + still_released += b; + } + } + let mut self_owned = 0usize; + let mut other_owned = 0usize; + for (&k, m) in &self.owned { + let b: usize = m.values().flatten().map(tex_bytes).sum(); + if k == owner { + self_owned += b; + } else { + other_owned += b; + } + } + let free_after = self.bytes.max((self.bytes + ready + self_owned).min(self.budget())); + self.leased + other_owned + still_released + free_after + } + /// Bytes of working textures kept for reuse: the free list, those waiting for open /// encoders and those kept by live `Enc`s. fn retained(&self) -> usize { @@ -595,6 +638,11 @@ pub struct GpuContext { /// Export frame → YUV 4:2:0 kernel (built on first use). pub(crate) yuv: std::sync::OnceLock>, readbacks: Arc, + /// Serializes `device.poll` across threads (`wgpu-core 30.0.1` can panic on concurrent + /// `Device::maintain` when two callers poll the same device at once). Shared by every + /// context on the same device. + poll_lock: Arc>, + memcap: crate::memcap::MemCap, } fn tex_entry(binding: u32) -> wgpu::BindGroupLayoutEntry { @@ -923,7 +971,9 @@ impl GpuContext { init_resource(&device, "dummy storage buffer", || buffer_init(&device, &queue, Some("dummy"), &[0u8; 16], wgpu::BufferUsages::STORAGE))?; let dummy_tex = init_resource(&device, "dummy texture view", || dummy.create_view(&Default::default()))?; let max_dim = device.limits().max_texture_dimension_2d; - let pool = Arc::new(Mutex::new(Pool { queue: Some(queue.clone()), ..Default::default() })); + let memcap = crate::memcap::MemCap::new(&info); + let pool_budget = if memcap.cap() > 0 { POOL_BUDGET.min((memcap.cap() / 4) as usize) } else { 0 }; + let pool = Arc::new(Mutex::new(Pool { queue: Some(queue.clone()), budget: pool_budget, ..Default::default() })); Ok(GpuContext { device, queue, @@ -955,6 +1005,8 @@ impl GpuContext { particle_states: Default::default(), deferred: None, readbacks, + poll_lock: Default::default(), + memcap, }) } @@ -988,10 +1040,25 @@ impl GpuContext { crate::readback::failure_notifier(&self.readbacks) } + /// Maximum time a blocking poll holds `poll_lock` in one slice so concurrent readbacks are not + /// starved behind a long `mem_gate` or readback wait. + const POLL_SLICE: std::time::Duration = std::time::Duration::from_millis(20); + /// All device polling routes through this failure bridge. pub(crate) fn poll_readbacks(&self, kind: wgpu::PollType) -> Result { self.readbacks.check()?; - self.device.poll(kind).map_err(|e| { + let res = { + let _guard = match kind { + wgpu::PollType::Poll => match self.poll_lock.try_lock() { + Ok(g) => g, + Err(std::sync::TryLockError::Poisoned(e)) => e.into_inner(), + Err(std::sync::TryLockError::WouldBlock) => return Ok(wgpu::PollStatus::Poll), + }, + _ => self.poll_lock.lock().unwrap_or_else(|e| e.into_inner()), + }; + self.device.poll(kind) + }; + res.map_err(|e| { let error = format!("GPU readback poll: {e}"); self.readbacks.retire(error.clone()); log::error!("{error}"); @@ -999,23 +1066,73 @@ impl GpuContext { }) } - /// Waits on `submission_index` up to `timeout` so wgpu frees resources dropped before that - /// submission. A timeout simply moves on without retiring the context or its readbacks; any - /// other poll error retires the readbacks through the failure path. + /// Waits on `submission_index` up to `timeout` in [`Self::POLL_SLICE`] increments so wgpu + /// frees resources dropped before that submission without holding `poll_lock` for seconds at a + /// time. A timeout simply moves on without retiring the context or its readbacks; any other + /// poll error retires the readbacks through the failure path. fn poll_wait(&self, submission_index: wgpu::SubmissionIndex, timeout: std::time::Duration) { if self.readbacks.check().is_err() { return; } - match self.device.poll(wgpu::PollType::Wait { submission_index: Some(submission_index), timeout: Some(timeout) }) { - Ok(_) | Err(wgpu::PollError::Timeout) => {} - Err(e) => { - let error = format!("GPU poll wait: {e}"); - self.readbacks.retire(error.clone()); - log::error!("{error}"); + let deadline = web_time::Instant::now() + timeout; + loop { + let remaining = deadline.saturating_duration_since(web_time::Instant::now()); + let slice = remaining.min(Self::POLL_SLICE); + let res = { + let _guard = self.poll_lock.lock().unwrap_or_else(|e| e.into_inner()); + self.device.poll(wgpu::PollType::Wait { submission_index: Some(submission_index.clone()), timeout: Some(slice) }) + }; + let done = match res { + Ok(_) => true, + Err(wgpu::PollError::Timeout) => false, + Err(e) => { + let error = format!("GPU poll wait: {e}"); + self.readbacks.retire(error.clone()); + log::error!("{error}"); + true + } + }; + if done || remaining <= slice { + break; } } } + /// Bytes held by wgpu's device allocator (`0` on backends that do not report). + pub(crate) fn allocated_bytes(&self) -> u64 { + self.device.generate_allocator_report().map_or(0, |r| r.total_allocated_bytes) + } + + /// Bytes held in long-lived or still-referenced GPU caches/leases that submitting and waiting + /// on `owner` (`closing_enc`) cannot free. + fn kept_bytes(&self, owner: u64, closing_enc: Option) -> u64 { + let uploads = self.uploads.lock().map(|u| u.bytes as u64).unwrap_or(0); + let pool = self.pool.lock().map(|p| p.kept_after_flush(owner, closing_enc) as u64).unwrap_or(0); + let staging: u64 = self.staging.lock().map(|s| s.iter().map(|(k, v)| k * v.len() as u64).sum()).unwrap_or(0); + let zeros: u64 = self.zeros.lock().map(|z| z.keys().map(|&(w, h)| w as u64 * h as u64 * 16).sum()).unwrap_or(0); + uploads + pool + staging + zeros + } + + #[cfg(test)] + pub(crate) fn memcap_waits(&self) -> u64 { + self.memcap.waits() + } + + #[cfg(test)] + pub(crate) fn memcap_warned(&self) -> bool { + self.memcap.warned() + } + + #[cfg(test)] + pub(crate) fn set_memcap_wait_timeout(&self, timeout: std::time::Duration) { + self.memcap.set_wait_timeout(timeout); + } + + #[cfg(test)] + pub(crate) fn lock_poll_for_test(&self) -> std::sync::MutexGuard<'_, ()> { + self.poll_lock.lock().unwrap_or_else(|e| e.into_inner()) + } + /// Start a bounded buffer mapping. Retired contexts reject new requests immediately. pub(crate) fn map_readback( &self, @@ -1050,25 +1167,52 @@ impl GpuContext { }); } - /// A bounded native readback: polling and receiving share one deadline. + /// A bounded native readback: polling and receiving share one deadline, started once + /// `poll_lock` is first acquired and polled in [`Self::POLL_SLICE`] slices so lock contention + /// never consumes the readback deadline or holds the mutex across the whole timeout. #[cfg(not(target_arch = "wasm32"))] pub(crate) fn read_buffer(&self, buffer: &wgpu::Buffer, row: usize, stride: usize, height: usize) -> Result, String> { - let deadline = std::time::Instant::now() + crate::readback::TIMEOUT; let (tx, rx) = std::sync::mpsc::channel(); self.map_readback(buffer, row, stride, height, move |r| { let _ = tx.send(r); }); + let mut deadline = None; loop { self.readbacks.check()?; - let remaining = deadline.saturating_duration_since(std::time::Instant::now()); + if let Ok(result) = rx.try_recv() { + return result; + } + let (remaining, slice, poll_res) = { + let _guard = self.poll_lock.lock().unwrap_or_else(|e| e.into_inner()); + self.readbacks.check()?; + let d = *deadline.get_or_insert_with(|| std::time::Instant::now() + crate::readback::TIMEOUT); + let remaining = d.saturating_duration_since(std::time::Instant::now()); + if remaining.is_zero() { + (remaining, std::time::Duration::ZERO, Ok(wgpu::PollStatus::Poll)) + } else { + let slice = remaining.min(Self::POLL_SLICE); + let res = self.device.poll(wgpu::PollType::Wait { submission_index: None, timeout: Some(slice) }); + (remaining, slice, res) + } + }; + if let Ok(result) = rx.try_recv() { + return result; + } if remaining.is_zero() { let error = "GPU readback receive timed out; context retired".to_string(); self.readbacks.retire(error.clone()); return Err(error); } - // Block until the GPU is done (bounded by the deadline: a timeout retires the - // context) rather than polling and sleeping, which adds latency to every readback. - self.poll_readbacks(wgpu::PollType::Wait { submission_index: None, timeout: Some(remaining) })?; + match poll_res { + Ok(_) => {} + Err(wgpu::PollError::Timeout) if remaining > slice => {} + Err(e) => { + let error = format!("GPU readback poll: {e}"); + self.readbacks.retire(error.clone()); + log::error!("{error}"); + return Err(error); + } + } // The callback may be running on another thread that polled the same device. match rx.recv_timeout(remaining.min(std::time::Duration::from_millis(1))) { Ok(result) => return result, @@ -1207,6 +1351,8 @@ impl GpuContext { readbacks: self.readbacks.clone(), software: self.software, yuv: std::sync::OnceLock::new(), + poll_lock: self.poll_lock.clone(), + memcap: self.memcap.fresh(), } } @@ -1253,7 +1399,16 @@ impl GpuContext { /// one is zeroed. fn pooled(&self, w: u32, h: u32, owner: u64) -> (GpuImage, bool) { let (w, h) = (w.max(1), h.max(1)); - let (reused, generation) = self.pool.lock().map(|mut p| (p.take_owned(owner, w, h).or_else(|| p.take(w, h)), p.generation)).unwrap_or((None, u64::MAX)); + let bytes = w as usize * h as usize * 16; + let (reused, generation) = self + .pool + .lock() + .map(|mut p| { + let t = p.take_owned(owner, w, h).or_else(|| p.take(w, h)); + p.leased += bytes; + (t, p.generation) + }) + .unwrap_or((None, u64::MAX)); let was_reused = reused.is_some(); let texture = reused.unwrap_or_else(|| self.texture(w, h)); let lease = Arc::new(Lease { texture: texture.clone(), pool: Arc::downgrade(&self.pool), owner, escaped: Default::default(), generation }); @@ -1461,6 +1616,7 @@ pub(crate) struct Enc<'g> { pub g: &'g GpuContext, /// The open command encoder and its id in the texture pool. enc: Option<(wgpu::CommandEncoder, u64)>, + last_submit: Option, pending: usize, /// A dispatch named a kernel the device doesn't have (see [`Enc::take_missing`]). missing: bool, @@ -1496,7 +1652,7 @@ impl Drop for Enc<'_> { impl<'g> Enc<'g> { pub fn new(g: &'g GpuContext) -> Enc<'g> { let owner = g.pool.lock().map(|mut p| p.register()).unwrap_or(0); - Enc { g, enc: None, pending: 0, missing: false, owner, lent: false } + Enc { g, enc: None, last_submit: None, pending: 0, missing: false, owner, lent: false } } /// An `Enc` for self-contained work (an effect chain): once it is dropped, its working @@ -1508,7 +1664,7 @@ impl<'g> Enc<'g> { /// effect's intermediates stayed allocated until the frame ended. pub(crate) fn lent(g: &'g GpuContext) -> Enc<'g> { let owner = g.pool.lock().map(|mut p| p.lend()).unwrap_or(0); - Enc { g, enc: None, pending: 0, missing: false, owner, lent: true } + Enc { g, enc: None, last_submit: None, pending: 0, missing: false, owner, lent: true } } /// `img` is used by this `Enc`: if another `Enc` made it, that one must not reuse it at once @@ -1693,7 +1849,7 @@ impl<'g> Enc<'g> { /// Submit recorded work. pub fn submit(&mut self) { if let Some((enc, id)) = self.enc.take() { - self.g.queue.submit([enc.finish()]); + self.last_submit = Some(self.g.queue.submit([enc.finish()])); if let Ok(mut p) = self.g.pool.lock() { p.close(id); } @@ -1701,6 +1857,31 @@ impl<'g> Enc<'g> { self.pending = 0; } + /// Submits and waits on this encoder's own submission when the device allocator is over the + /// memory cap and enough of that memory is reclaimable, so wgpu frees textures dropped so far. + pub fn mem_gate(&mut self) { + if !self.g.memcap.due() { + return; + } + let allocated = self.g.allocated_bytes(); + if allocated <= self.g.memcap.cap() { + let _ = self.g.memcap.should_wait(allocated, 0); + return; + } + let closing = self.enc.as_ref().map(|(_, id)| *id); + let kept = self.g.kept_bytes(self.owner, closing); + if !self.g.memcap.should_wait(allocated, kept) { + return; + } + self.submit(); + if let Ok(mut p) = self.g.pool.lock() { + p.flush_owned(self.owner); + } + let idx = self.last_submit.take().unwrap_or_else(|| self.g.queue.submit([])); + self.g.poll_wait(idx, self.g.memcap.wait_timeout()); + self.g.memcap.record_wait(self.g.allocated_bytes(), self.g.kept_bytes(self.owner, None)); + } + /// Read an image back to the CPU (submits and waits). `None` where readback is impossible. pub fn download(&mut self, img: &GpuImage) -> Option { self.uses(img); diff --git a/crates/gpu/src/effects.rs b/crates/gpu/src/effects.rs index 5a2b80f1c..b33c15e99 100644 --- a/crates/gpu/src/effects.rs +++ b/crates/gpu/src/effects.rs @@ -101,6 +101,7 @@ pub(crate) fn run_chain(e: &mut Enc, chain: &[FxStep], buf: &Buf, levels: Option e.take_missing(); for step in chain { b = apply(e, step.spec.id, &step.ctx, b)?; + e.mem_gate(); if e.take_missing() { return None; } @@ -158,6 +159,7 @@ impl effectcraft_render::FxTarget for GpuFx<'_, '_> { self.e.take_missing(); for step in steps { let Some(next) = apply(self.e, step.spec.id, &step.ctx, cur) else { return false }; + self.e.mem_gate(); // A kernel the device couldn't build: the effects render on the CPU (#613). if self.e.take_missing() { return false; diff --git a/crates/gpu/src/lib.rs b/crates/gpu/src/lib.rs index d0a688e49..1be095c3f 100644 --- a/crates/gpu/src/lib.rs +++ b/crates/gpu/src/lib.rs @@ -79,6 +79,7 @@ mod fx_tone; mod fx_transition; mod fx_vr; mod fx_warp; +mod memcap; mod ops; mod particles; mod readback; diff --git a/crates/gpu/src/memcap.rs b/crates/gpu/src/memcap.rs new file mode 100644 index 000000000..1c2ce6456 --- /dev/null +++ b/crates/gpu/src/memcap.rs @@ -0,0 +1,236 @@ +//! Memory back-pressure for frames in flight. +//! +//! wgpu destroys a texture or buffer we drop only once the submissions that used it have +//! completed, and pooled textures released while other frames' encoders are open stay pending +//! until then. With several 4K frames in flight, the frames record work faster than the GPU +//! retires it, and freed-but-not-destroyed memory runs the device out of memory (#879). Once the +//! allocator holds more than the cap and enough of that total is reclaimable (not held by the +//! upload cache, the texture pool, or live images), a frame submits its encoder and waits for its +//! own submission so wgpu frees what was dropped. Frames are not held back for memory that cannot +//! be reclaimed; a warning is logged once. +//! +//! wgpu has no safe query for the adapter's memory size (`max_buffer_size` is capped at `i32::MAX` +//! on Direct3D 12 and Linux/Mesa Vulkan), so the default cap is conservative per device type; +//! `EFFECTCRAFT_GPU_MEMCAP_MB` overrides it (0 = off). Backends without an allocator report +//! (Metal, WebGPU, GL) are never gated. + +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering::Relaxed}; +use std::time::Duration; + +/// Check the allocator only every this many gate calls (effect calls and layers): the report +/// lists every allocation, and between checks a frame adds at most a few textures. +const EVERY: u64 = 8; + +const GIB: u64 = 1 << 30; +const MIB: u64 = 1 << 20; + +/// Overrides the default cap: whole MiB, 0 turns the gate off. +const MEMCAP_ENV: &str = "EFFECTCRAFT_GPU_MEMCAP_MB"; + +pub(crate) struct MemCap { + cap: u64, + hysteresis: u64, + calls: AtomicU64, + waits: AtomicU64, + /// Reclaimable bytes remaining after the last wait: another wait fires only once reclaimable + /// memory grows by `hysteresis` above this floor (or drops below it first). + floor: AtomicU64, + /// Set after the first over-cap sample where non-reclaimable memory alone exceeds the cap. + warned: AtomicBool, + #[cfg(test)] + timeout_ns: AtomicU64, +} + +impl MemCap { + pub(crate) fn new(info: &wgpu::AdapterInfo) -> MemCap { + let env = std::env::var(MEMCAP_ENV).ok(); + Self::from_env(env.as_deref(), info.device_type) + } + + fn from_env(var: Option<&str>, device_type: wgpu::DeviceType) -> MemCap { + let cap = cap_from_env(var, device_type); + MemCap { + cap, + hysteresis: default_hysteresis(cap), + calls: AtomicU64::new(0), + waits: AtomicU64::new(0), + floor: AtomicU64::new(0), + warned: AtomicBool::new(false), + #[cfg(test)] + timeout_ns: AtomicU64::new(crate::readback::TIMEOUT.as_nanos() as u64), + } + } + + /// The same cap with fresh counters (a test context sharing another's device). + #[cfg(test)] + pub(crate) fn fresh(&self) -> MemCap { + MemCap { + cap: self.cap, + hysteresis: self.hysteresis, + calls: AtomicU64::new(0), + waits: AtomicU64::new(0), + floor: AtomicU64::new(0), + warned: AtomicBool::new(false), + timeout_ns: AtomicU64::new(self.timeout_ns.load(Relaxed)), + } + } + + pub(crate) fn cap(&self) -> u64 { + self.cap + } + + /// Whether to sample the allocator on this call. + pub(crate) fn due(&self) -> bool { + self.cap != 0 && self.calls.fetch_add(1, Relaxed).is_multiple_of(EVERY) + } + + /// Whether `allocated` bytes in the allocator with `kept` non-reclaimable bytes (upload + /// cache, pool free list, live frames) should trigger a submit and wait. + pub(crate) fn should_wait(&self, allocated: u64, kept: u64) -> bool { + if self.cap == 0 || allocated <= self.cap { + self.floor.store(0, Relaxed); + return false; + } + if kept > self.cap && !self.warned.swap(true, Relaxed) { + log::warn!( + target: "effectcraft_gpu", + "gpu: {} MiB held by live frames, the upload cache and the texture pool is over the {} MiB memory cap: waiting cannot free it, so frames are not held back for it ({MEMCAP_ENV} sets the cap)", + kept >> 20, + self.cap >> 20, + ); + } + let reclaimable = allocated.saturating_sub(kept); + let floor = self.floor.load(Relaxed); + if reclaimable < floor { + self.floor.store(reclaimable, Relaxed); + return false; + } + reclaimable >= floor.saturating_add(self.hysteresis) + } + + pub(crate) fn record_wait(&self, allocated_after: u64, kept_after: u64) { + self.floor.store(allocated_after.saturating_sub(kept_after), Relaxed); + let n = self.waits.fetch_add(1, Relaxed); + if n == 0 { + log::info!("gpu: allocator over the {} MiB memory cap, frames wait for the GPU to catch up", self.cap >> 20); + } + } + + pub(crate) fn wait_timeout(&self) -> Duration { + #[cfg(test)] + { + Duration::from_nanos(self.timeout_ns.load(Relaxed)) + } + #[cfg(not(test))] + { + crate::readback::TIMEOUT + } + } + + #[cfg(test)] + pub(crate) fn waits(&self) -> u64 { + self.waits.load(Relaxed) + } + + #[cfg(test)] + pub(crate) fn warned(&self) -> bool { + self.warned.load(Relaxed) + } + + #[cfg(test)] + pub(crate) fn set_wait_timeout(&self, timeout: Duration) { + self.timeout_ns.store(timeout.as_nanos() as u64, Relaxed); + } +} + +/// The cap from the `EFFECTCRAFT_GPU_MEMCAP_MB` value: whole MiB (0 turns the gate off), or the +/// default for `device_type` when unset, or (with a warning) when not a number or too large. +fn cap_from_env(var: Option<&str>, device_type: wgpu::DeviceType) -> u64 { + let Some(v) = var else { return default_cap(device_type) }; + match v.trim().parse::().ok().and_then(|mb| mb.checked_mul(MIB)) { + Some(cap) => cap, + None => { + let cap = default_cap(device_type); + log::warn!(target: "effectcraft_gpu", "{MEMCAP_ENV}={v:?} is not a whole number of MiB that fits in 64 bits: using {} MiB", cap / MIB); + cap + } + } +} + +/// Discrete cards: 3 GiB, low enough to fire on 4–8 GB cards with headroom for the driver and +/// desktop while leaving ~1 GiB above the default 1.5 GiB upload cache + 256 MB pool (larger +/// cards can raise it with `EFFECTCRAFT_GPU_MEMCAP_MB`). Integrated and software adapters share +/// system memory: 2 GiB. +fn default_cap(t: wgpu::DeviceType) -> u64 { + match t { + wgpu::DeviceType::DiscreteGpu => 3 * GIB, + _ => 2 * GIB, + } +} + +/// Minimum reclaimable bytes before a wait is worth draining an encoder (1/8 of the cap, +/// clamped between 1 MiB and 256 MiB, or 1/4 of caps smaller than 4 MiB). +fn default_hysteresis(cap: u64) -> u64 { + if cap == 0 { 0 } else { (cap / 8).clamp(MIB.min(cap / 4), 256 * MIB) } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn defaults_and_cadence() { + assert_eq!(default_cap(wgpu::DeviceType::DiscreteGpu), 3 * GIB); + assert_eq!(default_cap(wgpu::DeviceType::IntegratedGpu), 2 * GIB); + let m = MemCap::from_env(None, wgpu::DeviceType::IntegratedGpu); + assert_eq!(m.cap(), 2 * GIB); + let due = (0..2 * EVERY).filter(|_| m.due()).count(); + assert_eq!(due, 2); + let off = MemCap::from_env(Some("0"), wgpu::DeviceType::DiscreteGpu); + assert_eq!(off.cap(), 0); + assert!((0..EVERY).all(|_| !off.due())); + } + + #[test] + fn override_is_whole_mib_or_the_default() { + let cap = |v| MemCap::from_env(Some(v), wgpu::DeviceType::DiscreteGpu).cap(); + assert_eq!(cap("2048"), 2 * GIB); + assert_eq!(cap(" 8 "), 8 * MIB); + assert_eq!(cap("17592186044415"), 17592186044415 * MIB, "the largest value that fits"); + for bad in ["", "lots", "-1", "1.5", "17592186044416", "99999999999999999999"] { + assert_eq!(cap(bad), 3 * GIB, "{bad:?} falls back to the default"); + } + } + + #[test] + fn hysteresis_skips_non_reclaimable_and_unfreed_memory() { + let m = MemCap::from_env(Some("2048"), wgpu::DeviceType::IntegratedGpu); + // Upload cache + pool + live frames (2.15 GiB) sit over the 2 GiB cap with only 50 MiB + // reclaimable: waiting would free almost nothing, so the gate must stay quiet. + let kept = 2150 * MIB; + assert!(!m.should_wait(kept + 50 * MIB, kept)); + assert!(!m.should_wait(kept + 200 * MIB, kept)); + // Once 256 MiB of dropped textures pile up, the gate fires. + assert!(m.should_wait(kept + 256 * MIB, kept)); + // If a wait freed nothing (e.g. held by another thread's open encoder or timed out), + // the floor prevents firing again until another 256 MiB of reclaimable memory accumulates. + m.record_wait(kept + 256 * MIB, kept); + assert!(!m.should_wait(kept + 256 * MIB, kept)); + assert!(!m.should_wait(kept + 400 * MIB, kept)); + assert!(m.should_wait(kept + 512 * MIB, kept)); + // Once the other thread finishes and reclaimable memory drops, the floor drops with it. + assert!(!m.should_wait(kept + 16 * MIB, kept)); + assert!(m.should_wait(kept + (16 + 256) * MIB, kept)); + } + + #[test] + fn kept_memory_over_the_cap_warns_once_and_never_waits() { + let m = MemCap::from_env(Some("8"), wgpu::DeviceType::IntegratedGpu); + // Cap 8 MiB: kept alone over the cap with reclaimable under the hysteresis → no wait, warn once. + assert!(!m.should_wait(20 * MIB, 20 * MIB)); + assert!(m.warned()); + assert!(!m.should_wait(20 * MIB, 20 * MIB)); + assert!(!m.should_wait(20 * MIB, 19 * MIB + MIB / 2)); + assert!(m.should_wait(20 * MIB, 4 * MIB), "16 MiB reclaimable is over the hysteresis"); + } +} diff --git a/crates/gpu/src/tests.rs b/crates/gpu/src/tests.rs index 28c27b5c7..ef8673b5f 100644 --- a/crates/gpu/src/tests.rs +++ b/crates/gpu/src/tests.rs @@ -1586,3 +1586,214 @@ fn frames_outside_the_export_pipeline_age_the_pool() { assert!(ctx.pool_retained_bytes() <= budget); assert!(ctx.pool_evictions() > evicted0, "rendered frames did not age the pool: stale sizes were never evicted"); } + +/// #1550: with a small `EFFECTCRAFT_GPU_MEMCAP_MB` on a backend that reports allocator usage +/// (Vulkan, DX12), `Enc::mem_gate` stays quiet when the bytes over the cap are non-reclaimable +/// (live uploads), submits and waits when dropped working textures push reclaimable bytes over +/// the cap, carries on without retiring the context if a wait times out, warns once when a +/// frame's kept memory alone exceeds the cap, waits between two frames that fit only one at a +/// time, and recovers budget when a frame is dropped unfinished. +#[test] +fn mem_gate_reclaims_over_cap_and_tolerates_timeout() { + const ENV: &str = "EFFECTCRAFT_GPU_MEMCAP_TEST_CHILD"; + const SKIP: &str = "mem_gate_reclaims_over_cap_and_tolerates_timeout: skipped"; + if std::env::var_os(ENV).is_none() { + let exe = std::env::current_exe().expect("test binary path"); + let out = std::process::Command::new(exe) + .env(ENV, "1") + .env("EFFECTCRAFT_GPU_MEMCAP_MB", "8") + .args(["--exact", "tests::mem_gate_reclaims_over_cap_and_tolerates_timeout", "--nocapture"]) + .output() + .expect("spawn child test process"); + assert!( + out.status.success(), + "child test failed (status {:?}):\nstdout:\n{}\nstderr:\n{}", + out.status.code(), + String::from_utf8_lossy(&out.stdout), + String::from_utf8_lossy(&out.stderr), + ); + // The child's output is captured: pass its skip line on. + String::from_utf8_lossy(&out.stderr).lines().filter(|l| l.starts_with(SKIP)).for_each(|l| eprintln!("{l}")); + return; + } + + // EFFECTCRAFT_REQUIRE_GPU=1 (inherited from the parent) fails on a GPU that didn't start. + let g = Gpu::try_headless(); + let Some(g) = require_gpu("mem_gate child", g.as_ref()) else { + eprintln!("{SKIP}: no GPU ({})", g.as_ref().err().map_or("", String::as_str)); + return; + }; + let ctx = g.context(); + if ctx.allocated_bytes() == 0 { + // The backend has no allocator report (Metal, GL, WebGPU): the gate never engages. + eprintln!("{SKIP}: adapter {} reports no allocator usage", ctx.name); + return; + } + + // 1. Non-reclaimable memory (upload cache) over the 8 MiB cap must not trigger waits. + let img = pattern(7, 512, 512); // 4 MiB in RGBA f32 + let held: Vec<_> = (0..3) + .map(|i| { + let b = std::sync::Arc::new(effectcraft_effects::Buf { img: img.clone(), offset: [i as f64, 0.0], scale: 1.0 }); + ctx.upload_buf(&b).expect("upload"); + b + }) + .collect(); + let idx = ctx.queue.submit([]); + let _ = ctx.poll_readbacks(wgpu::PollType::Wait { submission_index: Some(idx), timeout: Some(std::time::Duration::from_secs(1)) }); + assert!(ctx.allocated_bytes() > (8 << 20), "uploads should put the allocator over the 8 MiB cap (got {} B)", ctx.allocated_bytes()); + { + let mut e = crate::context::Enc::new(ctx); + for _ in 0..16 { + e.mem_gate(); + } + } + assert_eq!(ctx.memcap_waits(), 0, "mem_gate must not wait when over-cap memory is held by live uploads"); + assert!(ctx.memcap_warned(), "uploads alone over the cap log the one-time warning"); + drop(held); + ctx.clear_uploads(); + ctx.pool_frame_end(); + + // 2. Dropped working textures over the 2 MiB pool budget and 8 MiB cap trigger a wait and + // free the dropped textures in wgpu's allocator. + let mut e = crate::context::Enc::new(ctx); + let mut peak_before_wait = 0u64; + for k in 0..16u32 { + let a = e.image(512 + k * 16, 256); // ~2 MiB each, distinct sizes + let b = e.image(512 + k * 16, 256); + e.copy_into(&a, &b, 0, 0); + drop((a, b)); + peak_before_wait = peak_before_wait.max(ctx.allocated_bytes()); + e.mem_gate(); + } + drop(e); + assert!(peak_before_wait > (8 << 20), "working textures should push allocator over 8 MiB before a wait (peak {} B)", peak_before_wait); + let waits = ctx.memcap_waits(); + assert!(waits > 0, "mem_gate should have waited once reclaimable bytes exceeded the cap"); + assert!( + ctx.allocated_bytes() < peak_before_wait, + "waiting should reclaim dropped textures (peak {} B, after {} B)", + peak_before_wait, + ctx.allocated_bytes() + ); + assert!(ctx.check_health().is_ok()); + ctx.pool_frame_end(); + + // 3. A frame bigger than the cap: live GpuImages count in Pool::leased (kept for every owner), + // so three live 512×512 RGBA f32 images (12 MiB) make kept alone exceed the 8 MiB cap. + // Waiting cannot free them; mem_gate must not add waits (warning already asserted in section 1). + let waits_before_big = ctx.memcap_waits(); + { + let mut e = crate::context::Enc::new(ctx); + let live = [e.image(512, 512), e.image(512, 512), e.image(512, 512)]; + for _ in 0..16 { + e.mem_gate(); + } + assert_eq!(ctx.memcap_waits(), waits_before_big, "a frame whose live images alone exceed the cap must not wait"); + assert!(ctx.check_health().is_ok()); + drop(live); + } + ctx.pool_frame_end(); + + // 4. Two frames that fit only one at a time: frame 1 creates ~6 MiB, copies, submits its + // work and drops its images, so wgpu holds them until the submission is done and polled. + // Frame 2 keeps ~6 MiB of new live images and waits for frame 1's reclaimable memory. + { + let mut e1 = crate::context::Enc::new(ctx); + let hold = [e1.image(512, 384), e1.image(512, 384)]; // 6 MiB + e1.copy_into(&hold[0], &hold[1], 0, 0); + e1.submit(); + drop((hold, e1)); + } + let allocated_after_f1 = ctx.allocated_bytes(); + let waits_before_pair = ctx.memcap_waits(); + let allocated_before_gate; + let downloaded; + { + let mut e2 = crate::context::Enc::new(ctx); + let a = e2.image(520, 384); + let b = e2.image(504, 384); // ~6 MiB together, sizes distinct from each other and frame 1 + e2.copy_into(&b, &a, 0, 0); + allocated_before_gate = ctx.allocated_bytes(); + assert!( + allocated_before_gate > (8 << 20), + "frame 1's unreclaimed memory ({allocated_after_f1} B) plus frame 2's live images should exceed the 8 MiB cap (got {allocated_before_gate} B)" + ); + for _ in 0..16 { + e2.mem_gate(); + } + assert!(ctx.memcap_waits() > waits_before_pair, "frame 2 should wait for frame 1's reclaimable memory (its own images stay live)"); + assert!( + ctx.allocated_bytes() < allocated_before_gate, + "waiting should reclaim frame 1's dropped textures (before {} B, after {} B)", + allocated_before_gate, + ctx.allocated_bytes() + ); + downloaded = e2.download(&a); + } + assert!(downloaded.is_some(), "frame 2 must finish with a correct download after waiting"); + ctx.pool_frame_end(); + + // 5. A failed / unfinished frame releases its budget: drop an Enc mid-flight, then a + // following frame must not wait once the pool has drained. + let baseline = ctx.allocated_bytes(); + { + let mut e = crate::context::Enc::new(ctx); + let _hold = [e.image(512, 512), e.image(512, 512), e.image(480, 480)]; + // Drop without finishing: same path as a frame that errors before submit. + } + ctx.pool_frame_end(); + let _ = ctx.poll_readbacks(wgpu::PollType::Poll); + assert!( + ctx.allocated_bytes() <= baseline + (2 << 20) + (1 << 20), + "dropped frame must release to baseline + pool budget (2 MiB) with slack (got {} B, baseline {} B)", + ctx.allocated_bytes(), + baseline + ); + let waits_after_drop = ctx.memcap_waits(); + { + let mut e = crate::context::Enc::lent(ctx); + for _ in 0..16 { + e.mem_gate(); + } + } + assert_eq!(ctx.memcap_waits(), waits_after_drop, "a following frame must not wait after a dropped frame released its memory"); + + // 6. A zero-timeout wait must not retire the context or break subsequent renders. + ctx.set_memcap_wait_timeout(std::time::Duration::ZERO); + { + let mut e = crate::context::Enc::lent(ctx); + for k in 0..16u32 { + let a = e.image(384 + k * 16, 256); + let b = e.image(384 + k * 16, 256); + e.copy_into(&a, &b, 0, 0); + drop((a, b)); + e.mem_gate(); + } + } + assert!(ctx.memcap_waits() > waits_after_drop, "mem_gate should have fired again under zero timeout"); + assert!(ctx.check_health().is_ok(), "a mem_gate wait timeout must carry on without retiring the context"); + let s = blend_scene(BitDepth::Bpc8, BlendMode::Screen); + let r = Renderer::new(&s.p, &Pattern, opts()); + assert!(g.render(&r, s.cid, Tick::ZERO).is_some(), "context renders normally after a zero-timeout mem_gate wait"); + + // 7. Non-blocking polls (`Gpu::poll`, `pool_frame_end`) skip when `poll_lock` is held by + // another thread, and `read_buffer` waits for `poll_lock` without consuming its deadline. + // Drain any dying pool textures first so the timed path takes Poll (try_lock) rather than Wait. + ctx.set_memcap_wait_timeout(crate::readback::TIMEOUT); + ctx.pool_frame_end(); + let (locked_tx, locked_rx) = std::sync::mpsc::channel(); + std::thread::scope(|sc| { + sc.spawn(|| { + let _guard = ctx.lock_poll_for_test(); + locked_tx.send(()).unwrap(); + std::thread::sleep(std::time::Duration::from_millis(60)); + }); + locked_rx.recv().unwrap(); + let t0 = std::time::Instant::now(); + g.poll(); + ctx.pool_frame_end(); + assert!(t0.elapsed() < std::time::Duration::from_millis(30), "non-blocking polls must not wait on a contended poll_lock (took {:?})", t0.elapsed()); + assert!(g.render(&r, s.cid, Tick::ZERO).is_some(), "readback waits for poll_lock and succeeds without retiring the context"); + }); +} diff --git a/crates/gpu/src/walk.rs b/crates/gpu/src/walk.rs index fdcb58dd4..c7212dc84 100644 --- a/crates/gpu/src/walk.rs +++ b/crates/gpu/src/walk.rs @@ -213,6 +213,12 @@ fn draw_collapsed<'a>( } fn draw_layer(e: &mut Enc, r: &Renderer, ctx: &EvalCtx, layer: &Layer, canvas: &mut GpuImage) -> Option<()> { + let out = draw_layer_on(e, r, ctx, layer, canvas); + e.mem_gate(); + out +} + +fn draw_layer_on(e: &mut Enc, r: &Renderer, ctx: &EvalCtx, layer: &Layer, canvas: &mut GpuImage) -> Option<()> { if layer.switches.adjustment { if draw_adjustment(e, r, ctx, layer, canvas).is_some() { return Some(()); diff --git a/crates/render/src/accel.rs b/crates/render/src/accel.rs index 9b7d29ec6..0d5fe1b6b 100644 --- a/crates/render/src/accel.rs +++ b/crates/render/src/accel.rs @@ -98,8 +98,9 @@ pub trait Accelerator: Send + Sync { fn pass_begin(&self) {} /// A rendered comp frame is done (the GPU pool ages sizes no frame asks for, and evicts /// only those once full). The multi-frame render paths call it once per comp frame: - /// [`crate::passes::comp_frame`] (exports; a field-rendered output frame is two), remote frames - /// and the preview workers. One-off renders (`Session::render`, single-frame export, + /// [`crate::passes::comp_frame`] (exports; a field-rendered output frame is two), the + /// export's 4:2:0 frames ([`Accelerator::comp_frame_yuv420`]), remote frames and the + /// preview workers. One-off renders (`Session::render`, single-frame export, /// thumbnails) don't, so they don't advance the pool's clock. Without it the pool never /// finds a stale size and keeps its first sizes, dropping newcomers. fn frame_end(&self) {}