diff --git a/vm/devices/virtio/virtio/src/common.rs b/vm/devices/virtio/virtio/src/common.rs index 311578a4a1..f4af845709 100644 --- a/vm/devices/virtio/virtio/src/common.rs +++ b/vm/devices/virtio/virtio/src/common.rs @@ -318,7 +318,14 @@ impl VirtioQueue { /// new data arrived during arming, it returns immediately without sleeping. /// On wakeup, kicks are suppressed to avoid unnecessary doorbells while /// the caller drains the queue. + /// + /// Returns `Poll::Pending` forever once the queue has failed, since no + /// further work can ever be fetched. This keeps callers that loop on + /// "kick, then retry" from spinning. pub fn poll_kick(&mut self, cx: &mut Context<'_>) -> Poll<()> { + if self.core.failed() { + return Poll::Pending; + } if self.core.arm_for_kick() { ready!(self.queue_event.wait().poll_unpin(cx)).expect("waits on Event cannot fail"); } @@ -424,6 +431,13 @@ impl VirtioQueue { } } +/// Yields each descriptor chain the guest makes available. +/// +/// After yielding a [`QueueError`] the queue is retired, and the stream parks +/// forever rather than ending: returning `None` would be ready synchronously on +/// every poll, so a caller looping over the stream would spin on it. Parking +/// makes it impossible for any caller to busy-loop, and the worker stays +/// cancellable. impl Stream for VirtioQueue { type Item = Result; diff --git a/vm/devices/virtio/virtio/src/queue.rs b/vm/devices/virtio/virtio/src/queue.rs index 8f65909477..0edf7692e3 100644 --- a/vm/devices/virtio/virtio/src/queue.rs +++ b/vm/devices/virtio/virtio/src/queue.rs @@ -203,6 +203,19 @@ pub(crate) struct QueueCoreGetWork { inner: QueueGetWorkInner, /// Whether kick notification is currently armed. armed: bool, + /// Whether a fatal error has retired the fetch side of the queue. + /// + /// Every [`QueueError`] raised while fetching work means the guest violated + /// the queue protocol, and the offending chain is rejected *without* being + /// consumed — the available index does not move. Re-reading the ring would + /// therefore hit the same descriptors and raise the same error forever, so a + /// caller that logs the error and keeps polling would spin without making + /// progress. Latch the failure instead: report it once, then report the + /// queue as permanently empty. + /// + /// This only retires fetching. Descriptors already consumed can still be + /// completed. + failed: bool, /// Consumed-but-not-completed ring capacity, in the format's native unit: /// buffers (heads) for split, descriptors (slots) for packed — the two forms /// of the virtio "Queue Size" limit (spec §2.7.1 vs §2.8.1). @@ -277,10 +290,17 @@ impl QueueCoreGetWork { mem, inner, armed: false, + failed: false, in_flight, }) } + /// Whether a fatal error has retired the fetch side of this queue. Once + /// set, no further work will ever be returned. See [`Self::failed`]. + pub fn failed(&self) -> bool { + self.failed + } + pub fn try_next_work(&mut self) -> Result, QueueError> { match self.try_peek_work() { Ok(Some(work)) => { @@ -296,7 +316,21 @@ impl QueueCoreGetWork { /// consume the peeked descriptor and move to the next one. Calling this /// again without advancing will return the same descriptor, but note that /// the guest may have modified the descriptor memory in the meantime. + /// + /// Returns `Ok(None)` forever once the queue has failed, so that the error + /// is reported exactly once. pub fn try_peek_work(&mut self) -> Result, QueueError> { + if self.failed { + return Ok(None); + } + let r = self.try_peek_work_inner(); + if r.is_err() { + self.failed = true; + } + r + } + + fn try_peek_work_inner(&mut self) -> Result, QueueError> { let index = match &mut self.inner { QueueGetWorkInner::Split(split) => split.is_available()?, QueueGetWorkInner::Packed(packed) => packed.is_available()?, diff --git a/vm/devices/virtio/virtio/src/tests.rs b/vm/devices/virtio/virtio/src/tests.rs index 7c1f1419ff..5bc5a8c486 100644 --- a/vm/devices/virtio/virtio/src/tests.rs +++ b/vm/devices/virtio/virtio/src/tests.rs @@ -5578,12 +5578,15 @@ fn new_bound_test_queue_result( Ok((mem, queue)) } -fn assert_in_flight_error(error: io::Error, in_flight: u16, requested: u16) { - let error = error +fn queue_error(error: &io::Error) -> Option<&QueueError> { + error .get_ref() - .and_then(|error| error.downcast_ref::()); + .and_then(|error| error.downcast_ref::()) +} + +fn assert_in_flight_error(error: io::Error, in_flight: u16, requested: u16) { assert!(matches!( - error, + queue_error(&error), Some(QueueError::TooManyInFlightDescriptors { in_flight: actual_in_flight, requested: actual_requested, @@ -5592,6 +5595,31 @@ fn assert_in_flight_error(error: io::Error, in_flight: u16, requested: u16) { )); } +/// Points descriptor 0 at itself and makes it available. The chain never +/// terminates, so the queue rejects it with [`QueueError::TooLong`]. +fn make_split_cyclic_chain(mem: &GuestMemory) { + use crate::test_helpers::make_available; + use crate::test_helpers::write_descriptor; + + write_descriptor( + mem, + BOUND_TEST_DESC_ADDR, + 0, + BOUND_TEST_DATA_ADDR, + 0x100, + DescriptorFlags::new().with_write(true).with_next(true), + 0, + ); + let mut avail_idx = 0; + make_available( + mem, + BOUND_TEST_AVAIL_ADDR, + BOUND_TEST_QUEUE_SIZE, + 0, + &mut avail_idx, + ); +} + fn next_error(queue: &mut VirtioQueue) -> io::Error { match queue.try_next() { Err(error) => error, @@ -5704,8 +5732,12 @@ async fn packed_queue_rejects_chain_exceeding_free_space(driver: DefaultDriver) assert_in_flight_error(next_error(&mut queue), 2, 3); } +/// Oversubscribing the ring retires the queue like any other protocol +/// violation. Completing outstanding work frees ring capacity but does not +/// revive it: the rejected chain was never consumed, so the queue would hand +/// out a descriptor the guest had already oversubscribed. #[async_test] -async fn split_queue_capacity_recovers_after_completion(driver: DefaultDriver) { +async fn split_queue_stays_retired_after_capacity_error(driver: DefaultDriver) { let (mem, mut queue) = new_bound_test_queue(&driver, false, None); make_split_bound_test_work(&mem, BOUND_TEST_QUEUE_SIZE + 1, 0); @@ -5714,8 +5746,14 @@ async fn split_queue_capacity_recovers_after_completion(driver: DefaultDriver) { works.push(queue.try_next().unwrap().expect("within queue size")); } assert_in_flight_error(next_error(&mut queue), BOUND_TEST_QUEUE_SIZE, 1); + + // Completions still publish to the used ring, so a device can drain its + // in-flight work after the failure. queue.complete(works.remove(0), 1); - let _ = queue.try_next().unwrap().unwrap(); + assert!( + matches!(queue.try_next(), Ok(None)), + "freeing capacity must not revive a retired queue" + ); } #[async_test] @@ -5788,3 +5826,50 @@ async fn queue_new_rejects_corrupt_saved_state(driver: DefaultDriver) { )); } } + +/// A rejected descriptor chain is never consumed, so the available index does +/// not move and the same chain would be re-read on the next call. Report the +/// failure once and then report the queue as empty, so that a device worker +/// which logs the error and keeps polling makes no further progress instead of +/// spinning on the same bad chain forever. +#[async_test] +async fn queue_reports_failure_only_once(driver: DefaultDriver) { + let (mem, mut queue) = new_bound_test_queue(&driver, false, None); + make_split_cyclic_chain(&mem); + + let Err(error) = queue.try_next() else { + panic!("a cyclic descriptor chain must be rejected"); + }; + assert!(matches!(queue_error(&error), Some(QueueError::TooLong))); + + assert!( + matches!(queue.try_next(), Ok(None)), + "failed queue must report no work rather than repeat the error" + ); + assert!( + matches!(queue.try_peek(), Ok(None)), + "failed queue must report no work rather than repeat the error" + ); +} + +/// Driven as a stream, a failed queue must park rather than yield the same +/// error on every poll. It must not end the stream either: `None` would be +/// ready synchronously on every poll, so a caller looping over the stream +/// would spin on it exactly as it would on the repeated error. +#[async_test] +async fn queue_stream_parks_after_failure(driver: DefaultDriver) { + let (mem, mut queue) = new_bound_test_queue(&driver, false, None); + make_split_cyclic_chain(&mem); + + let Some(Err(error)) = queue.next().await else { + panic!("the stream must yield the failure"); + }; + assert!(matches!(queue_error(&error), Some(QueueError::TooLong))); + + assert!( + poll_fn(|cx| std::task::Poll::Ready(queue.poll_next_unpin(cx))) + .await + .is_pending(), + "failed queue must park rather than repeat the error or end the stream" + ); +} diff --git a/vm/devices/virtio/virtio_blk/src/integration_tests.rs b/vm/devices/virtio/virtio_blk/src/integration_tests.rs index 1e2ae319a6..2efb6fa15f 100644 --- a/vm/devices/virtio/virtio_blk/src/integration_tests.rs +++ b/vm/devices/virtio/virtio_blk/src/integration_tests.rs @@ -11,15 +11,22 @@ use crate::VirtioBlkDevice; use disk_backend::Disk; use disk_backend::DiskError; use disk_backend::DiskIo; +use futures::future::Either; +use futures::future::select; use guestmem::GuestMemory; use guestmem::MemoryRead; use guestmem::MemoryWrite; use inspect::Inspect; use pal_async::DefaultDriver; +use pal_async::DefaultPool; use pal_async::async_test; +use pal_async::timer::PolledTimer; use pal_event::Event; use parking_lot::Mutex; use scsi_buffers::RequestBuffers; +use std::future::Future; +use std::pin::pin; +use std::time::Duration; use test_with_tracing::test; use virtio::QueueResources; use virtio::VirtioDevice; @@ -70,12 +77,26 @@ struct TestHarness { impl TestHarness { /// Create a harness with a RAM disk of the given size. fn new(driver: &DefaultDriver, disk: Disk, read_only: bool) -> Self { + Self::with_device_driver(driver, driver, disk, read_only) + } + + /// Like [`TestHarness::new`], but runs the device's worker task on + /// `device_driver` rather than on the test's own executor. A test that + /// must stay responsive while the worker misbehaves puts the device on a + /// separate thread's executor. + fn with_device_driver( + driver: &DefaultDriver, + device_driver: &DefaultDriver, + disk: Disk, + read_only: bool, + ) -> Self { let mem = GuestMemory::allocate(TOTAL_MEM_SIZE); init_avail_ring(&mem, AVAIL_ADDR); init_used_ring(&mem, USED_ADDR); - let driver_source = VmTaskDriverSource::new(SingleDriverBackend::new(driver.clone())); + let driver_source = + VmTaskDriverSource::new(SingleDriverBackend::new(device_driver.clone())); let device = VirtioBlkDevice::new(&driver_source, disk, read_only); let queue_event = Event::new(); @@ -510,6 +531,20 @@ fn ram_disk(size: u64, read_only: bool) -> Disk { disklayer_ram::ram_disk(size, read_only).unwrap() } +/// Awaits `fut`, panicking with `msg` if it does not complete within `timeout`. +async fn with_timeout( + driver: &DefaultDriver, + timeout: Duration, + msg: &str, + fut: F, +) -> F::Output { + let mut timer = PolledTimer::new(driver); + match select(pin!(fut), pin!(timer.sleep(timeout))).await { + Either::Left((output, _)) => output, + Either::Right(_) => panic!("{msg}"), + } +} + /// Write 1 sector then read it back. Verifies basic write and read roundtrip. #[async_test] async fn write_then_read_roundtrip(driver: DefaultDriver) { @@ -1284,3 +1319,62 @@ async fn bounce_buffer_write_read_roundtrip(driver: DefaultDriver) { .unwrap(); assert_eq!(read_status[0], VIRTIO_BLK_S_OK, "read should succeed"); } + +/// A descriptor whose `next` link points back at itself forms a chain that +/// never terminates, and the queue rejects it without consuming it — the +/// available index does not move, so the same chain is still there on the next +/// poll. A worker that logs the error and keeps polling therefore spins inside +/// a single `poll` call, never returning `Pending`. That burns a CPU forever +/// and, because the cancel future is only polled once the work future returns +/// `Pending`, leaves the worker task impossible to stop — wedging device +/// teardown, reset, save/restore, and inspect. +/// +/// The device runs on its own executor thread so that a spinning worker wedges +/// only that thread, letting this test fail on a timeout instead of hanging. +#[async_test] +async fn cyclic_descriptor_chain_does_not_wedge_worker(driver: DefaultDriver) { + let (_device_thread, device_driver) = DefaultPool::spawn_on_thread("virtio-blk-device"); + let disk = ram_disk(64 * 1024, false); + let mut harness = TestHarness::with_device_driver(&driver, &device_driver, disk, false); + harness.enable().await; + + // Run one valid request first, so the worker is known to be up and parked + // waiting for a kick by the time the bad chain arrives. + harness.post_flush_request(0); + let (used_id, _) = harness.wait_for_used().await; + assert_eq!(used_id, 0); + + // Descriptor 4 chains to itself. + let gpa = harness.alloc_data(REQ_HEADER_SIZE); + write_descriptor( + &harness.mem, + DESC_ADDR, + 4, + gpa, + REQ_HEADER_SIZE, + DescriptorFlags::new().with_next(true), + 4, + ); + make_available( + &harness.mem, + AVAIL_ADDR, + QUEUE_SIZE, + 4, + &mut harness.avail_idx, + ); + harness.queue_event.signal(); + + // Give the worker time to wake on the kick and reject the chain. + PolledTimer::new(&driver) + .sleep(Duration::from_millis(250)) + .await; + + // A worker spinning on the bad chain never observes the stop request. + with_timeout( + &driver, + Duration::from_secs(5), + "virtio-blk worker could not be stopped after an invalid descriptor chain", + harness.device.stop_queue(0), + ) + .await; +} diff --git a/vm/devices/virtio/virtio_blk/src/lib.rs b/vm/devices/virtio/virtio_blk/src/lib.rs index 3adf2f67f7..391c045080 100644 --- a/vm/devices/virtio/virtio_blk/src/lib.rs +++ b/vm/devices/virtio/virtio_blk/src/lib.rs @@ -169,7 +169,13 @@ impl AsyncRun for BlkWorker { state: &mut BlkQueueState, ) -> Result<(), task_control::Cancelled> { stop.until_stopped(async { - loop { + // Set once the queue can no longer produce work, because the guest + // violated the queue protocol. Fetching stops, but in-flight IOs + // keep draining so their descriptors are still completed; the + // worker exits once they are done. + let mut queue_done = false; + + while !(queue_done && self.ios.is_empty()) { enum Event { NewWork(Result), Completed(IoCompletion), @@ -181,7 +187,7 @@ impl AsyncRun for BlkWorker { return Poll::Ready(Event::Completed(completion)); } // Accept new work if under the depth limit. - if self.ios.len() < MAX_IO_DEPTH { + if !queue_done && self.ios.len() < MAX_IO_DEPTH { if let Poll::Ready(item) = state.queue.poll_next_unpin(cx) { let item = item.expect("virtio queue stream never ends"); return Poll::Ready(Event::NewWork(item)); @@ -201,10 +207,13 @@ impl AsyncRun for BlkWorker { })); } Event::NewWork(Err(err)) => { + // The queue is retired: the rejected chain is never + // consumed, so retrying would fail identically forever. tracelimit::error_ratelimited!( error = &err as &dyn std::error::Error, - "error reading from virtio queue" + "error reading from virtio queue, stopping worker" ); + queue_done = true; } Event::Completed(completion) => { self.finish_io(&mut state.queue, completion); diff --git a/vm/devices/virtio/virtio_vsock/src/lib.rs b/vm/devices/virtio/virtio_vsock/src/lib.rs index f8e311f8b8..f416f98f43 100644 --- a/vm/devices/virtio/virtio_vsock/src/lib.rs +++ b/vm/devices/virtio/virtio_vsock/src/lib.rs @@ -479,7 +479,7 @@ impl AsyncRun for VsockWorker { error = &err as &dyn std::error::Error, "error peeking virtio rx queue" ); - return false; + return; } }; @@ -502,10 +502,16 @@ impl AsyncRun for VsockWorker { r = state.tx_queue.select_next_some() => { match r { Ok(work) => self.handle_guest_tx(state, work), - Err(err) => tracing::error!( - error = &err as &dyn std::error::Error, - "error reading from virtio tx queue" - ), + Err(err) => { + // The queue is retired: the rejected chain is + // never consumed, so retrying would fail + // identically forever. + tracing::error!( + error = &err as &dyn std::error::Error, + "error reading from virtio tx queue, stopping worker" + ); + return; + } } } r = rx_ready => { @@ -541,8 +547,7 @@ impl AsyncRun for VsockWorker { }; } }) - .await?; - Ok(()) + .await } }