From 51f66491590ea7d0dfdaf8e79abd2321a72a7568 Mon Sep 17 00:00:00 2001 From: dailz Date: Mon, 13 Jul 2026 16:04:30 +0800 Subject: [PATCH] refactor(bin): dedupe av_err_to_string + receive_first_frame + drain_encoder via shared src/bin/common/mod.rs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Step 2a: eliminate cross-bench duplication identified by the Explore audit. Changes: - src/avhw/util.rs: av_err_to_string promoted pub(crate) -> pub (the only change to src/avhw/ in this whole refactor plan). - src/avhw/mod.rs: re-export av_err_to_string; #[allow(unused_imports)] silences rustc's per-bin unused-import false positive (the pub use is consumed by the bench bins, not by the main bin). - src/bin/common/mod.rs (new): shared receive_first_frame + drain_encoder. These were byte-identical between the two bench binaries modulo a type-path alias (ff::codec::encoder::video::Video vs ff::encoder::video::Video) and SAFETY-comment line wrapping. Both binaries now wire it via #[path = "common/mod.rs"] mod common;. - src/bin/vaapi_import_bench.rs: 1039 -> 947 LOC (av_err_to_string, receive_first_frame, drain_encoder all removed; 3 call sites updated). - src/bin/sw_encode_bench.rs: 614 -> 545 LOC (receive_first_frame, drain_encoder removed; 3 call sites updated). - use ffmpeg_next::packet::Mut moved to common/mod.rs (was needed only for pkt.as_mut_ptr() inside drain_encoder). Verification (all green): - cargo build --bins / cargo build --release - cargo test (79 lib + 3 integration = 82 pass, 1 ignored — unchanged) - cargo clippy --all-targets -- -D warnings - cargo fmt --check - Test counts unchanged from baseline --- src/avhw/mod.rs | 2 + src/avhw/util.rs | 2 +- src/bin/common/mod.rs | 76 +++++++++++++++++++++++++++++ src/bin/sw_encode_bench.rs | 79 +++--------------------------- src/bin/vaapi_import_bench.rs | 92 +++-------------------------------- 5 files changed, 92 insertions(+), 159 deletions(-) create mode 100644 src/bin/common/mod.rs diff --git a/src/avhw/mod.rs b/src/avhw/mod.rs index 0a6e0bd..0754491 100644 --- a/src/avhw/mod.rs +++ b/src/avhw/mod.rs @@ -63,6 +63,8 @@ pub use types::{ BitrateCommand, CpuNv12Frame, EncodeOutcome, EncodeStages, EncodedH264Frame, ResolutionChange, SwEncodeTiming, }; +#[allow(unused_imports)] +pub use util::av_err_to_string; pub(crate) use util::ff_err; // --------------------------------------------------------------------------- diff --git a/src/avhw/util.rs b/src/avhw/util.rs index c891c0a..b54dbc6 100644 --- a/src/avhw/util.rs +++ b/src/avhw/util.rs @@ -1,7 +1,7 @@ use ffmpeg_next::ffi; /// Convert an FFmpeg error code to a human-readable string. -pub(crate) fn av_err_to_string(err: i32) -> String { +pub fn av_err_to_string(err: i32) -> String { let mut buf = vec![0u8; 128]; // SAFETY: buf points to 128 writable bytes and lives for the duration of // av_strerror. diff --git a/src/bin/common/mod.rs b/src/bin/common/mod.rs new file mode 100644 index 0000000..e492151 --- /dev/null +++ b/src/bin/common/mod.rs @@ -0,0 +1,76 @@ +use std::time::Instant; + +use anyhow::{bail, Result}; +use ffmpeg_next as ff; +use ffmpeg_next::ffi; +use ffmpeg_next::packet::Mut; +use wl_webrtc::cap_portal::{CapPortal, PwCtrlEvent}; + +pub fn receive_first_frame(cap: &CapPortal) -> Result { + // Drain-and-wait loop that mirrors production's repeated-poll semantics + // (state_portal.rs::poll_and_encode driven by main.rs's outer loop), but with + // a single bounded 10s total deadline appropriate for a bench tool. Unlike a + // single 10s blocking wait, this loop actually iterates: each turn drains ALL + // pending control events (the ctrl channel is bounded to 8 — a single + // if-let would silently miss backlog) and then waits a short slice for a + // frame, so StreamEnded/Error arriving mid-wait are observed within ~200ms. + const TOTAL_DEADLINE: std::time::Duration = std::time::Duration::from_secs(10); + const WAIT_SLICE: std::time::Duration = std::time::Duration::from_millis(200); + let deadline = Instant::now() + TOTAL_DEADLINE; + loop { + while let Ok(ctrl) = cap.event_receiver().try_recv() { + match ctrl { + PwCtrlEvent::StreamEnded => bail!("PipeWire stream ended before first frame"), + PwCtrlEvent::FormatChanged { .. } => {} + PwCtrlEvent::Error(e) => bail!("PipeWire error: {e}"), + } + } + let remaining = match deadline.checked_duration_since(Instant::now()) { + Some(r) if !r.is_zero() => r, + _ => bail!("Timeout waiting for first frame (10s)"), + }; + let slice = remaining.min(WAIT_SLICE); + match cap.frame_receiver().recv_timeout(slice) { + Ok(frame) => return Ok(frame), + Err(crossbeam_channel::RecvTimeoutError::Timeout) => continue, + Err(crossbeam_channel::RecvTimeoutError::Disconnected) => { + bail!("PipeWire frame channel disconnected"); + } + } + } +} + +pub fn drain_encoder( + enc_video: &mut ff::encoder::video::Video, + octx: &mut ff::format::context::Output, +) -> Result<()> { + loop { + let mut pkt = ff::Packet::empty(); + // SAFETY: enc_video is the opened encoder; pkt is an empty Packet whose + // inner AVPacket pointer is valid. avcodec_receive_packet fills pkt with + // the next encoded packet, or returns EAGAIN/EOF when drained. + let ret = unsafe { ffi::avcodec_receive_packet(enc_video.as_mut_ptr(), pkt.as_mut_ptr()) }; + if ret < 0 { + if ret == ffi::AVERROR(ffi::EAGAIN) || ret == ffi::AVERROR_EOF { + break; + } + eprintln!("avcodec_receive_packet failed: {ret}"); + break; + } + + let enc_tb = enc_video.time_base(); + // SAFETY: octx.as_ptr() is a valid AVFormatContext; streams is a NULL-terminated + // array of AVStream*. We index [0] which exists because we created exactly one + // stream in setup. Reading time_base is a plain AVRational field access. + let stream_tb = unsafe { + let streams = (*octx.as_ptr()).streams; + let st = *streams.add(0); + ff::Rational::from((*st).time_base) + }; + pkt.rescale_ts(enc_tb, stream_tb); + pkt.set_stream(0); + pkt.write_interleaved(octx) + .map_err(|e| anyhow::anyhow!("write packet failed: {e}"))?; + } + Ok(()) +} diff --git a/src/bin/sw_encode_bench.rs b/src/bin/sw_encode_bench.rs index 278137d..3c248d4 100644 --- a/src/bin/sw_encode_bench.rs +++ b/src/bin/sw_encode_bench.rs @@ -15,11 +15,13 @@ use clap::Parser; use ffmpeg_next as ff; use ffmpeg_next::ffi; -use ffmpeg_next::packet::Mut; use wl_webrtc::args::Args; use wl_webrtc::cap_portal::{CapPortal, PwCtrlEvent}; +#[path = "common/mod.rs"] +mod common; + #[derive(Parser, Debug)] #[command( name = "sw_encode_bench", @@ -61,40 +63,6 @@ fn pix_fmt(p: ff::format::Pixel) -> ffi::AVPixelFormat { Into::::into(p) } -fn receive_first_frame(cap: &CapPortal) -> Result { - // Drain-and-wait loop that mirrors production's repeated-poll semantics - // (state_portal.rs::poll_and_encode driven by main.rs's outer loop), but with - // a single bounded 10s total deadline appropriate for a bench tool. Unlike a - // single 10s blocking wait, this loop actually iterates: each turn drains ALL - // pending control events (the ctrl channel is bounded to 8 — a single - // if-let would silently miss backlog) and then waits a short slice for a - // frame, so StreamEnded/Error arriving mid-wait are observed within ~200ms. - const TOTAL_DEADLINE: std::time::Duration = std::time::Duration::from_secs(10); - const WAIT_SLICE: std::time::Duration = std::time::Duration::from_millis(200); - let deadline = Instant::now() + TOTAL_DEADLINE; - loop { - while let Ok(ctrl) = cap.event_receiver().try_recv() { - match ctrl { - PwCtrlEvent::StreamEnded => bail!("PipeWire stream ended before first frame"), - PwCtrlEvent::FormatChanged { .. } => {} - PwCtrlEvent::Error(e) => bail!("PipeWire error: {e}"), - } - } - let remaining = match deadline.checked_duration_since(Instant::now()) { - Some(r) if !r.is_zero() => r, - _ => bail!("Timeout waiting for first frame (10s)"), - }; - let slice = remaining.min(WAIT_SLICE); - match cap.frame_receiver().recv_timeout(slice) { - Ok(frame) => return Ok(frame), - Err(crossbeam_channel::RecvTimeoutError::Timeout) => continue, - Err(crossbeam_channel::RecvTimeoutError::Disconnected) => { - bail!("PipeWire frame channel disconnected"); - } - } - } -} - fn main() -> Result<()> { let bench_args = BenchArgs::parse(); @@ -133,7 +101,7 @@ fn main() -> Result<()> { println!("[1/4] Portal connected, PipeWire stream active\n"); println!("[2/4] Waiting for first frame from PipeWire..."); - let first_frame = receive_first_frame(&cap)?; + let first_frame = common::receive_first_frame(&cap)?; let src_width = first_frame.width; let src_height = first_frame.height; @@ -462,7 +430,7 @@ fn main() -> Result<()> { } } - drain_encoder(&mut enc_video, &mut octx)?; + common::drain_encoder(&mut enc_video, &mut octx)?; stats .encode_us @@ -489,7 +457,7 @@ fn main() -> Result<()> { unsafe { ffi::avcodec_send_frame(enc_video.as_mut_ptr(), ptr::null()); } - drain_encoder(&mut enc_video, &mut octx)?; + common::drain_encoder(&mut enc_video, &mut octx)?; octx.write_trailer() .map_err(|e| anyhow::anyhow!("Failed to write trailer: {e}"))?; @@ -577,38 +545,3 @@ fn main() -> Result<()> { println!("Output written to: {}", bench_args.output); Ok(()) } - -fn drain_encoder( - enc_video: &mut ff::encoder::video::Video, - octx: &mut ff::format::context::Output, -) -> Result<()> { - loop { - let mut pkt = ff::Packet::empty(); - // SAFETY: enc_video is the opened encoder; pkt is an empty Packet whose - // inner AVPacket pointer is valid. avcodec_receive_packet fills pkt with - // the next encoded packet, or returns EAGAIN/EOF when drained. - let ret = unsafe { ffi::avcodec_receive_packet(enc_video.as_mut_ptr(), pkt.as_mut_ptr()) }; - if ret < 0 { - if ret == ffi::AVERROR(ffi::EAGAIN) || ret == ffi::AVERROR_EOF { - break; - } - eprintln!("avcodec_receive_packet failed: {ret}"); - break; - } - - let enc_tb = enc_video.time_base(); - // SAFETY: octx.as_ptr() is a valid AVFormatContext; streams is a NULL-terminated - // array of AVStream*. We index [0] which exists because we created exactly one - // stream in setup. Reading time_base is a plain AVRational field access. - let stream_tb = unsafe { - let streams = (*octx.as_ptr()).streams; - let st = *streams.add(0); - ff::Rational::from((*st).time_base) - }; - pkt.rescale_ts(enc_tb, stream_tb); - pkt.set_stream(0); - pkt.write_interleaved(octx) - .map_err(|e| anyhow::anyhow!("write packet failed: {e}"))?; - } - Ok(()) -} diff --git a/src/bin/vaapi_import_bench.rs b/src/bin/vaapi_import_bench.rs index 509e50a..9a26242 100644 --- a/src/bin/vaapi_import_bench.rs +++ b/src/bin/vaapi_import_bench.rs @@ -15,12 +15,14 @@ use clap::{Parser, ValueEnum}; use ffmpeg_next as ff; use ffmpeg_next::ffi; -use ffmpeg_next::packet::Mut; use wl_webrtc::args::Args; -use wl_webrtc::avhw::{import_dma_buf_to_vaapi, AvHwDevCtx, AvHwFrameCtx}; +use wl_webrtc::avhw::{av_err_to_string, import_dma_buf_to_vaapi, AvHwDevCtx, AvHwFrameCtx}; use wl_webrtc::cap_portal::{CapPortal, PwCtrlEvent}; +#[path = "common/mod.rs"] +mod common; + #[derive(Parser, Debug)] #[command(name = "vaapi_import_bench", about = "VAAPI DMA-BUF import benchmark")] struct BenchArgs { @@ -124,86 +126,6 @@ impl Drop for SwsContext { } } -fn av_err_to_string(ret: i32) -> String { - let mut buf = vec![0u8; 128]; - // SAFETY: buf is a 128-byte Vec initialized to zeros; av_strerror writes at most - // buf.len() bytes (including NUL) into the buffer. The ret value is an FFmpeg - // error code. We treat the buffer as `*mut i8` for the C string out-param. - unsafe { - ffi::av_strerror(ret, buf.as_mut_ptr() as *mut i8, buf.len()); - } - let end = buf.iter().position(|&b| b == 0).unwrap_or(buf.len()); - String::from_utf8_lossy(&buf[..end]).to_string() -} - -fn receive_first_frame(cap: &CapPortal) -> Result { - // Drain-and-wait loop that mirrors production's repeated-poll semantics - // (state_portal.rs::poll_and_encode driven by main.rs's outer loop), but with - // a single bounded 10s total deadline appropriate for a bench tool. Unlike a - // single 10s blocking wait, this loop actually iterates: each turn drains ALL - // pending control events (the ctrl channel is bounded to 8 — a single - // if-let would silently miss backlog) and then waits a short slice for a - // frame, so StreamEnded/Error arriving mid-wait are observed within ~200ms. - const TOTAL_DEADLINE: std::time::Duration = std::time::Duration::from_secs(10); - const WAIT_SLICE: std::time::Duration = std::time::Duration::from_millis(200); - let deadline = Instant::now() + TOTAL_DEADLINE; - loop { - while let Ok(ctrl) = cap.event_receiver().try_recv() { - match ctrl { - PwCtrlEvent::StreamEnded => bail!("PipeWire stream ended before first frame"), - PwCtrlEvent::FormatChanged { .. } => {} - PwCtrlEvent::Error(e) => bail!("PipeWire error: {e}"), - } - } - let remaining = match deadline.checked_duration_since(Instant::now()) { - Some(r) if !r.is_zero() => r, - _ => bail!("Timeout waiting for first frame (10s)"), - }; - let slice = remaining.min(WAIT_SLICE); - match cap.frame_receiver().recv_timeout(slice) { - Ok(frame) => return Ok(frame), - Err(crossbeam_channel::RecvTimeoutError::Timeout) => continue, - Err(crossbeam_channel::RecvTimeoutError::Disconnected) => { - bail!("PipeWire frame channel disconnected"); - } - } - } -} - -fn drain_encoder( - enc_video: &mut ff::codec::encoder::video::Video, - octx: &mut ff::format::context::Output, -) -> Result<()> { - loop { - let mut pkt = ff::Packet::empty(); - // SAFETY: enc_video is the opened encoder; pkt is an empty Packet whose inner - // AVPacket pointer is valid. avcodec_receive_packet fills pkt with the next - // encoded packet or returns EAGAIN/EOF when drained. - let ret = unsafe { ffi::avcodec_receive_packet(enc_video.as_mut_ptr(), pkt.as_mut_ptr()) }; - if ret < 0 { - if ret == ffi::AVERROR(ffi::EAGAIN) || ret == ffi::AVERROR_EOF { - break; - } - eprintln!("avcodec_receive_packet failed: {ret}"); - break; - } - let enc_tb = enc_video.time_base(); - // SAFETY: octx.as_ptr() is a valid AVFormatContext; streams is a NULL-terminated - // array; we index [0] which exists because we created exactly one stream in - // setup. Reading time_base is a plain AVRational field access. - let stream_tb = unsafe { - let streams = (*octx.as_ptr()).streams; - let st = *streams.add(0); - ff::Rational::from((*st).time_base) - }; - pkt.rescale_ts(enc_tb, stream_tb); - pkt.set_stream(0); - pkt.write_interleaved(octx) - .map_err(|e| anyhow::anyhow!("write packet failed: {e}"))?; - } - Ok(()) -} - fn create_software_encoder(output_path: &Path, width: u32, height: u32) -> Result { let output_cstr = CString::new(output_path.to_str().unwrap())?; let codec = ff::encoder::find_by_name("libx264") @@ -394,7 +316,7 @@ fn encode_yuv_frame(encoder: &mut SoftwareEncoder, pts: &mut i64) -> Result bail!("avcodec_send_frame failed: {r}"); } } - drain_encoder(&mut encoder.enc_video, &mut encoder.octx)?; + common::drain_encoder(&mut encoder.enc_video, &mut encoder.octx)?; Ok(t_encode.elapsed().as_micros() as u64) } @@ -403,7 +325,7 @@ fn finish_encoder(mut encoder: SoftwareEncoder) -> Result<()> { unsafe { ffi::avcodec_send_frame(encoder.enc_video.as_mut_ptr(), ptr::null()); } - drain_encoder(&mut encoder.enc_video, &mut encoder.octx)?; + common::drain_encoder(&mut encoder.enc_video, &mut encoder.octx)?; encoder .octx .write_trailer() @@ -904,7 +826,7 @@ fn main() -> Result<()> { println!("[1/3] Portal connected, PipeWire stream active\n"); println!("[2/3] Waiting for first frame from PipeWire..."); - let first_frame = receive_first_frame(&cap)?; + let first_frame = common::receive_first_frame(&cap)?; let src_width = first_frame.width; let src_height = first_frame.height;