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

@@ -3,3 +3,4 @@
 
 pub mod config;
 pub mod hyperhdr;
+pub mod slot;

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<u8>,
+}
+
+#[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<Inner>,
+    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<Instant>) -> 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();
+    }
+}