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..ba1ba29cfcf 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,61 @@ 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; + // `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, + } + } else if let Some(mapped) = data.data() { + let data_len = mapped.len(); + let data = mapped.as_mut_ptr(); + SourcePlane { + data, + data_len, + fd: -1, + map_offset: 0, + 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 +3454,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; @@ -3507,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]); } @@ -3567,9 +3614,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 { @@ -3578,6 +3627,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 new file mode 100644 index 00000000000..d8a09f450f4 --- /dev/null +++ b/crates/recording/src/sources/screen_capture/linux_pipewire_frames.rs @@ -0,0 +1,1173 @@ +//! 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, 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 +//! `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, + /// `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. + 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, + /// `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, + 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, + 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)); + } + 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 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(), + len, + libc::PROT_READ, + libc::MAP_SHARED, + owned, + offset, + ) + }; + 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, + map_offset, + 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, source.map_offset, 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, map_offset: usize, minimum_len: usize) -> anyhow::Result { + if fd < 0 { + bail!("PipeWire screen capture DMA-BUF had no file descriptor"); + } + 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}"); + } + 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, + map_offset: 0, + 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, + map_offset: 0, + 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 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"); + 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 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, + }; + 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, + map_offset: 0, + 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, + map_offset: 0, + 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 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(); + } + 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::*;