vsock/virtio-serial: pop all the received packets

src/crypto/Cargo.toml

@@ -9,7 +9,7 @@ cfg-if = "1.0"
 der = {version = "0.7.9", features = ["oid", "alloc", "derive"]}
 pki-types = { package = "rustls-pki-types", version = "1" }
 rust_std_stub = { path = "../std-support/rust-std-stub" }
-rustls = { version = "=0.23.12", default-features = false, features = ["ring" ], optional = true }
+rustls = { version = "0.23", default-features = false, features = ["ring" ], optional = true }
 ring = { path = "../../deps/td-shim/library/ring", default-features = false, features = ["alloc", "less-safe-getrandom-custom-or-rdrand"], optional = true }
 sys_time = { path = "../std-support/sys_time" }
 zeroize = "1.5.7"

src/devices/virtio_serial/src/lib.rs

@@ -854,6 +854,18 @@ impl VirtioSerial {
     fn port_queue_pop(port_id: u32) -> Option<Vec<u8>> {
         RECEIVE_QUEUES.lock().get_mut(&port_id)?.pop_front()
     }
+
+    fn can_recv(&self, port_id: u32) -> bool {
+        RECEIVE_QUEUES
+            .lock()
+            .get(&port_id)
+            .is_some_and(|q| !q.is_empty())
+            || self
+                .queues
+                .index(Self::port_queue_index(port_id) as usize)
+                .borrow_mut()
+                .can_pop()
+    }
 }
 
 /// Align `size` up to a page.

src/devices/virtio_serial/src/port.rs

@@ -64,26 +64,37 @@ impl VirtioSerialPort {
 
     pub fn recv(&mut self, data: &mut [u8]) -> Result<usize> {
         if self.cache.is_empty() {
-            let recv_bytes = SERIAL_DEVICE
-                .lock()
-                .get_mut()
-                .ok_or(VirtioSerialError::InvalidParameter)?
-                .dequeue(self.port_id, DEFAULT_TIMEOUT)?;
-            self.cache.push_back(recv_bytes);
+            loop {
+                let recv_bytes = SERIAL_DEVICE
+                    .lock()
+                    .get_mut()
+                    .ok_or(VirtioSerialError::InvalidParameter)?
+                    .dequeue(self.port_id, DEFAULT_TIMEOUT)?;
+                self.cache.push_back(recv_bytes);
+
+                if !SERIAL_DEVICE
+                    .lock()
+                    .get_mut()
+                    .ok_or(VirtioSerialError::InvalidParameter)?
+                    .can_recv(self.port_id)
+                {
+                    break;
+                }
+            }
         }
 
         let mut recvd = 0;
-        if !self.cache.is_empty() {
+        while !self.cache.is_empty() {
             let front = self.cache.front_mut().unwrap();
             let expect = data.len() - recvd;
             if front.len() <= expect {
-                data[..front.len()].copy_from_slice(front);
+                data[recvd..recvd + front.len()].copy_from_slice(front);
                 recvd += front.len();
                 self.cache.pop_front();
             } else {
                 data[recvd..].copy_from_slice(&front[..expect]);
                 front.drain(..expect);
-                recvd = expect;
+                recvd += expect;
             }
         }
 

src/devices/vsock/src/stream.rs

@@ -342,25 +342,40 @@ impl VsockStream {
             return Err(VsockError::Illegal);
         }
 
-        while self.data_queue.is_empty() {
-            self.recv_packet_connected()?;
+        if self.data_queue.is_empty() {
+            loop {
+                self.recv_packet_connected()?;
+
+                // If there are received vsock packets, continue to pop them out and insert to the
+                // `data_queue`. If there is no vsock packet left in the device, break the loop.
+                if !VSOCK_DEVICE
+                    .lock()
+                    .get_mut()
+                    .ok_or(VsockError::DeviceNotAvailable)?
+                    .transport
+                    .can_recv()
+                {
+                    break;
+                }
+            }
         }
 
-        let mut recvd = 0;
-        if !self.data_queue.is_empty() {
+        let mut used = 0;
+        while !self.data_queue.is_empty() && used < buf.len() {
             let head = self.data_queue.front_mut().unwrap();
-            if head.len() <= buf.len() {
-                buf[..head.len()].copy_from_slice(head);
-                recvd = head.len();
+            let free = buf.len() - used;
+            if head.len() <= free {
+                buf[used..used + head.len()].copy_from_slice(head);
+                used += head.len();
                 self.data_queue.pop_front();
             } else {
-                buf.copy_from_slice(&head[..buf.len()]);
-                recvd = buf.len();
-                head.drain(..buf.len());
+                buf[used..].copy_from_slice(&head[..free]);
+                used += free;
+                head.drain(..free);
             }
         }
 
-        Ok(recvd)
+        Ok(used)
     }
 
     fn reset(&mut self) -> Result {