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 = crate::memfd_create(name, libc::MFD_CLOEXEC | libc::MFD_ALLOW_SEALING);
143        if raw < 0 {
144            return Err(io::Error::last_os_error());
145        }
146        let fd = unsafe { OwnedFd::from_raw_fd(raw) };
147        if unsafe { libc::ftruncate(fd.as_raw_fd(), total as libc::off_t) } != 0 {
148            return Err(io::Error::last_os_error());
149        }
150        let map = unsafe { MmapMut::map_mut(&fd)? };
151        // Sealed once this writer's own mapping exists, so the read-only contract each peer
152        // receives the descriptor under is the kernel's: no peer maps it writable or resizes it
153        // (future-write sealing needs Linux 5.1; an older kernel keeps the size seals alone).
154        let size_seals = libc::F_SEAL_SHRINK | libc::F_SEAL_GROW | libc::F_SEAL_SEAL;
155        if unsafe {
156            libc::fcntl(
157                fd.as_raw_fd(),
158                libc::F_ADD_SEALS,
159                size_seals | libc::F_SEAL_FUTURE_WRITE,
160            )
161        } != 0
162        {
163            unsafe { libc::fcntl(fd.as_raw_fd(), libc::F_ADD_SEALS, size_seals) };
164        }
165        let mut ring = Ring {
166            fd,
167            map,
168            format,
169            n_slots,
170            slot_size,
171            next_slot: 0,
172            slot_seq: vec![0; n_slots as usize],
173            frame_seq: 0,
174            last_published: None,
175        };
176        ring.write_header();
177        Ok(ring)
178    }
179
180    pub fn fd(&self) -> RawFd {
181        self.fd.as_raw_fd()
182    }
183
184    pub fn format(&self) -> &RingFormat {
185        &self.format
186    }
187
188    pub fn n_slots(&self) -> u32 {
189        self.n_slots
190    }
191
192    pub fn slot_size(&self) -> usize {
193        self.slot_size
194    }
195
196    pub fn frame_seq(&self) -> u64 {
197        self.frame_seq
198    }
199
200    /// The bytes of the most recently published frame, valid until the next `publish`.
201    pub fn latest_frame(&self) -> Option<&[u8]> {
202        let (slot, used) = self.last_published?;
203        let data_off = DATA_OFFSET as usize + slot * self.slot_size;
204        Some(&self.map[data_off..data_off + used])
205    }
206
207    /// The configuration struct handed to each interposer client once, ahead of the doorbells.
208    pub fn config_bytes(&self) -> [u8; CONFIG_SIZE] {
209        let f = &self.format;
210        let words = [
211            SHM_MAGIC,
212            SHM_VERSION,
213            f.width,
214            f.height,
215            f.fourcc,
216            f.fps_num,
217            f.fps_den,
218            self.n_slots,
219            self.slot_size as u32,
220            DATA_OFFSET,
221            CTRL_OFFSET,
222            CTRL_STRIDE,
223            f.bytesperline,
224            f.sizeimage,
225        ];
226        let mut out = [0u8; CONFIG_SIZE];
227        for (i, w) in words.iter().enumerate() {
228            out[i * 4..i * 4 + 4].copy_from_slice(&w.to_le_bytes());
229        }
230        out
231    }
232
233    fn write_header(&mut self) {
234        let f = self.format;
235        let words = [
236            SHM_MAGIC,
237            SHM_VERSION,
238            f.width,
239            f.height,
240            f.fourcc,
241            f.fps_num,
242            f.fps_den,
243            self.n_slots,
244            self.slot_size as u32,
245            DATA_OFFSET,
246            f.bytesperline,
247            f.sizeimage,
248            0,
249            0,
250        ];
251        for (i, w) in words.iter().enumerate() {
252            self.map[i * 4..i * 4 + 4].copy_from_slice(&w.to_le_bytes());
253        }
254        self.map[HDR_LATEST_FRAME_SEQ..HDR_LATEST_FRAME_SEQ + 8]
255            .copy_from_slice(&0u64.to_le_bytes());
256    }
257
258    fn atomic_u32(&self, offset: usize) -> &AtomicU32 {
259        unsafe { &*(self.map.as_ptr().add(offset) as *const AtomicU32) }
260    }
261
262    fn atomic_u64(&self, offset: usize) -> &AtomicU64 {
263        unsafe { &*(self.map.as_ptr().add(offset) as *const AtomicU64) }
264    }
265
266    /// Publish one frame: `fill` writes the frame into the next slot and returns the byte count.
267    /// A `fill` returning 0 (or more than the slot holds) abandons the slot without publishing it.
268    pub fn publish<F: FnOnce(&mut [u8]) -> usize>(&mut self, ts_ns: u64, fill: F) -> bool {
269        let slot = self.next_slot as usize;
270        let ctrl = CTRL_OFFSET as usize + slot * CTRL_STRIDE as usize;
271        let data_off = DATA_OFFSET as usize + slot * self.slot_size;
272        let seq = self.slot_seq[slot];
273
274        self.atomic_u32(ctrl)
275            .store(seq.wrapping_add(1), Ordering::Release);
276        let slot_size = self.slot_size;
277        let used = fill(&mut self.map[data_off..data_off + slot_size]);
278        if used == 0 || used > slot_size {
279            self.atomic_u32(ctrl)
280                .store(seq.wrapping_add(2), Ordering::Release);
281            self.slot_seq[slot] = seq.wrapping_add(2);
282            return false;
283        }
284        self.frame_seq += 1;
285        self.atomic_u32(ctrl + 4)
286            .store(used as u32, Ordering::Relaxed);
287        self.atomic_u64(ctrl + 8)
288            .store(self.frame_seq, Ordering::Relaxed);
289        self.atomic_u64(ctrl + 16).store(ts_ns, Ordering::Relaxed);
290        self.atomic_u32(ctrl)
291            .store(seq.wrapping_add(2), Ordering::Release);
292        self.slot_seq[slot] = seq.wrapping_add(2);
293
294        self.atomic_u32(HDR_LATEST_SLOT)
295            .store(slot as u32, Ordering::Release);
296        self.atomic_u64(HDR_LATEST_FRAME_SEQ)
297            .store(self.frame_seq, Ordering::Release);
298        self.last_published = Some((slot, used));
299        self.next_slot = (self.next_slot + 1) % self.n_slots;
300        true
301    }
302
303    /// Read back a slot the way a client does, for tests: (bytesused, frame_seq, bytes).
304    #[cfg(test)]
305    pub fn read_latest(&self) -> Option<(u32, u64, Vec<u8>)> {
306        let fseq = self
307            .atomic_u64(HDR_LATEST_FRAME_SEQ)
308            .load(Ordering::Acquire);
309        if fseq == 0 {
310            return None;
311        }
312        let slot = self.atomic_u32(HDR_LATEST_SLOT).load(Ordering::Acquire) as usize;
313        let ctrl = CTRL_OFFSET as usize + slot * CTRL_STRIDE as usize;
314        let s1 = self.atomic_u32(ctrl).load(Ordering::Acquire);
315        let used = self.atomic_u32(ctrl + 4).load(Ordering::Relaxed) as usize;
316        let cseq = self.atomic_u64(ctrl + 8).load(Ordering::Relaxed);
317        let data_off = DATA_OFFSET as usize + slot * self.slot_size;
318        let bytes = self.map[data_off..data_off + used].to_vec();
319        let s2 = self.atomic_u32(ctrl).load(Ordering::Acquire);
320        if s1 != s2 || s1 & 1 == 1 {
321            return None;
322        }
323        Some((used as u32, cseq, bytes))
324    }
325}
326
327pub fn page_align(n: usize) -> usize {
328    let page = unsafe { libc::sysconf(libc::_SC_PAGESIZE) };
329    let page = if page > 0 { page as usize } else { 4096 };
330    n.div_ceil(page) * page
331}
332
333#[cfg(test)]
334mod tests {
335    use super::*;
336
337    /// A peer that receives the descriptor maps it read-only and nothing else: a writable map
338    /// or a resize is the kernel's to refuse, while this writer's own mapping keeps working.
339    #[test]
340    fn peers_cannot_write_or_resize_the_ring() {
341        let f = RingFormat::raw(V4L2_PIX_FMT_YUV420, 64, 48, 30, 1).unwrap();
342        let mut ring = Ring::new(f, 3).unwrap();
343        let fd = ring.fd();
344        let len = DATA_OFFSET as usize + ring.n_slots() as usize * ring.slot_size();
345        let map =
346            |prot| unsafe { libc::mmap(std::ptr::null_mut(), len, prot, libc::MAP_SHARED, fd, 0) };
347        let ro = map(libc::PROT_READ);
348        assert_ne!(ro, libc::MAP_FAILED, "a read-only map is what peers get");
349        unsafe { libc::munmap(ro, len) };
350        let seals = unsafe { libc::fcntl(fd, libc::F_GET_SEALS) };
351        assert!(
352            seals & libc::F_SEAL_SHRINK != 0 && seals & libc::F_SEAL_GROW != 0,
353            "seals {seals:#x}"
354        );
355        assert_ne!(
356            unsafe { libc::ftruncate(fd, 4096) },
357            0,
358            "the size is sealed"
359        );
360        if seals & libc::F_SEAL_FUTURE_WRITE != 0 {
361            assert_eq!(
362                map(libc::PROT_READ | libc::PROT_WRITE),
363                libc::MAP_FAILED,
364                "no writable map"
365            );
366        }
367        assert!(ring.publish(1, |dst| {
368            dst[0] = 7;
369            1
370        }));
371        assert_eq!(
372            ring.latest_frame().unwrap()[0],
373            7,
374            "the writer's own mapping still writes"
375        );
376    }
377
378    #[test]
379    fn header_and_config_layout() {
380        let f = RingFormat::raw(V4L2_PIX_FMT_YUV420, 640, 480, 30, 1).unwrap();
381        assert_eq!(f.bytesperline, 640);
382        assert_eq!(f.sizeimage, 640 * 480 * 3 / 2);
383        let ring = Ring::new(f, 3).unwrap();
384        let cfg = ring.config_bytes();
385        let w = |i: usize| u32::from_le_bytes(cfg[i * 4..i * 4 + 4].try_into().unwrap());
386        assert_eq!(w(0), SHM_MAGIC);
387        assert_eq!(w(1), SHM_VERSION);
388        assert_eq!((w(2), w(3)), (640, 480));
389        assert_eq!(w(4), V4L2_PIX_FMT_YUV420);
390        assert_eq!((w(5), w(6)), (30, 1));
391        assert_eq!(w(7), 3);
392        assert_eq!(w(8) as usize, page_align(f.sizeimage as usize));
393        assert_eq!(
394            (w(9), w(10), w(11)),
395            (DATA_OFFSET, CTRL_OFFSET, CTRL_STRIDE)
396        );
397        assert_eq!((w(12), w(13)), (f.bytesperline, f.sizeimage));
398        let hdr = |i: usize| u32::from_le_bytes(ring.map[i * 4..i * 4 + 4].try_into().unwrap());
399        assert_eq!((hdr(10), hdr(11)), (640, f.sizeimage));
400        assert_eq!(hdr(12), 0);
401        assert!(ring.read_latest().is_none());
402    }
403
404    #[test]
405    fn publish_cycles_slots_and_is_readable() {
406        let f = RingFormat::raw(V4L2_PIX_FMT_YUYV, 16, 8, 30, 1).unwrap();
407        let mut ring = Ring::new(f, 2).unwrap();
408        for i in 1..=5u8 {
409            assert!(ring.publish(u64::from(i) * 1000, |slot| {
410                slot[..f.sizeimage as usize].fill(i);
411                f.sizeimage as usize
412            }));
413            let (used, seq, bytes) = ring.read_latest().unwrap();
414            assert_eq!(used, f.sizeimage);
415            assert_eq!(seq, u64::from(i));
416            assert!(bytes.iter().all(|b| *b == i));
417        }
418        assert_eq!(ring.frame_seq(), 5);
419        assert!(!ring.publish(0, |_| 0));
420        assert_eq!(ring.frame_seq(), 5);
421    }
422
423    #[test]
424    fn slot_count_is_clamped() {
425        let f = RingFormat::raw(V4L2_PIX_FMT_NV12, 32, 32, 15, 1).unwrap();
426        assert_eq!(Ring::new(f, 1).unwrap().n_slots(), MIN_SLOTS);
427        assert_eq!(Ring::new(f, 99).unwrap().n_slots(), MAX_SLOTS);
428    }
429}
pixelflux/webcam/ring.rs

pixelflux/webcam/
ring.rs
