io: use io_uring for all io

There should no longer be any

- read
- write
- connect
- sendmsg
- recvmsg
- accept

calls in the codebase. Previously we were using a mix of io_uring and
these calls which had some negative effects: Since we were using the old
system calls, we had to set the file descriptors to non-blocking. But
our io_uring code did not handle EAGAIN. This lead to programs sometimes
being killed when the wayland IO was actually blocking.

Now all file descriptors are set to blocking, but io_uring makes it
non-blocking from our perspective. The one exception are evdev files
because they are read via libinput and libinput uses the old system
calls.

src/acceptor.rs

@@ -113,11 +113,7 @@ fn allocate_socket() -> Result<AllocatedSocket, AcceptorError> {
     };
     let mut fds = [None, None];
     for fd in &mut fds {
-        let socket = match uapi::socket(
+        let socket = match uapi::socket(c::AF_UNIX, c::SOCK_STREAM | c::SOCK_CLOEXEC, 0) {
-            c::AF_UNIX,
-            c::SOCK_STREAM | c::SOCK_NONBLOCK | c::SOCK_CLOEXEC,
-            0,
-        ) {
             Ok(f) => Rc::new(f),
             Err(e) => return Err(AcceptorError::SocketFailed(e.into())),
         };
@@ -172,31 +168,17 @@ impl Acceptor {
 
 async fn accept(fd: Rc<OwnedFd>, state: Rc<State>, secure: bool) {
     loop {
-        if let Err(e) = state.ring.readable(&fd).await {
+        let fd = match state.ring.accept(&fd, c::SOCK_CLOEXEC).await {
-            log::error!(
+            Ok(fd) => fd,
-                "Could not wait for the acceptor to become readable: {}",
+            Err(e) => {
-                ErrorFmt(e)
+                log::error!("Could not accept a client: {}", ErrorFmt(e));
-            );
-            break;
-        }
-        loop {
-            let fd = match uapi::accept4(
-                fd.raw(),
-                uapi::sockaddr_none_mut(),
-                c::SOCK_NONBLOCK | c::SOCK_CLOEXEC,
-            ) {
-                Ok((fd, _)) => fd,
-                Err(Errno(c::EAGAIN)) => break,
-                Err(e) => {
-                    log::error!("Could not accept a client: {}", ErrorFmt(OsError::from(e)));
-                    break;
-                }
-            };
-            let id = state.clients.id();
-            if let Err(e) = state.clients.spawn(id, &state, fd, secure) {
-                log::error!("Could not spawn a client: {}", ErrorFmt(e));
                 break;
             }
+        };
+        let id = state.clients.id();
+        if let Err(e) = state.clients.spawn(id, &state, fd, secure) {
+            log::error!("Could not spawn a client: {}", ErrorFmt(e));
+            break;
         }
     }
     state.ring.stop();

src/backends/metal.rs

@@ -211,7 +211,7 @@ fn dup_fd(fd: c::c_int) -> Result<Rc<OwnedFd>, MetalError> {
 }
 
 pub async fn create(state: &Rc<State>) -> Result<Rc<MetalBackend>, MetalError> {
-    let socket = match state.dbus.system() {
+    let socket = match state.dbus.system().await {
         Ok(s) => s,
         Err(e) => return Err(MetalError::DbusSystemSocket(e)),
     };

src/backends/metal/monitor.rs

@@ -213,11 +213,7 @@ impl MetalBackend {
                     return;
                 }
             };
-            if let Err(e) = set_nonblock(res.fd.raw()) {
+            let master = Rc::new(DrmMaster::new(&slf.state.ring, res.fd.clone()));
-                log::error!("Could set drm fd to non-blocking: {}", ErrorFmt(e));
-                return;
-            }
-            let master = Rc::new(DrmMaster::new(res.fd.clone()));
             let dev = match slf.create_drm_device(dev, &master) {
                 Ok(d) => d,
                 Err(e) => {

src/backends/metal/video.rs

@@ -1214,18 +1214,12 @@ impl MetalBackend {
 
     async fn handle_drm_events(self: Rc<Self>, dev: Rc<MetalDrmDeviceData>) {
         loop {
-            if let Err(e) = self.state.ring.readable(dev.dev.master.fd()).await {
+            match dev.dev.master.event().await {
-                log::error!("Could not register the DRM fd for reading: {}", ErrorFmt(e));
+                Ok(Some(e)) => self.handle_drm_event(e, &dev),
-                break;
+                Ok(None) => break,
-            }
+                Err(e) => {
-            loop {
+                    log::error!("Could not read DRM event: {}", ErrorFmt(e));
-                match dev.dev.master.event() {
+                    return;
-                    Ok(Some(e)) => self.handle_drm_event(e, &dev),
-                    Ok(None) => break,
-                    Err(e) => {
-                        log::error!("Could not read DRM event: {}", ErrorFmt(e));
-                        return;
-                    }
                 }
             }
         }

src/cli/idle.rs

@@ -1,7 +1,7 @@
 use {
     crate::{
         cli::{GlobalArgs, IdleArgs, IdleCmd, IdleSetArgs},
-        tools::tool_client::{Handle, ToolClient},
+        tools::tool_client::{with_tool_client, Handle, ToolClient},
         utils::{errorfmt::ErrorFmt, stack::Stack},
         wire::{jay_compositor, jay_idle, JayIdleId, WlSurfaceId},
     },
@@ -9,9 +9,10 @@ use {
 };
 
 pub fn main(global: GlobalArgs, args: IdleArgs) {
-    let tc = ToolClient::new(global.log_level.into());
+    with_tool_client(global.log_level.into(), |tc| async move {
-    let idle = Idle { tc: tc.clone() };
+        let idle = Idle { tc: tc.clone() };
-    tc.run(idle.run(args));
+        idle.run(args).await;
+    });
 }
 
 struct Idle {

src/cli/log.rs

@@ -1,7 +1,7 @@
 use {
     crate::{
         cli::{GlobalArgs, LogArgs},
-        tools::tool_client::{Handle, ToolClient},
+        tools::tool_client::{with_tool_client, Handle, ToolClient},
         utils::errorfmt::ErrorFmt,
         wire::{jay_compositor, jay_log_file},
     },
@@ -18,13 +18,14 @@ use {
 };
 
 pub fn main(global: GlobalArgs, args: LogArgs) {
-    let tc = ToolClient::new(global.log_level.into());
+    with_tool_client(global.log_level.into(), |tc| async move {
-    let logger = Rc::new(Log {
+        let logger = Rc::new(Log {
-        tc: tc.clone(),
+            tc: tc.clone(),
-        path: RefCell::new(None),
+            path: RefCell::new(None),
-        args,
+            args,
+        });
+        run(logger).await;
     });
-    tc.run(run(logger));
 }
 
 struct Log {

src/cli/quit.rs

@@ -1,11 +1,16 @@
 use {
-    crate::{cli::GlobalArgs, tools::tool_client::ToolClient, wire::jay_compositor::Quit},
+    crate::{
+        cli::GlobalArgs,
+        tools::tool_client::{with_tool_client, ToolClient},
+        wire::jay_compositor::Quit,
+    },
     std::rc::Rc,
 };
 
 pub fn main(global: GlobalArgs) {
-    let tc = ToolClient::new(global.log_level.into());
+    with_tool_client(global.log_level.into(), |tc| async move {
-    tc.run(run(tc.clone()));
+        run(tc).await;
+    });
 }
 
 async fn run(tc: Rc<ToolClient>) {

src/cli/screenshot.rs

@@ -2,7 +2,7 @@ use {
     crate::{
         cli::{GlobalArgs, ScreenshotArgs},
         format::XRGB8888,
-        tools::tool_client::{Handle, ToolClient},
+        tools::tool_client::{with_tool_client, Handle, ToolClient},
         utils::{errorfmt::ErrorFmt, queue::AsyncQueue},
         video::{
             dmabuf::{DmaBuf, DmaBufPlane},
@@ -21,12 +21,13 @@ use {
 };
 
 pub fn main(global: GlobalArgs, args: ScreenshotArgs) {
-    let tc = ToolClient::new(global.log_level.into());
+    with_tool_client(global.log_level.into(), |tc| async move {
-    let screenshot = Rc::new(Screenshot {
+        let screenshot = Rc::new(Screenshot {
-        tc: tc.clone(),
+            tc: tc.clone(),
-        args,
+            args,
+        });
+        run(screenshot).await;
     });
-    tc.run(run(screenshot));
 }
 
 struct Screenshot {

src/cli/seat_test.rs

@@ -2,7 +2,7 @@ use {
     crate::{
         cli::{GlobalArgs, SeatTestArgs},
         ifs::wl_seat::wl_pointer::{PendingScroll, CONTINUOUS, FINGER, WHEEL},
-        tools::tool_client::{Handle, ToolClient},
+        tools::tool_client::{with_tool_client, Handle, ToolClient},
         wire::{
             jay_compositor::{GetSeats, Seat, SeatEvents},
             jay_seat_events::{
@@ -16,13 +16,14 @@ use {
 };
 
 pub fn main(global: GlobalArgs, args: SeatTestArgs) {
-    let tc = ToolClient::new(global.log_level.into());
+    with_tool_client(global.log_level.into(), |tc| async move {
-    let screenshot = Rc::new(SeatTest {
+        let screenshot = Rc::new(SeatTest {
-        tc: tc.clone(),
+            tc: tc.clone(),
-        args,
+            args,
-        names: Default::default(),
+            names: Default::default(),
+        });
+        run(screenshot).await;
     });
-    tc.run(run(screenshot));
 }
 
 struct SeatTest {

src/cli/set_log_level.rs

@@ -1,19 +1,20 @@
 use {
     crate::{
         cli::{GlobalArgs, SetLogArgs},
-        tools::tool_client::ToolClient,
+        tools::tool_client::{with_tool_client, ToolClient},
         wire::jay_compositor::SetLogLevel,
     },
     std::rc::Rc,
 };
 
 pub fn main(global: GlobalArgs, args: SetLogArgs) {
-    let tc = ToolClient::new(global.log_level.into());
+    with_tool_client(global.log_level.into(), |tc| async move {
-    let logger = Rc::new(Log {
+        let logger = Rc::new(Log {
-        tc: tc.clone(),
+            tc: tc.clone(),
-        args,
+            args,
+        });
+        run(logger).await;
     });
-    tc.run(run(logger));
 }
 
 struct Log {

src/cli/unlock.rs

@@ -1,12 +1,17 @@
 use {
-    crate::{cli::GlobalArgs, tools::tool_client::ToolClient, wire::jay_compositor::Unlock},
+    crate::{
+        cli::GlobalArgs,
+        tools::tool_client::{with_tool_client, ToolClient},
+        wire::jay_compositor::Unlock,
+    },
     std::rc::Rc,
 };
 
 pub fn main(global: GlobalArgs) {
-    let tc = ToolClient::new(global.log_level.into());
+    with_tool_client(global.log_level.into(), |tc| async move {
-    let logger = Rc::new(Unlocker { tc: tc.clone() });
+        let logger = Rc::new(Unlocker { tc: tc.clone() });
-    tc.run(run(logger));
+        run(logger).await;
+    });
 }
 
 struct Unlocker {

src/client.rs

@@ -95,7 +95,7 @@ impl Clients {
         &self,
         id: ClientId,
         global: &Rc<State>,
-        socket: OwnedFd,
+        socket: Rc<OwnedFd>,
         secure: bool,
     ) -> Result<(), ClientError> {
         let (uid, pid) = {
@@ -123,7 +123,7 @@ impl Clients {
         &self,
         id: ClientId,
         global: &Rc<State>,
-        socket: OwnedFd,
+        socket: Rc<OwnedFd>,
         uid: c::uid_t,
         pid: c::pid_t,
         secure: bool,
@@ -133,7 +133,7 @@ impl Clients {
             id,
             state: global.clone(),
             checking_queue_size: Cell::new(false),
-            socket: Rc::new(socket),
+            socket,
             objects: Objects::new(),
             swapchain: Default::default(),
             flush_request: Default::default(),

src/compositor.rs

@@ -234,10 +234,10 @@ async fn start_compositor3(state: Rc<State>, test_future: Option<TestFuture>) {
 
     if backend.import_environment() {
         if let Some(acc) = state.acceptor.get() {
-            import_environment(&state, WAYLAND_DISPLAY, acc.socket_name());
+            import_environment(&state, WAYLAND_DISPLAY, acc.socket_name()).await;
         }
         for (key, val) in STATIC_VARS {
-            import_environment(&state, key, val);
+            import_environment(&state, key, val).await;
         }
     }
 

src/dbus.rs

@@ -8,12 +8,14 @@ use {
         },
         io_uring::{IoUring, IoUringError},
         utils::{
+            buf::DynamicBuf,
             bufio::{BufIo, BufIoError},
             clonecell::CloneCell,
             copyhashmap::CopyHashMap,
             numcell::NumCell,
             oserror::OsError,
             run_toplevel::RunToplevel,
+            stack::Stack,
             vecstorage::VecStorage,
             xrd::{xrd, XRD},
         },
@@ -89,11 +91,11 @@ pub enum DbusError {
     #[error("Could not create a socket")]
     Socket(#[source] OsError),
     #[error("Could not connect")]
-    Connect(#[source] OsError),
+    Connect(#[source] IoUringError),
     #[error("Could not write to the dbus socket")]
-    WriteError(#[source] OsError),
+    WriteError(#[source] IoUringError),
     #[error("Could not read from the dbus socket")]
-    ReadError(#[source] OsError),
+    ReadError(#[source] IoUringError),
     #[error("timeout")]
     IoUringError(#[source] Box<IoUringError>),
     #[error("Server did not accept our authentication")]
@@ -169,21 +171,25 @@ impl Dbus {
         self.session.clear();
     }
 
-    pub fn system(&self) -> Result<Rc<DbusSocket>, DbusError> {
+    pub async fn system(&self) -> Result<Rc<DbusSocket>, DbusError> {
-        self.system.get(
+        self.system
-            &self.eng,
+            .get(
-            &self.ring,
+                &self.eng,
-            "/var/run/dbus/system_bus_socket",
+                &self.ring,
-            "System bus",
+                "/var/run/dbus/system_bus_socket",
-        )
+                "System bus",
+            )
+            .await
     }
 
-    pub fn session(&self) -> Result<Rc<DbusSocket>, DbusError> {
+    pub async fn session(&self) -> Result<Rc<DbusSocket>, DbusError> {
         let sba = match self.user_path.as_deref() {
             None => return Err(DbusError::SessionBusAddressNotSet),
             Some(sba) => sba,
         };
-        self.session.get(&self.eng, &self.ring, sba, "Session bus")
+        self.session
+            .get(&self.eng, &self.ring, sba, "Session bus")
+            .await
     }
 }
 
@@ -203,6 +209,7 @@ pub struct DbusSocket {
     bus_name: &'static str,
     fd: Rc<OwnedFd>,
     ring: Rc<IoUring>,
+    in_bufs: Stack<Vec<u8>>,
     bufio: Rc<BufIo>,
     eng: Rc<AsyncEngine>,
     next_serial: NumCell<u32>,
@@ -361,7 +368,7 @@ pub struct Parser<'a> {
 
 pub struct Formatter<'a> {
     fds: &'a mut Vec<Rc<OwnedFd>>,
-    buf: &'a mut Vec<u8>,
+    buf: &'a mut DynamicBuf,
 }
 
 pub unsafe trait Message<'a>: Sized + 'a {
@@ -469,7 +476,7 @@ impl<T: Message<'static>> Reply<T> {
 
 impl<T: Message<'static>> Drop for Reply<T> {
     fn drop(&mut self) {
-        self.socket.bufio.add_buf(mem::take(&mut self.buf));
+        self.socket.in_bufs.push(mem::take(&mut self.buf));
     }
 }
 

src/dbus/auth.rs

@@ -1,16 +1,15 @@
 use {
     crate::{
         dbus::{incoming::handle_incoming, outgoing::handle_outgoing, DbusError, DbusSocket},
-        utils::{errorfmt::ErrorFmt, hex},
+        utils::{buf::Buf, errorfmt::ErrorFmt, hex},
     },
-    std::{io::Write, rc::Rc},
+    std::{ops::Deref, rc::Rc},
-    uapi::{c, Errno},
 };
 
 pub(super) async fn handle_auth(socket: Rc<DbusSocket>) {
     let mut auth = Auth {
         socket: socket.clone(),
-        buf: Box::new([0; BUF_SIZE]),
+        buf: Buf::new(BUF_SIZE),
         buf_start: 0,
         buf_stop: 0,
     };
@@ -22,7 +21,7 @@ const BUF_SIZE: usize = 128;
 struct Auth {
     socket: Rc<DbusSocket>,
 
-    buf: Box<[u8; BUF_SIZE]>,
+    buf: Buf,
     buf_start: usize,
     buf_stop: usize,
 }
@@ -50,61 +49,77 @@ impl Auth {
 
     async fn handle_auth(&mut self) -> Result<(), DbusError> {
         let uid = hex::to_hex(&uapi::getuid().to_string());
-        let mut out_buf = Vec::new();
+        let mut out_buf = Buf::new(128);
-        let _ = write!(out_buf, "\0AUTH EXTERNAL {}\r\n", uid);
+        {
-        self.write_buf(&mut out_buf).await?;
+            let buf = out_buf
+                .write_fmt(format_args!("\0AUTH EXTERNAL {}\r\n", uid))
+                .unwrap();
+            self.write_buf(buf).await?;
+        }
         let line = self.readline().await?;
         let (cmd, _) = line_to_cmd(&line);
         if cmd != "OK" {
             return Err(DbusError::Auth);
         }
-        let _ = write!(out_buf, "NEGOTIATE_UNIX_FD\r\n");
+        {
-        self.write_buf(&mut out_buf).await?;
+            let buf = out_buf
+                .write_fmt(format_args!("NEGOTIATE_UNIX_FD\r\n"))
+                .unwrap();
+            self.write_buf(buf).await?;
+        }
         let line = self.readline().await?;
         let (cmd, _) = line_to_cmd(&line);
         if cmd != "AGREE_UNIX_FD" {
             return Err(DbusError::UnixFd);
         }
-        let _ = write!(out_buf, "BEGIN\r\n");
+        {
-        self.write_buf(&mut out_buf).await?;
+            let buf = out_buf.write_fmt(format_args!("BEGIN\r\n")).unwrap();
+            self.write_buf(buf).await?;
+        }
         Ok(())
     }
 
     async fn readline(&mut self) -> Result<String, DbusError> {
         let mut s = String::new();
         loop {
-            for i in self.buf_start..self.buf_stop {
+            {
-                let c = self.buf[i % BUF_SIZE] as char;
+                let buf = self.buf.deref();
-                s.push(c);
+                for i in self.buf_start..self.buf_stop {
-                if c == '\n' {
+                    let c = buf[i % BUF_SIZE] as char;
-                    self.buf_start = i + 1;
+                    s.push(c);
-                    return Ok(s);
+                    if c == '\n' {
+                        self.buf_start = i + 1;
+                        return Ok(s);
+                    }
                 }
             }
             self.buf_start = 0;
             self.buf_stop = 0;
-            match uapi::read(self.socket.fd.raw(), &mut self.buf[..]) {
+            let res = self
-                Ok(n) => self.buf_stop = n.len(),
+                .socket
-                Err(Errno(c::EAGAIN)) => {
+                .ring
-                    self.socket.ring.readable(&self.socket.fd).await?;
+                .read(&self.socket.fd, self.buf.clone())
-                }
+                .await;
-                Err(e) => return Err(DbusError::ReadError(e.into())),
+            match res {
+                Ok(n) => self.buf_stop = n,
+                Err(e) => return Err(DbusError::ReadError(e)),
             }
         }
     }
 
-    async fn write_buf(&mut self, buf: &mut Vec<u8>) -> Result<(), DbusError> {
+    async fn write_buf(&mut self, mut buf: Buf) -> Result<(), DbusError> {
         let mut start = 0;
         while start < buf.len() {
-            match uapi::write(self.socket.fd.raw(), &buf[start..]) {
+            let res = self
+                .socket
+                .ring
+                .write(&self.socket.fd, buf.slice(start..), None)
+                .await;
+            match res {
                 Ok(n) => start += n,
-                Err(Errno(c::EAGAIN)) => {
+                Err(e) => return Err(DbusError::WriteError(e)),
-                    self.socket.ring.writable(&self.socket.fd).await?;
-                }
-                Err(e) => return Err(DbusError::WriteError(e.into())),
             }
         }
-        buf.clear();
         Ok(())
     }
 }

src/dbus/dynamic_type.rs

@@ -3,7 +3,10 @@ use {
         TY_ARRAY, TY_BOOLEAN, TY_BYTE, TY_DOUBLE, TY_INT16, TY_INT32, TY_INT64, TY_OBJECT_PATH,
         TY_SIGNATURE, TY_STRING, TY_UINT16, TY_UINT32, TY_UINT64, TY_UNIX_FD, TY_VARIANT,
     },
-    crate::dbus::{types::Variant, DbusError, DynamicType, Parser},
+    crate::{
+        dbus::{types::Variant, DbusError, DynamicType, Parser},
+        utils::buf::DynamicBuf,
+    },
     std::ops::Deref,
 };
 
@@ -88,7 +91,7 @@ impl DynamicType {
         }
     }
 
-    pub fn write_signature(&self, w: &mut Vec<u8>) {
+    pub fn write_signature(&self, w: &mut DynamicBuf) {
         let c = match self {
             DynamicType::U8 => TY_BYTE,
             DynamicType::Bool => TY_BOOLEAN,

src/dbus/formatter.rs

@@ -1,11 +1,14 @@
 use {
-    crate::dbus::{types::Variant, DbusType, Formatter},
+    crate::{
+        dbus::{types::Variant, DbusType, Formatter},
+        utils::buf::DynamicBuf,
+    },
     std::rc::Rc,
     uapi::{OwnedFd, Packed},
 };
 
 impl<'a> Formatter<'a> {
-    pub fn new(fds: &'a mut Vec<Rc<OwnedFd>>, buf: &'a mut Vec<u8>) -> Self {
+    pub fn new(fds: &'a mut Vec<Rc<OwnedFd>>, buf: &'a mut DynamicBuf) -> Self {
         Self { fds, buf }
     }
 

src/dbus/holder.rs

@@ -11,7 +11,7 @@ use {
 };
 
 impl DbusHolder {
-    pub(super) fn get(
+    pub(super) async fn get(
         self: &Rc<Self>,
         eng: &Rc<AsyncEngine>,
         ring: &Rc<IoUring>,
@@ -25,39 +25,35 @@ impl DbusHolder {
                 return Ok(c);
             }
         }
-        let socket = connect(eng, ring, addr, name, &self.run_toplevel)?;
+        let socket = connect(eng, ring, addr, name, &self.run_toplevel).await?;
         self.socket.set(Some(socket.clone()));
         Ok(socket)
     }
 }
 
-fn connect(
+async fn connect(
     eng: &Rc<AsyncEngine>,
     ring: &Rc<IoUring>,
     addr: &str,
     name: &'static str,
     run_toplevel: &Rc<RunToplevel>,
 ) -> Result<Rc<DbusSocket>, DbusError> {
-    let socket = match uapi::socket(
+    let fd = match uapi::socket(c::AF_UNIX, c::SOCK_STREAM | c::SOCK_CLOEXEC, 0) {
-        c::AF_UNIX,
+        Ok(s) => Rc::new(s),
-        c::SOCK_STREAM | c::SOCK_NONBLOCK | c::SOCK_CLOEXEC,
-        0,
-    ) {
-        Ok(s) => s,
         Err(e) => return Err(DbusError::Socket(e.into())),
     };
     let mut sadr: c::sockaddr_un = uapi::pod_zeroed();
     sadr.sun_family = c::AF_UNIX as _;
     let sun_path = uapi::as_bytes_mut(&mut sadr.sun_path[..]);
     sun_path[..addr.len()].copy_from_slice(addr.as_bytes());
-    if let Err(e) = uapi::connect(socket.raw(), &sadr) {
+    if let Err(e) = ring.connect(&fd, &sadr).await {
-        return Err(DbusError::Connect(e.into()));
+        return Err(DbusError::Connect(e));
     }
-    let fd = Rc::new(socket);
     let socket = Rc::new(DbusSocket {
         bus_name: name,
         fd: fd.clone(),
         ring: ring.clone(),
+        in_bufs: Default::default(),
         bufio: Rc::new(BufIo::new(&fd, ring)),
         eng: eng.clone(),
         next_serial: NumCell::new(1),

src/dbus/incoming.rs

@@ -52,8 +52,9 @@ impl Incoming {
     }
 
     async fn handle_msg(&mut self) -> Result<(), DbusError> {
-        let msg_buf_data = UnsafeCell::new(self.socket.bufio.buf());
+        let msg_buf_data = UnsafeCell::new(self.socket.in_bufs.pop().unwrap_or_default());
         let msg_buf = unsafe { msg_buf_data.get().deref_mut() };
+        msg_buf.clear();
         const FIXED_HEADER_SIZE: usize = 16;
         self.incoming
             .fill_msg_buf(FIXED_HEADER_SIZE, msg_buf)
@@ -235,7 +236,7 @@ impl Incoming {
         }
         let msg_buf = msg_buf_data.into_inner();
         if msg_buf.capacity() > 0 {
-            self.socket.bufio.add_buf(msg_buf);
+            self.socket.in_bufs.push(msg_buf);
         }
         Ok(())
     }

src/dbus/socket.rs

@@ -404,7 +404,13 @@ impl DbusSocket {
         msg.marshal(&mut fmt);
         let body_len = (buf.len() - body_start) as u32;
         buf[4..8].copy_from_slice(uapi::as_bytes(&body_len));
-        (BufIoMessage { fds, buf }, serial)
+        (
+            BufIoMessage {
+                fds,
+                buf: buf.unwrap(),
+            },
+            serial,
+        )
     }
 
     fn format_header(
@@ -489,7 +495,7 @@ where
     ) -> Result<(), DbusError> {
         let msg = <T::Generic<'a> as Message>::unmarshal(parser)?;
         (self.0)(Ok(&msg));
-        socket.bufio.add_buf(buf);
+        socket.in_bufs.push(buf);
         Ok(())
     }
 }

src/dbus/types.rs

@@ -1,8 +1,11 @@
 use {
-    crate::dbus::{
+    crate::{
-        DbusError, DbusType, DynamicType, Formatter, Parser, TY_ARRAY, TY_BOOLEAN, TY_BYTE,
+        dbus::{
-        TY_DOUBLE, TY_INT16, TY_INT32, TY_INT64, TY_OBJECT_PATH, TY_SIGNATURE, TY_STRING,
+            DbusError, DbusType, DynamicType, Formatter, Parser, TY_ARRAY, TY_BOOLEAN, TY_BYTE,
-        TY_UINT16, TY_UINT32, TY_UINT64, TY_UNIX_FD, TY_VARIANT,
+            TY_DOUBLE, TY_INT16, TY_INT32, TY_INT64, TY_OBJECT_PATH, TY_SIGNATURE, TY_STRING,
+            TY_UINT16, TY_UINT32, TY_UINT64, TY_UNIX_FD, TY_VARIANT,
+        },
+        utils::buf::DynamicBuf,
     },
     std::{borrow::Cow, ops::Deref, rc::Rc},
     uapi::{OwnedFd, Packed, Pod},
@@ -458,7 +461,7 @@ impl<'a> Variant<'a> {
         }
     }
 
-    pub fn write_signature(&self, w: &mut Vec<u8>) {
+    pub fn write_signature(&self, w: &mut DynamicBuf) {
         let c = match self {
             Variant::U8(..) => TY_BYTE,
             Variant::Bool(..) => TY_BOOLEAN,

src/forker.rs

@@ -50,7 +50,7 @@ pub struct ForkerProxy {
 }
 
 struct PidfdHandoff {
-    pidfd: Cell<Option<Result<(OwnedFd, c::pid_t), ForkerError>>>,
+    pidfd: Cell<Option<Result<(Rc<OwnedFd>, c::pid_t), ForkerError>>>,
     waiter: Cell<Option<Waker>>,
 }
 
@@ -81,14 +81,11 @@ impl ForkerProxy {
     }
 
     pub fn create() -> Result<Self, ForkerError> {
-        let (parent, child) = match uapi::socketpair(
+        let (parent, child) =
-            c::AF_UNIX,
+            match uapi::socketpair(c::AF_UNIX, c::SOCK_STREAM | c::SOCK_CLOEXEC, 0) {
-            c::SOCK_STREAM | c::SOCK_CLOEXEC | c::SOCK_NONBLOCK,
+                Ok(o) => o,
-            0,
+                Err(e) => return Err(ForkerError::Socketpair(e.into())),
-        ) {
+            };
-            Ok(o) => o,
-            Err(e) => return Err(ForkerError::Socketpair(e.into())),
-        };
         let pid = uapi::getpid();
         match fork_with_pidfd(false)? {
             Forked::Parent { pid, pidfd } => Ok(ForkerProxy {
@@ -135,7 +132,7 @@ impl ForkerProxy {
         })
     }
 
-    async fn pidfd(&self, id: u32) -> Result<(OwnedFd, c::pid_t), ForkerError> {
+    async fn pidfd(&self, id: u32) -> Result<(Rc<OwnedFd>, c::pid_t), ForkerError> {
         let handoff = Rc::new(PidfdHandoff {
             pidfd: Cell::new(None),
             waiter: Cell::new(None),
@@ -159,7 +156,7 @@ impl ForkerProxy {
         listenfd: Rc<OwnedFd>,
         wmfd: Rc<OwnedFd>,
         waylandfd: Rc<OwnedFd>,
-    ) -> Result<(OwnedFd, c::pid_t), ForkerError> {
+    ) -> Result<(Rc<OwnedFd>, c::pid_t), ForkerError> {
         self.fds
             .borrow_mut()
             .extend([stderr, dfd, listenfd, wmfd, waylandfd]);
@@ -394,10 +391,10 @@ impl Forker {
     fn handle_xwayland(self: &Rc<Self>, io: &mut IoIn, id: u32) {
         let stderr = io.pop_fd();
         let fds = vec![
-            io.pop_fd().unwrap(),
+            Rc::try_unwrap(io.pop_fd().unwrap()).unwrap(),
-            io.pop_fd().unwrap(),
+            Rc::try_unwrap(io.pop_fd().unwrap()).unwrap(),
-            io.pop_fd().unwrap(),
+            Rc::try_unwrap(io.pop_fd().unwrap()).unwrap(),
-            io.pop_fd().unwrap(),
+            Rc::try_unwrap(io.pop_fd().unwrap()).unwrap(),
         ];
         let (prog, args) = xwayland::build_args(&fds);
         let env = vec![("WAYLAND_SOCKET".to_string(), fds[3].raw().to_string())];
@@ -424,7 +421,7 @@ impl Forker {
         prog: String,
         args: Vec<String>,
         env: Vec<(String, String)>,
-        stderr: Option<OwnedFd>,
+        stderr: Option<Rc<OwnedFd>>,
         fds: Vec<OwnedFd>,
         pidfd_id: Option<u32>,
     ) {

src/forker/io.rs

@@ -8,6 +8,7 @@ use {
         forker::ForkerError,
         io_uring::IoUring,
         utils::{
+            buf::DynamicBuf,
             buffd::{BufFdIn, BufFdOut},
             vec_ext::VecExt,
         },
@@ -29,7 +30,7 @@ impl IoIn {
         }
     }
 
-    pub fn pop_fd(&mut self) -> Option<OwnedFd> {
+    pub fn pop_fd(&mut self) -> Option<Rc<OwnedFd>> {
         self.incoming.get_fd().ok()
     }
 
@@ -57,7 +58,7 @@ impl IoIn {
 
 pub struct IoOut {
     outgoing: BufFdOut,
-    scratch: Vec<u8>,
+    scratch: DynamicBuf,
     fds: Vec<Rc<OwnedFd>>,
 }
 
@@ -65,7 +66,7 @@ impl IoOut {
     pub fn new(fd: &Rc<OwnedFd>, ring: &Rc<IoUring>) -> Self {
         Self {
             outgoing: BufFdOut::new(fd, ring),
-            scratch: vec![],
+            scratch: DynamicBuf::new(),
             fds: vec![],
         }
     }
@@ -83,7 +84,12 @@ impl IoOut {
             Err(e) => return Err(ForkerError::EncodeFailed(e)),
         };
         self.scratch[..mem::size_of_val(&len)].copy_from_slice(uapi::as_bytes(&len));
-        match self.outgoing.flush2(&self.scratch, &mut self.fds).await {
+        let mut buf = self.scratch.borrow();
+        match self
+            .outgoing
+            .flush2(buf.buf.clone(), mem::take(&mut self.fds))
+            .await
+        {
             Ok(()) => Ok(()),
             Err(e) => Err(ForkerError::WriteFailed(e)),
         }

src/io_uring.rs

@@ -4,8 +4,9 @@ use {
         async_engine::AsyncEngine,
         io_uring::{
             ops::{
-                async_cancel::AsyncCancelTask, poll::PollTask, recvmsg::RecvmsgTask,
+                accept::AcceptTask, async_cancel::AsyncCancelTask, connect::ConnectTask,
-                sendmsg::SendmsgTask, timeout::TimeoutTask, write::WriteTask,
+                poll::PollTask, read_write::ReadWriteTask, recvmsg::RecvmsgTask,
+                sendmsg::SendmsgTask, timeout::TimeoutTask,
             },
             pending_result::PendingResults,
             sys::{
@@ -204,13 +205,15 @@ impl IoUring {
             pending_in_kernel: Default::default(),
             tasks: Default::default(),
             pending_results: Default::default(),
-            cached_writes: Default::default(),
+            cached_read_writes: Default::default(),
             cached_cancels: Default::default(),
             cached_polls: Default::default(),
             cached_sendmsg: Default::default(),
             cached_recvmsg: Default::default(),
             cached_timeouts: Default::default(),
             cached_cmsg_bufs: Default::default(),
+            cached_connects: Default::default(),
+            cached_accepts: Default::default(),
             fd_ids_scratch: Default::default(),
         });
         Ok(Rc::new(Self { ring: data }))
@@ -257,13 +260,15 @@ struct IoUringData {
 
     pending_results: PendingResults,
 
-    cached_writes: Stack<Box<WriteTask>>,
+    cached_read_writes: Stack<Box<ReadWriteTask>>,
     cached_cancels: Stack<Box<AsyncCancelTask>>,
     cached_polls: Stack<Box<PollTask>>,
     cached_sendmsg: Stack<Box<SendmsgTask>>,
     cached_recvmsg: Stack<Box<RecvmsgTask>>,
     cached_timeouts: Stack<Box<TimeoutTask>>,
     cached_cmsg_bufs: Stack<Buf>,
+    cached_connects: Stack<Box<ConnectTask>>,
+    cached_accepts: Stack<Box<AcceptTask>>,
 
     fd_ids_scratch: RefCell<Vec<c::c_int>>,
 }

src/io_uring/ops.rs

@@ -1,11 +1,13 @@
 use crate::{io_uring::IoUringError, utils::oserror::OsError};
 
+pub mod accept;
 pub mod async_cancel;
+pub mod connect;
 pub mod poll;
+pub mod read_write;
 pub mod recvmsg;
 pub mod sendmsg;
 pub mod timeout;
-pub mod write;
 
 pub type TaskResult<T> = Result<Result<T, OsError>, IoUringError>;
 

src/io_uring/ops/accept.rs

@@ -0,0 +1,67 @@
+use {
+    crate::io_uring::{
+        pending_result::PendingResult,
+        sys::{io_uring_sqe, IORING_OP_ACCEPT},
+        IoUring, IoUringData, IoUringError, Task, TaskResultExt,
+    },
+    std::rc::Rc,
+    uapi::{c, OwnedFd},
+};
+
+impl IoUring {
+    pub async fn accept(
+        &self,
+        fd: &Rc<OwnedFd>,
+        flags: c::c_int,
+    ) -> Result<Rc<OwnedFd>, IoUringError> {
+        self.ring.check_destroyed()?;
+        let id = self.ring.id();
+        let pr = self.ring.pending_results.acquire();
+        {
+            let mut pw = self.ring.cached_accepts.pop().unwrap_or_default();
+            pw.id = id.id;
+            pw.fd = fd.raw() as _;
+            pw.flags = flags as _;
+            pw.data = Some(Data {
+                pr: pr.clone(),
+                _fd: fd.clone(),
+            });
+            self.ring.schedule(pw);
+        }
+        Ok(pr.await.map(OwnedFd::new).map(Rc::new)).merge()
+    }
+}
+
+struct Data {
+    pr: PendingResult,
+    _fd: Rc<OwnedFd>,
+}
+
+#[derive(Default)]
+pub struct AcceptTask {
+    id: u64,
+    fd: i32,
+    flags: u32,
+    data: Option<Data>,
+}
+
+unsafe impl Task for AcceptTask {
+    fn id(&self) -> u64 {
+        self.id
+    }
+
+    fn complete(mut self: Box<Self>, ring: &IoUringData, res: i32) {
+        if let Some(data) = self.data.take() {
+            data.pr.complete(res);
+        }
+        ring.cached_accepts.push(self);
+    }
+
+    fn encode(&self, sqe: &mut io_uring_sqe) {
+        sqe.opcode = IORING_OP_ACCEPT;
+        sqe.fd = self.fd;
+        sqe.u2.addr = 0;
+        sqe.u1.addr2 = 0;
+        sqe.u3.accept_flags = self.flags;
+    }
+}

src/io_uring/ops/connect.rs

@@ -0,0 +1,77 @@
+use {
+    crate::io_uring::{
+        pending_result::PendingResult,
+        sys::{io_uring_sqe, IORING_OP_CONNECT},
+        IoUring, IoUringData, IoUringError, Task, TaskResultExt,
+    },
+    std::{mem, ptr, rc::Rc},
+    uapi::{c, OwnedFd, SockAddr},
+};
+
+impl IoUring {
+    pub async fn connect<T: SockAddr>(&self, fd: &Rc<OwnedFd>, t: &T) -> Result<(), IoUringError> {
+        self.ring.check_destroyed()?;
+        let id = self.ring.id();
+        let pr = self.ring.pending_results.acquire();
+        {
+            let mut pw = self.ring.cached_connects.pop().unwrap_or_default();
+            pw.id = id.id;
+            pw.fd = fd.raw() as _;
+            unsafe {
+                ptr::copy_nonoverlapping(t, &mut pw.sockaddr as *mut _ as *mut _, 1);
+            }
+            pw.addrlen = mem::size_of::<T>() as _;
+            pw.data = Some(Data {
+                pr: pr.clone(),
+                _fd: fd.clone(),
+            });
+            self.ring.schedule(pw);
+        }
+        Ok(pr.await.map(drop)).merge()
+    }
+}
+
+struct Data {
+    pr: PendingResult,
+    _fd: Rc<OwnedFd>,
+}
+
+pub struct ConnectTask {
+    id: u64,
+    fd: i32,
+    sockaddr: c::sockaddr_storage,
+    addrlen: u64,
+    data: Option<Data>,
+}
+
+impl Default for ConnectTask {
+    fn default() -> Self {
+        Self {
+            id: 0,
+            fd: 0,
+            sockaddr: uapi::pod_zeroed(),
+            addrlen: 0,
+            data: None,
+        }
+    }
+}
+
+unsafe impl Task for ConnectTask {
+    fn id(&self) -> u64 {
+        self.id
+    }
+
+    fn complete(mut self: Box<Self>, ring: &IoUringData, res: i32) {
+        if let Some(data) = self.data.take() {
+            data.pr.complete(res);
+        }
+        ring.cached_connects.push(self);
+    }
+
+    fn encode(&self, sqe: &mut io_uring_sqe) {
+        sqe.opcode = IORING_OP_CONNECT;
+        sqe.fd = self.fd;
+        sqe.u2.addr = &self.sockaddr as *const _ as _;
+        sqe.u1.off = self.addrlen;
+    }
+}

src/io_uring/ops/poll.rs

@@ -32,6 +32,7 @@ impl IoUring {
         self.poll(fd, c::POLLIN).await.merge()
     }
 
+    #[allow(dead_code)]
     pub async fn writable(&self, fd: &Rc<OwnedFd>) -> Result<c::c_short, IoUringError> {
         self.poll(fd, c::POLLOUT).await.merge()
     }

src/io_uring/ops/read_write.rs

@@ -1,10 +1,9 @@
 use {
     crate::{
         io_uring::{
-            ops::TaskResult,
             pending_result::PendingResult,
-            sys::{io_uring_sqe, IORING_OP_WRITE},
+            sys::{io_uring_sqe, IORING_OP_READ, IORING_OP_WRITE},
-            IoUring, IoUringData, Task,
+            IoUring, IoUringData, IoUringError, Task, TaskResultExt,
         },
         time::Time,
         utils::buf::Buf,
@@ -14,23 +13,38 @@ use {
 };
 
 impl IoUring {
+    pub async fn read(&self, fd: &Rc<OwnedFd>, buf: Buf) -> Result<usize, IoUringError> {
+        self.perform(fd, buf, None, IORING_OP_READ).await
+    }
+
     pub async fn write(
         &self,
         fd: &Rc<OwnedFd>,
         buf: Buf,
         timeout: Option<Time>,
-    ) -> TaskResult<usize> {
+    ) -> Result<usize, IoUringError> {
+        self.perform(fd, buf, timeout, IORING_OP_WRITE).await
+    }
+
+    async fn perform(
+        &self,
+        fd: &Rc<OwnedFd>,
+        buf: Buf,
+        timeout: Option<Time>,
+        opcode: u8,
+    ) -> Result<usize, IoUringError> {
         self.ring.check_destroyed()?;
         let id = self.ring.id();
         let pr = self.ring.pending_results.acquire();
         {
-            let mut pw = self.ring.cached_writes.pop().unwrap_or_default();
+            let mut pw = self.ring.cached_read_writes.pop().unwrap_or_default();
+            pw.opcode = opcode;
             pw.id = id.id;
             pw.has_timeout = timeout.is_some();
             pw.fd = fd.raw();
             pw.buf = buf.as_ptr() as _;
             pw.len = buf.len();
-            pw.data = Some(WriteTaskData {
+            pw.data = Some(ReadWriteTaskData {
                 _fd: fd.clone(),
                 _buf: buf,
                 res: pr.clone(),
@@ -40,27 +54,28 @@ impl IoUring {
                 self.schedule_timeout(time);
             }
         }
-        Ok(pr.await.map(|v| v as usize))
+        Ok(pr.await.map(|v| v as usize)).merge()
     }
 }
 
-struct WriteTaskData {
+struct ReadWriteTaskData {
     _fd: Rc<OwnedFd>,
     _buf: Buf,
     res: PendingResult,
 }
 
 #[derive(Default)]
-pub struct WriteTask {
+pub struct ReadWriteTask {
     id: u64,
     has_timeout: bool,
     fd: c::c_int,
     buf: usize,
     len: usize,
-    data: Option<WriteTaskData>,
+    data: Option<ReadWriteTaskData>,
+    opcode: u8,
 }
 
-unsafe impl Task for WriteTask {
+unsafe impl Task for ReadWriteTask {
     fn id(&self) -> u64 {
         self.id
     }
@@ -69,11 +84,11 @@ unsafe impl Task for WriteTask {
         if let Some(data) = self.data.take() {
             data.res.complete(res);
         }
-        ring.cached_writes.push(self);
+        ring.cached_read_writes.push(self);
     }
 
     fn encode(&self, sqe: &mut io_uring_sqe) {
-        sqe.opcode = IORING_OP_WRITE;
+        sqe.opcode = self.opcode;
         sqe.fd = self.fd as _;
         sqe.u1.off = !0;
         sqe.u2.addr = self.buf as _;

src/io_uring/ops/recvmsg.rs

@@ -16,7 +16,7 @@ impl IoUring {
         &self,
         fd: &Rc<OwnedFd>,
         bufs: &mut [Buf],
-        fds: &mut VecDeque<OwnedFd>,
+        fds: &mut VecDeque<Rc<OwnedFd>>,
     ) -> Result<usize, IoUringError> {
         self.ring.check_destroyed()?;
         let id = self.ring.id();
@@ -64,7 +64,7 @@ impl IoUring {
                         }
                     };
                     if (hdr.cmsg_level, hdr.cmsg_type) == (c::SOL_SOCKET, c::SCM_RIGHTS) {
-                        fds.extend(uapi::pod_iter(data).unwrap());
+                        fds.extend(uapi::pod_iter(data).unwrap().map(Rc::new));
                     }
                 }
                 return_cmsg!();

src/io_uring/ops/sendmsg.rs

@@ -17,12 +17,22 @@ use {
 };
 
 impl IoUring {
-    pub async fn sendmsg(
+    pub async fn sendmsg_one(
         &self,
         fd: &Rc<OwnedFd>,
         buf: Buf,
         fds: Vec<Rc<OwnedFd>>,
         timeout: Option<Time>,
+    ) -> Result<usize, IoUringError> {
+        self.sendmsg(fd, &mut [buf], fds, timeout).await
+    }
+
+    pub async fn sendmsg(
+        &self,
+        fd: &Rc<OwnedFd>,
+        bufs: &mut [Buf],
+        fds: Vec<Rc<OwnedFd>>,
+        timeout: Option<Time>,
     ) -> Result<usize, IoUringError> {
         self.ring.check_destroyed()?;
         let id = self.ring.id();
@@ -52,13 +62,17 @@ impl IoUring {
             }
             st.id = id.id;
             st.fd = fd.raw();
-            st.iovec.iov_base = buf.as_ptr() as _;
+            st.bufs.clear();
-            st.iovec.iov_len = buf.len() as _;
+            st.bufs.extend(bufs.iter_mut().map(|b| b.clone()));
-            st.msghdr.msg_iov = &st.iovec as *const _ as _;
+            st.iovecs.clear();
-            st.msghdr.msg_iovlen = 1;
+            st.iovecs.extend(bufs.iter().map(|b| c::iovec {
+                iov_base: b.as_ptr() as _,
+                iov_len: b.len(),
+            }));
+            st.msghdr.msg_iov = st.iovecs.as_ptr() as _;
+            st.msghdr.msg_iovlen = st.iovecs.len();
             st.data = Some(SendmsgTaskData {
                 _fd: fd.clone(),
-                _buf: buf,
                 res: pr.clone(),
             });
             st.has_timeout = timeout.is_some();
@@ -73,14 +87,14 @@ impl IoUring {
 
 struct SendmsgTaskData {
     _fd: Rc<OwnedFd>,
-    _buf: Buf,
     res: PendingResult,
 }
 
 pub struct SendmsgTask {
     id: u64,
-    iovec: c::iovec,
+    iovecs: Vec<c::iovec>,
     msghdr: c::msghdr,
+    bufs: Vec<Buf>,
     fd: i32,
     has_timeout: bool,
     fds: Vec<Rc<OwnedFd>>,
@@ -93,8 +107,9 @@ impl Default for SendmsgTask {
         unsafe {
             SendmsgTask {
                 id: 0,
-                iovec: MaybeUninit::zeroed().assume_init(),
+                iovecs: vec![],
                 msghdr: MaybeUninit::zeroed().assume_init(),
+                bufs: vec![],
                 fd: 0,
                 has_timeout: false,
                 fds: vec![],
@@ -112,6 +127,7 @@ unsafe impl Task for SendmsgTask {
 
     fn complete(mut self: Box<Self>, ring: &IoUringData, res: i32) {
         self.fds.clear();
+        self.bufs.clear();
         if let Some(data) = self.data.take() {
             data.res.complete(res);
         }

src/it/testrun.rs

@@ -40,16 +40,15 @@ impl TestRun {
     }
 
     async fn create_client2(self: &Rc<Self>) -> Result<Rc<TestClient>, TestError> {
-        let socket = uapi::socket(
+        let socket = uapi::socket(c::AF_UNIX, c::SOCK_STREAM | c::SOCK_CLOEXEC, 0)
-            c::AF_UNIX,
-            c::SOCK_STREAM | c::SOCK_CLOEXEC | c::SOCK_NONBLOCK,
-            0,
-        )
-        .to_os_error()
-        .with_context(|| "Could not create a unix socket")?;
-        let socket = Rc::new(socket);
-        uapi::connect(socket.raw(), &self.server_addr)
             .to_os_error()
+            .with_context(|| "Could not create a unix socket")?;
+        let socket = Rc::new(socket);
+        self.backend
+            .state
+            .ring
+            .connect(&socket, &self.server_addr)
+            .await
             .with_context(|| "Could not connect to the compositor")?;
         let mut obj_ids = Bitfield::default();
         obj_ids.take(0);

src/pipewire/pw_con.rs

@@ -1,7 +1,7 @@
 use {
     crate::{
         async_engine::{AsyncEngine, SpawnedFuture},
-        io_uring::IoUring,
+        io_uring::{IoUring, IoUringError},
         pipewire::{
             pw_formatter::{format, PwFormatter},
             pw_ifs::{
@@ -43,7 +43,7 @@ pub enum PwConError {
     #[error("Could not create a unix socket")]
     CreateSocket(#[source] OsError),
     #[error("Could not connect to the pipewire daemon")]
-    ConnectSocket(#[source] OsError),
+    ConnectSocket(#[source] IoUringError),
     #[error(transparent)]
     BufIoError(#[from] BufIoError),
     #[error("Server did not sent a required fd")]
@@ -194,7 +194,10 @@ impl PwCon {
                 log::trace!("{:#?}", parser.read_pod().unwrap());
             }
         }
-        self.io.send(BufIoMessage { fds, buf });
+        self.io.send(BufIoMessage {
+            fds,
+            buf: buf.unwrap(),
+        });
     }
 
     #[allow(dead_code)]
@@ -295,13 +298,8 @@ impl Drop for PwConHolder {
 }
 
 impl PwConHolder {
-    #[allow(dead_code)]
+    pub async fn new(eng: &Rc<AsyncEngine>, ring: &Rc<IoUring>) -> Result<Rc<Self>, PwConError> {
-    pub fn new(eng: &Rc<AsyncEngine>, ring: &Rc<IoUring>) -> Result<Rc<Self>, PwConError> {
+        let fd = match uapi::socket(c::AF_UNIX, c::SOCK_STREAM | c::SOCK_CLOEXEC, 0) {
-        let fd = match uapi::socket(
-            c::AF_UNIX,
-            c::SOCK_STREAM | c::SOCK_CLOEXEC | c::SOCK_NONBLOCK,
-            0,
-        ) {
             Ok(fd) => Rc::new(fd),
             Err(e) => return Err(PwConError::CreateSocket(e.into())),
         };
@@ -317,8 +315,8 @@ impl PwConHolder {
             let mut path = uapi::as_bytes_mut(&mut addr.sun_path[..]);
             let _ = write!(path, "{}/pipewire-0", xrd);
         }
-        if let Err(e) = uapi::connect(fd.raw(), &addr) {
+        if let Err(e) = ring.connect(&fd, &addr).await {
-            return Err(PwConError::ConnectSocket(e.into()));
+            return Err(PwConError::ConnectSocket(e));
         }
         let io = Rc::new(BufIo::new(&fd, ring));
         let data = Rc::new(PwCon {

src/pipewire/pw_formatter.rs

@@ -1,16 +1,19 @@
 use {
-    crate::pipewire::pw_pod::{
+    crate::{
-        PW_TYPE_Array, PW_TYPE_Bitmap, PW_TYPE_Bool, PW_TYPE_Bytes, PW_TYPE_Choice, PW_TYPE_Double,
+        pipewire::pw_pod::{
-        PW_TYPE_Fd, PW_TYPE_Float, PW_TYPE_Fraction, PW_TYPE_Id, PW_TYPE_Int, PW_TYPE_Long,
+            PW_TYPE_Array, PW_TYPE_Bitmap, PW_TYPE_Bool, PW_TYPE_Bytes, PW_TYPE_Choice,
-        PW_TYPE_None, PW_TYPE_Object, PW_TYPE_Rectangle, PW_TYPE_String, PW_TYPE_Struct,
+            PW_TYPE_Double, PW_TYPE_Fd, PW_TYPE_Float, PW_TYPE_Fraction, PW_TYPE_Id, PW_TYPE_Int,
-        PwChoiceType, PwPodObjectType, PwPodType, PwPropFlag,
+            PW_TYPE_Long, PW_TYPE_None, PW_TYPE_Object, PW_TYPE_Rectangle, PW_TYPE_String,
+            PW_TYPE_Struct, PwChoiceType, PwPodObjectType, PwPodType, PwPropFlag,
+        },
+        utils::buf::DynamicBuf,
     },
     std::rc::Rc,
     uapi::OwnedFd,
 };
 
 pub struct PwFormatter<'a> {
-    data: &'a mut Vec<u8>,
+    data: &'a mut DynamicBuf,
     fds: &'a mut Vec<Rc<OwnedFd>>,
     array: bool,
     first: bool,
@@ -260,7 +263,7 @@ impl<'a> PwFormatter<'a> {
 }
 
 pub struct PwObjectFormatter<'a> {
-    data: &'a mut Vec<u8>,
+    data: &'a mut DynamicBuf,
     fds: &'a mut Vec<Rc<OwnedFd>>,
 }
 
@@ -281,8 +284,14 @@ impl<'a> PwObjectFormatter<'a> {
     }
 }
 
-pub fn format<F>(buf: &mut Vec<u8>, fds: &mut Vec<Rc<OwnedFd>>, id: u32, opcode: u8, seq: u32, f: F)
+pub fn format<F>(
-where
+    buf: &mut DynamicBuf,
+    fds: &mut Vec<Rc<OwnedFd>>,
+    id: u32,
+    opcode: u8,
+    seq: u32,
+    f: F,
+) where
     F: FnOnce(&mut PwFormatter),
 {
     buf.clear();

src/pipewire/pw_ifs/pw_client_node.rs

@@ -30,7 +30,7 @@ use {
             },
         },
         utils::{
-            bitfield::Bitfield, bitflags::BitflagsExt, clonecell::CloneCell,
+            bitfield::Bitfield, bitflags::BitflagsExt, buf::TypedBuf, clonecell::CloneCell,
             copyhashmap::CopyHashMap, errorfmt::ErrorFmt,
         },
         video::dmabuf::DmaBuf,
@@ -798,17 +798,11 @@ impl PwClientNode {
         _activation: Rc<PwMemTyped<pw_node_activation>>,
         fd: Rc<OwnedFd>,
     ) {
+        let mut buf = TypedBuf::<u64>::new();
         loop {
             // unsafe {
             //     log::info!("transport = {:#?}", activation.read());
             // }
-            if let Err(e) = self.con.ring.readable(&fd).await {
-                log::error!(
-                    "Could not wait for transport to become readable: {}",
-                    ErrorFmt(e)
-                );
-                return;
-            }
             // log::info!("transport in");
             // for port in self.ports.lock().values() {
             //     for io in port.io_buffers.lock().values() {
@@ -820,11 +814,11 @@ impl PwClientNode {
             // unsafe {
             //     log::info!("state = {:#?}", activation.read().state[0]);
             // }
-            let mut n = 0u64;
+            if let Err(e) = self.con.ring.read(&fd, buf.buf()).await {
-            if let Err(e) = uapi::read(fd.raw(), &mut n) {
                 log::error!("Could not read from eventfd: {}", ErrorFmt(e));
                 return;
             }
+            let n = buf.t();
             if n > 1 {
                 log::warn!("Missed {} transport changes", n - 1);
             }

src/portal.rs

@@ -5,7 +5,7 @@ mod ptr_gui;
 
 use {
     crate::{
-        async_engine::{AsyncEngine, SpawnedFuture},
+        async_engine::AsyncEngine,
         cli::GlobalArgs,
         dbus::{
             Dbus, DbusSocket, BUS_DEST, BUS_PATH, DBUS_NAME_FLAG_DO_NOT_QUEUE,
@@ -26,10 +26,7 @@ use {
         wheel::Wheel,
         wire_dbus::org,
     },
-    std::{
+    std::rc::{Rc, Weak},
-        cell::Cell,
-        rc::{Rc, Weak},
-    },
     uapi::c,
 };
 
@@ -41,12 +38,6 @@ const PORTAL_ENDED: u32 = 2;
 
 pub fn run(global: GlobalArgs) {
     logger::Logger::install_stderr(global.log_level.into());
-    let xrd = match xrd() {
-        Some(xrd) => xrd,
-        _ => {
-            fatal!("XDG_RUNTIME_DIR is not set");
-        }
-    };
     let eng = AsyncEngine::new();
     let ring = match IoUring::new(&eng, 32) {
         Ok(r) => r,
@@ -54,13 +45,26 @@ pub fn run(global: GlobalArgs) {
             fatal!("Could not create an IO-uring: {}", ErrorFmt(e));
         }
     };
+    let _f = eng.spawn(run_async(eng.clone(), ring.clone()));
+    if let Err(e) = ring.run() {
+        fatal!("The IO-uring returned an error: {}", ErrorFmt(e));
+    }
+}
+
+async fn run_async(eng: Rc<AsyncEngine>, ring: Rc<IoUring>) {
+    let xrd = match xrd() {
+        Some(xrd) => xrd,
+        _ => {
+            fatal!("XDG_RUNTIME_DIR is not set");
+        }
+    };
     let wheel = match Wheel::new(&eng, &ring) {
         Ok(w) => w,
         Err(e) => {
             fatal!("Could not create a timer wheel: {}", ErrorFmt(e));
         }
     };
-    let pw_con = match PwConHolder::new(&eng, &ring) {
+    let pw_con = match PwConHolder::new(&eng, &ring).await {
         Ok(p) => p,
         Err(e) => {
             fatal!("Could not connect to pipewire: {}", ErrorFmt(e));
@@ -68,7 +72,7 @@ pub fn run(global: GlobalArgs) {
     };
     let (_rtl_future, rtl) = RunToplevel::install(&eng);
     let dbus = Dbus::new(&eng, &ring, &rtl);
-    let dbus = init_dbus_session(&dbus);
+    let dbus = init_dbus_session(&dbus).await;
     let state = Rc::new(PortalState {
         xrd,
         ring,
@@ -76,7 +80,6 @@ pub fn run(global: GlobalArgs) {
         wheel,
         pw_con: pw_con.con.clone(),
         displays: Default::default(),
-        watch_displays: Cell::new(None),
         dbus,
         screencasts: Default::default(),
         next_id: NumCell::new(1),
@@ -90,19 +93,14 @@ pub fn run(global: GlobalArgs) {
         add_screencast_dbus_members(&state, &obj);
         obj
     };
-    state
-        .watch_displays
-        .set(Some(state.eng.spawn(watch_displays(state.clone()))));
     state.pw_con.owner.set(Some(state.clone()));
-    if let Err(e) = state.ring.run() {
+    watch_displays(state.clone()).await;
-        fatal!("The IO-uring returned an error: {}", ErrorFmt(e));
-    }
 }
 
 const UNIQUE_NAME: &str = "org.freedesktop.impl.portal.desktop.jay";
 
-fn init_dbus_session(dbus: &Dbus) -> Rc<DbusSocket> {
+async fn init_dbus_session(dbus: &Dbus) -> Rc<DbusSocket> {
-    let session = match dbus.session() {
+    let session = match dbus.session().await {
         Ok(s) => s,
         Err(e) => {
             fatal!("Could not connect to dbus session daemon: {}", ErrorFmt(e));
@@ -141,7 +139,6 @@ struct PortalState {
     wheel: Rc<Wheel>,
     pw_con: Rc<PwCon>,
     displays: CopyHashMap<PortalDisplayId, Rc<PortalDisplay>>,
-    watch_displays: Cell<Option<SpawnedFuture<()>>>,
     dbus: Rc<DbusSocket>,
     screencasts: CopyHashMap<String, Rc<ScreencastSession>>,
     next_id: NumCell<u32>,

src/portal/ptl_display.rs

@@ -234,7 +234,7 @@ impl UsrJayOutputOwner for PortalOutput {
 
 impl UsrWlOutputOwner for PortalOutput {}
 
-fn maybe_add_display(state: &Rc<PortalState>, name: &str) {
+async fn maybe_add_display(state: &Rc<PortalState>, name: &str) {
     let tail = match name.strip_prefix("wayland-") {
         Some(t) => t,
         _ => return,
@@ -248,7 +248,7 @@ fn maybe_add_display(state: &Rc<PortalState>, name: &str) {
         _ => return,
     };
     let path = format!("{}/{}", state.xrd, name);
-    let con = match UsrCon::new(&state.ring, &state.wheel, &state.eng, &path, num) {
+    let con = match UsrCon::new(&state.ring, &state.wheel, &state.eng, &path, num).await {
         Ok(c) => c,
         Err(e) => {
             log::error!(
@@ -437,7 +437,7 @@ fn add_output(dpy: &Rc<PortalDisplay>, name: u32, version: u32) {
 }
 
 pub(super) async fn watch_displays(state: Rc<PortalState>) {
-    let inotify = Rc::new(uapi::inotify_init1(c::IN_CLOEXEC | c::IN_NONBLOCK).unwrap());
+    let inotify = Rc::new(uapi::inotify_init1(c::IN_CLOEXEC).unwrap());
     if let Err(e) = uapi::inotify_add_watch(inotify.raw(), state.xrd.as_str(), c::IN_CREATE) {
         log::error!(
             "Cannot watch directory `{}`: {}",
@@ -462,7 +462,7 @@ pub(super) async fn watch_displays(state: Rc<PortalState>) {
             }
         };
         if let Ok(s) = std::str::from_utf8(entry.file_name().as_bytes()) {
-            maybe_add_display(&state, s);
+            maybe_add_display(&state, s).await;
         }
     }
     let mut buf = vec![0u8; 4096];
@@ -480,7 +480,7 @@ pub(super) async fn watch_displays(state: Rc<PortalState>) {
         for event in events {
             if event.mask.contains(c::IN_CREATE) {
                 if let Ok(s) = std::str::from_utf8(event.name().as_ustr().as_bytes()) {
-                    maybe_add_display(&state, s);
+                    maybe_add_display(&state, s).await;
                 }
             }
         }

src/sighand.rs

@@ -2,11 +2,11 @@ use {
     crate::{
         async_engine::{AsyncEngine, SpawnedFuture},
         io_uring::IoUring,
-        utils::{errorfmt::ErrorFmt, oserror::OsError},
+        utils::{buf::TypedBuf, errorfmt::ErrorFmt, oserror::OsError},
     },
     std::rc::Rc,
     thiserror::Error,
-    uapi::{c, Errno, OwnedFd},
+    uapi::{c, OwnedFd},
 };
 
 #[derive(Debug, Error)]
@@ -28,7 +28,7 @@ pub fn install(
     if let Err(e) = uapi::pthread_sigmask(c::SIG_BLOCK, Some(&set), None) {
         return Err(SighandError::BlockFailed(e.into()));
     }
-    let fd = match uapi::signalfd_new(&set, c::SFD_CLOEXEC | c::SFD_NONBLOCK) {
+    let fd = match uapi::signalfd_new(&set, c::SFD_CLOEXEC) {
         Ok(fd) => Rc::new(fd),
         Err(e) => return Err(SighandError::CreateFailed(e.into())),
     };
@@ -36,34 +36,17 @@ pub fn install(
 }
 
 async fn handle_signals(fd: Rc<OwnedFd>, ring: Rc<IoUring>) {
-    let mut siginfo: c::signalfd_siginfo = uapi::pod_zeroed();
+    let mut buf = TypedBuf::<c::signalfd_siginfo>::new();
     loop {
-        if let Err(e) = ring.readable(&fd).await {
+        if let Err(e) = ring.read(&fd, buf.buf()).await {
-            log::error!(
+            log::error!("Could not read from signal fd: {}", ErrorFmt(e));
-                "Could not wait for signal fd to become readable: {}",
-                ErrorFmt(e)
-            );
             return;
         }
-        loop {
+        let sig = buf.t().ssi_signo as i32;
-            if let Err(e) = uapi::read(fd.raw(), &mut siginfo) {
+        log::info!("Received signal {}", sig);
-                match e {
+        if matches!(sig, c::SIGINT | c::SIGTERM) {
-                    Errno(c::EAGAIN) => break,
+            log::info!("Exiting");
-                    _ => {
+            ring.stop();
-                        log::error!(
-                            "Could not read from signal fd: {}",
-                            ErrorFmt(OsError::from(e))
-                        );
-                        return;
-                    }
-                }
-            }
-            let sig = siginfo.ssi_signo as i32;
-            log::info!("Received signal {}", sig);
-            if matches!(sig, c::SIGINT | c::SIGTERM) {
-                log::info!("Exiting");
-                ring.stop();
-            }
         }
     }
 }

src/tools/tool_client.rs

@@ -56,7 +56,7 @@ pub enum ToolClientError {
     #[error("The socket path is too long")]
     SocketPathTooLong,
     #[error("Could not connect to the compositor")]
-    Connect(#[source] OsError),
+    Connect(#[source] IoUringError),
     #[error("The message length is smaller than 8 bytes")]
     MsgLenTooSmall,
     #[error("The size of the message is not a multiple of 4")]
@@ -94,36 +94,53 @@ pub struct ToolClient {
     jay_compositor: Cell<Option<JayCompositorId>>,
 }
 
-impl ToolClient {
+pub fn with_tool_client<T, F>(level: Level, f: F)
-    pub fn new(level: Level) -> Rc<Self> {
+where
-        match Self::try_new(level) {
+    F: FnOnce(Rc<ToolClient>) -> T + 'static,
-            Ok(s) => s,
+    T: Future<Output = ()> + 'static,
-            Err(e) => {
+{
-                fatal!("Could not create a tool client: {}", ErrorFmt(e));
+    if let Err(e) = with_tool_client_(level, f) {
-            }
+        handle_error(e);
-        }
     }
+}
 
-    pub fn run<F>(&self, f: F)
+fn handle_error(e: ToolClientError) -> ! {
-    where
+    fatal!("Could not create a tool client: {}", ErrorFmt(e));
-        F: Future<Output = ()> + 'static,
+}
-    {
-        let _future = self.eng.spawn(async move {
-            f.await;
-            std::process::exit(0);
-        });
-        if let Err(e) = self.ring.run() {
-            fatal!("A fatal error occurred: {}", ErrorFmt(e));
-        }
-    }
 
-    pub fn try_new(level: Level) -> Result<Rc<Self>, ToolClientError> {
+fn with_tool_client_<T, F>(level: Level, f: F) -> Result<(), ToolClientError>
-        let logger = Logger::install_stderr(level);
+where
-        let eng = AsyncEngine::new();
+    F: FnOnce(Rc<ToolClient>) -> T + 'static,
-        let ring = match IoUring::new(&eng, 32) {
+    T: Future<Output = ()> + 'static,
-            Ok(e) => e,
+{
-            Err(e) => return Err(ToolClientError::CreateRing(e)),
+    let logger = Logger::install_stderr(level);
+    let eng = AsyncEngine::new();
+    let ring = match IoUring::new(&eng, 32) {
+        Ok(e) => e,
+        Err(e) => return Err(ToolClientError::CreateRing(e)),
+    };
+    let eng2 = eng.clone();
+    let ring2 = ring.clone();
+    let _f = eng.spawn(async move {
+        let tc = match ToolClient::try_new(logger, eng2, ring2).await {
+            Ok(tc) => tc,
+            Err(e) => handle_error(e),
         };
+        f(tc).await;
+        std::process::exit(0);
+    });
+    if let Err(e) = ring.run() {
+        fatal!("A fatal error occurred: {}", ErrorFmt(e));
+    }
+    Ok(())
+}
+
+impl ToolClient {
+    async fn try_new(
+        logger: Arc<Logger>,
+        eng: Rc<AsyncEngine>,
+        ring: Rc<IoUring>,
+    ) -> Result<Rc<Self>, ToolClientError> {
         let wheel = match Wheel::new(&eng, &ring) {
             Ok(w) => w,
             Err(e) => return Err(ToolClientError::CreateWheel(e)),
@@ -137,11 +154,7 @@ impl ToolClient {
             Err(_) => return Err(ToolClientError::WaylandDisplayNotSet),
         };
         let path = format_ustr!("{}/{}.jay", xrd, wd);
-        let socket = match uapi::socket(
+        let socket = match uapi::socket(c::AF_UNIX, c::SOCK_STREAM | c::SOCK_CLOEXEC, 0) {
-            c::AF_UNIX,
-            c::SOCK_STREAM | c::SOCK_CLOEXEC | c::SOCK_NONBLOCK,
-            0,
-        ) {
             Ok(s) => Rc::new(s),
             Err(e) => return Err(ToolClientError::CreateSocket(e.into())),
         };
@@ -153,8 +166,8 @@ impl ToolClient {
         let sun_path = uapi::as_bytes_mut(&mut addr.sun_path[..]);
         sun_path[..path.len()].copy_from_slice(path.as_bytes());
         sun_path[path.len()] = 0;
-        if let Err(e) = uapi::connect(socket.raw(), &addr) {
+        if let Err(e) = ring.connect(&socket, &addr).await {
-            return Err(ToolClientError::Connect(e.into()));
+            return Err(ToolClientError::Connect(e));
         }
         let mut obj_ids = Bitfield::default();
         obj_ids.take(0);

src/user_session.rs

@@ -18,8 +18,8 @@ pub enum UserSessionError {
     AcquireSessionBus(#[source] DbusError),
 }
 
-pub fn import_environment(state: &Rc<State>, key: &str, value: &str) {
+pub async fn import_environment(state: &Rc<State>, key: &str, value: &str) {
-    if let Err(e) = import_environment_(state, key, value) {
+    if let Err(e) = import_environment_(state, key, value).await {
         log::error!(
             "Could not import `{}={}` into the system environment: {}",
             key,
@@ -29,8 +29,12 @@ pub fn import_environment(state: &Rc<State>, key: &str, value: &str) {
     }
 }
 
-fn import_environment_(state: &Rc<State>, key: &str, value: &str) -> Result<(), UserSessionError> {
+async fn import_environment_(
-    let session = match state.dbus.session() {
+    state: &Rc<State>,
+    key: &str,
+    value: &str,
+) -> Result<(), UserSessionError> {
+    let session = match state.dbus.session().await {
         Ok(s) => s,
         Err(e) => return Err(UserSessionError::AcquireSessionBus(e)),
     };

src/utils/buf.rs

@@ -2,17 +2,25 @@ use {
     crate::utils::{numcell::NumCell, ptr_ext::PtrExt},
     std::{
         alloc::Layout,
+        cmp,
         collections::Bound,
+        fmt::Arguments,
+        io::{self, Write},
+        marker::PhantomData,
+        mem,
         ops::{Deref, DerefMut, Range, RangeBounds},
         ptr::NonNull,
         slice,
     },
+    uapi::Pod,
 };
 
 const METADATA_SIZE: u32 = 8;
 const METADATA_ALIGN: usize = 4;
+const SIZE_OFF: u32 = 0;
 const RC_OFF: u32 = 4;
 const RC_OFF_INV: u32 = METADATA_SIZE - RC_OFF;
+const SIZE_OFF_INV: u32 = METADATA_SIZE - SIZE_OFF;
 
 pub struct Buf {
     storage: NonNull<u8>,
@@ -118,9 +126,55 @@ impl Buf {
         self.len32() as _
     }
 
+    fn size32(&self) -> u32 {
+        unsafe {
+            *self
+                .storage
+                .as_ptr()
+                .sub(SIZE_OFF_INV as _)
+                .cast::<u32>()
+                .deref()
+        }
+    }
+
+    pub fn cap32(&self) -> u32 {
+        self.size32() - METADATA_SIZE
+    }
+
     pub fn as_ptr(&self) -> *mut u8 {
         unsafe { self.storage.as_ptr().add(self.range.start as _) }
     }
+
+    pub fn write_fmt(&mut self, args: Arguments) -> Result<Self, io::Error> {
+        let cap = self.len();
+        let mut buf = self.deref_mut();
+        buf.write_fmt(args)?;
+        let len = cap - buf.len();
+        Ok(self.slice(..len))
+    }
+
+    pub fn into_full(self) -> Self {
+        let new = Self {
+            storage: self.storage,
+            range: 0..self.cap32(),
+        };
+        mem::forget(self);
+        new
+    }
+
+    fn as_slice(&self) -> &[u8] {
+        unsafe { slice::from_raw_parts(self.as_ptr(), self.len()) }
+    }
+
+    fn as_slice_mut(&mut self) -> &mut [u8] {
+        unsafe { slice::from_raw_parts_mut(self.as_ptr(), self.len()) }
+    }
+}
+
+impl Default for Buf {
+    fn default() -> Self {
+        Self::new(0)
+    }
 }
 
 impl Deref for Buf {
@@ -128,14 +182,14 @@ impl Deref for Buf {
 
     fn deref(&self) -> &Self::Target {
         self.assert_unique();
-        unsafe { slice::from_raw_parts(self.as_ptr(), self.len()) }
+        self.as_slice()
     }
 }
 
 impl DerefMut for Buf {
     fn deref_mut(&mut self) -> &mut Self::Target {
         self.assert_unique();
-        unsafe { slice::from_raw_parts_mut(self.as_ptr(), self.len()) }
+        self.as_slice_mut()
     }
 }
 
@@ -153,3 +207,122 @@ impl Drop for Buf {
         }
     }
 }
+
+pub struct DynamicBuf {
+    buf: Buf,
+    len: usize,
+}
+
+impl DynamicBuf {
+    pub fn new() -> Self {
+        Self {
+            buf: Buf::new(0),
+            len: 0,
+        }
+    }
+
+    pub fn from_buf(buf: Buf) -> Self {
+        buf.assert_unique();
+        Self {
+            buf: buf.into_full(),
+            len: 0,
+        }
+    }
+
+    pub fn unwrap(mut self) -> Buf {
+        self.buf.slice(..self.len)
+    }
+
+    pub fn len(&self) -> usize {
+        self.len
+    }
+
+    pub fn reserve(&mut self, n: usize) {
+        if self.buf.len() - self.len < n {
+            let cap = self.len.checked_add(n).unwrap();
+            let cap = cmp::max(self.buf.len() * 2, cap);
+            let mut new = Buf::new(cap);
+            new[..self.len].copy_from_slice(&self.buf[..self.len]);
+            self.buf = new;
+        }
+    }
+
+    pub fn extend_from_slice(&mut self, buf: &[u8]) {
+        self.reserve(buf.len());
+        self.buf.as_slice_mut()[self.len..self.len + buf.len()].copy_from_slice(buf);
+        self.len += buf.len();
+    }
+
+    pub fn push(&mut self, b: u8) {
+        self.extend_from_slice(&[b]);
+    }
+
+    pub fn clear(&mut self) {
+        self.len = 0;
+    }
+
+    pub fn borrow(&mut self) -> BorrowedBuf<'_> {
+        BorrowedBuf {
+            buf: self.buf.slice(..self.len),
+            _phantom: Default::default(),
+        }
+    }
+}
+
+pub struct BorrowedBuf<'a> {
+    pub buf: Buf,
+    _phantom: PhantomData<&'a mut DynamicBuf>,
+}
+
+impl<'a> Drop for BorrowedBuf<'a> {
+    fn drop(&mut self) {
+        assert_eq!(self.buf.rc().get(), 2);
+    }
+}
+
+impl Write for DynamicBuf {
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        self.extend_from_slice(buf);
+        Ok(buf.len())
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        Ok(())
+    }
+}
+
+impl Deref for DynamicBuf {
+    type Target = [u8];
+
+    fn deref(&self) -> &Self::Target {
+        self.buf.as_slice()
+    }
+}
+
+impl DerefMut for DynamicBuf {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        self.buf.as_slice_mut()
+    }
+}
+
+pub struct TypedBuf<T: Pod> {
+    buf: Buf,
+    _phantom: PhantomData<T>,
+}
+
+impl<T: Pod> TypedBuf<T> {
+    pub fn new() -> Self {
+        Self {
+            buf: Buf::new(mem::size_of::<T>()),
+            _phantom: Default::default(),
+        }
+    }
+
+    pub fn buf(&mut self) -> Buf {
+        self.buf.clone()
+    }
+
+    pub fn t(&self) -> T {
+        uapi::pod_read(&self.buf[..]).unwrap()
+    }
+}

src/utils/buffd.rs

@@ -1,7 +1,4 @@
-use {
+use {crate::io_uring::IoUringError, thiserror::Error};
-    crate::{io_uring::IoUringError, utils::oserror::OsError},
-    thiserror::Error,
-};
 pub use {
     buf_in::BufFdIn,
     buf_out::{BufFdOut, OutBuffer, OutBufferSwapchain},
@@ -17,7 +14,7 @@ mod parser;
 #[derive(Debug, Error)]
 pub enum BufFdError {
     #[error("An IO error occurred")]
-    Io(#[source] OsError),
+    Io(#[source] IoUringError),
     #[error("An io-uring error occurred")]
     Ring(#[from] IoUringError),
     #[error("The peer did not send a file descriptor")]
@@ -31,5 +28,4 @@ pub enum BufFdError {
 }
 
 const BUF_SIZE: usize = 4096;
-const CMSG_BUF_SIZE: usize = 4096;
 const MAX_IN_FD: usize = 32;

src/utils/buffd/buf_in.rs

@@ -15,7 +15,7 @@ pub struct BufFdIn {
     fd: Rc<OwnedFd>,
     ring: Rc<IoUring>,
 
-    in_fd: VecDeque<OwnedFd>,
+    in_fd: VecDeque<Rc<OwnedFd>>,
 
     in_buf: Buf,
     in_left: usize,
@@ -86,7 +86,7 @@ impl BufFdIn {
         Ok(())
     }
 
-    pub fn get_fd(&mut self) -> Result<OwnedFd, BufFdError> {
+    pub fn get_fd(&mut self) -> Result<Rc<OwnedFd>, BufFdError> {
         match self.in_fd.pop_front() {
             Some(f) => Ok(f),
             None => Err(BufFdError::NoFd),

src/utils/buffd/buf_out.rs

@@ -4,17 +4,16 @@ use {
         time::Time,
         utils::{
             buf::Buf,
-            buffd::{BufFdError, BUF_SIZE, CMSG_BUF_SIZE},
+            buffd::{BufFdError, BUF_SIZE},
             oserror::OsError,
         },
     },
     std::{
         collections::VecDeque,
-        mem::{self, MaybeUninit},
+        mem::{self},
         rc::Rc,
-        slice,
     },
-    uapi::{c, Errno, OwnedFd},
+    uapi::{c, OwnedFd},
 };
 
 pub(super) const OUT_BUF_SIZE: usize = 2 * BUF_SIZE;
@@ -80,8 +79,6 @@ impl OutBufferSwapchain {
 pub struct BufFdOut {
     fd: Rc<OwnedFd>,
     ring: Rc<IoUring>,
-    cmsg_buf: Box<[MaybeUninit<u8>; CMSG_BUF_SIZE]>,
-    fd_ids: Vec<i32>,
 }
 
 impl BufFdOut {
@@ -89,8 +86,6 @@ impl BufFdOut {
         Self {
             fd: fd.clone(),
             ring: ring.clone(),
-            cmsg_buf: Box::new([MaybeUninit::uninit(); CMSG_BUF_SIZE]),
-            fd_ids: vec![],
         }
     }
 
@@ -131,7 +126,7 @@ impl BufFdOut {
                 buf = buffer.buf.slice(buffer.meta.read_pos..next_pos);
             }
         }
-        match self.ring.sendmsg(&self.fd, buf, fds, timeout).await {
+        match self.ring.sendmsg_one(&self.fd, buf, fds, timeout).await {
             Ok(n) => {
                 buffer.meta.read_pos += n;
                 Ok(())
@@ -144,59 +139,23 @@ impl BufFdOut {
 
     pub async fn flush2(
         &mut self,
-        buf: &[u8],
+        mut buf: Buf,
-        fds: &mut Vec<Rc<OwnedFd>>,
+        mut fds: Vec<Rc<OwnedFd>>,
     ) -> Result<(), BufFdError> {
         let mut read_pos = 0;
         while read_pos < buf.len() {
-            if self.flush_sync2(&mut read_pos, buf, fds)? {
+            let res = self
-                self.ring.writable(&self.fd).await?;
+                .ring
+                .sendmsg_one(&self.fd, buf.slice(read_pos..), mem::take(&mut fds), None)
+                .await;
+            match res {
+                Ok(n) => read_pos += n,
+                Err(IoUringError::OsError(OsError(c::ECONNRESET))) => {
+                    return Err(BufFdError::Closed)
+                }
+                Err(e) => return Err(BufFdError::Io(e)),
             }
         }
         Ok(())
     }
-
-    fn flush_sync2(
-        &mut self,
-        read_pos: &mut usize,
-        buf: &[u8],
-        fds: &mut Vec<Rc<OwnedFd>>,
-    ) -> Result<bool, BufFdError> {
-        let mut cmsg_len = 0;
-        let mut fds_opt = None;
-        if fds.len() > 0 {
-            self.fd_ids.clear();
-            self.fd_ids.extend(fds.iter().map(|f| f.raw()));
-            let hdr = c::cmsghdr {
-                cmsg_len: 0,
-                cmsg_level: c::SOL_SOCKET,
-                cmsg_type: c::SCM_RIGHTS,
-            };
-            let mut cmsg_buf = &mut self.cmsg_buf[..];
-            cmsg_len = uapi::cmsg_write(&mut cmsg_buf, hdr, &self.fd_ids[..]).unwrap();
-            fds_opt = Some(fds);
-        }
-        while *read_pos < buf.len() {
-            let buf = &buf[*read_pos..];
-            let hdr = uapi::Msghdr {
-                iov: slice::from_ref(&buf),
-                control: Some(&self.cmsg_buf[..cmsg_len]),
-                name: uapi::sockaddr_none_ref(),
-            };
-            let bytes_sent =
-                match uapi::sendmsg(self.fd.raw(), &hdr, c::MSG_DONTWAIT | c::MSG_NOSIGNAL) {
-                    Ok(b) => {
-                        if let Some(fds) = fds_opt.take() {
-                            fds.clear();
-                        }
-                        b
-                    }
-                    Err(Errno(c::EAGAIN)) => return Ok(true),
-                    Err(Errno(c::ECONNRESET)) => return Err(BufFdError::Closed),
-                    Err(e) => return Err(BufFdError::Io(e.into())),
-                };
-            *read_pos += bytes_sent;
-        }
-        Ok(false)
-    }
 }

src/utils/buffd/parser.rs

@@ -97,7 +97,7 @@ impl<'a, 'b> MsgParser<'a, 'b> {
 
     pub fn fd(&mut self) -> Result<Rc<OwnedFd>, MsgParserError> {
         match self.buf.get_fd() {
-            Ok(fd) => Ok(Rc::new(fd)),
+            Ok(fd) => Ok(fd),
             _ => Err(MsgParserError::MissingFd),
         }
     }

src/utils/bufio.rs

@@ -2,40 +2,33 @@ use {
     crate::{
         io_uring::{IoUring, IoUringError},
         utils::{
-            oserror::OsError,
+            buf::{Buf, DynamicBuf},
             queue::AsyncQueue,
             stack::Stack,
-            vec_ext::{UninitVecExt, VecExt},
-            vecstorage::VecStorage,
         },
     },
     std::{
         collections::VecDeque,
-        mem::{self, MaybeUninit},
+        mem::{self},
-        ptr::NonNull,
         rc::Rc,
     },
     thiserror::Error,
-    uapi::{c, Errno, MaybeUninitSliceExt, Msghdr, MsghdrMut, OwnedFd},
+    uapi::{c, OwnedFd},
 };
 
 #[derive(Debug, Error)]
 pub enum BufIoError {
     #[error("Could not write to the socket")]
-    FlushError(#[source] OsError),
+    FlushError(#[source] IoUringError),
     #[error("Could not read from the socket")]
-    ReadError(#[source] OsError),
+    ReadError(#[source] IoUringError),
-    #[error("Cannot wait for fd to become writable")]
-    Writable(#[source] IoUringError),
-    #[error("Cannot wait for fd to become readable")]
-    Readable(#[source] IoUringError),
     #[error("The socket is closed")]
     Closed,
 }
 
 pub struct BufIoMessage {
     pub fds: Vec<Rc<OwnedFd>>,
-    pub buf: Vec<u8>,
+    pub buf: Buf,
 }
 
 struct MessageOffset {
@@ -46,27 +39,24 @@ struct MessageOffset {
 pub struct BufIo {
     fd: Rc<OwnedFd>,
     ring: Rc<IoUring>,
-    bufs: Stack<Vec<u8>>,
+    bufs: Stack<Buf>,
     outgoing: AsyncQueue<BufIoMessage>,
 }
 
 pub struct BufIoIncoming {
     bufio: Rc<BufIo>,
 
-    buf: Box<[MaybeUninit<u8>; 4096]>,
+    buf: Buf,
     buf_start: usize,
     buf_end: usize,
     pub fds: VecDeque<Rc<OwnedFd>>,
-    cmsg: Box<[MaybeUninit<u8>; 256]>,
 }
 
 struct Outgoing {
     bufio: Rc<BufIo>,
 
     msgs: VecDeque<MessageOffset>,
-    cmsg: Vec<MaybeUninit<u8>>,
+    bufs: Vec<Buf>,
-    fds: Vec<c::c_int>,
-    iovecs: VecStorage<NonNull<[u8]>>,
 }
 
 impl BufIo {
@@ -83,14 +73,9 @@ impl BufIo {
         let _ = uapi::shutdown(self.fd.raw(), c::SHUT_RDWR);
     }
 
-    pub fn buf(&self) -> Vec<u8> {
+    pub fn buf(&self) -> DynamicBuf {
-        let mut buf = self.bufs.pop().unwrap_or_default();
+        let buf = self.bufs.pop().unwrap_or_default();
-        buf.clear();
+        DynamicBuf::from_buf(buf)
-        buf
-    }
-
-    pub fn add_buf(&self, buf: Vec<u8>) {
-        self.bufs.push(buf);
     }
 
     pub fn send(&self, msg: BufIoMessage) {
@@ -101,9 +86,7 @@ impl BufIo {
         let mut outgoing = Outgoing {
             bufio: self,
             msgs: Default::default(),
-            cmsg: vec![],
+            bufs: vec![],
-            fds: vec![],
-            iovecs: Default::default(),
         };
         outgoing.run().await
     }
@@ -111,11 +94,10 @@ impl BufIo {
     pub fn incoming(self: &Rc<Self>) -> BufIoIncoming {
         BufIoIncoming {
             bufio: self.clone(),
-            buf: Box::new([MaybeUninit::uninit(); 4096]),
+            buf: Buf::new(4096),
             buf_start: 0,
             buf_end: 0,
             fds: Default::default(),
-            cmsg: Box::new([MaybeUninit::uninit(); 256]),
         }
     }
 }
@@ -128,72 +110,42 @@ impl BufIoIncoming {
     ) -> Result<(), BufIoError> {
         while n > 0 {
             if self.buf_start == self.buf_end {
-                while let Err(e) = self.recvmsg() {
+                self.buf_start = 0;
-                    if e.0 != c::EAGAIN {
+                self.buf_end = 0;
-                        return Err(BufIoError::ReadError(e.into()));
+                let res = self
-                    }
+                    .bufio
-                    if let Err(e) = self.bufio.ring.readable(&self.bufio.fd).await {
+                    .ring
-                        return Err(BufIoError::Readable(e));
+                    .recvmsg(&self.bufio.fd, &mut [self.buf.clone()], &mut self.fds)
-                    }
+                    .await;
+                match res {
+                    Ok(n) => self.buf_end = n,
+                    Err(e) => return Err(BufIoError::ReadError(e)),
                 }
                 if self.buf_start == self.buf_end {
                     return Err(BufIoError::Closed);
                 }
             }
             let read = n.min(self.buf_end - self.buf_start);
-            let buf_start = self.buf_start % self.buf.len();
+            let buf_start = self.buf_start;
-            unsafe {
+            buf.extend_from_slice(&self.buf[buf_start..buf_start + read]);
-                buf.extend_from_slice(
-                    self.buf[buf_start..buf_start + read].slice_assume_init_ref(),
-                );
-            }
             n -= read;
             self.buf_start += read;
         }
         Ok(())
     }
-
-    fn recvmsg(&mut self) -> Result<(), Errno> {
-        self.buf_start = 0;
-        self.buf_end = 0;
-        let mut iov = [&mut self.buf[..]];
-        let mut hdr = MsghdrMut {
-            iov: &mut iov[..],
-            control: Some(&mut self.cmsg[..]),
-            name: uapi::sockaddr_none_mut(),
-            flags: 0,
-        };
-        let (ivec, _, mut cmsg) =
-            uapi::recvmsg(self.bufio.fd.raw(), &mut hdr, c::MSG_CMSG_CLOEXEC)?;
-        self.buf_end += ivec.len();
-        while cmsg.len() > 0 {
-            let (_, hdr, body) = uapi::cmsg_read(&mut cmsg)?;
-            if hdr.cmsg_level == c::SOL_SOCKET && hdr.cmsg_type == c::SCM_RIGHTS {
-                for fd in uapi::pod_iter(body)? {
-                    self.fds.push_back(Rc::new(OwnedFd::new(fd)));
-                }
-            }
-        }
-        Ok(())
-    }
 }
 
 impl Outgoing {
     async fn run(&mut self) -> Result<(), BufIoError> {
         loop {
             self.bufio.outgoing.non_empty().await;
-            while let Err(e) = self.try_flush() {
+            if let Err(e) = self.try_flush().await {
-                if e != Errno(c::EAGAIN) {
+                return Err(BufIoError::FlushError(e));
-                    return Err(BufIoError::FlushError(e.into()));
-                }
-                if let Err(e) = self.bufio.ring.writable(&self.bufio.fd).await {
-                    return Err(BufIoError::Writable(e));
-                }
             }
         }
     }
 
-    fn try_flush(&mut self) -> Result<(), Errno> {
+    async fn try_flush(&mut self) -> Result<(), IoUringError> {
         loop {
             while let Some(msg) = self.bufio.outgoing.try_pop() {
                 self.msgs.push_back(MessageOffset { msg, offset: 0 });
@@ -201,40 +153,23 @@ impl Outgoing {
             if self.msgs.is_empty() {
                 return Ok(());
             }
-            let mut iovecs = self.iovecs.take_as();
+            let mut fds = Vec::new();
-            let mut fds = &[][..];
             for msg in &mut self.msgs {
                 if msg.msg.fds.len() > 0 {
-                    if fds.len() > 0 || iovecs.len() > 0 {
+                    if fds.len() > 0 || self.bufs.len() > 0 {
                         break;
                     }
-                    fds = &msg.msg.fds;
+                    fds = mem::take(&mut msg.msg.fds);
                 }
-                iovecs.push(&msg.msg.buf[msg.offset..]);
+                self.bufs.push(msg.msg.buf.slice(msg.offset..));
             }
-            self.cmsg.clear();
+            let res = self
-            if fds.len() > 0 {
+                .bufio
-                self.fds.clear();
+                .ring
-                self.fds.extend(fds.iter().map(|f| f.raw()));
+                .sendmsg(&self.bufio.fd, &mut self.bufs, fds, None)
-                let cmsg_space = uapi::cmsg_space(fds.len() * mem::size_of::<c::c_int>());
+                .await;
-                self.cmsg.reserve(cmsg_space);
+            self.bufs.clear();
-                let (_, mut spare) = self.cmsg.split_at_spare_mut_bytes_ext();
+            let mut n = res?;
-                let hdr = c::cmsghdr {
-                    cmsg_len: 0,
-                    cmsg_level: c::SOL_SOCKET,
-                    cmsg_type: c::SCM_RIGHTS,
-                };
-                let len = uapi::cmsg_write(&mut spare, hdr, &self.fds[..]).unwrap();
-                self.cmsg.set_len_safe(len);
-            }
-            let msg = Msghdr {
-                iov: &iovecs[..],
-                control: Some(&self.cmsg[..]),
-                name: uapi::sockaddr_none_ref(),
-            };
-            let mut n = uapi::sendmsg(self.bufio.fd.raw(), &msg, c::MSG_DONTWAIT)?;
-            drop(iovecs);
-            self.msgs[0].msg.fds.clear();
             while n > 0 {
                 let len = self.msgs[0].msg.buf.len() - self.msgs[0].offset;
                 if n < len {

src/utils/timer.rs

@@ -1,9 +1,9 @@
 use {
     crate::{
         io_uring::{IoUring, IoUringError},
-        utils::oserror::OsError,
+        utils::{buf::TypedBuf, oserror::OsError},
     },
-    std::{rc::Rc, time::Duration},
+    std::{cell::RefCell, rc::Rc, time::Duration},
     thiserror::Error,
     uapi::{c, OwnedFd},
 };
@@ -13,7 +13,7 @@ pub enum TimerError {
     #[error("Could not create a timer")]
     CreateTimer(#[source] OsError),
     #[error("Could not read from a timer")]
-    TimerReadError(#[source] OsError),
+    TimerReadError(#[source] IoUringError),
     #[error("Could not set a timer")]
     SetTimer(#[source] OsError),
     #[error("The io-uring returned an error")]
@@ -23,24 +23,28 @@ pub enum TimerError {
 #[derive(Clone)]
 pub struct TimerFd {
     fd: Rc<OwnedFd>,
+    buf: Rc<RefCell<TypedBuf<u64>>>,
 }
 
 impl TimerFd {
     pub fn new(clock_id: c::c_int) -> Result<Self, TimerError> {
-        let fd = match uapi::timerfd_create(clock_id, c::TFD_CLOEXEC | c::TFD_NONBLOCK) {
+        let fd = match uapi::timerfd_create(clock_id, c::TFD_CLOEXEC) {
             Ok(fd) => Rc::new(fd),
             Err(e) => return Err(TimerError::CreateTimer(e.into())),
         };
-        Ok(Self { fd })
+        Ok(Self {
+            fd,
+            buf: Rc::new(RefCell::new(TypedBuf::new())),
+        })
     }
 
+    #[allow(clippy::await_holding_refcell_ref)]
     pub async fn expired(&self, ring: &IoUring) -> Result<u64, TimerError> {
-        ring.readable(&self.fd).await?;
+        let mut buf = self.buf.borrow_mut();
-        let mut buf = 0u64;
+        if let Err(e) = ring.read(&self.fd, buf.buf()).await {
-        if let Err(e) = uapi::read(self.fd.raw(), &mut buf) {
+            return Err(TimerError::TimerReadError(e));
-            return Err(TimerError::TimerReadError(e.into()));
         }
-        Ok(buf)
+        Ok(buf.t())
     }
 
     pub fn program(

src/video/drm.rs

@@ -26,12 +26,13 @@ use {
         rc::{Rc, Weak},
     },
     thiserror::Error,
-    uapi::{c, Errno, OwnedFd, Pod, Ustring},
+    uapi::{c, OwnedFd, Pod, Ustring},
 };
 
 use crate::{
     backend,
-    utils::{errorfmt::ErrorFmt, stack::Stack, syncqueue::SyncQueue, vec_ext::VecExt},
+    io_uring::{IoUring, IoUringError},
+    utils::{buf::Buf, errorfmt::ErrorFmt, stack::Stack, syncqueue::SyncQueue, vec_ext::VecExt},
     video::{
         dmabuf::DmaBuf,
         drm::sys::{get_version, DRM_CAP_CURSOR_HEIGHT, DRM_CAP_CURSOR_WIDTH},
@@ -100,7 +101,7 @@ pub enum DrmError {
     #[error("Could not convert prime fd to gem handle")]
     GemHandle(#[source] OsError),
     #[error("Could not read events from the drm fd")]
-    ReadEvents(#[source] OsError),
+    ReadEvents(#[source] IoUringError),
     #[error("Read invalid data from drm device")]
     InvalidRead,
     #[error("Could not determine the drm version")]
@@ -179,7 +180,8 @@ pub struct DrmMaster {
     u64_bufs: Stack<Vec<u64>>,
     gem_handles: RefCell<AHashMap<u32, Weak<GemHandle>>>,
     events: SyncQueue<DrmEvent>,
-    buf: RefCell<Box<[MaybeUninit<u8>; 1024]>>,
+    ring: Rc<IoUring>,
+    buf: RefCell<Buf>,
 }
 
 impl Debug for DrmMaster {
@@ -197,14 +199,15 @@ impl Deref for DrmMaster {
 }
 
 impl DrmMaster {
-    pub fn new(fd: Rc<OwnedFd>) -> Self {
+    pub fn new(ring: &Rc<IoUring>, fd: Rc<OwnedFd>) -> Self {
         Self {
             drm: Drm { fd },
             u32_bufs: Default::default(),
             u64_bufs: Default::default(),
             gem_handles: Default::default(),
             events: Default::default(),
-            buf: RefCell::new(Box::new([MaybeUninit::uninit(); 1024])),
+            ring: ring.clone(),
+            buf: RefCell::new(Buf::new(1024)),
         }
     }
 
@@ -370,13 +373,13 @@ impl DrmMaster {
         }
     }
 
-    pub fn event(&self) -> Result<Option<DrmEvent>, DrmError> {
+    #[allow(clippy::await_holding_refcell_ref)]
+    pub async fn event(&self) -> Result<Option<DrmEvent>, DrmError> {
         if self.events.is_empty() {
             let mut buf = self.buf.borrow_mut();
-            let mut buf = match uapi::read(self.raw(), buf.as_mut_slice()) {
+            let mut buf = match self.ring.read(self.drm.fd(), buf.clone()).await {
-                Ok(b) => b,
+                Ok(n) => &buf[..n],
-                Err(Errno(c::EAGAIN)) => return Ok(None),
+                Err(e) => return Err(DrmError::ReadEvents(e)),
-                Err(e) => return Err(DrmError::ReadEvents(e.into())),
             };
             while buf.len() > 0 {
                 let header: drm_event = match uapi::pod_read_init(buf) {
@@ -402,7 +405,7 @@ impl DrmMaster {
                     }
                     _ => {}
                 }
-                buf = &mut buf[len as usize..];
+                buf = &buf[len as usize..];
             }
         }
         Ok(self.events.pop())

src/wheel.rs

@@ -1,11 +1,11 @@
 use {
     crate::{
         async_engine::{AsyncEngine, SpawnedFuture},
-        io_uring::IoUring,
+        io_uring::{IoUring, IoUringError},
         time::{Time, TimeError},
         utils::{
-            copyhashmap::CopyHashMap, errorfmt::ErrorFmt, numcell::NumCell, oserror::OsError,
+            buf::TypedBuf, copyhashmap::CopyHashMap, errorfmt::ErrorFmt, numcell::NumCell,
-            stack::Stack,
+            oserror::OsError, stack::Stack,
         },
     },
     std::{
@@ -33,7 +33,7 @@ pub enum WheelError {
     #[error("The timer wheel is already destroyed")]
     Destroyed,
     #[error("Could not read from the timerfd")]
-    Read(#[source] OsError),
+    Read(#[source] IoUringError),
 }
 
 #[derive(Debug, Eq, PartialEq, Ord, PartialOrd)]
@@ -111,7 +111,7 @@ pub struct WheelData {
 
 impl Wheel {
     pub fn new(eng: &Rc<AsyncEngine>, ring: &Rc<IoUring>) -> Result<Rc<Self>, WheelError> {
-        let fd = match uapi::timerfd_create(c::CLOCK_MONOTONIC, c::TFD_CLOEXEC | c::TFD_NONBLOCK) {
+        let fd = match uapi::timerfd_create(c::CLOCK_MONOTONIC, c::TFD_CLOEXEC) {
             Ok(fd) => Rc::new(fd),
             Err(e) => return Err(WheelError::CreateFailed(e.into())),
         };
@@ -210,16 +210,9 @@ impl WheelData {
     }
 
     async fn dispatch(self: Rc<Self>) {
+        let mut n = TypedBuf::new();
         loop {
-            if let Err(e) = self.ring.readable(&self.fd).await {
+            if let Err(e) = self.dispatch_once(&mut n).await {
-                log::error!(
-                    "Could not wait for the timerfd to become readable: {}",
-                    ErrorFmt(e)
-                );
-                self.kill();
-                return;
-            }
-            if let Err(e) = self.dispatch_once() {
                 log::error!("Could not dispatch wheel expirations: {}", ErrorFmt(e));
                 self.kill();
                 return;
@@ -227,15 +220,9 @@ impl WheelData {
         }
     }
 
-    fn dispatch_once(&self) -> Result<(), WheelError> {
+    async fn dispatch_once(&self, n: &mut TypedBuf<u64>) -> Result<(), WheelError> {
-        let mut n = 0u64;
+        if let Err(e) = self.ring.read(&self.fd, n.buf()).await {
-        loop {
+            return Err(WheelError::Read(e));
-            if let Err(e) = uapi::read(self.fd.raw(), &mut n) {
-                if e.0 == c::EAGAIN {
-                    break;
-                }
-                return Err(WheelError::Read(e.into()));
-            }
         }
         let now = Time::now()?;
         let dist = now - self.start;

src/wl_usr.rs

@@ -5,7 +5,7 @@ use {
     crate::{
         async_engine::{AsyncEngine, SpawnedFuture},
         client::{EventFormatter, RequestParser, MIN_SERVER_ID},
-        io_uring::IoUring,
+        io_uring::{IoUring, IoUringError},
         object::{ObjectId, WL_DISPLAY_ID},
         utils::{
             asyncevent::AsyncEvent,
@@ -47,7 +47,7 @@ pub enum UsrConError {
     #[error("The socket path is too long")]
     SocketPathTooLong,
     #[error("Could not connect to the compositor")]
-    Connect(#[source] OsError),
+    Connect(#[source] IoUringError),
     #[error("The message length is smaller than 8 bytes")]
     MsgLenTooSmall,
     #[error("The size of the message is not a multiple of 4")]
@@ -84,18 +84,14 @@ pub trait UsrConOwner {
 }
 
 impl UsrCon {
-    pub fn new(
+    pub async fn new(
         ring: &Rc<IoUring>,
         wheel: &Rc<Wheel>,
         eng: &Rc<AsyncEngine>,
         path: &str,
         server_id: u32,
     ) -> Result<Rc<Self>, UsrConError> {
-        let socket = match uapi::socket(
+        let socket = match uapi::socket(c::AF_UNIX, c::SOCK_STREAM | c::SOCK_CLOEXEC, 0) {
-            c::AF_UNIX,
-            c::SOCK_STREAM | c::SOCK_CLOEXEC | c::SOCK_NONBLOCK,
-            0,
-        ) {
             Ok(s) => Rc::new(s),
             Err(e) => return Err(UsrConError::CreateSocket(e.into())),
         };
@@ -107,8 +103,8 @@ impl UsrCon {
         let sun_path = uapi::as_bytes_mut(&mut addr.sun_path[..]);
         sun_path[..path.len()].copy_from_slice(path.as_bytes());
         sun_path[path.len()] = 0;
-        if let Err(e) = uapi::connect(socket.raw(), &addr) {
+        if let Err(e) = ring.connect(&socket, &addr).await {
-            return Err(UsrConError::Connect(e.into()));
+            return Err(UsrConError::Connect(e));
         }
         let mut obj_ids = Bitfield::default();
         obj_ids.take(0);

src/xcon.rs

@@ -7,14 +7,17 @@ use {
     crate::{
         async_engine::{Phase, SpawnedFuture},
         compositor::DISPLAY,
+        io_uring::IoUringError,
         state::State,
         utils::{
+            buf::DynamicBuf,
             bufio::{BufIo, BufIoError, BufIoMessage},
             clonecell::CloneCell,
             errorfmt::ErrorFmt,
             numcell::NumCell,
             oserror::OsError,
             queue::AsyncQueue,
+            stack::Stack,
             vec_ext::VecExt,
         },
         wire_xcon::{
@@ -77,7 +80,7 @@ pub enum XconError {
     #[error("Could not create a unix socket")]
     CreateSocket(#[source] OsError),
     #[error("Could not connect to Xserver")]
-    ConnectSocket(#[source] OsError),
+    ConnectSocket(#[source] IoUringError),
     #[error("Could not retrive the hostname")]
     Hostname(#[source] OsError),
     #[error("Server did not send enough fds")]
@@ -170,6 +173,7 @@ impl Drop for Xcon {
 
 struct XconData {
     bufio: Rc<BufIo>,
+    in_bufs: Stack<Vec<u8>>,
     next_serial: NumCell<u64>,
     last_recv_serial: Cell<u64>,
     reply_handlers: RefCell<VecDeque<Box<dyn ReplyHandler>>>,
@@ -181,7 +185,7 @@ struct XconData {
 }
 
 pub struct Reply<T: Message<'static>> {
-    bufio: Rc<BufIo>,
+    socket: Rc<XconData>,
     buf: Vec<u8>,
     t: T::Generic<'static>,
 }
@@ -196,7 +200,7 @@ where
 }
 
 pub struct Event {
-    bufio: Rc<BufIo>,
+    socket: Rc<XconData>,
     ext: Option<Extension>,
     code: u16,
     buf: Vec<u8>,
@@ -236,7 +240,7 @@ impl Event {
 
 impl Drop for Event {
     fn drop(&mut self) {
-        self.bufio.add_buf(mem::take(&mut self.buf));
+        self.socket.in_bufs.push(mem::take(&mut self.buf));
     }
 }
 
@@ -249,7 +253,7 @@ impl<T: Message<'static>> Reply<T> {
 impl<T: Message<'static>> Drop for Reply<T> {
     fn drop(&mut self) {
         if self.buf.capacity() > 0 {
-            self.bufio.add_buf(mem::take(&mut self.buf));
+            self.socket.in_bufs.push(mem::take(&mut self.buf));
         }
     }
 }
@@ -259,11 +263,11 @@ unsafe trait ReplyHandler {
     fn serial(&self) -> u64;
     fn handle_result(
         self: Box<Self>,
-        bufio: &Rc<BufIo>,
+        socket: &Rc<XconData>,
         parser: &mut Parser<'static>,
         buf: Vec<u8>,
     ) -> Result<(), XconError>;
-    fn handle_noreply(self: Box<Self>, bufio: &Rc<BufIo>) -> Result<(), XconError>;
+    fn handle_noreply(self: Box<Self>, bufio: &Rc<XconData>) -> Result<(), XconError>;
     fn handle_error(self: Box<Self>, error: XconError);
 }
 
@@ -299,7 +303,7 @@ unsafe impl<T: Message<'static>> ReplyHandler for AsyncReplyHandler<T> {
 
     fn handle_result(
         self: Box<Self>,
-        bufio: &Rc<BufIo>,
+        socket: &Rc<XconData>,
         parser: &mut Parser<'static>,
         buf: Vec<u8>,
     ) -> Result<(), XconError> {
@@ -314,7 +318,7 @@ unsafe impl<T: Message<'static>> ReplyHandler for AsyncReplyHandler<T> {
         };
         log::trace!("result {:?}", msg);
         let reply = Reply {
-            bufio: bufio.clone(),
+            socket: socket.clone(),
             buf,
             t: msg,
         };
@@ -322,10 +326,10 @@ unsafe impl<T: Message<'static>> ReplyHandler for AsyncReplyHandler<T> {
         Ok(())
     }
 
-    fn handle_noreply(self: Box<Self>, bufio: &Rc<BufIo>) -> Result<(), XconError> {
+    fn handle_noreply(self: Box<Self>, socket: &Rc<XconData>) -> Result<(), XconError> {
         if TypeId::of::<T::Generic<'static>>() == TypeId::of::<()>() {
             let reply = Reply {
-                bufio: bufio.clone(),
+                socket: socket.clone(),
                 buf: vec![],
                 t: unsafe { ptr::read(&() as *const () as *const T::Generic<'static>) },
             };
@@ -404,16 +408,12 @@ impl Xcon {
             let mut path = uapi::as_bytes_mut(&mut addr.sun_path[..]);
             let _ = write!(path, "/tmp/.X11-unix/X{}", display);
         }
-        let fd = match uapi::socket(
+        let fd = match uapi::socket(c::AF_UNIX, c::SOCK_STREAM | c::SOCK_CLOEXEC, 0) {
-            c::AF_UNIX,
-            c::SOCK_STREAM | c::SOCK_CLOEXEC | c::SOCK_NONBLOCK,
-            0,
-        ) {
             Ok(fd) => Rc::new(fd),
             Err(e) => return Err(XconError::CreateSocket(e.into())),
         };
-        if let Err(e) = uapi::connect(fd.raw(), &addr) {
+        if let Err(e) = state.ring.connect(&fd, &addr).await {
-            return Err(XconError::ConnectSocket(e.into()));
+            return Err(XconError::ConnectSocket(e));
         }
         let mut hnbuf = [MaybeUninit::<u8>::uninit(); 256];
         let hn = match uapi::gethostname(&mut hnbuf[..]) {
@@ -443,6 +443,7 @@ impl Xcon {
     ) -> Result<Rc<Self>, XconError> {
         let data = Rc::new(XconData {
             bufio: Rc::new(BufIo::new(fd, &state.ring)),
+            in_bufs: Default::default(),
             next_serial: NumCell::new(1),
             last_recv_serial: Cell::new(0),
             reply_handlers: Default::default(),
@@ -475,9 +476,13 @@ impl Xcon {
             formatter.write_packed(auth_value.as_bytes());
             formatter.align(4);
         }
-        data.bufio.send(BufIoMessage { fds, buf });
+        data.bufio.send(BufIoMessage {
+            fds,
+            buf: buf.unwrap(),
+        });
         let mut incoming = data.bufio.incoming();
-        let mut buf = data.bufio.buf();
+        let mut buf = data.in_bufs.pop().unwrap_or_default();
+        buf.clear();
         incoming.fill_msg_buf(8, &mut buf).await?;
         let len = u16::from_ne_bytes([buf[6], buf[7]]) as usize * 4;
         incoming.fill_msg_buf(len, &mut buf).await?;
@@ -506,7 +511,7 @@ impl Xcon {
             xid_inc: 1 << setup.resource_id_mask.trailing_zeros(),
             xid_max: setup.resource_id_mask | setup.resource_id_base,
             setup: Reply {
-                bufio: data.bufio.clone(),
+                socket: data.clone(),
                 t: unsafe { mem::transmute(setup) },
                 buf,
             },
@@ -823,9 +828,12 @@ impl XconData {
     fn send<T: Message<'static>>(
         self: &Rc<Self>,
         fds: Vec<Rc<OwnedFd>>,
-        buf: Vec<u8>,
+        buf: DynamicBuf,
     ) -> (AsyncReply<T>, u64) {
-        self.bufio.send(BufIoMessage { fds, buf });
+        self.bufio.send(BufIoMessage {
+            fds,
+            buf: buf.unwrap(),
+        });
         let slot = Rc::new(AsyncReplySlot {
             data: Cell::new(None),
             waker: Cell::new(None),
@@ -852,7 +860,10 @@ impl XconData {
         let mut formatter = Formatter::new(&mut fds, &mut buf, 0);
         GetInputFocus {}.serialize(&mut formatter);
         formatter.write_request_length();
-        self.bufio.send(BufIoMessage { fds, buf });
+        self.bufio.send(BufIoMessage {
+            fds,
+            buf: buf.unwrap(),
+        });
         self.next_serial.fetch_add(1);
     }
 

src/xcon/formatter.rs

@@ -1,17 +1,17 @@
 use {
-    crate::xcon::Message,
+    crate::{utils::buf::DynamicBuf, xcon::Message},
     std::rc::Rc,
     uapi::{AssertPacked, OwnedFd, Packed},
 };
 
 pub struct Formatter<'a> {
     fds: &'a mut Vec<Rc<OwnedFd>>,
-    buf: &'a mut Vec<u8>,
+    buf: &'a mut DynamicBuf,
     ext_opcode: u8,
 }
 
 impl<'a> Formatter<'a> {
-    pub fn new(fds: &'a mut Vec<Rc<OwnedFd>>, buf: &'a mut Vec<u8>, ext_opcode: u8) -> Self {
+    pub fn new(fds: &'a mut Vec<Rc<OwnedFd>>, buf: &'a mut DynamicBuf, ext_opcode: u8) -> Self {
         Self {
             fds,
             buf,

src/xcon/incoming.rs

@@ -42,7 +42,8 @@ impl Incoming {
         const MAX_LENGTH_UNITS: usize = 0x4000 / 4;
         const MIN_MSG_SIZE: usize = 32;
 
-        let mut msg_buf = self.socket.bufio.buf();
+        let mut msg_buf = self.socket.in_bufs.pop().unwrap_or_default();
+        msg_buf.clear();
         self.incoming
             .fill_msg_buf(MIN_MSG_SIZE, &mut msg_buf)
             .await?;
@@ -60,7 +61,7 @@ impl Incoming {
                 if first.serial() < serial {
                     let handler = reply_handlers.pop_front().unwrap();
                     drop(reply_handlers);
-                    handler.handle_noreply(&self.socket.bufio)?;
+                    handler.handle_noreply(&self.socket)?;
                     reply_handlers = self.socket.reply_handlers.borrow_mut();
                 } else {
                     break;
@@ -140,7 +141,7 @@ impl Incoming {
                             Parser::new(msg_buf, fds)
                         };
                         handler.handle_result(
-                            &self.socket.bufio,
+                            &self.socket,
                             &mut parser,
                             mem::take(&mut msg_buf),
                         )?;
@@ -203,7 +204,7 @@ impl Incoming {
                     break 'handle_event;
                 };
                 self.socket.events.push(Event {
-                    bufio: self.socket.bufio.clone(),
+                    socket: self.socket.clone(),
                     ext,
                     code,
                     buf: mem::take(&mut msg_buf),
@@ -212,7 +213,7 @@ impl Incoming {
             }
         }
         if msg_buf.capacity() > 0 {
-            self.socket.bufio.add_buf(msg_buf);
+            self.socket.in_bufs.push(msg_buf);
         }
         Ok(())
     }

src/xwayland.rs

@@ -18,7 +18,7 @@ use {
         io_uring::IoUringError,
         state::State,
         user_session::import_environment,
-        utils::{errorfmt::ErrorFmt, oserror::OsError, tri::Try},
+        utils::{buf::Buf, errorfmt::ErrorFmt, oserror::OsError},
         wire::WlSurfaceId,
         xcon::XconError,
         xwayland::{
@@ -29,7 +29,7 @@ use {
     bstr::ByteSlice,
     std::{num::ParseIntError, rc::Rc},
     thiserror::Error,
-    uapi::{c, pipe2, Errno, OwnedFd},
+    uapi::{c, pipe2, OwnedFd},
 };
 
 #[derive(Debug, Error)]
@@ -115,7 +115,7 @@ pub async fn manage(state: Rc<State>) {
         log::info!("Allocated display :{} for Xwayland", xsocket.id);
         log::info!("Waiting for connection attempt");
         if state.backend.get().import_environment() {
-            import_environment(&state, DISPLAY, &display);
+            import_environment(&state, DISPLAY, &display).await;
         }
         if let Err(e) = state.ring.readable(&socket).await {
             log::error!("{}", ErrorFmt(e));
@@ -144,20 +144,12 @@ async fn run(
         Ok(p) => p,
         Err(e) => return Err(XWaylandError::Pipe(e.into())),
     };
-    let wm = uapi::socketpair(
+    let wm = uapi::socketpair(c::AF_UNIX, c::SOCK_STREAM | c::SOCK_CLOEXEC, 0);
-        c::AF_UNIX,
-        c::SOCK_STREAM | c::SOCK_CLOEXEC | c::SOCK_NONBLOCK,
-        0,
-    );
     let (wm1, wm2) = match wm {
         Ok(w) => w,
         Err(e) => return Err(XWaylandError::Socketpair(e.into())),
     };
-    let client = uapi::socketpair(
+    let client = uapi::socketpair(c::AF_UNIX, c::SOCK_STREAM | c::SOCK_CLOEXEC, 0);
-        c::AF_UNIX,
-        c::SOCK_STREAM | c::SOCK_CLOEXEC | c::SOCK_NONBLOCK,
-        0,
-    );
     let (client1, client2) = match client {
         Ok(w) => w,
         Err(e) => return Err(XWaylandError::Socketpair(e.into())),
@@ -177,9 +169,15 @@ async fn run(
         Err(e) => return Err(XWaylandError::ExecFailed(e)),
     };
     let client_id = state.clients.id();
-    let client = state
+    let client = state.clients.spawn2(
-        .clients
+        client_id,
-        .spawn2(client_id, state, client1, uapi::getuid(), pid, true, true);
+        state,
+        Rc::new(client1),
+        uapi::getuid(),
+        pid,
+        true,
+        true,
+    );
     let client = match client {
         Ok(c) => c,
         Err(e) => return Err(XWaylandError::SpawnClient(e)),
@@ -193,7 +191,7 @@ async fn run(
             Err(e) => return Err(XWaylandError::CreateWm(Box::new(e))),
         };
         let _wm = state.eng.spawn(wm.run());
-        state.ring.readable(&Rc::new(pidfd)).await?;
+        state.ring.readable(&pidfd).await?;
     }
     state.xwayland.queue.clear();
     stderr_read.await;
@@ -219,45 +217,21 @@ pub fn build_args(fds: &[OwnedFd]) -> (String, Vec<String>) {
 
 async fn log_xwayland(state: Rc<State>, stderr: OwnedFd) {
     let stderr = Rc::new(stderr);
-    let res = Errno::tri(|| {
-        uapi::fcntl_setfl(
-            stderr.raw(),
-            uapi::fcntl_getfl(stderr.raw())? | c::O_NONBLOCK,
-        )?;
-        Ok(())
-    });
-    if let Err(e) = res {
-        log::error!("Could not set stderr fd to nonblock: {}", ErrorFmt(e));
-        return;
-    }
     let mut buf = vec![];
-    let mut buf2 = [0; 128];
+    let mut buf2 = Buf::new(128);
     let mut done = false;
     while !done {
-        if let Err(e) = state.ring.readable(&stderr).await {
-            log::error!(
-                "Cannot wait for the xwayland stderr to become readable: {}",
-                ErrorFmt(e)
-            );
-            return;
-        }
         loop {
-            match uapi::read(stderr.raw(), &mut buf2[..]) {
+            match state.ring.read(&stderr, buf2.clone()).await {
-                Ok(buf2) if buf2.len() > 0 => {
+                Ok(n) if n > 0 => {
-                    buf.extend_from_slice(buf2);
+                    buf.extend_from_slice(&buf2[..n]);
                 }
                 Ok(_) => {
                     done = true;
                     break;
                 }
-                Err(Errno(c::EAGAIN)) => {
-                    break;
-                }
                 Err(e) => {
-                    log::error!(
+                    log::error!("Could not read from stderr fd: {}", ErrorFmt(e));
-                        "Could not read from stderr fd: {}",
-                        ErrorFmt(crate::utils::oserror::OsError::from(e))
-                    );
                     return;
                 }
             }

src/xwayland/xwm.rs

@@ -20,7 +20,7 @@ use {
                 WlSurface,
             },
         },
-        io_uring::{IoUring, IoUringError, TaskResultExt},
+        io_uring::{IoUring, IoUringError},
         rect::Rect,
         state::State,
         time::Time,
@@ -28,7 +28,7 @@ use {
         utils::{
             bitflags::BitflagsExt, buf::Buf, clonecell::CloneCell, copyhashmap::CopyHashMap,
             errorfmt::ErrorFmt, linkedlist::LinkedList, numcell::NumCell, oserror::OsError,
-            rc_eq::rc_eq, tri::Try,
+            rc_eq::rc_eq,
         },
         wire::{WlDataDeviceId, WlSurfaceId, ZwpPrimarySelectionDeviceV1Id},
         wire_xcon::{
@@ -67,11 +67,11 @@ use {
     std::{
         borrow::Cow,
         cell::{Cell, RefCell},
-        mem::{self, MaybeUninit},
+        mem::{self},
         ops::{Deref, DerefMut},
         rc::Rc,
     },
-    uapi::{c, Errno, OwnedFd},
+    uapi::{c, OwnedFd},
 };
 
 atoms! {
@@ -1607,15 +1607,6 @@ impl Wm {
                             break 'convert;
                         }
                     };
-                    let res = OsError::tri(|| {
-                        let fl = uapi::fcntl_getfl(rx.raw())?;
-                        uapi::fcntl_setfl(rx.raw(), fl | c::O_NONBLOCK)?;
-                        Ok(())
-                    });
-                    if let Err(e) = res {
-                        log::error!("Could not make pipe nonblocking: {}", e);
-                        break 'convert;
-                    }
                     success = None;
                     receive_data_offer::<T>(&offer.offer, &mt, Rc::new(tx));
                     let id = self.transfer_ids.fetch_add(1);
@@ -1726,7 +1717,7 @@ impl Wm {
                 let id = self.transfer_ids.fetch_add(1);
                 let transfer = XToWaylandTransfer {
                     id,
-                    data: data.slice(..),
+                    data: data.clone(),
                     fd: transfer.fd,
                     state: self.state.clone(),
                     shared: self.shared.clone(),
@@ -2467,7 +2458,7 @@ impl XToWaylandTransfer {
                 .state
                 .ring
                 .write(&self.fd, self.data.slice(pos..), Some(timeout));
-            match res.await.merge() {
+            match res.await {
                 Ok(n) => pos += n,
                 Err(IoUringError::OsError(OsError(c::ECANCELED))) => {
                     log::error!("Transfer timed out");
@@ -2499,10 +2490,10 @@ struct WaylandToXTransfer {
 impl WaylandToXTransfer {
     async fn run(self) {
         let mut success = false;
-        let mut buf = Box::new([MaybeUninit::<u8>::uninit(); 1024]);
+        let mut buf = Buf::new(1024);
         loop {
-            match uapi::read(self.fd.raw(), &mut buf[..]) {
+            match self.ring.read(&self.fd, buf.clone()).await {
-                Ok(n) if n.is_empty() => {
+                Ok(0) => {
                     success = true;
                     break;
                 }
@@ -2513,19 +2504,13 @@ impl WaylandToXTransfer {
                         property: self.property,
                         ty: self.ty,
                         format: 8,
-                        data: n,
+                        data: &buf[..n],
                     };
                     if let Err(e) = self.c.call(&cp).await {
                         log::error!("Could not append data to property: {}", ErrorFmt(e));
                         break;
                     }
                 }
-                Err(Errno(c::EAGAIN)) => {
-                    if let Err(e) = self.ring.readable(&self.fd).await {
-                        log::error!("Could not wait for fd to become readable: {}", ErrorFmt(e));
-                        break;
-                    }
-                }
                 Err(e) => {
                     log::error!("Could not read from wayland client: {}", ErrorFmt(e));
                     break;