pixelflux/webcam/
ring.rs

1//! Shared-memory frame ring consumed by the Selkies V4L2 interposer.
2//!
3//! The ring is an anonymous memfd mapped once by this writer and once, read-only, by every
4//! interposer client (it receives the fd over the control socket). Page 0 holds the header and the
5//! per-slot control blocks; frame bytes start at [`DATA_OFFSET`]. Every constant and offset below is
6//! mirrored by `addons/v4l2-interposer/v4l2_interposer.c`; the interposer refuses a connection whose
7//! `version` differs.
8//!
9//! Publishing is a single-writer seqlock per slot: the slot's `seq` is bumped to odd, the frame is
10//! written, `seq` is bumped to even, and only then does the header advertise the slot as the latest.
11//! Readers copy the newest slot and retry if `seq` changed or was odd, so no cross-process lock exists
12//! and a stalled reader never blocks the writer.
13
14use std::io;
15use std::os::fd::{AsRawFd, FromRawFd, OwnedFd, RawFd};
16use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
17
18use memmap2::MmapMut;
19
20pub const SHM_MAGIC: u32 = 0x434B_5753;
21pub const SHM_VERSION: u32 = 1;
22pub const CTRL_OFFSET: u32 = 128;
23pub const CTRL_STRIDE: u32 = 64;
24pub const DATA_OFFSET: u32 = 4096;
25pub const MAX_SLOTS: u32 = 4;
26pub const MIN_SLOTS: u32 = 2;
27/// Byte size of the on-connect configuration struct (`webcam_config_t`).
28pub const CONFIG_SIZE: usize = 64;
29
30const HDR_LATEST_SLOT: usize = 48;
31const HDR_LATEST_FRAME_SEQ: usize = 56;
32
33/// V4L2 fourcc codes the ring can advertise.
34pub const V4L2_PIX_FMT_YUV420: u32 = fourcc(b"YU12");
35pub const V4L2_PIX_FMT_NV12: u32 = fourcc(b"NV12");
36pub const V4L2_PIX_FMT_YUYV: u32 = fourcc(b"YUYV");
37pub const V4L2_PIX_FMT_MJPEG: u32 = fourcc(b"MJPG");
38
39pub const fn fourcc(c: &[u8; 4]) -> u32 {
40    (c[0] as u32) | ((c[1] as u32) << 8) | ((c[2] as u32) << 16) | ((c[3] as u32) << 24)
41}
42
43/// Geometry of the advertised device: a fixed-function camera with one format.
44#[derive(Clone, Copy, Debug, PartialEq, Eq)]
45pub struct RingFormat {
46    pub width: u32,
47    pub height: u32,
48    pub fourcc: u32,
49    pub fps_num: u32,
50    pub fps_den: u32,
51    /// Bytes per line of the first plane; 0 for compressed formats.
52    pub bytesperline: u32,
53    /// Maximum bytes of one frame, also the slot size.
54    pub sizeimage: u32,
55}
56
57impl RingFormat {
58    /// Build the format for a raw fourcc at `width` x `height`; `None` for an unknown fourcc.
59    pub fn raw(
60        fourcc_code: u32,
61        width: u32,
62        height: u32,
63        fps_num: u32,
64        fps_den: u32,
65    ) -> Option<Self> {
66        let (bytesperline, sizeimage) = match fourcc_code {
67            V4L2_PIX_FMT_YUV420 | V4L2_PIX_FMT_NV12 => (
68                width,
69                width * height + 2 * (width.div_ceil(2) * height.div_ceil(2)),
70            ),
71            V4L2_PIX_FMT_YUYV => (width * 2, width * 2 * height),
72            _ => return None,
73        };
74        Some(RingFormat {
75            width,
76            height,
77            fourcc: fourcc_code,
78            fps_num,
79            fps_den,
80            bytesperline,
81            sizeimage,
82        })
83    }
84
85    /// Build the format for the compressed fourcc (MJPEG): no stride, and a frame budget of two
86    /// bytes per pixel, what UVC cameras advertise for MJPEG and far above what a browser's JPEG
87    /// of a camera picture takes.
88    pub fn compressed(
89        fourcc_code: u32,
90        width: u32,
91        height: u32,
92        fps_num: u32,
93        fps_den: u32,
94    ) -> Option<Self> {
95        if fourcc_code != V4L2_PIX_FMT_MJPEG {
96            return None;
97        }
98        Some(RingFormat {
99            width,
100            height,
101            fourcc: fourcc_code,
102            fps_num,
103            fps_den,
104            bytesperline: 0,
105            sizeimage: width * height * 2,
106        })
107    }
108
109    /// Raw or compressed, by fourcc.
110    pub fn for_fourcc(
111        fourcc_code: u32,
112        width: u32,
113        height: u32,
114        fps_num: u32,
115        fps_den: u32,
116    ) -> Option<Self> {
117        Self::raw(fourcc_code, width, height, fps_num, fps_den)
118            .or_else(|| Self::compressed(fourcc_code, width, height, fps_num, fps_den))
119    }
120}
121
122/// Writer side of the ring. Not `Sync`: one thread publishes.
123pub struct Ring {
124    fd: OwnedFd,
125    map: MmapMut,
126    format: RingFormat,
127    n_slots: u32,
128    slot_size: usize,
129    next_slot: u32,
130    slot_seq: Vec<u32>,
131    frame_seq: u64,
132    last_published: Option<(usize, usize)>,
133}
134
135impl Ring {
136    /// Allocate and zero-initialize the memfd and write the header.
137    pub fn new(format: RingFormat, n_slots: u32) -> io::Result<Self> {
138        let n_slots = n_slots.clamp(MIN_SLOTS, MAX_SLOTS);
139        let slot_size = page_align(format.sizeimage.max(1) as usize);
140        let total = DATA_OFFSET as usize + n_slots as usize * slot_size;
141        let name = c"selkies-webcam-staging";
142        let raw: RawFd = unsafe {
143            libc::memfd_create(name.as_ptr(), libc::MFD_CLOEXEC | libc::MFD_ALLOW_SEALING)
144        };
145        if raw < 0 {
146            return Err(io::Error::last_os_error());
147        }
148        let fd = unsafe { OwnedFd::from_raw_fd(raw) };
149        if unsafe { libc::ftruncate(fd.as_raw_fd(), total as libc::off_t) } != 0 {
150            return Err(io::Error::last_os_error());
151        }
152        let map = unsafe { MmapMut::map_mut(&fd)? };
153        // Sealed once this writer's own mapping exists, so the read-only contract each peer
154        // receives the descriptor under is the kernel's: no peer maps it writable or resizes it
155        // (future-write sealing needs Linux 5.1; an older kernel keeps the size seals alone).
156        let size_seals = libc::F_SEAL_SHRINK | libc::F_SEAL_GROW | libc::F_SEAL_SEAL;
157        if unsafe {
158            libc::fcntl(
159                fd.as_raw_fd(),
160                libc::F_ADD_SEALS,
161                size_seals | libc::F_SEAL_FUTURE_WRITE,
162            )
163        } != 0
164        {
165            unsafe { libc::fcntl(fd.as_raw_fd(), libc::F_ADD_SEALS, size_seals) };
166        }
167        let mut ring = Ring {
168            fd,
169            map,
170            format,
171            n_slots,
172            slot_size,
173            next_slot: 0,
174            slot_seq: vec![0; n_slots as usize],
175            frame_seq: 0,
176            last_published: None,
177        };
178        ring.write_header();
179        Ok(ring)
180    }
181
182    pub fn fd(&self) -> RawFd {
183        self.fd.as_raw_fd()
184    }
185
186    pub fn format(&self) -> &RingFormat {
187        &self.format
188    }
189
190    pub fn n_slots(&self) -> u32 {
191        self.n_slots
192    }
193
194    pub fn slot_size(&self) -> usize {
195        self.slot_size
196    }
197
198    pub fn frame_seq(&self) -> u64 {
199        self.frame_seq
200    }
201
202    /// The bytes of the most recently published frame, valid until the next `publish`.
203    pub fn latest_frame(&self) -> Option<&[u8]> {
204        let (slot, used) = self.last_published?;
205        let data_off = DATA_OFFSET as usize + slot * self.slot_size;
206        Some(&self.map[data_off..data_off + used])
207    }
208
209    /// The configuration struct handed to each interposer client once, ahead of the doorbells.
210    pub fn config_bytes(&self) -> [u8; CONFIG_SIZE] {
211        let f = &self.format;
212        let words = [
213            SHM_MAGIC,
214            SHM_VERSION,
215            f.width,
216            f.height,
217            f.fourcc,
218            f.fps_num,
219            f.fps_den,
220            self.n_slots,
221            self.slot_size as u32,
222            DATA_OFFSET,
223            CTRL_OFFSET,
224            CTRL_STRIDE,
225            f.bytesperline,
226            f.sizeimage,
227        ];
228        let mut out = [0u8; CONFIG_SIZE];
229        for (i, w) in words.iter().enumerate() {
230            out[i * 4..i * 4 + 4].copy_from_slice(&w.to_le_bytes());
231        }
232        out
233    }
234
235    fn write_header(&mut self) {
236        let f = self.format;
237        let words = [
238            SHM_MAGIC,
239            SHM_VERSION,
240            f.width,
241            f.height,
242            f.fourcc,
243            f.fps_num,
244            f.fps_den,
245            self.n_slots,
246            self.slot_size as u32,
247            DATA_OFFSET,
248            f.bytesperline,
249            f.sizeimage,
250            0,
251            0,
252        ];
253        for (i, w) in words.iter().enumerate() {
254            self.map[i * 4..i * 4 + 4].copy_from_slice(&w.to_le_bytes());
255        }
256        self.map[HDR_LATEST_FRAME_SEQ..HDR_LATEST_FRAME_SEQ + 8]
257            .copy_from_slice(&0u64.to_le_bytes());
258    }
259
260    fn atomic_u32(&self, offset: usize) -> &AtomicU32 {
261        unsafe { &*(self.map.as_ptr().add(offset) as *const AtomicU32) }
262    }
263
264    fn atomic_u64(&self, offset: usize) -> &AtomicU64 {
265        unsafe { &*(self.map.as_ptr().add(offset) as *const AtomicU64) }
266    }
267
268    /// Publish one frame: `fill` writes the frame into the next slot and returns the byte count.
269    /// A `fill` returning 0 (or more than the slot holds) abandons the slot without publishing it.
270    pub fn publish<F: FnOnce(&mut [u8]) -> usize>(&mut self, ts_ns: u64, fill: F) -> bool {
271        let slot = self.next_slot as usize;
272        let ctrl = CTRL_OFFSET as usize + slot * CTRL_STRIDE as usize;
273        let data_off = DATA_OFFSET as usize + slot * self.slot_size;
274        let seq = self.slot_seq[slot];
275
276        self.atomic_u32(ctrl)
277            .store(seq.wrapping_add(1), Ordering::Release);
278        let slot_size = self.slot_size;
279        let used = fill(&mut self.map[data_off..data_off + slot_size]);
280        if used == 0 || used > slot_size {
281            self.atomic_u32(ctrl)
282                .store(seq.wrapping_add(2), Ordering::Release);
283            self.slot_seq[slot] = seq.wrapping_add(2);
284            return false;
285        }
286        self.frame_seq += 1;
287        self.atomic_u32(ctrl + 4)
288            .store(used as u32, Ordering::Relaxed);
289        self.atomic_u64(ctrl + 8)
290            .store(self.frame_seq, Ordering::Relaxed);
291        self.atomic_u64(ctrl + 16).store(ts_ns, Ordering::Relaxed);
292        self.atomic_u32(ctrl)
293            .store(seq.wrapping_add(2), Ordering::Release);
294        self.slot_seq[slot] = seq.wrapping_add(2);
295
296        self.atomic_u32(HDR_LATEST_SLOT)
297            .store(slot as u32, Ordering::Release);
298        self.atomic_u64(HDR_LATEST_FRAME_SEQ)
299            .store(self.frame_seq, Ordering::Release);
300        self.last_published = Some((slot, used));
301        self.next_slot = (self.next_slot + 1) % self.n_slots;
302        true
303    }
304
305    /// Read back a slot the way a client does, for tests: (bytesused, frame_seq, bytes).
306    #[cfg(test)]
307    pub fn read_latest(&self) -> Option<(u32, u64, Vec<u8>)> {
308        let fseq = self
309            .atomic_u64(HDR_LATEST_FRAME_SEQ)
310            .load(Ordering::Acquire);
311        if fseq == 0 {
312            return None;
313        }
314        let slot = self.atomic_u32(HDR_LATEST_SLOT).load(Ordering::Acquire) as usize;
315        let ctrl = CTRL_OFFSET as usize + slot * CTRL_STRIDE as usize;
316        let s1 = self.atomic_u32(ctrl).load(Ordering::Acquire);
317        let used = self.atomic_u32(ctrl + 4).load(Ordering::Relaxed) as usize;
318        let cseq = self.atomic_u64(ctrl + 8).load(Ordering::Relaxed);
319        let data_off = DATA_OFFSET as usize + slot * self.slot_size;
320        let bytes = self.map[data_off..data_off + used].to_vec();
321        let s2 = self.atomic_u32(ctrl).load(Ordering::Acquire);
322        if s1 != s2 || s1 & 1 == 1 {
323            return None;
324        }
325        Some((used as u32, cseq, bytes))
326    }
327}
328
329pub fn page_align(n: usize) -> usize {
330    let page = unsafe { libc::sysconf(libc::_SC_PAGESIZE) };
331    let page = if page > 0 { page as usize } else { 4096 };
332    n.div_ceil(page) * page
333}
334
335#[cfg(test)]
336mod tests {
337    use super::*;
338
339    /// A peer that receives the descriptor maps it read-only and nothing else: a writable map
340    /// or a resize is the kernel's to refuse, while this writer's own mapping keeps working.
341    #[test]
342    fn peers_cannot_write_or_resize_the_ring() {
343        let f = RingFormat::raw(V4L2_PIX_FMT_YUV420, 64, 48, 30, 1).unwrap();
344        let mut ring = Ring::new(f, 3).unwrap();
345        let fd = ring.fd();
346        let len = DATA_OFFSET as usize + ring.n_slots() as usize * ring.slot_size();
347        let map =
348            |prot| unsafe { libc::mmap(std::ptr::null_mut(), len, prot, libc::MAP_SHARED, fd, 0) };
349        let ro = map(libc::PROT_READ);
350        assert_ne!(ro, libc::MAP_FAILED, "a read-only map is what peers get");
351        unsafe { libc::munmap(ro, len) };
352        let seals = unsafe { libc::fcntl(fd, libc::F_GET_SEALS) };
353        assert!(
354            seals & libc::F_SEAL_SHRINK != 0 && seals & libc::F_SEAL_GROW != 0,
355            "seals {seals:#x}"
356        );
357        assert_ne!(
358            unsafe { libc::ftruncate(fd, 4096) },
359            0,
360            "the size is sealed"
361        );
362        if seals & libc::F_SEAL_FUTURE_WRITE != 0 {
363            assert_eq!(
364                map(libc::PROT_READ | libc::PROT_WRITE),
365                libc::MAP_FAILED,
366                "no writable map"
367            );
368        }
369        assert!(ring.publish(1, |dst| {
370            dst[0] = 7;
371            1
372        }));
373        assert_eq!(
374            ring.latest_frame().unwrap()[0],
375            7,
376            "the writer's own mapping still writes"
377        );
378    }
379
380    #[test]
381    fn header_and_config_layout() {
382        let f = RingFormat::raw(V4L2_PIX_FMT_YUV420, 640, 480, 30, 1).unwrap();
383        assert_eq!(f.bytesperline, 640);
384        assert_eq!(f.sizeimage, 640 * 480 * 3 / 2);
385        let ring = Ring::new(f, 3).unwrap();
386        let cfg = ring.config_bytes();
387        let w = |i: usize| u32::from_le_bytes(cfg[i * 4..i * 4 + 4].try_into().unwrap());
388        assert_eq!(w(0), SHM_MAGIC);
389        assert_eq!(w(1), SHM_VERSION);
390        assert_eq!((w(2), w(3)), (640, 480));
391        assert_eq!(w(4), V4L2_PIX_FMT_YUV420);
392        assert_eq!((w(5), w(6)), (30, 1));
393        assert_eq!(w(7), 3);
394        assert_eq!(w(8) as usize, page_align(f.sizeimage as usize));
395        assert_eq!(
396            (w(9), w(10), w(11)),
397            (DATA_OFFSET, CTRL_OFFSET, CTRL_STRIDE)
398        );
399        assert_eq!((w(12), w(13)), (f.bytesperline, f.sizeimage));
400        let hdr = |i: usize| u32::from_le_bytes(ring.map[i * 4..i * 4 + 4].try_into().unwrap());
401        assert_eq!((hdr(10), hdr(11)), (640, f.sizeimage));
402        assert_eq!(hdr(12), 0);
403        assert!(ring.read_latest().is_none());
404    }
405
406    #[test]
407    fn publish_cycles_slots_and_is_readable() {
408        let f = RingFormat::raw(V4L2_PIX_FMT_YUYV, 16, 8, 30, 1).unwrap();
409        let mut ring = Ring::new(f, 2).unwrap();
410        for i in 1..=5u8 {
411            assert!(ring.publish(i as u64 * 1000, |slot| {
412                slot[..f.sizeimage as usize].fill(i);
413                f.sizeimage as usize
414            }));
415            let (used, seq, bytes) = ring.read_latest().unwrap();
416            assert_eq!(used, f.sizeimage);
417            assert_eq!(seq, i as u64);
418            assert!(bytes.iter().all(|b| *b == i));
419        }
420        assert_eq!(ring.frame_seq(), 5);
421        assert!(!ring.publish(0, |_| 0));
422        assert_eq!(ring.frame_seq(), 5);
423    }
424
425    #[test]
426    fn slot_count_is_clamped() {
427        let f = RingFormat::raw(V4L2_PIX_FMT_NV12, 32, 32, 15, 1).unwrap();
428        assert_eq!(Ring::new(f, 1).unwrap().n_slots(), MIN_SLOTS);
429        assert_eq!(Ring::new(f, 99).unwrap().n_slots(), MAX_SLOTS);
430    }
431}
pixelflux/webcam/ring.rs

pixelflux/webcam/
ring.rs
