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

pixelflux/
recording_sink.rs
