From ee18227cac28caf0df65a932fef5071278ea69c8 Mon Sep 17 00:00:00 2001 From: Kyzer <36684609+kyzer1023@users.noreply.github.com> Date: Mon, 26 Oct 2026 03:55:51 +0000 Subject: [PATCH 1/3] perf(linux-pipewire): alias frames until encode; keep DMA-BUF mapped Linear PipeWire frames are read-only views of the producer buffer until the last frame reference drops, so the encoder can finish without a per-frame row copy. DMA-BUF mappings stay cached while the fd and size are stable. Instant-with-camera still copies because the alias is read-only and av_frame_make_writable allocates private pixels before the camera blend. A private copy is also used when the buffer pool has no spare slot. --- .../src/instant_recording/linux_camera.rs | 2 + .../src/sources/screen_capture/linux.rs | 321 ++--- .../screen_capture/linux_pipewire_frames.rs | 1034 +++++++++++++++++ .../src/sources/screen_capture/mod.rs | 2 + 4 files changed, 1213 insertions(+), 146 deletions(-) create mode 100644 crates/recording/src/sources/screen_capture/linux_pipewire_frames.rs diff --git a/crates/recording/src/instant_recording/linux_camera.rs b/crates/recording/src/instant_recording/linux_camera.rs index 84607cb807b..067011aafb4 100644 --- a/crates/recording/src/instant_recording/linux_camera.rs +++ b/crates/recording/src/instant_recording/linux_camera.rs @@ -1135,6 +1135,8 @@ impl Compositor { return None; } let layout = PackedRgbLayout::of(screen.format())?; + // PipeWire aliases are read-only, so the blend copies into private + // pixels. An owned screen frame is already writable and stays put. let writable = unsafe { ffmpeg::sys::av_frame_make_writable(screen.as_mut_ptr()) }; (writable >= 0).then_some(layout) } diff --git a/crates/recording/src/sources/screen_capture/linux.rs b/crates/recording/src/sources/screen_capture/linux.rs index 7ca0ef3bed0..4c9dae649a1 100644 --- a/crates/recording/src/sources/screen_capture/linux.rs +++ b/crates/recording/src/sources/screen_capture/linux.rs @@ -1,5 +1,6 @@ use super::cadence::FrameCadenceGate; use super::linux_dmabuf_import; +use super::linux_pipewire_frames::{self, LinearGeometry, PipewireFrames, SourcePlane}; use super::linux_shm::{self, ShmSegment}; use super::*; use crate::feeds::microphone::{self, MicrophoneFeed, MicrophoneFeedLock}; @@ -301,7 +302,8 @@ struct PipewireCaptureState { scaler: Option, video_info: VideoInfo, crop_bounds: Option, - video_tx: mpsc::Sender, + video_tx: Arc>>>, + frames: PipewireFrames, health_tx: output_pipeline::HealthSender, stop_requested: Arc, fatal_error: Arc>>, @@ -551,12 +553,17 @@ fn capture_pipewire_node( .map(|importer| pipewire_imported_format_params(input.fps, importer)) .unwrap_or_default(); + let frames = PipewireFrames::new(); + let video_tx = Arc::new(parking_lot::Mutex::new(Some(video_tx))); + let frames_shutdown = frames.clone(); + let video_tx_shutdown = video_tx.clone(); let state = PipewireCaptureState { format: Default::default(), scaler: None, video_info, crop_bounds: input.crop_bounds, video_tx, + frames, health_tx, stop_requested: stop_requested.clone(), fatal_error: fatal_error.clone(), @@ -580,6 +587,14 @@ fn capture_pipewire_node( ) .context("create PipeWire screen capture stream")?; + let (release_tx, release_rx) = pw::channel::channel(); + frames_shutdown.set_requeue_sender(release_tx); + frames_shutdown.set_stream(stream.as_raw_ptr()); + let _release_rx = release_rx.attach(thread_loop.loop_(), { + let frames = frames_shutdown.clone(); + move |buffer| frames.queue_released_buffer(buffer) + }); + let _listener = stream .add_local_listener_with_user_data(state) .state_changed(|_, state, _, new| { @@ -591,15 +606,31 @@ fn capture_pipewire_node( match update_pipewire_format(state, id, param) { Ok(PipewireFormatChange::DmaBuf) => { // A compositor that only shares DMA-BUFs (niri) still has to - // be told the consumer takes them. - if let Err(error) = request_dmabuf_buffers(stream) { + // be told the consumer takes them. Ask for a pool large + // enough that one buffer can stay borrowed by the encoder. + if let Err(error) = + request_capture_buffers(stream, 1 << spa::buffer::DataType::DmaBuf.as_raw()) + { + state.set_fatal_error(error.to_string()); + } + } + Ok(PipewireFormatChange::Memory) => { + let memory = (1 << spa::buffer::DataType::MemFd.as_raw()) + | (1 << spa::buffer::DataType::MemPtr.as_raw()); + if let Err(error) = request_capture_buffers(stream, memory) { state.set_fatal_error(error.to_string()); } } - Ok(PipewireFormatChange::None | PipewireFormatChange::Memory) => {} + Ok(PipewireFormatChange::None) => {} Err(error) => state.set_fatal_error(error.to_string()), } }) + .add_buffer(|_, state, _buffer| { + state.frames.add_pool_buffer(); + }) + .remove_buffer(|_, state, buffer| { + state.frames.remove_pool_buffer(buffer); + }) .process(|stream, state| { if state.stop_requested.load(Ordering::Relaxed) { return; @@ -660,6 +691,14 @@ fn capture_pipewire_node( stop_requested.store(true, Ordering::Relaxed); thread_loop.stop(); + // The muxer holds the latest frame until the next one replaces it, so an + // alias can outlive this thread. Drop the sender to let that drain finish, + // then destroy the stream only after the last reference is gone. + // `pw_stream_queue_buffer` is single-producer on this loop; a late release + // only posts the buffer address here. Destroying the stream earlier would + // unmap memory the encoder is still reading. + video_tx_shutdown.lock().take(); + let idle = frames_shutdown.wait_idle(Duration::from_secs(30)); let error = fatal_error.lock().take(); tracing::info!( @@ -670,6 +709,17 @@ fn capture_pipewire_node( "Linux Wayland PipeWire capture stopped" ); + if !idle { + tracing::error!("PipeWire screen capture stopped while a frame was still borrowed"); + std::mem::forget(_release_rx); + std::mem::forget(_listener); + std::mem::forget(stream); + std::mem::forget(core); + std::mem::forget(context); + std::mem::forget(thread_loop); + bail!("PipeWire screen capture stopped while a frame was still borrowed"); + } + if let Some(error) = error { Err(anyhow!(error)) } else { @@ -715,6 +765,7 @@ fn update_pipewire_format( ) })?; state.format = format; + state.frames.retire_all(); let dmabuf = param.as_object().is_ok_and(|object| { object @@ -737,18 +788,28 @@ fn update_pipewire_format( }) } -fn request_dmabuf_buffers(stream: &pw::stream::Stream) -> anyhow::Result<()> { - let dmabuf = 1 << spa::buffer::DataType::DmaBuf.as_raw(); +fn request_capture_buffers(stream: &pw::stream::Stream, data_type: i32) -> anyhow::Result<()> { let object = spa::pod::object!( spa::utils::SpaTypes::ObjectParamBuffers, spa::param::ParamType::Buffers, + spa::pod::Property::new( + spa::sys::SPA_PARAM_BUFFERS_buffers, + spa::pod::Value::Choice(spa::pod::ChoiceValue::Int(spa::utils::Choice( + spa::utils::ChoiceFlags::empty(), + spa::utils::ChoiceEnum::Range { + default: 8, + min: 1, + max: 32, + }, + ))), + ), spa::pod::Property::new( spa::sys::SPA_PARAM_BUFFERS_dataType, spa::pod::Value::Choice(spa::pod::ChoiceValue::Int(spa::utils::Choice( spa::utils::ChoiceFlags::empty(), spa::utils::ChoiceEnum::Flags { - default: dmabuf, - flags: vec![dmabuf], + default: data_type, + flags: vec![data_type], }, ))), ) @@ -758,109 +819,63 @@ fn request_dmabuf_buffers(stream: &pw::stream::Stream) -> anyhow::Result<()> { .ok_or_else(|| anyhow!("create PipeWire buffers parameter"))?; stream .update_params(&mut [pod]) - .context("ask PipeWire for DMA-BUF screen capture buffers") + .context("ask PipeWire for screen capture buffers") } -/// A DMA-BUF the producer did not map, mapped for reading one frame. LINEAR -/// buffers are plain rows a CPU can read once the kernel syncs the buffer. -struct DmaBufMapping { - fd: std::os::fd::RawFd, - pointer: *mut libc::c_void, - len: usize, +fn serialize_pod(object: spa::pod::Object, what: &str) -> anyhow::Result> { + Ok(spa::pod::serialize::PodSerializer::serialize( + std::io::Cursor::new(Vec::new()), + &spa::pod::Value::Object(object), + ) + .map_err(|error| anyhow!("serialize {what}: {error:?}"))? + .0 + .into_inner()) } -const DMA_BUF_IOCTL_SYNC: libc::c_ulong = 0x4008_6200; -const DMA_BUF_SYNC_READ: u64 = 1; -const DMA_BUF_SYNC_END: u64 = 1 << 2; - -impl DmaBufMapping { - fn map(fd: std::os::fd::RawFd, minimum_len: usize) -> anyhow::Result { - if fd < 0 { - bail!("PipeWire screen capture DMA-BUF had no file descriptor"); - } - let size = unsafe { libc::lseek(fd, 0, libc::SEEK_END) }; - let len = usize::try_from(size) - .ok() - .filter(|len| *len > 0) - .unwrap_or(minimum_len); - if len < minimum_len { - bail!( - "PipeWire screen capture DMA-BUF holds {len} bytes, the frame needs {minimum_len}" - ); - } - let pointer = unsafe { - libc::mmap( - std::ptr::null_mut(), - len, - libc::PROT_READ, - libc::MAP_SHARED, - fd, - 0, - ) - }; - if pointer == libc::MAP_FAILED { - return Err(std::io::Error::last_os_error()) - .context("map the PipeWire screen capture DMA-BUF"); - } - let mapping = Self { fd, pointer, len }; - mapping - .sync(DMA_BUF_SYNC_READ) - .context("sync the PipeWire screen capture DMA-BUF for reading")?; - Ok(mapping) - } - - /// The kernel's DMA-BUF read fence. A buffer that is not a DMA-BUF (a - /// memfd standing in for one) has none and answers ENOTTY. - fn sync(&self, flags: u64) -> std::io::Result<()> { - let sync = flags; - loop { - let result = unsafe { libc::ioctl(self.fd, DMA_BUF_IOCTL_SYNC, &sync as *const u64) }; - if result == 0 { - return Ok(()); - } - let error = std::io::Error::last_os_error(); - match error.raw_os_error() { - Some(libc::EINTR | libc::EAGAIN) => continue, - Some(libc::ENOTTY) => return Ok(()), - _ => return Err(error), - } - } - } - - fn bytes(&self) -> &[u8] { - unsafe { std::slice::from_raw_parts(self.pointer as *const u8, self.len) } - } +struct HeldBuffer<'a> { + stream: &'a pw::stream::Stream, + buffer: *mut pw::sys::pw_buffer, + leased: bool, } -impl Drop for DmaBufMapping { +impl Drop for HeldBuffer<'_> { fn drop(&mut self) { - if let Err(error) = self.sync(DMA_BUF_SYNC_READ | DMA_BUF_SYNC_END) { - tracing::warn!(%error, "ending the PipeWire screen capture DMA-BUF read failed"); + if self.leased || self.buffer.is_null() { + return; } unsafe { - libc::munmap(self.pointer, self.len); + self.stream.queue_raw_buffer(self.buffer); } } } -fn serialize_pod(object: spa::pod::Object, what: &str) -> anyhow::Result> { - Ok(spa::pod::serialize::PodSerializer::serialize( - std::io::Cursor::new(Vec::new()), - &spa::pod::Value::Object(object), - ) - .map_err(|error| anyhow!("serialize {what}: {error:?}"))? - .0 - .into_inner()) +unsafe fn pw_buffer_datas(buffer: *mut pw::sys::pw_buffer) -> &'static mut [spa::buffer::Data] { + unsafe { + if buffer.is_null() { + return &mut []; + } + let spa = (*buffer).buffer; + if spa.is_null() || (*spa).datas.is_null() || (*spa).n_datas == 0 { + return &mut []; + } + std::slice::from_raw_parts_mut((*spa).datas.cast(), (*spa).n_datas as usize) + } } fn process_pipewire_frame( stream: &pw::stream::Stream, state: &mut PipewireCaptureState, ) -> anyhow::Result> { - let Some(mut buffer) = stream.dequeue_buffer() else { + let buffer = unsafe { stream.dequeue_raw_buffer() }; + if buffer.is_null() { return Ok(None); + } + let mut held = HeldBuffer { + stream, + buffer, + leased: false, }; - let datas = buffer.datas_mut(); + let datas = unsafe { pw_buffer_datas(buffer) }; if datas.is_empty() { return Ok(None); } @@ -892,17 +907,30 @@ fn process_pipewire_frame( importer, &mut state.imported, )? + .map(|frame| (frame, false)) } else { - frame_from_pipewire_data(&mut datas[0], state.format, state.crop_bounds)? + frame_from_pipewire_data( + &mut datas[0], + state.format, + state.crop_bounds, + &state.frames, + buffer as usize, + state.frames.can_lease(), + )? }; - let Some(raw_frame) = raw_frame else { + let Some((raw_frame, leased)) = raw_frame else { return Ok(Some(StallSendOutcome::StalledAndDropped { waited_ms: 0 })); }; + held.leased = leased; let frame = prepare_pipewire_frame(raw_frame, &mut state.scaler, state.video_info)?; let timestamp = Timestamp::Instant(captured_at); + let mut video_tx = state.video_tx.lock(); + let Some(video_tx) = video_tx.as_mut() else { + return Ok(Some(StallSendOutcome::Disconnected)); + }; Ok(Some(send_with_stall_budget_futures( - &mut state.video_tx, + video_tx, FFmpegVideoFrame { inner: frame, timestamp, @@ -943,7 +971,10 @@ fn frame_from_pipewire_data( data: &mut spa::buffer::Data, format: spa::param::video::VideoInfoRaw, crop_bounds: Option, -) -> anyhow::Result> { + frames: &PipewireFrames, + pw_buffer: usize, + lease: bool, +) -> anyhow::Result> { let (pixel_format, bytes_per_pixel) = pipewire_pixel_format(format.format()).ok_or_else(|| { anyhow!( @@ -978,59 +1009,55 @@ fn frame_from_pipewire_data( let (crop_x, crop_y, crop_width, crop_height) = pipewire_crop(source_width, source_height, crop_bounds)?; let offset = chunk_offset as usize; - let dmabuf_fd = (data.type_() == spa::buffer::DataType::DmaBuf).then(|| data.fd() as i32); - let mapping; - let source: &[u8] = match data.data() { - Some(source) => source, - None => { - let Some(fd) = dmabuf_fd else { - bail!("PipeWire screen capture buffer was not memory-mapped"); - }; - mapping = DmaBufMapping::map(fd, offset + source_stride * source_height)?; - mapping.bytes() + let data_type = data.type_(); + let fd = data.fd(); + let dmabuf = data_type == spa::buffer::DataType::DmaBuf; + let memfd = data_type == spa::buffer::DataType::MemFd; + let plane = if (dmabuf || memfd) && fd >= 0 { + SourcePlane { + data: std::ptr::null_mut(), + data_len: 0, + fd, + sync_dma_buf: dmabuf, + pw_buffer, + } + } else if let Some(mapped) = data.data() { + let data_len = mapped.len(); + let data = mapped.as_mut_ptr(); + SourcePlane { + data, + data_len, + fd: -1, + sync_dma_buf: false, + pw_buffer, } + } else { + bail!("PipeWire screen capture buffer was not memory-mapped"); }; // DMA-BUF producers may leave the chunk size unset (the PipeWire DMA-BUF // docs tell consumers to ignore it), so the mapping bounds the frame. - let source_limit = if chunk_size > 0 && dmabuf_fd.is_none() { - offset - .checked_add(chunk_size as usize) - .map(|limit| limit.min(source.len())) - .ok_or_else(|| anyhow!("PipeWire screen capture frame size overflowed"))? + let limit = if chunk_size > 0 && !dmabuf { + Some( + offset + .checked_add(chunk_size as usize) + .ok_or_else(|| anyhow!("PipeWire screen capture frame size overflowed"))?, + ) } else { - source.len() + None }; - let row_bytes = crop_width * bytes_per_pixel; - - let mut frame = ffmpeg::frame::Video::new(pixel_format, crop_width as u32, crop_height as u32); - let target_stride = frame.stride(0); - if target_stride < row_bytes { - bail!( - "PipeWire target frame stride was too small: {} for {}x{}", - target_stride, - crop_width, - crop_height - ); - } - - for y in 0..crop_height { - let source_start = offset + (crop_y + y) * source_stride + crop_x * bytes_per_pixel; - let source_end = source_start + row_bytes; - if source_end > source_limit { - bail!( - "PipeWire screen capture frame was too small: {} bytes for {}x{}", - source.len(), - source_width, - source_height - ); - } - - let target_start = y * target_stride; - frame.data_mut(0)[target_start..target_start + row_bytes] - .copy_from_slice(&source[source_start..source_end]); - } - - Ok(Some(frame)) + let geometry = LinearGeometry { + pixel_format, + bytes_per_pixel, + source_stride, + crop_x, + crop_y, + crop_width, + crop_height, + offset, + limit, + }; + let (frame, leased) = linux_pipewire_frames::linear_frame(frames, &plane, &geometry, lease)?; + Ok(Some((frame, leased))) } /// A frame with a GPU-tiled modifier, read back through EGL as RGBA. @@ -3421,10 +3448,10 @@ mod system_audio_tests { #[cfg(test)] mod pipewire_frame_tests { use super::{ - FrameScaler, LinuxCaptureSource, PipewireNodeInput, VideoInfo, capture_pipewire_node, - frame_from_pipewire_data, pipewire_dmabuf_format_param, pipewire_format_param, - prefers_wayland_environment, prepare_pipewire_frame, spa, wayland_area_matches_display, - wayland_video_info, + FrameScaler, LinuxCaptureSource, PipewireFrames, PipewireNodeInput, VideoInfo, + capture_pipewire_node, frame_from_pipewire_data, pipewire_dmabuf_format_param, + pipewire_format_param, prefers_wayland_environment, prepare_pipewire_frame, spa, + wayland_area_matches_display, wayland_video_info, }; use std::os::fd::AsRawFd; @@ -3567,9 +3594,11 @@ mod pipewire_frame_tests { height: height as u32, }); format.set_modifier(0); - let frame = frame_from_pipewire_data(data, format, None) + let frames = PipewireFrames::new(); + let (frame, leased) = frame_from_pipewire_data(data, format, None, &frames, 0, true) .unwrap() .expect("a frame"); + assert!(leased); assert_eq!((frame.width(), frame.height()), (4, 3)); assert_eq!(frame.format(), ffmpeg::format::Pixel::BGRZ); for y in 0..height { diff --git a/crates/recording/src/sources/screen_capture/linux_pipewire_frames.rs b/crates/recording/src/sources/screen_capture/linux_pipewire_frames.rs new file mode 100644 index 00000000000..317b7e62b1b --- /dev/null +++ b/crates/recording/src/sources/screen_capture/linux_pipewire_frames.rs @@ -0,0 +1,1034 @@ +//! Linear PipeWire frames borrowed until the encoder is finished. +//! +//! A screencast buffer is a read-only view of the producer's memory. The +//! view stays alive until the last frame reference is dropped: the encoder +//! reads it, and the muxer also holds the latest frame so a static tail can +//! be repeated. DMA-BUF and memfd mappings are cached and replaced only when +//! the fd or the mapped size changes. +//! +//! Two paths still allocate a private writable copy: +//! - Instant mode with a camera. The alias is created with +//! `AV_BUFFER_FLAG_READONLY`, so `av_frame_make_writable` copies before the +//! camera is blended. Writing the alias would paint into the PipeWire buffer. +//! - The negotiated pool has no spare buffer (`pool < 2` or every buffer but +//! one is already borrowed). The muxer keeps the latest frame borrowed, so a +//! two-buffer pool copies in the steady state instead of stalling the +//! producer. A crop whose lines would make FFmpeg's line window extend past +//! the mapping is copied for the same reason. + +use std::{ + os::fd::RawFd, + sync::{ + Arc, + atomic::{AtomicPtr, AtomicUsize, Ordering}, + }, + time::{Duration, Instant}, +}; + +use anyhow::{Context, anyhow, bail}; +use parking_lot::{Condvar, Mutex}; +use pipewire as pw; + +const DMA_BUF_IOCTL_SYNC: libc::c_ulong = 0x4008_6200; +const DMA_BUF_SYNC_READ: u64 = 1; +const DMA_BUF_SYNC_END: u64 = 1 << 2; + +/// Leave at least one buffer with the producer. The muxer borrows the latest +/// frame until the next one replaces it, so a pool of two copies in the +/// steady state and a pool of one always copies. +pub(super) fn spare_buffer(pool: usize, inflight: usize) -> bool { + pool >= 2 && inflight + 1 < pool +} + +pub(super) struct SourcePlane { + /// Producer pointer when PipeWire already mapped a MemPtr. Null when this + /// plane is read through [`Self::fd`]. + pub data: *mut u8, + pub data_len: usize, + /// Negative when the plane has no fd of its own. + pub fd: i32, + pub sync_dma_buf: bool, + /// `pw_buffer` address, queued back when the last frame reference drops. + /// Zero in tests that have no PipeWire buffer. + pub pw_buffer: usize, +} + +#[derive(Clone, Copy)] +pub(super) struct LinearGeometry { + pub pixel_format: ffmpeg::format::Pixel, + pub bytes_per_pixel: usize, + pub source_stride: usize, + pub crop_x: usize, + pub crop_y: usize, + pub crop_width: usize, + pub crop_height: usize, + pub offset: usize, + /// Exclusive readable end. `None` uses the whole mapping. + pub limit: Option, +} + +struct MappedBuffer { + /// Producer fd. PipeWire may close it while an alias is still borrowed. + fd: RawFd, + /// Dup of `fd`, used for mmap and the read fence so that close cannot + /// recycle the descriptor under the encoder. + owned: RawFd, + len: usize, + pointer: usize, + in_use: bool, + retire: bool, +} + +struct DmaBufCache { + maps: Vec, + map_calls: usize, + unmap_calls: usize, +} + +impl DmaBufCache { + fn map(&mut self, fd: RawFd, minimum_len: usize) -> anyhow::Result<(*mut u8, usize)> { + let len = mapped_len(fd, minimum_len)?; + if let Some(index) = self.maps.iter().position(|map| map.fd == fd && !map.retire) { + if self.maps[index].len == len { + return Ok((self.maps[index].pointer as *mut u8, len)); + } + self.maps[index].retire = true; + self.unmap_retired(); + } + let owned = unsafe { libc::fcntl(fd, libc::F_DUPFD_CLOEXEC, 0) }; + if owned < 0 { + return Err(std::io::Error::last_os_error()) + .context("dup the PipeWire screen capture DMA-BUF"); + } + let pointer = unsafe { + libc::mmap( + std::ptr::null_mut(), + len, + libc::PROT_READ, + libc::MAP_SHARED, + owned, + 0, + ) + }; + if pointer == libc::MAP_FAILED { + unsafe { + libc::close(owned); + } + return Err(std::io::Error::last_os_error()) + .context("map the PipeWire screen capture DMA-BUF"); + } + let pointer = pointer as usize; + self.map_calls += 1; + self.maps.push(MappedBuffer { + fd, + owned, + len, + pointer, + in_use: false, + retire: false, + }); + Ok((pointer as *mut u8, len)) + } + + fn owned_fd(&self, pointer: *mut u8) -> Option { + let pointer = pointer as usize; + self.maps + .iter() + .find(|map| map.pointer == pointer) + .map(|map| map.owned) + } + + fn lease(&mut self, pointer: *mut u8) { + let pointer = pointer as usize; + if let Some(map) = self.maps.iter_mut().find(|map| map.pointer == pointer) { + map.in_use = true; + } + } + + fn release(&mut self, pointer: usize) { + if pointer == 0 { + return; + } + let Some(index) = self.maps.iter().position(|map| map.pointer == pointer) else { + return; + }; + self.maps[index].in_use = false; + if self.maps[index].retire { + self.unmap_at(index); + } + } + + #[cfg(test)] + fn unmap_idle(&mut self) { + let mut index = 0; + while index < self.maps.len() { + if self.maps[index].in_use { + index += 1; + } else { + self.unmap_at(index); + } + } + } + + fn retire_fd(&mut self, fd: RawFd) { + if fd < 0 { + return; + } + for map in &mut self.maps { + if map.fd == fd { + map.retire = true; + } + } + self.unmap_retired(); + } + + fn retire_all(&mut self) { + for map in &mut self.maps { + map.retire = true; + } + self.unmap_retired(); + } + + fn unmap_retired(&mut self) { + let mut index = 0; + while index < self.maps.len() { + if self.maps[index].retire && !self.maps[index].in_use { + self.unmap_at(index); + } else { + index += 1; + } + } + } + + fn unmap_at(&mut self, index: usize) { + let map = self.maps.remove(index); + unsafe { + libc::munmap(map.pointer as *mut libc::c_void, map.len); + libc::close(map.owned); + } + self.unmap_calls += 1; + } + + #[cfg(test)] + fn contains(&self, pointer: usize) -> bool { + self.maps.iter().any(|map| map.pointer == pointer) + } + + #[cfg(test)] + fn base_for(&self, fd: RawFd) -> Option<*const u8> { + self.maps + .iter() + .find(|map| map.fd == fd && !map.retire) + .map(|map| map.pointer as *const u8) + } +} + +impl Drop for DmaBufCache { + fn drop(&mut self) { + while !self.maps.is_empty() { + self.unmap_at(0); + } + } +} + +struct Inner { + cache: Mutex, + requeue: Mutex>>, + stream: AtomicPtr, + pool: AtomicUsize, + inflight: AtomicUsize, + inflight_lock: Mutex<()>, + inflight_cv: Condvar, + releases: AtomicUsize, + requeues: AtomicUsize, +} + +struct AliasedBuffer { + frames: Arc, + pw_buffer: usize, + fd: i32, + /// mmap base, or 0 when the alias borrows a MemPtr this cache does not own. + map_base: usize, + end_sync: bool, +} + +/// Shared owner of cached DMA-BUF mappings and the in-flight alias count. +#[derive(Clone)] +pub(super) struct PipewireFrames { + inner: Arc, +} + +impl PipewireFrames { + pub(super) fn new() -> Self { + Self { + inner: Arc::new(Inner { + cache: Mutex::new(DmaBufCache { + maps: Vec::new(), + map_calls: 0, + unmap_calls: 0, + }), + requeue: Mutex::new(None), + stream: AtomicPtr::new(std::ptr::null_mut()), + pool: AtomicUsize::new(0), + inflight: AtomicUsize::new(0), + inflight_lock: Mutex::new(()), + inflight_cv: Condvar::new(), + releases: AtomicUsize::new(0), + requeues: AtomicUsize::new(0), + }), + } + } + + pub(super) fn set_requeue_sender(&self, sender: pw::channel::Sender) { + *self.inner.requeue.lock() = Some(sender); + } + + pub(super) fn set_stream(&self, stream: *mut pw::sys::pw_stream) { + self.inner.stream.store(stream, Ordering::Release); + } + + /// `pw_stream_queue_buffer` is single-producer. The encoder thread only + /// posts the buffer address; this runs on the PipeWire loop thread. + pub(super) fn queue_released_buffer(&self, buffer: usize) { + if buffer == 0 { + return; + } + let stream = self.inner.stream.load(Ordering::Acquire); + if stream.is_null() { + return; + } + unsafe { + pw::sys::pw_stream_queue_buffer(stream, buffer as *mut pw::sys::pw_buffer); + } + } + + pub(super) fn add_pool_buffer(&self) { + self.inner.pool.fetch_add(1, Ordering::AcqRel); + } + + pub(super) fn remove_pool_buffer(&self, buffer: *mut pw::sys::pw_buffer) { + let mut current = self.inner.pool.load(Ordering::Acquire); + while current > 0 { + match self.inner.pool.compare_exchange_weak( + current, + current - 1, + Ordering::AcqRel, + Ordering::Acquire, + ) { + Ok(_) => break, + Err(observed) => current = observed, + } + } + let fd = unsafe { first_data_fd(buffer) }; + self.inner.cache.lock().retire_fd(fd); + } + + pub(super) fn retire_all(&self) { + self.inner.cache.lock().retire_all(); + } + + pub(super) fn can_lease(&self) -> bool { + spare_buffer( + self.inner.pool.load(Ordering::Acquire), + self.inner.inflight.load(Ordering::Acquire), + ) + } + + pub(super) fn wait_idle(&self, timeout: Duration) -> bool { + let deadline = Instant::now() + timeout; + let mut guard = self.inner.inflight_lock.lock(); + loop { + if self.inner.inflight.load(Ordering::Acquire) == 0 { + return true; + } + let now = Instant::now(); + if now >= deadline { + return false; + } + self.inner + .inflight_cv + .wait_for(&mut guard, deadline.saturating_duration_since(now)); + } + } + + #[cfg(test)] + pub(super) fn inflight(&self) -> usize { + self.inner.inflight.load(Ordering::Acquire) + } + + #[cfg(test)] + fn unmap_idle(&self) { + self.inner.cache.lock().unmap_idle(); + } + + fn begin_inflight(&self) { + let _guard = self.inner.inflight_lock.lock(); + self.inner.inflight.fetch_add(1, Ordering::AcqRel); + } + + fn end_inflight(&self) { + let _guard = self.inner.inflight_lock.lock(); + let previous = self.inner.inflight.fetch_sub(1, Ordering::AcqRel); + debug_assert!(previous > 0); + if previous == 1 { + self.inner.inflight_cv.notify_all(); + } + } +} + +impl Inner { + fn complete(&self, aliased: &AliasedBuffer) { + if aliased.end_sync && aliased.fd >= 0 { + if let Err(error) = dma_buf_sync(aliased.fd, DMA_BUF_SYNC_READ | DMA_BUF_SYNC_END) { + tracing::warn!( + %error, + "ending the PipeWire screen capture DMA-BUF read failed" + ); + } + } + self.cache.lock().release(aliased.map_base); + self.releases.fetch_add(1, Ordering::Release); + if aliased.pw_buffer != 0 { + self.requeues.fetch_add(1, Ordering::Release); + if let Some(sender) = self.requeue.lock().as_ref() { + let _ = sender.send(aliased.pw_buffer); + } + } + } +} + +unsafe extern "C" fn release_aliased_pipewire_frame(opaque: *mut libc::c_void, _data: *mut u8) { + if opaque.is_null() { + return; + } + let aliased = unsafe { Box::from_raw(opaque.cast::()) }; + aliased.frames.complete(&aliased); + aliased.frames.end_borrow(); +} + +impl Inner { + fn end_borrow(&self) { + let _guard = self.inflight_lock.lock(); + let previous = self.inflight.fetch_sub(1, Ordering::AcqRel); + debug_assert!(previous > 0); + if previous == 1 { + self.inflight_cv.notify_all(); + } + } +} + +/// Read one linear plane. `lease` borrows the producer buffer when the crop +/// fits in the mapping; otherwise the returned frame owns a private copy and +/// the caller queues the PipeWire buffer immediately. +pub(super) fn linear_frame( + frames: &PipewireFrames, + source: &SourcePlane, + geometry: &LinearGeometry, + lease: bool, +) -> anyhow::Result<(ffmpeg::frame::Video, bool)> { + let LinearGeometry { + pixel_format: _, + bytes_per_pixel, + source_stride, + crop_x, + crop_y, + crop_width, + crop_height, + offset, + limit, + } = *geometry; + if source_stride == 0 || crop_width == 0 || crop_height == 0 || bytes_per_pixel == 0 { + bail!("PipeWire screen capture frame had an empty stride"); + } + let row_bytes = crop_width + .checked_mul(bytes_per_pixel) + .ok_or_else(|| anyhow!("PipeWire screen capture frame size overflowed"))?; + if source_stride < row_bytes { + bail!( + "PipeWire screen capture frame stride was too small: {source_stride} for {crop_width}" + ); + } + let frame_bytes = frame_byte_end( + offset, + source_stride, + crop_x, + crop_y, + crop_height, + bytes_per_pixel, + row_bytes, + )?; + + let (base, len, sync_fd) = if source.data.is_null() && source.fd < 0 { + bail!("PipeWire screen capture buffer was not memory-mapped"); + } else if source.data.is_null() { + let (base, len) = frames.inner.cache.lock().map(source.fd, frame_bytes)?; + let sync_fd = if source.sync_dma_buf { + frames + .inner + .cache + .lock() + .owned_fd(base) + .unwrap_or(source.fd) + } else { + -1 + }; + (base, len, sync_fd) + } else if source.data_len < frame_bytes { + bail!( + "PipeWire screen capture frame was too small: {} bytes for the crop", + source.data_len + ); + } else { + let sync_fd = if source.sync_dma_buf { source.fd } else { -1 }; + (source.data, source.data_len, sync_fd) + }; + let readable = limit.unwrap_or(len).min(len); + if frame_bytes > readable { + bail!("PipeWire screen capture frame was too small: {readable} bytes for the crop"); + } + + if sync_fd >= 0 { + dma_buf_sync(sync_fd, DMA_BUF_SYNC_READ) + .context("sync the PipeWire screen capture DMA-BUF for reading")?; + } + + let start = offset + crop_y * source_stride + crop_x * bytes_per_pixel; + if lease && alias_window_fits(start, source_stride, crop_height, readable) { + if let Some(frame) = alias_frame( + frames, + source, + geometry, + PlaneWindow { + base, + len, + start, + row_bytes, + sync_fd, + }, + )? { + return Ok((frame, true)); + } + } + + let copied = match copy_linear_frame( + unsafe { std::slice::from_raw_parts(base, len) }, + geometry, + readable, + row_bytes, + ) { + Ok(frame) => frame, + Err(error) => { + end_dma_read(sync_fd); + return Err(error); + } + }; + end_dma_read(sync_fd); + Ok((copied, false)) +} + +fn end_dma_read(sync_fd: RawFd) { + if sync_fd < 0 { + return; + } + if let Err(error) = dma_buf_sync(sync_fd, DMA_BUF_SYNC_READ | DMA_BUF_SYNC_END) { + tracing::warn!( + %error, + "ending the PipeWire screen capture DMA-BUF read failed" + ); + } +} + +struct PlaneWindow { + base: *mut u8, + len: usize, + start: usize, + row_bytes: usize, + sync_fd: RawFd, +} + +fn alias_frame( + frames: &PipewireFrames, + source: &SourcePlane, + geometry: &LinearGeometry, + plane: PlaneWindow, +) -> anyhow::Result> { + let Ok(linesize) = i32::try_from(geometry.source_stride) else { + return Ok(None); + }; + if linesize as usize != geometry.source_stride || geometry.source_stride < plane.row_bytes { + return Ok(None); + } + let map_base = if source.fd >= 0 { + plane.base as usize + } else { + 0 + }; + frames.begin_inflight(); + let aliased = Box::new(AliasedBuffer { + frames: Arc::clone(&frames.inner), + pw_buffer: source.pw_buffer, + fd: plane.sync_fd, + map_base, + end_sync: plane.sync_fd >= 0, + }); + let opaque = Box::into_raw(aliased); + let buffer = unsafe { + ffmpeg::sys::av_buffer_create( + plane.base, + plane.len, + Some(release_aliased_pipewire_frame), + opaque.cast(), + ffmpeg::sys::AV_BUFFER_FLAG_READONLY as libc::c_int, + ) + }; + if buffer.is_null() { + drop(unsafe { Box::from_raw(opaque) }); + frames.end_inflight(); + return Ok(None); + } + if map_base != 0 { + frames.inner.cache.lock().lease(plane.base); + } + let mut frame = ffmpeg::frame::Video::empty(); + frame.set_format(geometry.pixel_format); + frame.set_width(geometry.crop_width as u32); + frame.set_height(geometry.crop_height as u32); + unsafe { + let raw = frame.as_mut_ptr(); + (*raw).data[0] = plane.base.add(plane.start); + (*raw).linesize[0] = linesize; + (*raw).buf[0] = buffer; + } + Ok(Some(frame)) +} + +fn copy_linear_frame( + source: &[u8], + geometry: &LinearGeometry, + readable: usize, + row_bytes: usize, +) -> anyhow::Result { + let mut frame = ffmpeg::frame::Video::new( + geometry.pixel_format, + geometry.crop_width as u32, + geometry.crop_height as u32, + ); + let target_stride = frame.stride(0); + if target_stride < row_bytes { + bail!( + "PipeWire target frame stride was too small: {} for {}x{}", + target_stride, + geometry.crop_width, + geometry.crop_height + ); + } + for y in 0..geometry.crop_height { + let source_start = geometry.offset + + (geometry.crop_y + y) * geometry.source_stride + + geometry.crop_x * geometry.bytes_per_pixel; + let source_end = source_start + row_bytes; + if source_end > readable || source_end > source.len() { + bail!( + "PipeWire screen capture frame was too small: {} bytes for the crop", + source.len() + ); + } + let target_start = y * target_stride; + frame.data_mut(0)[target_start..target_start + row_bytes] + .copy_from_slice(&source[source_start..source_end]); + } + Ok(frame) +} + +fn frame_byte_end( + offset: usize, + stride: usize, + crop_x: usize, + crop_y: usize, + crop_height: usize, + bytes_per_pixel: usize, + row_bytes: usize, +) -> anyhow::Result { + let last_row = crop_y + .checked_add(crop_height - 1) + .ok_or_else(|| anyhow!("PipeWire screen capture frame size overflowed"))?; + offset + .checked_add( + last_row + .checked_mul(stride) + .ok_or_else(|| anyhow!("PipeWire screen capture frame size overflowed"))?, + ) + .and_then(|start| start.checked_add(crop_x * bytes_per_pixel)) + .and_then(|start| start.checked_add(row_bytes)) + .ok_or_else(|| anyhow!("PipeWire screen capture frame size overflowed")) +} + +fn alias_window_fits(start: usize, stride: usize, height: usize, limit: usize) -> bool { + stride + .checked_mul(height) + .and_then(|bytes| start.checked_add(bytes)) + .is_some_and(|end| end <= limit) +} + +fn mapped_len(fd: RawFd, minimum_len: usize) -> anyhow::Result { + if fd < 0 { + bail!("PipeWire screen capture DMA-BUF had no file descriptor"); + } + let len = file_len(fd) + .ok() + .filter(|len| *len > 0) + .unwrap_or(minimum_len); + if len < minimum_len { + bail!("PipeWire screen capture DMA-BUF holds {len} bytes, the frame needs {minimum_len}"); + } + Ok(len) +} + +fn file_len(fd: RawFd) -> std::io::Result { + let current = unsafe { libc::lseek(fd, 0, libc::SEEK_CUR) }; + let size = unsafe { libc::lseek(fd, 0, libc::SEEK_END) }; + if current >= 0 { + unsafe { + libc::lseek(fd, current, libc::SEEK_SET); + } + } + if size < 0 { + return Err(std::io::Error::last_os_error()); + } + usize::try_from(size).map_err(|_| { + std::io::Error::new(std::io::ErrorKind::InvalidData, "PipeWire DMA-BUF length") + }) +} + +fn dma_buf_sync(fd: RawFd, flags: u64) -> std::io::Result<()> { + let sync = flags; + loop { + let result = unsafe { libc::ioctl(fd, DMA_BUF_IOCTL_SYNC, &sync as *const u64) }; + if result == 0 { + return Ok(()); + } + let error = std::io::Error::last_os_error(); + match error.raw_os_error() { + Some(libc::EINTR | libc::EAGAIN) => continue, + Some(libc::ENOTTY) => return Ok(()), + _ => return Err(error), + } + } +} + +unsafe fn first_data_fd(buffer: *mut pw::sys::pw_buffer) -> i32 { + if buffer.is_null() { + return -1; + } + let spa = unsafe { (*buffer).buffer }; + if spa.is_null() || unsafe { (*spa).datas.is_null() || (*spa).n_datas == 0 } { + return -1; + } + let fd = unsafe { (*(*spa).datas).fd }; + i32::try_from(fd).unwrap_or(-1) +} + +#[cfg(test)] +mod tests { + use std::os::fd::AsRawFd; + + use super::*; + + fn geometry( + _width: usize, + _height: usize, + stride: usize, + crop_x: usize, + crop_y: usize, + crop_width: usize, + crop_height: usize, + ) -> LinearGeometry { + LinearGeometry { + pixel_format: ffmpeg::format::Pixel::BGRA, + bytes_per_pixel: 4, + source_stride: stride, + crop_x, + crop_y, + crop_width, + crop_height, + offset: 0, + limit: None, + } + } + + fn memptr_plane(bytes: &mut [u8]) -> SourcePlane { + SourcePlane { + data: bytes.as_mut_ptr(), + data_len: bytes.len(), + fd: -1, + sync_dma_buf: false, + pw_buffer: 0, + } + } + + #[test] + fn spare_pool_keeps_one_buffer_for_the_producer() { + assert!(!spare_buffer(0, 0)); + assert!(!spare_buffer(1, 0)); + assert!(spare_buffer(2, 0)); + assert!(!spare_buffer(2, 1)); + assert!(spare_buffer(8, 1)); + assert!(spare_buffer(8, 6)); + assert!(!spare_buffer(8, 7)); + } + + #[test] + fn memptr_alias_points_at_the_producer_until_the_last_reference_drops() { + ffmpeg::init().expect("ffmpeg"); + let mut bytes = vec![0u8; 8 * 4]; + for (index, pixel) in bytes.chunks_exact_mut(4).enumerate() { + pixel.copy_from_slice(&[index as u8, 2, 3, 4]); + } + let frames = PipewireFrames::new(); + let plane = memptr_plane(&mut bytes); + let (frame, leased) = + linear_frame(&frames, &plane, &geometry(2, 2, 8, 0, 0, 2, 2), true).unwrap(); + assert!(leased); + assert_eq!(frame.data(0).as_ptr(), bytes.as_ptr()); + assert_eq!(frames.inflight(), 1); + assert_eq!(frames.inner.releases.load(Ordering::Acquire), 0); + + let mut duplicate = ffmpeg::frame::Video::empty(); + let status = unsafe { ffmpeg::sys::av_frame_ref(duplicate.as_mut_ptr(), frame.as_ptr()) }; + assert!(status >= 0); + drop(frame); + assert_eq!(frames.inner.releases.load(Ordering::Acquire), 0); + assert_eq!(duplicate.data(0)[0], 0); + assert_eq!(frames.inflight(), 1); + drop(duplicate); + assert_eq!(frames.inner.releases.load(Ordering::Acquire), 1); + assert_eq!(frames.inflight(), 0); + } + + #[test] + fn readonly_alias_is_copied_before_it_can_be_written() { + ffmpeg::init().expect("ffmpeg"); + let mut bytes = vec![7u8, 8, 9, 10, 0, 0, 0, 0]; + let frames = PipewireFrames::new(); + let plane = memptr_plane(&mut bytes); + let (mut frame, leased) = + linear_frame(&frames, &plane, &geometry(1, 1, 8, 0, 0, 1, 1), true).unwrap(); + assert!(leased); + let source = frame.data(0).as_ptr(); + let status = unsafe { ffmpeg::sys::av_frame_make_writable(frame.as_mut_ptr()) }; + assert!(status >= 0, "make_writable status {status}"); + assert_ne!(frame.data(0).as_ptr(), source); + assert_eq!(&frame.data(0)[..4], &[7, 8, 9, 10]); + assert_eq!(frames.inner.releases.load(Ordering::Acquire), 1); + assert_eq!(frames.inflight(), 0); + bytes[0] = 99; + assert_eq!(frame.data(0)[0], 7); + } + + #[test] + fn dmabuf_mapping_is_reused_until_size_changes_and_not_unmapped_while_borrowed() { + ffmpeg::init().expect("ffmpeg"); + let (width, height, stride) = (4usize, 2usize, 16usize); + let mut bytes = vec![0u8; stride * height]; + for y in 0..height { + for x in 0..width { + bytes[y * stride + x * 4] = (y * 10 + x) as u8; + } + } + let mut file = tempfile::tempfile().unwrap(); + std::io::Write::write_all(&mut file, &bytes).unwrap(); + let fd = file.as_raw_fd(); + let frames = PipewireFrames::new(); + let plane = SourcePlane { + data: std::ptr::null_mut(), + data_len: 0, + fd, + sync_dma_buf: true, + pw_buffer: 0, + }; + let (first, leased) = linear_frame( + &frames, + &plane, + &geometry(width, height, stride, 0, 0, width, height), + true, + ) + .unwrap(); + assert!(leased); + let pointer = first.data(0).as_ptr(); + assert_eq!(frames.inner.cache.lock().base_for(fd), Some(pointer)); + assert_eq!(frames.inner.cache.lock().map_calls, 1); + assert_eq!(&first.data(0)[..4], &bytes[..4]); + + frames.unmap_idle(); + assert!(frames.inner.cache.lock().contains(pointer as usize)); + assert_eq!(first.data(0)[0], bytes[0]); + + drop(first); + let (second, _) = linear_frame( + &frames, + &plane, + &geometry(width, height, stride, 0, 0, width, height), + true, + ) + .unwrap(); + assert_eq!(second.data(0).as_ptr(), pointer); + assert_eq!(frames.inner.cache.lock().map_calls, 1); + + file.set_len((stride * height * 2) as u64).unwrap(); + let (grown, _) = linear_frame( + &frames, + &plane, + &geometry(width, height, stride, 0, 0, width, height), + true, + ) + .unwrap(); + assert_ne!(grown.data(0).as_ptr(), pointer); + assert_eq!(second.data(0)[0], bytes[0]); + assert!(frames.inner.cache.lock().contains(pointer as usize)); + drop(second); + assert!(!frames.inner.cache.lock().contains(pointer as usize)); + assert_eq!(grown.data(0)[0], bytes[0]); + drop(grown); + } + + #[test] + fn a_crop_that_does_not_fit_the_line_window_is_copied() { + ffmpeg::init().expect("ffmpeg"); + let mut bytes = vec![0u8; 16 * 2]; + bytes[4] = 42; + let frames = PipewireFrames::new(); + let plane = memptr_plane(&mut bytes); + let (frame, leased) = + linear_frame(&frames, &plane, &geometry(4, 2, 16, 1, 0, 1, 2), true).unwrap(); + assert!(!leased); + let copied = frame.data(0).as_ptr() as usize; + let origin = bytes.as_ptr() as usize; + assert!(copied < origin || copied >= origin + bytes.len()); + assert_eq!(frame.data(0)[0], 42); + assert_eq!(frames.inflight(), 0); + assert_eq!(frames.inner.requeues.load(Ordering::Acquire), 0); + } + + #[test] + fn a_borrowed_pipewire_buffer_is_returned_once_on_the_last_drop() { + ffmpeg::init().expect("ffmpeg"); + let mut bytes = vec![1u8, 2, 3, 4]; + let frames = PipewireFrames::new(); + frames.add_pool_buffer(); + frames.add_pool_buffer(); + frames.add_pool_buffer(); + let mut plane = memptr_plane(&mut bytes); + plane.pw_buffer = 0x51ce; + let (frame, leased) = + linear_frame(&frames, &plane, &geometry(1, 1, 4, 0, 0, 1, 1), true).unwrap(); + assert!(leased); + assert!(frames.can_lease()); + let mut duplicate = ffmpeg::frame::Video::empty(); + let status = unsafe { ffmpeg::sys::av_frame_ref(duplicate.as_mut_ptr(), frame.as_ptr()) }; + assert!(status >= 0); + drop(frame); + assert_eq!(frames.inner.requeues.load(Ordering::Acquire), 0); + drop(duplicate); + assert_eq!(frames.inner.requeues.load(Ordering::Acquire), 1); + assert_eq!(frames.inflight(), 0); + } + + #[test] + fn four_k_bgra_alias_avoids_the_row_copy_and_keeps_the_mapping() { + ffmpeg::init().expect("ffmpeg"); + const WIDTH: usize = 3840; + const HEIGHT: usize = 2160; + const STRIDE: usize = WIDTH * 4; + let mut bytes = vec![0u8; STRIDE * HEIGHT]; + bytes[0] = 11; + bytes[STRIDE * HEIGHT - 1] = 22; + let frames = PipewireFrames::new(); + let geometry = geometry(WIDTH, HEIGHT, STRIDE, 0, 0, WIDTH, HEIGHT); + + let copy_ms = median_ms(|| { + let plane = SourcePlane { + data: bytes.as_mut_ptr(), + data_len: bytes.len(), + fd: -1, + sync_dma_buf: false, + pw_buffer: 1, + }; + let (frame, leased) = linear_frame(&frames, &plane, &geometry, false).unwrap(); + assert!(!leased); + std::hint::black_box(frame.data(0)[0]); + std::hint::black_box(frame.data(0)[frame.data(0).len() - 1]); + drop(frame); + }); + let alias_ms = median_ms(|| { + let plane = memptr_plane(&mut bytes); + let (frame, leased) = linear_frame(&frames, &plane, &geometry, true).unwrap(); + assert!(leased); + assert_eq!(frame.data(0).as_ptr(), bytes.as_ptr()); + std::hint::black_box(frame.data(0)[0]); + drop(frame); + }); + + let file = tempfile::tempfile().unwrap(); + file.set_len((STRIDE * HEIGHT) as u64).unwrap(); + let fd = file.as_raw_fd(); + let map_ms = median_ms(|| { + let mapped = PipewireFrames::new(); + let plane = SourcePlane { + data: std::ptr::null_mut(), + data_len: 0, + fd, + sync_dma_buf: true, + pw_buffer: 0, + }; + let (frame, leased) = linear_frame(&mapped, &plane, &geometry, true).unwrap(); + assert!(leased); + std::hint::black_box(frame.data(0).as_ptr()); + drop(frame); + drop(mapped); + }); + let cached = PipewireFrames::new(); + let plane = SourcePlane { + data: std::ptr::null_mut(), + data_len: 0, + fd, + sync_dma_buf: true, + pw_buffer: 0, + }; + let (warm, _) = linear_frame(&cached, &plane, &geometry, true).unwrap(); + drop(warm); + let cached_ms = median_ms(|| { + let (frame, _) = linear_frame(&cached, &plane, &geometry, true).unwrap(); + std::hint::black_box(frame.data(0).as_ptr()); + drop(frame); + }); + assert_eq!(cached.inner.cache.lock().map_calls, 1); + + let profile = if cfg!(debug_assertions) { + "debug" + } else { + "release" + }; + let report = format!( + "profile {profile}\n4K BGRA {WIDTH}x{HEIGHT}\nrow-copy median_ms {copy_ms:.4}\nalias setup+release median_ms {alias_ms:.4}\ntempfile mmap+munmap median_ms {map_ms:.4}\ncached remap median_ms {cached_ms:.4}\nmap_calls 1\nnote tempfile is not a GPU dma-buf: DMA_BUF_IOCTL_SYNC returns ENOTTY and the mapping is not a scanned-out framebuffer, so this mmap time is not the ~1.1 ms GPU map/unmap.\n" + ); + eprintln!("{report}"); + let _ = std::fs::create_dir_all("/opt/cursor/artifacts"); + let _ = std::fs::write("/opt/cursor/artifacts/pipewire-alias-4k.log", &report); + assert!(copy_ms.is_finite() && alias_ms.is_finite()); + assert!(alias_ms < copy_ms); + } + + fn median_ms(mut measure: impl FnMut()) -> f64 { + for _ in 0..2 { + measure(); + } + let mut samples = Vec::with_capacity(9); + for _ in 0..9 { + let start = Instant::now(); + measure(); + samples.push(start.elapsed().as_secs_f64() * 1000.0); + } + samples.sort_by(|left, right| left.partial_cmp(right).unwrap()); + samples[samples.len() / 2] + } +} diff --git a/crates/recording/src/sources/screen_capture/mod.rs b/crates/recording/src/sources/screen_capture/mod.rs index 688cee38b62..ce26160b350 100644 --- a/crates/recording/src/sources/screen_capture/mod.rs +++ b/crates/recording/src/sources/screen_capture/mod.rs @@ -25,6 +25,8 @@ mod linux; #[cfg(target_os = "linux")] mod linux_dmabuf_import; #[cfg(target_os = "linux")] +mod linux_pipewire_frames; +#[cfg(target_os = "linux")] mod linux_shm; #[cfg(target_os = "linux")] pub use linux::*; From 8f1b9cf0d6f699997fe52e31b68bb804dc07d6ac Mon Sep 17 00:00:00 2001 From: Kyzer <36684609+kyzer1023@users.noreply.github.com> Date: Mon, 26 Oct 2026 05:47:24 +0000 Subject: [PATCH 2/3] fix(linux-pipewire): honor memfd mapoffset and unflake the 4K alias test MemFd chunk offsets are relative to spa_data.mapoffset. Map that window, include it in the cache key, and bound the readable tail to the bytes after it so borrowed and copied frames see the producer's pixels. The 4K timing comparison is an ignored benchmark. The unit test keeps the pointer and mapping-count checks. --- .../src/sources/screen_capture/linux.rs | 70 ++++++++ .../screen_capture/linux_pipewire_frames.rs | 161 ++++++++++++++++-- 2 files changed, 220 insertions(+), 11 deletions(-) diff --git a/crates/recording/src/sources/screen_capture/linux.rs b/crates/recording/src/sources/screen_capture/linux.rs index 4c9dae649a1..82d1d711f63 100644 --- a/crates/recording/src/sources/screen_capture/linux.rs +++ b/crates/recording/src/sources/screen_capture/linux.rs @@ -1013,11 +1013,16 @@ fn frame_from_pipewire_data( let fd = data.fd(); let dmabuf = data_type == spa::buffer::DataType::DmaBuf; let memfd = data_type == spa::buffer::DataType::MemFd; + // `chunk.offset` is relative to the producer's mapping, which starts at + // `spa_data.mapoffset`. The fd mapping, its bounds, and the cache key use + // that base; `offset` stays inside the mapping. + let map_offset = data.as_raw().mapoffset as usize; let plane = if (dmabuf || memfd) && fd >= 0 { SourcePlane { data: std::ptr::null_mut(), data_len: 0, fd, + map_offset, sync_dma_buf: dmabuf, pw_buffer, } @@ -1028,6 +1033,7 @@ fn frame_from_pipewire_data( data, data_len, fd: -1, + map_offset: 0, sync_dma_buf: false, pw_buffer, } @@ -3607,6 +3613,70 @@ mod pipewire_frame_tests { } } + /// A packed memfd places this buffer at `spa_data.mapoffset`. `chunk.offset` + /// is relative to that mapping, so both the borrowed frame and the copy + /// have to start there rather than at byte zero of the fd. + #[test] + fn memfd_map_offset_is_applied_to_borrowed_and_copied_frames() { + let page = + usize::try_from(unsafe { libc::sysconf(libc::_SC_PAGESIZE) }).expect("page size"); + let (width, height, stride) = (4usize, 2usize, 32usize); + let chunk_offset = 48usize; + let map_offset = page; + let mut bytes = vec![0x5Au8; map_offset + chunk_offset + stride * height]; + for y in 0..height { + for x in 0..width { + let pixel = &mut bytes[map_offset + chunk_offset + y * stride + x * 4..][..4]; + pixel.copy_from_slice(&[x as u8, y as u8, 0x80, 0xff]); + } + } + let file = tempfile::tempfile().unwrap(); + std::io::Write::write_all(&mut &file, &bytes).unwrap(); + + let mut chunk = spa::sys::spa_chunk { + offset: chunk_offset as u32, + size: (stride * height) as u32, + stride: stride as i32, + flags: 0, + }; + let mut raw: spa::sys::spa_data = unsafe { std::mem::zeroed() }; + raw.type_ = spa::sys::SPA_DATA_MemFd; + raw.flags = spa::sys::SPA_DATA_FLAG_READABLE; + raw.fd = i64::from(file.as_raw_fd()); + raw.mapoffset = map_offset as u32; + raw.maxsize = (bytes.len() - map_offset) as u32; + raw.data = std::ptr::null_mut(); + raw.chunk = &mut chunk; + let data = unsafe { &mut *(&mut raw as *mut spa::sys::spa_data as *mut spa::buffer::Data) }; + + let mut format = spa::param::video::VideoInfoRaw::new(); + format.set_format(spa::param::video::VideoFormat::BGRx); + format.set_size(spa::utils::Rectangle { + width: width as u32, + height: height as u32, + }); + let frames = PipewireFrames::new(); + for lease in [true, false] { + let (frame, leased) = frame_from_pipewire_data(data, format, None, &frames, 0, lease) + .unwrap() + .expect("a frame"); + assert_eq!(leased, lease); + assert_eq!( + (frame.width(), frame.height()), + (width as u32, height as u32) + ); + for y in 0..height { + let row = &frame.data(0)[y * frame.stride(0)..][..width * 4]; + let source = map_offset + chunk_offset + y * stride; + assert_eq!( + row, + &bytes[source..source + width * 4], + "lease {lease} row {y}" + ); + } + } + } + #[test] fn dmabuf_offer_follows_shared_memory_with_a_mandatory_linear_modifier() { let shm = pipewire_format_param(30).unwrap(); diff --git a/crates/recording/src/sources/screen_capture/linux_pipewire_frames.rs b/crates/recording/src/sources/screen_capture/linux_pipewire_frames.rs index 317b7e62b1b..d8a09f450f4 100644 --- a/crates/recording/src/sources/screen_capture/linux_pipewire_frames.rs +++ b/crates/recording/src/sources/screen_capture/linux_pipewire_frames.rs @@ -4,7 +4,7 @@ //! view stays alive until the last frame reference is dropped: the encoder //! reads it, and the muxer also holds the latest frame so a static tail can //! be repeated. DMA-BUF and memfd mappings are cached and replaced only when -//! the fd or the mapped size changes. +//! the fd, the map offset, or the mapped size changes. //! //! Two paths still allocate a private writable copy: //! - Instant mode with a camera. The alias is created with @@ -47,6 +47,10 @@ pub(super) struct SourcePlane { pub data_len: usize, /// Negative when the plane has no fd of its own. pub fd: i32, + /// `spa_data.mapoffset` for an fd plane. Chunk offsets are relative to + /// this byte of the file, which is where the producer's mapping starts. + /// Zero when [`Self::data`] is already that mapping. + pub map_offset: usize, pub sync_dma_buf: bool, /// `pw_buffer` address, queued back when the last frame reference drops. /// Zero in tests that have no PipeWire buffer. @@ -70,6 +74,8 @@ pub(super) struct LinearGeometry { struct MappedBuffer { /// Producer fd. PipeWire may close it while an alias is still borrowed. fd: RawFd, + /// `spa_data.mapoffset` this mapping was established at. + map_offset: usize, /// Dup of `fd`, used for mmap and the read fence so that close cannot /// recycle the descriptor under the encoder. owned: RawFd, @@ -86,9 +92,18 @@ struct DmaBufCache { } impl DmaBufCache { - fn map(&mut self, fd: RawFd, minimum_len: usize) -> anyhow::Result<(*mut u8, usize)> { - let len = mapped_len(fd, minimum_len)?; - if let Some(index) = self.maps.iter().position(|map| map.fd == fd && !map.retire) { + fn map( + &mut self, + fd: RawFd, + map_offset: usize, + minimum_len: usize, + ) -> anyhow::Result<(*mut u8, usize)> { + let len = mapped_len(fd, map_offset, minimum_len)?; + if let Some(index) = self + .maps + .iter() + .position(|map| map.fd == fd && map.map_offset == map_offset && !map.retire) + { if self.maps[index].len == len { return Ok((self.maps[index].pointer as *mut u8, len)); } @@ -100,6 +115,9 @@ impl DmaBufCache { return Err(std::io::Error::last_os_error()) .context("dup the PipeWire screen capture DMA-BUF"); } + let offset = libc::off_t::try_from(map_offset).map_err(|_| { + anyhow!("PipeWire screen capture map offset does not fit in a file offset") + })?; let pointer = unsafe { libc::mmap( std::ptr::null_mut(), @@ -107,7 +125,7 @@ impl DmaBufCache { libc::PROT_READ, libc::MAP_SHARED, owned, - 0, + offset, ) }; if pointer == libc::MAP_FAILED { @@ -121,6 +139,7 @@ impl DmaBufCache { self.map_calls += 1; self.maps.push(MappedBuffer { fd, + map_offset, owned, len, pointer, @@ -461,7 +480,12 @@ pub(super) fn linear_frame( let (base, len, sync_fd) = if source.data.is_null() && source.fd < 0 { bail!("PipeWire screen capture buffer was not memory-mapped"); } else if source.data.is_null() { - let (base, len) = frames.inner.cache.lock().map(source.fd, frame_bytes)?; + let (base, len) = + frames + .inner + .cache + .lock() + .map(source.fd, source.map_offset, frame_bytes)?; let sync_fd = if source.sync_dma_buf { frames .inner @@ -670,14 +694,16 @@ fn alias_window_fits(start: usize, stride: usize, height: usize, limit: usize) - .is_some_and(|end| end <= limit) } -fn mapped_len(fd: RawFd, minimum_len: usize) -> anyhow::Result { +fn mapped_len(fd: RawFd, map_offset: usize, minimum_len: usize) -> anyhow::Result { if fd < 0 { bail!("PipeWire screen capture DMA-BUF had no file descriptor"); } - let len = file_len(fd) - .ok() - .filter(|len| *len > 0) - .unwrap_or(minimum_len); + let len = match file_len(fd) { + Ok(len) if len > 0 => len.checked_sub(map_offset).ok_or_else(|| { + anyhow!("PipeWire screen capture map offset {map_offset} is past the end of the buffer") + })?, + _ => minimum_len, + }; if len < minimum_len { bail!("PipeWire screen capture DMA-BUF holds {len} bytes, the frame needs {minimum_len}"); } @@ -761,6 +787,7 @@ mod tests { data: bytes.as_mut_ptr(), data_len: bytes.len(), fd: -1, + map_offset: 0, sync_dma_buf: false, pw_buffer: 0, } @@ -843,6 +870,7 @@ mod tests { data: std::ptr::null_mut(), data_len: 0, fd, + map_offset: 0, sync_dma_buf: true, pw_buffer: 0, }; @@ -933,6 +961,53 @@ mod tests { assert_eq!(frames.inflight(), 0); } + #[test] + fn memfd_map_offset_selects_the_producer_region_and_cache_entry() { + ffmpeg::init().expect("ffmpeg"); + let page = page_size(); + let (width, height, stride) = (4usize, 2usize, 16usize); + let chunk_offset = 16usize; + let mut bytes = vec![0u8; page * 2]; + bytes[..page].fill(0xA1); + bytes[page..].fill(0xB2); + bytes[chunk_offset] = 0x10; + bytes[page + chunk_offset] = 0x20; + let mut file = tempfile::tempfile().unwrap(); + std::io::Write::write_all(&mut file, &bytes).unwrap(); + let fd = file.as_raw_fd(); + let frames = PipewireFrames::new(); + let mut geometry = geometry(width, height, stride, 0, 0, width, height); + geometry.offset = chunk_offset; + let plane = |map_offset| SourcePlane { + data: std::ptr::null_mut(), + data_len: 0, + fd, + map_offset, + sync_dma_buf: false, + pw_buffer: 0, + }; + + let (borrowed, leased) = linear_frame(&frames, &plane(page), &geometry, true).unwrap(); + assert!(leased); + assert_eq!(borrowed.data(0)[0], 0x20); + let (copied, leased) = linear_frame(&frames, &plane(page), &geometry, false).unwrap(); + assert!(!leased); + assert_eq!(copied.data(0)[0], 0x20); + assert_ne!(copied.data(0).as_ptr(), borrowed.data(0).as_ptr()); + + let (region_a, _) = linear_frame(&frames, &plane(0), &geometry, true).unwrap(); + assert_eq!(region_a.data(0)[0], 0x10); + assert_eq!(borrowed.data(0)[0], 0x20); + assert_eq!(frames.inner.cache.lock().map_calls, 2); + + drop(borrowed); + drop(copied); + drop(region_a); + let (again, _) = linear_frame(&frames, &plane(page), &geometry, true).unwrap(); + assert_eq!(again.data(0)[0], 0x20); + assert_eq!(frames.inner.cache.lock().map_calls, 2); + } + #[test] fn four_k_bgra_alias_avoids_the_row_copy_and_keeps_the_mapping() { ffmpeg::init().expect("ffmpeg"); @@ -945,11 +1020,68 @@ mod tests { let frames = PipewireFrames::new(); let geometry = geometry(WIDTH, HEIGHT, STRIDE, 0, 0, WIDTH, HEIGHT); + let plane = memptr_plane(&mut bytes); + let (frame, leased) = linear_frame(&frames, &plane, &geometry, true).unwrap(); + assert!(leased); + assert_eq!(frame.data(0).as_ptr(), bytes.as_ptr()); + assert_eq!(frame.data(0)[0], 11); + assert_eq!(frame.data(0)[frame.data(0).len() - 1], 22); + drop(frame); + + let copied_plane = SourcePlane { + data: bytes.as_mut_ptr(), + data_len: bytes.len(), + fd: -1, + map_offset: 0, + sync_dma_buf: false, + pw_buffer: 1, + }; + let (copied, leased) = linear_frame(&frames, &copied_plane, &geometry, false).unwrap(); + assert!(!leased); + assert_ne!(copied.data(0).as_ptr(), bytes.as_ptr()); + assert_eq!(copied.data(0)[0], 11); + assert_eq!(copied.data(0)[copied.data(0).len() - 1], 22); + + let file = tempfile::tempfile().unwrap(); + file.set_len((STRIDE * HEIGHT) as u64).unwrap(); + let fd = file.as_raw_fd(); + let cached = PipewireFrames::new(); + let mapped = SourcePlane { + data: std::ptr::null_mut(), + data_len: 0, + fd, + map_offset: 0, + sync_dma_buf: true, + pw_buffer: 0, + }; + let (first, leased) = linear_frame(&cached, &mapped, &geometry, true).unwrap(); + assert!(leased); + let pointer = first.data(0).as_ptr(); + drop(first); + let (second, _) = linear_frame(&cached, &mapped, &geometry, true).unwrap(); + assert_eq!(second.data(0).as_ptr(), pointer); + assert_eq!(cached.inner.cache.lock().map_calls, 1); + } + + #[test] + #[ignore = "scheduler-sensitive timing; pointer and map-count checks stay in four_k_bgra_alias_avoids_the_row_copy_and_keeps_the_mapping"] + fn four_k_bgra_alias_is_faster_than_the_row_copy() { + ffmpeg::init().expect("ffmpeg"); + const WIDTH: usize = 3840; + const HEIGHT: usize = 2160; + const STRIDE: usize = WIDTH * 4; + let mut bytes = vec![0u8; STRIDE * HEIGHT]; + bytes[0] = 11; + bytes[STRIDE * HEIGHT - 1] = 22; + let frames = PipewireFrames::new(); + let geometry = geometry(WIDTH, HEIGHT, STRIDE, 0, 0, WIDTH, HEIGHT); + let copy_ms = median_ms(|| { let plane = SourcePlane { data: bytes.as_mut_ptr(), data_len: bytes.len(), fd: -1, + map_offset: 0, sync_dma_buf: false, pw_buffer: 1, }; @@ -977,6 +1109,7 @@ mod tests { data: std::ptr::null_mut(), data_len: 0, fd, + map_offset: 0, sync_dma_buf: true, pw_buffer: 0, }; @@ -991,6 +1124,7 @@ mod tests { data: std::ptr::null_mut(), data_len: 0, fd, + map_offset: 0, sync_dma_buf: true, pw_buffer: 0, }; @@ -1018,6 +1152,11 @@ mod tests { assert!(alias_ms < copy_ms); } + fn page_size() -> usize { + let page = unsafe { libc::sysconf(libc::_SC_PAGESIZE) }; + usize::try_from(page).expect("page size") + } + fn median_ms(mut measure: impl FnMut()) -> f64 { for _ in 0..2 { measure(); From d68cfa6cf192a66205a90cd0c12828d6fb2c142a Mon Sep 17 00:00:00 2001 From: Kyzer <36684609+kyzer1023@users.noreply.github.com> Date: Mon, 26 Oct 2026 07:30:27 +0000 Subject: [PATCH 3/3] fix(linux-pipewire): release niri DMA-BUF aliases before shutdown The GNOME reliability test kept captured frames until after the capture thread joined. Aliased DMA-BUFs hold the producer buffer until that last reference drops, so the thread hit its idle timeout and reported that capture stopped while a frame was still borrowed. Copy the pixels and drop the frames before joining. --- .../src/sources/screen_capture/linux.rs | 24 +++++++++++++++---- 1 file changed, 19 insertions(+), 5 deletions(-) diff --git a/crates/recording/src/sources/screen_capture/linux.rs b/crates/recording/src/sources/screen_capture/linux.rs index 82d1d711f63..ba1ba29cfcf 100644 --- a/crates/recording/src/sources/screen_capture/linux.rs +++ b/crates/recording/src/sources/screen_capture/linux.rs @@ -3540,19 +3540,33 @@ mod pipewire_frame_tests { Err(_) => std::thread::park_timeout(std::time::Duration::from_millis(10)), } } + // Borrowed DMA-BUF frames keep the producer buffer until the last + // reference drops. The capture thread waits for that before it + // destroys the stream, so copy the pixels and release the frames + // before joining it. + let mut samples = Vec::with_capacity(frames.len()); + for frame in &frames { + let mut pixel = [0u8; 4]; + let data = frame.data(0); + if data.len() >= 4 { + pixel.copy_from_slice(&data[..4]); + } + samples.push((frame.width(), frame.height(), pixel)); + } + drop(frames); + drop(video_rx); stop.cancel(); let result = capture.join().unwrap(); drop(child); assert!( - frames.len() >= 10, + samples.len() >= 10, "captured {} frames; capture ended with {result:?}", - frames.len() + samples.len() ); result.unwrap(); let mut levels = std::collections::BTreeSet::new(); - for frame in &frames { - assert_eq!((frame.width(), frame.height()), (640, 360)); - let pixel = &frame.data(0)[..4]; + for (width, height, pixel) in &samples { + assert_eq!((*width, *height), (640, 360)); assert_eq!(&pixel[1..3], &[0x40, 0x80], "{pixel:?}"); levels.insert(pixel[0]); }