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(fourcc_code: u32, width: u32, height: u32, fps_num: u32, fps_den: u32) -> Option<Self> {
60        let (bytesperline, sizeimage) = match fourcc_code {
61            V4L2_PIX_FMT_YUV420 | V4L2_PIX_FMT_NV12 => (width, width * height + 2 * (width.div_ceil(2) * height.div_ceil(2))),
62            V4L2_PIX_FMT_YUYV => (width * 2, width * 2 * height),
63            _ => return None,
64        };
65        Some(RingFormat { width, height, fourcc: fourcc_code, fps_num, fps_den, bytesperline, sizeimage })
66    }
67
68    /// Build the format for the compressed fourcc (MJPEG): no stride, and a frame budget of two
69    /// bytes per pixel, what UVC cameras advertise for MJPEG and far above what a browser's JPEG
70    /// of a camera picture takes.
71    pub fn compressed(fourcc_code: u32, width: u32, height: u32, fps_num: u32, fps_den: u32) -> Option<Self> {
72        if fourcc_code != V4L2_PIX_FMT_MJPEG {
73            return None;
74        }
75        Some(RingFormat { width, height, fourcc: fourcc_code, fps_num, fps_den, bytesperline: 0, sizeimage: width * height * 2 })
76    }
77
78    /// Raw or compressed, by fourcc.
79    pub fn for_fourcc(fourcc_code: u32, width: u32, height: u32, fps_num: u32, fps_den: u32) -> Option<Self> {
80        Self::raw(fourcc_code, width, height, fps_num, fps_den).or_else(|| Self::compressed(fourcc_code, width, height, fps_num, fps_den))
81    }
82}
83
84/// Writer side of the ring. Not `Sync`: one thread publishes.
85pub struct Ring {
86    fd: OwnedFd,
87    map: MmapMut,
88    format: RingFormat,
89    n_slots: u32,
90    slot_size: usize,
91    next_slot: u32,
92    slot_seq: Vec<u32>,
93    frame_seq: u64,
94    last_published: Option<(usize, usize)>,
95}
96
97impl Ring {
98    /// Allocate and zero-initialize the memfd and write the header.
99    pub fn new(format: RingFormat, n_slots: u32) -> io::Result<Self> {
100        let n_slots = n_slots.clamp(MIN_SLOTS, MAX_SLOTS);
101        let slot_size = page_align(format.sizeimage.max(1) as usize);
102        let total = DATA_OFFSET as usize + n_slots as usize * slot_size;
103        let name = c"selkies-webcam-staging";
104        let raw: RawFd = unsafe { libc::memfd_create(name.as_ptr(), libc::MFD_CLOEXEC) };
105        if raw < 0 {
106            return Err(io::Error::last_os_error());
107        }
108        let fd = unsafe { OwnedFd::from_raw_fd(raw) };
109        if unsafe { libc::ftruncate(fd.as_raw_fd(), total as libc::off_t) } != 0 {
110            return Err(io::Error::last_os_error());
111        }
112        let map = unsafe { MmapMut::map_mut(&fd)? };
113        let mut ring = Ring {
114            fd,
115            map,
116            format,
117            n_slots,
118            slot_size,
119            next_slot: 0,
120            slot_seq: vec![0; n_slots as usize],
121            frame_seq: 0,
122            last_published: None,
123        };
124        ring.write_header();
125        Ok(ring)
126    }
127
128    pub fn fd(&self) -> RawFd {
129        self.fd.as_raw_fd()
130    }
131
132    pub fn format(&self) -> &RingFormat {
133        &self.format
134    }
135
136    pub fn n_slots(&self) -> u32 {
137        self.n_slots
138    }
139
140    pub fn slot_size(&self) -> usize {
141        self.slot_size
142    }
143
144    pub fn frame_seq(&self) -> u64 {
145        self.frame_seq
146    }
147
148    /// The bytes of the most recently published frame, valid until the next `publish`.
149    pub fn latest_frame(&self) -> Option<&[u8]> {
150        let (slot, used) = self.last_published?;
151        let data_off = DATA_OFFSET as usize + slot * self.slot_size;
152        Some(&self.map[data_off..data_off + used])
153    }
154
155    /// The configuration struct handed to each interposer client once, ahead of the doorbells.
156    pub fn config_bytes(&self) -> [u8; CONFIG_SIZE] {
157        let f = &self.format;
158        let words = [
159            SHM_MAGIC, SHM_VERSION, f.width, f.height, f.fourcc, f.fps_num, f.fps_den,
160            self.n_slots, self.slot_size as u32, DATA_OFFSET, CTRL_OFFSET, CTRL_STRIDE,
161            f.bytesperline, f.sizeimage,
162        ];
163        let mut out = [0u8; CONFIG_SIZE];
164        for (i, w) in words.iter().enumerate() {
165            out[i * 4..i * 4 + 4].copy_from_slice(&w.to_le_bytes());
166        }
167        out
168    }
169
170    fn write_header(&mut self) {
171        let f = self.format;
172        let words = [
173            SHM_MAGIC, SHM_VERSION, f.width, f.height, f.fourcc, f.fps_num, f.fps_den,
174            self.n_slots, self.slot_size as u32, DATA_OFFSET, f.bytesperline, f.sizeimage,
175            0, 0,
176        ];
177        for (i, w) in words.iter().enumerate() {
178            self.map[i * 4..i * 4 + 4].copy_from_slice(&w.to_le_bytes());
179        }
180        self.map[HDR_LATEST_FRAME_SEQ..HDR_LATEST_FRAME_SEQ + 8].copy_from_slice(&0u64.to_le_bytes());
181    }
182
183    fn atomic_u32(&self, offset: usize) -> &AtomicU32 {
184        unsafe { &*(self.map.as_ptr().add(offset) as *const AtomicU32) }
185    }
186
187    fn atomic_u64(&self, offset: usize) -> &AtomicU64 {
188        unsafe { &*(self.map.as_ptr().add(offset) as *const AtomicU64) }
189    }
190
191    /// Publish one frame: `fill` writes the frame into the next slot and returns the byte count.
192    /// A `fill` returning 0 (or more than the slot holds) abandons the slot without publishing it.
193    pub fn publish<F: FnOnce(&mut [u8]) -> usize>(&mut self, ts_ns: u64, fill: F) -> bool {
194        let slot = self.next_slot as usize;
195        let ctrl = CTRL_OFFSET as usize + slot * CTRL_STRIDE as usize;
196        let data_off = DATA_OFFSET as usize + slot * self.slot_size;
197        let seq = self.slot_seq[slot];
198
199        self.atomic_u32(ctrl).store(seq.wrapping_add(1), Ordering::Release);
200        let slot_size = self.slot_size;
201        let used = fill(&mut self.map[data_off..data_off + slot_size]);
202        if used == 0 || used > slot_size {
203            self.atomic_u32(ctrl).store(seq.wrapping_add(2), Ordering::Release);
204            self.slot_seq[slot] = seq.wrapping_add(2);
205            return false;
206        }
207        self.frame_seq += 1;
208        self.atomic_u32(ctrl + 4).store(used as u32, Ordering::Relaxed);
209        self.atomic_u64(ctrl + 8).store(self.frame_seq, Ordering::Relaxed);
210        self.atomic_u64(ctrl + 16).store(ts_ns, Ordering::Relaxed);
211        self.atomic_u32(ctrl).store(seq.wrapping_add(2), Ordering::Release);
212        self.slot_seq[slot] = seq.wrapping_add(2);
213
214        self.atomic_u32(HDR_LATEST_SLOT).store(slot as u32, Ordering::Release);
215        self.atomic_u64(HDR_LATEST_FRAME_SEQ).store(self.frame_seq, Ordering::Release);
216        self.last_published = Some((slot, used));
217        self.next_slot = (self.next_slot + 1) % self.n_slots;
218        true
219    }
220
221    /// Read back a slot the way a client does, for tests: (bytesused, frame_seq, bytes).
222    #[cfg(test)]
223    pub fn read_latest(&self) -> Option<(u32, u64, Vec<u8>)> {
224        let fseq = self.atomic_u64(HDR_LATEST_FRAME_SEQ).load(Ordering::Acquire);
225        if fseq == 0 {
226            return None;
227        }
228        let slot = self.atomic_u32(HDR_LATEST_SLOT).load(Ordering::Acquire) as usize;
229        let ctrl = CTRL_OFFSET as usize + slot * CTRL_STRIDE as usize;
230        let s1 = self.atomic_u32(ctrl).load(Ordering::Acquire);
231        let used = self.atomic_u32(ctrl + 4).load(Ordering::Relaxed) as usize;
232        let cseq = self.atomic_u64(ctrl + 8).load(Ordering::Relaxed);
233        let data_off = DATA_OFFSET as usize + slot * self.slot_size;
234        let bytes = self.map[data_off..data_off + used].to_vec();
235        let s2 = self.atomic_u32(ctrl).load(Ordering::Acquire);
236        if s1 != s2 || s1 & 1 == 1 {
237            return None;
238        }
239        Some((used as u32, cseq, bytes))
240    }
241}
242
243pub fn page_align(n: usize) -> usize {
244    let page = unsafe { libc::sysconf(libc::_SC_PAGESIZE) };
245    let page = if page > 0 { page as usize } else { 4096 };
246    n.div_ceil(page) * page
247}
248
249#[cfg(test)]
250mod tests {
251    use super::*;
252
253    #[test]
254    fn header_and_config_layout() {
255        let f = RingFormat::raw(V4L2_PIX_FMT_YUV420, 640, 480, 30, 1).unwrap();
256        assert_eq!(f.bytesperline, 640);
257        assert_eq!(f.sizeimage, 640 * 480 * 3 / 2);
258        let ring = Ring::new(f, 3).unwrap();
259        let cfg = ring.config_bytes();
260        let w = |i: usize| u32::from_le_bytes(cfg[i * 4..i * 4 + 4].try_into().unwrap());
261        assert_eq!(w(0), SHM_MAGIC);
262        assert_eq!(w(1), SHM_VERSION);
263        assert_eq!((w(2), w(3)), (640, 480));
264        assert_eq!(w(4), V4L2_PIX_FMT_YUV420);
265        assert_eq!((w(5), w(6)), (30, 1));
266        assert_eq!(w(7), 3);
267        assert_eq!(w(8) as usize, page_align(f.sizeimage as usize));
268        assert_eq!((w(9), w(10), w(11)), (DATA_OFFSET, CTRL_OFFSET, CTRL_STRIDE));
269        assert_eq!((w(12), w(13)), (f.bytesperline, f.sizeimage));
270        let hdr = |i: usize| u32::from_le_bytes(ring.map[i * 4..i * 4 + 4].try_into().unwrap());
271        assert_eq!((hdr(10), hdr(11)), (640, f.sizeimage));
272        assert_eq!(hdr(12), 0);
273        assert!(ring.read_latest().is_none());
274    }
275
276    #[test]
277    fn publish_cycles_slots_and_is_readable() {
278        let f = RingFormat::raw(V4L2_PIX_FMT_YUYV, 16, 8, 30, 1).unwrap();
279        let mut ring = Ring::new(f, 2).unwrap();
280        for i in 1..=5u8 {
281            assert!(ring.publish(i as u64 * 1000, |slot| {
282                slot[..f.sizeimage as usize].fill(i);
283                f.sizeimage as usize
284            }));
285            let (used, seq, bytes) = ring.read_latest().unwrap();
286            assert_eq!(used, f.sizeimage);
287            assert_eq!(seq, i as u64);
288            assert!(bytes.iter().all(|b| *b == i));
289        }
290        assert_eq!(ring.frame_seq(), 5);
291        assert!(!ring.publish(0, |_| 0));
292        assert_eq!(ring.frame_seq(), 5);
293    }
294
295    #[test]
296    fn slot_count_is_clamped() {
297        let f = RingFormat::raw(V4L2_PIX_FMT_NV12, 32, 32, 15, 1).unwrap();
298        assert_eq!(Ring::new(f, 1).unwrap().n_slots(), MIN_SLOTS);
299        assert_eq!(Ring::new(f, 99).unwrap().n_slots(), MAX_SLOTS);
300    }
301}
pixelflux/webcam/ring.rs

pixelflux/webcam/
ring.rs
