devices/net: Rewrite using new RxQueueProducer/TxQueueConsumer

Rewrite the all of the backend (unixstream, unixgram, tap) in terms of the new
RxQueueProducer/TxQueueConsumer abstractions.

Signed-off-by: Matej Hrica <mhrica@redhat.com>

Author: mtjhrc

src/devices/src/virtio/net/backend.rs

@@ -18,29 +18,39 @@ pub enum ConnectError {
 #[allow(dead_code)]
 #[derive(Debug)]
 pub enum ReadError {
-    /// Nothing was written
-    NothingRead,
-    /// Another internal error occurred
+    /// Backend process not running (EPIPE)
+    ProcessNotRunning,
+    /// Internal I/O error
     Internal(nix::Error),
 }
 
 #[allow(dead_code)]
 #[derive(Debug)]
 pub enum WriteError {
-    /// Nothing was written, you can drop the frame or try to resend it later
-    NothingWritten,
-    /// Part of the buffer was written, the write has to be finished using try_finish_write
-    PartialWrite,
-    /// Passt doesnt seem to be running (received EPIPE)
+    /// Backend process not running (EPIPE)
     ProcessNotRunning,
-    /// Another internal error occurred
+    /// Internal I/O error
     Internal(nix::Error),
 }
 
+/// Network backend trait.
+///
+/// Backends own both the socket and the queue consumers. The send/recv methods
+/// operate on internal queues. EAGAIN is not an error - it just means nothing
+/// happened this call.
 pub trait NetBackend {
-    fn read_frame(&mut self, buf: &mut [u8]) -> Result<usize, ReadError>;
-    fn write_frame(&mut self, hdr_len: usize, buf: &mut [u8]) -> Result<(), WriteError>;
-    fn has_unfinished_write(&self) -> bool;
-    fn try_finish_write(&mut self, hdr_len: usize, buf: &[u8]) -> Result<(), WriteError>;
+    /// Send pending frames from the TX queue to the network.
+    ///
+    /// Pulls frames from internal TxQueueConsumer and sends using batched I/O.
+    /// EAGAIN returns Ok(()) - pending frames kept for retry.
+    fn send(&mut self) -> Result<(), WriteError>;
+
+    /// Receive frames from the network into the RX queue.
+    ///
+    /// Reads from socket into internal RxQueueProvider.
+    /// EAGAIN returns Ok(()).
+    fn recv(&mut self) -> Result<(), ReadError>;
+
+    /// Returns the raw socket fd for epoll registration.
     fn raw_socket_fd(&self) -> RawFd;
 }

src/devices/src/virtio/net/device.rs

@@ -22,23 +22,17 @@ use std::path::PathBuf;
 use utils::eventfd::{EventFd, EFD_NONBLOCK};
 use virtio_bindings::virtio_net::VIRTIO_NET_F_MAC;
 use virtio_bindings::virtio_ring::VIRTIO_RING_F_EVENT_IDX;
-use vm_memory::{ByteValued, GuestMemoryError, GuestMemoryMmap};
+use vm_memory::{ByteValued, GuestMemoryMmap};
 
 const VIRTIO_F_VERSION_1: u32 = 32;
 
-#[derive(Debug)]
-pub enum FrontendError {
-    DescriptorChainTooSmall,
-    EmptyQueue,
-    GuestMemory(GuestMemoryError),
-    QueueError(QueueError),
-    ReadOnlyDescriptor,
-}
+// FrontendError removed - no longer used with vectored I/O
 
 #[derive(Debug)]
 pub enum RxError {
     Backend(ReadError),
     DeviceError(DeviceError),
+    QueueError(QueueError),
 }
 
 #[derive(Debug)]
@@ -96,7 +90,7 @@ impl Net {
     ) -> Result<Self> {
         let avail_features = features as u64
             | (1 << VIRTIO_NET_F_MAC)
-            | (1 << VIRTIO_RING_F_EVENT_IDX)
+            // | (1 << VIRTIO_RING_F_EVENT_IDX)  // TODO: re-enable after debugging
             | (1 << VIRTIO_F_VERSION_1);
 
         let mut queue_evts = Vec::new();

src/devices/src/virtio/net/mod.rs

@@ -20,6 +20,8 @@ pub const TX_INDEX: usize = 1;
 
 mod backend;
 pub mod device;
+#[cfg(target_os = "macos")]
+mod socket_x;
 #[cfg(target_os = "linux")]
 mod tap;
 mod unixgram;
@@ -30,13 +32,10 @@ fn vnet_hdr_len() -> usize {
     mem::size_of::<virtio_net_hdr_v1>()
 }
 
-// This initializes to all 0 the virtio_net_hdr part of a buf and return the length of the header
-// https://docs.oasis-open.org/virtio/virtio/v1.1/csprd01/virtio-v1.1-csprd01.html#x1-2050006
-fn write_virtio_net_hdr(buf: &mut [u8]) -> usize {
-    let len = vnet_hdr_len();
-    buf[0..len].fill(0);
-    len
-}
+/// Default zeroed virtio_net_hdr_v1 (12 bytes) - used as prefix when receiving from backends
+/// that don't include vnet headers (e.g., passt/unixstream)
+/// https://docs.oasis-open.org/virtio/virtio/v1.1/csprd01/virtio-v1.1-csprd01.html#x1-2050006
+static DEFAULT_VNET_HDR: [u8; 12] = [0u8; 12];
 
 pub use self::device::Net;
 #[derive(Debug)]

src/devices/src/virtio/net/socket_x.rs

@@ -0,0 +1,81 @@
+// macOS-specific batch message syscalls (sendmsg_x/recvmsg_x)
+//
+// These are private Apple APIs that allow sending/receiving multiple messages
+// in a single syscall, similar to Linux's sendmmsg/recvmmsg.
+//
+// Reference: https://github.com/nirs/vmnet-helper/blob/main/socket_x.h
+
+#![allow(dead_code)]
+#![allow(non_camel_case_types)]
+
+#[cfg(target_os = "macos")]
+pub mod macos {
+    use libc::{c_int, c_uint, c_void, iovec, socklen_t};
+
+    /// Extended message header for batch operations.
+    /// Similar to msghdr but includes msg_datalen for output.
+    #[repr(C)]
+    pub struct msghdr_x {
+        pub msg_name: *mut c_void,
+        pub msg_namelen: socklen_t,
+        pub msg_iov: *mut iovec,
+        pub msg_iovlen: c_int,
+        pub msg_control: *mut c_void,
+        pub msg_controllen: socklen_t,
+        pub msg_flags: c_int,
+        pub msg_datalen: usize, // out: bytes transferred for this message
+    }
+
+    impl Default for msghdr_x {
+        fn default() -> Self {
+            Self {
+                msg_name: std::ptr::null_mut(),
+                msg_namelen: 0,
+                msg_iov: std::ptr::null_mut(),
+                msg_iovlen: 0,
+                msg_control: std::ptr::null_mut(),
+                msg_controllen: 0,
+                msg_flags: 0,
+                msg_datalen: 0,
+            }
+        }
+    }
+
+    extern "C" {
+        /// Send multiple datagrams in a single syscall.
+        ///
+        /// # Arguments
+        /// * `s` - Socket file descriptor
+        /// * `msgp` - Pointer to array of msghdr_x structures
+        /// * `cnt` - Number of messages to send
+        /// * `flags` - Only MSG_DONTWAIT is supported
+        ///
+        /// # Constraints
+        /// For each msghdr_x: msg_name, msg_namelen, msg_control, msg_controllen,
+        /// msg_flags, and msg_datalen must all be zero on input.
+        ///
+        /// # Returns
+        /// Number of datagrams sent, or -1 on error.
+        /// Each msghdr_x.msg_datalen is set to bytes sent for that message.
+        pub fn sendmsg_x(s: c_int, msgp: *const msghdr_x, cnt: c_uint, flags: c_int) -> isize;
+
+        /// Receive multiple datagrams in a single syscall.
+        ///
+        /// # Arguments
+        /// * `s` - Socket file descriptor
+        /// * `msgp` - Pointer to array of msghdr_x structures
+        /// * `cnt` - Maximum number of messages to receive
+        /// * `flags` - Only MSG_DONTWAIT is supported
+        ///
+        /// # Constraints
+        /// For each msghdr_x: msg_flags must be zero on input.
+        ///
+        /// # Returns
+        /// Number of datagrams received (may be less than cnt), or -1 on error.
+        /// Each msghdr_x.msg_datalen is set to bytes received for that message.
+        pub fn recvmsg_x(s: c_int, msgp: *mut msghdr_x, cnt: c_uint, flags: c_int) -> isize;
+    }
+}
+
+#[cfg(target_os = "macos")]
+pub use macos::*;

src/devices/src/virtio/net/tap.rs

@@ -2,35 +2,46 @@ use libc::{
     c_char, c_int, ifreq, IFF_NO_PI, IFF_TAP, IFF_VNET_HDR, TUN_F_CSUM, TUN_F_TSO4, TUN_F_TSO6,
     TUN_F_UFO,
 };
-use nix::fcntl::{fcntl, open, FcntlArg, OFlag};
+use nix::fcntl::{open, OFlag};
 use nix::sys::stat::Mode;
-use nix::unistd::{read, write};
+use nix::sys::uio::{readv, writev};
 use nix::{ioctl_write_int, ioctl_write_ptr};
-use std::os::fd::{AsRawFd, OwnedFd, RawFd};
+use std::os::fd::{AsFd, AsRawFd, OwnedFd, RawFd};
 use std::{io, mem, ptr};
+use utils::fd::SetNonblockingExt;
 use virtio_bindings::virtio_net::{
     VIRTIO_NET_F_GUEST_CSUM, VIRTIO_NET_F_GUEST_TSO4, VIRTIO_NET_F_GUEST_TSO6,
     VIRTIO_NET_F_GUEST_UFO,
 };
+use vm_memory::GuestMemoryMmap;
 
 use super::backend::{ConnectError, NetBackend, ReadError, WriteError};
-use super::{write_virtio_net_hdr, FRAME_HEADER_LEN};
+use crate::virtio::queue::Queue;
+use crate::virtio::rx_queue_producer::RxQueueProducer;
+use crate::virtio::tx_queue_consumer::{Consumed, TxQueueConsumer};
+use crate::virtio::InterruptTransport;
 
 ioctl_write_ptr!(tunsetiff, b'T', 202, c_int);
 ioctl_write_int!(tunsetoffload, b'T', 208);
 ioctl_write_ptr!(tunsetvnethdrsz, b'T', 216, c_int);
 
+const MAX_BATCH: usize = 256;
+
 pub struct Tap {
     fd: OwnedFd,
-    include_vnet_header: bool,
+    tx_consumer: TxQueueConsumer,
+    rx_producer: RxQueueProducer,
 }
 
 impl Tap {
     /// Create an endpoint using the file descriptor of a tap device
     pub fn new(
         tap_name: String,
         vnet_features: u64,
-        include_vnet_header: bool,
+        tx_queue: Queue,
+        rx_queue: Queue,
+        mem: GuestMemoryMmap,
+        interrupt: InterruptTransport,
     ) -> Result<Self, ConnectError> {
         let fd = match open("/dev/net/tun", OFlag::O_RDWR, Mode::empty()) {
             Ok(fd) => fd,
@@ -47,10 +58,9 @@ impl Tap {
             );
         }
 
-        req.ifr_ifru.ifru_flags = IFF_TAP as i16 | IFF_NO_PI as i16;
-        if include_vnet_header {
-            req.ifr_ifru.ifru_flags |= IFF_VNET_HDR as i16;
-        }
+        req.ifr_ifru.ifru_flags = IFF_TAP as i16 | IFF_NO_PI as i16 | IFF_VNET_HDR as i16;
+
+        log::info!("Tap::new() fd={} tap={}", fd.as_raw_fd(), tap_name);
 
         let mut offload_flags: u64 = 0;
         if (vnet_features & (1 << VIRTIO_NET_F_GUEST_CSUM)) != 0 {
@@ -71,7 +81,7 @@ impl Tap {
                 return Err(ConnectError::TunSetIff(io::Error::from(err)));
             }
 
-            // TODO(slp): replace hardcoded vnet size with cons
+            // TODO(slp): replace hardcoded vnet size with const
             if let Err(err) = tunsetvnethdrsz(fd.as_raw_fd(), &12) {
                 return Err(ConnectError::TunSetVnetHdrSz(io::Error::from(err)));
             }
@@ -81,67 +91,71 @@ impl Tap {
             }
         }
 
-        match fcntl(&fd, FcntlArg::F_GETFL) {
-            Ok(flags) => {
-                if let Err(e) = fcntl(
-                    &fd,
-                    FcntlArg::F_SETFL(OFlag::from_bits_truncate(flags) | OFlag::O_NONBLOCK),
-                ) {
-                    warn!("error switching to non-blocking: id={fd:?}, err={e}");
-                }
-            }
-            Err(e) => error!("couldn't obtain fd flags id={fd:?}, err={e}"),
-        };
+        if let Err(e) = fd.set_nonblocking(true) {
+            log::warn!("Failed to set O_NONBLOCK on tap: {e}");
+        }
+
+        let tx_consumer = TxQueueConsumer::new(tx_queue, mem.clone(), interrupt.clone());
+        let rx_provider = RxQueueProducer::new(rx_queue, mem, interrupt);
 
         Ok(Self {
             fd,
-            include_vnet_header,
+            tx_consumer,
+            rx_producer: rx_provider,
         })
     }
 }
 
 impl NetBackend for Tap {
-    /// Try to read a frame from the tap devie. If no bytes are available reports
-    /// ReadError::NothingRead.
-    fn read_frame(&mut self, buf: &mut [u8]) -> Result<usize, ReadError> {
-        let buf_offset = if !self.include_vnet_header {
-            write_virtio_net_hdr(buf)
-        } else {
-            0
-        };
-
-        let frame_length = match read(&self.fd, &mut buf[buf_offset..]) {
-            Ok(f) => f,
-            #[allow(unreachable_patterns)]
-            Err(nix::Error::EAGAIN | nix::Error::EWOULDBLOCK) => {
-                return Err(ReadError::NothingRead)
-            }
-            Err(e) => {
-                return Err(ReadError::Internal(e));
+    fn send(&mut self) -> Result<(), WriteError> {
+        let fd = self.fd.as_fd();
+
+        self.tx_consumer.feed(MAX_BATCH);
+
+        // Each descriptor chain is one packet. TAP's writev combines iovecs into
+        // a single packet (scatter-gather), so we can use it directly without
+        // flattening. One writev call per packet.
+        let _ = self.tx_consumer.consume(|frames| {
+            let mut total = 0usize;
+            for frame in frames.iter() {
+                if frame.is_empty() {
+                    continue;
+                }
+                match writev(fd, frame) {
+                    Ok(n) => total += n,
+                    Err(nix::errno::Errno::EAGAIN) => break,
+                    Err(nix::errno::Errno::EPIPE) => return Err(WriteError::ProcessNotRunning),
+                    Err(e) => {
+                        log::error!("Tap TX failed: {:?}", e);
+                        return Err(WriteError::Internal(e));
+                    }
+                }
             }
-        };
-        debug!("Read eth frame from tap: {frame_length} bytes");
-        Ok(buf_offset + frame_length)
-    }
+            Ok(Consumed::Bytes(total))
+        })?;
 
-    /// Try to write a frame to the tap device.
-    fn write_frame(&mut self, hdr_len: usize, buf: &mut [u8]) -> Result<(), WriteError> {
-        let buf_offset = if !self.include_vnet_header {
-            hdr_len
-        } else {
-            FRAME_HEADER_LEN
-        };
-        let ret = write(&self.fd, buf[buf_offset..]).map_err(WriteError::Internal)?;
-        debug!("Written frame size={}, written={}", buf.len(), ret);
         Ok(())
     }
 
-    fn has_unfinished_write(&self) -> bool {
-        false
-    }
+    fn recv(&mut self) -> Result<(), ReadError> {
+        let fd = self.fd.as_fd();
+
+        self.rx_producer.feed(MAX_BATCH);
+
+        self.rx_producer.produce(|chains, completer| {
+            for (i, chain) in chains.iter_mut().enumerate() {
+                if chain.is_empty() {
+                    warn!("Chain {i} was empty");
+                    break;
+                }
+
+                match readv(fd, chain) {
+                    Ok(n) => completer.complete(chain, i, n),
+                    Err(_) => break, // EAGAIN or error, stop receiving
+                }
+            }
+        });
 
-    fn try_finish_write(&mut self, _hdr_len: usize, _buf: &[u8]) -> Result<(), WriteError> {
-        // The tap backend doesn't do partial writes.
         Ok(())
     }