feat: add VIRTIO_NET_F_MRG_RXBUF path to virtio-net

Add new path for virtio-net if VIRTIO_NET_F_MRG_RXBUF
feature is negotiated.

Signed-off-by: Egor Lazarchuk <[email protected]>

Author: ShadowCurse

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

@@ -9,15 +9,16 @@
 use std::io::Read;
 use std::mem;
 use std::net::Ipv4Addr;
+use std::num::Wrapping;
 use std::sync::atomic::AtomicU32;
 use std::sync::{Arc, Mutex};
 
-use libc::EAGAIN;
+use libc::{iovec, EAGAIN};
 use log::{error, warn};
 use utils::eventfd::EventFd;
 use utils::net::mac::MacAddr;
 use utils::u64_to_usize;
-use vm_memory::GuestMemoryError;
+use vm_memory::{GuestMemory, GuestMemoryError};
 
 use crate::devices::virtio::device::{DeviceState, IrqTrigger, IrqType, VirtioDevice};
 use crate::devices::virtio::gen::virtio_blk::VIRTIO_F_VERSION_1;
@@ -33,7 +34,7 @@ use crate::devices::virtio::net::tap::Tap;
 use crate::devices::virtio::net::{
     gen, NetError, NetQueue, MAX_BUFFER_SIZE, NET_QUEUE_SIZES, RX_INDEX, TX_INDEX,
 };
-use crate::devices::virtio::queue::{DescriptorChain, Queue};
+use crate::devices::virtio::queue::{Descriptor, DescriptorChain, Queue};
 use crate::devices::virtio::{ActivateError, TYPE_NET};
 use crate::devices::{report_net_event_fail, DeviceError};
 use crate::dumbo::pdu::arp::ETH_IPV4_FRAME_LEN;
