pixelflux/webcam/
server.rs

1//! Control socket for the Selkies V4L2 interposer.
2//!
3//! Each application `open()` of the virtual device becomes one client connection. The server sends
4//! the ring configuration once, with the ring memfd attached as `SCM_RIGHTS` ancillary data, then
5//! only ever writes one-byte doorbells — one per published frame — so the interposer can block in
6//! `poll()`/`VIDIOC_DQBUF` on plain socket readability. Frame bytes never cross the socket. The
7//! interposer answers the handshake with a single byte; until that arrives a client gets no
8//! doorbells, and a closed connection is retired the moment its read end reports hangup.
9//!
10//! The accept/hangup loop runs on its own thread, woken through an eventfd for shutdown, so a
11//! process that opens the device while no frames flow is still served immediately.
12
13use std::io;
14use std::mem;
15use std::os::fd::{AsRawFd, FromRawFd, OwnedFd, RawFd};
16use std::os::unix::net::UnixListener;
17use std::ptr;
18use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
19use std::sync::{Arc, Mutex};
20use std::thread::{self, JoinHandle};
21
22use super::ring::CONFIG_SIZE;
23
24struct Client {
25    fd: OwnedFd,
26    /// Set once the interposer's handshake byte arrived; only ready clients receive doorbells.
27    ready: bool,
28}
29
30struct Shared {
31    clients: Mutex<Vec<Client>>,
32    ready_count: AtomicUsize,
33    stop: AtomicBool,
34}
35
36pub struct Server {
37    path: String,
38    shared: Arc<Shared>,
39    wake: OwnedFd,
40    thread: Option<JoinHandle<()>>,
41}
42
43impl Server {
44    /// Bind `path` (replacing a stale socket file) and start serving `config` + `ring_fd`.
45    ///
46    /// A file there that another account owns is refused: in a shared directory it is that
47    /// account's listener, and every camera client of this session would reach it.
48    pub fn bind(path: &str, config: [u8; CONFIG_SIZE], ring_fd: RawFd) -> io::Result<Self> {
49        use std::os::unix::fs::MetadataExt;
50        if let Some(dir) = std::path::Path::new(path).parent()
51            && !dir.as_os_str().is_empty()
52        {
53            std::fs::create_dir_all(dir)?;
54        }
55        match std::fs::symlink_metadata(path) {
56            Ok(meta) if meta.uid() != unsafe { libc::geteuid() } => {
57                return Err(io::Error::new(
58                    io::ErrorKind::PermissionDenied,
59                    format!("{path} belongs to uid {}, not this session", meta.uid()),
60                ));
61            }
62            Ok(_) => std::fs::remove_file(path)?,
63            Err(e) if e.kind() == io::ErrorKind::NotFound => {}
64            Err(e) => return Err(e),
65        }
66        let listener = UnixListener::bind(path)?;
67        listener.set_nonblocking(true)?;
68        let wake_raw = unsafe { libc::eventfd(0, libc::EFD_CLOEXEC | libc::EFD_NONBLOCK) };
69        if wake_raw < 0 {
70            return Err(io::Error::last_os_error());
71        }
72        let wake = unsafe { OwnedFd::from_raw_fd(wake_raw) };
73        let shared = Arc::new(Shared {
74            clients: Mutex::new(Vec::new()),
75            ready_count: AtomicUsize::new(0),
76            stop: AtomicBool::new(false),
77        });
78        let t_shared = shared.clone();
79        let t_wake = wake.as_raw_fd();
80        let thread = thread::Builder::new()
81            .name("pixelflux-webcam-ctl".into())
82            .spawn(move || serve(listener, t_wake, config, ring_fd, t_shared))?;
83        Ok(Server {
84            path: path.to_string(),
85            shared,
86            wake,
87            thread: Some(thread),
88        })
89    }
90
91    pub fn path(&self) -> &str {
92        &self.path
93    }
94
95    /// Number of interposer clients that completed the handshake.
96    pub fn client_count(&self) -> usize {
97        self.shared.ready_count.load(Ordering::Relaxed)
98    }
99
100    /// Wake every ready client: one byte per published frame. A full socket buffer means a
101    /// wakeup is already pending, so `EAGAIN` is not an error; a dead peer is retired here.
102    pub fn ring_doorbell(&self) {
103        let mut clients = self
104            .shared
105            .clients
106            .lock()
107            .unwrap_or_else(|e| e.into_inner());
108        let mut i = 0;
109        while i < clients.len() {
110            if !clients[i].ready {
111                i += 1;
112                continue;
113            }
114            let byte = [1u8];
115            let n = unsafe {
116                libc::send(
117                    clients[i].fd.as_raw_fd(),
118                    byte.as_ptr() as *const _,
119                    1,
120                    libc::MSG_DONTWAIT | libc::MSG_NOSIGNAL,
121                )
122            };
123            if n < 0 {
124                let err = io::Error::last_os_error();
125                match err.raw_os_error() {
126                    Some(libc::EAGAIN) | Some(libc::EINTR) => {}
127                    _ => {
128                        clients.remove(i);
129                        self.shared.ready_count.fetch_sub(1, Ordering::Relaxed);
130                        continue;
131                    }
132                }
133            }
134            i += 1;
135        }
136    }
137
138    fn wake_thread(&self) {
139        let one: u64 = 1;
140        unsafe { libc::write(self.wake.as_raw_fd(), &one as *const u64 as *const _, 8) };
141    }
142}
143
144impl Drop for Server {
145    /// Stop accepting, close every client (the interposer sees EOF and reports `ENODEV`), and remove
146    /// the socket file.
147    fn drop(&mut self) {
148        self.shared.stop.store(true, Ordering::Release);
149        self.wake_thread();
150        if let Some(t) = self.thread.take() {
151            let _ = t.join();
152        }
153        self.shared
154            .clients
155            .lock()
156            .unwrap_or_else(|e| e.into_inner())
157            .clear();
158        self.shared.ready_count.store(0, Ordering::Relaxed);
159        let _ = std::fs::remove_file(&self.path);
160    }
161}
162
163/// Send the configuration struct with the ring memfd attached as `SCM_RIGHTS`.
164fn send_config_with_fd(sock: RawFd, config: &[u8], fd: RawFd) -> io::Result<()> {
165    let mut iov = libc::iovec {
166        iov_base: config.as_ptr() as *mut libc::c_void,
167        iov_len: config.len(),
168    };
169    let space = unsafe { libc::CMSG_SPACE(mem::size_of::<RawFd>() as u32) } as usize;
170    let mut cbuf = vec![0u64; space.div_ceil(mem::size_of::<u64>())];
171    let mut msg: libc::msghdr = unsafe { mem::zeroed() };
172    msg.msg_iov = &mut iov;
173    msg.msg_iovlen = 1;
174    msg.msg_control = cbuf.as_mut_ptr() as *mut libc::c_void;
175    msg.msg_controllen = space as _;
176    unsafe {
177        let cmsg = libc::CMSG_FIRSTHDR(&msg);
178        (*cmsg).cmsg_level = libc::SOL_SOCKET;
179        (*cmsg).cmsg_type = libc::SCM_RIGHTS;
180        (*cmsg).cmsg_len = libc::CMSG_LEN(mem::size_of::<RawFd>() as u32) as _;
181        ptr::write_unaligned(libc::CMSG_DATA(cmsg) as *mut RawFd, fd);
182    }
183    loop {
184        let n = unsafe { libc::sendmsg(sock, &msg, libc::MSG_NOSIGNAL) };
185        if n < 0 {
186            let err = io::Error::last_os_error();
187            if err.raw_os_error() == Some(libc::EINTR) {
188                continue;
189            }
190            return Err(err);
191        }
192        if n as usize != config.len() {
193            return Err(io::Error::new(
194                io::ErrorKind::WriteZero,
195                "short config send",
196            ));
197        }
198        return Ok(());
199    }
200}
201
202fn serve(
203    listener: UnixListener,
204    wake: RawFd,
205    config: [u8; CONFIG_SIZE],
206    ring_fd: RawFd,
207    shared: Arc<Shared>,
208) {
209    let listen_fd = listener.as_raw_fd();
210    let mut pollfds: Vec<libc::pollfd> = Vec::new();
211    while !shared.stop.load(Ordering::Acquire) {
212        pollfds.clear();
213        pollfds.push(libc::pollfd {
214            fd: listen_fd,
215            events: libc::POLLIN,
216            revents: 0,
217        });
218        pollfds.push(libc::pollfd {
219            fd: wake,
220            events: libc::POLLIN,
221            revents: 0,
222        });
223        {
224            let clients = shared.clients.lock().unwrap_or_else(|e| e.into_inner());
225            for c in clients.iter() {
226                pollfds.push(libc::pollfd {
227                    fd: c.fd.as_raw_fd(),
228                    events: libc::POLLIN | libc::POLLRDHUP,
229                    revents: 0,
230                });
231            }
232        }
233        let n = unsafe { libc::poll(pollfds.as_mut_ptr(), pollfds.len() as libc::nfds_t, -1) };
234        if n < 0 {
235            if io::Error::last_os_error().raw_os_error() == Some(libc::EINTR) {
236                continue;
237            }
238            break;
239        }
240        if pollfds[1].revents != 0 {
241            let mut v: u64 = 0;
242            unsafe { libc::read(wake, &mut v as *mut u64 as *mut _, 8) };
243            if shared.stop.load(Ordering::Acquire) {
244                break;
245            }
246        }
247        if pollfds[0].revents != 0 {
248            accept_clients(&listener, &config, ring_fd, &shared);
249        }
250        let mut clients = shared.clients.lock().unwrap_or_else(|e| e.into_inner());
251        let mut idx = 0;
252        for pfd in pollfds.iter().skip(2) {
253            if idx >= clients.len() {
254                break;
255            }
256            if clients[idx].fd.as_raw_fd() != pfd.fd {
257                idx += 1;
258                continue;
259            }
260            if pfd.revents == 0 {
261                idx += 1;
262                continue;
263            }
264            let mut buf = [0u8; 64];
265            let r = unsafe {
266                libc::recv(
267                    clients[idx].fd.as_raw_fd(),
268                    buf.as_mut_ptr() as *mut _,
269                    buf.len(),
270                    libc::MSG_DONTWAIT,
271                )
272            };
273            let closed = if r > 0 {
274                if !clients[idx].ready {
275                    clients[idx].ready = true;
276                    shared.ready_count.fetch_add(1, Ordering::Relaxed);
277                }
278                false
279            } else if r == 0 {
280                true
281            } else {
282                !matches!(
283                    io::Error::last_os_error().raw_os_error(),
284                    Some(libc::EAGAIN) | Some(libc::EINTR)
285                )
286            };
287            if closed
288                || pfd.revents & (libc::POLLHUP | libc::POLLERR | libc::POLLRDHUP) != 0 && r <= 0
289            {
290                if clients[idx].ready {
291                    shared.ready_count.fetch_sub(1, Ordering::Relaxed);
292                }
293                clients.remove(idx);
294                continue;
295            }
296            idx += 1;
297        }
298    }
299}
300
301fn accept_clients(
302    listener: &UnixListener,
303    config: &[u8; CONFIG_SIZE],
304    ring_fd: RawFd,
305    shared: &Shared,
306) {
307    loop {
308        match listener.accept() {
309            Ok((stream, _)) => {
310                let _ = stream.set_nonblocking(true);
311                let fd = OwnedFd::from(stream);
312                match send_config_with_fd(fd.as_raw_fd(), config, ring_fd) {
313                    Ok(()) => {
314                        shared
315                            .clients
316                            .lock()
317                            .unwrap_or_else(|e| e.into_inner())
318                            .push(Client { fd, ready: false });
319                    }
320                    Err(e) => eprintln!("[webcam] interposer handshake failed: {}", e),
321                }
322            }
323            Err(e) if e.kind() == io::ErrorKind::WouldBlock => return,
324            Err(e) if e.kind() == io::ErrorKind::Interrupted => continue,
325            Err(e) => {
326                eprintln!("[webcam] accept failed: {}", e);
327                return;
328            }
329        }
330    }
331}
332
333#[cfg(test)]
334mod tests {
335    use super::*;
336    use std::io::{Read, Write};
337    use std::os::unix::net::UnixStream;
338    use std::time::{Duration, Instant};
339
340    fn recv_config(stream: &UnixStream) -> (Vec<u8>, RawFd) {
341        let mut data = vec![0u8; CONFIG_SIZE];
342        let mut iov = libc::iovec {
343            iov_base: data.as_mut_ptr() as *mut _,
344            iov_len: data.len(),
345        };
346        let space = unsafe { libc::CMSG_SPACE(4) } as usize;
347        let mut cbuf = vec![0u64; space.div_ceil(8)];
348        let mut msg: libc::msghdr = unsafe { mem::zeroed() };
349        msg.msg_iov = &mut iov;
350        msg.msg_iovlen = 1;
351        msg.msg_control = cbuf.as_mut_ptr() as *mut _;
352        msg.msg_controllen = space as _;
353        let n = unsafe { libc::recvmsg(stream.as_raw_fd(), &mut msg, 0) };
354        assert_eq!(n as usize, CONFIG_SIZE);
355        let cmsg = unsafe { libc::CMSG_FIRSTHDR(&msg) };
356        assert!(!cmsg.is_null());
357        let fd = unsafe { ptr::read_unaligned(libc::CMSG_DATA(cmsg) as *const RawFd) };
358        (data, fd)
359    }
360
361    #[test]
362    fn handshake_fd_and_doorbells() {
363        let dir = std::env::temp_dir().join(format!("pxwc-{}", std::process::id()));
364        let path = dir.join("cam.sock");
365        let ring = unsafe { libc::memfd_create(c"t".as_ptr(), libc::MFD_CLOEXEC) };
366        assert!(ring >= 0);
367        unsafe { libc::ftruncate(ring, 8192) };
368        let mut config = [0u8; CONFIG_SIZE];
369        config[..4].copy_from_slice(&0xDEADBEEFu32.to_le_bytes());
370        let server = Server::bind(path.to_str().unwrap(), config, ring).unwrap();
371
372        let mut client = UnixStream::connect(&path).unwrap();
373        let (data, fd) = recv_config(&client);
374        assert_eq!(&data[..4], &0xDEADBEEFu32.to_le_bytes());
375        assert!(fd >= 0);
376        let mut st: libc::stat = unsafe { mem::zeroed() };
377        assert_eq!(unsafe { libc::fstat(fd, &mut st) }, 0);
378        assert_eq!(st.st_size, 8192);
379        unsafe { libc::close(fd) };
380
381        server.ring_doorbell();
382        assert_eq!(server.client_count(), 0);
383        client.write_all(&[8u8]).unwrap();
384        let deadline = Instant::now() + Duration::from_secs(3);
385        while server.client_count() != 1 && Instant::now() < deadline {
386            thread::sleep(Duration::from_millis(5));
387        }
388        assert_eq!(server.client_count(), 1);
389        server.ring_doorbell();
390        client
391            .set_read_timeout(Some(Duration::from_secs(2)))
392            .unwrap();
393        let mut b = [0u8; 1];
394        client.read_exact(&mut b).unwrap();
395        assert_eq!(b[0], 1);
396
397        drop(client);
398        let deadline = Instant::now() + Duration::from_secs(3);
399        while server.client_count() != 0 && Instant::now() < deadline {
400            server.ring_doorbell();
401            thread::sleep(Duration::from_millis(5));
402        }
403        assert_eq!(server.client_count(), 0);
404        drop(server);
405        assert!(!path.exists());
406        unsafe { libc::close(ring) };
407        let _ = std::fs::remove_dir_all(&dir);
408    }
409}
pixelflux/webcam/server.rs

pixelflux/webcam/
server.rs
