diff --git a/compositor/src/handlers/screencopy.rs b/compositor/src/handlers/screencopy.rs index 0467851..040a4da 100644 --- a/compositor/src/handlers/screencopy.rs +++ b/compositor/src/handlers/screencopy.rs @@ -136,6 +136,15 @@ impl PendingScreencopy { pub(crate) fn fail(&self) { self.sink.fail(); } + + // The compositor-owned request this readback belongs to, if any. Used to + // drop readbacks whose request was already abandoned. + pub(crate) fn request_id(&self) -> Option { + match &self.sink { + CaptureSink::Shm { .. } => None, + CaptureSink::Owned { request, .. } => Some(*request), + } + } } pub(crate) fn capture_allowed() -> bool { diff --git a/compositor/src/handlers/screenshot.rs b/compositor/src/handlers/screenshot.rs index e5cc2ad..301982d 100644 --- a/compositor/src/handlers/screenshot.rs +++ b/compositor/src/handlers/screenshot.rs @@ -2,6 +2,7 @@ use std::path::{Path, PathBuf}; use std::sync::mpsc::SyncSender as ReplySender; +use std::time::{Duration, Instant}; use ferese_ipc::Response; use smithay::utils::{Buffer, Logical, Physical, Point, Rectangle, Size, Transform}; @@ -345,7 +346,7 @@ pub(crate) struct CaptureBuffer { } // A capture part completes exactly once, with either pixels or an error. The -// The compositor thread sends it over a channel that never fills while it is +// compositor thread sends it over a channel that never fills while it is // rendering; a dropped part would leave its request waiting indefinitely. #[derive(Debug)] pub(crate) struct PartOutcome { @@ -386,6 +387,7 @@ struct Request { state: RequestState, parts: Vec<(PartSpec, PartState)>, received: usize, + deadline: Instant, } pub(crate) enum Action { @@ -402,6 +404,11 @@ pub(crate) enum Action { const MAX_OUTSTANDING: usize = 8; const MAX_PENDING_BYTES: usize = 256 * 1024 * 1024; +// A request is answered or abandoned this long after it is admitted. Parts are +// produced by the render path, so an output that is never repainted would +// otherwise leave the request in Collecting forever, holding one of the few +// outstanding slots against every later request. +pub(crate) const REQUEST_TIMEOUT: Duration = Duration::from_secs(30); #[derive(Default)] pub(crate) struct Coordinator { @@ -457,6 +464,7 @@ impl Coordinator { .map(|spec| (spec, PartState::Pending)) .collect(), received: 0, + deadline: Instant::now() + REQUEST_TIMEOUT, }, ); Ok(id) @@ -572,6 +580,35 @@ impl Coordinator { self.reject_inner(request, error); } + // Answers every request that outlived its deadline. The returned ids let the + // caller drop readbacks that would otherwise publish into a request that no + // longer exists. + pub(crate) fn expire(&mut self) -> Vec { + let now = Instant::now(); + let expired: Vec = self + .requests + .iter() + .filter(|(_, entry)| now >= entry.deadline) + .map(|(id, _)| *id) + .collect(); + for id in &expired { + self.reject_inner( + *id, + "Screenshot timed out: an output was not redrawn in time", + ); + } + expired + } + + // The remaining time before the earliest deadline, so the caller can sleep + // exactly that long instead of polling on a fixed tick. + pub(crate) fn next_deadline(&self) -> Option { + self.requests + .values() + .map(|entry| entry.deadline.saturating_duration_since(Instant::now())) + .min() + } + pub(crate) fn terminate_all(&mut self) -> Vec { let ids: Vec = self.requests.keys().copied().collect(); for id in &ids { @@ -709,6 +746,11 @@ pub(crate) fn plan( let Some(clipped) = requested.intersection(output_rect) else { continue; }; + // Smithay's intersection already excludes a merely adjacent output, but + // a degenerate clip must never reach the hard failure below. + if clipped.size.w <= 0 || clipped.size.h <= 0 { + continue; + } // capture_region_for_geometry works in output-local logical space. let local = Rectangle { loc: ( @@ -718,13 +760,19 @@ pub(crate) fn plan( .into(), size: clipped.size, }; + // The clip above already intersected against this same logical rect, so + // a failure here means the two paths disagree. Report it rather than + // quietly returning a smaller image than was asked for. let Some(buffer) = crate::handlers::screencopy::capture_region_for_geometry( output.mode_size, output.scale, output.transform, Some(local), ) else { - continue; + return Err(format!( + "Could not compute a capture region for the output at {}x{}", + output.location.x, output.location.y + )); }; parts.push(PlannedPart { index, @@ -855,6 +903,24 @@ mod planning { assert!(parts.is_err(), "a region off the desktop is reported"); } + #[test] + fn an_output_adjacent_to_the_region_is_skipped_not_fatal() { + // The second output starts exactly where the region ends. It contributes + // nothing, so the capture must still succeed with just the first output + // rather than failing on a region it does not overlap. + let outputs = [ + output(0, 0, 1920, 1080, 1.0), + output(1920, 0, 1920, 1080, 1.0), + ]; + let parts = plan(®ion(100, 100, 1820, 980), &outputs).expect("captures the overlap"); + assert_eq!(parts.len(), 1, "only the overlapping output is read"); + assert_eq!(parts[0].index, 0); + assert_eq!( + (parts[0].logical.size.w, parts[0].logical.size.h), + (1820, 980) + ); + } + #[test] fn a_rotated_output_reads_back_with_swapped_axes() { let mut rotated = output(0, 0, 1920, 1080, 1.0); @@ -1127,6 +1193,76 @@ mod completion { assert!(!harness.coordinator.is_live(id)); } + #[test] + fn a_request_whose_output_is_never_redrawn_is_abandoned() { + // Nothing ever calls on_part, which is what an unrendered output does. + // Without a deadline the request would hold its slot forever and the + // client would never be answered. + let mut harness = Harness::new(2); + let id = harness.coordinator.next_id; + harness + .coordinator + .requests + .get_mut(&id) + .expect("tracked") + .deadline = Instant::now() - Duration::from_secs(1); + + assert_eq!(harness.coordinator.expire(), vec![id]); + let response = harness.response().expect("the caller is answered"); + assert_eq!(response.error.as_ref().unwrap().code, "screenshot_failed"); + assert!( + response + .error + .as_ref() + .unwrap() + .message + .contains("timed out"), + "the failure explains itself: {:?}", + response.error.as_ref().unwrap().message + ); + assert!(!harness.coordinator.is_live(id)); + // A late part for the abandoned request is discarded, not revived. + assert!(matches!( + harness.coordinator.on_part(id, 0, Ok(buffer(2, 2))), + Action::None + )); + assert!(harness.response().is_none()); + } + + #[test] + fn a_live_request_is_not_expired_early() { + let mut harness = Harness::new(1); + let id = harness.coordinator.next_id; + assert!(harness.coordinator.expire().is_empty(), "not yet due"); + assert!(harness.coordinator.is_live(id)); + assert!(harness.response().is_none(), "still unanswered, not failed"); + let remaining = harness.coordinator.next_deadline().expect("still armed"); + assert!( + remaining <= REQUEST_TIMEOUT && remaining > REQUEST_TIMEOUT - Duration::from_secs(1), + "the timer re-arms to roughly the full timeout, got {remaining:?}" + ); + } + + #[test] + fn expiring_frees_the_outstanding_slot() { + let (reply, _received) = sync_channel(1); + let mut coordinator = Coordinator::new(); + for _ in 0..MAX_OUTSTANDING { + coordinator + .admit(42, reply.clone(), vec![spec(2, 2)]) + .expect("within the request cap"); + } + for entry in coordinator.requests.values_mut() { + entry.deadline = Instant::now() - Duration::from_secs(1); + } + assert_eq!(coordinator.expire().len(), MAX_OUTSTANDING); + assert_eq!(coordinator.next_deadline(), None); + // The cap is usable again rather than permanently degraded. + coordinator + .admit(42, reply, vec![spec(2, 2)]) + .expect("a slot was released"); + } + #[test] fn admission_is_bounded_by_requests_and_bytes() { let (reply, _received) = sync_channel(1); diff --git a/compositor/src/handlers/screenshot_worker.rs b/compositor/src/handlers/screenshot_worker.rs index b74453d..9c8e737 100644 --- a/compositor/src/handlers/screenshot_worker.rs +++ b/compositor/src/handlers/screenshot_worker.rs @@ -10,7 +10,7 @@ use smithay::reexports::calloop::channel; use super::screenshot::{Canvas, OutputFrame, compose}; -const QUEUE_CAPACITY: usize = 4; +pub(crate) const QUEUE_CAPACITY: usize = 4; /// A client normally opens and unlinks its screenshot immediately. If it dies /// first, the file is left behind, so staged files are swept well after any /// realistic request could still be reading one. @@ -56,7 +56,10 @@ impl Worker { fn run(receiver: Receiver, results: channel::SyncSender) { while let Ok(job) = receiver.recv() { let result = encode(&job); - // The loop drains this channel, so a full queue only delays the worker. + // This send cannot block. Each request produces exactly one Encode, so at + // most QUEUE_CAPACITY jobs can be queued while one more is being encoded, + // and RESULT_QUEUE_CAPACITY is asserted to exceed that. Blocking here + // would stall the worker with no way for the loop to recover it. if results .send(Encoded { request: job.request, diff --git a/compositor/src/ipc.rs b/compositor/src/ipc.rs index f010735..14918f1 100644 --- a/compositor/src/ipc.rs +++ b/compositor/src/ipc.rs @@ -31,7 +31,19 @@ use smithay::output::Output; const REQUEST_QUEUE_CAPACITY: usize = 128; const MAX_CONNECTIONS: usize = 64; const RESULT_QUEUE_CAPACITY: usize = 8; +// The encode thread pushes one result per job, and at most QUEUE_CAPACITY jobs +// can be queued while a further one is in flight. Keeping the result queue +// larger than that is what stops the worker's send from ever blocking, which +// would otherwise wedge it whenever the event loop is busy. +const _: () = assert!( + RESULT_QUEUE_CAPACITY > crate::handlers::screenshot_worker::QUEUE_CAPACITY + 1, + "the result queue must outsize the job queue plus the in-flight job" +); const SWEEP_INTERVAL: Duration = crate::handlers::screenshot_worker::SWEEP_INTERVAL; +// Only consulted when no request is outstanding, so it bounds how late a newly +// admitted request can be noticed. With a request pending the timer re-arms to +// the exact remaining time instead. +const DEADLINE_IDLE: Duration = Duration::from_secs(1); #[derive(Debug)] struct IpcCall { @@ -136,6 +148,25 @@ pub(crate) fn init( timer::TimeoutAction::ToDuration(SWEEP_INTERVAL) })?; + // Readbacks are published by the render path, so a request whose outputs + // are not redrawn would otherwise wait forever. This re-arms to the exact + // remaining time whenever a request is outstanding, and idles otherwise. + event_loop.handle().insert_source( + timer::Timer::from_duration(DEADLINE_IDLE), + |_, _, state| { + // Abandoned requests have already been answered, so their + // readbacks are simply dropped: they are not failed, because + // that would publish into a request that no longer exists. + for request in state.screenshot.expire() { + state + .pending_screencopies + .retain(|capture| capture.request_id() != Some(request)); + } + let next = state.screenshot.next_deadline().unwrap_or(DEADLINE_IDLE); + timer::TimeoutAction::ToDuration(next) + }, + )?; + tracing::info!(path = %path.display(), "Ferese IPC is accepting connections"); let Some(worker) = worker else {