mirror of
https://github.com/storytold/filmcraft.git
synced 2026-10-24 10:10:05 -06:00
fix(export): prevent nested renderer-pool deadlocks
Bound outer export batches by actual renderer-pool capacity before nested Rayon work can steal a sibling frame. Cover image sequences, GIF, stepped export and overlapping audio with the original one-renderer regression.
Addresses #1956.
Full original GPU contribution preserved at `preserve/861-gpu-export-original` (674e8cd68d).
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
@@ -297,7 +297,8 @@ impl Exporter {
|
||||
(self.f1 - self.f0).max(0) as u64
|
||||
}
|
||||
|
||||
/// Frames per [`Self::step`] (default: one per core); 1 keeps web UI frames short.
|
||||
/// Requested frames per [`Self::step`] (default: one per core), capped by GPU renderer
|
||||
/// capacity and the overlap memory budget. 1 keeps web UI frames short.
|
||||
pub fn set_batch(&mut self, n: i64) {
|
||||
self.batch = n.max(1);
|
||||
}
|
||||
@@ -333,9 +334,10 @@ impl Exporter {
|
||||
self.ahead.as_ref().map_or(0, |a| a.frames.len())
|
||||
}
|
||||
|
||||
/// Frames in the next batch: [`Self::set_batch`], lowered to the memory budget when overlapping.
|
||||
/// Frames in the next batch, bounded before Rayon can steal nested frame tasks.
|
||||
fn batch_len(&self) -> i64 {
|
||||
if self.overlapping() { overlap_batch(self.batch, frame_bytes(self.pipe.w, self.pipe.h, self.pipe.hdr_out), self.budget) } else { self.batch }
|
||||
let batch = self.pipe.render_batch_limit(self.batch);
|
||||
if self.overlapping() { overlap_batch(batch, frame_bytes(self.pipe.w, self.pipe.h, self.pipe.hdr_out), self.budget) } else { batch }
|
||||
}
|
||||
|
||||
fn sample_at_frame(&self, f: i64, sr: u32) -> i64 {
|
||||
|
||||
@@ -1924,6 +1924,7 @@ pub fn export(project: &Arc<Project>, seq: ItemId, settings: &ExportSettings, so
|
||||
}
|
||||
f if f.is_image_sequence() || f == Format::Gif => {
|
||||
let pipe = pipeline::Pipeline::new(project.clone(), seq, settings, false)?;
|
||||
let batch = pipe.render_batch_limit(batch);
|
||||
let (f0, f1) = frame_span(pipe.rate, range);
|
||||
let count = (f1 - f0) as u64;
|
||||
let (w, h) = (pipe.w, pipe.h);
|
||||
|
||||
@@ -52,6 +52,8 @@ impl Geometry {
|
||||
/// GPU renderers shared by the export workers: a worker takes a free one for a frame and puts it
|
||||
/// back (a single renderer behind a mutex made the workers queue, GPU export ~20 % slower than CPU).
|
||||
struct RendererPool {
|
||||
/// Initial capacity, also the maximum number of frame tasks admitted at once.
|
||||
capacity: usize,
|
||||
renderers: std::sync::Mutex<Vec<Box<dyn FrameRenderer>>>,
|
||||
available: std::sync::Condvar,
|
||||
/// A renderer panicked: the pool is retired and the rest of the export renders on the CPU.
|
||||
@@ -62,13 +64,17 @@ impl RendererPool {
|
||||
/// `None` for an empty list (a pool nobody could take from).
|
||||
fn new(renderers: Vec<Box<dyn FrameRenderer>>) -> Option<Self> {
|
||||
(!renderers.is_empty()).then(|| Self {
|
||||
capacity: renderers.len(),
|
||||
renderers: std::sync::Mutex::new(renderers),
|
||||
available: std::sync::Condvar::new(),
|
||||
failed: std::sync::atomic::AtomicBool::new(false),
|
||||
})
|
||||
}
|
||||
|
||||
/// Run `f` with a free renderer, waiting for one if all are busy. `None` once a renderer has
|
||||
/// Run `f` with a free renderer. Frame batches must respect `Pipeline::render_batch_limit`:
|
||||
/// a renderer can run nested Rayon work, which may steal another frame on the same thread.
|
||||
/// Admitting more frame tasks than renderers could then deadlock this condition variable
|
||||
/// against the suspended outer frame's renderer. `None` once a renderer has
|
||||
/// panicked: the panicking renderer is dropped (its state can't be trusted) and so are the
|
||||
/// others as they come back, so the caller falls back to the CPU for the rest of the export.
|
||||
fn with<R>(&self, f: impl FnOnce(&mut dyn FrameRenderer) -> R) -> Option<R> {
|
||||
@@ -146,6 +152,16 @@ impl Drop for Pipeline {
|
||||
}
|
||||
|
||||
impl Pipeline {
|
||||
/// Bound outer frame tasks before spawning them, not inside a Rayon job. A renderer can
|
||||
/// invoke nested parallel decoding/conversion; those joins may steal another frame task.
|
||||
/// With at most one active batch and no more tasks than the initial renderer capacity,
|
||||
/// every stolen frame can acquire a renderer without waiting for an ancestor to return it.
|
||||
/// Keep the immutable capacity even after a failed renderer retires the pool.
|
||||
pub(crate) fn render_batch_limit(&self, requested: i64) -> i64 {
|
||||
let requested = requested.max(1);
|
||||
self.renderer.as_ref().map_or(requested, |pool| requested.min(i64::try_from(pool.capacity).unwrap_or(i64::MAX)))
|
||||
}
|
||||
|
||||
pub fn new(project: Arc<Project>, seq: ItemId, settings: &ExportSettings, hdr_out: bool) -> Result<Self> {
|
||||
let q = project.sequence(seq).ok_or(ExportError::NoSequence)?;
|
||||
q.settings.validate().map_err(ExportError::Unsupported)?;
|
||||
|
||||
@@ -0,0 +1,153 @@
|
||||
//! A renderer may enter Rayon while holding its pool slot (upload/readback conversion does).
|
||||
//! Yielding can then run another frame on the same thread. Exercise the actual export entry
|
||||
//! points with a one-slot factory, including hosts that request oversized stepped batches.
|
||||
//! Each case runs in a bounded child so a regression fails instead of hanging the test runner.
|
||||
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::time::Duration;
|
||||
|
||||
use filmcraft_export::{ExportSettings, Exporter, Format, FrameRenderer, GpuRendering, Progress, Step, export, gpu_render_stats, register_frame_renderer};
|
||||
use filmcraft_media::generators::GeneratorSource;
|
||||
use filmcraft_media::{Generator, MediaSource};
|
||||
use filmcraft_project::{ItemId, ItemKind, Label, MediaClip, MediaRef, Project, SequenceSettings, TrackKind};
|
||||
use filmcraft_render::{Image, RenderOptions, SourceMap, SourceProvider};
|
||||
use filmcraft_time::{FrameRate, Tick, TimeRange};
|
||||
use rayon::prelude::*;
|
||||
|
||||
const CHILD: &str = "FILMCRAFT_RENDERER_REENTRANCY_CASE";
|
||||
const FRAMES: i64 = 6;
|
||||
static CREATED: AtomicUsize = AtomicUsize::new(0);
|
||||
static RENDERED: AtomicUsize = AtomicUsize::new(0);
|
||||
|
||||
struct YieldingRenderer;
|
||||
|
||||
impl FrameRenderer for YieldingRenderer {
|
||||
fn render(&mut self, project: &Project, seq: ItemId, t: Tick, opts: RenderOptions, sources: &dyn SourceProvider) -> Option<Image> {
|
||||
// Deterministically offer this thread to the queued sibling frame before returning the
|
||||
// renderer. With one Rayon worker and one renderer, an uncapped batch deadlocks here.
|
||||
rayon::yield_now();
|
||||
let checksum: usize = (0..4096usize).into_par_iter().map(|x| x ^ 37).sum();
|
||||
assert!(checksum > 0);
|
||||
RENDERED.fetch_add(1, Ordering::Relaxed);
|
||||
Some(filmcraft_render::render_sequence(project, seq, t, opts, sources))
|
||||
}
|
||||
}
|
||||
|
||||
fn factory() -> Option<Box<dyn FrameRenderer>> {
|
||||
// A factory may stop before the requested renderer count (e.g. allocation pressure).
|
||||
// Batch safety must use the successfully created capacity, not the requested capacity.
|
||||
(CREATED.fetch_add(1, Ordering::Relaxed) == 0).then(|| Box::new(YieldingRenderer) as Box<dyn FrameRenderer>)
|
||||
}
|
||||
|
||||
fn project() -> (Arc<Project>, ItemId, SourceMap) {
|
||||
let mut project = Project::new("renderer reentrancy");
|
||||
let rate = FrameRate::FPS_24;
|
||||
let duration = rate.tick_of(FRAMES);
|
||||
let seq = project.new_sequence("test", SequenceSettings { width: 32, height: 32, frame_rate: rate, ..Default::default() }, 1, 1, None);
|
||||
let mut sources = SourceMap::default();
|
||||
for (generator, kind) in
|
||||
[(Generator::ColorMatte { color: [1.0, 0.0, 0.0, 1.0] }, TrackKind::Video), (Generator::Tone { hz: 440.0, db: -6.0 }, TrackKind::Audio)]
|
||||
{
|
||||
let source = GeneratorSource::new(generator, 32, 32, rate, duration);
|
||||
let info = source.info().clone();
|
||||
let id = project.add_item(
|
||||
&info.name.clone(),
|
||||
Label::Iris,
|
||||
ItemKind::Media(MediaClip {
|
||||
media: MediaRef::Generator(source.generator.clone()),
|
||||
info,
|
||||
interpret: Default::default(),
|
||||
mark_in: None,
|
||||
mark_out: None,
|
||||
markers: vec![],
|
||||
offline: false,
|
||||
proxy: None,
|
||||
identity: None,
|
||||
}),
|
||||
None,
|
||||
);
|
||||
let item = project.make_track_item(id, kind, Tick::ZERO, TimeRange::new(Tick::ZERO, duration), rate).unwrap();
|
||||
let sequence = project.sequence_mut(seq).unwrap();
|
||||
match kind {
|
||||
TrackKind::Video => sequence.video_tracks[0].items.push(item),
|
||||
TrackKind::Audio => sequence.audio_tracks[0].items.push(item),
|
||||
}
|
||||
sources.0.insert(id, Arc::new(source));
|
||||
}
|
||||
(Arc::new(project), seq, sources)
|
||||
}
|
||||
|
||||
fn run_case(case: &str) {
|
||||
let (threads, format, overlap, audio) = match case {
|
||||
"png" => (1, Format::PngSequence, false, false),
|
||||
"gif" => (1, Format::Gif, false, false),
|
||||
"stepped" => (1, Format::ProRes, false, false),
|
||||
"overlap-audio" => (8, Format::ProRes, true, true),
|
||||
_ => panic!("unknown regression case: {case}"),
|
||||
};
|
||||
register_frame_renderer(factory);
|
||||
let (project, seq, sources) = project();
|
||||
let dir = std::env::temp_dir().join(format!("fc-renderer-reentrancy-{}", std::process::id()));
|
||||
std::fs::create_dir(&dir).unwrap();
|
||||
let settings = ExportSettings {
|
||||
format,
|
||||
path: dir.join("output").to_string_lossy().into_owned(),
|
||||
gpu_rendering: GpuRendering::Auto,
|
||||
include_audio: audio,
|
||||
..Default::default()
|
||||
};
|
||||
let progress = Progress::default();
|
||||
rayon::ThreadPoolBuilder::new().num_threads(threads).build().unwrap().install(|| {
|
||||
if format == Format::ProRes {
|
||||
let mut exporter = Exporter::new(project.clone(), seq, &settings, &progress).unwrap();
|
||||
// Host overrides must not bypass the resource-capacity bound.
|
||||
exporter.set_batch(64);
|
||||
exporter.set_overlap(overlap);
|
||||
loop {
|
||||
if let Step::Done(report) = exporter.step(&sources, &progress).unwrap() {
|
||||
assert_eq!(report.frames, FRAMES as u64);
|
||||
break;
|
||||
}
|
||||
}
|
||||
if overlap {
|
||||
assert!(exporter.overlapped_steps() > 0);
|
||||
}
|
||||
} else {
|
||||
assert_eq!(export(&project, seq, &settings, &sources, &progress).unwrap().frames, FRAMES as u64);
|
||||
}
|
||||
});
|
||||
assert_eq!(RENDERED.load(Ordering::Relaxed), FRAMES as usize);
|
||||
assert_eq!(gpu_render_stats().frames, FRAMES as u64, "every frame uses the registered renderer");
|
||||
assert_eq!(gpu_render_stats().fallbacks, 0, "contention must not silently switch frames to CPU");
|
||||
assert_eq!(progress.done.load(Ordering::Relaxed), FRAMES as u64);
|
||||
std::fs::remove_dir_all(dir).unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn nested_rayon_renderers_finish_every_export_path() {
|
||||
if let Ok(case) = std::env::var(CHILD) {
|
||||
let (done, finished) = std::sync::mpsc::channel();
|
||||
let diagnostic = case.clone();
|
||||
let watchdog = std::thread::spawn(move || {
|
||||
if finished.recv_timeout(Duration::from_secs(30)).is_err() {
|
||||
eprintln!("{diagnostic}: renderer-pool reentrancy deadlocked (30-second deadline)");
|
||||
// Exit only this private synthetic test child, without signaling any process.
|
||||
std::process::exit(124);
|
||||
}
|
||||
});
|
||||
run_case(&case);
|
||||
done.send(()).unwrap();
|
||||
watchdog.join().unwrap();
|
||||
return;
|
||||
}
|
||||
for case in ["png", "gif", "stepped", "overlap-audio"] {
|
||||
let mut child = std::process::Command::new(std::env::current_exe().unwrap())
|
||||
.args(["--exact", "nested_rayon_renderers_finish_every_export_path", "--nocapture"])
|
||||
.env(CHILD, case)
|
||||
.spawn()
|
||||
.unwrap();
|
||||
let status = child.wait().unwrap();
|
||||
assert!(status.success(), "{case}: child failed with {status}");
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user