dailz/wl-webrtc
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor(bin): dedupe av_err_to_string + receive_first_frame + drain_encoder via shared src/bin/common/mod.rs

Browse Source 
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
This commit is contained in:
dailz
2026-07-13 16:04:30 +08:00
parent bc405c6d16
commit 51f6649159
5 changed files with 92 additions and 159 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/avhw/mod.rs
+2
View File
@@ -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;
// ---------------------------------------------------------------------------
src/avhw/util.rs
+1 -1
View File
@@ -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.
src/bin/common/mod.rs
+76
View File
@@ -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<wl_webrtc::cap_portal::PwDmaBufFrame> {
// 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(())
}
src/bin/sw_encode_bench.rs
+6 -73
View File
@@ -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::<ffi::AVPixelFormat>::into(p)
}
fn receive_first_frame(cap: &CapPortal) -> Result<wl_webrtc::cap_portal::PwDmaBufFrame> {
// 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(())
}
src/bin/vaapi_import_bench.rs
+7 -85
View File
@@ -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<wl_webrtc::cap_portal::PwDmaBufFrame> {
// 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<SoftwareEncoder> {
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<u64>
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;