From 83a1d33b814229ac1b352f807a662ebffcdbf0b7 Mon Sep 17 00:00:00 2001 From: chillfish8 Date: Mon, 5 Oct 2026 21:03:58 +0100 Subject: [PATCH] Report coordinator's error and bump version --- Cargo.lock | 6 +- Cargo.toml | 2 +- av-denoise/src/bin/file_mode.rs | 136 +++++++++++++++++++++++++++++--- docs/TUTORIAL-CLI.md | 2 +- 4 files changed, 128 insertions(+), 18 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index d1411fd..ac6f279 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -181,7 +181,7 @@ dependencies = [ [[package]] name = "av-denoise" -version = "0.4.1" +version = "0.4.2" dependencies = [ "anyhow", "av-decoders", @@ -204,7 +204,7 @@ dependencies = [ [[package]] name = "av-denoise-core" -version = "0.4.1" +version = "0.4.2" dependencies = [ "anyhow", "bon", @@ -221,7 +221,7 @@ dependencies = [ [[package]] name = "av-denoise-vs" -version = "0.4.1" +version = "0.4.2" dependencies = [ "anyhow", "av-denoise-core", diff --git a/Cargo.toml b/Cargo.toml index 32e1bca..5546ba9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,7 +3,7 @@ resolver = "3" members = ["av-denoise-core", "av-denoise", "av-denoise-vs"] [workspace.package] -version = "0.4.1" +version = "0.4.2" edition = "2024" license = "Apache-2.0" repository = "https://github.com/ChillFish8/av-denoise" diff --git a/av-denoise/src/bin/file_mode.rs b/av-denoise/src/bin/file_mode.rs index 1a4c712..33f5359 100644 --- a/av-denoise/src/bin/file_mode.rs +++ b/av-denoise/src/bin/file_mode.rs @@ -301,25 +301,67 @@ fn encode_scenes( give, ); - dispatch_frames(input, scenes, &job_tx, &take)?; + let dispatched = dispatch_frames(input, scenes, &job_tx, &take); // Closing the queue is what tells the workers there are no more scenes. drop(job_tx); - for h in worker_handles { - h.join() - .map_err(|e| anyhow::anyhow!("worker panicked: {e:?}"))??; - } - - coordinator + let workers_joined = join_workers(worker_handles); + let coordinator_joined = coordinator .join() - .map_err(|e| anyhow::anyhow!("coordinator panicked: {e:?}"))??; + .map_err(|panic| anyhow::anyhow!("coordinator panicked: {panic:?}")) + .and_then(|result| result); - Ok(()) + first_root_cause(coordinator_joined, workers_joined, dispatched) +} + +/// Picks which error a finished run reports. +/// +/// A failed write comes first, then worker errors, then dispatch errors. Lost frames come +/// last because a failed worker or dispatcher is what loses them. +fn first_root_cause( + coordinator: Result<(), anyhow::Error>, + workers: Result<(), anyhow::Error>, + dispatched: Result<(), anyhow::Error>, +) -> Result<(), anyhow::Error> { + let lost_frames = match coordinator { + Ok(()) => None, + Err(err) if err.is::() => Some(err), + Err(err) => return Err(err), + }; + + workers?; + dispatched?; + + lost_frames.map_or(Ok(()), Err) } type WorkerJoin = thread::JoinHandle>; +/// Joins every worker, logging each error or panic and returning the first. +fn join_workers(handles: Vec) -> Result<(), anyhow::Error> { + let mut first_error = None; + + for (worker_id, handle) in handles.into_iter().enumerate() { + let result = handle + .join() + .map_err(|panic| anyhow::anyhow!("worker panicked: {panic:?}")) + .and_then(|result| result); + + let Err(err) = result else { + continue; + }; + + tracing::error!(worker_id, error = %err, "worker failed"); + + if first_error.is_none() { + first_error = Some(err); + } + } + + first_error.map_or(Ok(()), Err) +} + /// Spawns `workers` worker threads over one shared scene queue. /// /// The queue is a rendezvous, so a scene is only offered when a worker is @@ -703,15 +745,36 @@ fn emit_frames( } if next_emit != total { - anyhow::bail!( - "wrote {next_emit} frames but expected {total}. Every worker disconnected \ - before the stream finished, so a frame index was likely lost" - ); + let lost = FramesLost { + written: next_emit, + expected: total, + }; + return Err(lost.into()); } Ok(()) } +/// Every worker hung up before the coordinator wrote all the frames it expected. +#[derive(Debug)] +struct FramesLost { + written: u64, + expected: u64, +} + +impl std::fmt::Display for FramesLost { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!( + f, + "wrote {} frames but expected {}. Every worker disconnected before the stream \ + finished, so a frame index was likely lost", + self.written, self.expected, + ) + } +} + +impl std::error::Error for FramesLost {} + /// Checks each plane's byte length against the layout, failing with an error naming which plane /// is wrong, the length found and the length expected. fn check_plane_lens(planes: &Planes, layout: FrameLayout) -> Result<(), anyhow::Error> { @@ -1138,6 +1201,53 @@ mod tests { msg.contains('1') && msg.contains('3'), "error should name frames written (1) vs expected (3): {msg}" ); + assert!(err.is::()); + } + + fn lost_frames() -> Result<(), anyhow::Error> { + let lost = FramesLost { + written: 1, + expected: 3, + }; + + Err(lost.into()) + } + + #[test] + fn a_failed_write_is_reported_over_the_worker_disconnect() { + let coordinator = Err(anyhow::anyhow!("broken pipe")); + let workers = Err(anyhow::anyhow!("coordinator disconnected")); + let dispatched = Err(anyhow::anyhow!("the coordinator stopped")); + + let err = first_root_cause(coordinator, workers, dispatched).expect_err("the run failed"); + + assert_eq!(err.to_string(), "broken pipe"); + } + + #[test] + fn a_worker_error_is_reported_over_lost_frames_and_dispatch() { + let workers = Err(anyhow::anyhow!("device lost")); + let dispatched = Err(anyhow::anyhow!("the worker holding scene 0 disconnected")); + + let err = first_root_cause(lost_frames(), workers, dispatched).expect_err("the run failed"); + + assert_eq!(err.to_string(), "device lost"); + } + + #[test] + fn a_dispatch_error_is_reported_over_lost_frames() { + let dispatched = Err(anyhow::anyhow!("decode failed")); + + let err = first_root_cause(lost_frames(), Ok(()), dispatched).expect_err("the run failed"); + + assert_eq!(err.to_string(), "decode failed"); + } + + #[test] + fn lost_frames_are_reported_when_nothing_else_failed() { + let err = first_root_cause(lost_frames(), Ok(()), Ok(())).expect_err("the run failed"); + + assert!(err.is::()); } /// Gated because it names the `Vulkan` accelerator variant, which only diff --git a/docs/TUTORIAL-CLI.md b/docs/TUTORIAL-CLI.md index 1852ce2..b2110d8 100644 --- a/docs/TUTORIAL-CLI.md +++ b/docs/TUTORIAL-CLI.md @@ -62,7 +62,7 @@ docker run --rm \ --device /dev/kfd --device /dev/dri \ --group-add video --group-add render \ -v "$PWD:/in:ro" \ - ghcr.io/chillfish8/av-denoise:vulkan-0.4.1 \ + ghcr.io/chillfish8/av-denoise:vulkan-0.4.2 \ nl4d --input /in/noisy.mkv \ | ffmpeg -f yuv4mpegpipe -i - -c:v ffv1 clean.mkv ```