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    pub fn bind(path: &str, config: [u8; CONFIG_SIZE], ring_fd: RawFd) -> io::Result<Self> {
46        if let Some(dir) = std::path::Path::new(path).parent()
47            && !dir.as_os_str().is_empty() {
48                std::fs::create_dir_all(dir)?;
49            }
50        let _ = std::fs::remove_file(path);
51        let listener = UnixListener::bind(path)?;
52        listener.set_nonblocking(true)?;
53        let wake_raw = unsafe { libc::eventfd(0, libc::EFD_CLOEXEC | libc::EFD_NONBLOCK) };
54        if wake_raw < 0 {
55            return Err(io::Error::last_os_error());
56        }
57        let wake = unsafe { OwnedFd::from_raw_fd(wake_raw) };
58        let shared = Arc::new(Shared {
59            clients: Mutex::new(Vec::new()),
60            ready_count: AtomicUsize::new(0),
61            stop: AtomicBool::new(false),
62        });
63        let t_shared = shared.clone();
64        let t_wake = wake.as_raw_fd();
65        let thread = thread::Builder::new()
66            .name("pixelflux-webcam-ctl".into())
67            .spawn(move || serve(listener, t_wake, config, ring_fd, t_shared))?;
68        Ok(Server { path: path.to_string(), shared, wake, thread: Some(thread) })
69    }
70
71    pub fn path(&self) -> &str {
72        &self.path
73    }
74
75    /// Number of interposer clients that completed the handshake.
76    pub fn client_count(&self) -> usize {
77        self.shared.ready_count.load(Ordering::Relaxed)
78    }
79
80    /// Wake every ready client: one byte per published frame. A full socket buffer means a
81    /// wakeup is already pending, so `EAGAIN` is not an error; a dead peer is retired here.
82    pub fn ring_doorbell(&self) {
83        let mut clients = self.shared.clients.lock().unwrap_or_else(|e| e.into_inner());
84        let mut i = 0;
85        while i < clients.len() {
86            if !clients[i].ready {
87                i += 1;
88                continue;
89            }
90            let byte = [1u8];
91            let n = unsafe {
92                libc::send(clients[i].fd.as_raw_fd(), byte.as_ptr() as *const _, 1,
93                           libc::MSG_DONTWAIT | libc::MSG_NOSIGNAL)
94            };
95            if n < 0 {
96                let err = io::Error::last_os_error();
97                match err.raw_os_error() {
98                    Some(libc::EAGAIN) | Some(libc::EINTR) => {}
99                    _ => {
100                        clients.remove(i);
101                        self.shared.ready_count.fetch_sub(1, Ordering::Relaxed);
102                        continue;
103                    }
104                }
105            }
106            i += 1;
107        }
108    }
109
110    fn wake_thread(&self) {
111        let one: u64 = 1;
112        unsafe { libc::write(self.wake.as_raw_fd(), &one as *const u64 as *const _, 8) };
113    }
114}
115
116impl Drop for Server {
117    /// Stop accepting, close every client (the interposer sees EOF and reports `ENODEV`) and remove
118    /// the socket file.
119    fn drop(&mut self) {
120        self.shared.stop.store(true, Ordering::Release);
121        self.wake_thread();
122        if let Some(t) = self.thread.take() {
123            let _ = t.join();
124        }
125        self.shared.clients.lock().unwrap_or_else(|e| e.into_inner()).clear();
126        self.shared.ready_count.store(0, Ordering::Relaxed);
127        let _ = std::fs::remove_file(&self.path);
128    }
129}
130
131/// Send the configuration struct with the ring memfd attached as `SCM_RIGHTS`.
132fn send_config_with_fd(sock: RawFd, config: &[u8], fd: RawFd) -> io::Result<()> {
133    let mut iov = libc::iovec { iov_base: config.as_ptr() as *mut libc::c_void, iov_len: config.len() };
134    let space = unsafe { libc::CMSG_SPACE(mem::size_of::<RawFd>() as u32) } as usize;
135    let mut cbuf = vec![0u64; space.div_ceil(mem::size_of::<u64>())];
136    let mut msg: libc::msghdr = unsafe { mem::zeroed() };
137    msg.msg_iov = &mut iov;
138    msg.msg_iovlen = 1;
139    msg.msg_control = cbuf.as_mut_ptr() as *mut libc::c_void;
140    msg.msg_controllen = space as _;
141    unsafe {
142        let cmsg = libc::CMSG_FIRSTHDR(&msg);
143        (*cmsg).cmsg_level = libc::SOL_SOCKET;
144        (*cmsg).cmsg_type = libc::SCM_RIGHTS;
145        (*cmsg).cmsg_len = libc::CMSG_LEN(mem::size_of::<RawFd>() as u32) as _;
146        ptr::write_unaligned(libc::CMSG_DATA(cmsg) as *mut RawFd, fd);
147    }
148    loop {
149        let n = unsafe { libc::sendmsg(sock, &msg, libc::MSG_NOSIGNAL) };
150        if n < 0 {
151            let err = io::Error::last_os_error();
152            if err.raw_os_error() == Some(libc::EINTR) {
153                continue;
154            }
155            return Err(err);
156        }
157        if n as usize != config.len() {
158            return Err(io::Error::new(io::ErrorKind::WriteZero, "short config send"));
159        }
160        return Ok(());
161    }
162}
163
164fn serve(listener: UnixListener, wake: RawFd, config: [u8; CONFIG_SIZE], ring_fd: RawFd, shared: Arc<Shared>) {
165    let listen_fd = listener.as_raw_fd();
166    let mut pollfds: Vec<libc::pollfd> = Vec::new();
167    while !shared.stop.load(Ordering::Acquire) {
168        pollfds.clear();
169        pollfds.push(libc::pollfd { fd: listen_fd, events: libc::POLLIN, revents: 0 });
170        pollfds.push(libc::pollfd { fd: wake, events: libc::POLLIN, revents: 0 });
171        {
172            let clients = shared.clients.lock().unwrap_or_else(|e| e.into_inner());
173            for c in clients.iter() {
174                pollfds.push(libc::pollfd { fd: c.fd.as_raw_fd(), events: libc::POLLIN | libc::POLLRDHUP, revents: 0 });
175            }
176        }
177        let n = unsafe { libc::poll(pollfds.as_mut_ptr(), pollfds.len() as libc::nfds_t, -1) };
178        if n < 0 {
179            if io::Error::last_os_error().raw_os_error() == Some(libc::EINTR) {
180                continue;
181            }
182            break;
183        }
184        if pollfds[1].revents != 0 {
185            let mut v: u64 = 0;
186            unsafe { libc::read(wake, &mut v as *mut u64 as *mut _, 8) };
187            if shared.stop.load(Ordering::Acquire) {
188                break;
189            }
190        }
191        if pollfds[0].revents != 0 {
192            accept_clients(&listener, &config, ring_fd, &shared);
193        }
194        let mut clients = shared.clients.lock().unwrap_or_else(|e| e.into_inner());
195        let mut idx = 0;
196        for pfd in pollfds.iter().skip(2) {
197            if idx >= clients.len() {
198                break;
199            }
200            if clients[idx].fd.as_raw_fd() != pfd.fd {
201                idx += 1;
202                continue;
203            }
204            if pfd.revents == 0 {
205                idx += 1;
206                continue;
207            }
208            let mut buf = [0u8; 64];
209            let r = unsafe { libc::recv(clients[idx].fd.as_raw_fd(), buf.as_mut_ptr() as *mut _, buf.len(), libc::MSG_DONTWAIT) };
210            let closed = if r > 0 {
211                if !clients[idx].ready {
212                    clients[idx].ready = true;
213                    shared.ready_count.fetch_add(1, Ordering::Relaxed);
214                }
215                false
216            } else if r == 0 {
217                true
218            } else {
219                !matches!(io::Error::last_os_error().raw_os_error(), Some(libc::EAGAIN) | Some(libc::EINTR))
220            };
221            if closed || pfd.revents & (libc::POLLHUP | libc::POLLERR | libc::POLLRDHUP) != 0 && r <= 0 {
222                if clients[idx].ready {
223                    shared.ready_count.fetch_sub(1, Ordering::Relaxed);
224                }
225                clients.remove(idx);
226                continue;
227            }
228            idx += 1;
229        }
230    }
231}
232
233fn accept_clients(listener: &UnixListener, config: &[u8; CONFIG_SIZE], ring_fd: RawFd, shared: &Shared) {
234    loop {
235        match listener.accept() {
236            Ok((stream, _)) => {
237                let _ = stream.set_nonblocking(true);
238                let fd = OwnedFd::from(stream);
239                match send_config_with_fd(fd.as_raw_fd(), config, ring_fd) {
240                    Ok(()) => {
241                        shared.clients.lock().unwrap_or_else(|e| e.into_inner()).push(Client { fd, ready: false });
242                    }
243                    Err(e) => eprintln!("[webcam] interposer handshake failed: {}", e),
244                }
245            }
246            Err(e) if e.kind() == io::ErrorKind::WouldBlock => return,
247            Err(e) if e.kind() == io::ErrorKind::Interrupted => continue,
248            Err(e) => {
249                eprintln!("[webcam] accept failed: {}", e);
250                return;
251            }
252        }
253    }
254}
255
256#[cfg(test)]
257mod tests {
258    use super::*;
259    use std::io::{Read, Write};
260    use std::os::unix::net::UnixStream;
261    use std::time::{Duration, Instant};
262
263    fn recv_config(stream: &UnixStream) -> (Vec<u8>, RawFd) {
264        let mut data = vec![0u8; CONFIG_SIZE];
265        let mut iov = libc::iovec { iov_base: data.as_mut_ptr() as *mut _, iov_len: data.len() };
266        let space = unsafe { libc::CMSG_SPACE(4) } as usize;
267        let mut cbuf = vec![0u64; space.div_ceil(8)];
268        let mut msg: libc::msghdr = unsafe { mem::zeroed() };
269        msg.msg_iov = &mut iov;
270        msg.msg_iovlen = 1;
271        msg.msg_control = cbuf.as_mut_ptr() as *mut _;
272        msg.msg_controllen = space as _;
273        let n = unsafe { libc::recvmsg(stream.as_raw_fd(), &mut msg, 0) };
274        assert_eq!(n as usize, CONFIG_SIZE);
275        let cmsg = unsafe { libc::CMSG_FIRSTHDR(&msg) };
276        assert!(!cmsg.is_null());
277        let fd = unsafe { ptr::read_unaligned(libc::CMSG_DATA(cmsg) as *const RawFd) };
278        (data, fd)
279    }
280
281    #[test]
282    fn handshake_fd_and_doorbells() {
283        let dir = std::env::temp_dir().join(format!("pxwc-{}", std::process::id()));
284        let path = dir.join("cam.sock");
285        let ring = unsafe { libc::memfd_create(c"t".as_ptr(), libc::MFD_CLOEXEC) };
286        assert!(ring >= 0);
287        unsafe { libc::ftruncate(ring, 8192) };
288        let mut config = [0u8; CONFIG_SIZE];
289        config[..4].copy_from_slice(&0xDEADBEEFu32.to_le_bytes());
290        let server = Server::bind(path.to_str().unwrap(), config, ring).unwrap();
291
292        let mut client = UnixStream::connect(&path).unwrap();
293        let (data, fd) = recv_config(&client);
294        assert_eq!(&data[..4], &0xDEADBEEFu32.to_le_bytes());
295        assert!(fd >= 0);
296        let mut st: libc::stat = unsafe { mem::zeroed() };
297        assert_eq!(unsafe { libc::fstat(fd, &mut st) }, 0);
298        assert_eq!(st.st_size, 8192);
299        unsafe { libc::close(fd) };
300
301        server.ring_doorbell();
302        assert_eq!(server.client_count(), 0);
303        client.write_all(&[8u8]).unwrap();
304        let deadline = Instant::now() + Duration::from_secs(3);
305        while server.client_count() != 1 && Instant::now() < deadline {
306            thread::sleep(Duration::from_millis(5));
307        }
308        assert_eq!(server.client_count(), 1);
309        server.ring_doorbell();
310        client.set_read_timeout(Some(Duration::from_secs(2))).unwrap();
311        let mut b = [0u8; 1];
312        client.read_exact(&mut b).unwrap();
313        assert_eq!(b[0], 1);
314
315        drop(client);
316        let deadline = Instant::now() + Duration::from_secs(3);
317        while server.client_count() != 0 && Instant::now() < deadline {
318            server.ring_doorbell();
319            thread::sleep(Duration::from_millis(5));
320        }
321        assert_eq!(server.client_count(), 0);
322        drop(server);
323        assert!(!path.exists());
324        unsafe { libc::close(ring) };
325        let _ = std::fs::remove_dir_all(&dir);
326    }
327}
pixelflux/webcam/server.rs

pixelflux/webcam/
server.rs
