feat(virtq): send logs over virtq

Signed-off-by: Tomasz Andrzejak <andreiltd@gmail.com>

Author: andreiltd

src/hyperlight_common/src/virtq/buffer.rs

@@ -33,7 +33,9 @@ pub enum AllocError {
     InvalidArg,
     #[error("Empty region")]
     EmptyRegion,
-    #[error("Out of memory")]
+    #[error("No space available")]
+    NoSpace,
+    #[error("Requested size exceeds pool capacity")]
     OutOfMemory,
     #[error("Overflow")]
     Overflow,

src/hyperlight_common/src/virtq/mod.rs

@@ -181,9 +181,13 @@ pub trait Notifier {
 #[derive(Error, Debug)]
 pub enum VirtqError {
     #[error("Ring error: {0}")]
-    RingError(#[from] RingError),
+    RingError(RingError),
     #[error("Allocation error: {0}")]
-    Alloc(#[from] AllocError),
+    Alloc(AllocError),
+    #[error("Ring or pool temporarily full")]
+    Backpressure,
+    #[error("Allocation exceeds pool capacity")]
+    OutOfMemory,
     #[error("Invalid token")]
     BadToken,
     #[error("Invalid chain received")]
@@ -202,6 +206,33 @@ pub enum VirtqError {
     NoReadableBuffer,
 }
 
+impl VirtqError {
+    /// Check if this error is transient or unrecoverable.
+    #[inline(always)]
+    pub fn is_transient(&self) -> bool {
+        matches!(self, Self::Backpressure)
+    }
+}
+
+impl From<RingError> for VirtqError {
+    fn from(e: RingError) -> Self {
+        match e {
+            RingError::WouldBlock => Self::Backpressure,
+            other => Self::RingError(other),
+        }
+    }
+}
+
+impl From<AllocError> for VirtqError {
+    fn from(e: AllocError) -> Self {
+        match e {
+            AllocError::NoSpace => Self::Backpressure,
+            AllocError::OutOfMemory => Self::OutOfMemory,
+            other => Self::Alloc(other),
+        }
+    }
+}
+
 /// Layout of a packed virtqueue ring in shared memory.
 ///
 /// Describes the memory addresses for the descriptor table and event suppression
@@ -424,7 +455,7 @@ pub(crate) mod test_utils {
             let addr = self.next.fetch_add(len as u64, Ordering::Relaxed);
             let end = addr + len as u64;
             if end > self.base + self.size as u64 {
-                return Err(AllocError::OutOfMemory);
+                return Err(AllocError::NoSpace);
             }
             Ok(Allocation { addr, len })
         }
@@ -794,6 +825,153 @@ mod tests {
         assert_eq!(&expected_first.1[..], b"resp1");
         assert_eq!(&expected_second.1[..], b"resp2");
     }
+
+    /// Helper: submit a ReadOnly entry (entry data, no completion).
+    fn send_readonly(
+        producer: &mut VirtqProducer<TestMem, TestNotifier, TestPool>,
+        entry_data: &[u8],
+    ) -> Token {
+        let mut se = producer.chain().entry(entry_data.len()).build().unwrap();
+        se.write_all(entry_data).unwrap();
+        producer.submit(se).unwrap()
+    }
+
+    #[test]
+    fn test_reclaim_frees_ring_slots() {
+        let ring = make_ring(4);
+        let (mut producer, mut consumer, _) = make_test_producer(&ring);
+
+        // Fill the ring with ReadOnly entries
+        send_readonly(&mut producer, b"a");
+        send_readonly(&mut producer, b"b");
+        send_readonly(&mut producer, b"c");
+        send_readonly(&mut producer, b"d");
+
+        // Ring is now full - next submit should fail with Backpressure
+        let mut se = producer.chain().entry(1).build().unwrap();
+        se.write_all(b"e").unwrap();
+        let res = producer.submit(se);
+        assert!(
+            matches!(res, Err(VirtqError::Backpressure)),
+            "expected Backpressure from full ring"
+        );
+
+        // Consumer acks all entries
+        while let Some((_, completion)) = consumer.poll(1024).unwrap() {
+            consumer.complete(completion).unwrap();
+        }
+
+        // Reclaim should free ring slots without losing data
+        let count = producer.reclaim().unwrap();
+        assert_eq!(count, 4, "expected 4 reclaimed entries");
+
+        // Ring should have space now
+        send_readonly(&mut producer, b"e");
+    }
+
+    #[test]
+    fn test_reclaim_buffers_rw_completions() {
+        let ring = make_ring(4);
+        let (mut producer, mut consumer, _) = make_test_producer(&ring);
+
+        // Submit a ReadWrite entry
+        let tok = send_readwrite(&mut producer, b"request", 64);
+
+        // Consumer processes and writes response
+        let (_, completion) = consumer.poll(1024).unwrap().unwrap();
+        let SendCompletion::Writable(mut wc) = completion else {
+            panic!("expected writable");
+        };
+        wc.write_all(b"response-data").unwrap();
+        consumer.complete(wc.into()).unwrap();
+
+        // Reclaim buffers the completion (doesn't discard it)
+        let count = producer.reclaim().unwrap();
+        assert_eq!(count, 1);
+
+        // poll() should return the buffered completion
+        let cqe = producer.poll().unwrap().unwrap();
+        assert_eq!(cqe.token, tok);
+        assert_eq!(&cqe.data[..], b"response-data");
+    }
+
+    #[test]
+    fn test_reclaim_then_poll_preserves_order() {
+        let ring = make_ring(8);
+        let (mut producer, mut consumer, _) = make_test_producer(&ring);
+
+        // Submit 3 entries: RO, RW, RO
+        let tok_ro1 = send_readonly(&mut producer, b"log1");
+        let tok_rw = send_readwrite(&mut producer, b"call", 64);
+        let tok_ro2 = send_readonly(&mut producer, b"log2");
+
+        // Consumer processes all 3
+        let (_, c1) = consumer.poll(1024).unwrap().unwrap();
+        consumer.complete(c1).unwrap(); // ack RO
+
+        let (_, c2) = consumer.poll(1024).unwrap().unwrap();
+        let SendCompletion::Writable(mut wc) = c2 else {
+            panic!("expected writable");
+        };
+        wc.write_all(b"result").unwrap();
+        consumer.complete(wc.into()).unwrap(); // complete RW
+
+        let (_, c3) = consumer.poll(1024).unwrap().unwrap();
+        consumer.complete(c3).unwrap(); // ack RO
+
+        // Reclaim all 3
+        let count = producer.reclaim().unwrap();
+        assert_eq!(count, 3);
+
+        // poll() returns them in order
+        let cqe1 = producer.poll().unwrap().unwrap();
+        assert_eq!(cqe1.token, tok_ro1);
+        assert!(cqe1.data.is_empty());
+
+        let cqe2 = producer.poll().unwrap().unwrap();
+        assert_eq!(cqe2.token, tok_rw);
+        assert_eq!(&cqe2.data[..], b"result");
+
+        let cqe3 = producer.poll().unwrap().unwrap();
+        assert_eq!(cqe3.token, tok_ro2);
+        assert!(cqe3.data.is_empty());
+
+        // No more
+        assert!(producer.poll().unwrap().is_none());
+    }
+
+    #[test]
+    fn test_reclaim_mixed_with_poll() {
+        let ring = make_ring(8);
+        let (mut producer, mut consumer, _) = make_test_producer(&ring);
+
+        // Submit and complete 2 entries
+        send_readonly(&mut producer, b"x");
+        let tok_rw = send_readwrite(&mut producer, b"y", 64);
+
+        let (_, c1) = consumer.poll(1024).unwrap().unwrap();
+        consumer.complete(c1).unwrap();
+
+        let (_, c2) = consumer.poll(1024).unwrap().unwrap();
+        let SendCompletion::Writable(mut wc) = c2 else {
+            panic!("expected writable");
+        };
+        wc.write_all(b"reply").unwrap();
+        consumer.complete(wc.into()).unwrap();
+
+        // poll() consumes first entry directly from ring
+        let cqe1 = producer.poll().unwrap().unwrap();
+        assert!(cqe1.data.is_empty());
+
+        // reclaim() buffers second entry
+        let count = producer.reclaim().unwrap();
+        assert_eq!(count, 1);
+
+        // poll() returns the buffered one
+        let cqe2 = producer.poll().unwrap().unwrap();
+        assert_eq!(cqe2.token, tok_rw);
+        assert_eq!(&cqe2.data[..], b"reply");
+    }
 }
 #[cfg(all(test, loom))]
 mod fuzz {

src/hyperlight_common/src/virtq/msg.rs

@@ -34,6 +34,24 @@ pub enum MsgKind {
     StreamEnd = 0x04,
     /// Cancel a pending request.
     Cancel = 0x05,
+    /// A guest log message (GuestLogData payload follows).
+    Log = 0x06,
+}
+
+impl TryFrom<u8> for MsgKind {
+    type Error = u8;
+
+    fn try_from(value: u8) -> Result<Self, Self::Error> {
+        match value {
+            0x01 => Ok(Self::Request),
+            0x02 => Ok(Self::Response),
+            0x03 => Ok(Self::StreamChunk),
+            0x04 => Ok(Self::StreamEnd),
+            0x05 => Ok(Self::Cancel),
+            0x06 => Ok(Self::Log),
+            other => Err(other),
+        }
+    }
 }
 
 /// Wire header for all virtqueue messages
@@ -72,4 +90,9 @@ impl VirtqMsgHeader {
             payload_len,
         }
     }
+
+    /// Parse the kind field into a [`MsgKind`] enum.
+    pub fn msg_kind(&self) -> Result<MsgKind, u8> {
+        MsgKind::try_from(self.kind)
+    }
 }

src/hyperlight_common/src/virtq/pool.rs

@@ -169,7 +169,7 @@ impl<const N: usize> Slab<N> {
             return Err(AllocError::OutOfMemory);
         }
 
-        let idx = self.find_slots(need_slots).ok_or(AllocError::OutOfMemory)?;
+        let idx = self.find_slots(need_slots).ok_or(AllocError::NoSpace)?;
         self.used_slots.insert_range(idx..idx + need_slots);
         let addr = self.addr_of(idx).ok_or(AllocError::Overflow)?;
 
@@ -463,7 +463,7 @@ impl<const L: usize, const U: usize> Inner<L, U> {
         if len <= L {
             match self.lower.alloc(len) {
                 Ok(alloc) => return Ok(alloc),
-                Err(AllocError::OutOfMemory) => {}
+                Err(AllocError::NoSpace) => {}
                 Err(e) => return Err(e),
             }
         }
@@ -614,7 +614,7 @@ impl BufferProvider for RecyclePool {
             return Err(AllocError::OutOfMemory);
         }
 
-        let addr = inner.free.pop().ok_or(AllocError::OutOfMemory)?;
+        let addr = inner.free.pop().ok_or(AllocError::NoSpace)?;
 
         Ok(Allocation {
             addr,
@@ -727,7 +727,7 @@ mod tests {
 
         // Next allocation should fail
         let result = slab.alloc(256);
-        assert!(matches!(result, Err(AllocError::OutOfMemory)));
+        assert!(matches!(result, Err(AllocError::NoSpace)));
 
         // Free one and retry
         slab.dealloc(a2).unwrap();
@@ -1287,7 +1287,7 @@ mod fuzz {
                         assert!(alloc.len >= *size);
                         allocations.push(alloc);
                     }
-                    Err(AllocError::OutOfMemory) => {}
+                    Err(AllocError::NoSpace | AllocError::OutOfMemory) => {}
                     Err(_) => {
                         return false;
                     }
@@ -1318,7 +1318,7 @@ mod fuzz {
                             assert!(new_alloc.len >= *new_size);
                             allocations[idx] = new_alloc;
                         }
-                        Err(AllocError::OutOfMemory) => {}
+                        Err(AllocError::NoSpace | AllocError::OutOfMemory) => {}
                         Err(_) => return false,
                     }
                 }

src/hyperlight_common/src/virtq/producer.rs

@@ -14,6 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
+use alloc::collections::VecDeque;
 use alloc::vec;
 use alloc::vec::Vec;
 
@@ -124,6 +125,7 @@ pub struct VirtqProducer<M, N, P> {
     notifier: N,
     pool: P,
     inflight: Vec<Option<Inflight>>,
+    pending: VecDeque<RecvCompletion>,
 }
 
 impl<M, N, P> VirtqProducer<M, N, P>
@@ -149,11 +151,15 @@ where
             pool,
             notifier,
             inflight,
+            pending: VecDeque::new(),
         }
     }
 
     /// Poll for a single completion from the device.
     ///
+    /// Returns buffered completions from prior [`reclaim`](Self::reclaim)
+    /// calls first, then checks the ring for new completions.
+    ///
     /// Returns `Ok(Some(completion))` if a completion is available, `Ok(None)` if no
     /// completions are ready (would block), or an error if the device misbehaved.
     ///
@@ -167,6 +173,39 @@ where
     /// - [`VirtqError::InvalidState`] - Device returned invalid descriptor ID or
     ///   wrote more data than the completion buffer capacity
     pub fn poll(&mut self) -> Result<Option<RecvCompletion>, VirtqError>
+    where
+        M: Send + 'static,
+        P: Send + 'static,
+    {
+        if let Some(cqe) = self.pending.pop_front() {
+            return Ok(Some(cqe));
+        }
+        self.poll_ring()
+    }
+
+    /// Reclaim ring slots and pool entries from completed descriptors.
+    ///
+    /// Processes all available used entries from the ring: frees entry
+    /// buffer allocations immediately, and buffers completion data for
+    /// later retrieval via [`poll`](Self::poll).
+    ///
+    /// Use this to free resources under backpressure without losing
+    /// completion data. Returns the number of entries reclaimed.
+    pub fn reclaim(&mut self) -> Result<usize, VirtqError>
+    where
+        M: Send + 'static,
+        P: Send + 'static,
+    {
+        let mut count = 0;
+        while let Some(cqe) = self.poll_ring()? {
+            self.pending.push_back(cqe);
+            count += 1;
+        }
+        Ok(count)
+    }
+
+    /// Poll one completion directly from the ring (bypassing pending buffer).
+    fn poll_ring(&mut self) -> Result<Option<RecvCompletion>, VirtqError>
     where
         M: Send + 'static,
         P: Send + 'static,
@@ -363,6 +402,7 @@ where
         self.inner.reset();
         self.pool.reset();
         self.inflight.fill(None);
+        self.pending.clear();
     }
 }