From 392e9cfeb03597e93234e04a8a4cd6f96bfc5ca5 Mon Sep 17 00:00:00 2001 From: Samuel Onoja Date: Sun, 25 Oct 2026 15:35:09 +0100 Subject: [PATCH] Avoid capture busy-waiting and redundant config and DMA-BUF checks --- compositor/src/backends/direct/feedback.rs | 31 ++++-- compositor/src/dmabuf_feedback.rs | 16 ++- compositor/src/dmabuf_feedback/tests.rs | 9 +- portal/src/lockdown.rs | 121 ++++++++++++++++++--- portal/src/stream.rs | 103 +++++++++++++++--- scripts/tests/test_portal_contracts.py | 14 +++ 6 files changed, 246 insertions(+), 48 deletions(-) diff --git a/compositor/src/backends/direct/feedback.rs b/compositor/src/backends/direct/feedback.rs index d733b1c..c069563 100644 --- a/compositor/src/backends/direct/feedback.rs +++ b/compositor/src/backends/direct/feedback.rs @@ -159,6 +159,7 @@ fn candidate_geometry( && has_storage() } +#[derive(Clone)] struct OutputView<'a> { global: DmabufGlobal, feedback: &'a OutputFeedback, @@ -187,16 +188,19 @@ fn select<'a>( } pub(crate) fn refresh(state: &mut Ferese) { + if state.dmabuf_feedback.requested().next().is_none() { + return; + } let mut preferred = HashMap::new(); if let Some(backend) = state.direct_backend.as_ref() { - for (surface, global) in state.dmabuf_feedback.requested() { - let id = Id::from(surface); - let outputs = backend - .devices - .values() - .flat_map(|device| device.outputs.values().map(move |output| (device, output))) - .filter(|(_, output)| state.output_id(&output.output).is_some()); - let views = outputs.map(|(device, output)| { + // Eligibility belongs to an output, so compute it once for all subscribers. + let outputs = backend + .devices + .values() + .flat_map(|device| device.outputs.values().map(move |output| (device, output))) + .filter(|(_, output)| state.output_id(&output.output).is_some()); + let views: Vec<_> = outputs + .map(|(device, output)| { let eligible = (backend.can_render() && device.drm.is_active() && output.power.can_render() @@ -233,9 +237,14 @@ pub(crate) fn refresh(state: &mut Ferese) { feedback: &output.dmabuf_feedback, eligible, } - }); - if let Some(feedback) = select(global, &id, views) { - preferred.insert(surface.clone(), feedback.clone()); + }) + .collect(); + if views.iter().any(|view| view.eligible.is_some()) { + for (surface, global) in state.dmabuf_feedback.requested() { + let id = Id::from(surface); + if let Some(feedback) = select(global, &id, views.iter().cloned()) { + preferred.insert(surface.clone(), feedback.clone()); + } } } } diff --git a/compositor/src/dmabuf_feedback.rs b/compositor/src/dmabuf_feedback.rs index c5849d7..5053508 100644 --- a/compositor/src/dmabuf_feedback.rs +++ b/compositor/src/dmabuf_feedback.rs @@ -1,7 +1,6 @@ //! Allocation hints retain the import global's renderer as a composition fallback. use std::collections::HashMap; -use smithay::reexports::wayland_server::Resource; use smithay::reexports::wayland_server::protocol::wl_surface::WlSurface; use smithay::wayland::compositor::with_states; use smithay::wayland::dmabuf::{DmabufFeedback, DmabufGlobal, SurfaceDmabufFeedbackState}; @@ -9,6 +8,7 @@ use smithay::wayland::dmabuf::{DmabufFeedback, DmabufGlobal, SurfaceDmabufFeedba struct Surface { global: DmabufGlobal, fallback: DmabufFeedback, + current: DmabufFeedback, } #[derive(Default)] @@ -33,6 +33,7 @@ impl FeedbackManager { Surface { global, fallback: fallback.clone(), + current: fallback.clone(), }, ); Some(fallback) @@ -45,17 +46,21 @@ impl FeedbackManager { } pub fn refresh(&mut self, preferred: &HashMap) { - self.surfaces.retain(|surface, _| surface.is_alive()); - for (surface, record) in &self.surfaces { + for (surface, record) in &mut self.surfaces { let feedback = if self.globals.contains_key(&record.global) { preferred.get(surface).unwrap_or(&record.fallback) } else { &record.fallback }; - // Smithay owns subscriptions, protocol messages and equality suppression. + if &record.current == feedback { + continue; + } + // Avoid locking surface state on every dispatch when its allocation + // hints are unchanged. Smithay still owns subscriptions and messages. with_states(surface, |states| { if let Some(state) = SurfaceDmabufFeedbackState::from_states(states) { state.set_feedback(feedback); + record.current = feedback.clone(); } }); } @@ -65,7 +70,7 @@ impl FeedbackManager { self.globals.remove(&global); for (surface, record) in self .surfaces - .iter() + .iter_mut() .filter(|(_, record)| record.global == global) { // V4 requires a main-device tranche. Retain last-known render @@ -74,6 +79,7 @@ impl FeedbackManager { with_states(surface, |states| { if let Some(state) = SurfaceDmabufFeedbackState::from_states(states) { state.set_feedback(&record.fallback); + record.current = record.fallback.clone(); } }); } diff --git a/compositor/src/dmabuf_feedback/tests.rs b/compositor/src/dmabuf_feedback/tests.rs index 04c8d63..bdc4183 100644 --- a/compositor/src/dmabuf_feedback/tests.rs +++ b/compositor/src/dmabuf_feedback/tests.rs @@ -161,7 +161,7 @@ fn subscriptions_update_deduplicate_and_retire_device_preferences() { state.display_handle.flush_clients().unwrap(); assert!( messages(&mut wire).is_empty(), - "Smithay suppresses unchanged feedback" + "unchanged feedback must not be published again" ); request(&mut wire, 8, 3, &[10, 2], None); @@ -227,6 +227,9 @@ fn subscriptions_update_deduplicate_and_retire_device_preferences() { ); drop(wire); dispatch(&mut events, &mut state); - state.dmabuf_feedback.refresh(&HashMap::new()); - assert_eq!(state.dmabuf_feedback.requested().count(), 0); + assert_eq!( + state.dmabuf_feedback.requested().count(), + 0, + "surface destruction must remove subscriptions without a refresh scan" + ); } diff --git a/portal/src/lockdown.rs b/portal/src/lockdown.rs index e7f8442..e3e24a3 100644 --- a/portal/src/lockdown.rs +++ b/portal/src/lockdown.rs @@ -23,11 +23,19 @@ const KEYS: [&str; 7] = [ ]; #[derive(Clone)] -pub(crate) struct Lockdown(Arc>); +pub(crate) struct Lockdown(Arc>); + +struct Policy { + values: [bool; 7], + source: Option, +} impl Lockdown { pub(crate) fn new() -> Self { - Self(Arc::new(RwLock::new(load().unwrap_or_default()))) + Self(Arc::new(RwLock::new(Policy { + values: load().unwrap_or_default(), + source: None, + }))) } pub(crate) async fn watch(self, connection: Connection) { @@ -36,17 +44,22 @@ impl Lockdown { loop { tokio::time::sleep(Duration::from_secs(1)).await; let mut current = self.0.write().await; - let Some(next) = load() else { continue }; - if *current == next { + let Some(path) = ferese_config::config_path() else { + continue; + }; + let Some(next) = load_changed(&path, &mut current.source) else { + continue; + }; + if current.values == next { continue; } let changed = KEYS .iter() .enumerate() - .filter(|(index, _)| current[*index] != next[*index]) + .filter(|(index, _)| current.values[*index] != next[*index]) .map(|(index, key)| (*key, Value::from(next[index]))) .collect::>(); - *current = next; + current.values = next; drop(current); let _ = zbus::fdo::Properties::properties_changed( &emitter, @@ -72,7 +85,9 @@ impl Lockdown { crate::backend::authorize(connection, &header).await?; let mut current = self.0.write().await; store(index, value).map_err(zbus::fdo::Error::Failed)?; - current[index] = value; + current.values[index] = value; + // Restoring a previously observed file must undo this write too. + current.source = None; Ok(()) } } @@ -81,7 +96,7 @@ impl Lockdown { impl Lockdown { #[zbus(property, name = "disable-printing")] async fn disable_printing(&self) -> bool { - self.0.read().await[0] + self.0.read().await.values[0] } #[zbus(property, name = "disable-printing")] @@ -96,7 +111,7 @@ impl Lockdown { #[zbus(property, name = "disable-save-to-disk")] async fn disable_save_to_disk(&self) -> bool { - self.0.read().await[1] + self.0.read().await.values[1] } #[zbus(property, name = "disable-save-to-disk")] @@ -111,7 +126,7 @@ impl Lockdown { #[zbus(property, name = "disable-application-handlers")] async fn disable_application_handlers(&self) -> bool { - self.0.read().await[2] + self.0.read().await.values[2] } #[zbus(property, name = "disable-application-handlers")] @@ -126,7 +141,7 @@ impl Lockdown { #[zbus(property, name = "disable-location")] async fn disable_location(&self) -> bool { - self.0.read().await[3] + self.0.read().await.values[3] } #[zbus(property, name = "disable-location")] @@ -141,7 +156,7 @@ impl Lockdown { #[zbus(property, name = "disable-camera")] async fn disable_camera(&self) -> bool { - self.0.read().await[4] + self.0.read().await.values[4] } #[zbus(property, name = "disable-camera")] @@ -156,7 +171,7 @@ impl Lockdown { #[zbus(property, name = "disable-microphone")] async fn disable_microphone(&self) -> bool { - self.0.read().await[5] + self.0.read().await.values[5] } #[zbus(property, name = "disable-microphone")] @@ -171,7 +186,7 @@ impl Lockdown { #[zbus(property, name = "disable-sound-output")] async fn disable_sound_output(&self) -> bool { - self.0.read().await[6] + self.0.read().await.values[6] } #[zbus(property, name = "disable-sound-output")] @@ -205,6 +220,21 @@ fn load() -> Option<[bool; 7]> { } fn load_path(path: &Path) -> Option<[bool; 7]> { + values(&ferese_config::Document::parse(&read_source(path)?).ok()?) +} + +fn load_changed(path: &Path, previous: &mut Option) -> Option<[bool; 7]> { + let source = read_source(path)?; + if previous.as_ref() == Some(&source) { + return None; + } + // Compare contents so atomic saves and edits with unchanged timestamps work. + // Remember invalid edits too, while retaining the last valid policy. + *previous = Some(source); + values(&ferese_config::Document::parse(previous.as_deref()?).ok()?) +} + +fn read_source(path: &Path) -> Option { let mut source = String::new(); match std::fs::File::open(path) { Ok(file) => { @@ -218,7 +248,7 @@ fn load_path(path: &Path) -> Option<[bool; 7]> { Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} Err(_) => return None, } - values(&ferese_config::Document::parse(&source).ok()?) + Some(source) } fn store(index: usize, value: bool) -> Result<(), String> { @@ -246,6 +276,67 @@ fn store_path(path: &Path, index: usize, value: bool) -> Result<(), String> { mod tests { use super::*; + #[test] + fn polling_reloads_only_changed_contents_and_recovers_after_invalid_edits() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("config.kdl"); + let mut source = None; + assert_eq!(load_changed(&path, &mut source), Some([false; 7])); + assert_eq!(load_changed(&path, &mut source), None); + + store_path(&path, 4, true).unwrap(); + assert!(load_changed(&path, &mut source).unwrap()[4]); + assert_eq!(load_changed(&path, &mut source), None); + + for invalid in ["portals {", "portals { lockdown { disable-camera 1; }; }"] { + std::fs::write(&path, invalid).unwrap(); + assert_eq!(load_changed(&path, &mut source), None); + assert_eq!(source.as_deref(), Some(invalid)); + assert_eq!(load_changed(&path, &mut source), None); + } + + // Replacing the file must recover even after a malformed edit. + let replacement = directory.path().join("replacement.kdl"); + std::fs::write( + &replacement, + "portals { lockdown { disable-camera #true; }; }", + ) + .unwrap(); + std::fs::rename(replacement, &path).unwrap(); + assert!(load_changed(&path, &mut source).unwrap()[4]); + std::fs::remove_file(&path).unwrap(); + assert_eq!(load_changed(&path, &mut source), Some([false; 7])); + store_path(&path, 4, true).unwrap(); + assert!(load_changed(&path, &mut source).unwrap()[4]); + } + + #[test] + fn polling_detects_same_size_edits_and_retries_failed_reads() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("config.kdl"); + let mut source = None; + std::fs::write(&path, "portals { lockdown { disable-camera #true ; }; }").unwrap(); + assert!(load_changed(&path, &mut source).unwrap()[4]); + let modified = std::fs::metadata(&path).unwrap().modified().unwrap(); + std::fs::write(&path, "portals { lockdown { disable-camera #false; }; }").unwrap(); + std::fs::File::options() + .write(true) + .open(&path) + .unwrap() + .set_modified(modified) + .unwrap(); + assert!(!load_changed(&path, &mut source).unwrap()[4]); + + std::fs::remove_file(&path).unwrap(); + std::fs::create_dir(&path).unwrap(); + let previous = source.clone(); + assert_eq!(load_changed(&path, &mut source), None); + assert_eq!(source, previous); + std::fs::remove_dir(&path).unwrap(); + store_path(&path, 4, true).unwrap(); + assert!(load_changed(&path, &mut source).unwrap()[4]); + } + #[test] fn policy_save_respects_other_config_writers_and_preserves_their_changes() { let directory = tempfile::tempdir().unwrap(); diff --git a/portal/src/stream.rs b/portal/src/stream.rs index df35643..87d6b55 100644 --- a/portal/src/stream.rs +++ b/portal/src/stream.rs @@ -6,7 +6,7 @@ use std::ptr::NonNull; use std::rc::Rc; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; -use std::time::Instant; +use std::time::{Duration, Instant}; use pipewire as pw; use pipewire::properties::properties; @@ -537,16 +537,18 @@ pub fn run( } }) .register()?; - let capture_io = capture.clone(); + // Readiness wakes the loop; dispatch once at its top, alongside deadlines + // and PipeWire callbacks, instead of draining the same queue twice. 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 frame_timer = mainloop.loop_().add_timer(|_| {}); let advertised = formats(size, &rate, dma); let mut parameters: Vec<_> = advertised .iter() @@ -602,27 +604,100 @@ pub fn run( let deadline = next_frame.max(data.borrow().next_capture); if can_capture && Instant::now() >= deadline { stream.trigger_process()?; - next_frame = Instant::now() + rate.interval(); + next_frame = next_frame_deadline(deadline, 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 { - pw::loop_::Timeout::Infinite - }; - mainloop.loop_().iterate(timeout); + update_frame_timer( + &frame_timer, + can_capture.then(|| next_frame.max(data.borrow().next_capture)), + )?; + mainloop.loop_().iterate(pw::loop_::Timeout::Infinite); } let _ = stream.disconnect(); Ok(()) } +fn next_frame_deadline(deadline: Instant, now: Instant, interval: Duration) -> Instant { + let next = deadline + interval; + if next > now { + // Preserve cadence when a timer wakes slightly late. + next + } else { + // Resume after a pause or slow frame without a burst of catch-up frames. + now + interval + } +} + +fn update_frame_timer( + timer: &pw::loop_::TimerSource<'_>, + deadline: Option, +) -> Result<(), spa::utils::result::Error> { + // Loop timeouts truncate to milliseconds and spin near a frame deadline. + // A timer preserves sub-millisecond waits without changing frame pacing. + let delay = deadline.map(|deadline| { + deadline + .saturating_duration_since(Instant::now()) + // Zero disarms a PipeWire timer; an overdue frame must wake instead. + .max(Duration::from_nanos(1)) + }); + timer.update_timer(delay, None).into_result()?; + Ok(()) +} + #[cfg(test)] mod tests { use super::*; + #[test] + fn frame_clock_preserves_cadence_but_does_not_catch_up_after_stalls() { + let start = Instant::now(); + let interval = Duration::from_nanos(16_666_667); + let mut deadline = start; + for frame in 1..=240 { + deadline = + next_frame_deadline(deadline, deadline + Duration::from_micros(200), interval); + assert_eq!(deadline, start + interval * frame); + } + for delay in [interval, Duration::from_secs(2)] { + let resumed = deadline + delay; + assert_eq!( + next_frame_deadline(deadline, resumed, interval), + resumed + interval + ); + } + } + + #[test] + fn frame_timer_wakes_once_for_due_frames_and_disarms_while_paused() { + let mainloop = pw::main_loop::MainLoopRc::new(None).unwrap(); + let expirations = Rc::new(std::cell::Cell::new(0)); + let received = expirations.clone(); + let timer = mainloop.loop_().add_timer(move |count| { + received.set(received.get() + count); + }); + for delay in [Duration::ZERO, Duration::from_micros(100)] { + update_frame_timer(&timer, Some(Instant::now() + delay)).unwrap(); + let expected = expirations.get() + 1; + let limit = Instant::now() + Duration::from_secs(1); + while expirations.get() < expected && Instant::now() < limit { + mainloop + .loop_() + .iterate(pw::loop_::Timeout::Finite(Duration::from_millis(10))); + } + assert_eq!(expirations.get(), expected); + mainloop + .loop_() + .iterate(pw::loop_::Timeout::Finite(Duration::from_millis(5))); + assert_eq!(expirations.get(), expected, "frame timer must be one-shot"); + } + update_frame_timer(&timer, Some(Instant::now())).unwrap(); + update_frame_timer(&timer, None).unwrap(); + let previous = expirations.get(); + mainloop + .loop_() + .iterate(pw::loop_::Timeout::Finite(Duration::from_millis(5))); + assert_eq!(expirations.get(), previous, "paused capture must not wake"); + } + #[test] fn buffer_types_intersect_obs_masks_in_either_negotiation_order() { // PipeWire expands OBS's MemPtr bit to also permit MemFd transport. diff --git a/scripts/tests/test_portal_contracts.py b/scripts/tests/test_portal_contracts.py index 8a72185..ca87250 100644 --- a/scripts/tests/test_portal_contracts.py +++ b/scripts/tests/test_portal_contracts.py @@ -174,6 +174,20 @@ def private_bus_checks(standalone=False): assert "disable-printing #true" in config.read_text() call("org.freedesktop.DBus.Properties", "Set", GLib.Variant("(ssv)", ("org.freedesktop.impl.portal.Lockdown", "disable-printing", GLib.Variant("b", False)))) + # A D-Bus write must invalidate the polling cache. Restoring the + # previously observed file still has to undo the in-memory write. + cached_policy = config.read_text() + time.sleep(1.2) + call("org.freedesktop.DBus.Properties", "Set", GLib.Variant("(ssv)", ("org.freedesktop.impl.portal.Lockdown", "disable-printing", GLib.Variant("b", True)))) + config.write_text(cached_policy) + deadline = time.monotonic() + 5 + while time.monotonic() < deadline: + printing, = call("org.freedesktop.DBus.Properties", "Get", GLib.Variant("(ss)", ("org.freedesktop.impl.portal.Lockdown", "disable-printing")), "(v)") + if printing is False: + break + time.sleep(.02) + assert printing is False, "restoring cached policy did not undo the D-Bus write" + policy_signals = [] subscription = bus.signal_subscribe(NAME, "org.freedesktop.DBus.Properties", "PropertiesChanged", PATH, None, Gio.DBusSignalFlags.NONE, lambda *args: policy_signals.append(args[-1].unpack()))