chore(API): Infrastructure to use endpoints with fd. passing

util/src/mio/mio.rs

@@ -0,0 +1,39 @@
+use mio::net::{UnixListener, UnixStream};
+use rustix::fd::RawFd;
+
+use crate::fd::claim_fd;
+
+pub mod interest {
+    use mio::Interest;
+    pub const R: Interest = Interest::READABLE;
+    pub const W: Interest = Interest::WRITABLE;
+    pub const RW: Interest = R.add(W);
+}
+
+pub trait UnixListenerExt: Sized {
+    fn claim_fd(fd: RawFd) -> anyhow::Result<Self>;
+}
+
+impl UnixListenerExt for UnixListener {
+    fn claim_fd(fd: RawFd) -> anyhow::Result<Self> {
+        use std::os::unix::net::UnixListener as StdUnixListener;
+
+        let sock = StdUnixListener::from(claim_fd(fd)?);
+        sock.set_nonblocking(true)?;
+        Ok(UnixListener::from_std(sock))
+    }
+}
+
+pub trait UnixStreamExt: Sized {
+    fn claim_fd(fd: RawFd) -> anyhow::Result<Self>;
+}
+
+impl UnixStreamExt for UnixStream {
+    fn claim_fd(fd: RawFd) -> anyhow::Result<Self> {
+        use std::os::unix::net::UnixStream as StdUnixStream;
+
+        let sock = StdUnixStream::from(claim_fd(fd)?);
+        sock.set_nonblocking(true)?;
+        Ok(UnixStream::from_std(sock))
+    }
+}

util/src/mio/mod.rs

@@ -0,0 +1,12 @@
+#[allow(clippy::module_inception)]
+mod mio;
+pub use mio::*;
+
+#[cfg(feature = "experiment_file_descriptor_passing")]
+mod uds_recv_fd;
+#[cfg(feature = "experiment_file_descriptor_passing")]
+mod uds_send_fd;
+#[cfg(feature = "experiment_file_descriptor_passing")]
+pub use uds_recv_fd::*;
+#[cfg(feature = "experiment_file_descriptor_passing")]
+pub use uds_send_fd::*;

util/src/mio/uds_recv_fd.rs

@@ -0,0 +1,123 @@
+use std::{
+    borrow::{Borrow, BorrowMut},
+    collections::VecDeque,
+    io::Read,
+    marker::PhantomData,
+    os::fd::OwnedFd,
+};
+use uds::UnixStreamExt as FdPassingExt;
+
+use crate::fd::{claim_fd_inplace, IntoStdioErr};
+
+pub struct ReadWithFileDescriptors<const MAX_FDS: usize, Sock, BorrowSock, BorrowFds>
+where
+    Sock: FdPassingExt,
+    BorrowSock: Borrow<Sock>,
+    BorrowFds: BorrowMut<VecDeque<OwnedFd>>,
+{
+    socket: BorrowSock,
+    fds: BorrowFds,
+    _sock_dummy: PhantomData<Sock>,
+}
+
+impl<const MAX_FDS: usize, Sock, BorrowSock, BorrowFds>
+    ReadWithFileDescriptors<MAX_FDS, Sock, BorrowSock, BorrowFds>
+where
+    Sock: FdPassingExt,
+    BorrowSock: Borrow<Sock>,
+    BorrowFds: BorrowMut<VecDeque<OwnedFd>>,
+{
+    pub fn new(socket: BorrowSock, fds: BorrowFds) -> Self {
+        let _sock_dummy = PhantomData;
+        Self {
+            socket,
+            fds,
+            _sock_dummy,
+        }
+    }
+
+    pub fn into_parts(self) -> (BorrowSock, BorrowFds) {
+        let Self { socket, fds, .. } = self;
+        (socket, fds)
+    }
+
+    pub fn socket(&self) -> &Sock {
+        self.socket.borrow()
+    }
+
+    pub fn fds(&self) -> &VecDeque<OwnedFd> {
+        self.fds.borrow()
+    }
+
+    pub fn fds_mut(&mut self) -> &mut VecDeque<OwnedFd> {
+        self.fds.borrow_mut()
+    }
+}
+
+impl<const MAX_FDS: usize, Sock, BorrowSock, BorrowFds>
+    ReadWithFileDescriptors<MAX_FDS, Sock, BorrowSock, BorrowFds>
+where
+    Sock: FdPassingExt,
+    BorrowSock: BorrowMut<Sock>,
+    BorrowFds: BorrowMut<VecDeque<OwnedFd>>,
+{
+    pub fn socket_mut(&mut self) -> &mut Sock {
+        self.socket.borrow_mut()
+    }
+}
+
+impl<const MAX_FDS: usize, Sock, BorrowSock, BorrowFds> Read
+    for ReadWithFileDescriptors<MAX_FDS, Sock, BorrowSock, BorrowFds>
+where
+    Sock: FdPassingExt,
+    BorrowSock: Borrow<Sock>,
+    BorrowFds: BorrowMut<VecDeque<OwnedFd>>,
+{
+    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
+        // Calculate space for additional file descriptors
+        let have_fds_before_read = self.fds().len();
+        let free_fd_slots = MAX_FDS.saturating_sub(have_fds_before_read);
+
+        // Allocate a buffer for file descriptors
+        let mut fd_buf = [0; MAX_FDS];
+        let fd_buf = &mut fd_buf[..free_fd_slots];
+
+        // Read from the unix socket
+        let (bytes_read, fds_read) = self.socket.borrow().recv_fds(buf, fd_buf)?;
+        let fd_buf = &fd_buf[..fds_read];
+
+        // Process the file descriptors
+        let mut fd_iter = fd_buf.iter();
+
+        // Try claiming all the file descriptors
+        let mut claim_fd_result = Ok(bytes_read);
+        self.fds_mut().reserve(fd_buf.len());
+        for fd in fd_iter.by_ref() {
+            match claim_fd_inplace(*fd) {
+                Ok(owned) => self.fds_mut().push_back(owned),
+                Err(e) => {
+                    // Abort on error and pass to error handler
+                    // Note that claim_fd_inplace is responsible for closing this particular
+                    // file descriptor if claiming it fails
+                    claim_fd_result = Err(e.into_stdio_err());
+                    break;
+                }
+            }
+        }
+
+        // Return if we where able to claim all file descriptors
+        if claim_fd_result.is_ok() {
+            return claim_fd_result;
+        };
+
+        // An error occurred while claiming fds
+        self.fds_mut().truncate(have_fds_before_read); // Close fds successfully claimed
+
+        // Close the remaining fds
+        for fd in fd_iter {
+            unsafe { rustix::io::close(*fd) };
+        }
+
+        claim_fd_result
+    }
+}