@@ -128,6 +129,8 @@ pub struct Net {
 
     rx_bytes_read: usize,
     rx_frame_buf: [u8; MAX_BUFFER_SIZE],
+    rx_iov: Vec<iovec>,
+    rx_heads_info: Vec<(u16, u32, usize)>,
 
     tx_frame_headers: [u8; frame_hdr_len()],
 
@@ -145,6 +148,12 @@ pub struct Net {
     pub(crate) metrics: Arc<NetDeviceMetrics>,
 }
 
+// This is needed for rx_iov as it uses iovec struct
+// which contains *mut c_void type.
+// SAFETY:
+// Safe as only vmm thread will access rx_iov field.
+unsafe impl Send for Net {}
+
 impl Net {
     /// Create a new virtio network device with the given TAP interface.
     pub fn new_with_tap(
@@ -161,6 +170,7 @@ impl Net {
             | 1 << VIRTIO_NET_F_HOST_TSO4
             | 1 << VIRTIO_NET_F_HOST_UFO
             | 1 << VIRTIO_F_VERSION_1
+            | 1 << VIRTIO_NET_F_MRG_RXBUF
             | 1 << VIRTIO_RING_F_EVENT_IDX;
 
         let mut config_space = ConfigSpace::default();
@@ -190,6 +200,8 @@ impl Net {
             rx_deferred_frame: false,
             rx_bytes_read: 0,
             rx_frame_buf: [0u8; MAX_BUFFER_SIZE],
+            rx_iov: Vec::new(),
+            rx_heads_info: Vec::new(),
             tx_frame_headers: [0u8; frame_hdr_len()],
             irq_trigger: IrqTrigger::new().map_err(NetError::EventFd)?,
             config_space,
@@ -528,7 +540,7 @@ impl Net {
         if !self.has_feature(u64::from(VIRTIO_NET_F_MRG_RXBUF)) {
             self.process_rx_orig()
         } else {
-            unimplemented!();
+            self.process_rx_iov()
         }
     }
 
@@ -568,6 +580,167 @@ impl Net {
         self.signal_used_queue(NetQueue::Rx)
     }
 
+    fn process_rx_iov(&mut self) -> Result<(), DeviceError> {
+        self.rx_iov.clear();
+        self.rx_heads_info.clear();
+
+        let mem = self.device_state.mem().unwrap();
+
+        // Preparing iov vector with all available buffers
+        // from all available descriptor chains
+        #[repr(C)]
+        struct AvailRing {
+            flags: u16,
+            idx: u16,
+            ring: [u16; 256],
+            used_element: u16,
+        }
+        // SAFETY:
+        // avail_ring in the queue is a valid guest address
+        let avail_ring: &AvailRing = unsafe {
+            std::mem::transmute(
+                mem.get_host_address(self.queues[RX_INDEX].avail_ring)
+                    .unwrap(),
+            )
+        };
+        // SAFETY:
+        // desc_table in the queue is a valid guest address
+        let desc_table: &[Descriptor; 256] = unsafe {
+            std::mem::transmute(
+                mem.get_host_address(self.queues[RX_INDEX].desc_table)
+                    .unwrap(),
+            )
+        };
+
+        let avail_idx = self.queues[RX_INDEX].avail_idx(mem);
+        let actual_size = self.queues[RX_INDEX].actual_size();
+        let mut next_avail = self.queues[RX_INDEX].next_avail;
+
+        while next_avail.0 != avail_idx.0 {
+            let index = next_avail.0 % actual_size;
+            let desc_index = avail_ring.ring[index as usize];
+
+            let mut desc = &desc_table[desc_index as usize];
+
+            let mut chain_capacity = desc.len;
+            let mut iov_len = 1;
+            self.rx_iov.push(iovec {
+                iov_base: mem
+                    .get_host_address(vm_memory::GuestAddress(desc.addr))
+                    .unwrap()
+                    .cast(),
+                iov_len: desc.len as usize,
+            });
+
+            while desc.flags & crate::devices::virtio::queue::VIRTQ_DESC_F_NEXT != 0 {
+                desc = &desc_table[desc.next as usize];
+                chain_capacity += desc.len;
+                iov_len += 1;
+                self.rx_iov.push(iovec {
+                    iov_base: mem
+                        .get_host_address(vm_memory::GuestAddress(desc.addr))
+                        .unwrap()
+                        .cast(),
+                    iov_len: desc.len as usize,
+                });
+            }
+
+            self.rx_heads_info
+                .push((desc_index, chain_capacity, iov_len));
+
+            next_avail += Wrapping(1);
+        }
+
+        // If threre are no buffers, there is
+        // nothing for us to do.
+        if self.rx_iov.is_empty() {
+            return Ok(());
+        }
+
+        let mut iov_slice = &self.rx_iov[..];
+        let mut heads_info_slice = &self.rx_heads_info[..];
+        loop {
+            // If we used all iovs, we cannot read
+            // anything else.
+            if iov_slice.is_empty() {
+                break;
+            }
+
+            match self.tap.read_iovec(iov_slice).map_err(NetError::IO) {
+                Ok(mut bytes_written) => {
+                    let mut used_iovs = 0;
+                    let mut used_heads = 0;
+                    // Calculate how manu descriptor heads we used for this write.
+                    loop {
+                        if used_heads == heads_info_slice.len() {
+                            break;
+                        }
+
+                        let (head_index, iov_capacity, iov_len) = heads_info_slice[used_heads];
+                        used_heads += 1;
+                        used_iovs += iov_len;
+
+                        if bytes_written < iov_capacity as usize {
+                            self.queues[RX_INDEX]
+                                .add_used(mem, head_index, bytes_written as u32)
+                                .unwrap();
+                            break;
+                        } else {
+                            self.queues[RX_INDEX]
+                                .add_used(mem, head_index, iov_capacity)
+                                .unwrap();
+                            bytes_written -= iov_capacity as usize;
+                        };
+                    }
+
+                    // Update number of descrptor heads used to store
+                    // a packet.
+                    // SAFETY:
+                    // The iov_base is valid userspace address.
+                    #[allow(clippy::transmute_ptr_to_ref)]
+                    let header: &mut virtio_net_hdr_v1 =
+                        unsafe { std::mem::transmute(iov_slice[0].iov_base) };
+                    header.num_buffers = used_heads as u16;
+
+                    iov_slice = &iov_slice[used_iovs..];
+                    heads_info_slice = &heads_info_slice[used_heads..];
+
+                    // Update the number of used descriptor heads
+                    self.queues[RX_INDEX].next_avail += Wrapping(used_heads as u16);
+
+                    // Tell the guest about used descriptor heads.
+                    if self.queues[RX_INDEX].uses_notif_suppression {
+                        if self.queues[RX_INDEX].len(mem) == 0 {
+                            let next_avail = self.queues[RX_INDEX].next_avail.0;
+                            self.queues[RX_INDEX].set_avail_event(next_avail, mem);
+                        }
+                    } else {
+                        let next_avail = self.queues[RX_INDEX].next_avail.0;
+                        self.queues[RX_INDEX].set_avail_event(next_avail, mem);
+                    }
+                }
+                Err(NetError::IO(err)) => {
+                    // The tap device is non-blocking, so any error aside from EAGAIN is
+                    // unexpected.
+                    match err.raw_os_error() {
+                        Some(err) if err == EAGAIN => (),
+                        _ => {
+                            error!("Failed to read tap: {:?}", err);
+                            self.metrics.tap_read_fails.inc();
+                            return Err(DeviceError::FailedReadTap);
+                        }
+                    };
+                    break;
+                }
+                Err(err) => {
+                    error!("Spurious error in network RX: {:?}", err);
+                }
+            }
+        }
+        self.signal_used_queue(NetQueue::Rx)?;
+        Ok(())
+    }
+
     // Process the deferred frame first, then continue reading from tap.
     fn handle_deferred_frame(&mut self) -> Result<(), DeviceError> {
         if self.rate_limited_rx_single_frame() {