fix(net/vsock): receiving logic

Cargo.toml

@@ -71,14 +71,14 @@ allocator = { git = "https://github.com/arceos-org/allocator.git", tag = "v0.1.2
     "axerrno",
 ] }
 axbacktrace = "0.1"
-axdriver_base = { git = "https://github.com/arceos-org/axdriver_crates.git", tag = "dev-v01" }
-axdriver_block = { git = "https://github.com/arceos-org/axdriver_crates.git", tag = "dev-v01" }
-axdriver_display = { git = "https://github.com/arceos-org/axdriver_crates.git", tag = "dev-v01" }
-axdriver_input = { git = "https://github.com/arceos-org/axdriver_crates.git", tag = "dev-v01" }
-axdriver_net = { git = "https://github.com/arceos-org/axdriver_crates.git", tag = "dev-v01" }
-axdriver_pci = { git = "https://github.com/arceos-org/axdriver_crates.git", tag = "dev-v01" }
-axdriver_virtio = { git = "https://github.com/arceos-org/axdriver_crates.git", tag = "dev-v01" }
-axdriver_vsock = { git = "https://github.com/arceos-org/axdriver_crates.git", tag = "dev-v01" }
+axdriver_base = { git = "https://github.com/arceos-org/axdriver_crates.git", tag = "v0.1.2-dev.0" }
+axdriver_block = { git = "https://github.com/arceos-org/axdriver_crates.git", tag = "v0.1.2-dev.0" }
+axdriver_display = { git = "https://github.com/arceos-org/axdriver_crates.git", tag = "v0.1.2-dev.0" }
+axdriver_input = { git = "https://github.com/arceos-org/axdriver_crates.git", tag = "v0.1.2-dev.0" }
+axdriver_net = { git = "https://github.com/arceos-org/axdriver_crates.git", tag = "v0.1.2-dev.0" }
+axdriver_pci = { git = "https://github.com/arceos-org/axdriver_crates.git", tag = "v0.1.2-dev.0" }
+axdriver_virtio = { git = "https://github.com/arceos-org/axdriver_crates.git", tag = "v0.1.2-dev.0" }
+axdriver_vsock = { git = "https://github.com/arceos-org/axdriver_crates.git", tag = "v0.1.2-dev.0" }
 axerrno = "0.2"
 axfs-ng-vfs = "0.1"
 axio = "0.3.0-pre.1"

modules/axnet/src/device/vsock.rs

