A vsock reader wired through tokio `AsyncFd` is never woken after the first readable event

[archevel]

On Hermit after PR #2434 is merged (which makes accep return a real connection socket), a non-blocking AF_VSOCK socket driven through tokio::io::unix::AsyncFd stops being woken after the first readable event. The first read works; every subsequent readable().await on the same fd hangs forever, even though the peer has sent more data.

The guest listens on a vsock port, accepts one connection, and reads in "waves" (await readiness, drain to WouldBlock) and responds with "received" back. The host sends test, waits, then sends test again. Observed guest output:

REPRO: listening on vsock port 9999
REPRO: accepted connection
REPRO: before awaiting readable. will hang on next line after second call
REPRO: after awaiting readable, we only reach this first call
REPRO: wave 1 = "test\n"
REPRO: awaiting wave 2...
REPRO: before awaiting readable. will hang on next line after second call

Wave 1 is read and responded too. Wave 2 never wakes the reader.

A control using the blocking vsock path (no tokio) — the existing hermit-rs vsock example — works fine on the same VM, so this is specific to the AsyncFd/SourceFd readiness path, not the vsock transport or the kernel's accept/read implementation.

Root cause

mio's poll(2) selector is edge-triggered by construction. After an fd's event fires, it clears that fd's interest from the pollfd mio-1.2.1/src/sys/unix/selector/poll.rs:

// Remove the interest which just got triggered the IoSourceState's do_io
// wrapper used with this selector will add back the interest using
// reregister.
poll_fd.events &= !poll_fd.revents;

As the comment says, it relies on IoSourceState::do_io to re-arm the interest.
But do_io only reregisters when the I/O closure returns WouldBlock
mio-1.2.1/src/sys/unix/selector/poll.rs:

if err.kind() == io::ErrorKind::WouldBlock {
    self.inner.as_ref().map_or(Ok(()), |state| {
        state
        .selector
        .reregister(state.fd, state.token, state.interests)
    })?;
}

tokio's AsyncFd registers the fd via mio::unix::SourceFd (a raw fd) and never routes I/O through mio's IoSource, so do_io is never called. The interest is cleared once and never restored. The next readable() therefore waits on an interest mask that no longer contains the read bit, and never fires.

NOTE: This does not affect TcpStream/UdpSocket: tokio wraps them in mio's native mio::net::TcpStream/UdpSocket, which hold the fd as an IoSource and route all I/O through IoSource::do_io — the method that reregisters interest on WouldBlock (poll.rs:733). Only raw fds registered via SourceFd (as AsyncFd does, and as this vsock example does) miss the rearm and get stuck

Reproducer

A self-contained main.rs example below. It needs the code in the second block raw_vsock.rs to compile and run.

// main.rs
use tokio::io::unix::AsyncFd;

async fn read_wave(afd: &AsyncFd<RawVsock>) -> std::io::Result<Vec<u8>> {
    loop {
        println!("REPRO: before awaiting readable. will hang on next line after second call");
        let mut guard = afd.readable().await?;
        println!("REPRO: after awaiting readable, we only reach this first call");
        let mut out = Vec::new();
        let mut buf = [0u8; 256];
        loop {
            match guard.try_io(|inner| {
                use std::io::Read as _;
                (&*inner.get_ref()).read(&mut buf)
            }) {
                Ok(Ok(0)) => return Ok(out), // EOF
                Ok(Ok(n)) => out.extend_from_slice(&buf[..n]),
                Ok(Err(e)) => return Err(e),
                Err(_would_block) => {
                    if !out.is_empty() {
                        return Ok(out);
                    }
                    break; // spurious wake with no data: re-await
                }
            }
        }
    }
}

async fn run() -> std::io::Result<()> {
    let listener = listen_vsock(PORT)?;
    println!("REPRO: listening on vsock port {PORT}");

    let conn = accept_one(&listener).await?;
    println!("REPRO: accepted connection");

    // Wave 1 works: the fresh `AsyncFd` registration arms the interest.
    let first = read_wave(&conn).await?;
    println!("REPRO: wave 1 = {:?}", String::from_utf8_lossy(&first));
    write_all(&conn, b"received").await?;

    // Wave 2 hangs: mio cleared the interest after wave 1 and `AsyncFd`
    // (SourceFd) is never re-armed, so this `readable()` never wakes.
    println!("REPRO: awaiting wave 2...");
    let second = read_wave(&conn).await?;
    println!("REPRO: wave 2 = {:?}", String::from_utf8_lossy(&second));
    write_all(&conn, b"received again").await?;

    println!("REPRO: responded to wave 2 -- OK (bug NOT present)");
    Ok(())
}

And for completeness the supporting code:

// raw_vsock.rs

use std::io::{self, Read, Write};
use std::mem::size_of;
use std::os::fd::{AsRawFd, FromRawFd, OwnedFd, RawFd};

#[cfg(target_os = "hermit")]
use hermit_abi::{
    accept, bind, fcntl, listen, read, sockaddr, sockaddr_vm, socket, socklen_t, write, AF_VSOCK,
    F_SETFL, O_NONBLOCK, SOCK_STREAM, VMADDR_CID_ANY,
};

use tokio::io::unix::AsyncFd;

