aria/feresaria
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf(capture): add SceneLens GPU streams and retained scene reuse

Browse Source 
Use persistent privacy-filtered capture targets and client-owned GPU buffers with fence completion, backpressure, negotiated frame rates and shared-memory fallback.

Invalidate output views on scene changes while preserving inactive-window callbacks and capture revocation. Include OBS memory-mask negotiation and native pixel, transport and recording regressions.
This commit is contained in:
Samuel Onoja committed 2026-10-24 19:58:54 +01:00
1 parent f49f62583d
commit 533cd74c94
40 files changed
+4704 -704
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

portal/Cargo.toml
+3 -1
View File
@@ -40,6 +40,8 @@ serde.workspace = true
serde_json.workspace = true
tokio = { version = "1", features = ["rt-multi-thread", "macros", "time", "sync", "process", "io-util", "net"] }
wayland-client.workspace = true
wayland-protocols = { version = "0.32", features = ["client", "unstable"] }
wayland-protocols = { version = "0.32", features = ["client", "unstable", "staging"] }
wayland-protocols-wlr.workspace = true
zbus = { version = "5.19.0", default-features = false, features = ["tokio"] }
gbm = { version = "0.18", default-features = false }
portal/src/capture.rs
+8 -258
View File
@@ -1,19 +1,13 @@
use std::fs::File;
use std::os::fd::{AsFd, AsRawFd, FromRawFd};
use std::os::fd::AsRawFd;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::{Duration, Instant};
use ferese_protocols::window_capture::v1::client::ferese_window_capture_manager_v1 as window_manager;
use memmap2::MmapMut;
use serde::{Deserialize, Serialize};
use wayland_client::protocol::{wl_buffer, wl_output, wl_registry, wl_shm, wl_shm_pool};
use wayland_client::protocol::{wl_output, wl_registry, wl_shm};
use wayland_client::{Connection, Dispatch, EventQueue, QueueHandle, WEnum};
use wayland_protocols_wlr::screencopy::v1::client::{
zwlr_screencopy_frame_v1 as frame, zwlr_screencopy_manager_v1 as manager,
};
pub const BUSY: &str = "Window capture snapshot pool is busy";
use wayland_protocols_wlr::screencopy::v1::client::zwlr_screencopy_manager_v1 as manager;
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct Source {
@@ -34,53 +28,23 @@ impl Source {
struct Output {
global: u32,
proxy: wl_output::WlOutput,
_proxy: wl_output::WlOutput,
source: Source,
transform: wl_output::Transform,
}
pub struct Frame {
pub logical_size: Option<(u32, u32)>,
pub width: u32,
pub height: u32,
pub stride: u32,
pub pixels: Vec<u8>,
}
struct Buffer {
_file: File,
map: MmapMut,
proxy: wl_buffer::WlBuffer,
width: u32,
height: u32,
stride: u32,
format: wl_shm::Format,
}
impl Drop for Buffer {
fn drop(&mut self) {
self.proxy.destroy();
}
}
#[derive(Default)]
struct State {
outputs: Vec<Output>,
manager: Option<manager::ZwlrScreencopyManagerV1>,
window_manager: Option<window_manager::FereseWindowCaptureManagerV1>,
shm: Option<wl_shm::WlShm>,
buffer: Option<Buffer>,
result: Option<Result<(), String>>,
flipped: bool,
logical_size: Option<(u32, u32)>,
busy: bool,
}
pub struct Capture {
connection: Connection,
queue: EventQueue<State>,
state: State,
selected_output: Option<u32>,
}
impl Capture {
@@ -96,7 +60,6 @@ impl Capture {
connection,
queue,
state: State::default(),
selected_output: None,
};
// Two syncs deliver registry bindings and output metadata without an
// uninterruptible roundtrip on a stalled compositor.
@@ -134,14 +97,6 @@ impl Capture {
.map(|output| output.global)
}
pub fn pin_output(&mut self, name: &str, generation: u32) -> Result<(), String> {
if self.generation(name) != Some(generation) {
return Err("Shared monitor was replaced".into());
}
self.selected_output = Some(generation);
Ok(())
}
pub fn validate_sources(
&mut self,
selected: &[(String, u32)],
@@ -170,99 +125,6 @@ impl Capture {
.collect()
}
pub fn frame(
&mut self,
name: &str,
cursor: bool,
stop: &AtomicBool,
mut pixels: Vec<u8>,
) -> Result<Frame, String> {
self.state.result = None;
self.state.flipped = false;
self.state.logical_size = None;
self.state.busy = false;
let (frame, transform) = if let Some(id) = name
.strip_prefix("window:")
.and_then(|id| id.parse::<u64>().ok())
{
let manager = self
.state
.window_manager
.as_ref()
.ok_or("Window capture is unavailable")?;
(
manager.capture_window(
(id >> 32) as u32,
id as u32,
cursor as u32,
&self.queue.handle(),
(),
),
wl_output::Transform::Normal,
)
} else {
let output = self
.state
.outputs
.iter()
.find(|o| o.source.name == name)
.ok_or("Shared monitor disconnected")?;
if self
.selected_output
.is_some_and(|selected| selected != output.global)
{
return Err("Shared monitor was replaced".into());
}
self.selected_output = Some(output.global);
let manager = self
.state
.manager
.as_ref()
.ok_or("Monitor capture is unavailable")?;
(
manager.capture_output(cursor as i32, &output.proxy, &self.queue.handle(), ()),
output.transform,
)
};
let deadline = Instant::now() + Duration::from_secs(5);
while self.state.result.is_none() {
if let Err(error) = self.dispatch(stop, deadline) {
frame.destroy();
return Err(error);
}
}
frame.destroy();
self.state
.result
.take()
.expect("the capture dispatch loop exits after storing its result")?;
let buffer = self
.state
.buffer
.as_ref()
.ok_or("Capture supplied no buffer")?;
let (width, height) = copy_oriented_pixels(
&buffer.map,
buffer.width,
buffer.height,
buffer.stride,
transform,
self.state.flipped,
&mut pixels,
);
Ok(Frame {
logical_size: self.state.logical_size,
width,
height,
stride: width * 4,
pixels,
})
}
fn dispatch(&mut self, stop: &AtomicBool, deadline: Instant) -> Result<(), String> {
if stop.load(Ordering::Relaxed) {
return Err("Capture cancelled".into());
@@ -321,7 +183,7 @@ impl Dispatch<wl_registry::WlRegistry, ()> for State {
"wl_output" => state.outputs.push(Output {
global: name,
transform: wl_output::Transform::Normal,
proxy: registry.bind(name, version.min(4), qh, name),
_proxy: registry.bind(name, version.min(4), qh, name),
source: Source {
name: format!("output-{name}"),
label: "Display".into(),
@@ -387,100 +249,6 @@ impl Dispatch<wl_output::WlOutput, u32> for State {
}
}
impl Dispatch<frame::ZwlrScreencopyFrameV1, ()> for State {
fn event(
state: &mut Self,
frame: &frame::ZwlrScreencopyFrameV1,
event: frame::Event,
_: &(),
_: &Connection,
qh: &QueueHandle<Self>,
) {
match event {
frame::Event::Buffer {
format: WEnum::Value(format),
width,
height,
stride,
} => {
let result = (|| {
if format != wl_shm::Format::Argb8888 && format != wl_shm::Format::Xrgb8888 {
return Err("Unsupported capture pixel format".into());
}
let bytes = buffer_size(width, height, stride)?;
if let Some(buffer) = &state.buffer
&& (buffer.width, buffer.height, buffer.stride, buffer.format)
== (width, height, stride, format)
{
frame.copy(&buffer.proxy);
return Ok(());
}
// SAFETY: fixed NUL-terminated name and a newly owned memfd.
let fd = unsafe {
libc::memfd_create(c"ferese-screencast".as_ptr(), libc::MFD_CLOEXEC)
};
if fd < 0 {
return Err(std::io::Error::last_os_error().to_string());
}
// SAFETY: memfd_create succeeded above and returned a fresh descriptor whose
// ownership transfers to File.
let file = unsafe { File::from_raw_fd(fd) };
file.set_len(bytes as u64).map_err(|e| e.to_string())?;
// SAFETY: the file has the checked mapping length and lives with the mapping.
let map = unsafe { MmapMut::map_mut(&file) }.map_err(|e| e.to_string())?;
let pool = state
.shm
.as_ref()
.ok_or("Missing shared memory global")?
.create_pool(file.as_fd(), bytes as i32, qh, ());
let proxy = pool.create_buffer(
0,
width as i32,
height as i32,
stride as i32,
format,
qh,
(),
);
pool.destroy();
frame.copy(&proxy);
state.buffer = Some(Buffer {
_file: file,
map,
proxy,
width,
height,
stride,
format,
});
Ok::<(), String>(())
})();
if let Err(e) = result {
state.result = Some(Err(e));
}
}
frame::Event::Ready { .. } => state.result = Some(Ok(())),
frame::Event::Failed => {
state.result = Some(Err(if state.busy {
BUSY.into()
} else {
"Capture ended: session locked, source closed, or compositor unavailable".into()
}))
}
frame::Event::Flags {
flags: WEnum::Value(flags),
} => state.flipped = flags.contains(frame::Flags::YInvert),
_ => (),
}
}
}
fn swaps_axes(transform: wl_output::Transform) -> bool {
matches!(
transform,
@@ -494,6 +262,7 @@ fn swaps_axes(transform: wl_output::Transform) -> bool {
// Screencopy supplies output-buffer coordinates. Normalize both the output's
// presentation transform and the independent YInvert storage flag before
// publishing upright, tightly packed BGRx frames to PipeWire.
#[cfg(test)]
fn copy_oriented_pixels(
source: &[u8],
width: u32,
@@ -545,6 +314,7 @@ fn copy_oriented_pixels(
(out_width, out_height)
}
#[cfg(test)]
fn buffer_size(width: u32, height: u32, stride: u32) -> Result<usize, String> {
let row = width.checked_mul(4).ok_or("Capture dimensions overflow")?;
let bytes = stride
@@ -572,27 +342,8 @@ impl Dispatch<wayland_client::protocol::wl_callback::WlCallback, Arc<AtomicBool>
}
wayland_client::delegate_noop!(State: ignore wl_shm::WlShm);
wayland_client::delegate_noop!(State: ignore wl_shm_pool::WlShmPool);
wayland_client::delegate_noop!(State: ignore wl_buffer::WlBuffer);
wayland_client::delegate_noop!(State: ignore manager::ZwlrScreencopyManagerV1);
impl Dispatch<window_manager::FereseWindowCaptureManagerV1, ()> for State {
fn event(
state: &mut Self,
_: &window_manager::FereseWindowCaptureManagerV1,
event: window_manager::Event,
_: &(),
_: &Connection,
_: &QueueHandle<Self>,
) {
match event {
window_manager::Event::Geometry { width, height, .. } => {
state.logical_size = Some((width, height))
}
window_manager::Event::Busy { .. } => state.busy = true,
_ => (),
}
}
}
wayland_client::delegate_noop!(State: ignore window_manager::FereseWindowCaptureManagerV1);
#[cfg(test)]
mod tests {
@@ -622,7 +373,6 @@ mod tests {
connection,
queue,
state: State::default(),
selected_output: None,
};
capture.validate_sources(&[], &flag).unwrap_err()
}
portal/src/permissions.rs
+39 -26
View File
@@ -3,6 +3,7 @@ use std::path::Path;
use std::sync::Arc;
use std::time::Duration;
use ferese_ipc::events::Window;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::UnixStream;
use tokio::sync::RwLock;
@@ -197,7 +198,7 @@ impl Background {
// Tokio owns the socket so cancelling the watch task also closes an idle stream.
struct BackgroundEvents {
stream: UnixStream,
windows: serde_json::Value,
windows: Vec<Window>,
changed: bool,
}
@@ -238,7 +239,7 @@ impl BackgroundEvents {
.map_err(|_| "Event subscription timed out".to_owned())??;
Ok(Self {
stream,
windows: serde_json::json!([]),
windows: Vec::new(),
changed: false,
})
}
@@ -257,10 +258,9 @@ impl BackgroundEvents {
self.changed = true;
}
Change::FocusChanged { focus } => {
for window in self.windows.as_array_mut().into_iter().flatten() {
window["focused"] = serde_json::json!(
focus["window"].is_number() && window["id"] == focus["window"]
);
let focused = focus["window"].as_u64();
for window in &mut self.windows {
window.focused = focused == Some(window.id);
}
self.changed = true;
}
@@ -364,14 +364,15 @@ async fn background_consent(app: &str, name: &str) -> Result<Option<Options>, St
)])))
}
fn app_states(windows: &serde_json::Value) -> HashMap<String, u32> {
fn app_states(windows: &[Window]) -> HashMap<String, u32> {
let mut apps: HashMap<String, u32> = HashMap::new();
for window in windows.as_array().into_iter().flatten() {
let Some(app) = window["app_id"].as_str().filter(|app| !app.is_empty()) else {
for window in windows {
let app = window.app_id.as_str();
if app.is_empty() {
continue;
};
}
let app = app.strip_suffix(".desktop").unwrap_or(app).to_owned();
let state = if window["focused"] == true { 2 } else { 1 };
let state = if window.focused { 2 } else { 1 };
apps.entry(app)
.and_modify(|old| *old = (*old).max(state))
.or_insert(state);
@@ -480,6 +481,18 @@ fn desktop_argument(argument: &str) -> Result<String, String> {
mod tests {
use super::*;
fn window(id: u64, app: &str, focused: bool, workspace: u64) -> Window {
Window {
id,
app_id: app.into(),
title: format!("Window {id}"),
focused,
workspace,
floating: false,
fullscreen: false,
}
}
async fn send_frame(peer: &mut UnixStream, value: &impl serde::Serialize) {
let mut bytes = Vec::new();
ferese_ipc::write_frame(&mut bytes, value).unwrap();
@@ -516,10 +529,10 @@ mod tests {
desktop: Box::new(ferese_ipc::events::Snapshot {
outputs: serde_json::json!([]),
workspaces: serde_json::json!([]),
windows: serde_json::json!([{"id":1,"app_id":"one.desktop","focused":true}]),
windows: vec![window(1, "one.desktop", true, 1)],
focus: serde_json::json!({"window":1}),
config: Default::default(),
theme: serde_json::Value::Null,
theme: serde_json::Value::Null.into(),
lock: ferese_ipc::events::LockState {
phase: "unlocked".into(),
sleeping: false,
@@ -533,10 +546,10 @@ mod tests {
events.next().await.unwrap().unwrap(),
HashMap::from([("one".into(), 2)])
);
let windows = serde_json::json!([
{"id":1,"app_id":"one.desktop","focused":true},
{"id":2,"app_id":"two","focused":false}
]);
let windows = vec![
window(1, "one.desktop", true, 1),
window(2, "two", false, 2),
];
send_frame(
&mut peer,
&Event {
@@ -577,7 +590,7 @@ mod tests {
generation: 2,
last: true,
change: Change::WindowsChanged {
windows: serde_json::json!([]),
windows: Vec::new(),
},
},
)
@@ -595,7 +608,7 @@ mod tests {
let (stream, mut peer) = UnixStream::pair().unwrap();
let mut events = BackgroundEvents {
stream,
windows: serde_json::json!([]),
windows: Vec::new(),
changed: false,
};
peer.write_u32((ferese_ipc::MAX_PAYLOAD_SIZE + 1) as u32)
@@ -608,17 +621,17 @@ mod tests {
#[test]
fn application_state_groups_windows_and_keeps_inactive_workspaces_running() {
let windows = serde_json::json!([
{"app_id":"org.test.App.desktop","focused":false,"mapped":false},
{"app_id":"org.test.App","focused":true,"mapped":true},
{"app_id":"org.test.Other","focused":false,"mapped":true},
{"app_id":"","focused":false}
]);
let windows = vec![
window(1, "org.test.App.desktop", false, 2),
window(2, "org.test.App", true, 1),
window(3, "org.test.Other", false, 3),
window(4, "", false, 1),
];
assert_eq!(
app_states(&windows),
HashMap::from([("org.test.App".into(), 2), ("org.test.Other".into(), 1)])
);
assert!(app_states(&serde_json::json!([])).is_empty());
assert!(app_states(&[]).is_empty());
}
#[test]
portal/src/recorder.rs
+168 -9
View File
@@ -4,8 +4,8 @@ use std::io::{Read, Write};
use std::os::fd::AsRawFd;
use std::os::unix::fs::OpenOptionsExt;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use crate::recording::{Audio, Recording};
@@ -515,20 +515,29 @@ fn encode(
.is_some_and(|event| matches!(event.view(), gst::EventView::Eos(_)))
{
ended.store(true, Ordering::Relaxed);
// Finish through the encoder queue after appending the last
// image's timestamp, even when the capture source vanished.
return gst::PadProbeReturn::Drop;
}
gst::PadProbeReturn::Ok
});
let announced = Arc::new(AtomicBool::new(false));
let notice = announced.clone();
let last_frame = Arc::new(Mutex::new(None));
let observed_frame = last_frame.clone();
source
.static_pad("src")
.ok_or("Capture has no video output")?
.add_probe(gst::PadProbeType::BUFFER, move |_, info| {
if info.buffer().is_some_and(|b| b.size() > 0)
&& !notice.swap(true, Ordering::Relaxed)
{
event("recording", None, None);
.add_probe(gst::PadProbeType::BUFFER, move |pad, info| {
if let Some(buffer) = info.buffer().filter(|buffer| buffer.size() > 0) {
*observed_frame
.lock()
.expect("last recording frame mutex poisoned") =
Some((buffer.to_owned(), pad.current_caps()));
if !notice.swap(true, Ordering::Relaxed) {
event("recording", None, None);
}
}
gst::PadProbeReturn::Ok
});
@@ -548,10 +557,35 @@ fn encode(
finishing = Some(Instant::now());
// Inject after the capture source: it may already have disappeared
// when the portal closes on lock or monitor removal.
let _ = queue
let input = queue
.static_pad("sink")
.ok_or("Encoder queue has no input")?
.send_event(gst::event::Eos::new());
.ok_or("Encoder queue has no input")?;
{
// Serialize at the queue input, where an idle source holds
// no stream lock. The source pad itself can stay locked while
// waiting for its next frame, so locking it would stall stop.
let _input_lock = input.stream_lock();
let last = last_frame
.lock()
.expect("last recording frame mutex poisoned")
.take();
if let (Some((mut buffer, caps)), Some(now)) =
(last, pipeline.current_running_time())
&& caps == input.current_caps()
&& buffer.pts().is_some_and(|pts| {
now > pts + buffer.duration().unwrap_or(gst::ClockTime::ZERO)
})
{
// make_mut copies the buffer header while sharing its
// owned pixel memory; it does not copy a full video frame.
let tail = buffer.make_mut();
tail.set_pts(now);
tail.set_dts(None);
tail.set_duration(gst::ClockTime::from_nseconds(1));
let _ = input.chain(buffer);
}
let _ = input.send_event(gst::event::Eos::new());
}
for (source, queue) in &audio_inputs {
// Audio devices are still alive when video is revoked.
// Let each source order EOS after its own startup events and
@@ -656,6 +690,131 @@ mod tests {
assert_video_decodes(&result.unwrap());
}
#[test]
#[ignore = "requires GStreamer video plugins"]
fn encoder_preserves_the_duration_of_a_static_capture() {
gst::init().unwrap();
for (source_eos, with_audio) in [(false, false), (true, false), (false, true)] {
let directory = tempfile::tempdir().unwrap();
let source = gst::ElementFactory::make("appsrc")
.property("is-live", true)
.property("do-timestamp", true)
.property("format", gst::Format::Time)
.property(
"caps",
gst::Caps::builder("video/x-raw")
.field("format", "BGRx")
.field("width", 64i32)
.field("height", 48i32)
.field("framerate", gst::Fraction::new(30, 1))
.build(),
)
.build()
.unwrap();
let stop = Arc::new(AtomicBool::new(false));
let producer = source.clone();
let request_stop = stop.clone();
let worker = std::thread::spawn(move || {
std::thread::sleep(Duration::from_millis(150));
let mut image = gst::Buffer::with_size(64 * 48 * 4).unwrap();
image
.get_mut()
.unwrap()
.map_writable()
.unwrap()
.as_mut_slice()
.fill(0x80);
assert_eq!(
producer.emit_by_name::<gst::FlowReturn>("push-buffer", &[&image]),
gst::FlowReturn::Ok
);
// The source stays alive without producing a duplicate image.
std::thread::sleep(Duration::from_millis(if with_audio { 4000 } else { 1200 }));
if source_eos {
assert_eq!(
producer.emit_by_name::<gst::FlowReturn>("end-of-stream", &[]),
gst::FlowReturn::Ok
);
} else {
request_stop.store(true, Ordering::Relaxed);
}
});
let audio = if with_audio {
vec![
gst::ElementFactory::make("audiotestsrc")
.property("is-live", true)
.build()
.unwrap(),
]
} else {
Vec::new()
};
let path = encode(source, audio, None, directory.path(), stop).unwrap();
worker.join().unwrap();
let audio_branch = if with_audio {
"demux.audio_0 ! queue ! opusdec ! audioconvert ! audioresample ! audio/x-raw,format=F32LE,rate=48000,channels=1 ! fakesink name=audio sync=false"
} else {
""
};
let decoder = gst::parse::launch(&format!("filesrc location={} ! matroskademux name=demux demux.video_0 ! queue ! vp8dec ! fakesink name=video sync=false {audio_branch}", path.display())).unwrap().downcast::<gst::Pipeline>().unwrap();
let audio_bytes = Arc::new(std::sync::atomic::AtomicU64::new(0));
if with_audio {
let bytes = audio_bytes.clone();
decoder
.by_name("audio")
.unwrap()
.static_pad("sink")
.unwrap()
.add_probe(gst::PadProbeType::BUFFER, move |_, info| {
if let Some(buffer) = info.buffer() {
bytes.fetch_add(buffer.size() as u64, Ordering::Relaxed);
}
gst::PadProbeReturn::Ok
});
}
let end = Arc::new(std::sync::atomic::AtomicU64::new(0));
let observed = end.clone();
decoder
.by_name("video")
.unwrap()
.static_pad("sink")
.unwrap()
.add_probe(gst::PadProbeType::BUFFER, move |_, info| {
if let Some(buffer) = info.buffer() {
let time = buffer.pts().unwrap_or(gst::ClockTime::ZERO)
+ buffer.duration().unwrap_or(gst::ClockTime::ZERO);
observed.fetch_max(time.nseconds(), Ordering::Relaxed);
}
gst::PadProbeReturn::Ok
});
decoder.set_state(gst::State::Playing).unwrap();
let result = decoder.bus().unwrap().timed_pop_filtered(
gst::ClockTime::from_seconds(5),
&[gst::MessageType::Eos, gst::MessageType::Error],
);
decoder.set_state(gst::State::Null).unwrap();
assert!(
matches!(
result.as_ref().map(|message| message.view()),
Some(gst::MessageView::Eos(_))
),
"static recording did not decode: {result:?}"
);
if with_audio {
assert!(
audio_bytes.load(Ordering::Relaxed) >= 3 * 48_000 * 4,
"static video interrupted audio: {} decoded bytes",
audio_bytes.load(Ordering::Relaxed)
);
}
assert!(
end.load(Ordering::Relaxed) >= 1_000_000_000,
"recording lost the idle interval: last decoded frame ends at {} ns",
end.load(Ordering::Relaxed)
);
}
}
#[test]
#[ignore = "requires GStreamer video plugins"]
fn encoder_finishes_a_real_webm() {
portal/src/stream.rs
+533 -252
View File
@@ -1,7 +1,12 @@
use std::cell::RefCell;
use std::collections::HashMap;
use std::io::{Read, Write};
use std::os::fd::AsRawFd;
use std::ptr::NonNull;
use std::rc::Rc;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use std::time::Instant;
use pipewire as pw;
use pipewire::properties::properties;
@@ -9,7 +14,10 @@ use pipewire::spa;
use serde::{Deserialize, Serialize};
use spa::pod::Pod;
use crate::capture::{Capture, Frame};
mod capture;
mod rate;
use capture::{Capture, Completion};
use rate::FrameRate;
#[derive(Debug, Serialize, Deserialize)]
pub struct Ready {
@@ -20,80 +28,94 @@ pub struct Ready {
pub height: u32,
}
struct PoolBuffer {
generation: u64,
buffer: capture::Buffer,
}
#[derive(Clone, Copy)]
struct HeldBuffer {
raw: NonNull<pw::sys::pw_buffer>,
generation: u64,
}
struct Data {
latest: Arc<Mutex<Frame>>,
capture: Rc<RefCell<Capture>>,
stop: Arc<AtomicBool>,
streaming: Arc<AtomicBool>,
fresh: Arc<AtomicBool>,
announced: bool,
rate: Arc<FrameRate>,
buffers: HashMap<usize, PoolBuffer>,
held: Option<HeldBuffer>,
generation: u64,
sequence: u64,
active: bool,
announced: bool,
negotiated: (u32, u32),
dma: bool,
force_shm: bool,
renegotiate: bool,
negotiating: bool,
next_capture: Instant,
}
// pipewire-rs exposes read-only metadata through Buffer. Own a dequeued raw
// buffer here so header writes retain exclusive access and every exit requeues it.
struct OutputBuffer<'a> {
stream: &'a pw::stream::Stream,
raw: std::ptr::NonNull<pw::sys::pw_buffer>,
}
impl<'a> OutputBuffer<'a> {
fn dequeue(stream: &'a pw::stream::Stream) -> Option<Self> {
// SAFETY: called on the owning PipeWire loop; the guard queues exactly once.
let raw = std::ptr::NonNull::new(unsafe { stream.dequeue_raw_buffer() })?;
Some(Self { stream, raw })
}
fn data(&mut self) -> Option<&mut spa::buffer::Data> {
// SAFETY: the dequeued buffer is exclusively ours until Drop. SPA Data
// is repr(transparent), and PipeWire owns the mapped allocation.
unsafe {
let buffer = self.raw.as_ref().buffer.as_mut()?;
if buffer.n_datas == 0 || buffer.datas.is_null() {
return None;
impl Data {
fn complete(&mut self, stream: &pw::stream::Stream, completion: Completion) {
let Some(held) = self.held.take() else {
return;
};
let Some(pool) = self
.buffers
.get(&(held.raw.as_ptr() as usize))
.filter(|pool| pool.generation == held.generation)
else {
// The completed image belongs to a retired pool. Refresh once more
// so the replacement pool receives its own initial image, then
// resume damage waiting. Never queue the retired raw pointer.
if let Err(error) = self.capture.borrow_mut().refresh() {
eprintln!("Could not refresh replacement capture buffers: {error}");
self.stop.store(true, Ordering::Relaxed);
}
buffer.datas.cast::<spa::buffer::Data>().as_mut()
}
}
fn timestamp(&mut self, sequence: u64) {
// SAFETY: mutable access is exclusive, the metadata lookup checks its size,
// and clock_gettime writes only to our initialized stack timespec.
unsafe {
let buffer = self.raw.as_ref().buffer;
if buffer.is_null() {
return;
return;
};
let mut deliver = matches!(completion, Completion::Ready)
&& self.active
&& !self.capture.borrow().stopped();
match completion {
Completion::Retry => {
self.renegotiate = self.negotiated != self.capture.borrow().constraints().size
}
let header = spa::sys::spa_buffer_find_meta_data(
buffer,
spa::sys::SPA_META_Header,
std::mem::size_of::<spa::sys::spa_meta_header>(),
)
.cast::<spa::sys::spa_meta_header>();
if let Some(header) = header.as_mut() {
let mut now = libc::timespec {
tv_sec: 0,
tv_nsec: 0,
};
header.pts = if libc::clock_gettime(libc::CLOCK_MONOTONIC, &mut now) == 0 {
now.tv_sec * 1_000_000_000 + now.tv_nsec
Completion::Failed(error) => {
if self.dma && !self.capture.borrow().stopped() {
eprintln!("DMA-BUF capture unavailable ({error}); using shared memory");
self.force_shm = true;
self.renegotiate = true;
} else {
-1
};
header.flags = 0;
header.offset = 0;
header.dts_offset = 0;
header.seq = sequence;
eprintln!("{error}");
self.stop.store(true, Ordering::Relaxed);
}
deliver = false;
}
Completion::Ready => (),
}
}
}
impl Drop for OutputBuffer<'_> {
fn drop(&mut self) {
// SAFETY: raw came from this stream and has not yet been queued.
let raw = held.raw;
// SAFETY: this pointer is still registered with the same allocation
// generation and exclusively dequeued by us. All callbacks run on the
// owning PipeWire loop; remove_buffer invalidates the registry entry.
unsafe {
self.stream.queue_raw_buffer(self.raw.as_ptr());
let buffer = raw.as_ref().buffer.as_mut().expect("registered SPA buffer");
let data = &mut *buffer.datas;
if let Some(chunk) = data.chunk.as_mut() {
chunk.offset = 0;
chunk.size = if deliver { pool.buffer.bytes } else { 0 };
chunk.stride = pool.buffer.stride as i32;
}
if deliver {
timestamp(raw, self.sequence);
self.sequence = self.sequence.wrapping_add(1);
}
stream.queue_raw_buffer(raw.as_ptr());
}
}
}
fn int_property(key: u32, value: i32) -> spa::pod::Property {
spa::pod::Property {
key,
@@ -111,7 +133,7 @@ fn pod(object: spa::pod::Object) -> Vec<u8> {
.into_inner()
}
fn video_format(width: u32, height: u32) -> Vec<u8> {
fn video_format(width: u32, height: u32, rate: &FrameRate) -> Vec<u8> {
pod(spa::pod::object!(
spa::utils::SpaTypes::ObjectParamFormat,
spa::param::ParamType::EnumFormat,
@@ -137,46 +159,136 @@ fn video_format(width: u32, height: u32) -> Vec<u8> {
),
spa::pod::property!(
spa::param::format::FormatProperties::VideoFramerate,
Choice,
Range,
Fraction,
spa::utils::Fraction { num: 30, denom: 1 }
rate.fraction(),
spa::utils::Fraction { num: 1, denom: 1 },
spa::utils::Fraction {
num: rate.maximum,
denom: 1
}
)
))
}
fn timestamp(raw: std::ptr::NonNull<pw::sys::pw_buffer>, sequence: u64) {
// SAFETY: mutable access is exclusive, the metadata lookup checks its size,
// and clock_gettime writes only to our initialized stack timespec.
unsafe {
let buffer = raw.as_ref().buffer;
if buffer.is_null() {
return;
}
let header = spa::sys::spa_buffer_find_meta_data(
buffer,
spa::sys::SPA_META_Header,
std::mem::size_of::<spa::sys::spa_meta_header>(),
)
.cast::<spa::sys::spa_meta_header>();
if let Some(header) = header.as_mut() {
let mut now = libc::timespec {
tv_sec: 0,
tv_nsec: 0,
};
header.pts = if libc::clock_gettime(libc::CLOCK_MONOTONIC, &mut now) == 0 {
now.tv_sec * 1_000_000_000 + now.tv_nsec
} else {
-1
};
header.flags = 0;
header.offset = 0;
header.dts_offset = 0;
header.seq = sequence;
}
}
}
fn buffer_params((width, height): (u32, u32), dma: bool) -> Vec<u8> {
pod(spa::pod::object!(
spa::utils::SpaTypes::ObjectParamBuffers,
spa::param::ParamType::Buffers,
int_property(spa::sys::SPA_PARAM_BUFFERS_buffers, 4),
int_property(spa::sys::SPA_PARAM_BUFFERS_blocks, 1),
int_property(
spa::sys::SPA_PARAM_BUFFERS_size,
(width * height * 4) as i32
),
int_property(spa::sys::SPA_PARAM_BUFFERS_stride, (width * 4) as i32),
int_property(spa::sys::SPA_PARAM_BUFFERS_align, 16),
// This is a set of supported memory types, not an exact integer.
// Consumers such as OBS offer a scalar mask with several bits;
// a flags choice lets SPA intersect it with our one supported type.
spa::pod::Property {
key: spa::sys::SPA_PARAM_BUFFERS_dataType,
flags: spa::pod::PropertyFlags::empty(),
value: spa::pod::Value::Choice(spa::pod::ChoiceValue::Int(spa::utils::Choice(
spa::utils::ChoiceFlags::empty(),
spa::utils::ChoiceEnum::Flags {
default: 1
<< if dma {
spa::sys::SPA_DATA_DmaBuf
} else {
spa::sys::SPA_DATA_MemFd
},
flags: Vec::new(),
},
),)),
}
))
}
fn formats(size: (u32, u32), rate: &FrameRate, dma: bool) -> Vec<Vec<u8>> {
let shm = video_format(size.0, size.1, rate);
if !dma {
return vec![shm];
}
let (_, mut format) = spa::pod::deserialize::PodDeserializer::deserialize_any_from(&shm)
.expect("serialized video format");
if let spa::pod::Value::Object(ref mut object) = format {
object.properties.push(spa::pod::Property {
key: spa::sys::SPA_FORMAT_VIDEO_modifier,
flags: spa::pod::PropertyFlags::MANDATORY,
value: spa::pod::Value::Long(0),
});
}
let dma =
spa::pod::serialize::PodSerializer::serialize(std::io::Cursor::new(Vec::new()), &format)
.expect("serializable video format")
.0
.into_inner();
vec![dma, shm]
}
pub fn run(
name: String,
cursor: bool,
generation: Option<u32>,
) -> Result<(), Box<dyn std::error::Error>> {
let configured_rate = std::env::var("FERESE_CAPTURE_MAX_FPS").ok();
let rate = Arc::new(FrameRate::new(configured_rate.as_deref())?);
let stop = Arc::new(AtomicBool::new(false));
let (cancel_read, mut cancel_write) = std::os::unix::net::UnixStream::pair()?;
cancel_read.set_nonblocking(true)?;
let pipe_stop = stop.clone();
std::thread::spawn(move || {
let mut byte = [0u8];
let _ = std::io::stdin().read(&mut byte);
pipe_stop.store(true, Ordering::Relaxed);
let _ = cancel_write.write_all(&[1]);
});
let mut capture = Capture::connect(&stop)?;
if let Some(generation) = generation {
capture.pin_output(&name, generation)?;
}
let deadline = Instant::now() + Duration::from_secs(5);
let first = loop {
match capture.frame(&name, cursor, &stop, Vec::new()) {
Err(error) if error == crate::capture::BUSY && Instant::now() < deadline => {
std::thread::sleep(Duration::from_millis(10));
}
result => break result?,
}
};
let (width, height) = (first.width, first.height);
let logical_size = first.logical_size;
let resizable = name.starts_with("window:");
let latest = Arc::new(Mutex::new(first));
let streaming = Arc::new(AtomicBool::new(false));
let fresh = Arc::new(AtomicBool::new(false));
let capture = Rc::new(RefCell::new(Capture::connect(
&name, cursor, generation, &stop,
)?));
let size = capture.borrow().constraints().size;
let logical_size = capture.borrow().logical_size();
let force_shm = std::env::var_os("FERESE_CAPTURE_FORCE_SHM").is_some_and(|value| value == "1");
let attempted_dma = !force_shm && capture.borrow().dma_supported();
// Verify GBM allocation before advertising DMA-BUF to the consumer. Failure
// here leaves the same capture session usable through its SHM constraints.
let dma = attempted_dma && capture.borrow_mut().allocate(true).is_ok();
let force_shm = force_shm || (attempted_dma && !dma);
pw::init();
let mainloop = pw::main_loop::MainLoopRc::new(None)?;
let context = pw::context::ContextRc::new(&mainloop, None)?;
let core = context.connect_rc(None)?;
@@ -184,37 +296,46 @@ pub fn run(
&core,
"Ferese ScreenCast",
properties! {
"media.class" => "Video/Source",
"media.type" => "Video",
"media.category" => "Capture",
"media.role" => "Screen",
"node.name" => format!("ferese.screencast.{}", std::process::id()),
"node.description" => format!("Ferese — {name}"),
"node.virtual" => "true",
"node.want-driver" => "true",
"node.always-process" => "false",
"node.pause-on-idle" => "true",
"node.latency" => "1/30",
"media.class" => "Video/Source", "media.type" => "Video", "media.category" => "Capture",
"media.role" => "Screen", "node.name" => format!("ferese.screencast.{}", std::process::id()),
"node.description" => format!("Ferese — {name}"), "node.virtual" => "true",
"node.want-driver" => "true", "node.always-process" => "false", "node.pause-on-idle" => "true",
"node.latency" => format!("{}/{}", rate.fraction().denom, rate.fraction().num),
},
)?;
let format_frames = latest.clone();
let data = Data {
latest: latest.clone(),
let data = Rc::new(RefCell::new(Data {
capture: capture.clone(),
stop: stop.clone(),
streaming: streaming.clone(),
fresh: fresh.clone(),
announced: false,
rate: rate.clone(),
buffers: HashMap::new(),
held: None,
generation: 0,
sequence: 0,
negotiated: (width, height),
};
active: false,
announced: false,
negotiated: size,
dma,
force_shm,
renegotiate: false,
negotiating: false,
next_capture: Instant::now(),
}));
let _listener = stream
.add_local_listener_with_user_data(data)
.add_local_listener_with_user_data(data.clone())
.state_changed(move |stream, data, _, state| {
data.fresh.store(false, Ordering::SeqCst);
data.streaming.store(
matches!(state, pw::stream::StreamState::Streaming),
Ordering::Relaxed,
);
let mut data = data.borrow_mut();
let active = matches!(state, pw::stream::StreamState::Streaming);
if active
&& !data.active
&& let Err(error) = data.capture.borrow_mut().refresh()
{
eprintln!("Could not resume capture: {error}");
data.stop.store(true, Ordering::Relaxed);
}
if active && !data.active {
data.next_capture = Instant::now();
}
data.active = active;
match state {
pw::stream::StreamState::Paused | pw::stream::StreamState::Streaming
if !data.announced =>
@@ -227,16 +348,16 @@ pub fn run(
serde_json::to_string(&Ready {
logical_size,
node,
width,
height
width: size.0,
height: size.1
})
.expect("typed configuration and IPC values must serialize to JSON")
.expect("serializable capture readiness")
);
let _ = std::io::stdout().flush();
}
}
pw::stream::StreamState::Unconnected if data.announced => {
data.stop.store(true, Ordering::Relaxed);
data.stop.store(true, Ordering::Relaxed)
}
pw::stream::StreamState::Error(error) => {
eprintln!("PipeWire stream: {error}");
@@ -245,44 +366,51 @@ pub fn run(
_ => (),
}
})
.param_changed(move |stream, data, id, param| {
.param_changed(|stream, shared, id, param| {
if id != spa::param::ParamType::Format.as_raw() {
return;
}
let Some(param) = param else { return };
let Some(param) = param else {
return;
};
let mut format = spa::param::video::VideoInfoRaw::default();
let mut data = shared.borrow_mut();
if format.parse(param).is_err()
|| format.size().width == 0
|| format.size().height == 0
|| (!resizable && (format.size().width, format.size().height) != (width, height))
|| u64::from(format.size().width) * u64::from(format.size().height) * 4
> 128 * 1024 * 1024
|| format.format() != spa::param::video::VideoFormat::BGRx
|| (format.size().width, format.size().height)
!= data.capture.borrow().constraints().size
|| data.rate.negotiate(format.framerate()).is_err()
{
data.stop.store(true, Ordering::Relaxed);
return;
}
let (width, height) = (format.size().width, format.size().height);
data.negotiated = (width, height);
let buffers = pod(spa::pod::object!(
spa::utils::SpaTypes::ObjectParamBuffers,
spa::param::ParamType::Buffers,
int_property(spa::sys::SPA_PARAM_BUFFERS_buffers, 4),
int_property(spa::sys::SPA_PARAM_BUFFERS_blocks, 1),
int_property(
spa::sys::SPA_PARAM_BUFFERS_size,
(width * height * 4) as i32
),
int_property(spa::sys::SPA_PARAM_BUFFERS_stride, (width * 4) as i32),
int_property(spa::sys::SPA_PARAM_BUFFERS_align, 16)
));
data.negotiated = (format.size().width, format.size().height);
data.dma = format
.flags()
.contains(spa::param::video::VideoFlags::MODIFIER);
if data.dma && (data.force_shm || format.modifier() != 0) {
data.stop.store(true, Ordering::Relaxed);
return;
}
data.negotiating = false;
let selected = data.rate.fraction();
let properties =
properties! { "node.latency" => format!("{}/{}", selected.denom, selected.num) };
// SAFETY: this loop owns the stream; PipeWire copies the live dictionary.
unsafe {
pw::sys::pw_stream_update_properties(
stream.as_raw_ptr(),
properties.dict().as_raw_ptr(),
);
}
let buffers = buffer_params(data.negotiated, data.dma);
let header = pod(spa::pod::object!(
spa::utils::SpaTypes::ObjectParamMeta,
spa::param::ParamType::Meta,
spa::pod::Property {
key: spa::sys::SPA_PARAM_META_type,
flags: spa::pod::PropertyFlags::empty(),
value: spa::pod::Value::Id(spa::utils::Id(spa::sys::SPA_META_Header)),
value: spa::pod::Value::Id(spa::utils::Id(spa::sys::SPA_META_Header))
},
int_property(
spa::sys::SPA_PARAM_META_size,
@@ -291,145 +419,298 @@ pub fn run(
));
if stream
.update_params(&mut [
Pod::from_bytes(&buffers)
.expect("SPA pod bytes were produced by the matching pod serializer"),
Pod::from_bytes(&header)
.expect("SPA pod bytes were produced by the matching pod serializer"),
Pod::from_bytes(&buffers).expect("serialized buffer constraints"),
Pod::from_bytes(&header).expect("serialized metadata constraints"),
])
.is_err()
{
data.stop.store(true, Ordering::Relaxed);
}
})
.process(|stream, data| {
let Some(mut buffer) = OutputBuffer::dequeue(stream) else {
return;
};
let Some(target) = buffer.data() else {
return;
};
if data.stop.load(Ordering::Relaxed) || !data.fresh.load(Ordering::SeqCst) {
*target.chunk_mut().size_mut() = 0;
return;
}
let frame = data.latest.lock().expect(
"capture frame state must remain consistent across producer and PipeWire callbacks",
);
if (frame.width, frame.height) != data.negotiated {
*target.chunk_mut().size_mut() = 0;
return;
}
let row = frame.width as usize * 4;
let bytes = row * frame.height as usize;
let Some(destination) = target.data() else {
return;
};
if destination.len() < bytes {
.add_buffer(|_, shared, raw| {
let mut data = shared.borrow_mut();
if raw.is_null() || data.buffers.len() >= 8 {
data.stop.store(true, Ordering::Relaxed);
return;
}
for (dst, src) in destination[..bytes]
.chunks_exact_mut(row)
.zip(frame.pixels.chunks_exact(frame.stride as usize))
let result = data.capture.borrow_mut().allocate(data.dma);
match result {
Ok(buffer) => {
// SAFETY: PipeWire grants mutable access during add_buffer.
// ALLOC_BUFFERS delegates fd ownership to us until remove_buffer.
unsafe {
let Some(spa) = (*raw).buffer.as_mut() else {
data.stop.store(true, Ordering::Relaxed);
return;
};
if spa.n_datas != 1 || spa.datas.is_null() {
data.stop.store(true, Ordering::Relaxed);
return;
}
let target = &mut *spa.datas;
target.type_ = if data.dma {
spa::sys::SPA_DATA_DmaBuf
} else {
spa::sys::SPA_DATA_MemFd
};
target.flags = spa::sys::SPA_DATA_FLAG_READWRITE;
target.fd = i64::from(buffer.fd.as_raw_fd());
target.mapoffset = 0;
target.maxsize = buffer.bytes;
target.data = std::ptr::null_mut();
if let Some(chunk) = target.chunk.as_mut() {
chunk.offset = 0;
chunk.size = 0;
chunk.stride = buffer.stride as i32;
}
}
data.generation = data.generation.wrapping_add(1);
let generation = data.generation;
data.buffers
.insert(raw as usize, PoolBuffer { generation, buffer });
}
Err(error) if data.dma => {
eprintln!("DMA-BUF allocation failed ({error}); using shared memory");
data.force_shm = true;
data.renegotiate = true;
}
Err(error) => {
eprintln!("Capture allocation failed: {error}");
data.stop.store(true, Ordering::Relaxed);
}
}
})
.remove_buffer(|_, shared, raw| {
shared.borrow_mut().buffers.remove(&(raw as usize));
})
.process(|stream, shared| {
let mut data = shared.borrow_mut();
if !data.active
|| data.held.is_some()
|| data.negotiating
|| Instant::now() < data.next_capture
|| data.stop.load(Ordering::Relaxed)
{
dst.copy_from_slice(&src[..row]);
return;
}
// SAFETY: called only on the owning loop. The registry and held token
// retain exclusive ownership until completion, removal or disconnect.
let Some(raw) = NonNull::new(unsafe { stream.dequeue_raw_buffer() }) else {
return;
};
let key = raw.as_ptr() as usize;
let result = data
.buffers
.get(&key)
.filter(|pool| pool.buffer.size == data.capture.borrow().constraints().size)
.map(|pool| {
(
pool.generation,
data.capture.borrow_mut().submit(&pool.buffer),
)
});
if let Some((generation, Ok(()))) = result {
data.held = Some(HeldBuffer { raw, generation });
data.next_capture = Instant::now() + data.rate.interval();
} else {
// SAFETY: this buffer was just dequeued and hasn't been handed to
// Wayland. Return it empty; old dimensions cannot reach consumers.
unsafe {
if let Some(chunk) = (*(*raw.as_ptr()).buffer)
.datas
.as_mut()
.and_then(|data| data.chunk.as_mut())
{
chunk.size = 0;
}
stream.queue_raw_buffer(raw.as_ptr());
}
}
*target.chunk_mut().offset_mut() = 0;
*target.chunk_mut().size_mut() = bytes as u32;
*target.chunk_mut().stride_mut() = row as i32;
buffer.timestamp(data.sequence);
data.sequence = data.sequence.wrapping_add(1);
})
.register()?;
let format = video_format(width, height);
let capture_io = capture.clone();
let _capture_io = mainloop.loop_().add_io(
capture.borrow().poll_fd()?,
spa::support::system::IoFlags::IN,
move |_| capture_io.borrow_mut().dispatch(),
);
let _cancel_io =
mainloop
.loop_()
.add_io(cancel_read, spa::support::system::IoFlags::IN, |_| {});
let advertised = formats(size, &rate, dma);
let mut parameters: Vec<_> = advertised
.iter()
.map(|format| Pod::from_bytes(format).expect("serialized video format"))
.collect();
// RT_PROCESS stays disabled: PipeWire callbacks, Wayland dispatch and the
// buffer registry all belong to this single main loop.
stream.connect(
spa::utils::Direction::Output,
None,
pw::stream::StreamFlags::MAP_BUFFERS | pw::stream::StreamFlags::DRIVER,
&mut [Pod::from_bytes(&format)
.expect("SPA pod bytes were produced by the matching pod serializer")],
pw::stream::StreamFlags::ALLOC_BUFFERS | pw::stream::StreamFlags::DRIVER,
&mut parameters,
)?;
let interval = Duration::from_nanos(1_000_000_000 / 30);
let capture_stop = stop.clone();
let capture_streaming = streaming.clone();
let capture_thread = std::thread::spawn(move || {
let mut spare = Vec::new();
let mut checked = Instant::now() - Duration::from_secs(1);
while !capture_stop.load(Ordering::Relaxed) {
let start = Instant::now();
if capture_streaming.load(Ordering::Relaxed)
|| checked.elapsed() >= Duration::from_secs(1)
{
checked = Instant::now();
match capture.frame(&name, cursor, &capture_stop, std::mem::take(&mut spare)) {
Ok(frame) if resizable || (frame.width == width && frame.height == height) => {
spare = std::mem::replace(&mut *latest.lock().expect("capture frame state must remain consistent across producer and PipeWire callbacks"), frame).pixels;
fresh.store(capture_streaming.load(Ordering::Relaxed), Ordering::SeqCst);
}
Ok(_) => {
eprintln!("Shared monitor changed size; start sharing again");
capture_stop.store(true, Ordering::Relaxed);
}
Err(error) if error == crate::capture::BUSY => (),
Err(error) => {
eprintln!("{error}");
capture_stop.store(true, Ordering::Relaxed);
}
}
}
std::thread::sleep(interval.saturating_sub(start.elapsed()));
}
});
let mut next_frame = Instant::now();
let mut requested_size = (width, height);
let mut renegotiation_error = None;
while !stop.load(Ordering::Relaxed) {
let size = {
let frame = format_frames.lock().expect(
"capture frame state must remain consistent across producer and PipeWire callbacks",
while !stop.load(Ordering::Relaxed) && !capture.borrow().stopped() {
capture.borrow_mut().dispatch();
let completion = capture.borrow_mut().completion();
if let Some(completion) = completion {
data.borrow_mut().complete(&stream, completion);
}
let changed = capture.borrow_mut().take_changed();
if changed && !name.starts_with("window:") && capture.borrow().constraints().size != size {
eprintln!("Shared monitor changed size; start sharing again");
break;
}
let renegotiate = {
let mut data = data.borrow_mut();
std::mem::take(&mut data.renegotiate) || changed
};
if renegotiate {
data.borrow_mut().negotiating = true;
let advertised = formats(
capture.borrow().constraints().size,
&rate,
!data.borrow().force_shm && capture.borrow().dma_supported(),
);
(frame.width, frame.height)
};
if resizable && size != requested_size {
let format = video_format(size.0, size.1);
if let Err(error) = stream.update_params(&mut [Pod::from_bytes(&format)
.expect("SPA pod bytes were produced by the matching pod serializer")])
{
renegotiation_error = Some(error);
stop.store(true, Ordering::Relaxed);
break;
}
requested_size = size;
let mut params: Vec<_> = advertised
.iter()
.map(|format| Pod::from_bytes(format).expect("serialized video format"))
.collect();
stream.update_params(&mut params)?;
}
let active = streaming.load(Ordering::Relaxed);
let timeout = if active {
next_frame
.saturating_duration_since(Instant::now())
.min(Duration::from_millis(10))
let can_capture = {
let data = data.borrow();
data.active && data.held.is_none() && !data.renegotiate && !data.negotiating
};
let deadline = next_frame.max(data.borrow().next_capture);
if can_capture && Instant::now() >= deadline {
stream.trigger_process()?;
next_frame = Instant::now() + rate.interval();
}
let timeout = if can_capture {
pw::loop_::Timeout::Finite(
next_frame
.max(data.borrow().next_capture)
.saturating_duration_since(Instant::now()),
)
} else {
Duration::from_millis(10)
pw::loop_::Timeout::Infinite
};
mainloop
.loop_()
.iterate(pw::loop_::Timeout::Finite(timeout));
if streaming.load(Ordering::Relaxed) && Instant::now() >= next_frame {
let _ = stream.trigger_process();
next_frame = Instant::now() + interval;
} else if !streaming.load(Ordering::Relaxed) {
next_frame = Instant::now();
}
mainloop.loop_().iterate(timeout);
}
let _ = stream.disconnect();
let _ = capture_thread.join();
if let Some(error) = renegotiation_error {
return Err(error.into());
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn buffer_types_intersect_obs_masks_in_either_negotiation_order() {
// PipeWire expands OBS's MemPtr bit to also permit MemFd transport.
let obs_mask = (1 << spa::sys::SPA_DATA_MemPtr)
| (1 << spa::sys::SPA_DATA_MemFd)
| (1 << spa::sys::SPA_DATA_DmaBuf);
let consumer = pod(spa::pod::object!(
spa::utils::SpaTypes::ObjectParamBuffers,
spa::param::ParamType::Buffers,
int_property(spa::sys::SPA_PARAM_BUFFERS_dataType, obs_mask)
));
for dma in [false, true] {
let producer = buffer_params((735, 840), dma);
for reverse in [false, true] {
let (offer, filter) = if reverse {
(&consumer, &producer)
} else {
(&producer, &consumer)
};
let mut bytes = vec![0; 1024];
let builder = spa::pod::builder::Builder::new(&mut bytes);
let mut result = std::ptr::null_mut();
// SAFETY: both input pods and the builder's writable storage
// remain alive throughout filtering and reading the result.
let status = unsafe {
spa::sys::spa_pod_filter(
builder.as_raw_ptr(),
&mut result,
Pod::from_bytes(offer).unwrap().as_raw_ptr(),
Pod::from_bytes(filter).unwrap().as_raw_ptr(),
)
};
assert!(
status >= 0,
"SPA rejected dma={dma} reverse={reverse}: {status}"
);
assert!(!result.is_null());
// SAFETY: successful filtering initialized this pod in the
// still-live builder storage; no subsequent writes occur.
let filtered = unsafe { Pod::from_raw(result) };
let (_, value) = spa::pod::deserialize::PodDeserializer::deserialize_any_from(
filtered.as_bytes(),
)
.unwrap();
let spa::pod::Value::Object(object) = value else {
panic!("buffer parameters must remain an object")
};
let types = object
.properties
.iter()
.find(|property| property.key == spa::sys::SPA_PARAM_BUFFERS_dataType)
.unwrap();
let expected = 1
<< if dma {
spa::sys::SPA_DATA_DmaBuf
} else {
spa::sys::SPA_DATA_MemFd
};
match &types.value {
spa::pod::Value::Int(mask) => assert_eq!(*mask, expected),
spa::pod::Value::Choice(spa::pod::ChoiceValue::Int(spa::utils::Choice(
_,
choice,
))) => match choice {
spa::utils::ChoiceEnum::Flags { default, .. }
| spa::utils::ChoiceEnum::None(default) => assert_eq!(*default, expected),
other => panic!("unexpected filtered choice: {other:?}"),
},
other => panic!("unexpected filtered mask: {other:?}"),
}
}
}
}
#[test]
fn dma_is_an_explicit_format_with_a_shared_memory_alternative() {
let rate = FrameRate::new(None).unwrap();
let offers = formats((640, 480), &rate, true);
assert_eq!(offers.len(), 2);
for (index, offer) in offers.iter().enumerate() {
let (_, value) =
spa::pod::deserialize::PodDeserializer::deserialize_any_from(offer).unwrap();
let spa::pod::Value::Object(object) = value else {
panic!("video format must be an object");
};
let modifier = object
.properties
.iter()
.find(|property| property.key == spa::sys::SPA_FORMAT_VIDEO_modifier);
if index == 0 {
let modifier = modifier.expect("DMA-BUF requires an explicit modifier");
assert!(modifier.flags.contains(spa::pod::PropertyFlags::MANDATORY));
assert_eq!(modifier.value, spa::pod::Value::Long(0));
} else {
assert!(
modifier.is_none(),
"SHM must not inherit the DMA-BUF modifier"
);
}
assert!(object.properties.iter().any(|property| property.key
== spa::sys::SPA_FORMAT_VIDEO_framerate
&& matches!(property.value, spa::pod::Value::Choice(_))));
}
assert_eq!(formats((640, 480), &rate, false), vec![offers[1].clone()]);
}
}
portal/src/stream/capture.rs
+627
View File
@@ -0,0 +1,627 @@
use std::collections::HashMap;
use std::fs::File;
use std::os::fd::{AsFd, AsRawFd, FromRawFd, OwnedFd};
use std::os::unix::fs::MetadataExt;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::{Duration, Instant};
use ferese_protocols::window_capture::v1::client::ferese_window_capture_manager_v1 as window_manager;
use wayland_client::protocol::{wl_buffer, wl_output, wl_registry, wl_shm, wl_shm_pool};
use wayland_client::{Connection, Dispatch, EventQueue, QueueHandle, WEnum};
use wayland_protocols::ext::image_capture_source::v1::client::{
ext_image_capture_source_v1 as source,
ext_output_image_capture_source_manager_v1 as output_manager,
};
use wayland_protocols::ext::image_copy_capture::v1::client::{
ext_image_copy_capture_frame_v1 as frame, ext_image_copy_capture_manager_v1 as manager,
ext_image_copy_capture_session_v1 as session,
};
use wayland_protocols::wp::linux_dmabuf::zv1::client::{
zwp_linux_buffer_params_v1 as params, zwp_linux_dmabuf_v1 as dmabuf,
};
#[derive(Clone, Debug, Default, PartialEq)]
pub(super) struct Constraints {
pub size: (u32, u32),
pub shm: bool,
pub device: Option<u64>,
pub linear: bool,
}
impl Constraints {
pub(super) fn validate(&self) -> Result<(), String> {
let (width, height) = self.size;
if width == 0
|| height == 0
|| u64::from(width)
.checked_mul(u64::from(height))
.and_then(|pixels| pixels.checked_mul(4))
.is_none_or(|bytes| bytes > 128 * 1024 * 1024)
|| !self.shm
{
return Err("Invalid capture buffer constraints".into());
}
Ok(())
}
}
#[derive(Debug)]
pub(super) enum Completion {
Ready,
Retry,
Failed(String),
}
pub(super) struct Buffer {
pub proxy: wl_buffer::WlBuffer,
pub fd: OwnedFd,
pub stride: u32,
pub bytes: u32,
pub size: (u32, u32),
_bo: Option<gbm::BufferObject<()>>,
}
impl Drop for Buffer {
fn drop(&mut self) {
self.proxy.destroy();
}
}
#[derive(Default)]
struct State {
outputs: HashMap<u32, (wl_output::WlOutput, String)>,
manager: Option<manager::ExtImageCopyCaptureManagerV1>,
windows: Option<window_manager::FereseWindowCaptureManagerV1>,
output_manager: Option<output_manager::ExtOutputImageCaptureSourceManagerV1>,
shm: Option<wl_shm::WlShm>,
dma: Option<dmabuf::ZwpLinuxDmabufV1>,
pending_constraints: Constraints,
constraints: Option<Constraints>,
logical_size: Option<(u32, u32)>,
changed: bool,
stopped: bool,
completion: Option<Completion>,
error: Option<String>,
selected: Option<u32>,
}
pub(super) struct Capture {
connection: Connection,
queue: EventQueue<State>,
state: State,
session: session::ExtImageCopyCaptureSessionV1,
source: source::ExtImageCaptureSourceV1,
frame: Option<frame::ExtImageCopyCaptureFrameV1>,
pub device: Option<gbm::Device<File>>,
device_id: Option<u64>,
}
impl Capture {
pub(super) fn connect(
name: &str,
cursor: bool,
generation: Option<u32>,
stop: &AtomicBool,
) -> Result<Self, String> {
let connection = Connection::connect_to_env().map_err(|error| error.to_string())?;
let mut queue = connection.new_event_queue();
let qh = queue.handle();
let mut state = State::default();
connection.display().get_registry(&qh, ());
// Startup alone uses bounded roundtrips. Streaming dispatch is integrated
// with PipeWire's fd loop and never waits for an unchanged source.
let deadline = Instant::now() + Duration::from_secs(5);
for _ in 0..2 {
let done = std::sync::Arc::new(AtomicBool::new(false));
connection.display().sync(&qh, done.clone());
while !done.load(Ordering::Relaxed) {
if stop.load(Ordering::Relaxed) || Instant::now() >= deadline {
return Err("Capture startup cancelled or timed out".into());
}
dispatch(&connection, &mut queue, &mut state, 20)?;
}
}
let source = if let Some(id) = name
.strip_prefix("window:")
.and_then(|value| value.parse::<u64>().ok())
{
state
.windows
.as_ref()
.ok_or("Persistent window capture is unavailable")?
.create_source((id >> 32) as u32, id as u32, &qh, ())
} else {
let (&id, (output, _)) = state
.outputs
.iter()
.find(|(_, (_, output_name))| output_name == name)
.ok_or("Shared display disconnected")?;
if generation.is_some_and(|generation| generation != id) {
return Err("Shared display was replaced".into());
}
state.selected = Some(id);
state
.output_manager
.as_ref()
.ok_or("Persistent display capture is unavailable")?
.create_source(output, &qh, ())
};
let options = if cursor {
manager::Options::PaintCursors
} else {
manager::Options::empty()
};
let session = state
.manager
.as_ref()
.ok_or("Persistent screen capture is unavailable")?
.create_session(&source, options, &qh, ());
while state.constraints.is_none() && !state.stopped {
if stop.load(Ordering::Relaxed) || Instant::now() >= deadline {
return Err("Capture startup cancelled or timed out".into());
}
dispatch(&connection, &mut queue, &mut state, 20)?;
}
if state.stopped {
return Err("Capture source is unavailable".into());
}
let constraints = state
.constraints
.as_ref()
.expect("capture constraints received");
constraints.validate()?;
let device_id = constraints.device;
let device = constraints
.device
.and_then(open_device)
.and_then(|file| gbm::Device::new(file).ok());
Ok(Self {
connection,
queue,
state,
session,
source,
frame: None,
device,
device_id,
})
}
pub(super) fn poll_fd(&self) -> Result<OwnedFd, String> {
self.connection
.backend()
.poll_fd()
.try_clone_to_owned()
.map_err(|error| error.to_string())
}
pub(super) fn constraints(&self) -> &Constraints {
self.state
.constraints
.as_ref()
.expect("capture constraints received")
}
pub(super) fn logical_size(&self) -> Option<(u32, u32)> {
self.state.logical_size
}
pub(super) fn stopped(&self) -> bool {
self.state.stopped || self.state.error.is_some()
}
pub(super) fn take_changed(&mut self) -> bool {
std::mem::take(&mut self.state.changed)
}
pub(super) fn dma_supported(&self) -> bool {
self.constraints().linear && self.device.is_some() && self.state.dma.is_some()
}
pub(super) fn dispatch(&mut self) {
if let Err(error) = dispatch(&self.connection, &mut self.queue, &mut self.state, 0) {
self.state.error = Some(error);
}
let device = self.constraints().device;
if device != self.device_id {
self.device = device
.and_then(open_device)
.and_then(|file| gbm::Device::new(file).ok());
self.device_id = device;
}
}
pub(super) fn completion(&mut self) -> Option<Completion> {
let completion = self.state.completion.take()?;
if let Some(frame) = self.frame.take() {
frame.destroy();
}
Some(completion)
}
pub(super) fn refresh(&mut self) -> Result<(), String> {
self.state
.windows
.as_ref()
.ok_or("Capture refresh is unavailable")?
.refresh_source(&self.source);
self.connection.flush().map_err(|error| error.to_string())
}
pub(super) fn submit(&mut self, buffer: &Buffer) -> Result<(), String> {
if self.frame.is_some() || self.stopped() {
return Err("Capture is already pending or stopped".into());
}
let frame = self.session.create_frame(&self.queue.handle(), ());
frame.attach_buffer(&buffer.proxy);
// PipeWire may return any pool buffer. Mark it fully stale; only source
// damage wakes the session, then the compositor fills this whole buffer.
frame.damage_buffer(0, 0, buffer.size.0 as i32, buffer.size.1 as i32);
frame.capture();
self.frame = Some(frame);
self.connection.flush().map_err(|error| error.to_string())
}
pub(super) fn allocate(&mut self, dma: bool) -> Result<Buffer, String> {
let (width, height) = self.constraints().size;
let qh = self.queue.handle();
if dma {
let device = self.device.as_ref().ok_or("No DMA-BUF allocation device")?;
let bo = device
.create_buffer_object::<()>(
width,
height,
gbm::Format::Xrgb8888,
gbm::BufferObjectFlags::RENDERING | gbm::BufferObjectFlags::LINEAR,
)
.map_err(|error| error.to_string())?;
if bo.plane_count() != 1 || bo.modifier() != gbm::Modifier::Linear {
return Err("Allocator did not supply a linear single-plane buffer".into());
}
let fd = bo.fd().map_err(|error| error.to_string())?;
let stride = bo.stride_for_plane(0);
let offset = bo.offset(0);
if offset != 0 {
return Err("Unsupported DMA-BUF plane offset".into());
}
let bytes = stride
.checked_mul(height)
.ok_or("DMA-BUF allocation is too large")?;
let params = self
.state
.dma
.as_ref()
.ok_or("Missing DMA-BUF global")?
.create_params(&qh, ());
params.add(fd.as_fd(), 0, offset, stride, 0, 0);
let proxy = params.create_immed(
width as i32,
height as i32,
gbm::Format::Xrgb8888 as u32,
params::Flags::empty(),
&qh,
(),
);
params.destroy();
return Ok(Buffer {
proxy,
fd,
stride,
bytes,
size: (width, height),
_bo: Some(bo),
});
}
let stride = width.checked_mul(4).ok_or("SHM row is too large")?;
let bytes = stride
.checked_mul(height)
.filter(|bytes| *bytes <= 128 * 1024 * 1024)
.ok_or("SHM buffer is too large")?;
// SAFETY: this returns a new owned anonymous file with a fixed C name.
let fd = unsafe {
libc::memfd_create(
c"ferese-stream".as_ptr(),
libc::MFD_CLOEXEC | libc::MFD_ALLOW_SEALING,
)
};
if fd < 0 {
return Err(std::io::Error::last_os_error().to_string());
}
// SAFETY: memfd_create succeeded and ownership transfers to File.
let file = unsafe { File::from_raw_fd(fd) };
file.set_len(u64::from(bytes))
.map_err(|error| error.to_string())?;
// SAFETY: this file is owned here and has its final size. The compositor
// can map it but neither process needs to resize a pool buffer.
if unsafe {
libc::fcntl(
file.as_raw_fd(),
libc::F_ADD_SEALS,
libc::F_SEAL_SHRINK | libc::F_SEAL_GROW | libc::F_SEAL_SEAL,
)
} < 0
{
return Err(std::io::Error::last_os_error().to_string());
}
let pool = self
.state
.shm
.as_ref()
.ok_or("Missing SHM global")?
.create_pool(file.as_fd(), bytes as i32, &qh, ());
let proxy = pool.create_buffer(
0,
width as i32,
height as i32,
stride as i32,
wl_shm::Format::Xrgb8888,
&qh,
(),
);
pool.destroy();
Ok(Buffer {
proxy,
fd: file.into(),
stride,
bytes,
size: (width, height),
_bo: None,
})
}
}
impl Drop for Capture {
fn drop(&mut self) {
if let Some(frame) = self.frame.take() {
frame.destroy();
}
self.session.destroy();
self.source.destroy();
let _ = self.connection.flush();
}
}
fn open_device(device: u64) -> Option<File> {
std::fs::read_dir("/dev/dri")
.ok()?
.filter_map(Result::ok)
.find_map(|entry| {
if !entry.file_name().as_encoded_bytes().starts_with(b"renderD")
|| entry.metadata().ok()?.rdev() != device
{
return None;
}
File::options()
.read(true)
.write(true)
.open(entry.path())
.ok()
})
}
fn dispatch(
connection: &Connection,
queue: &mut EventQueue<State>,
state: &mut State,
timeout: i32,
) -> Result<(), String> {
queue
.dispatch_pending(state)
.map_err(|error| error.to_string())?;
connection.flush().map_err(|error| error.to_string())?;
if let Some(read) = queue.prepare_read() {
let mut fd = libc::pollfd {
fd: read.connection_fd().as_raw_fd(),
events: libc::POLLIN,
revents: 0,
};
// SAFETY: poll receives one live descriptor owned by the read guard.
let ready = unsafe { libc::poll(&mut fd, 1, timeout) };
if ready > 0 {
read.read().map_err(|error| error.to_string())?;
} else if ready < 0
&& std::io::Error::last_os_error().kind() != std::io::ErrorKind::Interrupted
{
return Err(std::io::Error::last_os_error().to_string());
}
}
queue
.dispatch_pending(state)
.map_err(|error| error.to_string())?;
Ok(())
}
impl Dispatch<wl_registry::WlRegistry, ()> for State {
fn event(
state: &mut Self,
registry: &wl_registry::WlRegistry,
event: wl_registry::Event,
_: &(),
_: &Connection,
qh: &QueueHandle<Self>,
) {
match event {
wl_registry::Event::Global {
name,
interface,
version,
} => match interface.as_str() {
"wl_output" => {
state.outputs.insert(
name,
(
registry.bind(name, version.min(4), qh, name),
format!("output-{name}"),
),
);
}
"ext_image_copy_capture_manager_v1" => {
state.manager = Some(registry.bind(name, 1, qh, ()))
}
"ext_output_image_capture_source_manager_v1" => {
state.output_manager = Some(registry.bind(name, 1, qh, ()))
}
"ferese_window_capture_manager_v1" if version >= 2 => {
state.windows = Some(registry.bind(name, 2, qh, ()))
}
"wl_shm" => state.shm = Some(registry.bind(name, 1, qh, ())),
"zwp_linux_dmabuf_v1" if version >= 3 => {
state.dma = Some(registry.bind(name, version.min(4), qh, ()))
}
_ => (),
},
wl_registry::Event::GlobalRemove { name } => {
state.outputs.remove(&name);
if state.selected == Some(name) {
state.stopped = true;
}
}
_ => (),
}
}
}
impl Dispatch<session::ExtImageCopyCaptureSessionV1, ()> for State {
fn event(
state: &mut Self,
_: &session::ExtImageCopyCaptureSessionV1,
event: session::Event,
_: &(),
_: &Connection,
_: &QueueHandle<Self>,
) {
match event {
session::Event::BufferSize { width, height } => {
state.pending_constraints.size = (width, height)
}
session::Event::ShmFormat {
format: WEnum::Value(wl_shm::Format::Xrgb8888),
} => state.pending_constraints.shm = true,
session::Event::DmabufDevice { device }
if device.len() == std::mem::size_of::<libc::dev_t>() =>
{
let mut bytes = [0u8; 8];
bytes[..device.len()].copy_from_slice(&device);
state.pending_constraints.device = Some(u64::from_ne_bytes(bytes));
}
session::Event::DmabufFormat { format, modifiers }
if format == gbm::Format::Xrgb8888 as u32 =>
{
state.pending_constraints.linear |=
modifiers.chunks_exact(8).any(|bytes| bytes == [0; 8]);
}
session::Event::Done => {
let constraints = std::mem::take(&mut state.pending_constraints);
match constraints.validate() {
Ok(()) => {
state.changed |= state
.constraints
.as_ref()
.is_some_and(|old| old != &constraints);
state.constraints = Some(constraints);
}
Err(error) => state.error = Some(error),
}
}
session::Event::Stopped => state.stopped = true,
_ => (),
}
}
}
impl Dispatch<frame::ExtImageCopyCaptureFrameV1, ()> for State {
fn event(
state: &mut Self,
_: &frame::ExtImageCopyCaptureFrameV1,
event: frame::Event,
_: &(),
_: &Connection,
_: &QueueHandle<Self>,
) {
match event {
frame::Event::Ready => state.completion = Some(Completion::Ready),
frame::Event::Failed {
reason: WEnum::Value(frame::FailureReason::BufferConstraints),
} => state.completion = Some(Completion::Retry),
frame::Event::Failed { reason } => {
state.completion = Some(Completion::Failed(format!("Capture failed: {reason:?}")))
}
frame::Event::Transform { transform }
if transform != WEnum::Value(wl_output::Transform::Normal) =>
{
state.error = Some("Capture returned a non-upright buffer".into())
}
_ => (),
}
}
}
impl Dispatch<window_manager::FereseWindowCaptureManagerV1, ()> for State {
fn event(
state: &mut Self,
_: &window_manager::FereseWindowCaptureManagerV1,
event: window_manager::Event,
_: &(),
_: &Connection,
_: &QueueHandle<Self>,
) {
if let window_manager::Event::SourceGeometry { width, height, .. } = event {
state.logical_size = Some((width, height));
}
}
}
impl Dispatch<wayland_client::protocol::wl_callback::WlCallback, std::sync::Arc<AtomicBool>>
for State
{
fn event(
_: &mut Self,
_: &wayland_client::protocol::wl_callback::WlCallback,
_: wayland_client::protocol::wl_callback::Event,
done: &std::sync::Arc<AtomicBool>,
_: &Connection,
_: &QueueHandle<Self>,
) {
done.store(true, Ordering::Relaxed);
}
}
impl Dispatch<wl_output::WlOutput, u32> for State {
fn event(
state: &mut Self,
_: &wl_output::WlOutput,
event: wl_output::Event,
id: &u32,
_: &Connection,
_: &QueueHandle<Self>,
) {
if let wl_output::Event::Name { name } = event
&& let Some((_, output_name)) = state.outputs.get_mut(id)
{
*output_name = name;
}
}
}
wayland_client::delegate_noop!(State: ignore wl_shm::WlShm);
wayland_client::delegate_noop!(State: ignore wl_shm_pool::WlShmPool);
wayland_client::delegate_noop!(State: ignore wl_buffer::WlBuffer);
wayland_client::delegate_noop!(State: ignore manager::ExtImageCopyCaptureManagerV1);
wayland_client::delegate_noop!(State: ignore output_manager::ExtOutputImageCaptureSourceManagerV1);
wayland_client::delegate_noop!(State: ignore source::ExtImageCaptureSourceV1);
wayland_client::delegate_noop!(State: ignore dmabuf::ZwpLinuxDmabufV1);
wayland_client::delegate_noop!(State: ignore params::ZwpLinuxBufferParamsV1);
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn buffer_constraints_are_bounded_without_integer_overflow() {
let mut constraints = Constraints {
size: (640, 480),
shm: true,
..Constraints::default()
};
assert!(constraints.validate().is_ok());
constraints.size = (16384, 2048);
assert!(constraints.validate().is_ok());
for size in [(0, 1), (1, 0), (16384, 2049), (u32::MAX, u32::MAX)] {
constraints.size = size;
assert!(constraints.validate().is_err(), "{size:?}");
}
constraints.size = (640, 480);
constraints.shm = false;
assert!(constraints.validate().is_err());
}
}
portal/src/stream/rate.rs
+85
View File
@@ -0,0 +1,85 @@
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;
use pipewire::spa::utils::Fraction;
/// The negotiated capture clock, independent of the output's refresh rate.
pub(super) struct FrameRate {
current: AtomicU64,
pub(super) maximum: u32,
}
impl FrameRate {
pub(super) fn new(value: Option<&str>) -> Result<Self, String> {
let maximum = match value {
None => 60,
Some(value) => value
.parse::<u32>()
.ok()
.filter(|value| (1..=240).contains(value))
.ok_or("FERESE_CAPTURE_MAX_FPS must be an integer from 1 to 240")?,
};
Ok(Self {
current: AtomicU64::new((u64::from(maximum.min(30)) << 32) | 1),
maximum,
})
}
pub(super) fn negotiate(&self, rate: Fraction) -> Result<(), &'static str> {
if rate.denom == 0
|| rate.num < rate.denom
|| u64::from(rate.num) > u64::from(self.maximum) * u64::from(rate.denom)
{
return Err("Negotiated capture rate is outside the advertised range");
}
self.current.store(
(u64::from(rate.num) << 32) | u64::from(rate.denom),
Ordering::Relaxed,
);
Ok(())
}
pub(super) fn fraction(&self) -> Fraction {
let packed = self.current.load(Ordering::Relaxed);
Fraction {
num: (packed >> 32) as u32,
denom: packed as u32,
}
}
pub(super) fn interval(&self) -> Duration {
let rate = self.fraction();
Duration::from_nanos(1_000_000_000 * u64::from(rate.denom) / u64::from(rate.num))
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn configuration_bounds_and_default_clock() {
let rate = FrameRate::new(None).unwrap();
assert_eq!(rate.maximum, 60);
assert_eq!(rate.interval(), Duration::from_nanos(33_333_333));
assert_eq!(FrameRate::new(Some("15")).unwrap().fraction().num, 15);
for invalid in ["", "0", "241", "-1", "60.0", " 60"] {
assert!(FrameRate::new(Some(invalid)).is_err(), "{invalid}");
}
}
#[test]
fn consumer_negotiation_preserves_fractional_rates() {
let rate = FrameRate::new(None).unwrap();
rate.negotiate(Fraction {
num: 60_000,
denom: 1001,
})
.unwrap();
assert_eq!(rate.interval(), Duration::from_nanos(16_683_333));
for (num, denom) in [(0, 1), (1, 0), (1, 2), (61, 1), (u32::MAX, 1)] {
assert!(rate.negotiate(Fraction { num, denom }).is_err());
assert_eq!(rate.fraction().num, 60_000);
}
}
}