From b3993be5cde6e0b4eede117c5820242c19d1c0c3 Mon Sep 17 00:00:00 2001 From: Oleksandr Berezovskyi Date: Sat, 12 Sep 2026 10:09:23 +0300 Subject: [PATCH] feat(slot): add the latest-frame-wins hand-off between threads The PipeWire thread publishes into a single slot and the network thread takes from it, swapping buffers instead of copying so the steady state does not allocate. An older pending frame is dropped when a newer one arrives. --- src/lib.rs | 1 + src/slot.rs | 187 ++++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 188 insertions(+) create mode 100644 src/slot.rs diff --git a/src/lib.rs b/src/lib.rs index ac0347d..6ca8ba1 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -3,3 +3,4 @@ pub mod config; pub mod hyperhdr; +pub mod slot; diff --git a/src/slot.rs b/src/slot.rs new file mode 100644 index 0000000..6c0782c --- /dev/null +++ b/src/slot.rs @@ -0,0 +1,187 @@ +//! "Latest frame wins" hand-off between the PipeWire thread and the network thread. +//! +//! Buffers are swapped rather than copied, so in steady state no allocations happen. + +use std::sync::{Condvar, Mutex, MutexGuard}; +use std::time::Instant; + +/// An RGB24 frame, rows without padding. +#[derive(Debug, Default, Clone, PartialEq, Eq)] +pub struct Frame { + pub width: u32, + pub height: u32, + pub rgb: Vec, +} + +#[derive(Debug, Default)] +struct Inner { + frame: Frame, + /// `frame` was published and not taken yet. + pending: bool, + source_gone: bool, + wake: bool, + shutdown: bool, +} + +/// What happened since the last [`FrameSlot::wait`]. +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] +pub struct Update { + pub shutdown: bool, + /// The source disappeared. Handle before `new_frame`: such a frame is from a new source. + pub source_gone: bool, + /// The caller's frame was replaced with a newer one. + pub new_frame: bool, + /// Someone called [`FrameSlot::wake`]. + pub woken: bool, +} + +#[derive(Debug, Default)] +pub struct FrameSlot { + inner: Mutex, + cond: Condvar, +} + +impl FrameSlot { + pub fn new() -> Self { + Self::default() + } + + fn lock(&self) -> MutexGuard<'_, Inner> { + // A panic while holding this lock cannot leave the plain data inconsistent. + self.inner.lock().unwrap_or_else(|e| e.into_inner()) + } + + /// Publishes `frame` by swapping it into the slot; `frame` receives the previous buffer + /// for reuse. Returns `true` if an untaken frame was replaced. + pub fn publish(&self, frame: &mut Frame) -> bool { + let mut inner = self.lock(); + std::mem::swap(&mut inner.frame, frame); + let superseded = std::mem::replace(&mut inner.pending, true); + drop(inner); + self.cond.notify_all(); + superseded + } + + /// The source is gone; drops a pending frame. + pub fn source_gone(&self) { + let mut inner = self.lock(); + inner.pending = false; + inner.source_gone = true; + drop(inner); + self.cond.notify_all(); + } + + pub fn wake(&self) { + self.lock().wake = true; + self.cond.notify_all(); + } + + pub fn shutdown(&self) { + self.lock().shutdown = true; + self.cond.notify_all(); + } + + /// Waits until an event occurs or `deadline` passes. A newer frame is swapped into `frame`. + pub fn wait(&self, frame: &mut Frame, deadline: Option) -> Update { + let mut inner = self.lock(); + loop { + if inner.shutdown || inner.source_gone || inner.pending || inner.wake { + break; + } + match deadline { + None => inner = self.cond.wait(inner).unwrap_or_else(|e| e.into_inner()), + Some(deadline) => { + let now = Instant::now(); + if now >= deadline { + break; + } + inner = self + .cond + .wait_timeout(inner, deadline - now) + .unwrap_or_else(|e| e.into_inner()) + .0; + } + } + } + let new_frame = std::mem::take(&mut inner.pending); + if new_frame { + std::mem::swap(&mut inner.frame, frame); + } + Update { + shutdown: inner.shutdown, + source_gone: std::mem::take(&mut inner.source_gone), + new_frame, + woken: std::mem::take(&mut inner.wake), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Arc; + use std::time::Duration; + + fn frame(tag: u8) -> Frame { + Frame { + width: 1, + height: 1, + rgb: vec![tag, tag, tag], + } + } + + #[test] + fn latest_wins_and_buffers_rotate() { + let slot = FrameSlot::new(); + let mut producer = frame(1); + assert!(!slot.publish(&mut producer)); + producer = frame(2); + assert!(slot.publish(&mut producer)); + // The producer got the previous buffer back. + assert_eq!(producer.rgb, [1, 1, 1]); + + let mut consumer = Frame::default(); + let u = slot.wait(&mut consumer, Some(Instant::now())); + assert!(u.new_frame); + assert_eq!(consumer.rgb, [2, 2, 2]); + + let u = slot.wait(&mut consumer, Some(Instant::now())); + assert!(!u.new_frame); + assert_eq!(consumer.rgb, [2, 2, 2]); + } + + #[test] + fn source_gone_drops_pending_frame() { + let slot = FrameSlot::new(); + slot.publish(&mut frame(1)); + slot.source_gone(); + let mut consumer = Frame::default(); + let u = slot.wait(&mut consumer, None); + assert!(u.source_gone && !u.new_frame); + + slot.source_gone(); + slot.publish(&mut frame(3)); + let u = slot.wait(&mut consumer, None); + assert!(u.source_gone && u.new_frame); + assert_eq!(consumer.rgb, [3, 3, 3]); + } + + #[test] + fn wait_times_out_and_wakes() { + let slot = Arc::new(FrameSlot::new()); + let mut consumer = Frame::default(); + let start = Instant::now(); + let u = slot.wait(&mut consumer, Some(start + Duration::from_millis(30))); + assert_eq!(u, Update::default()); + assert!(start.elapsed() >= Duration::from_millis(30)); + + let s = slot.clone(); + let t = std::thread::spawn(move || { + std::thread::sleep(Duration::from_millis(20)); + s.shutdown(); + }); + let u = slot.wait(&mut consumer, None); + assert!(u.shutdown); + t.join().unwrap(); + } +}