@@ -1,3 +1,4 @@
+use alloc::collections::VecDeque;
 use core::{
     sync::atomic::{AtomicBool, AtomicU64, Ordering},
     time::Duration,
@@ -12,6 +13,9 @@ use crate::{alloc::string::ToString, vsock::connection_manager::VSOCK_CONN_MANAG
 
 // we need a global and static only one vsock device
 static VSOCK_DEVICE: Mutex<Option<AxVsockDevice>> = Mutex::new(None);
+static PENDING_EVENTS: Mutex<VecDeque<VsockDriverEvent>> = Mutex::new(VecDeque::new());
+
+const VSOCK_RX_TMPBUF_SIZE: usize = 0x1000; // 4KiB buffer for vsock receive
 
 /// Registers a vsock device. Only one vsock device can be registered.
 pub fn register_vsock_device(dev: AxVsockDevice) -> AxResult {
@@ -90,7 +94,7 @@ pub fn stop_vsock_poll() {
     }
     *count -= 1;
     let new_count = *count;
-    debug!("stop_vsock_poll: ref_count -> {}", new_count);
+    debug!("stop_vsock_poll: ref_count -> {new_count}");
 }
 
 fn vsock_poll_loop() {
@@ -118,10 +122,7 @@ async fn poll_interfaces_adaptive() -> AxResult<()> {
 
     let (idle_count, interval_us) = POLL_FREQUENCY.stats();
     if idle_count > 0 && idle_count % 10 == 0 {
-        trace!(
-            "Poll frequency: idle_count={}, interval={}μs",
-            idle_count, interval_us
-        );
+        trace!("Poll frequency: idle_count={idle_count}, interval={interval_us}μs",);
     }
     axtask::future::sleep(interval).await;
     Ok(())
@@ -131,43 +132,80 @@ fn poll_vsock_interfaces() -> AxResult<bool> {
     let mut guard = VSOCK_DEVICE.lock();
     let dev = guard.as_mut().ok_or(AxError::NotFound)?;
     let mut event_count = 0;
-    let mut buf = alloc::vec![0; 0x1000]; // 4KiB buffer for receiving data
+    let mut buf = alloc::vec![0; VSOCK_RX_TMPBUF_SIZE];
+
+    // Process pending events first
+    // Use core::mem::take to atomically move all events out and empty the global queue
+    let pending_events = core::mem::take(&mut *PENDING_EVENTS.lock());
+    for event in pending_events {
+        handle_vsock_event(event, dev, &mut buf);
+    }
 
     loop {
-        match dev.poll_event(&mut buf) {
+        match dev.poll_event() {
             Ok(None) => break, // no more events
             Ok(Some(event)) => {
                 event_count += 1;
-                handle_vsock_event(event, &buf);
+                handle_vsock_event(event, dev, &mut buf);
             }
             Err(e) => {
-                info!("Failed to poll vsock event: {:?}", e);
+                info!("Failed to poll vsock event: {e:?}");
                 break;
             }
         }
     }
     Ok(event_count > 0)
 }
 
-fn handle_vsock_event(event: VsockDriverEvent, buf: &[u8]) {
+fn handle_vsock_event(event: VsockDriverEvent, dev: &mut AxVsockDevice, buf: &mut [u8]) {
     let mut manager = VSOCK_CONN_MANAGER.lock();
-    debug!("Handling vsock event: {:?}", event);
+    debug!("Handling vsock event: {event:?}");
 
     match event {
         VsockDriverEvent::ConnectionRequest(conn_id) => {
-            let _ = manager.on_connection_request(conn_id);
+            if let Err(e) = manager.on_connection_request(conn_id) {
+                info!("Connection request failed: {conn_id:?}, error={e:?}");
+            }
         }
 
         VsockDriverEvent::Received(conn_id, len) => {
-            let _ = manager.on_data_received(conn_id, &buf[..len]);
+            let free_space = if let Some(conn) = manager.get_connection(conn_id) {
+                conn.lock().rx_buffer_free()
+            } else {
+                info!("Received data for unknown connection: {conn_id:?}");
+                return;
+            };
+
+            if free_space == 0 {
+                PENDING_EVENTS
+                    .lock()
+                    .push_back(VsockDriverEvent::Received(conn_id, len));
+                return;
+            }
+
+            let max_read = core::cmp::min(free_space, buf.len());
+            match dev.recv(conn_id, &mut buf[..max_read]) {
+                Ok(read_len) => {
+                    if let Err(e) = manager.on_data_received(conn_id, &buf[..read_len]) {
+                        info!("Failed to handle received data: conn_id={conn_id:?}, error={e:?}",);
+                    }
+                }
+                Err(e) => {
+                    info!("Failed to receive vsock data: conn_id={conn_id:?}, error={e:?}",);
+                }
+            }
         }
 
         VsockDriverEvent::Disconnected(conn_id) => {
-            let _ = manager.on_disconnected(conn_id);
+            if let Err(e) = manager.on_disconnected(conn_id) {
+                info!("Failed to handle disconnection: {conn_id:?}, error={e:?}",);
+            }
         }
 
         VsockDriverEvent::Connected(conn_id) => {
-            let _ = manager.on_connected(conn_id);
+            if let Err(e) = manager.on_connected(conn_id) {
+                info!("Failed to handle connection established: {conn_id:?}, error={e:?}",);
+            }
         }
 
         VsockDriverEvent::Unknown => warn!("Received unknown vsock event"),