//! Background frame rendering and the RAM preview cache. //! //! Frames render on a dedicated thread pool from immutable `Arc` snapshots, so the UI //! never waits for the compositor (a playing preview's frames and idle caching on one thread //! each, so the frames started first are done first). Results land in a memory-budgeted cache keyed by the comp's //! *content* ([`comp_content`]: the comp and everything it uses), frame, scale, view and render //! options; the timeline draws the cached range as the green cache bar. An edit keeps the frames //! of every comp it doesn't touch, and undo finds the frames of the state it returns to. //! Switches that don't change pixels (Audio, Lock, Shy, Hide Shy Layers) keep the frames too. //! When the budget is full the least recently shown frames go first. //! //! wasm32 has no threads: jobs wait in the same priority queue and [`Frames::pump`] hands them //! to a [`RemoteFrames`] (the browser's frame worker, a second engine instance fed with project //! diffs, `effectcraft_engine::remote`) or, without one, renders them on the UI thread between //! egui frames, within a time budget. //! //! With the disk cache on (Settings ▸ Media & Disk Cache), every CPU-rendered frame is also //! written to disk under a content key (the comp's content hash, frame, scale, view and render //! options), and a job first looks there: frames survive restarts and project reopenings. The //! timeline shows disk-only frames as a blue cache bar. //! //! With Mercury GPU Acceleration (a [`Gpu`] on egui-wgpu's device and the project's renderer set //! to the GPU) frames stay on the GPU: the compositor writes an RGBA8 texture that the viewer //! registers with egui-wgpu and draws directly; pixels are read back only when something needs //! them (Info panel, eyedroppers, histograms). use std::collections::{BTreeSet, HashMap, HashSet, VecDeque}; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; use effectcraft_engine::project::{Comp, ItemId, ItemKind, LayerSource, Node, Project, PropGroup}; use effectcraft_engine::render::disk_cache::{self, DiskCache}; use effectcraft_engine::render::{CancellableFootage, ExprHost, FootageSource, LayerCache, RenderOpts, Renderer}; use effectcraft_engine::time::Tick; use effectcraft_gpu::{DisplayFrame, FrameError, Gpu}; use rayon::prelude::*; /// Comp identities (and the project snapshots behind them) the RAM preview keeps frames of. const MAX_KEEPERS: usize = 64; /// GPU frames kept for RAM preview (bytes of video memory) at first. wgpu can't tell how much /// video memory there is: each time the device runs out, the budget halves (to /// [`MIN_GPU_BUDGET`]) and the frame renders on the CPU (#106). const GPU_BUDGET: usize = 1 << 30; const MIN_GPU_BUDGET: usize = 64 << 20; /// A rendered viewer frame: CPU pixels or a GPU texture. #[derive(Clone)] pub enum FrameImage { Cpu(Arc), Gpu(Arc), } impl FrameImage { pub fn size(&self) -> [usize; 2] { match self { FrameImage::Cpu(c) => c.size, FrameImage::Gpu(f) => [f.width as usize, f.height as usize], } } fn bytes(&self) -> usize { self.size()[0] * self.size()[1] * 4 } fn is_gpu(&self) -> bool { matches!(self, FrameImage::Gpu(_)) } } /// A viewer frame: its comp's content, the frame, scale, view and render options. `revision` is /// the project revision it was asked for at (job bookkeeping); it is not part of the identity, so /// a later revision with the same content finds the frame. #[derive(Clone, Copy, Debug)] pub struct FrameKey { pub revision: u64, /// The comp's content key ([`Frames::content_of`]). pub content: u64, pub comp: u64, pub frame: i64, /// Render scale × 1000. pub scale: u32, /// Hash of the 3D view camera (0 = the comp's active camera). pub view: u64, /// Hash of the other render options (Draft / Fast Previews, shadows, nested switches…). pub opts: u64, } impl FrameKey { /// Everything but the revision. fn id(&self) -> (u64, u64, i64, u32, u64, u64) { (self.content, self.comp, self.frame, self.scale, self.view, self.opts) } /// Whether `self` is a frame of the same comp content, scale, view and options as `base`. pub fn same_series(&self, base: &FrameKey) -> bool { FrameKey { frame: base.frame, ..*self } == *base } } impl PartialEq for FrameKey { fn eq(&self, o: &FrameKey) -> bool { self.id() == o.id() } } impl Eq for FrameKey {} impl std::hash::Hash for FrameKey { fn hash(&self, h: &mut H) { self.id().hash(h); } } /// Where a playing preview is and what it plays next (see [`Frames::set_playhead`]): the RAM /// preview renders the frames it reaches first before later ones, and keeps them over frames it /// reaches later or not at all (#595). #[derive(Clone, Copy, Debug)] pub struct Playhead { /// The frame series that plays ([`FrameKey::same_series`]; its `frame` is ignored). pub series: FrameKey, /// The frame on screen. pub cur: i64, /// The preview's range, skip step and looping (as in its `PreviewPlan`). pub start: i64, pub end: i64, pub step: i64, pub looping: bool, } impl Playhead { /// How many frames the preview shows before it reaches `k` (0: on screen); `None` for /// frames it doesn't reach (other series, outside the range, behind it without looping). pub fn ahead(&self, k: &FrameKey) -> Option { if !k.same_series(&self.series) || k.frame < self.start || k.frame > self.end { return None; } let step = self.step.max(1); if k.frame >= self.cur { return Some((k.frame - self.cur) / step); } // Looping: through the end of the range and around from its start. self.looping.then(|| (self.end.saturating_sub(self.cur) / step).saturating_add(1).saturating_add((k.frame - self.start) / step)) } } struct Cache { map: HashMap, order: VecDeque, bytes: usize, /// Bytes of GPU frames (also counted in `bytes`). gpu_bytes: usize, budget: usize, /// Bytes of GPU frames to keep at most ([`GPU_BUDGET`]). gpu_budget: usize, /// The playing preview, if any: evictions spare the frames it shows next. playhead: Option, /// Scale of the frame on screen (the newest urgent request): frames of other scales go first. scale: Option, } impl Cache { fn insert(&mut self, k: FrameKey, img: FrameImage) { let (sz, gpu) = (img.bytes(), img.is_gpu()); if let Some(old) = self.map.insert(k, img) { self.forget(&old); } else { self.order.push_back(k); } self.bytes += sz; if gpu { self.gpu_bytes += sz; } self.evict_to_budget(); self.evict_gpu(); } /// The frame to evict next: the least recently shown, but while a preview plays, the least /// recently shown of the frames it doesn't reach, then the one it reaches last. Plain LRU /// evicted the frames just ahead of the playhead first (rendered a while ago, not shown yet) /// to keep the ones just shown, so a preview longer than the cache played into a gap that /// prefetch kept refilling while it evicted the next frames (#595). Frames of another scale /// than the one on screen go first (a zoom or resolution change). fn victim(&self) -> Option { if let Some(s) = self.scale && let Some(i) = self.order.iter().position(|k| k.scale != s) { return Some(i); } let Some(p) = &self.playhead else { return (!self.order.is_empty()).then_some(0) }; let mut farthest: Option<(i64, usize)> = None; for (i, k) in self.order.iter().enumerate() { match p.ahead(k) { None => return Some(i), Some(d) if farthest.is_none_or(|(f, _)| d > f) => farthest = Some((d, i)), Some(_) => {} } } farthest.map(|(_, i)| i) } /// Video memory: drop the oldest GPU frames past their own budget, frames of another scale /// than the one on screen first (a zoom or resolution change). fn evict_gpu(&mut self) { while self.gpu_bytes > self.gpu_budget { let is_gpu = |k: &FrameKey| self.map.get(k).is_some_and(FrameImage::is_gpu); let other_scale = self.order.iter().position(|k| Some(k.scale) != self.scale && is_gpu(k)); let Some(i) = other_scale.or_else(|| self.order.iter().position(is_gpu)) else { break }; if let Some(old) = self.order.remove(i) { self.remove(&old); } } } /// The GPU ran out of memory: keep half as many GPU frames from now on. fn gpu_out_of_memory(&mut self) { self.gpu_budget = (self.gpu_budget / 2).max(MIN_GPU_BUDGET); self.evict_gpu(); } fn forget(&mut self, img: &FrameImage) { self.bytes -= img.bytes(); if img.is_gpu() { self.gpu_bytes -= img.bytes(); } } fn remove(&mut self, k: &FrameKey) { if let Some(i) = self.map.remove(k) { self.forget(&i); } } /// Shown again: last in line for eviction. fn touch(&mut self, k: &FrameKey) { if let Some(i) = self.order.iter().rposition(|o| o == k) && let Some(k) = self.order.remove(i) { self.order.push_back(k); } } /// Drop every frame of the identities in `gone`. fn drop_contents(&mut self, gone: &HashSet) { let keys: Vec = self.map.keys().filter(|k| gone.contains(&k.content)).copied().collect(); for k in &keys { self.remove(k); } self.order.retain(|k| !gone.contains(&k.content)); } /// Move the frames of identity `from` that `keep` accepts to identity `to` (in place in the /// eviction order), unless `to` already has them. fn carry(&mut self, from: u64, to: u64, keep: impl Fn(i64) -> bool) { for k in self.order.iter_mut().filter(|k| k.content == from && keep(k.frame)) { let moved = FrameKey { content: to, ..*k }; if !self.map.contains_key(&moved) && let Some(img) = self.map.remove(k) { self.map.insert(moved, img); *k = moved; } } } fn evict_to_budget(&mut self) { while self.bytes > self.budget { let Some(old) = self.victim().and_then(|i| self.order.remove(i)) else { break }; self.remove(&old); } } } /// What a render job needs (all cheap clones). #[derive(Clone)] pub struct RenderSource { pub project: Arc, pub footage: Arc, pub expr: Option>, pub layer_cache: Arc, /// GPU compositor on the viewer's device (Mercury GPU Acceleration), if available. pub gpu: Option, /// Keep GPU frames on the GPU for the viewer (native egui-wgpu textures). Off on the /// desktop: Show Channel / exposure / region-of-interest drawing read the CPU frame texture, /// so GPU frames are read back (the compositing still runs on the GPU). On in the browser /// while the viewer draws plainly, because there frames can't be read back synchronously /// (`EffectcraftApp::gpu_display`). pub gpu_display: bool, /// The persistent disk cache, when enabled. pub disk: Option>, } /// A [`comp_content`] identity and, when it was found in another project, that project /// ([`Frames::identity`]). type Identity = (u64, Option>); /// Content keys of (project revision, comp, virtual-file generation). The generation is one /// counter bumped when a browser file is replaced, so a lookup stays a map read. Hashing the /// new bytes happens only when that generation misses. type ContentKeys = Arc>>; #[cfg(not(target_arch = "wasm32"))] #[derive(Default)] struct ContentKeyJobs { running: bool, pending: VecDeque<(Arc, u64, u64)>, } fn content_key(keys: &ContentKeys, project: &Project, revision: u64, comp: u64, epoch: u64, virtual_id: Option>) -> u128 { if let Some(k) = keys.lock().ok().and_then(|m| m.get(&(revision, comp, epoch)).copied()) { return k; } let k = disk_cache::comp_content_key_with(project, ItemId(comp), virtual_id); if let Ok(mut m) = keys.lock() { if m.len() > 256 { m.clear(); } m.insert((revision, comp, epoch), k); } k } /// The identity of a comp's pixels in `project`, cheap enough to take at every revision: the /// addresses of the comps it draws (an edit replaces a comp's `Arc`: `Arc::make_mut` copies it /// while another snapshot holds it, and [`Frames::content_of`] keeps one for every identity it /// hands out), and by value the project settings and the footage, solids and proxies it uses. /// With expressions in its comps (which can read anything) every item counts. pub fn comp_content(project: &Project, comp: ItemId) -> u64 { comp_content_with(project, comp, project, |_, c| c) } /// [`comp_content`] with each comp's version selected for both its sources and its address. /// Sources removed by undo are read from `base`. fn comp_content_with<'a>(project: &'a Project, comp: ItemId, base: &'a Project, version: impl Fn(ItemId, &'a Arc) -> &'a Arc) -> u64 { use std::hash::{Hash, Hasher}; fn has_expression(g: &PropGroup) -> bool { g.children.iter().any(|c| match c { Node::Prop(p) => p.expr.is_some(), Node::Group(sub) => has_expression(sub), }) } let mut h = std::collections::hash_map::DefaultHasher::new(); format!("{:?}", project.settings).hash(&mut h); let item = |id| project.item(id).or_else(|| base.item(id)); let mut seen = BTreeSet::new(); let mut stack = vec![comp]; let mut expressions = false; while let Some(id) = stack.pop() { if !seen.insert(id) { continue; } let Some(ItemKind::Comp(c)) = item(id).map(|i| &i.kind) else { continue }; let c = version(id, c); for l in &c.layers { if let LayerSource::Comp { item } | LayerSource::Footage { item } | LayerSource::Solid { item } = l.source { stack.push(item); } expressions = expressions || has_expression(&l.props); } } let ids: Vec = if expressions { project.items.keys().copied().collect() } else { seen.into_iter().collect() }; for id in ids { let Some(it) = item(id) else { continue }; id.0.hash(&mut h); if expressions { // (`comp("Name")`, `footage("Name")`) it.name.hash(&mut h); } match &it.kind { ItemKind::Comp(c) => (Arc::as_ptr(version(id, c)) as usize).hash(&mut h), k => format!("{k:?}").hash(&mut h), } format!("{:?}", it.proxy).hash(&mut h); } comp.0.hash(&mut h); h.finish() } /// Hash of render options (part of [`FrameKey`] and of the disk key). The scale counts as the /// key's thousandths ([`FrameKey::scale`]): a series rebuilt from the key (the cache bars) has /// the same options as the frames rendered at 1/3 = 0.333… (#595). pub fn opts_hash(o: &RenderOpts) -> u64 { let o = RenderOpts { scale: (o.scale * 1000.0).round() / 1000.0, ..*o }; let mut h = disk_cache::Hash128::default(); h.write(format!("{o:?}").as_bytes()); h.finish() as u64 } /// Disk key of a viewer frame (render options included: region of interest, draft…). fn frame_disk_key( keys: &ContentKeys, project: &Project, key: &FrameKey, opts_hash: u64, epoch: u64, virtual_id: Option>, ) -> u128 { let content = content_key(keys, project, key.revision, key.comp, epoch, virtual_id); disk_cache::frame_key(content, key.frame, key.scale, key.view ^ opts_hash) } /// A frame for a [`RemoteFrames`] renderer. pub struct RemoteJob { pub project: Arc, pub revision: u64, pub comp: ItemId, pub t: Tick, pub opts: RenderOpts, /// The frame's disk-cache key, when the renderer keeps a disk cache /// ([`RemoteFrames::disk`]): served from it on a hit, stored there after a render. pub disk_key: Option, } /// Premultiplied RGBA8 pixels from a [`RemoteFrames`] renderer. pub struct RemoteFrame { pub width: u32, pub height: u32, pub rgba: Vec, /// Render time (ms). pub ms: f64, } /// Called once with a remote frame (or why there is none). pub type RemoteDone = Box) + Send>; /// Renders viewer frames elsewhere (the browser's frame worker), so the UI thread never does. pub trait RemoteFrames: Send + Sync { /// Frames it can render at the same time (0 when it is not usable: frames then render on /// the UI thread again). fn slots(&self) -> usize; /// Start rendering `job`; `done` runs when the frame arrives. fn start(&self, job: RemoteJob, done: RemoteDone); /// Drop cached intermediate results (Edit ▸ Purge). fn purge(&self) {} /// It renders GPU effects (its own GPU device): frames with effects go to it rather than /// the UI thread's GPU compositor, which would run their effects on the CPU. fn gpu(&self) -> bool { false } /// It keeps a disk cache of frames (the browser's Origin Private File System). fn disk(&self) -> bool { false } /// The disk cache holds the frame of this key (the blue cache bar). fn disk_contains(&self, _key: u128) -> bool { false } } /// Whether a comp (or a comp it nests) has a layer with enabled effects. pub fn comp_has_effects(project: &Project, comp: ItemId) -> bool { fn walk(p: &Project, comp: ItemId, depth: usize) -> bool { let Some(c) = p.comp(comp) else { return false }; c.layers.iter().any(|l| { l.switches.effects && l.effects().is_some_and(|fx| fx.groups().any(|g| g.enabled)) || depth < 16 && matches!(l.source, effectcraft_engine::project::LayerSource::Comp { item } if walk(p, item, depth + 1)) }) } walk(project, comp, 0) } pub struct Frames { cache: Arc>, /// Keys queued or rendering. inflight: Arc>>, /// Pending jobs, picked by priority when a pool thread frees up (the viewer's frame first). queue: Arc>, #[cfg(not(target_arch = "wasm32"))] /// None if the OS refused the threads: jobs then go to rayon's global pool. pool: Option, #[cfg(not(target_arch = "wasm32"))] /// One thread each, for a playing preview's frames and idle caching (see [`Queue::idle_lanes`]). Empty if /// the OS refused the threads: those frames then render on `pool` too. lanes: Vec, ctx: Option, /// Render time of the last viewer (urgent) frame in ms, for the Info panel / perf readout. pub last_ms: Arc>, content_keys: ContentKeys, /// Bumped when a virtual file is replaced. Lookups key on it instead of statting footage. virtual_epoch: Arc, /// Size and content hash for paths with no native metadata. Called only on a cache miss. virtual_id: Option Option<(u64, u64)> + Send + Sync>>, #[cfg(not(target_arch = "wasm32"))] content_key_jobs: Arc>, gpu_retired: Arc, /// [`comp_content`] identities by (revision, comp), with the project that keeps an identity /// found by [`Frames::identity`] in another project. identities: Mutex>, /// The last new identity of each comp and its project (see [`Frames::identity`]). bases: Mutex)>>, /// A project snapshot for each identity handed out, with when it was last asked for (see /// [`Frames::content_of`]); the counter. keepers: Mutex<(u64, HashMap)>)>, /// Where frames render instead of the UI thread (wasm32), and how many it is rendering. remote: Option>, remote_busy: Arc>, } impl Default for Frames { fn default() -> Self { #[cfg(not(target_arch = "wasm32"))] let threads = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(4).clamp(2, 16); #[cfg(not(target_arch = "wasm32"))] let lanes: Vec = (0..threads) .map_while(|i| { rayon::ThreadPoolBuilder::new() .num_threads(1) .thread_name(move |_| format!("ec-preview-{i}")) .panic_handler(|_| log::error!("a frame render panicked; that frame was skipped")) .build() .ok() }) .collect(); #[cfg(not(target_arch = "wasm32"))] let queue = Queue { idle_lanes: (0..lanes.len()).collect(), lanes: lanes.len(), ..Default::default() }; #[cfg(target_arch = "wasm32")] let queue = Queue::default(); Frames { cache: Arc::new(Mutex::new(Cache { map: HashMap::new(), order: VecDeque::new(), bytes: 0, gpu_bytes: 0, budget: 3 << 30, gpu_budget: GPU_BUDGET, playhead: None, scale: None })), inflight: Arc::new(Mutex::new(HashSet::new())), queue: Arc::new(Mutex::new(queue)), #[cfg(not(target_arch = "wasm32"))] // A panicking job would abort the whole process (rayon's default for `spawn`): log it // and lose only that frame (the panic hook has logged the message). pool: rayon::ThreadPoolBuilder::new() .num_threads(threads) .thread_name(|i| format!("ec-frame-{i}")) .panic_handler(|_| log::error!("a frame render panicked; that frame was skipped")) .build() .ok(), #[cfg(not(target_arch = "wasm32"))] lanes, ctx: None, last_ms: Arc::new(Mutex::new(0.0)), content_keys: Arc::default(), virtual_epoch: Arc::new(AtomicU64::new(0)), virtual_id: None, #[cfg(not(target_arch = "wasm32"))] content_key_jobs: Arc::default(), gpu_retired: Arc::default(), identities: Mutex::default(), bases: Mutex::default(), keepers: Mutex::default(), remote: None, remote_busy: Arc::default(), } } } /// Premultiplied f32 → egui premultiplied Color32 (alpha kept, so the viewer can show the /// transparency grid or the comp background colour underneath). pub fn to_color_image(img: &effectcraft_engine::render::Image) -> egui::ColorImage { let px: Vec = img.data.par_iter().map(|p| display_pixel(*p)).collect(); egui::ColorImage::new([img.width as usize, img.height as usize], px) } pub(crate) fn display_pixel(pixel: [f32; 4]) -> egui::Color32 { let alpha = if pixel[3].is_nan() { 0.0 } else { pixel[3].clamp(0.0, 1.0) }; let component = |value: f32| (value.clamp(0.0, alpha) * 255.0 + 0.5) as u8; egui::Color32::from_rgba_premultiplied(component(pixel[0]), component(pixel[1]), component(pixel[2]), (alpha * 255.0 + 0.5) as u8) } /// The whole factor an image of `size` is averaged down by for a texture: at least `min_k`, and /// the smallest that fits its sides in `max_side` (frames of comps wider than the GPU's texture /// limit, #201). pub fn fit_factor(size: [usize; 2], max_side: usize, min_k: usize) -> usize { size[0].max(size[1]).div_ceil(max_side.max(1)).max(min_k).max(1) } /// The size `size` is shown at in a texture whose sides can't exceed `max_side`. pub fn fitted_size(size: [usize; 2], max_side: usize) -> [usize; 2] { let k = fit_factor(size, max_side, 1); [size[0].div_ceil(k), size[1].div_ceil(k)] } /// The texture coordinates of the image within a texture averaged down by `k`: its last texels /// cover fewer pixels than the factor, so drawn over the image's area with these the texels sit /// exactly where their pixels are. pub fn fitted_uv(size: [usize; 2], k: usize) -> egui::Rect { let k = k.max(1); let f = |n: usize| (n as f64 / (n.div_ceil(k).max(1) * k) as f64) as f32; egui::Rect::from_min_max(egui::pos2(0.0, 0.0), egui::pos2(f(size[0]), f(size[1]))) } /// `img` as a texture the renderer accepts: unchanged when it fits in `max_side`, else averaged /// down by a whole factor (premultiplied pixels average correctly). The viewer draws it over the /// same area, so only detail beyond the GPU's texture limit is lost, instead of the upload /// failing (#201). pub fn fit_texture(img: egui::ColorImage, max_side: usize) -> egui::ColorImage { fit_texture_by(img, max_side, 1) } /// [`fit_texture`] averaging down by at least `min_k` (see [`fit_factor`]). pub fn fit_texture_by(img: egui::ColorImage, max_side: usize, min_k: usize) -> egui::ColorImage { let k = fit_factor(img.size, max_side, min_k); if k == 1 { return img; } average_down(&img, k) } /// A shared frame as texture data (see [`fit_texture_by`]): the frame itself when it needs no /// averaging, without copying its pixels. The viewer puts every frame a preview shows on screen; /// copying a 4K frame first held the UI thread for about a millisecond each time (#595). pub fn fit_shared_texture_by(img: &Arc, max_side: usize, min_k: usize) -> egui::ImageData { let k = fit_factor(img.size, max_side, min_k); if k == 1 { return egui::ImageData::Color(img.clone()); } average_down(img, k).into() } /// `img` averaged down by `k` (rounded up at the edges). fn average_down(img: &egui::ColorImage, k: usize) -> egui::ColorImage { let [w, h] = img.size; let (fw, fh) = (w.div_ceil(k), h.div_ceil(k)); let px: Vec = (0..fw * fh) .into_par_iter() .map(|i| { let (x0, y0) = ((i % fw) * k, (i / fw) * k); let (mut sum, mut n) = ([0u64; 4], 0u64); for y in y0..(y0 + k).min(h) { for c in img.pixels.get(y * w + x0..y * w + (x0 + k).min(w)).unwrap_or_default() { for (s, v) in sum.iter_mut().zip(c.to_array()) { *s += v as u64; } n += 1; } } let n = n.max(1); let [r, g, b, a] = sum.map(|s| ((s + n / 2) / n) as u8); egui::Color32::from_rgba_premultiplied(r, g, b, a) }) .collect(); egui::ColorImage::new([fw, fh], px) } /// The largest texture side the renderer accepts. pub fn max_texture_side(ctx: &egui::Context) -> usize { ctx.input(|i| i.max_texture_side) } /// `ctx.load_texture` for frame-sized images: fitted to the renderer's texture limit first. pub fn load_fitted(ctx: &egui::Context, name: impl Into, img: egui::ColorImage, opts: egui::TextureOptions) -> egui::TextureHandle { ctx.load_texture(name, fit_texture(img, max_texture_side(ctx)), opts) } impl Frames { /// Retire only GPU preview state; documents, CPU frames and running job leases survive. /// Publication checks the same latch while holding the cache lock, so a late GPU frame /// cannot restore an invalid texture after this removal. pub fn retire_gpu(&self) { self.gpu_retired.store(true, Ordering::Release); if let Ok(mut cache) = self.cache.lock() { let keys: Vec<_> = cache.map.iter().filter_map(|(key, image)| image.is_gpu().then_some(*key)).collect(); for key in keys { cache.remove(&key); } let Cache { map, order, .. } = &mut *cache; order.retain(|key| map.contains_key(key)); } if let Some(ctx) = &self.ctx { ctx.request_repaint(); } } /// Actual native frame-worker capacity, including the OS fallback pool. #[cfg(not(target_arch = "wasm32"))] pub fn parallelism(&self) -> usize { self.pool.as_ref().map_or_else(rayon::current_num_threads, rayon::ThreadPool::current_num_threads) } #[cfg(target_arch = "wasm32")] pub fn parallelism(&self) -> usize { self.remote_slots().max(1) } pub fn set_context(&mut self, ctx: &egui::Context) { if self.ctx.is_none() { self.ctx = Some(ctx.clone()); } } /// The egui context (to request repaints from callbacks). pub fn context(&self) -> Option { self.ctx.clone() } /// The cached frame, which becomes the most recently used. pub fn get(&self, k: &FrameKey) -> Option { let mut c = self.cache.lock().ok()?; let img = c.map.get(k).cloned()?; c.touch(k); Some(img) } /// Prefer the exact requested frame. While an edit renders, advance to the newest completed /// revision of the same comp, time, scale, view and render options instead of freezing on the /// image from before the drag. Input may arrive before the viewer draws on every UI frame, /// so even a fast worker cannot always finish the *latest* revision in time. /// /// Never move backwards when workers finish out of order, and never substitute another time /// or view while scrubbing or switching comps. The exact content lookup above also means an /// undo can immediately reuse an old cache entry, regardless of its original revision. pub fn get_for_viewer(&self, wanted: &FrameKey, shown: Option<&FrameKey>) -> Option<(FrameKey, FrameImage)> { if let Some(img) = self.get(wanted) { return Some((*wanted, img)); } let shown = shown?; if shown.comp != wanted.comp || shown.frame != wanted.frame || shown.view != wanted.view || shown.revision >= wanted.revision { return None; } let mut c = self.cache.lock().ok()?; let newest = c .map .keys() .filter(|k| { k.revision > shown.revision && k.revision < wanted.revision && k.comp == wanted.comp && k.frame == wanted.frame && k.scale == wanted.scale && k.view == wanted.view && k.opts == wanted.opts }) .max_by_key(|k| k.revision) .copied()?; let img = c.map.get(&newest).cloned()?; c.touch(&newest); Some((newest, img)) } /// The [`comp_content`] identity of `comp` in `project` at `revision` (taken once per /// revision). The project is kept while frames of that identity are cached or rendering, so /// the comp addresses it hashes can't be reused by other comps meanwhile. pub fn content_of(&self, project: &Arc, revision: u64, comp: ItemId) -> u64 { let known = self.identities.lock().ok().and_then(|m| m.get(&(revision, comp.0)).cloned()); let (id, base) = known.unwrap_or_else(|| { let found = self.identity(project, comp); if let Ok(mut m) = self.identities.lock() { if m.len() > 256 { m.clear(); } m.insert((revision, comp.0), found.clone()); } found }); if let Ok(mut guard) = self.keepers.lock() { let (clock, k) = &mut *guard; *clock += 1; let now = *clock; k.entry(id).or_insert_with(|| (now, base.unwrap_or_else(|| project.clone()))).0 = now; if k.len() > MAX_KEEPERS { let inflight: HashSet = self.inflight.lock().map(|s| s.iter().map(|f| f.content).collect()).unwrap_or_default(); if let Ok(mut c) = self.cache.lock() { // Snapshots of identities without frames go; past the limit the least // recently asked for go too, with their frames (a long drag leaves many). let cached: HashSet = c.map.keys().map(|f| f.content).collect(); k.retain(|i, _| *i == id || cached.contains(i) || inflight.contains(i)); if k.len() > MAX_KEEPERS { let mut ages: Vec<(u64, u64)> = k.iter().filter(|(i, _)| **i != id && !inflight.contains(i)).map(|(i, (t, _))| (*t, *i)).collect(); ages.sort_unstable(); let gone: HashSet = ages.into_iter().take(k.len().saturating_sub(MAX_KEEPERS / 2)).map(|(_, i)| i).collect(); k.retain(|i, _| !gone.contains(i)); c.drop_contents(&gone); } } } } id } /// The [`comp_content`] identity of `comp` in `project`, or the comp's last new identity when /// every comp it draws is the same or draws the same pixels there ([`Comp::same_pixels`]: /// only Audio, Lock or Shy switches changed), so toggling those keeps the cached frames /// (#103). That identity hashes the addresses of the comps in its project, which is /// returned to keep them. fn identity(&self, project: &Arc, comp: ItemId) -> Identity { let plain = comp_content(project, comp); let Ok(mut bases) = self.bases.lock() else { return (plain, None) }; // Markers are pixels only to the expressions that read them (#1083). let markers = project.reads_markers(); if let Some((base_id, base)) = bases.get(&comp.0) && *base_id != plain && comp_content_with(project, comp, project, |id, c| match base.item(id).map(|i| &i.kind) { Some(ItemKind::Comp(b)) if Arc::ptr_eq(b, c) || b.same_pixels_with(c, markers) => b, _ => c, }) == *base_id { return (*base_id, Some(base.clone())); } // Only layers of this comp changed: its frames outside their span (and a frame either // side for rounding) keep drawing the same pixels, so a cut or trim recaches only the // layers it touched (#1641). if let Some((base_id, base)) = bases.get(&comp.0) && let (Some(old), Some(new)) = (base.comp(comp), project.comp(comp)) && comp_content_with(project, comp, base, |id, c| match base.item(id).map(|i| &i.kind) { Some(ItemKind::Comp(b)) if id == comp || Arc::ptr_eq(b, c) || b.same_pixels_with(c, markers) => b, _ => c, }) == *base_id && let Some((s, e)) = new.changed_span(old) && let Ok(mut cache) = self.cache.lock() { let (first, last) = (new.frame_rate.frame_at(s).saturating_sub(1), new.frame_rate.frame_at(e).saturating_add(1)); cache.carry(*base_id, plain, |f| f < first || f > last); } if bases.len() > MAX_KEEPERS { bases.clear(); } bases.insert(comp.0, (plain, project.clone())); (plain, None) } pub fn is_cached(&self, k: &FrameKey) -> bool { self.cache.lock().map(|c| c.map.contains_key(k)).unwrap_or(false) } pub fn inflight(&self) -> usize { self.inflight.lock().map(|s| s.len()).unwrap_or(0) } /// Cached frame numbers of the series of `base` (its comp content, scale, view and render /// options) — for the cache bar. pub fn cached_frames(&self, base: &FrameKey) -> Vec { let Ok(c) = self.cache.lock() else { return vec![] }; let mut v: Vec = c.map.keys().filter(|k| k.same_series(base)).map(|k| k.frame).collect(); v.sort_unstable(); v } /// Frames of the series of `base` in the disk cache but not in RAM — the blue cache bar. /// `opts` are the viewer's render options (part of the disk key). Native queries return /// no blue frames until the content key is ready; hashing never blocks timeline painting. pub fn disk_frames(&self, src: &RenderSource, base: &FrameKey, range: std::ops::Range, opts: &RenderOpts) -> Vec { let start = range.start.clamp(0, 100_000); let end = range.end.clamp(0, 100_000); if start >= end { return vec![]; } let remote = self.remote.as_ref().filter(|r| r.disk()); let contains = |k: u128| match (&src.disk, remote) { (Some(dc), _) => dc.contains(disk_cache::Kind::Frame, k), (None, Some(r)) => r.disk_contains(k), (None, None) => false, }; if src.disk.is_none() && remote.is_none() { return vec![]; } let epoch = self.virtual_epoch.load(Ordering::Relaxed); #[cfg(not(target_arch = "wasm32"))] let content = { let known = self.content_keys.lock().unwrap_or_else(std::sync::PoisonError::into_inner).get(&(base.revision, base.comp, epoch)).copied(); let Some(content) = known else { self.request_content_key(src.project.clone(), base.revision, base.comp); return vec![]; }; content }; #[cfg(target_arch = "wasm32")] let content = content_key(&self.content_keys, &src.project, base.revision, base.comp, epoch, self.virtual_id.as_deref()); let ram: HashSet = self.cached_frames(base).into_iter().collect(); let oh = opts_hash(opts); (start..end).filter(|f| !ram.contains(f)).filter(|f| contains(disk_cache::frame_key(content, *f, base.scale, base.view ^ oh))).collect() } #[cfg(not(target_arch = "wasm32"))] fn request_content_key(&self, project: Arc, revision: u64, comp: u64) { let mut jobs = self.content_key_jobs.lock().unwrap_or_else(std::sync::PoisonError::into_inner); // Coalesce rapid edits. One worker hashes at most one revision at a time; retained // snapshots are bounded even with multiple timelines open. jobs.pending.retain(|(_, _, c)| *c != comp); if jobs.pending.len() >= 8 { jobs.pending.pop_front(); } jobs.pending.push_back((project, revision, comp)); if jobs.running { return; } jobs.running = true; drop(jobs); let (jobs, keys, ctx) = (self.content_key_jobs.clone(), self.content_keys.clone(), self.ctx.clone()); let (epoch, id) = (self.virtual_epoch.clone(), self.virtual_id.clone()); let work = move || loop { let next = { let mut jobs = jobs.lock().unwrap_or_else(std::sync::PoisonError::into_inner); let next = jobs.pending.pop_back(); if next.is_none() { jobs.running = false; } next }; let Some((project, revision, comp)) = next else { return }; let epoch = epoch.load(Ordering::Relaxed); let id = id.clone(); if std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| content_key(&keys, &project, revision, comp, epoch, id.as_deref()))).is_err() { log::error!("could not compute a preview disk-cache key; the cache indicator will retry"); } if let Some(ctx) = &ctx { ctx.request_repaint(); } }; match &self.pool { Some(pool) => pool.spawn(work), None => rayon::spawn(work), } } /// Change the RAM preview cache budget (Settings ▸ Memory & CPU), evicting the oldest /// frames when it shrinks. pub fn set_budget(&self, bytes: usize) { if let Ok(mut c) = self.cache.lock() { c.budget = bytes; c.evict_to_budget(); } } pub fn budget(&self) -> usize { self.cache.lock().map(|c| c.budget).unwrap_or(0) } pub fn clear(&self) { if let Some(r) = &self.remote { r.purge(); } if let Ok(mut c) = self.cache.lock() { c.map.clear(); c.order.clear(); c.bytes = 0; c.gpu_bytes = 0; } // Disk keys hash the footage files' size and time, but are remembered per revision; a // file reloaded from disk without a new revision (a re-rendered image sequence) must be // keyed again, or the disk cache serves the frames from before. if let Ok(mut k) = self.content_keys.lock() { k.clear(); } } /// The playing preview's position (`None` when no preview plays): prefetch renders the /// frames it reaches first before the others, evictions spare the frames it shows next, and /// queued prefetch frames of its series it no longer reaches are dropped (#595). pub fn set_playhead(&self, playhead: Option) { if let Ok(mut c) = self.cache.lock() { c.playhead = playhead; } let Ok(mut q) = self.queue.lock() else { return }; // A preview starting or stopping moves the queued prefetch frames between the lanes and // the pool (see [`Queue::idle_lanes`]). #[cfg(not(target_arch = "wasm32"))] let moved = q.playhead.is_some() != playhead.is_some(); q.playhead = playhead; if let Some(p) = playhead && let Ok(mut inf) = self.inflight.lock() { q.jobs.retain(|j| { let gone = !j.urgent && j.key.same_series(&p.series) && p.ahead(&j.key).is_none(); if gone { inf.remove(&j.key); } !gone }); } #[cfg(not(target_arch = "wasm32"))] if moved { self.start_lanes(&mut q, usize::MAX); let pool_jobs = q.jobs.iter().filter(|j| !q.on_lane(j)).count().min(self.parallelism()); drop(q); for _ in 0..pool_jobs { self.start_pool(); } } } /// Start up to `n` idle lanes on the queued jobs (see [`Queue::idle_lanes`]). #[cfg(not(target_arch = "wasm32"))] fn start_lanes(&self, q: &mut Queue, n: usize) { if self.remote.is_some() { return; } let jobs = q.jobs.iter().filter(|j| q.on_lane(j)).count(); for _ in 0..n.min(jobs) { let Some(lane) = q.idle_lanes.pop() else { break }; let Some(pool) = self.lanes.get(lane) else { break }; let w = self.worker(); pool.spawn(move || w.drain(Some(lane))); } } /// Have a pool thread render the queued jobs that are the pool's. #[cfg(not(target_arch = "wasm32"))] fn start_pool(&self) { if self.remote.is_some() { return; } let w = self.worker(); let job = move || w.drain(None); match &self.pool { Some(pool) => pool.spawn(job), None => rayon::spawn(job), } } /// Queue a prefetch frame (no-op if cached or already queued/rendering). pub fn request(&self, src: &RenderSource, key: FrameKey, comp: ItemId, t: Tick, opts: RenderOpts) { self.request_with(src, key, comp, t, opts, Want::Prefetch); } /// Queue a frame nobody waits for (Cache Frames When Idle): as [`Frames::request`], but it /// renders on a lane, one thread, and leaves the pool to the viewer's frame (see /// [`Queue::idle_lanes`]). pub fn request_background(&self, src: &RenderSource, key: FrameKey, comp: ItemId, t: Tick, opts: RenderOpts) { self.request_with(src, key, comp, t, opts, Want::Background); } /// Queue the frame the viewer is showing: it jumps ahead of every prefetch job. pub fn request_urgent(&self, src: &RenderSource, key: FrameKey, comp: ItemId, t: Tick, opts: RenderOpts) { self.request_with(src, key, comp, t, opts, Want::Viewer); } /// An urgent (viewer) frame is queued or rendering. pub fn urgent_pending(&self) -> bool { self.queue.lock().map(|q| q.urgent_running > 0 || q.jobs.iter().any(|j| j.urgent)).unwrap_or(false) } fn request_with(&self, src: &RenderSource, key: FrameKey, comp: ItemId, t: Tick, opts: RenderOpts, want: Want) { let urgent = want == Want::Viewer; // The frame on screen sets the scale evictions keep: a prefetch or a passive viewer at // another zoom must not evict the active viewer's frames. let cached = self .cache .lock() .map(|mut c| { if urgent { c.scale = Some(key.scale); } c.map.contains_key(&key) }) .unwrap_or(false); if cached { return; } let Ok(mut q) = self.queue.lock() else { return }; let Ok(mut inf) = self.inflight.lock() else { return }; // Jobs for an older project revision can never be shown: drop them. let before = q.jobs.len(); q.jobs.retain(|j| j.key.revision >= key.revision); if q.jobs.len() < before { inf.retain(|k| k.revision >= key.revision || q.running.contains(k)); } if urgent { // Only the newest viewer frame is urgent. The comp's earlier viewer frames that // haven't started are dropped (scrubbing: rendering every frame the pointer passed // would hold the workers the frame under it needs); frames already rendering finish // into the cache. During edits the viewer advances through compatible completed // revisions; scrubbing holds its last frame until the requested time is in (#284). // Other comps' viewer frames become prefetch. q.jobs.retain(|j| { let superseded = j.urgent && j.key != key && j.key.comp == key.comp; if superseded { inf.remove(&j.key); } !superseded }); for j in q.jobs.iter_mut() { j.urgent = j.key == key; } // The comp's viewer frames of other times that are rendering are called off too: a // movie seeking for each of them held the decoder the frame under the pointer // waited for, so it came seconds after the pointer stopped (#329). Another revision // of the same frame (a live edit) finishes, as the viewer shows it meanwhile. for (k, cancel) in &q.urgent_cancels { if k.comp == key.comp && k.frame != key.frame { cancel.store(true, Ordering::Relaxed); } } } if !inf.insert(key) { drop(inf); drop(q); // Queued for a lane and now the viewer's frame: it is the pool's. #[cfg(not(target_arch = "wasm32"))] if urgent { self.start_pool(); } return; } q.seq += 1; let seq = q.seq; let job = Job { key, src: src.clone(), comp, t, opts, urgent, background: want == Want::Background, seq, cancel: Arc::default() }; // A lane's frame: a busy lane takes the next job when it is done, an idle one starts now. #[cfg(not(target_arch = "wasm32"))] let on_lane = q.on_lane(&job); q.jobs.push(job); #[cfg(not(target_arch = "wasm32"))] if on_lane { self.start_lanes(&mut q, 1); } drop(inf); drop(q); // One spawn per job; each spawn renders whichever queued jobs matter most right now. #[cfg(not(target_arch = "wasm32"))] if !on_lane { self.start_pool(); } #[cfg(target_arch = "wasm32")] if let Some(ctx) = &self.ctx { // rendered by `pump` after this UI frame ctx.request_repaint(); } } fn worker(&self) -> Worker { Worker { queue: self.queue.clone(), cache: self.cache.clone(), inflight: self.inflight.clone(), ctx: self.ctx.clone(), last: self.last_ms.clone(), content_keys: self.content_keys.clone(), virtual_epoch: self.virtual_epoch.clone(), virtual_id: self.virtual_id.clone(), gpu_retired: self.gpu_retired.clone(), } } /// Browser file table: `epoch` bumps on each replacement, and `id` is the size and content /// hash used when a content key is rebuilt. Neither runs on a cache hit. pub fn set_virtual_media(&mut self, epoch: Arc, id: Arc Option<(u64, u64)> + Send + Sync>) { self.virtual_epoch = epoch; self.virtual_id = Some(id); if let Ok(mut keys) = self.content_keys.lock() { keys.clear(); } } /// Render frames with `remote` from now on (`None`: on this thread again). pub fn set_remote(&mut self, remote: Option>) { self.remote = remote; } /// The remote renderer keeps a disk cache (the browser's). pub fn remote_disk(&self) -> bool { self.remote.as_ref().is_some_and(|r| r.disk()) } /// Whether frames go to a usable [`RemoteFrames`]. pub fn remote_active(&self) -> bool { self.remote.as_ref().is_some_and(|r| r.slots() > 0) } /// Frames the remote renderer renders at once (0 without one). pub fn remote_slots(&self) -> usize { self.remote.as_ref().map(|r| r.slots()).unwrap_or(0) } /// Frames being rendered remotely. pub fn remote_busy(&self) -> usize { self.remote_busy.lock().map(|b| *b).unwrap_or(0) } /// Hand queued frames to the remote renderer while it has free slots, most important first /// (the viewer's frame, then prefetch in request order; a viewer frame requested while every /// slot is busy goes to the first one that frees up). Frames that the GPU compositor can keep /// on the GPU render here instead (no readback). Returns how many were started. pub fn dispatch_remote(&self) -> usize { let Some(remote) = self.remote.clone() else { return 0 }; let slots = remote.slots(); let w = self.worker(); let mut started = 0; loop { let busy = self.remote_busy(); if busy >= slots { break; } let job = { let Ok(mut q) = self.queue.lock() else { break }; let Some(i) = q.best() else { break }; let job = q.jobs.swap_remove(i); q.running.insert(job.key); if job.urgent { q.urgent_running += 1; } job }; // The GPU compositor here keeps the frame on the GPU, but runs layer effects on the // CPU (this thread can't wait for readbacks): frames with effects go to workers // that render them on their own GPU. let gpu_here = job.src.gpu.is_some() && job.src.gpu_display && !(remote.gpu() && comp_has_effects(&job.src.project, job.comp)); if gpu_here { let t0 = web_time::Instant::now(); if let Some(img) = w.render_gpu(&job) { w.finish(&job, img, t0); continue; } } if let Ok(mut b) = self.remote_busy.lock() { *b += 1; } started += 1; let (key, urgent) = (job.key, job.urgent); let fw = self.worker(); let busy = self.remote_busy.clone(); let t0 = web_time::Instant::now(); let epoch = w.virtual_epoch.load(Ordering::Relaxed); let disk_key = remote.disk().then(|| frame_disk_key(&w.content_keys, &job.src.project, &key, opts_hash(&job.opts), epoch, w.virtual_id.as_deref())); let rj = RemoteJob { project: job.src.project.clone(), revision: key.revision, comp: job.comp, t: job.t, opts: job.opts, disk_key }; remote.start( rj, Box::new(move |r: Result| { if let Ok(mut b) = busy.lock() { *b = b.saturating_sub(1); } match r { Ok(f) => { let img = egui::ColorImage::from_rgba_premultiplied([f.width as usize, f.height as usize], &f.rgba); fw.finish_key(key, urgent, FrameImage::Cpu(Arc::new(img)), t0); } Err(e) => { // Not rendered (a stale revision, a lost worker): the viewer asks again. log::debug!("remote frame: {e}"); fw.release(key, urgent); } } }), ); } started } /// Render queued frames, most important first. With a usable [`RemoteFrames`] they are only /// handed over ([`Frames::dispatch_remote`]); otherwise they render on this thread until the /// queue is empty or `budget` has passed (at least one frame). Returns whether frames are /// still queued or rendering remotely. Only wasm32 needs this (native frames render on the /// pool); elsewhere it does nothing. pub fn pump(&self, budget: std::time::Duration) -> bool { if cfg!(not(target_arch = "wasm32")) { return false; } if self.remote_active() { self.dispatch_remote(); return self.queue.lock().map(|q| !q.jobs.is_empty()).unwrap_or(false) || self.remote_busy() > 0; } let t0 = web_time::Instant::now(); let w = self.worker(); while w.run_next(None) { if t0.elapsed() >= budget { break; } } self.queue.lock().map(|q| !q.jobs.is_empty()).unwrap_or(false) } } /// Renders queued jobs (on a pool thread, or the UI thread on wasm32). struct Worker { queue: Arc>, cache: Arc>, inflight: Arc>>, ctx: Option, last: Arc>, content_keys: ContentKeys, virtual_epoch: Arc, virtual_id: Option Option<(u64, u64)> + Send + Sync>>, gpu_retired: Arc, } #[cfg(not(target_arch = "wasm32"))] thread_local! { /// This pool thread is inside [`Worker::drain`] (rendering a frame). static DRAINING: std::cell::Cell = const { std::cell::Cell::new(false) }; } /// Clears [`DRAINING`] when the drain ends, even by a panic. #[cfg(not(target_arch = "wasm32"))] struct Draining; #[cfg(not(target_arch = "wasm32"))] impl Drop for Draining { fn drop(&mut self) { DRAINING.set(false); } } impl Worker { /// Render queued jobs, most important first, until the queue is empty. /// /// A pool thread waiting inside a render (on a parallel loop's join) runs other pool work /// meanwhile, including the spawns of newer requests: each would start a whole frame render /// on top of the one waiting, with all of its buffers. Dragging a layer asks for a new frame /// every UI frame, so renders piled up on every thread until memory ran out (#647). A spawn /// that lands inside a render returns at once instead; the render below it takes the queued /// jobs when it is done, so each thread renders one frame at a time. /// /// `lane`: the lane this runs on (see [`Queue::idle_lanes`]), `None` on the pool. #[cfg(not(target_arch = "wasm32"))] fn drain(&self, lane: Option) { if DRAINING.get() { return; } DRAINING.set(true); let _reset = Draining; while self.run_next(lane) {} } /// Render the queued job that matters most right now, on lane `lane` (`None`: on /// the pool, or the UI thread). `false` when the queue had no job for it; the lane is then /// idle again. fn run_next(&self, lane: Option) -> bool { let Worker { queue, content_keys, .. } = self; let job = { let Ok(mut q) = queue.lock() else { return false }; let Some(i) = q.next(lane.is_some()) else { q.idle_lanes.extend(lane); return false; }; let job = q.jobs.swap_remove(i); q.running.insert(job.key); if job.urgent { q.urgent_running += 1; q.urgent_cancels.push((job.key, job.cancel.clone())); } job }; let t0 = web_time::Instant::now(); let epoch = self.virtual_epoch.load(Ordering::Relaxed); let disk = job .src .disk .as_ref() .map(|dc| (dc.clone(), frame_disk_key(content_keys, &job.src.project, &job.key, opts_hash(&job.opts), epoch, self.virtual_id.as_deref()))); let from_disk = disk.as_ref().and_then(|(dc, k)| dc.get_frame(*k)); let disk_hit = from_disk.is_some(); let ci = match from_disk { Some(f) => FrameImage::Cpu(Arc::new(egui::ColorImage::from_rgba_premultiplied([f.width as usize, f.height as usize], &f.rgba))), None => match self.render_guarded(&job).filter(|_| !job.cancel.load(Ordering::Relaxed)) { Some(ci) => ci, // Called off (frames it read may be missing) or failed: not cached, and not // "in progress" forever: the viewer asks for it again. None => { self.release(job.key, job.urgent); return true; } }, }; // Publish the shared frame and wake the viewer before optional RGBA conversion // and compression. Persistence still runs on this worker, using the same snapshot. self.finish(&job, ci.clone(), t0); if !disk_hit && let (Some((dc, k)), FrameImage::Cpu(img)) = (&disk, &ci) { let rgba: Vec = img.pixels.iter().flat_map(|c| c.to_array()).collect(); dc.put_frame(*k, img.size[0] as u32, img.size[1] as u32, &rgba); if let Some(ctx) = &self.ctx { ctx.request_repaint(); } } true } /// Store a finished frame and release its job. fn finish(&self, job: &Job, img: FrameImage, t0: web_time::Instant) { self.finish_key(job.key, job.urgent, img, t0); } fn finish_key(&self, key: FrameKey, urgent: bool, img: FrameImage, t0: web_time::Instant) { if urgent && let Ok(mut l) = self.last.lock() { *l = t0.elapsed().as_secs_f64() * 1000.0; } if let Ok(mut c) = self.cache.lock() && (!img.is_gpu() || !self.gpu_retired.load(Ordering::Acquire)) { c.insert(key, img); } self.release(key, urgent); } /// The job of `key` is no longer queued or running (rendered or dropped). fn release(&self, key: FrameKey, urgent: bool) { if let Ok(mut inf) = self.inflight.lock() { inf.remove(&key); } if let Ok(mut q) = self.queue.lock() { q.running.remove(&key); if urgent { q.urgent_running = q.urgent_running.saturating_sub(1); if let Some(i) = q.urgent_cancels.iter().position(|(k, _)| *k == key) { q.urgent_cancels.swap_remove(i); } } } if let Some(ctx) = &self.ctx { ctx.request_repaint(); } } /// [`Worker::render_job`], again on the CPU if it panicked (the panic hook has logged it); /// `None` if that panicked too. fn render_guarded(&self, job: &Job) -> Option { let attempt = |gpu: bool| std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| self.render_job(job, gpu))).ok(); attempt(true).or_else(|| { log::error!("a frame render panicked (frame {})", job.key.frame); job.src.gpu.as_ref().and_then(|_| attempt(false)) }) } /// A failed scoped attempt retries on CPU. Retired contexts cannot recover by reducing /// their budget; only a healthy device's allocation failure uses the upstream backoff. fn gpu_attempt_failed(&self, gpu: &Gpu, e: &FrameError) { if gpu.context().check_health().is_err() { log::warn!("the GPU context is retired ({e}); the frame renders on the CPU"); return; } // Not memory pressure: keeping fewer GPU frames wouldn't help (#1319). if let FrameError::Validation(_) = e { log::warn!("the GPU rejected the frame ({e}); it renders on the CPU, the next frame tries the GPU again"); return; } log::warn!("the GPU ran out of memory ({e}); the frame renders on the CPU and fewer GPU frames are kept"); if let Ok(mut c) = self.cache.lock() { c.gpu_out_of_memory(); } } /// The frame on the GPU, kept there for the viewer, if the compositor can do it alone. fn render_gpu(&self, job: &Job) -> Option { if self.gpu_retired.load(Ordering::Acquire) { return None; } let g = job.src.gpu.as_ref()?; let footage = job.footage(); let mut r = Renderer::new(&job.src.project, &footage, job.opts); r.expr = job.src.expr.as_deref(); r.cache = Some(&job.src.layer_cache); r.accel = Some(g as &dyn effectcraft_engine::render::Accelerator); r.active_accel()?; match g.within_memory(|| g.render_display(&r, job.comp, job.t)) { Ok(f) => f.map(|f| FrameImage::Gpu(Arc::new(f))), Err(e) => { self.gpu_attempt_failed(g, &e); None } } } /// Render a frame; `gpu: false` keeps it off the GPU compositor. fn render_job(&self, job: &Job, gpu: bool) -> FrameImage { let footage = job.footage(); let mut r = Renderer::new(&job.src.project, &footage, job.opts); r.expr = job.src.expr.as_deref(); let gpu_allowed = gpu && !self.gpu_retired.load(Ordering::Acquire); // A failed GPU attempt may have populated intermediate entries before its scope // reported failure. The CPU retry must compute its own pixels from the snapshot. r.cache = gpu_allowed.then_some(job.src.layer_cache.as_ref()); let gpu = job.src.gpu.as_ref().filter(|_| gpu_allowed); r.accel = gpu.map(|g| g as &dyn effectcraft_engine::render::Accelerator); // The GPU leaves the frame in a texture for the viewer; otherwise (Software Only, no // adapter, or a frame the GPU cannot finish here) the CPU renders it. // Auto (Mercury GPU Acceleration) shows each comp on whichever compositor measured // faster for it (light comps composite faster on the CPU). if let (Some(g), Some(a)) = (gpu, r.active_accel()) && job.src.gpu_display { // The browser cannot wait for the GPU to time it (and its CPU path is slow): always // the GPU there. let auto = if cfg!(target_arch = "wasm32") { None } else { r.frame_auto(a, job.comp, true) }; if auto.is_none_or(|(p, key)| p.choose(key)) { let t0 = web_time::Instant::now(); match g.within_memory(|| g.render_display(&r, job.comp, job.t)) { Ok(Some(f)) => { if let Some((p, key)) = auto { // Time the GPU's work, not just its recording. g.wait(); if g.context().check_health().is_err() { return self.render_job(job, false); } p.record(key, true, t0.elapsed().as_secs_f64() * 1000.0); } return FrameImage::Gpu(Arc::new(f)); } Ok(None) => { if let Some((p, key)) = auto { p.declined(key); } } Err(e) => { self.gpu_attempt_failed(g, &e); return self.render_job(job, false); } } } let t0 = web_time::Instant::now(); let img = to_color_image(&r.comp_frame_cpu(job.comp, job.t)); if let Some((p, key)) = auto { p.record(key, false, t0.elapsed().as_secs_f64() * 1000.0); } return FrameImage::Cpu(Arc::new(img)); } if let Some(g) = gpu { // The GPU composites and the frame is read back. return match g.within_memory(|| r.comp_frame(job.comp, job.t)) { Ok(img) => FrameImage::Cpu(Arc::new(to_color_image(&img))), Err(e) => { self.gpu_attempt_failed(g, &e); self.render_job(job, false) } }; } FrameImage::Cpu(Arc::new(to_color_image(&r.comp_frame(job.comp, job.t)))) } } /// Who asked for a frame (see [`Frames::request_with`]). #[derive(Clone, Copy, PartialEq, Eq)] enum Want { Viewer, Prefetch, Background, } struct Job { key: FrameKey, src: RenderSource, comp: ItemId, t: Tick, opts: RenderOpts, urgent: bool, /// Nobody waits for it (see [`Frames::request_background`]). background: bool, seq: u64, /// Set when the viewer no longer waits for this frame (see [`Frames::request_urgent`]). cancel: Arc, } impl Job { /// The footage this job's render reads, called off with the job. fn footage(&self) -> CancellableFootage<'_> { CancellableFootage { inner: self.src.footage.as_ref(), cancel: self.cancel.clone() } } } #[derive(Default)] struct Queue { jobs: Vec, running: HashSet, urgent_running: usize, /// The running viewer (urgent) frames and how to call them off. urgent_cancels: Vec<(FrameKey, Arc)>, seq: u64, /// The playing preview, if any: its prefetch renders nearest first. playhead: Option, /// How many lanes there are, and the ones not rendering. A lane is one thread that renders /// one frame at a time. The frames of a playing preview and the frames nobody waits for /// (Cache Frames When Idle) render on the lanes; the pool keeps the viewer's frame and the /// few frames prefetched around it while paused. On the pool a frame's parallel work is /// shared: a thread that took a piece of one frame and waited inside it started rendering /// the queued frames on top, and went back to its piece only once the queue was empty. The /// frames a preview started first, the ones right ahead of the playhead, were done last: /// the preview stood still on a gap while the frames further ahead turned green. Idle /// caching left the same gaps, and held up the viewer's frame when it shared a piece. lanes: usize, idle_lanes: Vec, } impl Queue { /// Whether `j` renders on a lane (else on the pool). fn on_lane(&self, j: &Job) -> bool { self.lanes > 0 && !j.urgent && (j.background || self.playhead.is_some()) } /// The job a lane (`lane`) or the pool renders next: the [`Queue::best`] of its own. fn next(&self, lane: bool) -> Option { self.best_of(|j| self.on_lane(j) == lane) } /// The job that matters most: urgent first (newest), then, while a preview plays, the /// prefetch frame it reaches first, then prefetch in request order. In request order alone, /// frames asked for again after an eviction (the ones right ahead of the playhead) waited /// behind every frame further ahead (#595). fn best(&self) -> Option { self.best_of(|_| true) } /// [`Queue::best`] among the jobs `of` accepts. fn best_of(&self, of: impl Fn(&Job) -> bool) -> Option { let p = self.playhead.as_ref(); self.jobs .iter() .enumerate() .filter(|(_, j)| of(j)) .max_by_key(|(_, j)| { let near = if j.urgent { None } else { p.and_then(|p| p.ahead(&j.key)) }; (j.urgent, near.is_some(), near.map_or(0, |d| -d), if j.urgent { j.seq as i64 } else { -(j.seq as i64) }) }) .map(|(i, _)| i) } } #[cfg(test)] mod tests;