Reuse Rxbuffer and update qemu_args.sh for vsock

.github/workflows/kernel_test.yml

@@ -83,28 +83,3 @@ jobs:
       - name: Regression Test (Linux EFI Handover Boot Protocol)
         id: regression_test_linux
         run: make run AUTO_TEST=regression ENABLE_KVM=0 BOOT_PROTOCOL=multiboot2 RELEASE=1
-
-  vsock-test:
-    runs-on: ubuntu-latest
-    timeout-minutes: 15
-    steps:
-      - uses: actions/checkout@v4
-
-      - name: Run Vsock Server on Host
-        id: host_vsock_server
-        run: |
-          sudo modprobe vhost_vsock
-          sudo apt-get install socat
-          echo "Run vsock server on host...."
-          socat -dd vsock-listen:1234 SYSTEM:"sleep 3s; echo 'Hello from host';sleep 2s;kill -INT $$" &
-      - name: Run Vsock Client and Server on Guest
-        id: guest_vsock_client_server
-        run: |
-          docker run  --privileged --network=host --device=/dev/kvm -v ./:/root/asterinas asterinas/asterinas:0.4.2 \
-          make run AUTO_TEST=vsock ENABLE_KVM=0 QEMU_MACHINE=microvm RELEASE_MODE=1 &
-      - name: Run Vsock Client on Host
-        id: host_vsock_client
-        run: |
-          sleep 5m
-          echo "Run vsock client on host...."
-          echo "Hello from host" | socat -dd - vsock-connect:3:4321

Makefile

@@ -35,14 +35,8 @@ CARGO_OSDK_ARGS += --init-args="/regression/run_regression_test.sh"
 else ifeq ($(AUTO_TEST), boot)
 CARGO_OSDK_ARGS += --init-args="/regression/boot_hello.sh"
 else ifeq ($(AUTO_TEST), vsock)
+export VSOCK=1
 CARGO_OSDK_ARGS += --init-args="/regression/run_vsock_test.sh"
-ifeq ($(SCHEME), microvm)
-CARGO_OSDK_ARGS += --qemu-args="-device vhost-vsock-device,guest-cid=3"
-else ifeq ($(SCHEME), iommu)
-CARGO_OSDK_ARGS += --qemu-args="-device vhost-vsock-pci,id=vhost-vsock-pci0,guest-cid=3,disable-legacy=on,disable-modern=off,iommu_platform=on,ats=on"
-else
-CARGO_OSDK_ARGS += --qemu-args="-device vhost-vsock-pci,id=vhost-vsock-pci0,guest-cid=3,disable-legacy=on,disable-modern=off"
-endif
 endif
 
 ifeq ($(RELEASE_LTO), 1)

kernel/comps/network/src/buffer.rs