fn last_err() -> io::Error {
    io::Error::last_os_error()
}

fn make_sockaddr_vm(cid: u32, port: u32) -> sockaddr_vm {
    sockaddr_vm {
        #[cfg(target_os = "hermit")]
        svm_len: size_of::<sockaddr_vm>() as u8,
        svm_reserved1: 0,
        svm_family: AF_VSOCK as _,
        svm_cid: cid,
        svm_port: port,
        svm_zero: [0; 4],
    }
}

fn set_nonblocking(fd: RawFd) -> io::Result<()> {
    if unsafe { fcntl(fd, F_SETFL, O_NONBLOCK) } < 0 {
        return Err(last_err());
    }
    Ok(())
}

pub struct RawVsock {
    fd: OwnedFd,
}

impl RawVsock {
    fn from_raw(fd: RawFd) -> io::Result<Self> {
        set_nonblocking(fd)?;
        Ok(Self {
            fd: unsafe { OwnedFd::from_raw_fd(fd) },
        })
    }
}

impl AsRawFd for RawVsock {
    fn as_raw_fd(&self) -> RawFd {
        self.fd.as_raw_fd()
    }
}

impl Read for &RawVsock {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        let r = unsafe { read(self.fd.as_raw_fd(), buf.as_mut_ptr(), buf.len()) };
        if r < 0 {
            Err(last_err())
        } else {
            Ok(r as usize)
        }
    }
}

impl Write for &RawVsock {
    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
        let r = unsafe { write(self.fd.as_raw_fd(), buf.as_ptr(), buf.len()) };
        if r < 0 {
            Err(last_err())
        } else {
            Ok(r as usize)
        }
    }
    fn flush(&mut self) -> io::Result<()> {
        Ok(())
    }
}

/// Bind + listen on `VMADDR_CID_ANY:port` and return the listener as an
/// `AsyncFd`.
pub fn listen_vsock(port: u32) -> io::Result<AsyncFd<RawVsock>> {
    let lfd = unsafe { socket(AF_VSOCK, SOCK_STREAM, 0) };
    if lfd < 0 {
        return Err(last_err());
    }
    let addr = make_sockaddr_vm(VMADDR_CID_ANY, port);
    if unsafe {
        bind(
            lfd,
            &addr as *const _ as *const sockaddr,
            size_of::<sockaddr_vm>() as socklen_t,
        )
    } < 0
    {
        return Err(last_err());
    }
    if unsafe { listen(lfd, 1) } < 0 {
        return Err(last_err());
    }
    AsyncFd::new(RawVsock::from_raw(lfd)?)
}

pub async fn accept_one(listener: &AsyncFd<RawVsock>) -> io::Result<AsyncFd<RawVsock>> {
    loop {
        let mut guard = listener.readable().await?;
        let mut peer = make_sockaddr_vm(0, 0);
        let mut len = size_of::<sockaddr_vm>() as socklen_t;
        match guard.try_io(|inner| {
            let fd = unsafe {
                accept(
                    inner.get_ref().as_raw_fd(),
                    &mut peer as *mut _ as *mut sockaddr,
                    &mut len as *mut socklen_t,
                )
            };
            if fd < 0 {
                Err(last_err())
            } else {
                Ok(fd)
            }
        }) {
            Ok(Ok(fd)) => return AsyncFd::new(RawVsock::from_raw(fd)?),
            Ok(Err(e)) => return Err(e),
            Err(_would_block) => continue,
        }
    }
}

/// Write all of `buf`, awaiting writability between partial writes.
pub async fn write_all(afd: &AsyncFd<RawVsock>, mut buf: &[u8]) -> io::Result<()> {
    while !buf.is_empty() {
        let mut guard = afd.writable().await?;
        match guard.try_io(|inner| (&*inner.get_ref()).write(buf)) {
            Ok(Ok(n)) => buf = &buf[n..],
            Ok(Err(e)) => return Err(e),
            Err(_would_block) => continue,
        }
    }
    Ok(())
}

Build the above with hermit and run it in qemu:

sudo qemu-system-x86_64 -enable-kvm -cpu host -smp 1 -m 512M -machine q35 \
  -display none -serial stdio \
  -kernel kernel/hermit-loader-x86_64 \
  -initrd target/x86_64-unknown-hermit/debug/vsock-async-fd-issue \
  -device vhost-vsock-pci,disable-legacy=on,guest-cid=3

Use e.g. socat - VSOCK-CONNECT:3:9999 to connect and send a message that get's echoed back. Then try sending a second message. The AsyncFd will then cause a hang.

Workaround

Consumers can force a re-arm by deregistering and reregistering the fd after each drained read — in practice, dropping and recreating the AsyncFd over the same raw fd (into_inner() deregisters without closing the fd; AsyncFd::new reregisters with fresh interest). clear_ready() alone is not enough: it resets tokio's internal readiness but not mio's pollfd interest mask.

Possible directions

• Make the poll(2) selector re-arm interest for fds registered via SourceFd (not just those going through IoSource::do_io) — i.e. fix it in mio.
• Or handle the SourceFd + poll(2)-selector combination in the Hermit integration so AsyncFd-based code works without a per-read re-arm.
• Or switch to epoll for hermit.