OpenVMM: virtio-fs queues never serve requests after a restore behind the post-restore input gate

From 9fa024aa4df2439af3f8546ef4d5e86518f74d2c Mon Sep 17 00:00:00 2001
From: Enrique Saurez <[email protected]>
Date: Sun, 25 Oct 2026 21:51:16 -0700
Subject: [PATCH] virtiofs: wait for request admission instead of ending a
 queue worker

A virtio-fs queue worker ended as soon as it found request admission
closed. A restore under the post-restore input gate quiesces device
input before the VM's state units start, so the queues of a microVM
virtio-fs device start with admission closed: each worker ended at
once, task_control marked it done, and nothing started it again once
the guest acknowledged the restore and input resumed. Every request of
the guest, from its FUSE_INIT on, then waited forever, so a filesystem
attached at a restore, such as one bound to a dormant slot, could be
mounted but never used.

The worker now waits for admission to reopen, until it is stopped:
RequestAdmission keeps the wakers of the workers that wait, and
resume() wakes them. Admission still precedes queue.next(), so a closed
gate leaves descriptors on the virtqueue, and a save still finds no
request in flight.

A new test starts a microVM device's queue while input is quiesced,
checks that a posted FUSE_INIT stays on the virtqueue, resumes input,
and waits for the reply. Without the change, the reply never comes.

Co-authored-by: Copilot <[email protected]>
Copilot-Session: 34e6ffe3-3ebf-42db-bfca-e0ecfade6873
---
 .../virtio/virtiofs/src/integration_tests.rs  | 83 +++++++++++++++++--
 vm/devices/virtio/virtiofs/src/virtio.rs      | 51 +++++++++++-
 2 files changed, 123 insertions(+), 11 deletions(-)

diff --git a/vm/devices/virtio/virtiofs/src/integration_tests.rs b/vm/devices/virtio/virtiofs/src/integration_tests.rs
index 831b8df..375ce5d 100644
--- a/vm/devices/virtio/virtiofs/src/integration_tests.rs
+++ b/vm/devices/virtio/virtiofs/src/integration_tests.rs
@@ -8,12 +8,17 @@
 //! as a guest kernel would.
 
 use crate::VirtioFs;
+use crate::profile::MICROVM_ATTACHMENT_ID;
+use crate::profile::MicroVmOwnerMode;
+use crate::profile::microvm_root_identity;
 use crate::virtio::VirtioFsDevice;
 use fuse::protocol::*;
 use guestmem::GuestMemory;
 use pal_async::DefaultDriver;
 use pal_async::async_test;
+use pal_async::timer::PolledTimer;
 use pal_event::Event;
+use std::time::Duration;
 use test_with_tracing::test;
 use virtio::QueueResources;
 use virtio::VirtioDevice;