@@ -17,14 +17,19 @@ use crate::dma_pool::{DmaPool, DmaSegment};
 pub struct TxBuffer {
     dma_stream: DmaStream,
     nbytes: usize,
+    pool: &'static SpinLock<LinkedList<DmaStream>>,
 }
 
 impl TxBuffer {
-    pub fn new<H: Pod>(header: &H, packet: &[u8]) -> Self {
+    pub fn new<H: Pod>(
+        header: &H,
+        packet: &[u8],
+        pool: &'static SpinLock<LinkedList<DmaStream>>,
+    ) -> Self {
         let header = header.as_bytes();
         let nbytes = header.len() + packet.len();
 
-        let dma_stream = if let Some(stream) = get_tx_stream_from_pool(nbytes) {
+        let dma_stream = if let Some(stream) = get_tx_stream_from_pool(nbytes, pool) {
             stream
         } else {
             let segment = {
@@ -38,7 +43,11 @@ impl TxBuffer {
         writer.write(&mut VmReader::from(header));
         writer.write(&mut VmReader::from(packet));
 
-        let tx_buffer = Self { dma_stream, nbytes };
+        let tx_buffer = Self {
+            dma_stream,
+            nbytes,
+            pool,
+        };
         tx_buffer.sync();
         tx_buffer
     }
@@ -64,9 +73,7 @@ impl HasDaddr for TxBuffer {
 
 impl Drop for TxBuffer {
     fn drop(&mut self) {
-        TX_BUFFER_POOL
+        self.pool
-            .get()
-            .unwrap()
             .lock_irq_disabled()
             .push_back(self.dma_stream.clone());
     }
@@ -79,9 +86,9 @@ pub struct RxBuffer {
 }
 
 impl RxBuffer {
-    pub fn new(header_len: usize) -> Self {
+    pub fn new(header_len: usize, pool: &Arc<DmaPool>) -> Self {
-        assert!(header_len <= RX_BUFFER_LEN);
+        assert!(header_len <= pool.segment_size());
-        let segment = RX_BUFFER_POOL.get().unwrap().alloc_segment().unwrap();
+        let segment = pool.alloc_segment().unwrap();
         Self {
             segment,
             header_len,
@@ -109,6 +116,16 @@ impl RxBuffer {
             .limit(self.packet_len)
     }
 
+    pub fn buf(&self) -> VmReader<'_> {
+        self.segment
+            .sync(0..self.header_len + self.packet_len)
+            .unwrap();
+        self.segment
+            .reader()
+            .unwrap()
+            .limit(self.header_len + self.packet_len)
+    }
+
     pub const fn buf_len(&self) -> usize {
         self.segment.size()
     }
@@ -121,11 +138,14 @@ impl HasDaddr for RxBuffer {
 }
 
 const RX_BUFFER_LEN: usize = 4096;
-static RX_BUFFER_POOL: Once<Arc<DmaPool>> = Once::new();
+pub static RX_BUFFER_POOL: Once<Arc<DmaPool>> = Once::new();
-static TX_BUFFER_POOL: Once<SpinLock<LinkedList<DmaStream>>> = Once::new();
+pub static TX_BUFFER_POOL: Once<SpinLock<LinkedList<DmaStream>>> = Once::new();
 
-fn get_tx_stream_from_pool(nbytes: usize) -> Option<DmaStream> {
+fn get_tx_stream_from_pool(
-    let mut pool = TX_BUFFER_POOL.get().unwrap().lock_irq_disabled();
+    nbytes: usize,
+    tx_buffer_pool: &'static SpinLock<LinkedList<DmaStream>>,
+) -> Option<DmaStream> {
+    let mut pool = tx_buffer_pool.lock_irq_disabled();
     let mut cursor = pool.cursor_front_mut();
     while let Some(current) = cursor.current() {
         if current.nbytes() >= nbytes {

kernel/comps/network/src/dma_pool.rs

@@ -127,6 +127,11 @@ impl DmaPool {
     fn num_pages(&self) -> usize {
         self.all_pages.lock_irq_disabled().len()
     }
+
+    /// Return segment size in pool
+    pub fn segment_size(&self) -> usize {
+        self.segment_size
+    }
 }
 
 #[derive(Debug)]

kernel/comps/network/src/lib.rs

@@ -17,7 +17,7 @@ use core::{any::Any, fmt::Debug};
 
 use aster_frame::sync::SpinLock;
 use aster_util::safe_ptr::Pod;
-pub use buffer::{RxBuffer, TxBuffer};
+pub use buffer::{RxBuffer, TxBuffer, RX_BUFFER_POOL, TX_BUFFER_POOL};
 use component::{init_component, ComponentInitError};
 pub use dma_pool::DmaSegment;
 use smoltcp::phy;

kernel/comps/virtio/src/device/network/device.rs

@@ -4,7 +4,10 @@ use alloc::{boxed::Box, string::ToString, sync::Arc};
 use core::{fmt::Debug, hint::spin_loop, mem::size_of};
 
 use aster_frame::{offset_of, sync::SpinLock, trap::TrapFrame};
-use aster_network::{AnyNetworkDevice, EthernetAddr, RxBuffer, TxBuffer, VirtioNetError};
+use aster_network::{
+    AnyNetworkDevice, EthernetAddr, RxBuffer, TxBuffer, VirtioNetError, RX_BUFFER_POOL,
+    TX_BUFFER_POOL,
+};
 use aster_util::{field_ptr, slot_vec::SlotVec};
 use log::debug;
 use smoltcp::phy::{DeviceCapabilities, Medium};
@@ -55,7 +58,8 @@ impl NetworkDevice {
 
         let mut rx_buffers = SlotVec::new();
         for i in 0..QUEUE_SIZE {
-            let rx_buffer = RxBuffer::new(size_of::<VirtioNetHdr>());
+            let rx_pool = RX_BUFFER_POOL.get().unwrap();
+            let rx_buffer = RxBuffer::new(size_of::<VirtioNetHdr>(), rx_pool);
             // FIEME: Replace rx_buffer with VM segment-based data structure to use dma mapping.
             let token = recv_queue.add_dma_buf(&[], &[&rx_buffer])?;
             assert_eq!(i, token);
@@ -128,7 +132,8 @@ impl NetworkDevice {
         rx_buffer.set_packet_len(len as usize);
         // FIXME: Ideally, we can reuse the returned buffer without creating new buffer.
         // But this requires locking device to be compatible with smoltcp interface.
-        let new_rx_buffer = RxBuffer::new(size_of::<VirtioNetHdr>());
+        let rx_pool = RX_BUFFER_POOL.get().unwrap();
+        let new_rx_buffer = RxBuffer::new(size_of::<VirtioNetHdr>(), rx_pool);
         self.add_rx_buffer(new_rx_buffer)?;
         Ok(rx_buffer)
     }
@@ -137,7 +142,8 @@ impl NetworkDevice {
     /// FIEME: Replace tx_buffer with VM segment-based data structure to use dma mapping.
     fn send(&mut self, packet: &[u8]) -> Result<(), VirtioNetError> {
         let header = VirtioNetHdr::default();
-        let tx_buffer = TxBuffer::new(&header, packet);
+        let tx_pool = TX_BUFFER_POOL.get().unwrap();
+        let tx_buffer = TxBuffer::new(&header, packet, tx_pool);
 
         let token = self
             .send_queue

kernel/comps/virtio/src/device/socket/buffer.rs

@@ -2,147 +2,16 @@
 
 use alloc::{collections::LinkedList, sync::Arc};
 
-use align_ext::AlignExt;
 use aster_frame::{
+    mm::{DmaDirection, DmaStream},
     sync::SpinLock,
-    vm::{Daddr, DmaDirection, DmaStream, HasDaddr, VmAllocOptions, VmReader, VmWriter, PAGE_SIZE},
 };
-use aster_network::dma_pool::{DmaPool, DmaSegment};
+use aster_network::dma_pool::DmaPool;
-use pod::Pod;
 use spin::Once;
 
-pub struct TxBuffer {
+const RX_BUFFER_LEN: usize = 4096;
-    dma_stream: DmaStream,
+pub static RX_BUFFER_POOL: Once<Arc<DmaPool>> = Once::new();
-    nbytes: usize,
+pub static TX_BUFFER_POOL: Once<SpinLock<LinkedList<DmaStream>>> = Once::new();
-}
-
-impl TxBuffer {
-    pub fn new<H: Pod>(header: &H, packet: &[u8]) -> Self {
-        let header = header.as_bytes();
-        let nbytes = header.len() + packet.len();
-
-        let dma_stream = if let Some(stream) = get_tx_stream_from_pool(nbytes) {
-            stream
-        } else {
-            let segment = {
-                let nframes = (nbytes.align_up(PAGE_SIZE)) / PAGE_SIZE;
-                VmAllocOptions::new(nframes).alloc_contiguous().unwrap()
-            };
-            DmaStream::map(segment, DmaDirection::ToDevice, false).unwrap()
-        };
-
-        let mut writer = dma_stream.writer().unwrap();
-        writer.write(&mut VmReader::from(header));
-        writer.write(&mut VmReader::from(packet));
-
-        let tx_buffer = Self { dma_stream, nbytes };
-        tx_buffer.sync();
-        tx_buffer
-    }
-
-    pub fn writer(&self) -> VmWriter<'_> {
-        self.dma_stream.writer().unwrap().limit(self.nbytes)
-    }
-
-    fn sync(&self) {
-        self.dma_stream.sync(0..self.nbytes).unwrap();
-    }
-
-    pub fn nbytes(&self) -> usize {
-        self.nbytes
-    }
-}
-
-impl HasDaddr for TxBuffer {
-    fn daddr(&self) -> Daddr {
-        self.dma_stream.daddr()
-    }
-}
-
-impl Drop for TxBuffer {
-    fn drop(&mut self) {
-        TX_BUFFER_POOL
-            .get()
-            .unwrap()
-            .lock_irq_disabled()
-            .push_back(self.dma_stream.clone());
-    }
-}
-
-pub struct RxBuffer {
-    segment: DmaSegment,
-    header_len: usize,
-    packet_len: usize,
-}
-
-impl RxBuffer {
-    pub fn new(header_len: usize) -> Self {
-        assert!(header_len <= RX_BUFFER_LEN);
-        let segment = RX_BUFFER_POOL.get().unwrap().alloc_segment().unwrap();
-        Self {
-            segment,
-            header_len,
-            packet_len: 0,
-        }
-    }
-
-    pub const fn packet_len(&self) -> usize {
-        self.packet_len
-    }
-
-    pub fn set_packet_len(&mut self, packet_len: usize) {
-        assert!(self.header_len + packet_len <= RX_BUFFER_LEN);
-        self.packet_len = packet_len;
-    }
-
-    pub fn packet(&self) -> VmReader<'_> {
-        self.segment
-            .sync(self.header_len..self.header_len + self.packet_len)
-            .unwrap();
-        self.segment
-            .reader()
-            .unwrap()
-            .skip(self.header_len)
-            .limit(self.packet_len)
-    }
-
-    pub fn buf(&self) -> VmReader<'_> {
-        self.segment
-            .sync(0..self.header_len + self.packet_len)
-            .unwrap();
-        self.segment
-            .reader()
-            .unwrap()
-            .limit(self.header_len + self.packet_len)
-    }
-
-    pub const fn buf_len(&self) -> usize {
-        self.segment.size()
-    }
-}
-
-impl HasDaddr for RxBuffer {
-    fn daddr(&self) -> Daddr {
-        self.segment.daddr()
-    }
-}
-
-pub const RX_BUFFER_LEN: usize = 4096;
-static RX_BUFFER_POOL: Once<Arc<DmaPool>> = Once::new();
-static TX_BUFFER_POOL: Once<SpinLock<LinkedList<DmaStream>>> = Once::new();
-
-fn get_tx_stream_from_pool(nbytes: usize) -> Option<DmaStream> {
-    let mut pool = TX_BUFFER_POOL.get().unwrap().lock_irq_disabled();
-    let mut cursor = pool.cursor_front_mut();
-    while let Some(current) = cursor.current() {
-        if current.nbytes() >= nbytes {
-            return cursor.remove_current();
-        }
-        cursor.move_next();
-    }
-
-    None
-}
 
 pub fn init() {
     const POOL_INIT_SIZE: usize = 32;

kernel/comps/virtio/src/device/socket/device.rs

@@ -3,13 +3,13 @@
 use alloc::{boxed::Box, string::ToString, sync::Arc, vec, vec::Vec};
 use core::{fmt::Debug, hint::spin_loop, mem::size_of};
 
-use aster_frame::{offset_of, sync::SpinLock, trap::TrapFrame, vm::VmWriter};
+use aster_frame::{mm::VmWriter, offset_of, sync::SpinLock, trap::TrapFrame};
+use aster_network::{RxBuffer, TxBuffer};
 use aster_util::{field_ptr, slot_vec::SlotVec};
 use log::debug;
 use pod::Pod;
 
 use super::{
-    buffer::RxBuffer,
     config::{VirtioVsockConfig, VsockFeatures},
     connect::{ConnectionInfo, VsockEvent},
     error::SocketError,
@@ -18,7 +18,10 @@ use super::{
 };
 use crate::{
     device::{
-        socket::{buffer::TxBuffer, handle_recv_irq, register_device},
+        socket::{
+            buffer::{RX_BUFFER_POOL, TX_BUFFER_POOL},
+            handle_recv_irq, register_device,
+        },
         VirtioDeviceError,
     },
     queue::{QueueError, VirtQueue},
@@ -68,7 +71,8 @@ impl SocketDevice {
         // Allocate and add buffers for the RX queue.
         let mut rx_buffers = SlotVec::new();
         for i in 0..QUEUE_SIZE {
-            let rx_buffer = RxBuffer::new(size_of::<VirtioVsockHdr>());
+            let rx_pool = RX_BUFFER_POOL.get().unwrap();
+            let rx_buffer = RxBuffer::new(size_of::<VirtioVsockHdr>(), rx_pool);
             let token = recv_queue.add_dma_buf(&[], &[&rx_buffer])?;
             assert_eq!(i, token);
             assert_eq!(rx_buffers.put(rx_buffer) as u16, i);
@@ -187,7 +191,8 @@ impl SocketDevice {
     ) -> Result<(), SocketError> {
         debug!("Sent packet {:?}. Op {:?}", header, header.op());
         debug!("buffer in send_packet_to_tx_queue: {:?}", buffer);
-        let tx_buffer = TxBuffer::new(header, buffer);
+        let tx_pool = TX_BUFFER_POOL.get().unwrap();
+        let tx_buffer = TxBuffer::new(header, buffer, tx_pool);
 
         let token = self.send_queue.add_dma_buf(&[&tx_buffer], &[])?;
 
@@ -269,7 +274,8 @@ impl SocketDevice {
             .ok_or(QueueError::WrongToken)?;
         rx_buffer.set_packet_len(len as usize);
 
-        let new_rx_buffer = RxBuffer::new(size_of::<VirtioVsockHdr>());
+        let rx_pool = RX_BUFFER_POOL.get().unwrap();
+        let new_rx_buffer = RxBuffer::new(size_of::<VirtioVsockHdr>(), rx_pool);
         self.add_rx_buffer(new_rx_buffer, token)?;
 
         Ok(rx_buffer)

kernel/comps/virtio/src/dma_buf.rs

@@ -3,8 +3,6 @@
 use aster_frame::mm::{DmaCoherent, DmaStream, DmaStreamSlice, HasDaddr};
 use aster_network::{DmaSegment, RxBuffer, TxBuffer};
 
-use crate::device;
-
 /// A DMA-capable buffer.
 ///
 /// Any type implements this trait should also implements `HasDaddr` trait,
@@ -50,15 +48,3 @@ impl DmaBuf for RxBuffer {
         self.buf_len()
     }
 }
-
-impl DmaBuf for device::socket::buffer::TxBuffer {
-    fn len(&self) -> usize {
-        self.nbytes()
-    }
-}
-
-impl DmaBuf for device::socket::buffer::RxBuffer {
-    fn len(&self) -> usize {
-        self.buf_len()
-    }
-}

regression/apps/vsock/vsock_server.c

@@ -7,7 +7,6 @@
 #include <unistd.h>
 #include <string.h>
 
-#define CID 3
 #define PORT 4321
 
 int main()
@@ -23,8 +22,9 @@ int main()
 		return -1;
 	}
 	printf("\nCreate socket successfully\n");
+
 	serv_addr.svm_family = AF_VSOCK;
-	serv_addr.svm_cid = CID;
+	serv_addr.svm_cid = VMADDR_CID_ANY;
 	serv_addr.svm_port = PORT;
 
 	if (bind(sock, (struct sockaddr *)&serv_addr, sizeof(serv_addr)) < 0) {

tools/qemu_args.sh

@@ -5,6 +5,7 @@
 # This script is used to generate QEMU arguments for OSDK.
 # The positional argument $1 is the scheme.
 # A switch "-ovmf" can be passed as an argument to enable OVMF.
+# The positional argument $2 can be passed as "vsock" to trigger vsock module.
 
 RAND_PORT_NUM1=$(shuf -i 1024-65535 -n 1)
 RAND_PORT_NUM2=$(shuf -i 1024-65535 -n 1)
@@ -61,6 +62,24 @@ MICROVM_QEMU_ARGS="\
     -device virtconsole,chardev=mux \
 "
 
+if [ "$VSOCK" = "1" ]; then
+    # RAND_CID=$(shuf -i 3-65535 -n 1)
+    RAND_CID=3
+    echo "[$1] Launched QEMU VM with CID $RAND_CID" 1>&2
+    if [ "$1" = "microvm" ]; then
+        MICROVM_QEMU_ARGS="
+            $MICROVM_QEMU_ARGS \
+            -device vhost-vsock-device,guest-cid=$RAND_CID \
+        "
+    else
+        QEMU_ARGS="
+            $QEMU_ARGS \
+            -device vhost-vsock-pci,id=vhost-vsock-pci0,guest-cid=$RAND_CID,disable-legacy=on,disable-modern=off$IOMMU_DEV_EXTRA \
+        "
+    fi
+fi
+
+
 if [ "$1" = "microvm" ]; then
     QEMU_ARGS=$MICROVM_QEMU_ARGS
     echo $QEMU_ARGS