util/src/mio/uds_send_fd.rs

@@ -0,0 +1,121 @@
+use rustix::fd::{AsFd, AsRawFd};
+use std::{
+    borrow::{Borrow, BorrowMut},
+    cmp::min,
+    collections::VecDeque,
+    io::Write,
+    marker::PhantomData,
+};
+use uds::UnixStreamExt as FdPassingExt;
+
+use crate::{repeat, return_if};
+
+pub struct WriteWithFileDescriptors<Sock, Fd, BorrowSock, BorrowFds>
+where
+    Sock: FdPassingExt,
+    Fd: AsFd,
+    BorrowSock: Borrow<Sock>,
+    BorrowFds: BorrowMut<VecDeque<Fd>>,
+{
+    socket: BorrowSock,
+    fds: BorrowFds,
+    _sock_dummy: PhantomData<Sock>,
+    _fd_dummy: PhantomData<Fd>,
+}
+
+impl<Sock, Fd, BorrowSock, BorrowFds> WriteWithFileDescriptors<Sock, Fd, BorrowSock, BorrowFds>
+where
+    Sock: FdPassingExt,
+    Fd: AsFd,
+    BorrowSock: Borrow<Sock>,
+    BorrowFds: BorrowMut<VecDeque<Fd>>,
+{
+    pub fn new(socket: BorrowSock, fds: BorrowFds) -> Self {
+        let _sock_dummy = PhantomData;
+        let _fd_dummy = PhantomData;
+        Self {
+            socket,
+            fds,
+            _sock_dummy,
+            _fd_dummy,
+        }
+    }
+
+    pub fn into_parts(self) -> (BorrowSock, BorrowFds) {
+        let Self { socket, fds, .. } = self;
+        (socket, fds)
+    }
+
+    pub fn socket(&self) -> &Sock {
+        self.socket.borrow()
+    }
+
+    pub fn fds(&self) -> &VecDeque<Fd> {
+        self.fds.borrow()
+    }
+
+    pub fn fds_mut(&mut self) -> &mut VecDeque<Fd> {
+        self.fds.borrow_mut()
+    }
+}
+
+impl<Sock, Fd, BorrowSock, BorrowFds> WriteWithFileDescriptors<Sock, Fd, BorrowSock, BorrowFds>
+where
+    Sock: FdPassingExt,
+    Fd: AsFd,
+    BorrowSock: BorrowMut<Sock>,
+    BorrowFds: BorrowMut<VecDeque<Fd>>,
+{
+    pub fn socket_mut(&mut self) -> &mut Sock {
+        self.socket.borrow_mut()
+    }
+}
+
+impl<Sock, Fd, BorrowSock, BorrowFds> Write
+    for WriteWithFileDescriptors<Sock, Fd, BorrowSock, BorrowFds>
+where
+    Sock: FdPassingExt,
+    Fd: AsFd,
+    BorrowSock: Borrow<Sock>,
+    BorrowFds: BorrowMut<VecDeque<Fd>>,
+{
+    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+        // At least one byte of real data should be sent when sending ancillary data. -- unix(7)
+        return_if!(buf.is_empty(), Ok(0));
+
+        // The kernel constant SCM_MAX_FD defines a limit on the number of file descriptors
+        // in the array.  Attempting to  send  an  array  larger  than  this  limit  causes
+        // sendmsg(2)  to fail with the error EINVAL.  SCM_MAX_FD has the value 253 (or 255
+        // before Linux 2.6.38).
+        // -- unix(7)
+        const SCM_MAX_FD: usize = 253;
+        let buf = match self.fds().len() <= SCM_MAX_FD {
+            false => &buf[..1], // Force caller to immediately call write() again to send its data
+            true => buf,
+        };
+
+        // Allocate the buffer for the file descriptor array
+        let fd_no = min(SCM_MAX_FD, self.fds().len());
+        let mut fd_buf = [0; SCM_MAX_FD]; // My kingdom for alloca(3)
+        let fd_buf = &mut fd_buf[..fd_no];
+
+        // Fill the file descriptor array
+        for (raw, fancy) in fd_buf.iter_mut().zip(self.fds().iter()) {
+            *raw = fancy.as_fd().as_raw_fd();
+        }
+
+        // Send data and file descriptors
+        let bytes_written = self.socket().send_fds(buf, fd_buf)?;
+
+        // Drop the file descriptors from the Deque
+        repeat!(fd_no, {
+            self.fds_mut().pop_front();
+        });
+
+        Ok(bytes_written)
+    }
+
+    fn flush(&mut self) -> std::io::Result<()> {
+        Ok(())
+    }
+}