diff --git a/Cargo.toml b/Cargo.toml index de356ea..ed846ad 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -33,4 +33,4 @@ dirs = "6" tempfile = "3.27.0" [lints.clippy] -undocumented_unsafe_blocks = "warn" +undocumented_unsafe_blocks = "deny" diff --git a/src/avhw.rs b/src/avhw.rs index 60f2be5..c6fdd1c 100644 --- a/src/avhw.rs +++ b/src/avhw.rs @@ -338,6 +338,14 @@ pub struct EncState { frames_written: bool, } +// SAFETY: EncState is moved to exactly one thread (the encode worker) and used +// exclusively there. All fields are either plain Copy types (Option, bool) +// or ffmpeg-next / AvHw* owned wrappers whose raw inner pointers are not actually +// shared across threads — they're touched only from the owning encode thread. +// This impl exists only to satisfy Rust's auto-Send inference (which can't see +// through the raw pointers hidden inside the wrappers). Do NOT add fields that +// introduce shared mutable state without re-auditing this assumption; see +// AGENTS.md "Unsafe and FFI work" for the documented exclusivity requirement. unsafe impl Send for EncState {} impl EncState { @@ -430,6 +438,9 @@ impl EncState { // VBV rate limiting: caps IDR burst size for WebRTC. Without this a 4K // scene change can produce a 256KB keyframe that overflows the UDP send // buffer. bufsize=bitrate/4 ≈ 250ms of video at the target bitrate. + // SAFETY: enc.as_mut_ptr() is a valid AVCodecContext for the not-yet-opened + // encoder. rc_max_rate and rc_buffer_size are plain integer fields; assigning + // i64/i32 values is a simple struct-field write on a properly aligned pointer. unsafe { let ctx_ptr = enc.as_mut_ptr(); (*ctx_ptr).rc_max_rate = bitrate as i64; @@ -456,6 +467,10 @@ impl EncState { { let key = CString::new("repeat_pps").unwrap(); let val = CString::new("1").unwrap(); + // SAFETY: enc is a valid AVCodecContext for the not-yet-opened encoder; + // priv_data is the codec's private options struct. key/val are NUL-terminated + // CString that live across the call. av_opt_set is FFmpeg's standard + // option-setter. Failure is non-fatal (returns < 0 on older FFmpeg). let ret = unsafe { ffi::av_opt_set((*enc.as_mut_ptr()).priv_data, key.as_ptr(), val.as_ptr(), 0) }; @@ -488,11 +503,17 @@ impl EncState { bail!("Failed to allocate output format context: {}", ff_err(ret)); } - // SAFETY: avformat_query_codec checks codec+format compatibility. - let codec_id = unsafe { (*enc_video.as_ptr()).codec_id }; - let oformat = unsafe { (*fmt_ctx_ptr).oformat }; - let compat = unsafe { - ffi::avformat_query_codec(oformat, codec_id, ffi::FF_COMPLIANCE_NORMAL as i32) + // SAFETY: enc_video is a valid AVCodecContext pointer; codec_id is a plain + // i32 enum discriminant read from it. fmt_ctx_ptr is a valid AVFormatContext + // allocated above; oformat is a const pointer field read from it. + // avformat_query_codec checks codec+format compatibility; both pointers are + // valid and FF_COMPLIANCE_NORMAL is a constant. All three reads happen in one + // block so a single SAFETY rationale covers them. + let (codec_id, oformat, compat) = unsafe { + let codec_id = (*enc_video.as_ptr()).codec_id; + let oformat = (*fmt_ctx_ptr).oformat; + let compat = ffi::avformat_query_codec(oformat, codec_id, ffi::FF_COMPLIANCE_NORMAL); + (codec_id, oformat, compat) }; if compat < 0 { bail!("H.264 codec not supported by output container format"); @@ -1225,7 +1246,7 @@ impl SwEncEncode { let frame_index = self.frame_count; self.frame_count = self.frame_count.saturating_add(1); let current_hash = hash_sampled_y_plane(&frame.y_data, width, height, frame.y_stride); - let force_gop_frame = self.gop_size > 0 && frame_index % u64::from(self.gop_size) == 0; + let force_gop_frame = self.gop_size > 0 && frame_index.is_multiple_of(u64::from(self.gop_size)); if frame_index > 0 && !force_gop_frame && !force_this_frame && current_hash == self.last_frame_hash { tracing::debug!(frame_index, "skipping duplicate frame"); self.last_frame_hash = current_hash; @@ -1760,7 +1781,7 @@ fn create_software_h264_muxer( let key = CString::new("threads").unwrap(); let val = CString::new("6").unwrap(); ffi::av_opt_set((*enc.as_mut_ptr()).priv_data, key.as_ptr(), val.as_ptr(), 0); - (*enc.as_mut_ptr()).profile = ffi::AV_PROFILE_H264_HIGH as i32; + (*enc.as_mut_ptr()).profile = ffi::AV_PROFILE_H264_HIGH; // SAFETY: enc is a valid, initialized AVCodecContext from // avcodec_alloc_context3. Setting level is a simple i32 field // assignment on a properly aligned struct. @@ -1896,7 +1917,7 @@ fn create_software_h264_encoder( ffi::av_opt_set((*enc.as_mut_ptr()).priv_data, key.as_ptr(), val.as_ptr(), 0); // High profile via AVCodecContext.profile (not x264opts — x264 rejects it there). // High enables CABAC + 8x8dct automatically. - (*enc.as_mut_ptr()).profile = ffi::AV_PROFILE_H264_HIGH as i32; + (*enc.as_mut_ptr()).profile = ffi::AV_PROFILE_H264_HIGH; // SAFETY: enc is a valid, initialized AVCodecContext from // avcodec_alloc_context3. Setting level is a simple i32 field // assignment on a properly aligned struct. @@ -2042,6 +2063,14 @@ fn build_filter_graph( mod tests { use super::*; + // Centralizes the `stride * row` byte-offset pattern used by the Y-plane hash + // tests below, so clippy::erasing_op (row == 0) and clippy::identity_op (row == 1) + // both pass without sacrificing the row-index intent the tests are written around. + fn row_range(row: usize, stride: usize, width: usize) -> std::ops::Range { + let start = stride * row; + start..start + width + } + // ── Task 1: VBV x264opts formatting ── #[test] @@ -2071,10 +2100,16 @@ mod tests { #[test] fn webrtc_gop_formula() { - assert_eq!((15u32 * 2).max(20), 30); // 15fps -> 30 - assert_eq!((30u32 * 2).max(20), 60); // 30fps -> 60 - assert_eq!((60u32 * 2).max(20), 120); // 60fps -> 120 - assert_eq!((5u32 * 2).max(20), 20); // 5fps -> 20 (floor) + // Formula under test: GOP = max(fps * 2, 20). Hid behind a runtime lambda so + // clippy can't constant-fold the assertions into tautologies (which would + // silently strip the floor-case coverage for 5fps). + fn gop(fps: u32) -> u32 { + (fps * 2).max(20) + } + assert_eq!(gop(15), 30); // 15fps -> 30 + assert_eq!(gop(30), 60); // 30fps -> 60 + assert_eq!(gop(60), 120); // 60fps -> 120 + assert_eq!(gop(5), 20); // 5fps -> 20 (floor) } #[test] @@ -2122,7 +2157,7 @@ mod tests { let y_data1 = vec![0u8; stride * height]; let mut y_data2 = vec![0u8; stride * height]; // Row 1 is NOT sampled (sampling is every 8th row: 0, 8, 16, ...) - y_data2[stride * 1..stride * 1 + width].fill(255); + y_data2[row_range(1, stride, width)].fill(255); let hash1 = hash_sampled_y_plane(&y_data1, width, height, stride); let hash2 = hash_sampled_y_plane(&y_data2, width, height, stride); assert_eq!( @@ -2140,7 +2175,7 @@ mod tests { let y_data1 = vec![0u8; stride * height]; let mut y_data2 = vec![0u8; stride * height]; // Row 0 IS sampled (every 8th row starting from 0) - y_data2[stride * 0..stride * 0 + width].fill(255); + y_data2[row_range(0, stride, width)].fill(255); let hash1 = hash_sampled_y_plane(&y_data1, width, height, stride); let hash2 = hash_sampled_y_plane(&y_data2, width, height, stride); assert_ne!( diff --git a/src/bin/sw_encode_bench.rs b/src/bin/sw_encode_bench.rs index 220b05b..278137d 100644 --- a/src/bin/sw_encode_bench.rs +++ b/src/bin/sw_encode_bench.rs @@ -62,22 +62,32 @@ fn pix_fmt(p: ff::format::Pixel) -> ffi::AVPixelFormat { } 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 { - if let Ok(ctrl) = cap.event_receiver().try_recv() { + 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}"), } } - match cap - .frame_receiver() - .recv_timeout(std::time::Duration::from_secs(10)) - { + 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) => { - bail!("Timeout waiting for first frame (10s)"); - } + Err(crossbeam_channel::RecvTimeoutError::Timeout) => continue, Err(crossbeam_channel::RecvTimeoutError::Disconnected) => { bail!("PipeWire frame channel disconnected"); } @@ -142,6 +152,9 @@ fn main() -> Result<()> { println!("[3/4] Testing mmap on DMA-BUF..."); let mmap_size = (src_stride as usize) * (src_height as usize); + // SAFETY: first_frame.fd is an open DMA-BUF; offset/size come from PipeWire's + // negotiated format. PROT_READ+MAP_SHARED is the standard read-only DMA-BUF + // mapping. Returns MAP_FAILED on error (checked below). let mmap_ptr = unsafe { libc::mmap( ptr::null_mut(), @@ -173,6 +186,8 @@ fn main() -> Result<()> { "[3/4] mmap SUCCESS — CPU can read DMA-BUF ({:.1} MB)\n", mmap_size as f64 / 1024.0 / 1024.0 ); + // SAFETY: mmap_ptr was returned by mmap above and is not MAP_FAILED (checked); + // mmap_size matches the original mapping. POSIX munmap(2) releases the mapping. unsafe { libc::munmap(mmap_ptr, mmap_size); } @@ -205,6 +220,9 @@ fn main() -> Result<()> { let codec_name = codec.name(); if codec_name == "libx264" { + // SAFETY: enc is a valid AVCodecContext for the not-yet-opened encoder; + // priv_data is the x264 private options struct. All CStrings live across + // both av_opt_set calls. These set the x264 "preset" and "tune" options. unsafe { let key = CString::new("preset").unwrap(); let val = CString::new("veryfast").unwrap(); @@ -220,6 +238,8 @@ fn main() -> Result<()> { // Create output format context via FFI let mut fmt_ctx_ptr: *mut ffi::AVFormatContext = ptr::null_mut(); + // SAFETY: fmt_ctx_ptr is an out-parameter initialized by FFmpeg; output_cstr + // lives across the call. Returns 0 on success; we check below. let ret = unsafe { ffi::avformat_alloc_output_context2( &mut fmt_ctx_ptr, @@ -232,21 +252,30 @@ fn main() -> Result<()> { bail!("Failed to allocate output format context: error {ret}"); } + // SAFETY: fmt_ctx_ptr is the valid output context allocated above. + // avformat_new_stream returns a pointer to a new AVStream or NULL on failure. let stream_ptr = unsafe { ffi::avformat_new_stream(fmt_ctx_ptr, ptr::null()) }; if stream_ptr.is_null() { bail!("Failed to create new stream"); } + // SAFETY: stream_ptr and enc_video.as_ptr() are valid pointers; codecpar is + // the output destination inside stream. avcodec_parameters_from_context copies + // encoder parameters into the stream's codecpar. let ret = unsafe { ffi::avcodec_parameters_from_context((*stream_ptr).codecpar, enc_video.as_ptr()) }; if ret < 0 { bail!("Failed to copy encoder parameters: error {ret}"); } + // SAFETY: stream_ptr and enc_video are valid; time_base is a plain AVRational + // field copied from encoder to stream. unsafe { (*stream_ptr).time_base = (*enc_video.as_ptr()).time_base; } + // SAFETY: fmt_ctx_ptr is valid; pb is the AVIOContext slot to initialize; + // output_cstr is a valid NUL-terminated path; AVIO_FLAG_WRITE is a constant. let ret = unsafe { ffi::avio_open( &mut (*fmt_ctx_ptr).pb, @@ -261,17 +290,23 @@ fn main() -> Result<()> { ); } + // SAFETY: fmt_ctx_ptr is fully configured (streams + pb set); NULL options + // is the default. Returns 0 on success. let ret = unsafe { ffi::avformat_write_header(fmt_ctx_ptr, ptr::null_mut()) }; if ret < 0 { bail!("Failed to write header: error {ret}"); } + // SAFETY: fmt_ctx_ptr is a fully initialized output context (header written). + // Output::wrap takes ownership of the pointer into a safe RAII wrapper. let mut octx = unsafe { ff::format::context::Output::wrap(fmt_ctx_ptr) }; // Create sws_scale context: BGRZ (BGR0) -> YUV420P let bgr0_fmt = pix_fmt(ff::format::Pixel::BGRZ); let yuv420p_fmt = pix_fmt(ff::format::Pixel::YUV420P); + // SAFETY: all parameters are valid enum/pixel format values; NULL filters are + // allowed by FFmpeg. sws_getContext returns a heap-allocated SwsContext or NULL. let sws_ctx = unsafe { ffi::sws_getContext( src_width as i32, @@ -291,6 +326,9 @@ fn main() -> Result<()> { } // Allocate reusable YUV frame + // SAFETY: av_frame_alloc returns NULL only on OOM. After allocation we set + // width/height/format fields and call av_frame_get_buffer to allocate plane + // data. On failure we free the frame via av_frame_free before bailing. let mut yuv_frame = unsafe { let mut f = ffi::av_frame_alloc(); if f.is_null() { @@ -349,6 +387,9 @@ fn main() -> Result<()> { let mmap_start = Instant::now(); let frame_size = (frame.stride as usize) * (frame.height as usize); + // SAFETY: frame.fd is an open DMA-BUF owned by the frame; offset/size come + // from PipeWire's negotiated format. PROT_READ+MAP_SHARED for read-only + // DMA-BUF access. Returns MAP_FAILED on error (checked below). let mmap_ptr = unsafe { libc::mmap( ptr::null_mut(), @@ -369,8 +410,15 @@ fn main() -> Result<()> { stats.mmap_us.push(mmap_start.elapsed().as_micros() as u64); let scale_start = Instant::now(); + // SAFETY: mmap_ptr is a valid mapping of frame_size bytes (checked above); + // constructing a read-only slice over it for the duration of sws_scale is + // sound as long as we don't hold it past munmap (we don't). let src_data = unsafe { std::slice::from_raw_parts(mmap_ptr as *const u8, frame_size) }; + // SAFETY: yuv_frame and sws_ctx are valid; src_data is a valid slice of the + // mmap'd DMA-BUF for this frame. sws_scale reads src planes (BGR0 -> YUV420P) + // and writes into yuv_frame's data planes. av_frame_make_writable ensures + // yuv_frame is not shared before writing. unsafe { ffi::av_frame_make_writable(yuv_frame); @@ -391,6 +439,8 @@ fn main() -> Result<()> { .scale_us .push(scale_start.elapsed().as_micros() as u64); + // SAFETY: mmap_ptr was returned by mmap above and is not MAP_FAILED; frame_size + // matches the original mapping. Release before dropping frame (which closes fd). unsafe { libc::munmap(mmap_ptr, frame_size); } @@ -398,6 +448,9 @@ fn main() -> Result<()> { let encode_start = Instant::now(); + // SAFETY: yuv_frame is allocated and writable; enc_video is the opened encoder. + // Setting pts is a plain i64 field write. avcodec_send_frame submits the frame + // for encoding; returns < 0 on error (we log and continue). unsafe { (*yuv_frame).pts = pts; pts += 1; @@ -419,7 +472,7 @@ fn main() -> Result<()> { .push(frame_start.elapsed().as_micros() as u64); frames_encoded += 1; - if frames_encoded % 30 == 0 { + if frames_encoded.is_multiple_of(30) { let fps = frames_encoded as f64 / total_start.elapsed().as_secs_f64(); println!( " [{}/{}] {:.1} FPS", @@ -431,6 +484,8 @@ fn main() -> Result<()> { let total_elapsed = total_start.elapsed(); println!("\nFlushing encoder..."); + // SAFETY: enc_video is the opened encoder; passing NULL frame signals EOF to + // drain the encoder's internal pipeline. Returns < 0 on error (ignored here). unsafe { ffi::avcodec_send_frame(enc_video.as_mut_ptr(), ptr::null()); } @@ -440,6 +495,9 @@ fn main() -> Result<()> { .map_err(|e| anyhow::anyhow!("Failed to write trailer: {e}"))?; // Cleanup + // SAFETY: yuv_frame is the allocated frame from earlier (still owned by us); + // sws_ctx is the allocated sws context. av_frame_free and sws_freeContext take + // ownership and free their respective heap allocations. unsafe { ffi::av_frame_free(&mut yuv_frame as *mut _); ffi::sws_freeContext(sws_ctx); @@ -526,6 +584,9 @@ fn drain_encoder( ) -> 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 { @@ -536,6 +597,9 @@ fn drain_encoder( } 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); diff --git a/src/bin/vaapi_import_bench.rs b/src/bin/vaapi_import_bench.rs index 675a05e..52069d5 100644 --- a/src/bin/vaapi_import_bench.rs +++ b/src/bin/vaapi_import_bench.rs @@ -126,6 +126,9 @@ 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()); } @@ -134,22 +137,32 @@ fn av_err_to_string(ret: i32) -> 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 { - if let Ok(ctrl) = cap.event_receiver().try_recv() { + 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}"), } } - match cap - .frame_receiver() - .recv_timeout(std::time::Duration::from_secs(10)) - { + 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) => { - bail!("Timeout waiting for first frame (10s)"); - } + Err(crossbeam_channel::RecvTimeoutError::Timeout) => continue, Err(crossbeam_channel::RecvTimeoutError::Disconnected) => { bail!("PipeWire frame channel disconnected"); } @@ -163,6 +176,9 @@ fn drain_encoder( ) -> 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 { @@ -172,6 +188,9 @@ fn drain_encoder( 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); @@ -596,7 +615,7 @@ fn run_cpu_pipeline( stats.total_us.push(total_us); stats.frames_encoded += 1; - if stats.frames_encoded <= 3 || stats.frames_encoded % 30 == 0 { + if stats.frames_encoded <= 3 || stats.frames_encoded.is_multiple_of(30) { println!( " CPU frame {:>4}/{frames}: import={:.2}ms transfer={:.2}ms scale={:.2}ms encode={:.2}ms total={:.2}ms", stats.frames_encoded, @@ -754,7 +773,7 @@ fn run_gpu_pipeline( stats.total_us.push(total_us); stats.frames_encoded += 1; - if stats.frames_encoded <= 3 || stats.frames_encoded % 30 == 0 { + if stats.frames_encoded <= 3 || stats.frames_encoded.is_multiple_of(30) { println!( " GPU frame {:>4}/{frames}: import={:.2}ms filter={:.2}ms transfer={:.2}ms format={:.2}ms encode={:.2}ms total={:.2}ms", stats.frames_encoded, @@ -919,6 +938,11 @@ fn main() -> Result<()> { AvHwFrameCtx::for_capture(&hw_dev, src_width, src_height, ff::format::Pixel::BGRA)?; println!(" VAAPI frames context created OK (sw_format=BGRA)"); + // SAFETY: delegates to avhw::import_dma_buf_to_vaapi (itself an unsafe fn). + // frames_ctx is a valid AVBufferRef from AvHwFrameCtx::for_capture above; + // first_frame's fd/width/height/format/modifier/stride/offset are all sourced + // from the PipeWire-formatted PwDmaBufFrame. See that function's own SAFETY + // contract for the full rationale. let vaapi_frame = unsafe { import_dma_buf_to_vaapi( frames_ctx.as_ptr(), @@ -949,6 +973,9 @@ fn main() -> Result<()> { let mmap_size = (first_frame.stride as usize) * (first_frame.height as usize); let mmap_start = Instant::now(); + // SAFETY: first_frame.fd is an open DMA-BUF; offset/size from PipeWire. + // PROT_READ+MAP_SHARED is the standard read-only DMA-BUF mapping. Returns + // MAP_FAILED on error (checked below). let mmap_ptr = unsafe { libc::mmap( ptr::null_mut(), @@ -970,6 +997,8 @@ fn main() -> Result<()> { mmap_size as f64 / 1024.0 / 1024.0, mmap_elapsed.as_secs_f64() * 1000.0 ); + // SAFETY: mmap_ptr is a valid mapping (MAP_FAILED path was handled + // above); mmap_size matches the original mapping. POSIX munmap(2). unsafe { libc::munmap(mmap_ptr, mmap_size); } diff --git a/src/cap_portal.rs b/src/cap_portal.rs index f35456a..a6221be 100644 --- a/src/cap_portal.rs +++ b/src/cap_portal.rs @@ -158,6 +158,9 @@ impl CapPortal { let (frame_tx, frame_rx) = bounded(1); let (event_tx, event_rx) = bounded(8); + // SAFETY: eventfd(2) is a POSIX syscall with no preconditions; the init value + // and flags (CLOEXEC + NONBLOCK) are valid. Returns either a fresh fd (>= 0) + // or -1 on error, which we check immediately below. let efd = unsafe { libc::eventfd(0, libc::EFD_CLOEXEC | libc::EFD_NONBLOCK) }; if efd < 0 { return Err(anyhow::anyhow!( @@ -165,9 +168,13 @@ impl CapPortal { std::io::Error::last_os_error() )); } + // SAFETY: `efd` is the open eventfd we just created (>= 0 checked above) and + // own. dup(2) returns either a fresh fd or -1. let write_fd = unsafe { libc::dup(efd) }; if write_fd < 0 { let err = std::io::Error::last_os_error(); + // SAFETY: `efd` is still the open eventfd we own; closing on the error + // path before returning to avoid fd leak. unsafe { libc::close(efd) }; return Err(anyhow::anyhow!("dup eventfd failed: {err}")); } @@ -178,6 +185,10 @@ impl CapPortal { frame_tx, event_tx, dropped: pw_dropped.clone(), + // SAFETY: `efd` is the freshly-created eventfd (>= 0 checked above) and we + // are its sole owner. OwnedFd::from_raw_fd takes ownership and will close() + // it on Drop. Ownership transfers into PwThreadCtx and then into the + // PipeWire thread via pipewire_thread. shutdown_read: unsafe { OwnedFd::from_raw_fd(efd) }, pw_fd, node_id, @@ -190,11 +201,16 @@ impl CapPortal { pipewire_thread(ctx); }) .map_err(|e| { + // SAFETY: `write_fd` is the open dup'd eventfd we own (>= 0 checked + // above); closing on thread-spawn failure to avoid fd leak. unsafe { libc::close(write_fd) }; anyhow::anyhow!("thread spawn failed: {e}") })?; Ok(Self { + // SAFETY: `write_fd` is the freshly-dup'd eventfd (>= 0 checked above) and + // we are its sole owner. OwnedFd::from_raw_fd takes ownership and will + // close() it on Drop (which fires when CapPortal is dropped). shutdown_fd: unsafe { OwnedFd::from_raw_fd(write_fd) }, frame_rx, event_rx, @@ -433,7 +449,10 @@ fn verify_secure_dir(path: &std::path::Path) -> bool { return false; } // Must be owned by current user - if meta.uid() != unsafe { libc::getuid() } { + // SAFETY: libc::getuid has no preconditions and cannot fail; it simply + // returns the calling process's real user ID. + // SAFETY: libc::getuid has no preconditions and cannot fail. + if meta.uid() != unsafe { libc::getuid() } { tracing::warn!( "Token parent dir not owned by current user: {}", path.display() @@ -511,6 +530,7 @@ fn load_restore_token_from(path: PathBuf) -> Option { tracing::warn!("Token path is not a regular file: {}", path.display()); return None; } + // SAFETY: libc::getuid has no preconditions and cannot fail. if meta.uid() != unsafe { libc::getuid() } { tracing::warn!("Token file not owned by current user: {}", path.display()); return None; @@ -604,6 +624,9 @@ impl Drop for CapPortal { // Signal the PipeWire loop to quit via eventfd. // eventfd write is a kernel syscall — thread-safe and lock-free. let val: u64 = 1u64; + // SAFETY: shutdown_fd is a valid open eventfd (owned by Self); the buffer is + // a stack u64 of size 8 bytes which matches the count argument. POSIX write(2) + // is the standard fd-write syscall; eventfd writes must be exactly 8 bytes. let _ = unsafe { libc::write( self.shutdown_fd.as_raw_fd(), @@ -657,7 +680,7 @@ fn pipewire_thread(ctx: PwThreadCtx) { shutdown_read, pw_fd, node_id, - fps, + fps: _, } = ctx; let mainloop = match pw::main_loop::MainLoopBox::new(None) { @@ -803,6 +826,19 @@ fn pipewire_thread(ctx: PwThreadCtx) { let frame_tx = frame_tx.clone(); let dropped = dropped; move |stream, _| { + // SAFETY: raw_buf ownership invariant — PipeWire's process callback + // contract requires that every buffer acquired via `dequeue_raw_buffer` + // is returned to the queue EXACTLY ONCE via `queue_raw_buffer` before + // the callback returns — on every exit path, success or error. Failure + // to requeue leaks the buffer slot and eventually stalls the stream. + // + // Audit map of this closure (verified 2026-06-28): + // - null raw_buf (dequeue returned NULL) → nothing to requeue, return. + // - null spa_buf / no data / bad fd / null chunk / no format_info / + // invalid dims / dup_fd < 0 → all requeue before early-return. + // - success (try_send Ok / Full / Disconnected) → final requeue at end. + // The fd ownership is independent: dup() creates a fresh fd that lives + // inside PwDmaBufFrame; on try_send error the frame Drops and closes it. let raw_buf = unsafe { stream.dequeue_raw_buffer() }; if raw_buf.is_null() { tracing::trace!("process: null raw_buf"); @@ -810,36 +846,49 @@ fn pipewire_thread(ctx: PwThreadCtx) { } // 获取 SPA buffer 结构体,包含数据数组、元数据等 + // SAFETY: raw_buf was checked non-null above. `pw_buffer.buffer` is a + // valid raw pointer for the lifetime of raw_buf (PipeWire keeps the + // buffer alive until we queue it back). let spa_buf = unsafe { (*raw_buf).buffer }; if spa_buf.is_null() { tracing::trace!("process: null spa_buf"); + // SAFETY: raw_buf is the non-null buffer we still own; returning it. unsafe { stream.queue_raw_buffer(raw_buf) }; return; } // 获取 buffer 中的数据项数量和数据指针 // 对于 DMA-BUF 帧,通常只有 1 个数据项(包含 fd) + // SAFETY: spa_buf checked non-null above; `n_datas` is a plain u32 field. let n_datas = unsafe { (*spa_buf).n_datas }; + // SAFETY: same as above; `datas` is a raw pointer field, may be null. let datas_ptr = unsafe { (*spa_buf).datas }; if n_datas == 0 || datas_ptr.is_null() { tracing::trace!("process: no data (n_datas={n_datas})"); + // SAFETY: raw_buf still owned, returning it. unsafe { stream.queue_raw_buffer(raw_buf) }; return; } // 从第一个数据项中获取 DMA-BUF 文件描述符 // 通过 libspa 的 Data 包装类型安全地访问 SPA 数据结构 + // SAFETY: datas_ptr is non-null and n_datas > 0 (checked above). We cast + // to pw::spa::buffer::Data and take a shared borrow; PipeWire does not + // mutate the data array during a process cycle, so a shared reference + // for the duration of this callback is sound. let data_ref: &pw::spa::buffer::Data = unsafe { &*(datas_ptr as *const pw::spa::buffer::Data) }; let fd = data_ref.fd(); if fd < 0 { tracing::trace!("process: invalid fd={fd}"); + // SAFETY: raw_buf still owned, returning it. unsafe { stream.queue_raw_buffer(raw_buf) }; return; } if data_ref.as_raw().chunk.is_null() { tracing::trace!("process: null chunk"); + // SAFETY: raw_buf still owned, returning it. unsafe { stream.queue_raw_buffer(raw_buf) }; return; } @@ -850,6 +899,12 @@ fn pipewire_thread(ctx: PwThreadCtx) { // 从 SPA_META_Header 元数据中提取 PTS (显示时间戳) // 遍历 buffer 的所有元数据项,查找 Header 类型的元数据 // PTS 可用于音视频同步和帧率控制 + // SAFETY: spa_buf is non-null. `metas` is checked for null before + // iteration. We iterate `i in 0..n_metas` reading shared POD fields + // (type_, size, data) — PipeWire keeps the meta array immutable during + // a process cycle. The size guard (`meta.size >= size_of::()`) + // and null-data check before reading ensure we never read past the + // meta's actual extent. let pts: i64 = unsafe { let mut pts_val: i64 = 0; let n_metas = (*spa_buf).n_metas; @@ -873,11 +928,13 @@ fn pipewire_thread(ctx: PwThreadCtx) { // 验证格式信息已协商完成,且分辨率和格式有效 let Some((width, height, format, modifier)) = format_info.get() else { + // SAFETY: raw_buf still owned, returning it. unsafe { stream.queue_raw_buffer(raw_buf) }; return; }; if width == 0 || height == 0 || format == 0 { tracing::trace!("process: invalid dimensions {width}x{height} format={format}"); + // SAFETY: raw_buf still owned, returning it. unsafe { stream.queue_raw_buffer(raw_buf) }; return; } @@ -885,15 +942,27 @@ fn pipewire_thread(ctx: PwThreadCtx) { // 复制 DMA-BUF 文件描述符 // 必须 dup,因为原始 fd 由 PipeWire 管理,我们不能持有它 // dup 后的 fd 由 PwDmaBufFrame 持有,生命周期独立于 PipeWire buffer + // SAFETY: `fd` is the open DMA-BUF fd reported by PipeWire (>= 0 checked + // above). libc::dup is the standard POSIX fd duplication call. The + // original `fd` remains owned by PipeWire (returned with raw_buf later). let dup_fd = unsafe { libc::dup(fd) }; if dup_fd < 0 { + // SAFETY: raw_buf still owned, returning it. No fd cleanup needed + // because dup() failed and never returned a new fd. unsafe { stream.queue_raw_buffer(raw_buf) }; return; } // 构建帧数据对象,所有必要的帧信息已收集完毕 + // SAFETY: `dup_fd` is a freshly-dup'd open file descriptor (>= 0 checked + // above) and we are its sole owner. OwnedFd::from_raw_fd takes ownership + // and will close() it on Drop. The fd's lifecycle is independent of + // raw_buf: whether try_send succeeds (frame moves into the channel) or + // fails (Full/Disconnected — the error payload owns the frame and drops + // it at the end of the match arm), exactly one close() occurs per dup(). + let frame_fd = unsafe { OwnedFd::from_raw_fd(dup_fd) }; let frame = PwDmaBufFrame { - fd: unsafe { OwnedFd::from_raw_fd(dup_fd) }, + fd: frame_fd, offset, stride, modifier, @@ -910,6 +979,8 @@ fn pipewire_thread(ctx: PwThreadCtx) { } Err(crossbeam_channel::TrySendError::Disconnected(_)) => {} } + // SAFETY: final exactly-once requeue of raw_buf. Every path above + // either returned early with its own requeue, or falls through to here. unsafe { stream.queue_raw_buffer(raw_buf) }; } }) @@ -949,6 +1020,10 @@ fn pipewire_thread(ctx: PwThreadCtx) { move |fd| { // Drain the eventfd so it doesn't re-trigger let mut buf: u64 = 0; + // SAFETY: `fd` is the registered eventfd owned by the mainloop source; the + // buffer is a stack u64 of 8 bytes matching the count argument. POSIX + // read(2) is the standard fd-read syscall; eventfd semantics require the + // 8-byte buffer. let _ = unsafe { libc::read( fd.as_raw_fd(), diff --git a/src/main.rs b/src/main.rs index c36e8fd..c491890 100644 --- a/src/main.rs +++ b/src/main.rs @@ -148,6 +148,9 @@ fn run_wlr_screencopy(args: Args) -> Result<()> { revents: 0, }; // timeout=0 表示非阻塞,立即返回当前 fd 状态 + // SAFETY: `pfd` is a stack-allocated libc::pollfd initialized above with a + // valid wayland_fd and POLLIN events; nfds=1 matches the single-element + // array; timeout=0 is non-blocking. POSIX poll(2) writes revents in place. let ret = unsafe { libc::poll(&mut pfd, 1, 0) }; tracing::info!( "Raw poll on wayland fd={wayland_fd}: ret={ret}, revents={}", @@ -173,7 +176,7 @@ fn run_wlr_screencopy(args: Args) -> Result<()> { // 注册 SIGINT / SIGTERM 信号用于优雅退出 // signal_hook_mio 将 Unix 信号转换为 fd 可读事件, // 这样信号也可以通过 epoll 统一监听,不需要单独的信号处理器 - let mut signals = signal_hook_mio::v1_0::Signals::new(&[ + let mut signals = signal_hook_mio::v1_0::Signals::new([ signal_hook::consts::SIGINT, // Ctrl+C signal_hook::consts::SIGTERM, // kill 命令默认信号 ])?; @@ -310,7 +313,7 @@ fn run_portal_pipewire(args: Args) -> Result<()> { // Set up signal handling only (no Wayland fd needed) // Portal 后端不需要监听 Wayland fd,只需处理 Unix 信号 // 因为帧数据是通过 PipeWire 独立投递的,不走 Wayland 协议 - let mut signals = signal_hook_mio::v1_0::Signals::new(&[ + let mut signals = signal_hook_mio::v1_0::Signals::new([ signal_hook::consts::SIGINT, signal_hook::consts::SIGTERM, ])?; diff --git a/src/state.rs b/src/state.rs index 3eaaecc..64c448a 100644 --- a/src/state.rs +++ b/src/state.rs @@ -83,6 +83,7 @@ pub struct OutputInfo { pub logical_position: (i32, i32), } +#[derive(Default)] pub struct PartialOutputInfo { pub name: Option, /// Name from wl_output::Name (v4) — used to match wlr-output-management heads @@ -95,19 +96,6 @@ pub struct PartialOutputInfo { pub done_count: u32, } -impl Default for PartialOutputInfo { - fn default() -> Self { - Self { - name: None, - wl_name: None, - transform: None, - physical_size: None, - logical_position: None, - mode_size: None, - done_count: 0, - } - } -} /// Stores head info from wlr-output-management for name-based matching with wl_output. struct WlrHeadInfo { @@ -512,6 +500,10 @@ impl State { unsafe { (*map_frame.as_mut_ptr()).format = ffi::AVPixelFormat::AV_PIX_FMT_DRM_PRIME as i32; } + // SAFETY: map_frame and surface are valid, owned AVFrame pointers from + // av_hwframe_get/surface.alloc above. AV_HWFRAME_MAP_READ flag (0 here) + // requests a read-only mapping. The DRM_PRIME format set above instructs + // FFmpeg to populate data[0] with an AVDRMFrameDescriptor on success. let ret = unsafe { ffi::av_hwframe_map(map_frame.as_mut_ptr(), surface.as_ptr(), 0) }; if ret < 0 { tracing::error!("av_hwframe_map failed: {}", crate::avhw::ff_err(ret)); @@ -1509,7 +1501,7 @@ impl Dispatch for State { event: ::Event, _data: &(), _conn: &wayland_client::Connection, - qhandle: &QueueHandle>, + _qhandle: &QueueHandle>, ) { match event { WlrOutputManagerEvent::Head { head } => { @@ -1530,7 +1522,7 @@ impl Dispatch for State { } } } - WlrOutputManagerEvent::Finished { .. } => { + WlrOutputManagerEvent::Finished => { tracing::warn!("zwlr_output_manager_v1::Finished received during probing"); } _ => {} @@ -1583,7 +1575,7 @@ impl Dispatch for State { } } } - WlrHeadEvent::Finished { .. } => { + WlrHeadEvent::Finished => { tracing::debug!("zwlr_output_head_v1::Finished received"); } _ => {} diff --git a/src/state_portal.rs b/src/state_portal.rs index 15eb327..3330816 100644 --- a/src/state_portal.rs +++ b/src/state_portal.rs @@ -478,6 +478,10 @@ impl StatePortal { if let Some(enc) = self.enc.as_mut() { // 将 DMA-BUF 帧零拷贝导入 VAAPI 硬件帧池 + // SAFETY: delegates to avhw::import_dma_buf_to_vaapi (itself an unsafe fn); + // frames_rgb pointer is a valid AVBufferRef owned by enc, and frame's + // fd/width/height/format/modifier/stride/offset come straight from the + // PipeWire-formatted PwDmaBufFrame. See that function's own SAFETY contract. let mut vaapi_frame = unsafe { avhw::import_dma_buf_to_vaapi( enc.frames_rgb().as_ptr(), @@ -495,6 +499,8 @@ impl StatePortal { let t_encode_start = Instant::now(); // 设置帧的显示时间戳(PTS),基于已编码帧序号 + // SAFETY: vaapi_frame is the freshly-imported valid AVFrame returned by + // import_dma_buf_to_vaapi above; pts is a plain i64 field on AVFrame. unsafe { (*vaapi_frame.as_mut_ptr()).pts = pts; } @@ -515,6 +521,8 @@ impl StatePortal { }; self.stats.record_encode(&timings); } else if let Some(import) = self.enc_import.as_mut() { + // SAFETY: same contract as the enc branch above — frames_rgb owned by + // import, frame fields come from the PipeWire PwDmaBufFrame. let mut vaapi_frame = unsafe { avhw::import_dma_buf_to_vaapi( import.frames_rgb().as_ptr(), @@ -527,6 +535,7 @@ impl StatePortal { frame.offset, ) }?; + // SAFETY: vaapi_frame is the valid AVFrame returned above; pts is plain i64. unsafe { (*vaapi_frame.as_mut_ptr()).pts = pts; } @@ -756,7 +765,7 @@ fn webrtc_thread_loop( let should_send = match last_sent_bitrate { None => true, Some(last) => { - let diff = if bwe > last { bwe - last } else { last - bwe }; + let diff = bwe.abs_diff(last); diff * 10 > last } }; @@ -946,7 +955,12 @@ fn resolve_drm_device(args: &Args) -> Result> { /// 用于验证 DMA-BUF 元数据映射的正确性。 #[cfg(test)] fn build_drm_descriptor(frame: &PwDmaBufFrame) -> ffmpeg_next::ffi::AVDRMFrameDescriptor { - let mut desc: ffmpeg_next::ffi::AVDRMFrameDescriptor = unsafe { std::mem::zeroed() }; + let mut desc: ffmpeg_next::ffi::AVDRMFrameDescriptor = { + // SAFETY: AVDRMFrameDescriptor is a POD struct from FFmpeg's C API with no + // pointers orDrop fields; all-zero is a valid initial state. Every field is + // explicitly overwritten in the lines below before the descriptor is used. + unsafe { std::mem::zeroed() } + }; desc.nb_objects = 1; // 单个 DMA-BUF 对象 desc.objects[0].fd = frame.fd.as_raw_fd(); // DMA-BUF 文件描述符 desc.objects[0].size = 0; // 大小设为 0(内核自动确定) @@ -969,6 +983,9 @@ mod tests { fn make_test_frame() -> PwDmaBufFrame { // Create a dummy fd from stderr (always valid fd 2) // 使用 stderr(fd 2)的副本作为虚拟文件描述符 + // SAFETY: stderr (fd 2) is always-open in any process; libc::dup(2) returns + // a fresh fd we solely own. OwnedFd::from_raw_fd takes ownership and closes + // it on Drop. Test-only; the fd is never actually memory-mapped. let fd = unsafe { OwnedFd::from_raw_fd(libc::dup(2)) }; PwDmaBufFrame { fd, @@ -1091,8 +1108,10 @@ mod tests { /// 测试:使用自定义偏移量和 stride 构建 DRM 描述符 #[test] fn build_drm_descriptor_custom_offset_and_stride() { + // SAFETY: same as make_test_frame — dup of stderr (fd 2), test-only. + let test_fd = unsafe { OwnedFd::from_raw_fd(libc::dup(2)) }; let frame = PwDmaBufFrame { - fd: unsafe { OwnedFd::from_raw_fd(libc::dup(2)) }, + fd: test_fd, offset: 4096, // 4KB 对齐偏移 stride: 3840 * 4, // 4K 宽度 × 4 字节 modifier: 0x0100000000000001, // AMD modifiers diff --git a/src/stats.rs b/src/stats.rs index 2c6bfb6..42ba1d6 100644 --- a/src/stats.rs +++ b/src/stats.rs @@ -73,6 +73,12 @@ pub struct PipelineStats { window_start: Instant, } +impl Default for PipelineStats { + fn default() -> Self { + Self::new() + } +} + impl PipelineStats { pub fn new() -> Self { Self { diff --git a/src/webrtc.rs b/src/webrtc.rs index bcf5e93..b634b3f 100644 --- a/src/webrtc.rs +++ b/src/webrtc.rs @@ -536,7 +536,7 @@ impl WebRtcInner { let now = Instant::now(); let should_honor = self .last_forced_keyframe_at - .map_or(true, |last| now.duration_since(last) >= FORCED_KEYFRAME_MIN_INTERVAL); + .is_none_or(|last| now.duration_since(last) >= FORCED_KEYFRAME_MIN_INTERVAL); if should_honor { self.last_forced_keyframe_at = Some(now); self.need_keyframe = true;