devices/net: Rewrite virtio-net in terms of new batch_queue utilities

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

Author: mtjhrc

src/devices/Cargo.toml

@@ -8,7 +8,7 @@ edition = "2021"
 tee = []
 amd-sev = ["blk", "tee"]
 tdx = ["blk", "tee"]
-net = []
+net = ["batch_queue"]
 batch_queue = []
 blk = []
 efi = ["blk", "net"]
@@ -25,7 +25,7 @@ crossbeam-channel = ">=0.5.15"
 libc = ">=0.2.39"
 libloading = "0.8"
 log = "0.4.0"
-nix = { version = "0.30.1", features = ["ioctl", "net", "poll", "socket", "fs"] }
+nix = { version = "0.30.1", features = ["ioctl", "net", "poll", "socket", "fs", "uio"] }
 pw = { package = "pipewire", version = "0.8.0", optional = true }
 rand = "0.9.2"
 thiserror = { version = "2.0", optional = true }

src/devices/src/virtio/mod.rs

@@ -12,6 +12,8 @@ use std::io::Error as IOError;
 
 #[cfg(not(feature = "tee"))]
 pub mod balloon;
+#[cfg(feature = "batch_queue")]
+pub mod batch_queue;
 #[allow(dead_code)]
 #[allow(non_camel_case_types)]
 pub mod bindings;
@@ -36,6 +38,8 @@ mod queue;
 pub mod rng;
 #[cfg(feature = "snd")]
 pub mod snd;
+#[cfg(all(feature = "batch_queue", test))]
+pub(crate) mod test_utils;
 pub mod vsock;
 
 #[cfg(not(feature = "tee"))]

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

@@ -20,25 +20,18 @@ use std::cmp;
 use std::io::Write;
 use std::os::fd::RawFd;
 use std::path::PathBuf;
-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 virtio_bindings::{virtio_net::VIRTIO_NET_F_MAC, virtio_ring::VIRTIO_RING_F_EVENT_IDX};
+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)]

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

@@ -18,6 +18,8 @@ pub static QUEUE_CONFIG: [QueueConfig; NUM_QUEUES] = [QueueConfig::new(QUEUE_SIZ
 
 mod backend;
 pub mod device;
+#[cfg(target_os = "macos")]
+mod socket_x;
 #[cfg(target_os = "linux")]
 mod tap;
 mod unixgram;
@@ -28,13 +30,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::*;