pixelflux/
recording_sink.rs

1//! Unix-socket video fan-out for external recording.
2//!
3//! Frames are intercepted at the delivery layer (not inside each encoder), so the tap works
4//! uniformly for every full-frame encoder. The pixelflux wire header is skipped so consumers
5//! receive a plain elementary stream that is directly muxable: Annex-B for H.264 and H.265,
6//! a temporal-unit OBU stream for AV1 (each unit opened by a temporal delimiter), and IVF for
7//! VP8 and VP9, whose raw frames carry no framing of their own.
8//!
9//! The tap must never perturb the live viewer transport, and it never copies frame bytes:
10//! stripe payloads are `Arc`-shared, so the encode thread only clones a handle into a bounded
11//! per-client channel drained by a dedicated writer thread. A slow or stalled recorder blocks
12//! nothing but itself and is dropped once its queue overflows. A newly connected client arms
13//! [`RecordingSink::should_force_idr`] so the next encode emits an IDR it can decode from.
14
15use std::fs;
16use std::io::{ErrorKind, Write};
17use std::os::fd::AsRawFd;
18use std::os::unix::net::UnixListener;
19use std::sync::atomic::{AtomicBool, Ordering};
20use std::sync::mpsc::{SyncSender, TrySendError, sync_channel};
21use std::sync::{Arc, Mutex};
22use std::thread;
23use std::time::Duration;
24
25use crate::encoders::codec::{Codec, VIDEO_HEADER_LEN, WIRE_VIDEO};
26use crate::encoders::frame_rate::FrameRate;
27use crate::encoders::software::EncodedStripe;
28
29/// Per-write timeout on a client stream; a stalled write surfaces as a soft error that
30/// [`write_all_frame`] retries, keeping the writer thread responsive to teardown.
31const WRITE_TIMEOUT: Duration = Duration::from_millis(100);
32
33/// How often the non-blocking accept loop retries when no client is waiting.
34const ACCEPT_POLL_INTERVAL: Duration = Duration::from_millis(50);
35
36/// Per-client backlog bound. A recorder that falls this far behind is dropped rather than
37/// allowed to grow memory or push back on the shared encode thread.
38const CLIENT_QUEUE_CAP: usize = 256;
39
40/// A queued frame: the container bytes the stream needs ahead of it (an IVF file or frame
41/// header, an AV1 temporal delimiter; usually nothing), the `Arc`-shared payload, and the
42/// byte offset where the recordable stream starts (past the wire header, or `0` when the
43/// payload is bare).
44struct QueuedFrame {
45    prefix: Vec<u8>,
46    #[expect(
47        clippy::rc_buffer,
48        reason = "the encoder's buffer as it hands it over; an `Arc<[u8]>` would copy every frame"
49    )]
50    data: Arc<Vec<u8>>,
51    offset: usize,
52}
53
54/// The sink's handle to one connected recorder: the feed end of its bounded queue, a kill
55/// switch for its writer thread, and how many frames it has been sent (the IVF frame index).
56/// The socket itself is owned solely by that writer thread.
57struct ClientHandle {
58    tx: SyncSender<QueuedFrame>,
59    stop: Arc<AtomicBool>,
60    frames: u64,
61}
62
63/// Unix-socket fan-out that broadcasts every encoded video frame to connected consumers.
64///
65/// A listener thread accepts connections and gives each its own bounded queue and writer thread
66/// (see [`ClientHandle`]) so one slow reader cannot stall the others or the encode thread.
67pub struct RecordingSink {
68    /// Filesystem path of the Unix socket; removed on drop.
69    path: String,
70    /// Feed handles for the connected clients, shared with the accept thread.
71    clients: Arc<Mutex<Vec<ClientHandle>>>,
72    /// Signals the accept thread to exit; set in [`Drop`].
73    shutdown: Arc<AtomicBool>,
74    /// Flipped to `true` each time a new client connects; consumed by [`should_force_idr`].
75    ///
76    /// [`should_force_idr`]: RecordingSink::should_force_idr
77    client_connected: Arc<AtomicBool>,
78    /// The capture frame rate an IVF header declares.
79    fps: FrameRate,
80}
81
82impl RecordingSink {
83    /// Bind a Unix socket at `settings_path`, or return `None` when no path is configured or the
84    /// bind fails. Recording is an optional tap that must never take the pipeline down, so a bind
85    /// error is logged and swallowed.
86    pub fn try_bind(settings_path: &str, fps: f64) -> Option<Arc<Self>> {
87        if settings_path.is_empty() {
88            return None;
89        }
90        match Self::bind(settings_path.to_string(), FrameRate::of(fps)) {
91            Ok(sink) => Some(Arc::new(sink)),
92            Err(e) => {
93                eprintln!("[recording_sink] bind failed: {:?}", e);
94                None
95            }
96        }
97    }
98
99    /// Create the socket and spawn the accept thread. Each accepted connection gets a write
100    /// timeout, a bounded queue, and a writer thread; the sink keeps only the feed handle.
101    fn bind(path: String, fps: FrameRate) -> std::io::Result<Self> {
102        let _ = fs::remove_file(&path);
103        let listener = UnixListener::bind(&path)?;
104        crate::webcam::server::restrict_to_owner(&path)?;
105        listener.set_nonblocking(true)?;
106
107        let clients: Arc<Mutex<Vec<ClientHandle>>> = Arc::new(Mutex::new(Vec::new()));
108        let shutdown = Arc::new(AtomicBool::new(false));
109        let client_connected = Arc::new(AtomicBool::new(false));
110
111        let clients_acc = clients.clone();
112        let shutdown_acc = shutdown.clone();
113        let client_connected_acc = client_connected.clone();
114        let path_log = path.clone();
115
116        thread::spawn(move || {
117            eprintln!("[recording_sink] listening on {}", path_log);
118            while !shutdown_acc.load(Ordering::Relaxed) {
119                match listener.accept() {
120                    Ok((stream, _)) => {
121                        if !crate::webcam::server::peer_is_session_user(stream.as_raw_fd()) {
122                            continue;
123                        }
124                        if let Err(e) = stream.set_write_timeout(Some(WRITE_TIMEOUT)) {
125                            eprintln!("[recording_sink] set_write_timeout failed: {:?}", e);
126                            continue;
127                        }
128
129                        let (tx, rx) = sync_channel::<QueuedFrame>(CLIENT_QUEUE_CAP);
130                        let stop = Arc::new(AtomicBool::new(false));
131                        let stop_writer = stop.clone();
132                        thread::spawn(move || {
133                            let mut stream = stream;
134                            for QueuedFrame {
135                                prefix,
136                                data,
137                                offset,
138                            } in rx.iter()
139                            {
140                                if stop_writer.load(Ordering::Relaxed) {
141                                    break;
142                                }
143                                let written = if prefix.is_empty() {
144                                    write_all_frame(&mut stream, &data[offset..], &stop_writer)
145                                } else {
146                                    write_all_frame(&mut stream, &prefix, &stop_writer).and_then(
147                                        |()| {
148                                            write_all_frame(
149                                                &mut stream,
150                                                &data[offset..],
151                                                &stop_writer,
152                                            )
153                                        },
154                                    )
155                                };
156                                if let Err(e) = written {
157                                    eprintln!(
158                                        "[recording_sink] writer thread exiting; write failed: {:?}",
159                                        e
160                                    );
161                                    break;
162                                }
163                            }
164                        });
165
166                        let mut guard = clients_acc.lock().unwrap();
167                        guard.push(ClientHandle {
168                            tx,
169                            stop,
170                            frames: 0,
171                        });
172                        client_connected_acc.store(true, Ordering::Relaxed);
173                        eprintln!("[recording_sink] client connected; total {}", guard.len());
174                    }
175                    Err(e) if e.kind() == ErrorKind::WouldBlock => {
176                        thread::sleep(ACCEPT_POLL_INTERVAL);
177                    }
178                    Err(e) => {
179                        eprintln!("[recording_sink] accept error: {:?}", e);
180                        thread::sleep(Duration::from_millis(500));
181                    }
182                }
183            }
184            eprintln!("[recording_sink] listener thread exiting");
185        });
186
187        Ok(Self {
188            path,
189            clients,
190            shutdown,
191            client_connected,
192            fps,
193        })
194    }
195
196    /// Returns `true` exactly once after a new client connects, signaling that the next encode
197    /// should produce an IDR so the consumer starts from a clean reference frame.
198    pub fn should_force_idr(&self) -> bool {
199        self.client_connected.swap(false, Ordering::Relaxed)
200    }
201
202    /// Delivery-layer tap for one encoded frame: its video frame, JPEG carrying none. The wire
203    /// header is skipped via the queued offset, and the container bytes the codec's stream needs
204    /// are prefixed per client: an IVF file header on a VP8/VP9 client's first frame and an IVF
205    /// frame header on every one, a temporal delimiter ahead of an AV1 unit that lacks one.
206    ///
207    /// Never blocks and never copies the payload: the `Arc` is cloned into each client's
208    /// bounded queue with `try_send`, and a client whose queue is full or whose writer
209    /// died is dropped.
210    pub fn write_frame(&self, stripes: &[EncodedStripe], full_width: i32, full_height: i32) {
211        let Some(stripe) = stripes
212            .iter()
213            .find(|s| s.codec.is_video() && !s.data.is_empty())
214        else {
215            return;
216        };
217        let offset = if stripe.data.len() >= VIDEO_HEADER_LEN && stripe.data[0] == WIRE_VIDEO {
218            VIDEO_HEADER_LEN
219        } else {
220            0
221        };
222        if stripe.data.len() == offset {
223            return;
224        }
225        let payload = &stripe.data[offset..];
226
227        let mut clients = self.clients.lock().unwrap();
228        if clients.is_empty() {
229            return;
230        }
231        let mut to_remove: Vec<usize> = Vec::new();
232        for (idx, client) in clients.iter_mut().enumerate() {
233            let prefix = stream_prefix(
234                stripe.codec,
235                payload,
236                client.frames,
237                (full_width as u16, full_height as u16),
238                self.fps,
239            );
240            client.frames += 1;
241            let queued = QueuedFrame {
242                prefix,
243                data: stripe.data.clone(),
244                offset,
245            };
246            match client.tx.try_send(queued) {
247                Ok(()) => {}
248                Err(TrySendError::Full(_)) => {
249                    eprintln!("[recording_sink] dropping slow client (idx {})", idx);
250                    to_remove.push(idx);
251                }
252                Err(TrySendError::Disconnected(_)) => {
253                    to_remove.push(idx);
254                }
255            }
256        }
257        for idx in to_remove.into_iter().rev() {
258            let removed = clients.swap_remove(idx);
259            removed.stop.store(true, Ordering::Relaxed);
260        }
261    }
262}
263
264impl Drop for RecordingSink {
265    /// Stop accepting, release every writer thread (set each `stop`, then drop its sender so an
266    /// idle writer parked on `rx.iter()` wakes), and remove the socket file.
267    fn drop(&mut self) {
268        self.shutdown.store(true, Ordering::Relaxed);
269        if let Ok(mut clients) = self.clients.lock() {
270            for client in clients.iter() {
271                client.stop.store(true, Ordering::Relaxed);
272            }
273            clients.clear();
274        }
275        let _ = fs::remove_file(&self.path);
276    }
277}
278
279/// The container bytes a client's stream needs ahead of its `index`th frame: an IVF file
280/// header (on the first frame) and frame header for VP8 and VP9, a temporal delimiter for an
281/// AV1 unit that does not open with one, nothing for the Annex-B codecs. The IVF header's time
282/// base is one frame at `fps`, which the frame index counts in.
283fn stream_prefix(
284    codec: Codec,
285    payload: &[u8],
286    index: u64,
287    size: (u16, u16),
288    fps: FrameRate,
289) -> Vec<u8> {
290    match codec {
291        Codec::Vp8 | Codec::Vp9 => {
292            let mut prefix = Vec::with_capacity(44);
293            if index == 0 {
294                prefix.extend_from_slice(b"DKIF");
295                prefix.extend_from_slice(&0u16.to_le_bytes());
296                prefix.extend_from_slice(&32u16.to_le_bytes());
297                prefix.extend_from_slice(if codec == Codec::Vp8 {
298                    b"VP80"
299                } else {
300                    b"VP90"
301                });
302                prefix.extend_from_slice(&size.0.to_le_bytes());
303                prefix.extend_from_slice(&size.1.to_le_bytes());
304                prefix.extend_from_slice(&fps.num.to_le_bytes());
305                prefix.extend_from_slice(&fps.den.to_le_bytes());
306                prefix.extend_from_slice(&0u32.to_le_bytes());
307                prefix.extend_from_slice(&0u32.to_le_bytes());
308            }
309            prefix.extend_from_slice(&(payload.len() as u32).to_le_bytes());
310            prefix.extend_from_slice(&index.to_le_bytes());
311            prefix
312        }
313        Codec::Av1 if payload.first().is_none_or(|&b| (b >> 3) & 0x0f != 2) => vec![0x12, 0x00],
314        _ => Vec::new(),
315    }
316}
317
318/// Write one whole frame to a recorder's socket, resuming across the soft timeouts a slow reader
319/// induces so a partial NAL or OBU is never left behind. Aborts if `stop` is set (the client was
320/// dropped by [`RecordingSink::write_frame`]) or a hard error occurs.
321fn write_all_frame<W: Write>(stream: &mut W, buf: &[u8], stop: &AtomicBool) -> std::io::Result<()> {
322    let mut written = 0usize;
323    while written < buf.len() {
324        if stop.load(Ordering::Relaxed) {
325            return Err(std::io::Error::other("writer stopped (client dropped)"));
326        }
327        match stream.write(&buf[written..]) {
328            Ok(0) => {
329                return Err(std::io::Error::new(
330                    ErrorKind::WriteZero,
331                    "failed to write whole frame",
332                ));
333            }
334            Ok(n) => written += n,
335            Err(ref e) if e.kind() == ErrorKind::TimedOut => {}
336            Err(ref e) if e.kind() == ErrorKind::WouldBlock => {
337                thread::sleep(Duration::from_millis(5));
338            }
339            Err(ref e) if e.kind() == ErrorKind::Interrupted => {}
340            Err(e) => return Err(e),
341        }
342    }
343    Ok(())
344}
345
346#[cfg(test)]
347mod prefix_tests {
348    use super::*;
349
350    /// The Annex-B codecs need nothing ahead of a frame; VP8 and VP9 get an IVF file header
351    /// on a client's first frame and a frame header on every one, carrying the size and the
352    /// frame index; an AV1 unit is opened with a temporal delimiter only when it lacks one.
353    #[test]
354    fn prefixes_follow_the_codec() {
355        assert!(
356            stream_prefix(
357                Codec::H264,
358                &[0, 0, 1, 0x65],
359                0,
360                (1280, 720),
361                FrameRate::of(60.0)
362            )
363            .is_empty()
364        );
365        assert!(
366            stream_prefix(
367                Codec::H265,
368                &[0, 0, 1, 0x26, 1],
369                3,
370                (1280, 720),
371                FrameRate::of(60.0)
372            )
373            .is_empty()
374        );
375
376        let first = stream_prefix(
377            Codec::Vp9,
378            &[0x82, 0x49, 0x83],
379            0,
380            (1280, 720),
381            FrameRate::of(60.0),
382        );
383        assert_eq!(first.len(), 44);
384        assert_eq!(&first[..4], b"DKIF");
385        assert_eq!(&first[8..12], b"VP90");
386        assert_eq!(u16::from_le_bytes([first[12], first[13]]), 1280);
387        assert_eq!(u16::from_le_bytes([first[14], first[15]]), 720);
388        assert_eq!(
389            u32::from_le_bytes([first[16], first[17], first[18], first[19]]),
390            60
391        );
392        assert_eq!(
393            u32::from_le_bytes([first[20], first[21], first[22], first[23]]),
394            1
395        );
396        assert_eq!(
397            u32::from_le_bytes([first[32], first[33], first[34], first[35]]),
398            3
399        );
400        assert_eq!(u64::from_le_bytes(first[36..44].try_into().unwrap()), 0);
401        let later = stream_prefix(Codec::Vp8, &[0u8; 100], 7, (1280, 720), FrameRate::of(60.0));
402        assert_eq!(later.len(), 12);
403        assert_eq!(u32::from_le_bytes(later[..4].try_into().unwrap()), 100);
404        assert_eq!(u64::from_le_bytes(later[4..].try_into().unwrap()), 7);
405        assert_eq!(
406            &stream_prefix(Codec::Vp8, &[], 0, (64, 64), FrameRate::of(30.0))[8..12],
407            b"VP80"
408        );
409
410        assert!(
411            stream_prefix(
412                Codec::Av1,
413                &[0x12, 0x00, 0x32, 0x01, 0x10],
414                0,
415                (64, 64),
416                FrameRate::of(30.0)
417            )
418            .is_empty()
419        );
420        assert_eq!(
421            stream_prefix(
422                Codec::Av1,
423                &[0x32, 0x01, 0x10],
424                0,
425                (64, 64),
426                FrameRate::of(30.0)
427            ),
428            vec![0x12, 0x00]
429        );
430        assert_eq!(
431            stream_prefix(Codec::Av1, &[], 0, (64, 64), FrameRate::of(30.0)),
432            vec![0x12, 0x00]
433        );
434        let ntsc = stream_prefix(
435            Codec::Vp8,
436            &[],
437            0,
438            (64, 64),
439            FrameRate::of(60000.0 / 1001.0),
440        );
441        assert_eq!(
442            u32::from_le_bytes(ntsc[16..20].try_into().unwrap()),
443            60000,
444            "59.94 fps is 1001/60000 s a frame"
445        );
446        assert_eq!(u32::from_le_bytes(ntsc[20..24].try_into().unwrap()), 1001);
447    }
448}
449
450#[cfg(test)]
451mod cost_tests {
452    //! The sink's isolation contract, measured: feeding a frame must cost nothing when no
453    //! recorder is connected (empty-clients early return), microseconds when a healthy
454    //! recorder drains its queue, and stay bounded (a lock + `try_send`, never a blocking
455    //! write) when a recorder stalls completely — until the bounded queue overflows and the
456    //! client is dropped, returning the tap to idle cost.
457
458    use super::*;
459    use std::io::Read;
460    use std::os::unix::net::UnixStream;
461    use std::time::Instant;
462
463    fn frame(len: usize) -> EncodedStripe {
464        let mut data = vec![0u8; len];
465        data[0] = WIRE_VIDEO;
466        EncodedStripe {
467            data: Arc::new(data),
468            codec: Codec::H264,
469            stripe_y_start: 0,
470            stripe_height: 720,
471            frame_id: 0,
472            timing: Default::default(),
473            reference: Default::default(),
474        }
475    }
476
477    /// The time each of `n` feeds took, in microseconds, sorted.
478    fn feed_timed(sink: &RecordingSink, n: usize, len: usize) -> Vec<f64> {
479        let f = frame(len);
480        let mut us: Vec<f64> = (0..n)
481            .map(|_| {
482                let t = Instant::now();
483                sink.write_frame(std::slice::from_ref(&f), 1280, 720);
484                let elapsed = t.elapsed().as_secs_f64() * 1e6;
485                thread::sleep(Duration::from_micros(200));
486                elapsed
487            })
488            .collect();
489        us.sort_by(f64::total_cmp);
490        us
491    }
492
493    /// Each phase is judged by the median and the 99th percentile of its feeds, which one
494    /// preemption of the test thread does not move, where a mean or a maximum would.
495    #[test]
496    fn stalled_recorder_isolation_cost() {
497        let path = format!("/tmp/pf-sink-cost-{}.sock", std::process::id());
498        let sink = RecordingSink::try_bind(&path, 60.0).expect("bind");
499
500        // Idle: no client connected.
501        let idle = feed_timed(&sink, 500, 100_000);
502
503        // Healthy: a client draining as fast as it can.
504        let mut healthy = UnixStream::connect(&path).expect("connect");
505        thread::sleep(Duration::from_millis(200));
506        let drain = thread::spawn(move || {
507            let mut buf = vec![0u8; 1 << 20];
508            while healthy.read(&mut buf).map(|n| n > 0).unwrap_or(false) {}
509        });
510        let healthy = feed_timed(&sink, 500, 100_000);
511
512        // Stalled: a connected client that never reads. The socket buffer fills, then the
513        // bounded queue fills, then the client is dropped (~256 frames later).
514        let reader = UnixStream::connect(&path).expect("connect");
515        thread::sleep(Duration::from_millis(200));
516        let stalled = feed_timed(&sink, 500, 100_000);
517        drop(reader);
518
519        let median = |us: &[f64]| us[us.len() / 2];
520        let p99 = |us: &[f64]| us[us.len() * 99 / 100];
521        for (phase, us) in [
522            ("idle   ", &idle),
523            ("healthy", &healthy),
524            ("stalled", &stalled),
525        ] {
526            println!(
527                "[sink-cost] {phase} median {:.3}us p99 {:.3}us",
528                median(us),
529                p99(us)
530            );
531        }
532        drop(sink);
533        let _ = drain.join();
534
535        assert!(
536            median(&idle) < 5.0,
537            "idle feed should be sub-5us, was {:.3}us",
538            median(&idle)
539        );
540        assert!(
541            median(&healthy) < 100.0,
542            "healthy feed should be tens of us, was {:.3}us",
543            median(&healthy)
544        );
545        assert!(
546            p99(&stalled) < 10_000.0,
547            "a stalled recorder must never block the tap >10ms, was {:.3}us",
548            p99(&stalled)
549        );
550    }
551}
pixelflux/recording_sink.rs

pixelflux/
recording_sink.rs