@@ -64,15 +69,42 @@ struct TestHarness {
 impl TestHarness {
     fn new(driver: &DefaultDriver) -> Self {
         let tmpdir = tempfile::tempdir().unwrap();
+        let fs = VirtioFs::new(tmpdir.path(), None).unwrap();
+        let driver_source = VmTaskDriverSource::new(SingleDriverBackend::new(driver.clone()));
+        let device = VirtioFsDevice::new(&driver_source, "testfs", fs, 0, None);
+        Self::with_device(driver, device, tmpdir)
+    }
 
+    /// Creates a harness for the fixed microVM device of a writable host
+    /// directory, whose requests run as the VMM.
+    fn new_microvm(driver: &DefaultDriver) -> Self {
+        let tmpdir = tempfile::tempdir().unwrap();
+        let driver_source = VmTaskDriverSource::new(SingleDriverBackend::new(driver.clone()));
+        let device = VirtioFsDevice::new_microvm_hostfs(
+            &driver_source,
+            MICROVM_ATTACHMENT_ID.to_owned(),
+            microvm_root_identity(tmpdir.path()).unwrap(),
+            false,
+            Vec::new(),
+            Vec::new(),
+            Vec::new(),
+            MicroVmOwnerMode::Vmm,
+            tmpdir.path(),
+            None,
+        )
+        .unwrap();
+        Self::with_device(driver, device, tmpdir)
+    }
+
+    fn with_device(
+        driver: &DefaultDriver,
+        device: VirtioFsDevice,
+        tmpdir: tempfile::TempDir,
+    ) -> Self {
         let mem = GuestMemory::allocate(TOTAL_MEM_SIZE);
         init_avail_ring(&mem, AVAIL_ADDR);
         init_used_ring(&mem, USED_ADDR);
 
-        let fs = VirtioFs::new(tmpdir.path(), None).unwrap();
-        let driver_source = VmTaskDriverSource::new(SingleDriverBackend::new(driver.clone()));
-        let device = VirtioFsDevice::new(&driver_source, "testfs", fs, 0, None);
-
         let queue_event = Event::new();
         let interrupt_event = Event::new();
 
@@ -277,8 +309,9 @@ fn read_response<T: FromBytes>(&self, resp_gpa: u64) -> T {
         T::read_from_bytes(&buf).unwrap()
     }
 
-    /// Send FUSE_INIT and wait for the response. Panics on failure.
-    async fn fuse_init(&mut self, head_desc: u16) {
+    /// Post FUSE_INIT. Returns `(unique, resp_gpa)`, as
+    /// [`post_fuse_request`](Self::post_fuse_request) does.
+    fn post_fuse_init(&mut self, head_desc: u16) -> (u64, u64) {
         let init_args = fuse_init_in {
             major: FUSE_KERNEL_VERSION,
             minor: FUSE_KERNEL_MINOR_VERSION,
@@ -289,9 +322,12 @@ async fn fuse_init(&mut self, head_desc: u16) {
         };
 
         let resp_size = OUT_HEADER_SIZE + size_of::<fuse_init_out>() as u32;
-        let (unique, resp_gpa) =
-            self.post_fuse_request(head_desc, FUSE_INIT, 0, init_args.as_bytes(), resp_size);
+        self.post_fuse_request(head_desc, FUSE_INIT, 0, init_args.as_bytes(), resp_size)
+    }
 
+    /// Wait for the response to the FUSE_INIT `unique`, whose response buffer
+    /// is at `resp_gpa`. Panics on failure.
+    async fn wait_for_init(&mut self, unique: u64, resp_gpa: u64) {
         let (_used_id, used_len) = self.wait_for_used().await;
         assert!(used_len > 0, "FUSE_INIT response should not be empty");
 
@@ -303,6 +339,12 @@ async fn fuse_init(&mut self, head_desc: u16) {
         assert_eq!(init_out.major, FUSE_KERNEL_VERSION);
     }
 
+    /// Send FUSE_INIT and wait for the response. Panics on failure.
+    async fn fuse_init(&mut self, head_desc: u16) {
+        let (unique, resp_gpa) = self.post_fuse_init(head_desc);
+        self.wait_for_init(unique, resp_gpa).await;
+    }
+
     /// Return the path to the temp directory backing the filesystem.
     fn tmpdir_path(&self) -> &std::path::Path {
         self._tmpdir.path()
@@ -319,6 +361,31 @@ async fn fuse_init_succeeds(driver: DefaultDriver) {
     harness.fuse_init(0).await;
 }
 
+/// A microVM device whose queue starts while host input is quiesced, as a
+/// restore starts it behind its post-restore input gate, leaves a request on
+/// the virtqueue until input resumes, and then serves it.
+#[async_test]
+async fn microvm_queue_started_while_input_is_quiesced_serves_requests_once_it_resumes(
+    driver: DefaultDriver,
+) {
+    let mut harness = TestHarness::new_microvm(&driver);
+    harness.device.quiesce_input().await.unwrap();
+    harness.enable().await;
+
+    let (unique, resp_gpa) = harness.post_fuse_init(0);
+    PolledTimer::new(&driver)
+        .sleep(Duration::from_millis(100))
+        .await;
+    let used_idx: u16 = harness.mem.read_plain(USED_ADDR + 2).unwrap();
+    assert_eq!(
+        used_idx, 0,
+        "the device served a request while input was quiesced"
+    );
+
+    harness.device.resume_input().await.unwrap();
+    harness.wait_for_init(unique, resp_gpa).await;
+}
+
 /// GETATTR on the root inode returns a directory after INIT.
 #[async_test]
 async fn getattr_root_returns_directory(driver: DefaultDriver) {
diff --git a/vm/devices/virtio/virtiofs/src/virtio.rs b/vm/devices/virtio/virtiofs/src/virtio.rs
index 31960ac..a44afb6 100644
--- a/vm/devices/virtio/virtiofs/src/virtio.rs
+++ b/vm/devices/virtio/virtiofs/src/virtio.rs
@@ -17,6 +17,9 @@
 use std::io;
 use std::io::Write;
 use std::sync::Arc;
+use std::task::Context;
+use std::task::Poll;
+use std::task::Waker;
 use task_control::AsyncRun;
 use task_control::Cancelled;
 use task_control::StopTask;
@@ -99,6 +102,8 @@ pub struct VirtioFsDevice {
 struct AdmissionState {
     accepting: bool,
     in_flight: u32,
+    /// The wakers of the queue workers that wait for admission to reopen.
+    waiters: Vec<Waker>,
 }
 
 pub(crate) struct RequestAdmission {
@@ -115,10 +120,12 @@ fn new() -> Self {
             state: Mutex::new(AdmissionState {
                 accepting: true,
                 in_flight: 0,
+                waiters: Vec::new(),
             }),
         }
     }
 
+    #[cfg(test)]
     fn accept(self: &Arc<Self>) -> Option<RequestGuard> {
         let mut state = self.state.lock();
         if !state.accepting {
@@ -130,14 +137,45 @@ fn accept(self: &Arc<Self>) -> Option<RequestGuard> {
         })
     }
 
+    /// Admits one request, or, while admission is closed, registers the
+    /// caller's waker so that [`Self::resume`] wakes it. Returns `None` if the
+    /// in-flight count would overflow.
+    fn poll_accept(self: &Arc<Self>, cx: &mut Context<'_>) -> Poll<Option<RequestGuard>> {
+        let mut state = self.state.lock();
+        if !state.accepting {
+            if !state
+                .waiters
+                .iter()
+                .any(|waker| waker.will_wake(cx.waker()))
+            {
+                state.waiters.push(cx.waker().clone());
+            }
+            return Poll::Pending;
+        }
+        let Some(in_flight) = state.in_flight.checked_add(1) else {
+            return Poll::Ready(None);
+        };
+        state.in_flight = in_flight;
+        Poll::Ready(Some(RequestGuard {
+            admission: Arc::clone(self),
+        }))
+    }
+
     pub(crate) fn quiesce(&self) {
         let mut state = self.state.lock();
         state.accepting = false;
     }
 
+    /// Reopens admission, and wakes the queue workers that wait for it.
     pub(crate) fn resume(&self) {
-        let mut state = self.state.lock();
-        state.accepting = true;
+        let waiters = {
+            let mut state = self.state.lock();
+            state.accepting = true;
+            std::mem::take(&mut state.waiters)
+        };
+        for waker in waiters {
+            waker.wake();
+        }
     }
 
     pub(crate) fn verify_drained(&self) -> anyhow::Result<()> {
@@ -413,7 +451,14 @@ async fn run(
             // Admission must precede queue.next(): after save closes the
             // gate, an unowned descriptor must remain on the virtqueue rather
             // than being dequeued and falsely completed with an empty reply.
-            let Some(request_guard) = self.admission.accept() else {
+            // While the gate is closed, as when a restore starts the queues
+            // behind its post-restore input gate, the worker waits for it to
+            // reopen rather than ending, since nothing would start it again.
+            let admission = &self.admission;
+            let Some(request_guard) = stop
+                .until_stopped(std::future::poll_fn(|cx| admission.poll_accept(cx)))
+                .await?
+            else {
                 break;
             };
             let work = stop.until_stopped(state.queue.next()).await?;
-- 
2.55.0.windows.5