test new iov approach

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

Author: ShadowCurse

src/vmm/src/devices/virtio/iovec.rs

@@ -40,9 +40,9 @@ type IoVecVec = SmallVec<[iovec; 4]>;
 #[derive(Debug)]
 pub struct IoVecBuffer {
     // container of the memory regions included in this IO vector
-    vecs: IoVecVec,
+    pub vecs: IoVecVec,
     // Total length of the IoVecBuffer
-    len: u32,
+    pub len: u32,
 }
 
 impl IoVecBuffer {
@@ -53,9 +53,9 @@ impl IoVecBuffer {
 
         let mut next_descriptor = Some(head);
         while let Some(desc) = next_descriptor {
-            if desc.is_write_only() {
-                return Err(IoVecError::WriteOnlyDescriptor);
-            }
+            // if desc.is_write_only() {
+            //     return Err(IoVecError::WriteOnlyDescriptor);
+            // }
 
             // We use get_slice instead of `get_host_address` here in order to have the whole
             // range of the descriptor chain checked, i.e. [addr, addr + len) is a valid memory

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

@@ -9,6 +9,7 @@
 use std::io::Read;
 use std::mem;
 use std::net::Ipv4Addr;
+use std::os::fd::AsRawFd;
 use std::sync::atomic::AtomicU32;
 use std::sync::{Arc, Mutex};
 
@@ -286,6 +287,7 @@ impl Net {
         };
 
         if queue.prepare_kick(mem) {
+            // warn!("trigger: {:?}", queue_type);
             self.irq_trigger
                 .trigger_irq(IrqType::Vring)
                 .map_err(|err| {
@@ -405,12 +407,19 @@ impl Net {
 
         let queue = &mut self.queues[RX_INDEX];
 
-        // let mut i = 0;
+        let mut i = 0;
+        let mut nd = 0;
         let mut bytes = &self.rx_frame_buf[self.rx_bytes_send..self.rx_bytes_read];
         warn!("queue size: {}", queue.len(mem));
-        warn!("do_write_frame_to_guest bytes len: {}", bytes.len());
+        warn!(
+            "do_write_frame_to_guest bytes s: {} e: {} len: {}",
+            self.rx_bytes_send,
+            self.rx_bytes_read,
+            bytes.len()
+        );
+
         loop {
-            // i += 1;
+            i += 1;
             // warn!("do_write_frame_to_guest iteration: {i}");
             let head_descriptor = queue.pop_or_enable_notification(mem).ok_or_else(|| {
                 self.metrics.no_rx_avail_buffer.inc();
@@ -469,6 +478,7 @@ impl Net {
                         //     self.rx_descriptors_used
                         // );
 
+                        nd = self.rx_descriptors_used;
                         self.rx_bytes_send = 0;
                         self.rx_header_addr = None;
                         self.rx_descriptors_used = 0;
@@ -483,6 +493,7 @@ impl Net {
                 }
             }
         }
+        warn!("used {i} descriptor chains, {nd} descriptors");
         Ok(())
     }
 
@@ -545,8 +556,15 @@ impl Net {
         }
 
         let _metric = net_metrics.tap_write_agg.record_latency_metrics();
+        warn!(
+            "+++++++++++ writing to the tap iov with len: {} ++++++++",
+            frame_iovec.iovec_count()
+        );
+        for v in frame_iovec.vecs.iter() {
+            warn!("iov len: {}", v.iov_len);
+        }
         match Self::write_tap(tap, frame_iovec) {
-            Ok(_) => {
+            Ok(l) => {
                 let len = u64::from(frame_iovec.len());
                 net_metrics.tx_bytes_count.add(len);
                 net_metrics.tx_packets_count.inc();
@@ -574,20 +592,26 @@ impl Net {
             }
         } else {
             self.rx_bytes_read = self.read_tap().map_err(NetError::IO)?;
+            warn!("read {} bytes from tap", self.rx_bytes_read);
         }
         Ok(())
     }
 
     fn process_rx(&mut self) -> Result<(), DeviceError> {
-        // warn!("process_rx");
-        // Read as many frames as possible.
+        self.process_rx_iov()
+    }
 
+    fn process_rx_orig(&mut self) -> Result<(), DeviceError> {
+        warn!("------- process_rx ----------");
         let t = std::time::Instant::now();
+
+        // Read as many frames as possible.
         loop {
             match self.read_from_mmds_or_tap() {
                 Ok(_) => {
                     self.metrics.rx_count.inc();
                     if !self.rate_limited_rx_single_frame() {
+                        warn!("setting deferred_frame");
                         self.rx_deferred_frame = true;
                         break;
                     }
@@ -603,6 +627,7 @@ impl Net {
                             return Err(DeviceError::FailedReadTap);
                         }
                     };
+                    warn!("net IO error: {:?}", err);
                     break;
                 }
                 Err(err) => {
@@ -617,9 +642,107 @@ impl Net {
         self.signal_used_queue(NetQueue::Rx)
     }
 
+    fn process_rx_iov(&mut self) -> Result<(), DeviceError> {
+        warn!("------- process_rx ----------");
+        let t = std::time::Instant::now();
+
+        let mem = self.device_state.mem().unwrap();
+
+        let queue_len = self.queues[RX_INDEX].len(mem);
+        warn!("Queue len: {queue_len}");
+        if queue_len == 0 {
+            return Ok(());
+        }
+
+        let mut used_heads = 0;
+        loop {
+            for _ in 0..used_heads {
+                _ = self.queues[RX_INDEX].pop(mem);
+            }
+
+            let mut the_iov = IoVecBuffer {
+                vecs: Default::default(),
+                len: 0,
+            };
+            let mut index_iov = Vec::new();
+            let mut q = 0;
+            while let Some(head_descriptor) = self.queues[RX_INDEX].pop(mem) {
+                q += 1;
+                let hi = head_descriptor.index;
+                let iov = IoVecBuffer::from_descriptor_chain(head_descriptor).unwrap();
+                the_iov.vecs.extend_from_slice(&iov.vecs);
+                the_iov.len += iov.len;
+                index_iov.push((hi, iov));
+            }
+            for _ in 0..q {
+                self.queues[RX_INDEX].undo_pop();
+            }
+            warn!(
+                "all_iov: vecs: {}, len: {}",
+                the_iov.vecs.len(),
+                the_iov.len
+            );
+            if index_iov.is_empty() {
+                return Ok(());
+            }
+
+            match self.tap.read_iovec(&the_iov).map_err(NetError::IO) {
+                Ok(mut bytes_written) => {
+                    warn!("bytes written: {bytes_written}");
+                    let mut heads = 0;
+                    loop {
+                        if heads == index_iov.len() {
+                            warn!("exausted all {heads} iovecs");
+                            break;
+                        }
+
+                        let (head_index, iov) = &index_iov[heads];
+
+                        if bytes_written < iov.len() as usize {
+                            warn!(
+                                "head with index: {head_index} used {bytes_written} out of {} bytes",
+                                iov.len()
+                            );
+                            _ = self.queues[RX_INDEX].try_enable_notification(mem);
+                            self.queues[RX_INDEX]
+                                .add_used(mem, *head_index, bytes_written as u32)
+                                .unwrap();
+                            break;
+                        } else {
+                            warn!(
+                                "head with index: {head_index} used all out of {} bytes",
+                                iov.len()
+                            );
+                            _ = self.queues[RX_INDEX].try_enable_notification(mem);
+                            self.queues[RX_INDEX]
+                                .add_used(mem, *head_index, iov.len())
+                                .unwrap();
+                            bytes_written -= iov.len() as usize;
+                        };
+                        heads += 1;
+                    }
+                    warn!("heads: {heads}");
+                    #[allow(clippy::transmute_ptr_to_ref)]
+                    let header: &mut virtio_net_hdr_v1 =
+                        unsafe { std::mem::transmute(the_iov.vecs[0].iov_base) };
+                    header.num_buffers = heads as u16 + 1;
+                    used_heads += heads + 1;
+                }
+                Err(e) => {
+                    break;
+                }
+            }
+        }
+
+        warn!("process_rx took: {}ns", t.elapsed().as_nanos());
+
+        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> {
-        warn!("handle_deferred_frame");
+        // warn!("handle_deferred_frame");
         if self.rate_limited_rx_single_frame() {
             self.rx_deferred_frame = false;
             // process_rx() was interrupted possibly before consuming all
@@ -919,6 +1042,7 @@ impl VirtioDevice for Net {
     }
 
     fn activate(&mut self, mem: GuestMemoryMmap) -> Result<(), ActivateError> {
+        // warn!("NET ACKED FEATURES: {:#x}", self.acked_features);
         let event_idx = self.has_feature(u64::from(VIRTIO_RING_F_EVENT_IDX));
         if event_idx {
             for queue in &mut self.queues {

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

@@ -185,6 +185,20 @@ impl Tap {
         Ok(())
     }
 
+    /// Read an `IoVecBuffer` from tap
+    pub(crate) fn read_iovec(&mut self, buffer: &IoVecBuffer) -> Result<usize, IoError> {
+        let iovcnt = i32::try_from(buffer.iovec_count()).unwrap();
+        let iov = buffer.as_iovec_ptr();
+
+        // SAFETY: `readv` is safe. Called with a valid tap fd, the iovec pointer and length
+        // is provide by the `IoVecBuffer` implementation and we check the return value.
+        let ret = unsafe { libc::readv(self.tap_file.as_raw_fd(), iov, iovcnt) };
+        if ret == -1 {
+            return Err(IoError::last_os_error());
+        }
+        Ok(usize::try_from(ret).unwrap())
+    }
+
     /// Write an `IoVecBuffer` to tap
     pub(crate) fn write_iovec(&mut self, buffer: &IoVecBuffer) -> Result<usize, IoError> {
         let iovcnt = i32::try_from(buffer.iovec_count()).unwrap();