#3335: Render Queue reads frames back once, pipelines GPU frames and picks the faster renderer

This commit is contained in:
vangel76
2026-11-09 00:55:30 +01:00
parent a41ff5015c
commit 4649c27ef9
15 changed files with 973 additions and 69 deletions
+3
View File
@@ -863,6 +863,9 @@ fn save_current_preview(s: &mut Session, p: &Value) -> Result<Value> {
storage: s.storage_quota.as_deref(),
label: "Save Current Preview".into(),
nested_switches: s.prefs.general.switches_affect_nested_comps,
caches: None,
renderer: None,
};
let r = exporter.export(&job, &mut |_, _| true).map_err(EngineError::Other)?;
s.toast(format!("Saved preview to {}", r.path));
+3
View File
@@ -145,6 +145,9 @@ fn prerender(
storage: None,
label: format!("Pre-render {}", l.name),
nested_switches: cx.nested_switches,
caches: None,
renderer: None,
};
let r = exporter.export(&job, progress)?;
Ok(Prerendered { path: r.path, width: comp.width, height: comp.height, frame_rate: comp.frame_rate, duration: end - start })
+39
View File
@@ -34,6 +34,40 @@ pub struct ExportJob<'a> {
pub label: String,
/// Settings ▸ General ▸ Switches Affect Nested Comps.
pub nested_switches: bool,
/// The layer caches of the Render Queue run this job belongs to (#3335): consecutive jobs
/// start warm, and the run trims the accelerator's pooled textures once at its end. `None`:
/// a one-off export (its own caches, trimmed when it ends).
pub caches: Option<&'a RunCaches>,
/// The compositor the queue item's output modules render on (#3335): the first module
/// chooses (measuring a GPU project's first frames) and stores it here, the others follow.
pub renderer: Option<&'a Mutex<Option<effectcraft_render::Backend>>>,
}
/// What the jobs of one Render Queue run keep between them (#3335): a layer cache per
/// compositor (a buffer an effect made on the GPU is not the CPU's), together at the budget
/// each job's single cache had before.
pub struct RunCaches {
pub gpu: effectcraft_render::LayerCache,
pub cpu: effectcraft_render::LayerCache,
}
impl Default for RunCaches {
fn default() -> Self {
let half = effectcraft_render::LayerCache::default().budget() / 2;
RunCaches { gpu: effectcraft_render::LayerCache::new(half), cpu: effectcraft_render::LayerCache::new(half) }
}
}
/// Calls [`effectcraft_render::Accelerator::render_end`] when a Render Queue run ends, however
/// it ends (done, failed, stopped, panicked): its jobs leave the pooled textures to each other.
struct RunEnd<'a>(Option<&'a dyn effectcraft_render::Accelerator>);
impl Drop for RunEnd<'_> {
fn drop(&mut self) {
if let Some(a) = self.0 {
a.render_end();
}
}
}
/// What an export wrote.
@@ -317,6 +351,8 @@ async fn run_work(w: Work, shared: &JobShared, notify: &mut dyn FnMut(&JobShared
let mut s = lock(&shared.state);
s.items_total = w.items.len();
}
let caches = RunCaches::default();
let _run_end = RunEnd(w.accel.as_deref());
for (k, modules) in w.items.iter().enumerate() {
let Some((item, _)) = modules.first() else { continue };
if shared.cancel.load(Ordering::Relaxed) {
@@ -337,6 +373,7 @@ async fn run_work(w: Work, shared: &JobShared, notify: &mut dyn FnMut(&JobShared
// Every output module encodes the same frames (Composition ▸ Add Output Module); the
// first module's file is the item's output.
let mut r: Result<ExportResult, String> = Err("no output module".into());
let renderer = Mutex::new(None);
for (mi, (mitem, path)) in modules.iter().enumerate() {
let name = w.project.item(mitem.comp).map(|i| i.name.clone()).unwrap_or_default();
let index = w.project.render_queue.iter().position(|i| i.id == mitem.id).map_or(0, |i| i + 1);
@@ -350,6 +387,8 @@ async fn run_work(w: Work, shared: &JobShared, notify: &mut dyn FnMut(&JobShared
storage: w.storage.as_deref(),
label: format!("#{index} {name} (output module {})", mi + 1),
nested_switches: w.nested_switches,
caches: Some(&caches),
renderer: Some(&renderer),
};
let rr = w
.exporter
+58
View File
@@ -248,3 +248,61 @@ fn locked_resize_uses_height_for_height_only_output_module_updates() {
let comp = s.project.comp(item.comp).unwrap();
assert_eq!(item.output.output_size(comp, &item.settings), (32, 32), "explicit width retains precedence");
}
#[test]
fn a_queue_run_keeps_its_jobs_warm() {
// #3335: every job started with a fresh layer cache and trimmed the GPU's texture pool when
// it ended, so each item's first frames were cold. A run's jobs share its caches; the pool is
// trimmed once, when the run ends; an item's output modules share one compositor choice.
use std::sync::atomic::{AtomicUsize, Ordering};
struct Recorder(Mutex<Vec<(Option<usize>, Option<usize>)>>);
impl Exporter for Recorder {
fn formats(&self) -> Vec<OutputFormat> {
OutputFormat::ALL.to_vec()
}
fn export(&self, job: &ExportJob, _: &mut dyn FnMut(u64, u64) -> bool) -> Result<ExportResult, String> {
let caches = job.caches.map(|c| c as *const _ as usize);
let slot = job.renderer.map(|r| {
// The first module chooses; the second sees its choice.
r.lock().unwrap().get_or_insert(effectcraft_render::Backend::Cpu);
r as *const _ as usize
});
self.0.lock().unwrap().push((caches, slot));
Ok(ExportResult { path: job.path.to_string(), ..Default::default() })
}
}
#[derive(Default)]
struct Ends(AtomicUsize);
impl effectcraft_render::Accelerator for Ends {
fn name(&self) -> String {
"ends".into()
}
fn comp_frame(&self, _: &effectcraft_render::Renderer, _: effectcraft_project::ItemId, _: effectcraft_time::Tick) -> Option<effectcraft_raster::Image> {
None
}
fn supports_effect(&self, _: &str) -> bool {
false
}
fn effects(&self, _: &[effectcraft_render::FxStep], _: &effectcraft_effects::Buf, _: Option<f32>) -> Option<effectcraft_effects::Buf> {
None
}
fn render_end(&self) {
self.0.fetch_add(1, Ordering::SeqCst);
}
}
let mut s = demo();
let ex = Arc::new(Recorder(Mutex::new(vec![])));
s.exporter = Some(ex.clone());
let ends = Arc::new(Ends::default());
s.accel = Some(ends.clone());
s.execute("renderQueue.add", json!({"output": crate::tests::stubs::tmp_out("rq/warm_a.mp4")})).unwrap();
s.execute("render.addOutputModule", json!({"format": "png"})).unwrap();
s.execute("renderQueue.add", json!({"output": crate::tests::stubs::tmp_out("rq/warm_b.mp4")})).unwrap();
s.execute("renderQueue.render", json!({})).unwrap();
let jobs = ex.0.lock().unwrap().clone();
assert_eq!(jobs.len(), 3, "{jobs:?}");
assert!(jobs.iter().all(|(c, r)| c.is_some() && r.is_some()), "{jobs:?}");
assert!(jobs.iter().all(|(c, _)| *c == jobs[0].0), "every job of the run shares its caches");
assert_eq!(jobs[0].1, jobs[1].1, "an item's output modules share one choice");
assert_eq!(ends.0.load(Ordering::SeqCst), 1, "the pool is trimmed once, after the run");
}
+34 -1
View File
@@ -53,6 +53,7 @@ pub use effectcraft_project::render_queue;
pub use effectcraft_project::render_queue::StorageQuota;
pub use out::Sink;
pub(crate) use pipeline::Cx;
pub use pipeline::{CACHE_BUDGET, RENDERER_ENV, pick_renderer};
#[derive(Debug, thiserror::Error)]
pub enum ExportError {
@@ -107,6 +108,37 @@ pub struct JobOptions<'a> {
pub storage: Option<&'a dyn StorageQuota>,
/// Overflow folders, in order.
pub overflow: Vec<String>,
/// Layer caches shared by the consecutive jobs of a Render Queue run (#3335): the job
/// renders through them instead of fresh ones, and leaves the accelerator's pooled working
/// textures for the next job (whoever holds the caches calls
/// [`effectcraft_render::Accelerator::render_end`] after the last one). `None`: the job
/// starts cold and trims the pool when it ends, as a one-off export does.
pub caches: Option<JobCaches<'a>>,
/// The compositor of the job's frames (`Backend::Gpu` or `Backend::Cpu`; `Auto` counts as unset), shared by the output modules of
/// one queue item: when it holds a choice the job renders on it; otherwise the job stores
/// the compositor it chose there, so the item's other output modules encode the same
/// frames. `None`: the job decides alone.
pub renderer: Option<&'a std::sync::Mutex<Option<effectcraft_render::Backend>>>,
}
/// The layer caches of a Render Queue run (see [`JobOptions::caches`]): one per compositor,
/// because a buffer an effect made on the GPU is not the CPU's, and neither may stand in for
/// the other.
#[derive(Clone, Copy)]
pub struct JobCaches<'a> {
pub gpu: &'a effectcraft_render::LayerCache,
pub cpu: &'a effectcraft_render::LayerCache,
}
/// The compositor a job's frames render on: the GPU's (Mercury GPU Acceleration, with an
/// accelerator) or the CPU's frame-parallel one (Mercury Software Only). The whole job renders
/// on one of them, never a mix: the two differ slightly. A GPU project with an accelerator
/// measures both on its first frames and renders the job on the faster (#3335); the choice is
/// in the render log. `EFFECTCRAFT_RQ_RENDERER=gpu|cpu|auto` overrides the measurement.
#[derive(Clone, Copy, Debug, PartialEq)]
pub enum JobRenderer {
Gpu,
Cpu,
}
/// Progress after a batch of frames.
@@ -176,7 +208,8 @@ pub fn export(job: &Job, progress: &mut dyn FnMut(&Progress) -> bool) -> Result<
pub async fn export_async(job: &Job<'_>, progress: &mut dyn FnMut(&Progress) -> bool) -> Result<Report> {
let t0 = Instant::now();
let cx = Cx::new(job);
let render = RenderEnd(job.accel);
// A job of a Render Queue run leaves the pooled working textures warm for the next one.
let render = RenderEnd(job.accel.filter(|_| job.options.caches.is_none()));
let mut r = export_with(&cx, progress, t0).await;
drop(render);
if matches!(r, Err(ExportError::Cancelled)) {
+313 -28
View File
@@ -37,6 +37,42 @@ pub(crate) fn gpu_inflight(var: Option<&str>) -> usize {
}
}
/// Overrides the Render Queue's measured choice of compositor for GPU projects with an
/// accelerator (#3335): `gpu` (no measurement, the GPU), `cpu` (Mercury Software Only) or
/// `auto` (measure, the default).
pub const RENDERER_ENV: &str = "EFFECTCRAFT_RQ_RENDERER";
/// Frames each compositor renders, timed, when a job measures which is faster (after one
/// untimed warm-up frame each: layer cache misses, uploads, kernel builds).
const PROBE_FRAMES: usize = 2;
/// Jobs shorter than this render on the project's compositor without measuring: the frames
/// the slower one renders for the measurement would cost more than they could save.
const PROBE_MIN_JOB: u64 = 12;
/// Bytes each of a job's two layer caches (one per compositor) may hold: together what the
/// job's single cache held before (`LayerCache::default()`).
pub const CACHE_BUDGET: usize = 1 << 29;
/// The faster compositor from measured milliseconds per frame (`None`: not measured, or the
/// GPU stopped). As the viewer's Auto ([`effectcraft_render::AutoPick::MARGIN`]), the CPU wins
/// only when clearly faster: ties and noise stay on the GPU.
pub fn pick_renderer(gpu_ms: Option<f64>, cpu_ms: Option<f64>) -> crate::JobRenderer {
match (gpu_ms, cpu_ms) {
(Some(g), Some(c)) if g.is_finite() && c.is_finite() && c < g * effectcraft_render::AutoPick::MARGIN => crate::JobRenderer::Cpu,
(None, Some(_)) => crate::JobRenderer::Cpu,
_ => crate::JobRenderer::Gpu,
}
}
/// How a job's compositor was chosen, for the render log.
#[derive(Clone, Copy, Debug, PartialEq)]
pub(crate) struct Choice {
pub renderer: crate::JobRenderer,
/// Measured ms per frame on each (`None`: not measured).
pub gpu_ms: Option<f64>,
pub cpu_ms: Option<f64>,
pub why: &'static str,
}
/// Caps CPU frames in flight (#1615): a whole number of frames, at least 1. Unset, CPU renders
/// keep one frame per thread in flight as before. A heavy comp can need gigabytes of working
/// memory per frame, so 16 threads at once can get the process OOM-killed; the headless CLI
@@ -168,8 +204,10 @@ pub(crate) struct Cx<'a> {
pub project: Project,
/// Processed layer buffers shared by the job's frames, as the viewer's layer cache: static
/// and transform-only layers render once per job (the GPU compositor then uploads them
/// once). Dropped with the job.
cache: LayerCache,
/// once). One per compositor; the job's own (dropped with it) or a Render Queue run's.
caches: Caches<'a>,
/// The compositor the job renders on, once chosen ([`Cx::backend`]).
choice: Mutex<Option<Choice>>,
/// (output frame, seconds) of every rendered frame (Plus Per Frame Info logs).
pub frame_times: Option<Mutex<Vec<(u64, f64)>>>,
/// Files written to an overflow folder.
@@ -177,6 +215,23 @@ pub(crate) struct Cx<'a> {
written: Mutex<Vec<String>>,
}
/// A job's layer caches: its own, or those of the Render Queue run it belongs to.
enum Caches<'a> {
Own { gpu: Box<LayerCache>, cpu: Box<LayerCache> },
Run(crate::JobCaches<'a>),
}
impl Caches<'_> {
fn of(&self, r: crate::JobRenderer) -> &LayerCache {
match (self, r) {
(Caches::Own { gpu, .. }, crate::JobRenderer::Gpu) => gpu,
(Caches::Own { cpu, .. }, crate::JobRenderer::Cpu) => cpu,
(Caches::Run(c), crate::JobRenderer::Gpu) => c.gpu,
(Caches::Run(c), crate::JobRenderer::Cpu) => c.cpu,
}
}
}
impl<'a> std::ops::Deref for Cx<'a> {
type Target = Job<'a>;
fn deref(&self) -> &Job<'a> {
@@ -222,10 +277,85 @@ impl<'a> Cx<'a> {
// layer and effect quantises at it), not just the finished frame (#2022).
project.settings.bit_depth = job.settings.color_depth.resolve(project.settings.bit_depth);
let frame_times = (job.options.log == RenderLog::PlusPerFrameInfo).then(|| Mutex::new(Vec::new()));
let cache = LayerCache::default();
let caches = match job.options.caches {
Some(c) => Caches::Run(c),
None => Caches::Own { gpu: Box::new(LayerCache::new(CACHE_BUDGET)), cpu: Box::new(LayerCache::new(CACHE_BUDGET)) },
};
// With deferred readbacks, buffers computed from a missed pass's placeholders are not kept.
cache.set_gate(job.accel.and_then(|a| a.miss_gate()));
Cx { job, project, cache, frame_times, overflowed: Mutex::new(vec![]), written: Mutex::new(vec![]) }
for r in [crate::JobRenderer::Gpu, crate::JobRenderer::Cpu] {
caches.of(r).set_gate(job.accel.and_then(|a| a.miss_gate()));
}
let cx = Cx { job, project, caches, choice: Mutex::new(None), frame_times, overflowed: Mutex::new(vec![]), written: Mutex::new(vec![]) };
if let Some(c) = cx.fixed_choice(std::env::var(RENDERER_ENV).ok().as_deref()) {
cx.decide(c);
}
cx
}
/// The project's compositor: the GPU's for Mercury GPU Acceleration with an accelerator.
fn project_renderer(&self) -> crate::JobRenderer {
if self.project.settings.gpu_acceleration && self.job.accel.is_some() { crate::JobRenderer::Gpu } else { crate::JobRenderer::Cpu }
}
/// The job's compositor when it is decided without measuring: the queue item's first
/// output module's, the project's when there is nothing to choose between (Software Only,
/// no accelerator, deferred readbacks), or `env` ([`RENDERER_ENV`]). `None`: measure.
fn fixed_choice(&self, env: Option<&str>) -> Option<Choice> {
let fixed = |renderer, why| Some(Choice { renderer, gpu_ms: None, cpu_ms: None, why });
if self.project_renderer() == crate::JobRenderer::Cpu {
return fixed(crate::JobRenderer::Cpu, "the project's renderer");
}
if self.deferred() {
return fixed(crate::JobRenderer::Gpu, "the project's renderer");
}
let shared = self.job.options.renderer.and_then(|s| *s.lock().unwrap_or_else(std::sync::PoisonError::into_inner));
let shared = match shared {
Some(Backend::Gpu) => Some(crate::JobRenderer::Gpu),
Some(Backend::Cpu) => Some(crate::JobRenderer::Cpu),
Some(Backend::Auto) | None => None,
};
if let Some(r) = shared {
return Some(Choice { renderer: r, gpu_ms: None, cpu_ms: None, why: "as the queue item's first output module" });
}
match env.map(|v| v.trim().to_ascii_lowercase()).as_deref() {
Some("gpu") => fixed(crate::JobRenderer::Gpu, "set by EFFECTCRAFT_RQ_RENDERER"),
Some("cpu") => fixed(crate::JobRenderer::Cpu, "set by EFFECTCRAFT_RQ_RENDERER"),
None | Some("" | "auto") => None,
Some(v) => {
log::warn!(target: "effectcraft_export", "{RENDERER_ENV}={v:?} is not gpu, cpu or auto: measuring");
None
}
}
}
/// Settle the job's compositor (once): remember it for the queue item's other output
/// modules and say so in the log.
fn decide(&self, c: Choice) {
let mut slot = self.choice.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
if slot.is_some() {
return;
}
*slot = Some(c);
if let Some(shared) = self.job.options.renderer {
let mut shared = shared.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
if shared.is_none_or(|b| b == Backend::Auto) {
*shared = Some(match c.renderer {
crate::JobRenderer::Gpu => Backend::Gpu,
crate::JobRenderer::Cpu => Backend::Cpu,
});
}
}
log::info!(target: "effectcraft_export", "{}: {}", self.job.options.label, describe_choice(&c));
}
/// The job's choice, once made.
pub(crate) fn choice(&self) -> Option<Choice> {
*self.choice.lock().unwrap_or_else(std::sync::PoisonError::into_inner)
}
/// The compositor frames render on: the choice, or the project's until it is made.
fn backend(&self) -> crate::JobRenderer {
self.choice().map_or_else(|| self.project_renderer(), |c| c.renderer)
}
pub fn comp(&self) -> Option<&Comp> {
@@ -234,12 +364,22 @@ impl<'a> Cx<'a> {
/// The renderer of the job's frames: the Render Settings' options, the job's accelerator
/// and layer cache.
#[cfg(test)]
fn renderer(&self) -> Renderer<'_> {
self.renderer_on(self.backend())
}
/// [`Cx::renderer`] on compositor `on`, with that compositor's layer cache.
fn renderer_on(&self, on: crate::JobRenderer) -> Renderer<'_> {
let s = self.job.settings;
// The whole job renders on the project's renderer (Mercury GPU Acceleration or Software
// Only), not on whichever compositor Auto timed faster for each frame: the CPU and GPU
// compositors differ slightly, so a per-frame choice made renders differ run to run.
let backend = if self.project.settings.gpu_acceleration { Backend::Gpu } else { Backend::Cpu };
// The whole job renders on one compositor (the project's, or the faster one measured on
// the first frames, #3335), not on whichever compositor Auto timed faster for each
// frame: the CPU and GPU compositors differ slightly, so a per-frame choice made renders
// differ within a render.
let backend = match on {
crate::JobRenderer::Gpu => Backend::Gpu,
crate::JobRenderer::Cpu => Backend::Cpu,
};
let opts = RenderOpts {
scale: self.comp().map(|c| s.effective_scale(c)).unwrap_or_else(|| s.resolution.clamp(0.01, 4.0)),
motion_blur: s.motion_blur_override() != SwitchOverride::OffForAll,
@@ -258,13 +398,13 @@ impl<'a> Cx<'a> {
let mut r = Renderer::new(&self.project, self.job.footage, opts);
r.expr = self.job.expr;
r.accel = self.job.accel;
r.cache = Some(&self.cache);
r.cache = Some(self.caches.of(on));
r
}
async fn render_at(&self, t: Tick) -> Image {
async fn render_at(&self, t: Tick, on: crate::JobRenderer) -> Image {
// With deferred GPU readbacks (a browser worker) the frame renders in passes.
effectcraft_render::passes::comp_frame(&self.renderer(), self.job.comp, t).await
effectcraft_render::passes::comp_frame(&self.renderer_on(on), self.job.comp, t).await
}
/// Whether frames render in passes (an accelerator with deferred readbacks).
@@ -283,10 +423,91 @@ impl<'a> Cx<'a> {
}
return out;
}
self.render_frames(ks, |k, on| self.frame_on(comp, k, on), f)
}
/// `render` then `encode` over the frames `ks` on the job's compositor, results in `ks`
/// order. The first call of a job that still has to choose its compositor measures both on
/// its first frames ([`Cx::probe`]) and keeps the faster one's.
fn render_frames<I: Send, T: Send>(
&self,
ks: Vec<u64>,
render: impl Fn(u64, crate::JobRenderer) -> I + Sync + Send,
encode: impl Fn(u64, I) -> T + Sync + Send,
) -> Vec<T> {
use rayon::prelude::*;
let probed = self.probe(&ks, &render);
// The GPU cap is for the GPU's memory: once it has stopped for good, frames render on the
// CPU, in parallel or under the CPU cap (#1615), which holds throughout.
let gpu_live = || self.job.accel.is_none_or(|a| a.available());
map_frames_while(ks, self.inflight(), gpu_live, |k| self.frame(comp, k), f)
let on = self.backend();
if probed.is_empty() {
return map_frames_while(ks, self.inflight(), gpu_live, |k| render(k, on), encode);
}
let rest: Vec<u64> = ks.iter().copied().filter(|k| !probed.iter().any(|(p, _)| p == k)).collect();
let mut done: Vec<(u64, T)> = probed.into_par_iter().map(|(k, i)| (k, encode(k, i))).collect();
done.extend(rest.iter().copied().zip(map_frames_while(rest.clone(), self.inflight(), gpu_live, |k| render(k, on), &encode)));
// Back in `ks` order.
let mut out = Vec::with_capacity(ks.len());
for k in &ks {
if let Some(i) = done.iter().position(|(d, _)| d == k) {
out.push(done.swap_remove(i).1);
}
}
out
}
/// Choose the job's compositor by timing both on the first frames of `ks` (#3335), as the
/// viewer's Auto does per comp: frame 0 warms each up (layer cache misses, uploads, kernel
/// builds) untimed, then the GPU renders the next [`PROBE_FRAMES`] under its in-flight cap
/// and the CPU the [`PROBE_FRAMES`] after them in parallel. The CPU renders more frames at
/// once in the job than here, so the measurement leans towards the GPU, as the margin of
/// [`pick_renderer`] does: the CPU is chosen only when clearly faster. Returns
/// the frames the faster one rendered (the slower one's are rendered again); empty when
/// the choice was made already or is made without measuring (a short job).
fn probe<I: Send>(&self, ks: &[u64], render: &(impl Fn(u64, crate::JobRenderer) -> I + Sync + Send)) -> Vec<(u64, I)> {
if self.choice().is_some() {
return vec![];
}
let total = self.comp().map_or(0, |c| self.job.settings.frame_count(c));
if total < PROBE_MIN_JOB {
self.decide(Choice { renderer: crate::JobRenderer::Gpu, gpu_ms: None, cpu_ms: None, why: "the project's renderer (too few frames to measure)" });
return vec![];
}
// Small batches (few threads) measure one frame each; a batch too small for even that
// (a resumed sequence skipping frames) renders on the GPU and leaves it to the next.
let n = PROBE_FRAMES.min(ks.len().saturating_sub(1) / 2);
if n == 0 {
return vec![];
}
let (Some(&first), Some(gpu_ks), Some(cpu_ks)) = (ks.first(), ks.get(1..1 + n), ks.get(1 + n..1 + 2 * n)) else { return vec![] };
let logged = || self.frame_times.as_ref().map_or(0, |ft| ft.lock().unwrap_or_else(std::sync::PoisonError::into_inner).len());
let cpu_cap = self.inflight().cpu;
let timed = |on: crate::JobRenderer, frames: &[u64], caps: Inflight| {
let warm = render(first, on);
let t0 = Instant::now();
let gpu_live = || on == crate::JobRenderer::Cpu || self.job.accel.is_some_and(|a| a.available());
let items = map_frames_while(frames.to_vec(), caps, gpu_live, |k| render(k, on), |k, i| (k, i));
let ms = t0.elapsed().as_secs_f64() * 1e3 / frames.len().max(1) as f64;
(std::iter::once((first, warm)).chain(items).collect::<Vec<_>>(), ms)
};
let log0 = logged();
let (gpu, gpu_ms) = timed(crate::JobRenderer::Gpu, gpu_ks, Inflight { gpu: Some(self.gpu_inflight()), cpu: cpu_cap });
let log1 = logged();
let (cpu, cpu_ms) = timed(crate::JobRenderer::Cpu, cpu_ks, Inflight { gpu: None, cpu: cpu_cap });
// A GPU that stopped meanwhile rendered its frames on the CPU: it is no choice.
let gpu_ms = self.job.accel.is_some_and(|a| a.available()).then_some(gpu_ms);
let renderer = pick_renderer(gpu_ms, Some(cpu_ms));
// The per-frame log keeps the kept frames' times only.
if let Some(ft) = &self.frame_times {
let mut v = ft.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
let (a, b) = if renderer == crate::JobRenderer::Gpu { (log1, v.len()) } else { (log0, log1) };
if a <= b && b <= v.len() {
v.drain(a..b);
}
}
self.decide(Choice { renderer, gpu_ms, cpu_ms: Some(cpu_ms), why: "measured on the first frames" });
if renderer == crate::JobRenderer::Gpu { gpu } else { cpu }
}
/// [`Cx::frames`] for `ks` while `encode` runs beside them on `beside`'s threads, so the
@@ -338,9 +559,15 @@ impl<'a> Cx<'a> {
/// GPU frames rendered at once (#879), decided once per batch: with a GPU accelerator every
/// frame in flight holds its own working textures, so a whole batch rendered at once can
/// exhaust VRAM (or shared memory, which crashes instead of falling back). `None` when the
/// frames render on the CPU: no accelerator, or a Software Only project.
/// frames render on the CPU: no accelerator, a Software Only project, or a job that chose
/// the CPU's compositor (#3335), whose frames render in parallel.
fn gpu_cap(&self) -> Option<usize> {
(self.job.accel.is_some() && self.project.settings.gpu_acceleration).then(|| gpu_inflight(std::env::var(GPU_INFLIGHT_ENV).ok().as_deref()))
(self.job.accel.is_some() && self.backend() == crate::JobRenderer::Gpu).then(|| self.gpu_inflight())
}
/// GPU frames in flight ([`GPU_INFLIGHT_ENV`] or [`GPU_INFLIGHT_DEFAULT`]).
fn gpu_inflight(&self) -> usize {
gpu_inflight(std::env::var(GPU_INFLIGHT_ENV).ok().as_deref())
}
/// The GPU and CPU frames-in-flight caps every frame loop shares (#1615).
@@ -349,18 +576,28 @@ impl<'a> Cx<'a> {
}
/// Output frame `i`: sampled (fields, pulldown), at the colour depth, cropped and resized.
#[cfg(test)]
pub fn frame(&self, comp: &Comp, i: u64) -> Image {
effectcraft_render::passes::block_on(self.frame_async(comp, i))
self.frame_on(comp, i, self.backend())
}
/// [`Cx::frame`] on compositor `on`.
fn frame_on(&self, comp: &Comp, i: u64, on: crate::JobRenderer) -> Image {
effectcraft_render::passes::block_on(self.frame_async_on(comp, i, on))
}
/// [`Cx::frame`] as a future (deferred readbacks: the frame renders in passes).
pub async fn frame_async(&self, comp: &Comp, i: u64) -> Image {
self.frame_async_on(comp, i, self.backend()).await
}
async fn frame_async_on(&self, comp: &Comp, i: u64, on: crate::JobRenderer) -> Image {
let t0 = Instant::now();
let img = match self.job.settings.sample(comp, i) {
FrameSample::Progressive(t) => self.render_at(t).await,
FrameSample::Progressive(t) => self.render_at(t, on).await,
FrameSample::Fields { first, second, upper_first } => {
let a = self.render_at(first).await;
if second == first { a } else { interleave(&a, &self.render_at(second).await, upper_first) }
let a = self.render_at(first, on).await;
if second == first { a } else { interleave(&a, &self.render_at(second, on).await, upper_first) }
}
};
let img = self.crop_resize(img);
@@ -374,7 +611,7 @@ impl<'a> Cx<'a> {
/// by the accelerator ([`effectcraft_render::Accelerator::comp_frame_yuv420`]). Only for
/// frames that need nothing done after the render (progressive, at the project's depth, no
/// crop or resize, rendered at exactly `w`×`h`). `None`: render it with [`Cx::frame`].
pub fn yuv420_frame(&self, comp: &Comp, i: u64, w: u32, h: u32) -> Option<Vec<u8>> {
fn yuv420_frame(&self, comp: &Comp, i: u64, w: u32, h: u32, on: crate::JobRenderer) -> Option<Vec<u8>> {
if self.deferred() {
return None;
}
@@ -392,7 +629,7 @@ impl<'a> Cx<'a> {
return None;
}
let t0 = Instant::now();
let r = self.renderer();
let r = self.renderer_on(on);
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);
@@ -419,12 +656,9 @@ impl<'a> Cx<'a> {
fallback: impl Fn(u64, Image) -> T + Sync + Send,
) -> Vec<T> {
// As in `frames`: once the GPU has stopped for good, the rest render in parallel.
let gpu_live = || self.job.accel.is_none_or(|a| a.available());
map_frames_while(
self.render_frames(
ks,
self.inflight(),
gpu_live,
|k| self.yuv420_frame(comp, k, w, h).ok_or_else(|| self.frame(comp, k)),
|k, on| self.yuv420_frame(comp, k, w, h, on).ok_or_else(|| self.frame_on(comp, k, on)),
|k, r| match r {
Ok(p) => planes(p),
Err(img) => fallback(k, img),
@@ -544,6 +778,9 @@ impl<'a> Cx<'a> {
Ok(r) => s.push_str(&format!("Result: Done ({} frames, {}×{}, {} bytes) in {seconds:.2} s\n", r.frames, r.width, r.height, r.bytes)),
Err(e) => s.push_str(&format!("Result: Failed\nError: {e}\n")),
}
if let Some(c) = self.choice() {
s.push_str(&format!("Renderer: {}\n", describe_choice(&c)));
}
let over = self.overflowed.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
if !over.is_empty() {
s.push_str(&format!("Storage overflow: {} file(s) written to overflow folders (first: {})\n", over.len(), over[0]));
@@ -575,6 +812,20 @@ impl<'a> Cx<'a> {
}
}
/// The render log's account of a job's compositor: which, why, and what was measured.
pub(crate) fn describe_choice(c: &Choice) -> String {
let name = match c.renderer {
crate::JobRenderer::Gpu => "Mercury GPU Acceleration",
crate::JobRenderer::Cpu => "Mercury Software Only",
};
let ms = |v: Option<f64>| v.map_or_else(|| "n/a".to_string(), |v| format!("{v:.1} ms/frame"));
if c.gpu_ms.is_none() && c.cpu_ms.is_none() {
format!("{name} ({})", c.why)
} else {
format!("{name} ({}: GPU {}, CPU {})", c.why, ms(c.gpu_ms), ms(c.cpu_ms))
}
}
/// Weave two fields: the dominant field's lines (even lines when `upper_first`) from `first`.
pub(crate) fn interleave(first: &Image, second: &Image, upper_first: bool) -> Image {
let mut out = first.clone();
@@ -860,6 +1111,40 @@ mod tests {
assert_eq!(Cx::new(&job).gpu_cap(), None, "no accelerator");
}
#[test]
fn a_job_that_chose_the_cpu_is_not_capped() {
// #3335: the CPU's frame-parallel compositor is the point of choosing it; a GPU project
// that measured the CPU faster must not keep rendering one frame at a time.
use effectcraft_project::render_queue::{OutputFormat, OutputModule, RenderSettings};
let (settings, output) = (RenderSettings::default(), OutputModule::for_format(OutputFormat::PngSequence));
let accel = NoAccel;
let mut p = Project::default();
p.settings.gpu_acceleration = true;
let job = Job {
project: &p,
footage: &effectcraft_render::NoFootage,
expr: None,
accel: Some(&accel),
comp: effectcraft_project::ItemId(0),
settings: &settings,
output: &output,
path: "/c/f_[#####].png",
sink: None,
nested_switches: true,
options: Default::default(),
};
let cx = Cx::new(&job);
if std::env::var_os(RENDERER_ENV).is_none() {
assert_eq!(cx.choice(), None, "a GPU project measures first");
assert!(cx.gpu_cap().is_some());
}
cx.decide(Choice { renderer: crate::JobRenderer::Cpu, gpu_ms: Some(117.0), cpu_ms: Some(68.0), why: "measured on the first frames" });
assert_eq!(cx.backend(), crate::JobRenderer::Cpu);
assert_eq!(cx.gpu_cap(), None);
assert_eq!(cx.renderer().active_accel().map(|a| a.name()), None, "CPU frames don't use the accelerator");
assert_eq!(describe_choice(&cx.choice().unwrap()), "Mercury Software Only (measured on the first frames: GPU 117.0 ms/frame, CPU 68.0 ms/frame)");
}
#[test]
fn cpu_inflight_is_off_unless_a_positive_number() {
assert_eq!(cpu_inflight(None), None);
@@ -1002,7 +1287,7 @@ mod tests {
let cx = Cx::new(&job);
let comp = cx.comp().unwrap();
cx.frame(comp, 0);
let renders = cx.cache.stats().misses;
let renders = cx.caches.of(cx.backend()).stats().misses;
assert!(renders > 0);
for i in 1..10 {
let img = cx.frame(comp, i);
@@ -1012,7 +1297,7 @@ mod tests {
uncached.cache = None;
assert_eq!(img.data, uncached.comp_frame(cid, t).data, "frame {i}");
}
let stats = cx.cache.stats();
let stats = cx.caches.of(cx.backend()).stats();
assert_eq!(stats.misses, renders, "the solid rendered on the first frame only");
assert!(stats.hits >= 9, "{stats:?}");
}
+3 -1
View File
@@ -104,6 +104,8 @@ fn project() -> (Project, ItemId) {
fn run(p: &Project, cid: ItemId, accel: Option<&dyn Accelerator>, s: &RenderSettings, path: &Path) -> Vec<u8> {
let om = OutputModule { bitrate_kbps: 4000, ..OutputModule::for_format(OutputFormat::H264) };
let path = path.to_string_lossy().to_string();
// These tests are about the accelerator's own path: no measuring against the CPU (#3335).
let gpu = std::sync::Mutex::new(Some(effectcraft_render::Backend::Gpu));
let job = Job {
project: p,
footage: &NoFootage,
@@ -114,7 +116,7 @@ fn run(p: &Project, cid: ItemId, accel: Option<&dyn Accelerator>, s: &RenderSett
output: &om,
path: &path,
sink: None,
options: Default::default(),
options: effectcraft_export::JobOptions { renderer: Some(&gpu), ..Default::default() },
nested_switches: true,
};
export(&job, &mut |_| true).expect("export");
+147
View File
@@ -501,3 +501,150 @@ fn a_job_renders_on_one_compositor() {
assert_eq!(cpu.frames.load(Ordering::SeqCst), 0);
assert_eq!(c[0].get_pixel(10, 44).0, [255, 0, 0, 255], "the CPU drew the red block");
}
/// A GPU compositor stand-in that renders the frame exactly as the CPU does, after `delay`, and
/// either as it is or solid green (`green`), counting its frames.
struct TimedAccel {
delay: std::time::Duration,
green: bool,
frames: AtomicUsize,
}
impl TimedAccel {
fn new(delay_ms: u64, green: bool) -> TimedAccel {
TimedAccel { delay: std::time::Duration::from_millis(delay_ms), green, frames: AtomicUsize::new(0) }
}
}
impl Accelerator for TimedAccel {
fn name(&self) -> String {
"timed".into()
}
fn comp_frame(&self, r: &Renderer, comp: ItemId, t: Tick) -> Option<Image> {
self.frames.fetch_add(1, Ordering::SeqCst);
std::thread::sleep(self.delay);
let c = r.project.comp(comp)?;
Some(if self.green { Image::filled(c.width, c.height, [0.0, 1.0, 0.0, 1.0]) } else { r.comp_frame_cpu(comp, t) })
}
fn supports_effect(&self, _id: &str) -> bool {
false
}
fn effects(&self, _chain: &[FxStep], _buf: &Buf, _levels: Option<f32>) -> Option<Buf> {
None
}
}
/// The environment forces the compositor or the GPU frames in flight: the measured choice
/// isn't what these tests check then.
fn measurement_overridden() -> bool {
std::env::var_os(effectcraft_export::RENDERER_ENV).is_some() || std::env::var_os("EFFECTCRAFT_GPU_INFLIGHT").is_some()
}
/// [`run_on`] on a pool of 8 threads (a batch of 8 frames: room to measure both compositors on
/// the first one, whatever the machine), returning the PNGs and the render log.
fn run_pooled(p: &Project, cid: ItemId, s: &RenderSettings, path: &str, options: JobOptions, accel: &dyn Accelerator) -> (Vec<image::RgbaImage>, String) {
let om = OutputModule::for_format(OutputFormat::PngSequence);
let pool = rayon::ThreadPoolBuilder::new().num_threads(8).build().unwrap();
let options = JobOptions { log: RenderLog::PlusSettings, ..options };
let (r, files) = pool.install(|| run_on(p, cid, s, &om, path, options, (Some(accel), false)));
let n = r.expect("export").frames as usize;
let log = files.lock().unwrap().iter().find(|(p, _)| p.ends_with("_RenderLog.txt")).map(|(_, d)| String::from_utf8(d.clone()).unwrap()).expect("log");
((0..n).map(|i| png(&files, i)).collect(), log)
}
#[test]
fn a_transfer_bound_gpu_job_renders_on_the_cpu() {
// #3335: a GPU project whose GPU frames are slower than the CPU's renders the whole job on
// the CPU's frame-parallel compositor, measured on its first frames, and says so in the log.
if measurement_overridden() {
return;
}
let (mut p, cid) = project(FrameRate::new(25, 1), |_, _| {});
p.settings.gpu_acceleration = true;
let s = custom(0.0, 0.8);
let slow = TimedAccel::new(60, true);
let (frames, log) = run_pooled(&p, cid, &s, "/r/a_[#####].png", JobOptions::default(), &slow);
assert_eq!(frames.len(), 20);
// The GPU rendered its warm-up and timed frames only, and none of them is in the output.
assert_eq!(slow.frames.load(Ordering::SeqCst), 3);
for (i, f) in frames.iter().enumerate() {
assert_eq!(f.get_pixel(10, 44).0, [255, 0, 0, 255], "frame {i} is not the CPU's");
}
assert!(log.contains("Renderer: Mercury Software Only (measured on the first frames: GPU "), "{log}");
// The queue item's next output module follows the first one's choice without measuring.
let shared = Mutex::new(None);
let opts = || JobOptions { renderer: Some(&shared), ..Default::default() };
let slow = TimedAccel::new(60, true);
run_pooled(&p, cid, &s, "/r/b_[#####].png", opts(), &slow);
assert_eq!(*shared.lock().unwrap(), Some(effectcraft_render::Backend::Cpu));
let again = TimedAccel::new(0, true);
let (frames, log) = run_pooled(&p, cid, &s, "/r/c_[#####].png", opts(), &again);
assert_eq!(again.frames.load(Ordering::SeqCst), 0);
assert!(frames.iter().all(|f| f.get_pixel(10, 44).0 == [255, 0, 0, 255]));
assert!(log.contains("Renderer: Mercury Software Only (as the queue item's first output module)"), "{log}");
}
#[test]
fn a_gpu_job_that_measures_no_faster_on_the_cpu_stays_on_the_gpu() {
if measurement_overridden() {
return;
}
// A slow CPU (a large comp of eight rotating, translucent full-frame layers, composited
// anew every frame) against a GPU that only fills the frame: every frame is the GPU's.
let (mut p, cid) = project(FrameRate::new(25, 1), |p, comp| {
comp.width = 1920;
comp.height = 1080;
let big = p.add_item("Big", Label::Red, None, ItemKind::Solid(Solid { color: [0.2, 0.4, 0.6], width: 1920, height: 1080, pixel_aspect: 1.0 }));
for k in 0..8 {
let mut l = build::layer(p, comp, "Big", LayerSource::Solid { item: big }, (1920, 1080), None);
l.props.prop_mut("transform/rotation").unwrap().keys =
vec![Keyframe::new(Tick::ZERO, Value::Scalar(k as f64)), Keyframe::new(Tick::from_seconds_f64(1.0), Value::Scalar(90.0 + k as f64))];
l.props.prop_mut("transform/opacity").unwrap().value = Value::Scalar(50.0);
comp.layers.insert(0, l);
}
});
p.settings.gpu_acceleration = true;
let s = custom(0.0, 0.6);
let fast = TimedAccel::new(0, true);
let (frames, log) = run_pooled(&p, cid, &s, "/r/d_[#####].png", JobOptions::default(), &fast);
assert_eq!(frames.len(), 15);
assert_eq!(fast.frames.load(Ordering::SeqCst), 15, "every kept frame, warm-up and measured ones included, is the GPU's");
assert!(frames.iter().all(|f| f.pixels().all(|px| px.0 == [0, 255, 0, 255])));
assert!(log.contains("Renderer: Mercury GPU Acceleration (measured on the first frames: GPU "), "{log}");
}
#[test]
fn gpu_cpu_and_measured_jobs_write_the_same_frames() {
// #3335: where the two compositors agree, a job is the same whichever renders it: the GPU
// (forced), the CPU (Software Only) or the measured choice.
if measurement_overridden() {
return;
}
let (mut p, cid) = project(FrameRate::new(25, 1), |_, _| {});
let s = custom(0.0, 0.8);
let echo = TimedAccel::new(5, false);
p.settings.gpu_acceleration = false;
let (cpu, log) = run_pooled(&p, cid, &s, "/i/a_[#####].png", JobOptions::default(), &echo);
assert!(log.contains("Renderer: Mercury Software Only (the project's renderer)"), "{log}");
p.settings.gpu_acceleration = true;
let gpu_slot = Mutex::new(Some(effectcraft_render::Backend::Gpu));
let (gpu, _) = run_pooled(&p, cid, &s, "/i/b_[#####].png", JobOptions { renderer: Some(&gpu_slot), ..Default::default() }, &echo);
let (auto, _) = run_pooled(&p, cid, &s, "/i/c_[#####].png", JobOptions::default(), &echo);
assert_eq!(cpu.len(), 20);
assert_eq!(cpu, gpu);
assert_eq!(cpu, auto);
}
#[test]
fn renderer_choice_follows_the_measurement() {
use effectcraft_export::{JobRenderer, pick_renderer};
assert_eq!(pick_renderer(Some(117.0), Some(68.0)), JobRenderer::Cpu);
assert_eq!(pick_renderer(Some(68.0), Some(117.0)), JobRenderer::Gpu);
// Ties and noise stay on the GPU: the CPU must be clearly faster.
assert_eq!(pick_renderer(Some(100.0), Some(90.0)), JobRenderer::Gpu);
assert_eq!(pick_renderer(Some(100.0), Some(80.0)), JobRenderer::Cpu);
// Unmeasured or nonsense: the GPU, unless the GPU stopped while measuring.
assert_eq!(pick_renderer(None, None), JobRenderer::Gpu);
assert_eq!(pick_renderer(Some(f64::NAN), Some(1.0)), JobRenderer::Gpu);
assert_eq!(pick_renderer(None, Some(50.0)), JobRenderer::Cpu);
}
+1 -1
View File
@@ -801,7 +801,7 @@ fn read_buffer(e: &mut Enc, buf: &wgpu::Buffer, size: u64) -> Option<Vec<u8>> {
});
e.record(&[]).copy_buffer_to_buffer(buf, 0, &rb, 0, size);
e.submit();
let out = e.g.read_buffer(&rb, row, row, 1).map_err(|e| log::error!("gpu 3d readback: {e}")).ok()?;
let out = e.g.read_buffer_after(&rb, e.last_submission(), row, row, 1).map_err(|e| log::error!("gpu 3d readback: {e}")).ok()?;
Some(out)
}
+144 -32
View File
@@ -1198,18 +1198,99 @@ impl GpuContext {
/// 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<Vec<u8>, String> {
self.read_buffer_after(buffer, None, row, stride, height)
}
/// [`Self::read_buffer`] waiting only for `submission` (the one that wrote `buffer`), not
/// for every submission so far: another frame's GPU work submitted meanwhile doesn't hold
/// this readback up (#3335). `None`: wait for all submitted work.
#[cfg(not(target_arch = "wasm32"))]
pub(crate) fn read_buffer_after(
&self,
buffer: &wgpu::Buffer,
submission: Option<wgpu::SubmissionIndex>,
row: usize,
stride: usize,
height: usize,
) -> Result<Vec<u8>, String> {
self.read_mapped(buffer, submission, |view| crate::readback::packed(view, row, stride, height))
}
/// Map `buffer`, wait for the mapping (only for `submission` when given) and hand the
/// mapped bytes to `read`, on this thread, before unmapping: `read` copies the rows
/// straight into their destination, once (#3335). The copy happens outside `poll_lock`, so
/// it doesn't hold up other threads' readbacks the way copying inside the map callback
/// (which runs during a device poll) would.
#[cfg(not(target_arch = "wasm32"))]
pub(crate) fn read_mapped<R>(
&self,
buffer: &wgpu::Buffer,
submission: Option<wgpu::SubmissionIndex>,
read: impl FnOnce(&[u8]) -> Result<R, String>,
) -> Result<R, String> {
let (tx, rx) = std::sync::mpsc::channel();
self.map_readback(buffer, row, stride, height, move |r| {
let ticket = self.readbacks.start(move |r| {
let _ = tx.send(r);
})?;
if !ticket.active() {
return Err("GPU readback cancelled".into());
}
let b = buffer.clone();
buffer.map_async(wgpu::MapMode::Read, .., move |r| {
// Timed out or cancelled: nobody reads the mapping.
if !ticket.active() {
b.unmap();
return;
}
// Mapped: the waiting thread reads the rows and unmaps (an empty vector = mapped).
ticket.complete(r.map(|_| Vec::new()).map_err(|e| format!("GPU buffer mapping: {e}")));
});
// On any failure, unmap (harmless when not mapped; cancels a pending map) so a mapped
// buffer can never be handed out again.
let fail = |e: String| {
buffer.unmap();
e
};
self.wait_mapped(&rx, submission).map_err(fail)?.map_err(fail)?;
let result = {
let view = buffer.get_mapped_range(..).map_err(|e| format!("GPU mapped range: {e}"));
view.and_then(|view| read(&view))
};
buffer.unmap();
result
}
/// Poll until `rx` reports the mapping (see [`Self::read_mapped`]): 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. While another thread holds `poll_lock` it is
/// polling the device, which delivers this mapping too: wait for it instead of queueing.
#[cfg(not(target_arch = "wasm32"))]
fn wait_mapped(
&self,
rx: &std::sync::mpsc::Receiver<Result<Vec<u8>, String>>,
submission: Option<wgpu::SubmissionIndex>,
) -> Result<Result<Vec<u8>, String>, String> {
let mut deadline = None;
loop {
self.readbacks.check()?;
if let Ok(result) = rx.try_recv() {
return result;
return Ok(result);
}
let (remaining, slice, poll_res) = {
let _guard = self.poll_lock.lock().unwrap_or_else(|e| e.into_inner());
let _guard = match self.poll_lock.try_lock() {
Ok(g) => g,
Err(std::sync::TryLockError::Poisoned(e)) => e.into_inner(),
Err(std::sync::TryLockError::WouldBlock) => {
match rx.recv_timeout(std::time::Duration::from_millis(1)) {
Ok(result) => return Ok(result),
Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {}
Err(e) => return Err(format!("GPU readback completion channel: {e}")),
}
// The ticket's own deadline still bounds a wait that never gets the lock.
continue;
}
};
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());
@@ -1217,12 +1298,12 @@ impl GpuContext {
(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) });
let res = self.device.poll(wgpu::PollType::Wait { submission_index: submission.clone(), timeout: Some(slice) });
(remaining, slice, res)
}
};
if let Ok(result) = rx.try_recv() {
return result;
return Ok(result);
}
if remaining.is_zero() {
let error = "GPU readback receive timed out; context retired".to_string();
@@ -1241,7 +1322,7 @@ impl GpuContext {
}
// 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,
Ok(result) => return Ok(result),
Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {}
Err(e) => return Err(format!("GPU readback completion channel: {e}")),
}
@@ -1253,6 +1334,28 @@ impl GpuContext {
Err("synchronous GPU readback is unavailable in the browser".into())
}
#[cfg(target_arch = "wasm32")]
pub(crate) fn read_buffer_after(
&self,
_buffer: &wgpu::Buffer,
_submission: Option<wgpu::SubmissionIndex>,
_row: usize,
_stride: usize,
_height: usize,
) -> Result<Vec<u8>, String> {
Err("synchronous GPU readback is unavailable in the browser".into())
}
#[cfg(target_arch = "wasm32")]
pub(crate) fn read_mapped<R>(
&self,
_buffer: &wgpu::Buffer,
_submission: Option<wgpu::SubmissionIndex>,
_read: impl FnOnce(&[u8]) -> Result<R, String>,
) -> Result<R, String> {
Err("synchronous GPU readback is unavailable in the browser".into())
}
/// Switch to deferred readbacks (see [`crate::deferred`]): readbacks are never waited for.
pub fn set_deferred(&mut self, on: bool) {
self.readbacks.cancel_pending("GPU readback mode changed".into());
@@ -1952,6 +2055,11 @@ impl<'g> Enc<'g> {
dst
}
/// The last submission of this encoder's work (what a readback after it waits for).
pub(crate) fn last_submission(&self) -> Option<wgpu::SubmissionIndex> {
self.last_submit.clone()
}
/// Submit recorded work.
pub fn submit(&mut self) {
if let Some((enc, id)) = self.enc.take() {
@@ -1994,13 +2102,13 @@ impl<'g> Enc<'g> {
let layout = crate::readback::Layout::new(img.width, img.height, 16, self.g.device.limits().max_buffer_size)
.map_err(|e| log::error!("gpu image readback: {e}"))
.ok()?;
let mut data = Vec::new();
data.try_reserve_exact(layout.len / 16).map_err(|e| log::error!("gpu image allocation: {e}")).ok()?;
data.resize(layout.len / 16, [0.0; 4]);
let mut out = Image { width: img.width, height: img.height, data };
let dst: &mut [u8] = bytemuck::cast_slice_mut(&mut out.data);
self.read_texture_into(&img.texture, [0, 0, img.width, img.height], 16, dst)?;
Some(out)
// The mapped rows go straight into the image's pixels: one copy (#3335).
let data: Vec<[f32; 4]> = self.read_texture_with(&img.texture, [0, 0, img.width, img.height], 16, crate::readback::rows::<[f32; 4]>)?;
if data.len() != layout.len / 16 {
log::error!("gpu image readback size mismatch");
return None;
}
Some(Image { width: img.width, height: img.height, data })
}
/// Tightly packed texel bytes of a texture (`bpp` bytes per pixel).
@@ -2013,27 +2121,32 @@ impl<'g> Enc<'g> {
pub fn read_rect(&mut self, texture: &wgpu::Texture, x: u32, y: u32, w: u32, h: u32, bpp: u32) -> Option<Vec<u8>> {
let layout =
crate::readback::Layout::new(w, h, bpp, self.g.device.limits().max_buffer_size).map_err(|e| log::error!("gpu texture readback: {e}")).ok()?;
let mut out = Vec::new();
out.try_reserve_exact(layout.len).map_err(|e| log::error!("gpu readback allocation: {e}")).ok()?;
out.resize(layout.len, 0);
self.read_texture_into(texture, [x, y, w, h], bpp, &mut out)?;
let out: Vec<u8> = self.read_texture_with(texture, [x, y, w, h], bpp, crate::readback::rows::<u8>)?;
if out.len() != layout.len {
log::error!("gpu texture readback size mismatch");
return None;
}
Some(out)
}
/// [`Enc::read_rect`] into `out` (`w × bpp × h` bytes), through a reused staging buffer.
/// `rect` is `[x, y, w, h]`.
fn read_texture_into(&mut self, texture: &wgpu::Texture, rect: [u32; 4], bpp: u32, out: &mut [u8]) -> Option<()> {
/// Read `rect` (`[x, y, w, h]`) of `texture` back through a reused staging buffer: `read`
/// gets the mapped bytes, the row length, the padded row stride and the row count, and
/// copies the rows into their destination itself (one copy, #3335). Waits only for this
/// encoder's own submission, so frames rendering on other threads don't delay it.
fn read_texture_with<R>(
&mut self,
texture: &wgpu::Texture,
rect: [u32; 4],
bpp: u32,
read: impl FnOnce(&[u8], usize, usize, usize) -> Result<R, String>,
) -> Option<R> {
let [x, y, w, h] = rect;
if !self.g.can_wait() {
return None;
}
self.g.transfers.count(false, out.len());
let layout =
crate::readback::Layout::new(w, h, bpp, self.g.device.limits().max_buffer_size).map_err(|e| log::error!("gpu texture readback: {e}")).ok()?;
if out.len() != layout.len {
log::error!("gpu readback destination size mismatch");
return None;
}
self.g.transfers.count(false, layout.len);
let row = layout.row as usize;
let padded = layout.stride;
let size = layout.size;
@@ -2051,12 +2164,11 @@ impl<'g> Enc<'g> {
wgpu::Extent3d { width: w, height: h, depth_or_array_layers: 1 },
);
self.submit();
let bytes = self.g.read_buffer(&buf, row, padded as usize, h as usize).map_err(|e| log::error!("gpu texture readback: {e}")).ok()?;
if bytes.len() != out.len() {
log::error!("gpu texture readback size mismatch");
return None;
}
out.copy_from_slice(&bytes);
let out = self
.g
.read_mapped(&buf, self.last_submit.clone(), |view| read(view, row, padded as usize, h as usize))
.map_err(|e| log::error!("gpu texture readback: {e}"))
.ok()?;
// Keep a few staging buffers per size (frames read back at the same size every time).
// Layers whose bounds change every frame read back at a new size each time: drop the
// other sizes once over budget, or mapped host memory grows for the whole render (#2081).
@@ -2070,7 +2182,7 @@ impl<'g> Enc<'g> {
v.push(buf);
}
}
Some(())
Some(out)
}
/// [`Enc::read_texture`] without waiting for the GPU (the browser's main thread can't): `done`
+41 -5
View File
@@ -367,20 +367,32 @@ thread_local! {
/// Validate every row before publishing tightly packed bytes.
pub(crate) fn packed(view: &[u8], row: usize, stride: usize, height: usize) -> Result<Vec<u8>, Error> {
if row == 0 || height == 0 || stride < row {
rows(view, row, stride, height)
}
/// The `height` rows of `row` bytes each, `stride` bytes apart in the mapped `view`, as tightly
/// packed `T`s: one copy, straight from the mapped rows into a vector reserved once (no
/// zero-filled buffer first, no intermediate byte vector). Every row is validated before any
/// is copied. A row that is not aligned for `T` is decoded value by value (same bytes).
pub(crate) fn rows<T: bytemuck::Pod>(view: &[u8], row: usize, stride: usize, height: usize) -> Result<Vec<T>, Error> {
let unit = std::mem::size_of::<T>().max(1);
if row == 0 || height == 0 || stride < row || !row.is_multiple_of(unit) {
return Err("invalid GPU readback layout".into());
}
let size = row.checked_mul(height).ok_or("GPU readback output size overflow")?;
let size = row.checked_mul(height).ok_or("GPU readback output size overflow")? / unit;
let end = (height - 1).checked_mul(stride).and_then(|s| s.checked_add(row)).ok_or("GPU readback mapped size overflow")?;
if view.len() < end {
return Err(format!("short GPU readback: {} bytes, needs {end}", view.len()));
}
let mut out = Vec::new();
let mut out: Vec<T> = Vec::new();
out.try_reserve_exact(size).map_err(|e| format!("allocating GPU readback output: {e}"))?;
for y in 0..height {
let start = y.checked_mul(stride).ok_or("GPU readback row overflow")?;
let bytes = view.get(start..start + row).ok_or("short GPU readback row")?;
out.extend_from_slice(bytes);
let bytes = start.checked_add(row).and_then(|e| view.get(start..e)).ok_or("short GPU readback row")?;
match bytemuck::try_cast_slice::<u8, T>(bytes) {
Ok(values) => out.extend_from_slice(values),
Err(_) => out.extend(bytes.chunks_exact(unit).map(bytemuck::pod_read_unaligned::<T>)),
}
}
Ok(out)
}
@@ -658,4 +670,28 @@ mod tests {
assert!(packed(&[0; 7], 4, 4, 2).is_err());
assert_eq!(packed(&[1, 2, 0, 0, 3, 4], 2, 4, 2).unwrap(), vec![1, 2, 3, 4]);
}
/// #3335: rows go from the mapped buffer straight into the destination type, skipping the
/// row padding, whether or not the mapped bytes are aligned for it.
#[test]
fn rows_copy_once_into_the_destination_type() {
// Two rows of two RGBA f32 pixels, padded to a 48-byte stride.
let px = |i: f32| [i, i + 0.25, i + 0.5, 1.0];
let want = vec![px(1.0), px(2.0), px(3.0), px(4.0)];
let mut view = Vec::new();
for r in want.chunks(2) {
view.extend_from_slice(bytemuck::cast_slice(r));
view.extend_from_slice(&[0xAB; 16]);
}
assert_eq!(rows::<[f32; 4]>(&view, 32, 48, 2).unwrap(), want);
// The same bytes one byte off any f32 alignment decode the same.
let mut shifted = vec![0u8];
shifted.extend_from_slice(&view);
assert_eq!(rows::<[f32; 4]>(&shifted[1..], 32, 48, 2).unwrap(), want);
// A row that isn't whole pixels, a short last row and a stride under the row fail.
assert!(rows::<[f32; 4]>(&view, 30, 48, 2).is_err());
assert!(rows::<[f32; 4]>(&view[..view.len() - 17], 32, 48, 2).is_err());
assert!(rows::<[f32; 4]>(&view, 32, 16, 2).is_err());
assert_eq!(rows::<u8>(&[1, 2, 9, 3, 4], 2, 3, 2).unwrap(), vec![1, 2, 3, 4]);
}
}
+43
View File
@@ -1247,6 +1247,49 @@ fn readback_staging_stays_bounded_across_sizes() {
assert!(held <= (256 << 20) + 2 * 1024 * 16 * 376, "staging buffers hold {held} bytes");
}
/// #3335: a readback copies the mapped rows straight into the image (no zero-filled buffer,
/// no intermediate byte vector), skipping each row's padding: odd widths, sub-rectangles and
/// readbacks from several threads at once (each waiting for its own submission) return the
/// uploaded pixels exactly.
#[test]
fn single_copy_readback_returns_the_exact_pixels() {
let Some(g) = gpu() else { return };
if !g.ctx.can_wait() {
return;
}
let image = |w: u32, h: u32, seed: f32| {
let mut img = Image::new(w, h);
for (i, p) in img.data.iter_mut().enumerate() {
let v = i as f32 * 0.37 + seed;
*p = [v, -v, v * 1e-3, 1.0 / (1.0 + v)];
}
img
};
// 37 × 16 bytes = 592 per row: padded to 768 in the staging buffer.
let src = image(37, 5, 0.5);
let up = g.ctx.upload_image(&src).expect("upload");
let mut e = crate::context::Enc::new(&g.ctx);
let back = e.download(&up).expect("download");
assert_eq!((back.width, back.height), (37, 5));
assert_eq!(back.data, src.data);
let rect = e.read_rect(&up.texture, 3, 1, 5, 2, 16).expect("read_rect");
let want: Vec<[f32; 4]> = (1..3).flat_map(|y| (3..8).map(move |x| (x, y))).map(|(x, y)| src.data[src.idx(x, y)]).collect();
assert_eq!(rect, bytemuck::cast_slice::<[f32; 4], u8>(&want));
// Frames on several threads, each with its own encoder and submission.
std::thread::scope(|s| {
for t in 0..4 {
s.spawn(move || {
for k in 0..3 {
let src = image(61 + t, 7 + k, (t * 10 + k) as f32);
let up = g.ctx.upload_image(&src).expect("upload");
let back = crate::context::Enc::new(&g.ctx).download(&up).expect("download");
assert_eq!(back.data, src.data, "thread {t} frame {k}");
}
});
}
});
}
/// A static adjustment layer with a time-dependent effect (Noise) keeps its footprint key
/// across frames (the footprint is the source through the masks, not the effects), so the GPU
/// uploads nothing after the first frame; matches the CPU.
+1 -1
View File
@@ -147,7 +147,7 @@ pub(crate) fn record(e: &mut Enc, yp: &YuvPipe, img: &GpuImage, bg: [f32; 3]) ->
pub(crate) fn convert(e: &mut Enc, yp: &YuvPipe, img: &GpuImage, bg: [f32; 3]) -> Option<Vec<u8>> {
let stg = record(e, yp, img, bg)?;
let n = planes_len(img.width, img.height) as usize;
e.g.read_buffer(&stg, n, n, 1).map_err(|err| log::error!(target: "effectcraft_gpu", "yuv420 readback: {err}")).ok()
e.g.read_buffer_after(&stg, e.last_submission(), n, n, 1).map_err(|err| log::error!(target: "effectcraft_gpu", "yuv420 readback: {err}")).ok()
}
#[cfg(test)]
+2
View File
@@ -62,6 +62,8 @@ impl Exporter for FileExporter {
label: job.label.clone(),
storage: job.storage,
overflow: job.project.render_prefs.overflow_folders.clone(),
caches: job.caches.map(|c| effectcraft_export::JobCaches { gpu: &c.gpu, cpu: &c.cpu }),
renderer: job.renderer,
},
nested_switches: job.nested_switches,
};
+141
View File
@@ -0,0 +1,141 @@
//! Render Queue benchmark (#3335): a 1080p 32 bpc job on the GPU, the CPU or the measured choice.
//!
//! `cargo run --release -p effectcraft-ui-egui --example rq_bench -- [options]`
//!
//! It builds a comp like a compositing shot: a 1920 × 1080 OpenEXR sequence (float RGBA with a
//! Cryptomatte ID channel and manifest, generated once into `target/rq-bench/media`, no
//! committed media) under Gaussian Blur, Curves and the Cryptomatte effect and a moving
//! translucent solid, at 32 bpc. It queues the comp twice (the second item shows a warm start) and
//! renders the queue as the app does, printing each item's milliseconds per frame and a hash of
//! the written pixels (equal hashes: identical output).
//!
//! Options: `--renderer gpu|cpu` (the project's renderer, default gpu; with gpu the Render
//! Queue measures both unless `EFFECTCRAFT_RQ_RENDERER` says otherwise), `--frames N` (default
//! 60), `--format png|exr|tiff` (default png), `--out DIR` (default `target/rq-bench/out`).
use std::sync::Arc;
use effectcraft_engine::Session;
use effectcraft_engine::raster::channels3d::{crypto_float, crypto_hash};
use serde_json::{Value, json};
const W: usize = 1920;
const H: usize = 1080;
fn arg(name: &str) -> Option<String> {
let a: Vec<String> = std::env::args().collect();
a.iter().position(|x| x == name).and_then(|i| a.get(i + 1).cloned())
}
fn run(s: &mut Session, cmd: &str, p: Value) -> Value {
s.execute(cmd, p).unwrap_or_else(|e| panic!("{cmd}: {e}"))
}
/// The plate's object at pixel (x, y) in frame `f`: 24 bands that drift sideways.
fn object(x: usize, y: usize, f: usize) -> String {
format!("obj{}", ((x + f * 7) / 80 + y / 270 * 24) % 96)
}
/// Frame `f` of the plate: float colour over 1.0 (32 bpc), alpha, and Cryptomatte IDs.
fn plate(path: &std::path::Path, f: usize) {
use exr::prelude::*;
let px = |c: usize| -> Vec<f32> {
(0..W * H)
.map(|i| {
let (x, y) = (i % W, i / W);
let v = ((x as f32 * 0.003 + f as f32 * 0.05 + c as f32).sin() * 0.5 + 0.5) * (1.0 + y as f32 / H as f32);
if c == 3 { 1.0 } else { v }
})
.collect()
};
let ids: Vec<f32> = (0..W * H).map(|i| crypto_float(crypto_hash(&object(i % W, i / W, f)))).collect();
let ch = |n: &str, v: Vec<f32>| AnyChannel::new(n, FlatSamples::F32(v));
let channels = AnyChannels::sort(
vec![ch("R", px(0)), ch("G", px(1)), ch("B", px(2)), ch("A", px(3)), ch("CryptoObject00.R", ids), ch("CryptoObject00.G", vec![1.0; W * H])].into(),
);
let mut layer = Layer::new((W, H), LayerAttributes::default(), Encoding::FAST_LOSSLESS, channels);
let names: serde_json::Map<String, Value> = (0..96).map(|k| format!("obj{k}")).map(|n| (n.clone(), json!(format!("{:08x}", crypto_hash(&n))))).collect();
for (key, v) in [("name", "CryptoObject".to_string()), ("manifest", Value::Object(names).to_string())] {
layer.attributes.other.insert(Text::from(format!("cryptomatte/f834d0a/{key}").as_str()), AttributeValue::Text(Text::from(v.as_str())));
}
Image::from_layer(layer).write().to_file(path).expect("write plate");
}
/// FNV-1a over every written frame file, in name order (equal: identical output).
fn hash_dir(dir: &std::path::Path) -> (usize, u64) {
let mut names: Vec<_> = std::fs::read_dir(dir).map(|d| d.filter_map(|e| e.ok()).map(|e| e.path()).collect()).unwrap_or_default();
names.retain(|p| !p.to_string_lossy().ends_with(".txt"));
names.sort();
let mut h = 0xcbf2_9ce4_8422_2325u64;
for p in &names {
for b in std::fs::read(p).unwrap_or_default() {
h = (h ^ b as u64).wrapping_mul(0x100_0000_01b3);
}
}
(names.len(), h)
}
fn main() {
let root = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../target/rq-bench");
let frames: usize = arg("--frames").and_then(|n| n.parse().ok()).unwrap_or(60);
let format = arg("--format").unwrap_or_else(|| "png".into());
let gpu = arg("--renderer").as_deref() != Some("cpu");
let out = arg("--out").map(std::path::PathBuf::from).unwrap_or_else(|| root.join("out"));
let media = root.join("media");
std::fs::create_dir_all(&media).expect("media folder");
let first = media.join("plate_0000.exr");
for f in 0..frames {
let p = media.join(format!("plate_{f:04}.exr"));
if !p.exists() {
if f == 0 {
eprintln!("writing the {frames}-frame EXR plate (once)…");
}
plate(&p, f);
}
}
let mut s = effectcraft_host::session();
let accel = effectcraft_gpu::Gpu::try_headless().unwrap_or_else(|e| panic!("no GPU: {e}"));
eprintln!("GPU: {}", effectcraft_engine::render::Accelerator::name(&accel));
s.accel = Some(Arc::new(accel));
let r = run(&mut s, "file.import", json!({"paths": [first.to_string_lossy()], "sequence": true}));
let footage = r["items"][0].as_u64().expect("the plate");
let secs = frames as f64 / 30.0;
run(&mut s, "comp.new", json!({"name": "Shot", "width": W, "height": H, "frameRate": 30, "duration": secs}));
let plate = run(&mut s, "layer.addItem", json!({"item": footage}))["layer"].clone();
run(&mut s, "effect.apply", json!({"layers": [plate], "effect": "Gaussian Blur"}));
run(&mut s, "prop.set", json!({"layer": plate, "path": "effects/#1/blurriness", "value": 12.0}));
run(&mut s, "effect.apply", json!({"layers": [plate], "effect": "Curves"}));
run(&mut s, "effect.apply", json!({"layers": [plate], "effect": "Cryptomatte"}));
let solid = run(&mut s, "layer.newSolid", json!({"color": "#d04020", "width": 400, "height": 300}))["layer"].clone();
run(&mut s, "prop.addKey", json!({"layer": solid, "path": "transform/position", "time": 0.0, "value": [200.0, 540.0]}));
run(&mut s, "prop.addKey", json!({"layer": solid, "path": "transform/position", "time": secs, "value": [1720.0, 540.0]}));
run(&mut s, "prop.set", json!({"layer": solid, "path": "transform/opacity", "value": 70.0}));
let mut p = (*s.project).clone();
p.settings.bit_depth = effectcraft_engine::project::BitDepth::Bpc32;
p.settings.gpu_acceleration = gpu;
s.project = Arc::new(p);
let _ = std::fs::remove_dir_all(&out);
let dirs = [out.join("a"), out.join("b")];
for d in &dirs {
std::fs::create_dir_all(d).expect("output folder");
let path = d.join(format!("f_[####].{format}"));
run(&mut s, "renderQueue.add", json!({"comp": "Shot", "format": format, "output": path.to_string_lossy()}));
}
let t0 = std::time::Instant::now();
run(&mut s, "renderQueue.render", json!({"wait": true}));
let total = t0.elapsed().as_secs_f64();
println!(
"renderer {} (EFFECTCRAFT_RQ_RENDERER={:?}, EFFECTCRAFT_GPU_INFLIGHT={:?}), {frames} frames {W}x{H} 32 bpc, {format}",
if gpu { "gpu" } else { "cpu" },
std::env::var("EFFECTCRAFT_RQ_RENDERER").ok(),
std::env::var("EFFECTCRAFT_GPU_INFLIGHT").ok()
);
for (i, (item, d)) in s.project.render_queue.iter().zip(&dirs).enumerate() {
let secs = item.render_time.unwrap_or(f64::NAN);
let (n, h) = hash_dir(d);
println!(" item {}: {:?}, {:.1} ms/frame, {n} files, hash {h:016x}", i + 1, item.status, secs * 1000.0 / frames as f64);
}
println!(" queue: {total:.2} s");
}