From 7f5f9608ade601d46c1664847d0b41122d64ae00 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Daniel=20Gra=C3=B1a?= Date: Thu, 24 Sep 2026 11:21:34 -0300 Subject: [PATCH 1/3] vmm: add a Threaded block IO engine Every virtio device is serviced by the single VMM event loop thread. With io_engine=Sync the block device does its reads, writes and fsyncs inline on that thread, so while the storage behind a drive is slow nothing else on the event loop runs: virtio-net stops being serviced and guest network latency climbs to the disk's latency. io_engine=Async avoids that but needs io_uring. Add io_engine=Threaded: the same blocking IO as Sync, performed by a worker thread per drive, with completions coming back through an eventfd the way the Async engine does. Sync and Async are left as they are. - The worker performs requests one at a time, in submission order, using the Sync engine's own read/write/flush on a clone of the backing file, so flushes still cover the writes submitted before them. Reads mark the guest pages they fill as dirty. - Completion signals are coalesced while more requests are queued, for up to 200us after the previous signal, so a burst of completions raises one guest interrupt; a slow backing file still gets one per request. - At most 128 requests are in flight; beyond that the engine reports itself throttled and the device resumes the queue on completion. - drain()/drain_and_flush() wait for the worker, so snapshots and device teardown see every request finished. A drive update goes through the same queue: earlier requests use the old file, later ones the new one. - The engine type is saved in snapshots like the other two. - The worker is started when the drive is created, before the VMM thread installs its seccomp filter, so it installs its own: a new, optional "block_io" thread category. Filter files without it fall back to the "vmm" filter. The engine lives in io/threaded_io.rs and its device-side completion handling in threaded.rs; the changes to existing files are limited to the engine type enums, their match arms and the event registration. --- .../seccomp/aarch64-unknown-linux-musl.json | 163 +++++ resources/seccomp/unimplemented.json | 5 + .../seccomp/x86_64-unknown-linux-musl.json | 163 +++++ src/firecracker/src/main.rs | 8 + src/firecracker/src/seccomp.rs | 29 +- src/firecracker/swagger/firecracker.yaml | 5 +- .../src/devices/virtio/block/virtio/device.rs | 6 +- .../virtio/block/virtio/event_handler.rs | 11 + .../src/devices/virtio/block/virtio/io/mod.rs | 40 ++ .../virtio/block/virtio/io/threaded_io.rs | 647 ++++++++++++++++++ .../src/devices/virtio/block/virtio/mod.rs | 3 + .../devices/virtio/block/virtio/persist.rs | 4 + .../devices/virtio/block/virtio/test_utils.rs | 4 + .../devices/virtio/block/virtio/threaded.rs | 330 +++++++++ 14 files changed, 1412 insertions(+), 6 deletions(-) create mode 100644 src/vmm/src/devices/virtio/block/virtio/io/threaded_io.rs create mode 100644 src/vmm/src/devices/virtio/block/virtio/threaded.rs diff --git a/resources/seccomp/aarch64-unknown-linux-musl.json b/resources/seccomp/aarch64-unknown-linux-musl.json index 26dd661e46b..0e105b0d9c9 100644 --- a/resources/seccomp/aarch64-unknown-linux-musl.json +++ b/resources/seccomp/aarch64-unknown-linux-musl.json @@ -1094,5 +1094,168 @@ ] } ] + }, + "block_io": { + "default_action": "trap", + "filter_action": "allow", + "filter": [ + { + "syscall": "exit" + }, + { + "syscall": "exit_group" + }, + { + "syscall": "read", + "comment": "Reads from the backing file" + }, + { + "syscall": "write", + "comment": "Writes to the backing file, and signals completions through an eventfd" + }, + { + "syscall": "fsync", + "comment": "Flush requests" + }, + { + "syscall": "close", + "comment": "Closes the previous backing file after a drive update" + }, + { + "syscall": "brk", + "comment": "Called for expanding the heap" + }, + { + "syscall": "clock_gettime", + "comment": "Used for metrics and logging, via the helpers in utils/src/time.rs. It's not called on some platforms, because of vdso optimisations." + }, + { + "syscall": "lseek", + "comment": "Positions the backing file before each read or write" + }, + { + "syscall": "mremap", + "comment": "Used for re-allocating large memory regions, for example vectors" + }, + { + "syscall": "munmap", + "comment": "Used for freeing memory" + }, + { + "syscall": "rt_sigprocmask", + "comment": "rt_sigprocmask is used by libc::abort during a panic to block and unblock signals" + }, + { + "syscall": "rt_sigreturn", + "comment": "rt_sigreturn is needed in case a fault does occur, so that the signal handler can return. Otherwise we get stuck in a fault loop." + }, + { + "syscall": "sigaltstack", + "comment": "sigaltstack is used by Rust stdlib to remove alternative signal stack during thread teardown." + }, + { + "syscall": "futex", + "comment": "Used for synchronization (during thread teardown when joining multiple vcpu threads at once)", + "args": [ + { + "index": 1, + "type": "dword", + "op": "eq", + "val": 0, + "comment": "FUTEX_WAIT" + } + ] + }, + { + "syscall": "futex", + "comment": "Used for synchronization (during thread teardown)", + "args": [ + { + "index": 1, + "type": "dword", + "op": "eq", + "val": 1, + "comment": "FUTEX_WAKE" + } + ] + }, + { + "syscall": "futex", + "comment": "Used for synchronization", + "args": [ + { + "index": 1, + "type": "dword", + "op": "eq", + "val": 128, + "comment": "FUTEX_WAIT_PRIVATE" + } + ] + }, + { + "syscall": "futex", + "comment": "Used for synchronization", + "args": [ + { + "index": 1, + "type": "dword", + "op": "eq", + "val": 137, + "comment": "FUTEX_WAIT_BITSET_PRIVATE" + } + ] + }, + { + "syscall": "futex", + "comment": "Used for synchronization", + "args": [ + { + "index": 1, + "type": "dword", + "op": "eq", + "val": 129, + "comment": "FUTEX_WAKE_PRIVATE" + } + ] + }, + { + "syscall": "madvise", + "comment": "Used by the VirtIO balloon device and by musl for some customer workloads. It is also used by aws-lc during random number generation. They setup a memory page that mark with MADV_WIPEONFORK to be able to detect forks. They also call it with -1 to see if madvise is supported in certain platforms." + }, + { + "syscall": "mmap", + "comment": "Used by the allocator", + "args": [ + { + "index": 3, + "type": "dword", + "op": "eq", + "val": 34, + "comment": "libc::MAP_ANONYMOUS | libc::MAP_PRIVATE" + } + ] + }, + { + "syscall": "tkill", + "comment": "tkill is used by libc::abort during a panic to raise SIGABRT", + "args": [ + { + "index": 1, + "type": "dword", + "op": "eq", + "val": 6, + "comment": "SIGABRT" + } + ] + }, + { + "syscall": "sched_yield", + "comment": "Used by the rust standard library in std::sync::mpmc. Firecracker uses mpsc channels from this module for inter-thread communication" + }, + { + "syscall": "restart_syscall", + "comment": "automatically issued by the kernel when specific timing-related syscalls (e.g. nanosleep) get interrupted by SIGSTOP" + } + ] } } diff --git a/resources/seccomp/unimplemented.json b/resources/seccomp/unimplemented.json index a919df15519..f733949931f 100644 --- a/resources/seccomp/unimplemented.json +++ b/resources/seccomp/unimplemented.json @@ -13,5 +13,10 @@ "default_action": "allow", "filter_action": "trap", "filter": [] + }, + "block_io": { + "default_action": "allow", + "filter_action": "trap", + "filter": [] } } diff --git a/resources/seccomp/x86_64-unknown-linux-musl.json b/resources/seccomp/x86_64-unknown-linux-musl.json index dcd6753a4c5..c742f6e6f0c 100644 --- a/resources/seccomp/x86_64-unknown-linux-musl.json +++ b/resources/seccomp/x86_64-unknown-linux-musl.json @@ -1226,5 +1226,168 @@ ] } ] + }, + "block_io": { + "default_action": "trap", + "filter_action": "allow", + "filter": [ + { + "syscall": "exit" + }, + { + "syscall": "exit_group" + }, + { + "syscall": "read", + "comment": "Reads from the backing file" + }, + { + "syscall": "write", + "comment": "Writes to the backing file, and signals completions through an eventfd" + }, + { + "syscall": "fsync", + "comment": "Flush requests" + }, + { + "syscall": "close", + "comment": "Closes the previous backing file after a drive update" + }, + { + "syscall": "brk", + "comment": "Called for expanding the heap" + }, + { + "syscall": "clock_gettime", + "comment": "Used for metrics and logging, via the helpers in utils/src/time.rs. It's not called on some platforms, because of vdso optimisations." + }, + { + "syscall": "lseek", + "comment": "Positions the backing file before each read or write" + }, + { + "syscall": "mremap", + "comment": "Used for re-allocating large memory regions, for example vectors" + }, + { + "syscall": "munmap", + "comment": "Used for freeing memory" + }, + { + "syscall": "rt_sigprocmask", + "comment": "rt_sigprocmask is used by libc::abort during a panic to block and unblock signals" + }, + { + "syscall": "rt_sigreturn", + "comment": "rt_sigreturn is needed in case a fault does occur, so that the signal handler can return. Otherwise we get stuck in a fault loop." + }, + { + "syscall": "sigaltstack", + "comment": "sigaltstack is used by Rust stdlib to remove alternative signal stack during thread teardown." + }, + { + "syscall": "futex", + "comment": "Used for synchronization (during thread teardown when joining multiple vcpu threads at once)", + "args": [ + { + "index": 1, + "type": "dword", + "op": "eq", + "val": 0, + "comment": "FUTEX_WAIT" + } + ] + }, + { + "syscall": "futex", + "comment": "Used for synchronization (during thread teardown)", + "args": [ + { + "index": 1, + "type": "dword", + "op": "eq", + "val": 1, + "comment": "FUTEX_WAKE" + } + ] + }, + { + "syscall": "futex", + "comment": "Used for synchronization", + "args": [ + { + "index": 1, + "type": "dword", + "op": "eq", + "val": 128, + "comment": "FUTEX_WAIT_PRIVATE" + } + ] + }, + { + "syscall": "futex", + "comment": "Used for synchronization", + "args": [ + { + "index": 1, + "type": "dword", + "op": "eq", + "val": 137, + "comment": "FUTEX_WAIT_BITSET_PRIVATE" + } + ] + }, + { + "syscall": "futex", + "comment": "Used for synchronization", + "args": [ + { + "index": 1, + "type": "dword", + "op": "eq", + "val": 129, + "comment": "FUTEX_WAKE_PRIVATE" + } + ] + }, + { + "syscall": "madvise", + "comment": "Used by the VirtIO balloon device and by musl for some customer workloads. It is also used by aws-lc during random number generation. They setup a memory page that mark with MADV_WIPEONFORK to be able to detect forks. They also call it with -1 to see if madvise is supported in certain platforms." + }, + { + "syscall": "mmap", + "comment": "Used by the allocator", + "args": [ + { + "index": 3, + "type": "dword", + "op": "eq", + "val": 34, + "comment": "libc::MAP_ANONYMOUS | libc::MAP_PRIVATE" + } + ] + }, + { + "syscall": "tkill", + "comment": "tkill is used by libc::abort during a panic to raise SIGABRT", + "args": [ + { + "index": 1, + "type": "dword", + "op": "eq", + "val": 6, + "comment": "SIGABRT" + } + ] + }, + { + "syscall": "sched_yield", + "comment": "Used by the rust standard library in std::sync::mpmc. Firecracker uses mpsc channels from this module for inter-thread communication" + }, + { + "syscall": "restart_syscall", + "comment": "automatically issued by the kernel when specific timing-related syscalls (e.g. nanosleep) get interrupted by SIGSTOP" + } + ] } } diff --git a/src/firecracker/src/main.rs b/src/firecracker/src/main.rs index 739214999a4..b02c4d22f00 100644 --- a/src/firecracker/src/main.rs +++ b/src/firecracker/src/main.rs @@ -363,6 +363,14 @@ fn main_exec() -> Result<(), MainError> { .and_then(seccomp::get_filters) .map_err(MainError::SeccompFilter)?; + // Threaded block IO engine workers install their own filter. + if let Some(filter) = seccomp_filters + .get("block_io") + .or_else(|| seccomp_filters.get("vmm")) + { + vmm::devices::virtio::block::virtio::set_worker_seccomp_filter(filter.clone()); + } + let vmm_config_json = arguments .single_value("config-file") .map(fs::read_to_string) diff --git a/src/firecracker/src/seccomp.rs b/src/firecracker/src/seccomp.rs index 421220a7b5f..f1095106a8c 100644 --- a/src/firecracker/src/seccomp.rs +++ b/src/firecracker/src/seccomp.rs @@ -8,6 +8,9 @@ use std::path::Path; use vmm::seccomp::{BpfThreadMap, DeserializationError, deserialize_binary, get_empty_filters}; const THREAD_CATEGORIES: [&str; 3] = ["vmm", "api", "vcpu"]; +/// Thread categories a filter file may leave out. A missing "block_io" filter, for the threaded +/// block IO engine's workers, falls back to the "vmm" one, which allows the same IO. +const OPTIONAL_THREAD_CATEGORIES: [&str; 1] = ["block_io"]; /// Error retrieving seccomp filters. #[derive(Debug, thiserror::Error, displaydoc::Display)] @@ -78,9 +81,11 @@ fn get_custom_filters(reader: R) -> Result Result { - let (filters, invalid_filters): (BpfThreadMap, BpfThreadMap) = map - .into_iter() - .partition(|(k, _)| THREAD_CATEGORIES.contains(&k.as_str())); + let (filters, invalid_filters): (BpfThreadMap, BpfThreadMap) = + map.into_iter().partition(|(k, _)| { + THREAD_CATEGORIES.contains(&k.as_str()) + || OPTIONAL_THREAD_CATEGORIES.contains(&k.as_str()) + }); if !invalid_filters.is_empty() { // build the error message let mut thread_categories_string = @@ -143,6 +148,15 @@ mod tests { assert_eq!(filter_thread_categories(map).unwrap().len(), 3); + // correct categories, including the optional ones + let mut map = BpfThreadMap::new(); + map.insert("vcpu".to_string(), Arc::new(vec![])); + map.insert("vmm".to_string(), Arc::new(vec![])); + map.insert("api".to_string(), Arc::new(vec![])); + map.insert("block_io".to_string(), Arc::new(vec![])); + + assert_eq!(filter_thread_categories(map).unwrap().len(), 4); + // invalid categories let mut map = BpfThreadMap::new(); map.insert("vcpu".to_string(), Arc::new(vec![])); @@ -168,6 +182,15 @@ mod tests { } } + #[test] + fn test_default_filters() { + // Debug builds compile an empty policy, so only release builds check the real one. + let filters = get_default_filters().unwrap(); + for category in THREAD_CATEGORIES.iter().chain(&OPTIONAL_THREAD_CATEGORIES) { + assert!(filters.contains_key(*category), "missing {category}"); + } + } + #[test] fn test_seccomp_config() { assert!(matches!( diff --git a/src/firecracker/swagger/firecracker.yaml b/src/firecracker/swagger/firecracker.yaml index 828bb086198..663ac236c4e 100644 --- a/src/firecracker/swagger/firecracker.yaml +++ b/src/firecracker/swagger/firecracker.yaml @@ -1220,9 +1220,10 @@ definitions: type: string description: Type of the IO engine used by the device. "Async" is supported on - host kernels newer than 5.10.51. + host kernels newer than 5.10.51. "Threaded" does blocking IO like + "Sync", but from a worker thread per drive. This field is optional for virtio-block config and should be omitted for vhost-user-block configuration. - enum: ["Sync", "Async"] + enum: ["Sync", "Async", "Threaded"] default: "Sync" # VhostUserBlock specific parameters diff --git a/src/vmm/src/devices/virtio/block/virtio/device.rs b/src/vmm/src/devices/virtio/block/virtio/device.rs index ecdd8ee4f6d..2d3c14b653a 100644 --- a/src/vmm/src/devices/virtio/block/virtio/device.rs +++ b/src/vmm/src/devices/virtio/block/virtio/device.rs @@ -50,6 +50,8 @@ pub enum FileEngineType { /// Use a Sync engine, based on blocking system calls. #[default] Sync, + /// Use a Threaded engine: blocking system calls, made from a worker thread per drive. + Threaded, } /// Helper object for setting up all `Block` fields derived from its backing file. @@ -272,7 +274,7 @@ macro_rules! unwrap_async_file_engine_or_return { ($file_engine: expr) => { match $file_engine { FileEngine::Async(engine) => engine, - FileEngine::Sync(_) => { + _ => { error!("The block device doesn't use an async IO engine"); return; } @@ -560,6 +562,7 @@ impl VirtioBlock { match self.disk.file_engine { FileEngine::Sync(_) => FileEngineType::Sync, FileEngine::Async(_) => FileEngineType::Async, + FileEngine::Threaded(_) => FileEngineType::Threaded, } } @@ -579,6 +582,7 @@ impl VirtioBlock { if let FileEngine::Async(ref _engine) = self.disk.file_engine { self.process_async_completion_queue(); } + self.process_threaded_completion_queue(); } } diff --git a/src/vmm/src/devices/virtio/block/virtio/event_handler.rs b/src/vmm/src/devices/virtio/block/virtio/event_handler.rs index 9f02862f814..06f2d982473 100644 --- a/src/vmm/src/devices/virtio/block/virtio/event_handler.rs +++ b/src/vmm/src/devices/virtio/block/virtio/event_handler.rs @@ -13,6 +13,7 @@ impl VirtioBlock { const PROCESS_QUEUE: u32 = 1; const PROCESS_RATE_LIMITER: u32 = 2; const PROCESS_ASYNC_COMPLETION: u32 = 3; + const PROCESS_THREADED_COMPLETION: u32 = 4; fn register_runtime_events(&self, ops: &mut EventOps) { if let Err(err) = ops.add(Events::with_data( @@ -38,6 +39,15 @@ impl VirtioBlock { { error!("Failed to register IO engine completion event: {}", err); } + if let FileEngine::Threaded(ref engine) = self.disk.file_engine + && let Err(err) = ops.add(Events::with_data( + engine.completion_evt(), + Self::PROCESS_THREADED_COMPLETION, + EventSet::IN, + )) + { + error!("Failed to register IO engine completion event: {}", err); + } } fn register_activate_event(&self, ops: &mut EventOps) { @@ -88,6 +98,7 @@ impl MutEventSubscriber for VirtioBlock { Self::PROCESS_QUEUE => self.process_queue_event(), Self::PROCESS_RATE_LIMITER => self.process_rate_limiter_event(), Self::PROCESS_ASYNC_COMPLETION => self.process_async_completion_event(), + Self::PROCESS_THREADED_COMPLETION => self.process_threaded_completion_event(), _ => warn!("Block: Spurious event received: {:?}", source), } } else { diff --git a/src/vmm/src/devices/virtio/block/virtio/io/mod.rs b/src/vmm/src/devices/virtio/block/virtio/io/mod.rs index b7aa8061d76..50ad2e7295c 100644 --- a/src/vmm/src/devices/virtio/block/virtio/io/mod.rs +++ b/src/vmm/src/devices/virtio/block/virtio/io/mod.rs @@ -3,12 +3,14 @@ pub mod async_io; pub mod sync_io; +pub mod threaded_io; use std::fmt::Debug; use std::fs::File; pub use self::async_io::{AsyncFileEngine, AsyncIoError}; pub use self::sync_io::{SyncFileEngine, SyncIoError}; +pub use self::threaded_io::{ThreadedFileEngine, ThreadedIoError}; use crate::devices::virtio::block::virtio::PendingRequest; use crate::devices::virtio::block::virtio::device::FileEngineType; use crate::vstate::memory::{GuestAddress, GuestMemoryMmap}; @@ -31,12 +33,15 @@ pub enum BlockIoError { Sync(SyncIoError), /// Async error: {0} Async(AsyncIoError), + /// Threaded error: {0} + Threaded(ThreadedIoError), } impl BlockIoError { pub fn is_throttling_err(&self) -> bool { match self { BlockIoError::Async(AsyncIoError::IoUring(err)) => err.is_throttling_err(), + BlockIoError::Threaded(ThreadedIoError::QueueFull) => true, _ => false, } } @@ -54,6 +59,7 @@ pub enum FileEngine { #[allow(unused)] Async(AsyncFileEngine), Sync(SyncFileEngine), + Threaded(ThreadedFileEngine), } impl FileEngine { @@ -63,6 +69,9 @@ impl FileEngine { AsyncFileEngine::from_file(file).map_err(BlockIoError::Async)?, )), FileEngineType::Sync => Ok(FileEngine::Sync(SyncFileEngine::from_file(file))), + FileEngineType::Threaded => Ok(FileEngine::Threaded( + ThreadedFileEngine::from_file(file).map_err(BlockIoError::Threaded)?, + )), } } @@ -70,6 +79,9 @@ impl FileEngine { match self { FileEngine::Async(engine) => engine.update_file(file).map_err(BlockIoError::Async)?, FileEngine::Sync(engine) => engine.update_file(file), + FileEngine::Threaded(engine) => { + engine.update_file(file).map_err(BlockIoError::Threaded)? + } }; Ok(()) @@ -80,6 +92,7 @@ impl FileEngine { match self { FileEngine::Async(engine) => engine.file(), FileEngine::Sync(engine) => engine.file(), + FileEngine::Threaded(engine) => engine.file(), } } @@ -99,6 +112,13 @@ impl FileEngine { error: BlockIoError::Async(err.error), }), }, + FileEngine::Threaded(engine) => match engine.push_read(offset, mem, addr, count, req) { + Ok(_) => Ok(FileEngineOk::Submitted), + Err(err) => Err(RequestError { + req: err.req, + error: BlockIoError::Threaded(err.error), + }), + }, FileEngine::Sync(engine) => match engine.read(offset, mem, addr, count) { Ok(count) => Ok(FileEngineOk::Executed(RequestOk { req, count })), Err(err) => Err(RequestError { @@ -125,6 +145,15 @@ impl FileEngine { error: BlockIoError::Async(err.error), }), }, + FileEngine::Threaded(engine) => { + match engine.push_write(offset, mem, addr, count, req) { + Ok(_) => Ok(FileEngineOk::Submitted), + Err(err) => Err(RequestError { + req: err.req, + error: BlockIoError::Threaded(err.error), + }), + } + } FileEngine::Sync(engine) => match engine.write(offset, mem, addr, count) { Ok(count) => Ok(FileEngineOk::Executed(RequestOk { req, count })), Err(err) => Err(RequestError { @@ -147,6 +176,13 @@ impl FileEngine { error: BlockIoError::Async(err.error), }), }, + FileEngine::Threaded(engine) => match engine.push_flush(req) { + Ok(_) => Ok(FileEngineOk::Submitted), + Err(err) => Err(RequestError { + req: err.req, + error: BlockIoError::Threaded(err.error), + }), + }, FileEngine::Sync(engine) => match engine.flush() { Ok(_) => Ok(FileEngineOk::Executed(RequestOk { req, count: 0 })), Err(err) => Err(RequestError { @@ -161,6 +197,7 @@ impl FileEngine { match self { FileEngine::Async(engine) => engine.drain(discard).map_err(BlockIoError::Async), FileEngine::Sync(_engine) => Ok(()), + FileEngine::Threaded(engine) => engine.drain(discard).map_err(BlockIoError::Threaded), } } @@ -170,6 +207,9 @@ impl FileEngine { engine.drain_and_flush(discard).map_err(BlockIoError::Async) } FileEngine::Sync(engine) => engine.flush().map_err(BlockIoError::Sync), + FileEngine::Threaded(engine) => engine + .drain_and_flush(discard) + .map_err(BlockIoError::Threaded), } } } diff --git a/src/vmm/src/devices/virtio/block/virtio/io/threaded_io.rs b/src/vmm/src/devices/virtio/block/virtio/io/threaded_io.rs new file mode 100644 index 00000000000..f8d4061026c --- /dev/null +++ b/src/vmm/src/devices/virtio/block/virtio/io/threaded_io.rs @@ -0,0 +1,647 @@ +// Copyright 2026 Fly.io, Inc. +// SPDX-License-Identifier: Apache-2.0 + +//! Threaded file engine. +//! +//! Blocking I/O, like the sync engine, but not on the thread that submits it: each engine owns a +//! worker thread that performs the requests one at a time, in submission order, and reports +//! completions through an eventfd, the way the async engine does. A slow backing file then stalls +//! only its own worker, not the event loop and every other device serviced there, and no +//! io_uring support is needed. + +use std::fs::File; +use std::sync::{Arc, OnceLock, mpsc}; +use std::thread; +use std::time::{Duration, Instant}; + +use vmm_sys_util::eventfd::EventFd; + +use super::sync_io::{SyncFileEngine, SyncIoError}; +use crate::devices::virtio::block::virtio::PendingRequest; +use crate::devices::virtio::block::virtio::io::RequestError; +use crate::logger::error; +use crate::seccomp::{BpfProgram, BpfProgramRef}; +use crate::vstate::memory::{GuestAddress, GuestMemoryExtension, GuestMemoryMmap}; + +/// Maximum number of requests submitted to the worker and not yet popped. The engine reports +/// itself as throttled beyond this, and the device resumes processing its queue once completions +/// come back. +pub const THREADED_IO_MAX_IN_FLIGHT: usize = 128; + +/// How long a finished request may wait for the requests queued behind it before the device is +/// told about it. +const COMPLETION_SIGNAL_DELAY: Duration = Duration::from_micros(200); + +/// Filter the worker threads apply to themselves, see [`set_worker_seccomp_filter`]. +static WORKER_SECCOMP_FILTER: OnceLock> = OnceLock::new(); + +/// Set the seccomp filter each worker thread applies to itself when it starts. +/// +/// Workers are started whenever a drive is created, which is before the VMM thread installs its +/// own filter, so they cannot rely on inheriting it. Only the first call has an effect. +pub fn set_worker_seccomp_filter(filter: Arc) { + // Ignoring the error is what makes later calls no-ops. + let _ = WORKER_SECCOMP_FILTER.set(filter); +} + +fn worker_seccomp_filter() -> Option> { + WORKER_SECCOMP_FILTER.get().map(|filter| filter.as_slice()) +} + +#[derive(Debug, thiserror::Error, displaydoc::Display)] +pub enum ThreadedIoError { + /// IO: {0} + Io(SyncIoError), + /// EventFd: {0} + EventFd(std::io::Error), + /// Cloning the backing file: {0} + FileClone(std::io::Error), + /// Spawning the IO worker thread: {0} + Spawn(std::io::Error), + /// Too many requests in flight + QueueFull, + /// The IO worker thread is gone + WorkerGone, +} + +/// A finished request, as reported by the worker thread. +#[derive(Debug)] +pub struct ThreadedCompletion { + pub req: PendingRequest, + pub result: Result, +} + +#[derive(Debug)] +enum Io { + Read { + offset: u64, + mem: GuestMemoryMmap, + addr: GuestAddress, + count: u32, + }, + Write { + offset: u64, + mem: GuestMemoryMmap, + addr: GuestAddress, + count: u32, + }, + Flush, +} + +impl Io { + fn execute(self, file: &mut SyncFileEngine) -> Result { + match self { + Io::Read { + offset, + mem, + addr, + count, + } => { + let count = file.read(offset, &mem, addr, count)?; + // The guest memory was written from this thread, so account for it in the dirty + // bitmap before the device gets to see the completion. + mem.mark_dirty(addr, count as usize); + Ok(count) + } + Io::Write { + offset, + mem, + addr, + count, + } => file.write(offset, &mem, addr, count), + Io::Flush => file.flush().map(|_| 0), + } + } +} + +#[derive(Debug)] +enum Op { + Io { + io: Io, + req: PendingRequest, + }, + /// Switch to a new backing file. Requests submitted before it use the old one. + UpdateFile(File), + /// Acknowledged once every request submitted before it has completed. + Barrier(mpsc::SyncSender<()>), + Exit, +} + +/// Coalesces the completion eventfd writes of the worker thread. +#[derive(Debug)] +struct CompletionSignal { + evt: EventFd, + pending: bool, + last: Instant, +} + +impl CompletionSignal { + fn new(evt: EventFd) -> Self { + CompletionSignal { + evt, + pending: false, + last: Instant::now(), + } + } + + /// Signal pending completions if the last signal is old enough. + fn maybe_flush(&mut self) { + if self.last.elapsed() >= COMPLETION_SIGNAL_DELAY { + self.flush(); + } + } + + /// Signal pending completions. + fn flush(&mut self) { + if !self.pending { + return; + } + if let Err(err) = self.evt.write(1) { + error!("Failed to signal block IO completion: {:?}", err); + } + self.pending = false; + self.last = Instant::now(); + } +} + +fn run_worker( + mut file: SyncFileEngine, + ops: mpsc::Receiver, + completions: mpsc::Sender, + completion_evt: EventFd, +) { + if let Some(filter) = worker_seccomp_filter() + && let Err(err) = crate::seccomp::apply_filter(filter) + { + panic!("Failed to set the requested seccomp filters on the block IO worker: {err}"); + } + + let mut signal = CompletionSignal::new(completion_evt); + let mut next = None; + loop { + let op = match next.take() { + Some(op) => op, + None => { + // Never go to sleep on a completion the device has not been told about. + signal.flush(); + match ops.recv() { + Ok(op) => op, + Err(mpsc::RecvError) => break, + } + } + }; + + let completion = match op { + Op::Io { io, req } => ThreadedCompletion { + req, + result: io.execute(&mut file).map_err(ThreadedIoError::Io), + }, + Op::UpdateFile(new_file) => { + file.update_file(new_file); + continue; + } + Op::Barrier(ack) => { + // The submitter may have given up waiting; nothing to do about it here. + let _ = ack.send(()); + continue; + } + Op::Exit => break, + }; + + if completions.send(completion).is_err() { + break; + } + signal.pending = true; + + // Hold the signal back while more requests are queued, so that the device handles a + // burst of completions at once instead of raising an interrupt for each. Only while the + // previous signal is recent, though: a slow backing file gets one after every request. + match ops.try_recv() { + Ok(op) => { + next = Some(op); + signal.maybe_flush(); + } + Err(_) => signal.flush(), + } + } + signal.flush(); +} + +/// Front end of the threaded engine, used from the thread that owns the device. +#[derive(Debug)] +pub struct ThreadedFileEngine { + file: File, + ops: mpsc::Sender, + completions: mpsc::Receiver, + completion_evt: EventFd, + in_flight: usize, + worker: Option>, +} + +impl ThreadedFileEngine { + pub fn from_file(file: File) -> Result { + let completion_evt = EventFd::new(libc::EFD_NONBLOCK).map_err(ThreadedIoError::EventFd)?; + let worker_evt = completion_evt + .try_clone() + .map_err(ThreadedIoError::EventFd)?; + let worker_file = + SyncFileEngine::from_file(file.try_clone().map_err(ThreadedIoError::FileClone)?); + let (ops, worker_ops) = mpsc::channel(); + let (worker_completions, completions) = mpsc::channel(); + + let worker = thread::Builder::new() + .name("fc_blk_io".to_string()) + .spawn(move || run_worker(worker_file, worker_ops, worker_completions, worker_evt)) + .map_err(ThreadedIoError::Spawn)?; + + Ok(ThreadedFileEngine { + file, + ops, + completions, + completion_evt, + in_flight: 0, + worker: Some(worker), + }) + } + + #[cfg(test)] + pub fn file(&self) -> &File { + &self.file + } + + /// Update the backing file of the engine + pub fn update_file(&mut self, file: File) -> Result<(), ThreadedIoError> { + let worker_file = file.try_clone().map_err(ThreadedIoError::FileClone)?; + self.ops + .send(Op::UpdateFile(worker_file)) + .map_err(|_| ThreadedIoError::WorkerGone)?; + self.file = file; + Ok(()) + } + + pub fn completion_evt(&self) -> &EventFd { + &self.completion_evt + } + + fn push(&mut self, io: Io, req: PendingRequest) -> Result<(), RequestError> { + if self.in_flight >= THREADED_IO_MAX_IN_FLIGHT { + return Err(RequestError { + req, + error: ThreadedIoError::QueueFull, + }); + } + + match self.ops.send(Op::Io { io, req }) { + Ok(()) => { + self.in_flight += 1; + Ok(()) + } + Err(mpsc::SendError(Op::Io { req, .. })) => Err(RequestError { + req, + error: ThreadedIoError::WorkerGone, + }), + Err(_) => unreachable!("sent an IO op"), + } + } + + pub fn push_read( + &mut self, + offset: u64, + mem: &GuestMemoryMmap, + addr: GuestAddress, + count: u32, + req: PendingRequest, + ) -> Result<(), RequestError> { + let io = Io::Read { + offset, + mem: mem.clone(), + addr, + count, + }; + self.push(io, req) + } + + pub fn push_write( + &mut self, + offset: u64, + mem: &GuestMemoryMmap, + addr: GuestAddress, + count: u32, + req: PendingRequest, + ) -> Result<(), RequestError> { + let io = Io::Write { + offset, + mem: mem.clone(), + addr, + count, + }; + self.push(io, req) + } + + pub fn push_flush(&mut self, req: PendingRequest) -> Result<(), RequestError> { + self.push(Io::Flush, req) + } + + /// Pop a finished request, if there is one. + pub fn pop(&mut self) -> Option { + let completion = self.completions.try_recv().ok()?; + self.in_flight -= 1; + Some(completion) + } + + /// Wait for every submitted request to complete. Their completions are left to be popped, + /// unless `discard` is set. + pub fn drain(&mut self, discard: bool) -> Result<(), ThreadedIoError> { + if self.in_flight > 0 { + let (ack, done) = mpsc::sync_channel(1); + self.ops + .send(Op::Barrier(ack)) + .map_err(|_| ThreadedIoError::WorkerGone)?; + done.recv().map_err(|_| ThreadedIoError::WorkerGone)?; + } + + if discard { + while self.pop().is_some() {} + } + + Ok(()) + } + + pub fn drain_and_flush(&mut self, discard: bool) -> Result<(), ThreadedIoError> { + self.drain(discard)?; + + // Sync data out to physical media on host. The worker holds no data of its own, so the + // file descriptor here reaches everything it wrote. + self.file + .sync_all() + .map_err(|err| ThreadedIoError::Io(SyncIoError::SyncAll(err))) + } +} + +impl Drop for ThreadedFileEngine { + fn drop(&mut self) { + // The worker only exits once it gets here, so everything submitted before is finished. + let _ = self.ops.send(Op::Exit); + if let Some(worker) = self.worker.take() + && worker.join().is_err() + { + error!("The block IO worker thread panicked"); + } + } +} + +#[cfg(test)] +mod tests { + #![allow(clippy::undocumented_unsafe_blocks)] + use std::os::unix::ffi::OsStrExt; + use std::os::unix::fs::MetadataExt; + + use vm_memory::GuestMemoryRegion; + use vmm_sys_util::tempfile::TempFile; + + use super::*; + use crate::utils::u64_to_usize; + use crate::vmm_config::machine_config::HugePageConfig; + use crate::vstate::memory; + use crate::vstate::memory::{Bitmap, Bytes, GuestMemory, GuestRegionMmapExt}; + + const FILE_LEN: u32 = 1024; + // 2 pages of memory should be enough to test read/write ops and also dirty tracking. + const MEM_LEN: usize = 8192; + + fn create_mem() -> GuestMemoryMmap { + GuestMemoryMmap::from_regions( + memory::anonymous( + [(GuestAddress(0), MEM_LEN)].into_iter(), + true, + HugePageConfig::None, + ) + .unwrap() + .into_iter() + .map(|region| GuestRegionMmapExt::dram_from_mmap_region(region, 0)) + .collect(), + ) + .unwrap() + } + + fn check_dirty_mem(mem: &GuestMemoryMmap, addr: GuestAddress, len: u32, dirty: bool) { + let bitmap = mem.find_region(addr).unwrap().bitmap(); + for offset in addr.0..addr.0 + u64::from(len) { + assert_eq!(bitmap.dirty_at(u64_to_usize(offset)), dirty); + } + } + + fn new_engine() -> ThreadedFileEngine { + ThreadedFileEngine::from_file(TempFile::new().unwrap().into_file()).unwrap() + } + + fn assert_completed(engine: &mut ThreadedFileEngine, count: u32) { + engine.drain(false).unwrap(); + assert_eq!(engine.pop().unwrap().result.unwrap(), count); + } + + fn wait_for_signal(engine: &ThreadedFileEngine) { + let deadline = Instant::now() + Duration::from_secs(5); + while engine.completion_evt().read().is_err() { + assert!(Instant::now() < deadline, "completion never signalled"); + thread::sleep(Duration::from_millis(1)); + } + } + + #[test] + fn test_read_write_flush() { + let mut engine = new_engine(); + let data = vmm_sys_util::rand::rand_alphanumerics(FILE_LEN as usize) + .as_bytes() + .to_vec(); + + // Partial write and read, at the end of guest memory. + let partial_len = 50; + let addr = GuestAddress(MEM_LEN as u64 - u64::from(partial_len)); + let mem = create_mem(); + mem.write(&data, addr).unwrap(); + engine + .push_write(0, &mem, addr, partial_len, PendingRequest::default()) + .unwrap(); + assert_completed(&mut engine, partial_len); + let mem = create_mem(); + engine + .push_read(0, &mem, addr, partial_len, PendingRequest::default()) + .unwrap(); + assert_completed(&mut engine, partial_len); + let mut buf = vec![0u8; partial_len as usize]; + mem.read_slice(&mut buf, addr).unwrap(); + assert_eq!(buf, data[..partial_len as usize]); + + // Full write and read, at an offset. + let mem = create_mem(); + mem.write(&data, GuestAddress(0)).unwrap(); + engine + .push_write( + 100, + &mem, + GuestAddress(0), + FILE_LEN, + PendingRequest::default(), + ) + .unwrap(); + assert_completed(&mut engine, FILE_LEN); + let mem = create_mem(); + engine + .push_read( + 100, + &mem, + GuestAddress(0), + FILE_LEN, + PendingRequest::default(), + ) + .unwrap(); + assert_completed(&mut engine, FILE_LEN); + let mut buf = vec![0u8; FILE_LEN as usize]; + mem.read_slice(&mut buf, GuestAddress(0)).unwrap(); + assert_eq!(buf, data); + + // Reads from the worker thread are accounted for in the dirty bitmap. + check_dirty_mem(&mem, GuestAddress(0), FILE_LEN, true); + check_dirty_mem(&mem, GuestAddress(4096), 4096, false); + + // Out of bounds guest memory fails the request, not the engine. + engine + .push_read( + 0, + &mem, + GuestAddress(MEM_LEN as u64), + FILE_LEN, + PendingRequest::default(), + ) + .unwrap(); + engine.drain(false).unwrap(); + assert!(matches!( + engine.pop().unwrap().result, + Err(ThreadedIoError::Io(SyncIoError::Transfer(_))) + )); + + engine.push_flush(PendingRequest::default()).unwrap(); + assert_completed(&mut engine, 0); + engine.drain_and_flush(true).unwrap(); + } + + #[test] + fn test_completions_are_signalled() { + let mem = create_mem(); + let mut engine = new_engine(); + + // Submitting returns before the IO is done: completions only show up through the + // completion eventfd, once the worker has finished them. + for _ in 0..10 { + engine + .push_write( + 0, + &mem, + GuestAddress(0), + FILE_LEN, + PendingRequest::default(), + ) + .unwrap(); + } + let mut completed = 0; + while completed < 10 { + wait_for_signal(&engine); + while let Some(completion) = engine.pop() { + assert_eq!(completion.result.unwrap(), FILE_LEN); + completed += 1; + } + } + assert!(engine.pop().is_none()); + } + + #[test] + fn test_signal_before_idle() { + let mem = create_mem(); + let mut engine = new_engine(); + + // Queue ops that complete nothing behind a request: the worker must still signal the + // request's completion before it goes idle. + engine + .push_write( + 0, + &mem, + GuestAddress(0), + FILE_LEN, + PendingRequest::default(), + ) + .unwrap(); + engine + .update_file(TempFile::new().unwrap().into_file()) + .unwrap(); + + wait_for_signal(&engine); + assert_eq!(engine.pop().unwrap().result.unwrap(), FILE_LEN); + } + + #[test] + fn test_throttling() { + let mut engine = new_engine(); + + for _ in 0..THREADED_IO_MAX_IN_FLIGHT { + engine.push_flush(PendingRequest::default()).unwrap(); + } + // Completed but not yet popped requests still count as in flight. + engine.drain(false).unwrap(); + let err = engine.push_flush(PendingRequest::default()).unwrap_err(); + assert!(matches!(err.error, ThreadedIoError::QueueFull)); + + // Popping a completion makes room for one more request. + engine.pop().unwrap(); + engine.push_flush(PendingRequest::default()).unwrap(); + let err = engine.push_flush(PendingRequest::default()).unwrap_err(); + assert!(matches!(err.error, ThreadedIoError::QueueFull)); + + // Discarding all completions makes room for all of them. + engine.drain(true).unwrap(); + assert!(engine.pop().is_none()); + for _ in 0..THREADED_IO_MAX_IN_FLIGHT { + engine.push_flush(PendingRequest::default()).unwrap(); + } + engine.drain(true).unwrap(); + } + + #[test] + fn test_update_file() { + let mem = create_mem(); + let old = TempFile::new().unwrap(); + let new = TempFile::new().unwrap(); + let mut engine = ThreadedFileEngine::from_file(old.as_file().try_clone().unwrap()).unwrap(); + + let data = vmm_sys_util::rand::rand_alphanumerics(FILE_LEN as usize) + .as_bytes() + .to_vec(); + mem.write(&data, GuestAddress(0)).unwrap(); + + // Requests submitted before the update go to the old file, the ones after to the new one, + // without waiting for the first to complete. + engine + .push_write( + 0, + &mem, + GuestAddress(0), + FILE_LEN, + PendingRequest::default(), + ) + .unwrap(); + engine + .update_file(new.as_file().try_clone().unwrap()) + .unwrap(); + engine + .push_write(0, &mem, GuestAddress(0), 10, PendingRequest::default()) + .unwrap(); + engine.drain(true).unwrap(); + + assert_eq!(old.as_file().metadata().unwrap().len(), u64::from(FILE_LEN)); + assert_eq!(new.as_file().metadata().unwrap().len(), 10); + assert_eq!( + engine.file().metadata().unwrap().ino(), + new.as_file().metadata().unwrap().ino() + ); + } +} diff --git a/src/vmm/src/devices/virtio/block/virtio/mod.rs b/src/vmm/src/devices/virtio/block/virtio/mod.rs index 9e97d6d3897..8cd130f76ee 100644 --- a/src/vmm/src/devices/virtio/block/virtio/mod.rs +++ b/src/vmm/src/devices/virtio/block/virtio/mod.rs @@ -10,6 +10,9 @@ pub mod metrics; pub mod persist; pub mod request; pub mod test_utils; +mod threaded; + +pub use self::io::threaded_io::set_worker_seccomp_filter; use vm_memory::GuestMemoryError; diff --git a/src/vmm/src/devices/virtio/block/virtio/persist.rs b/src/vmm/src/devices/virtio/block/virtio/persist.rs index 380fe1de0e8..cfb890582fa 100644 --- a/src/vmm/src/devices/virtio/block/virtio/persist.rs +++ b/src/vmm/src/devices/virtio/block/virtio/persist.rs @@ -30,6 +30,8 @@ pub enum FileEngineTypeState { Sync, /// Async File Engine. Async, + /// Threaded File Engine. + Threaded, } impl From for FileEngineTypeState { @@ -37,6 +39,7 @@ impl From for FileEngineTypeState { match file_engine_type { FileEngineType::Sync => FileEngineTypeState::Sync, FileEngineType::Async => FileEngineTypeState::Async, + FileEngineType::Threaded => FileEngineTypeState::Threaded, } } } @@ -46,6 +49,7 @@ impl From for FileEngineType { match file_engine_type_state { FileEngineTypeState::Sync => FileEngineType::Sync, FileEngineTypeState::Async => FileEngineType::Async, + FileEngineTypeState::Threaded => FileEngineType::Threaded, } } } diff --git a/src/vmm/src/devices/virtio/block/virtio/test_utils.rs b/src/vmm/src/devices/virtio/block/virtio/test_utils.rs index e4f23c6a038..8025f89d8ef 100644 --- a/src/vmm/src/devices/virtio/block/virtio/test_utils.rs +++ b/src/vmm/src/devices/virtio/block/virtio/test_utils.rs @@ -122,6 +122,10 @@ pub fn simulate_queue_and_async_completion_events(b: &mut VirtioBlock, expected_ FileEngine::Sync(_) => { simulate_queue_event(b, Some(expected_irq)); } + FileEngine::Threaded(_) => { + simulate_queue_event(b, None); + super::threaded::simulate_threaded_completion_event(b, expected_irq); + } } } diff --git a/src/vmm/src/devices/virtio/block/virtio/threaded.rs b/src/vmm/src/devices/virtio/block/virtio/threaded.rs new file mode 100644 index 00000000000..11e21d013cf --- /dev/null +++ b/src/vmm/src/devices/virtio/block/virtio/threaded.rs @@ -0,0 +1,330 @@ +// Copyright 2026 Fly.io, Inc. +// SPDX-License-Identifier: Apache-2.0 + +//! Block device support for the threaded IO engine: handing the requests its worker thread +//! finished back to the guest. + +use super::device::VirtioBlock; +use super::io::{BlockIoError, FileEngine}; +use super::request::IoErr; +use crate::devices::virtio::transport::VirtioInterruptType; +use crate::logger::{IncMetric, error}; + +impl VirtioBlock { + /// Handle the threaded engine's completion eventfd. + pub(crate) fn process_threaded_completion_event(&mut self) { + let FileEngine::Threaded(engine) = &self.disk.file_engine else { + error!("The block device doesn't use a threaded IO engine"); + return; + }; + + if let Err(err) = engine.completion_evt().read() { + error!("Failed to get threaded completion event: {:?}", err); + return; + } + self.process_threaded_completion_queue(); + + if self.is_io_engine_throttled { + self.is_io_engine_throttled = false; + self.process_queue(0).unwrap() + } + } + + /// Add every request the worker finished to the used ring, and notify the guest. + pub(crate) fn process_threaded_completion_queue(&mut self) { + let FileEngine::Threaded(engine) = &mut self.disk.file_engine else { + return; + }; + + // This is safe since we checked in the event handler that the device is activated. + let active_state = self.device_state.active_state().unwrap(); + let queue = &mut self.queues[0]; + + while let Some(completion) = engine.pop() { + let res = completion + .result + .map_err(|err| IoErr::FileEngine(BlockIoError::Threaded(err))); + let finished = completion.req.finish(&active_state.mem, res, &self.metrics); + queue + .add_used(finished.desc_idx, finished.num_bytes_to_mem) + .unwrap_or_else(|err| { + error!( + "Failed to add available descriptor head {}: {}", + finished.desc_idx, err + ) + }); + } + queue.advance_used_ring_idx(); + + if queue.prepare_kick() { + active_state + .interrupt + .trigger(VirtioInterruptType::Queue(0)) + .unwrap_or_else(|_| { + self.metrics.event_fails.inc(); + }); + } + } +} + +/// Wait for the threaded engine to finish everything submitted, then handle its completion event +/// and check whether the guest got an interrupt. +#[cfg(test)] +pub fn simulate_threaded_completion_event(b: &mut VirtioBlock, expected_irq: bool) { + use crate::devices::virtio::device::VirtioDevice; + + b.disk.file_engine.drain(false).unwrap(); + b.process_threaded_completion_event(); + assert_eq!( + b.interrupt_trigger() + .has_pending_interrupt(VirtioInterruptType::Queue(0)), + expected_irq + ); +} + +#[cfg(test)] +mod tests { + use std::sync::{Arc, Mutex}; + + use event_manager::{EventManager, SubscriberOps}; + use vmm_sys_util::tempfile::TempFile; + + use super::*; + use crate::devices::virtio::block::persist::BlockConstructorArgs; + use crate::devices::virtio::block::virtio::device::{FileEngineType, VirtioBlockConfig}; + use crate::devices::virtio::block::virtio::io::threaded_io::THREADED_IO_MAX_IN_FLIGHT; + use crate::devices::virtio::block::virtio::test_utils::{ + default_block, read_blk_req_descriptors, set_queue, simulate_queue_event, + }; + use crate::devices::virtio::block::virtio::{ + CacheType, RequestHeader, VIRTIO_BLK_S_OK, VIRTIO_BLK_T_FLUSH, VIRTIO_BLK_T_IN, + VIRTIO_BLK_T_OUT, + }; + use crate::devices::virtio::device::VirtioDevice; + use crate::devices::virtio::queue::{VIRTQ_DESC_F_NEXT, VIRTQ_DESC_F_WRITE}; + use crate::devices::virtio::test_utils::{VirtQueue, default_interrupt, default_mem}; + use crate::snapshot::{Persist, Snapshot}; + use crate::vstate::memory::{Address, Bytes, GuestAddress}; + + fn add_flush_requests_batch(block: &mut VirtioBlock, vq: &VirtQueue, count: u16) { + let mem = vq.memory(); + vq.avail.idx.set(0); + vq.used.idx.set(0); + set_queue(block, 0, vq.create_queue()); + + let hdr_addr = vq + .end() + .checked_align_up(std::mem::align_of::() as u64) + .unwrap(); + mem.write_obj(RequestHeader::new(VIRTIO_BLK_T_FLUSH, 0), hdr_addr) + .unwrap(); + let mut status_addr = hdr_addr + .checked_add(std::mem::size_of::() as u64) + .unwrap() + .checked_align_up(4) + .unwrap(); + + for i in 0..count { + let idx = i * 2; + let hdr_desc = &vq.dtable[idx as usize]; + hdr_desc.addr.set(hdr_addr.0); + hdr_desc.flags.set(VIRTQ_DESC_F_NEXT); + hdr_desc.next.set(idx + 1); + + let status_desc = &vq.dtable[idx as usize + 1]; + status_desc.addr.set(status_addr.0); + status_desc.flags.set(VIRTQ_DESC_F_WRITE); + status_desc.len.set(4); + status_addr = status_addr.checked_add(4).unwrap(); + + vq.avail.ring[i as usize].set(idx); + vq.avail.idx.set(i + 1); + } + } + + fn check_flush_requests_batch(count: u16, vq: &VirtQueue) { + assert_eq!(vq.used.idx.get(), count); + for i in 0..count { + let used = vq.used.ring[i as usize].get(); + let status_addr = vq.dtable[used.id as usize + 1].addr.get(); + assert_eq!(used.len, 1); + assert_eq!( + u32::from( + vq.memory() + .read_obj::(GuestAddress(status_addr)) + .unwrap() + ), + VIRTIO_BLK_S_OK + ); + } + } + + #[test] + fn test_engine_type() { + let block = default_block(FileEngineType::Threaded); + assert!(matches!(block.disk.file_engine, FileEngine::Threaded(_))); + assert_eq!(block.file_engine_type(), FileEngineType::Threaded); + assert_eq!(block.config().file_engine_type, FileEngineType::Threaded); + } + + #[test] + fn test_read_write() { + let mut block = default_block(FileEngineType::Threaded); + let mem = default_mem(); + let vq = VirtQueue::new(GuestAddress(0), &mem, 16); + set_queue(&mut block, 0, vq.create_queue()); + block.activate(mem.clone(), default_interrupt()).unwrap(); + read_blk_req_descriptors(&vq); + + let request_type_addr = GuestAddress(vq.dtable[0].addr.get()); + let data_addr = GuestAddress(vq.dtable[1].addr.get()); + let status_addr = GuestAddress(vq.dtable[2].addr.get()); + + // Write. + mem.write_obj::(VIRTIO_BLK_T_OUT, request_type_addr) + .unwrap(); + vq.dtable[1].flags.set(VIRTQ_DESC_F_NEXT); + vq.dtable[1].len.set(512); + mem.write_obj::(123_456_789, data_addr).unwrap(); + // Submitting does not complete the request... + simulate_queue_event(&mut block, Some(false)); + assert_eq!(vq.used.idx.get(), 0); + // ...the worker's completion does. + simulate_threaded_completion_event(&mut block, true); + assert_eq!(vq.used.idx.get(), 1); + assert_eq!(vq.used.ring[0].get().len, 1); + assert_eq!(mem.read_obj::(status_addr).unwrap(), VIRTIO_BLK_S_OK); + + // Read it back. + vq.used.idx.set(0); + set_queue(&mut block, 0, vq.create_queue()); + mem.write_obj::(VIRTIO_BLK_T_IN, request_type_addr) + .unwrap(); + vq.dtable[1] + .flags + .set(VIRTQ_DESC_F_NEXT | VIRTQ_DESC_F_WRITE); + mem.write_obj::(0, data_addr).unwrap(); + simulate_queue_event(&mut block, Some(false)); + simulate_threaded_completion_event(&mut block, true); + assert_eq!(vq.used.idx.get(), 1); + assert_eq!(vq.used.ring[0].get().len, 513); + assert_eq!(mem.read_obj::(status_addr).unwrap(), VIRTIO_BLK_S_OK); + assert_eq!(mem.read_obj::(data_addr).unwrap(), 123_456_789); + } + + #[test] + fn test_throttling() { + let limit = u16::try_from(THREADED_IO_MAX_IN_FLIGHT).unwrap(); + let mut block = default_block(FileEngineType::Threaded); + let mem = default_mem(); + let vq = VirtQueue::new(GuestAddress(0), &mem, limit * 4); + block.queues[0] = vq.create_queue(); + block.activate(mem.clone(), default_interrupt()).unwrap(); + + // Up to the limit, everything is submitted. + add_flush_requests_batch(&mut block, &vq, limit); + simulate_queue_event(&mut block, Some(false)); + assert!(!block.is_io_engine_throttled); + simulate_threaded_completion_event(&mut block, true); + check_flush_requests_batch(limit, &vq); + + // Beyond it, the device stops until completions come back, then resumes the queue. + add_flush_requests_batch(&mut block, &vq, limit + 10); + simulate_queue_event(&mut block, Some(false)); + assert!(block.is_io_engine_throttled); + simulate_threaded_completion_event(&mut block, true); + assert!(!block.is_io_engine_throttled); + check_flush_requests_batch(limit, &vq); + simulate_threaded_completion_event(&mut block, true); + assert!(!block.is_io_engine_throttled); + check_flush_requests_batch(limit + 10, &vq); + } + + #[test] + fn test_prepare_save() { + let mut block = default_block(FileEngineType::Threaded); + let mem = default_mem(); + let vq = VirtQueue::new(GuestAddress(0), &mem, 16); + block.queues[0] = vq.create_queue(); + block.activate(mem.clone(), default_interrupt()).unwrap(); + + add_flush_requests_batch(&mut block, &vq, 5); + simulate_queue_event(&mut block, None); + block.prepare_save(); + + // Every request submitted to the worker was finished and handed back to the guest. + check_flush_requests_batch(5, &vq); + } + + #[test] + fn test_event_handler() { + let mut event_manager = EventManager::new().unwrap(); + let mut block = default_block(FileEngineType::Threaded); + let mem = default_mem(); + let vq = VirtQueue::new(GuestAddress(0), &mem, 16); + set_queue(&mut block, 0, vq.create_queue()); + read_blk_req_descriptors(&vq); + mem.write_obj::(VIRTIO_BLK_T_OUT, GuestAddress(vq.dtable[0].addr.get())) + .unwrap(); + vq.dtable[1].flags.set(VIRTQ_DESC_F_NEXT); + vq.dtable[1].len.set(512); + + let block = Arc::new(Mutex::new(block)); + event_manager.add_subscriber(block.clone()); + block + .lock() + .unwrap() + .activate(mem.clone(), default_interrupt()) + .unwrap(); + // Process the activate event. + assert_eq!(event_manager.run_with_timeout(50).unwrap(), 1); + + // The queue event submits the request, and the completion event, registered by the + // device, finishes it. + block.lock().unwrap().queue_evts[0].write(1).unwrap(); + for _ in 0..10 { + event_manager.run_with_timeout(100).unwrap(); + if vq.used.idx.get() == 1 { + break; + } + } + assert_eq!(vq.used.idx.get(), 1); + assert_eq!( + mem.read_obj::(GuestAddress(vq.dtable[2].addr.get())) + .unwrap(), + VIRTIO_BLK_S_OK + ); + } + + #[test] + fn test_persistence() { + let f = TempFile::new().unwrap(); + f.as_file().set_len(0x1000).unwrap(); + let block = VirtioBlock::new(VirtioBlockConfig { + drive_id: "threaded".to_string(), + path_on_host: f.as_path().to_str().unwrap().to_string(), + is_root_device: false, + partuuid: None, + is_read_only: false, + cache_type: CacheType::Unsafe, + rate_limiter: None, + file_engine_type: FileEngineType::Threaded, + }) + .unwrap(); + + let mut snapshot = vec![0; 4096]; + Snapshot::new(block.save()) + .save(&mut snapshot.as_mut_slice()) + .unwrap(); + let restored = VirtioBlock::restore( + BlockConstructorArgs { mem: default_mem() }, + &Snapshot::load_without_crc_check(snapshot.as_slice()) + .unwrap() + .data, + ) + .unwrap(); + + assert_eq!(restored.file_engine_type(), FileEngineType::Threaded); + assert!(matches!(restored.disk.file_engine, FileEngine::Threaded(_))); + } +} From a849576cb930d302b4fa208466e8f3affd9f158b Mon Sep 17 00:00:00 2001 From: Akshit Garg Date: Thu, 24 Sep 2026 10:16:21 +0530 Subject: [PATCH 2/3] feat(rate-limiter): adapt refill timer delay - Schedule depleted buckets based on their token deficit, within configured bounds - Apply a 5 ms minimum refill delay to virtio block limiters and restore it after persistence (cherry picked from commit 8202cfb2906814e124fc8a5a3d6f510d8603dc23) --- .../src/devices/virtio/block/virtio/device.rs | 8 +- .../src/devices/virtio/block/virtio/mod.rs | 3 + .../devices/virtio/block/virtio/persist.rs | 12 +- src/vmm/src/rate_limiter/mod.rs | 144 +++++++++++++++++- src/vmm/src/rate_limiter/persist.rs | 2 + 5 files changed, 165 insertions(+), 4 deletions(-) diff --git a/src/vmm/src/devices/virtio/block/virtio/device.rs b/src/vmm/src/devices/virtio/block/virtio/device.rs index 2d3c14b653a..e21e297b828 100644 --- a/src/vmm/src/devices/virtio/block/virtio/device.rs +++ b/src/vmm/src/devices/virtio/block/virtio/device.rs @@ -21,7 +21,10 @@ use vmm_sys_util::eventfd::EventFd; use super::io::async_io; use super::request::*; -use super::{BLOCK_QUEUE_SIZES, SECTOR_SHIFT, SECTOR_SIZE, VirtioBlockError, io as block_io}; +use super::{ + BLOCK_QUEUE_SIZES, RATE_LIMITER_MIN_REFILL_DELAY, SECTOR_SHIFT, SECTOR_SIZE, VirtioBlockError, + io as block_io, +}; use crate::devices::virtio::ActivateError; use crate::devices::virtio::block::CacheType; use crate::devices::virtio::block::virtio::metrics::{BlockDeviceMetrics, BlockMetricsPerDevice}; @@ -293,12 +296,13 @@ impl VirtioBlock { config.file_engine_type, )?; - let rate_limiter = config + let mut rate_limiter: RateLimiter = config .rate_limiter .map(RateLimiterConfig::try_into) .transpose() .map_err(VirtioBlockError::RateLimiter)? .unwrap_or_default(); + rate_limiter.set_min_refill_delay(RATE_LIMITER_MIN_REFILL_DELAY); let mut avail_features = (1u64 << VIRTIO_F_VERSION_1) | (1u64 << VIRTIO_RING_F_EVENT_IDX); diff --git a/src/vmm/src/devices/virtio/block/virtio/mod.rs b/src/vmm/src/devices/virtio/block/virtio/mod.rs index 8cd130f76ee..fc0985c4110 100644 --- a/src/vmm/src/devices/virtio/block/virtio/mod.rs +++ b/src/vmm/src/devices/virtio/block/virtio/mod.rs @@ -32,6 +32,9 @@ pub const BLOCK_QUEUE_SIZES: [u16; BLOCK_NUM_QUEUES] = [FIRECRACKER_MAX_QUEUE_SI // So we can use 128 IO_URING entries without ever triggering a FullSq Error. /// Maximum number of io uring entries we allow in the queue. pub const IO_URING_NUM_ENTRIES: u16 = 128; +/// Minimum wait before retrying a depleted rate limiter. The limiter otherwise waits +/// until the failed request's tokens have refilled, rather than a fixed 100ms. +pub const RATE_LIMITER_MIN_REFILL_DELAY: std::time::Duration = std::time::Duration::from_millis(5); /// Errors the block device can trigger. #[derive(Debug, thiserror::Error, displaydoc::Display)] diff --git a/src/vmm/src/devices/virtio/block/virtio/persist.rs b/src/vmm/src/devices/virtio/block/virtio/persist.rs index cfb890582fa..cde43b0fecf 100644 --- a/src/vmm/src/devices/virtio/block/virtio/persist.rs +++ b/src/vmm/src/devices/virtio/block/virtio/persist.rs @@ -91,8 +91,9 @@ impl Persist<'_> for VirtioBlock { state: &Self::State, ) -> Result { let is_read_only = state.virtio_state.avail_features & (1u64 << VIRTIO_BLK_F_RO) != 0; - let rate_limiter = RateLimiter::restore((), &state.rate_limiter_state) + let mut rate_limiter = RateLimiter::restore((), &state.rate_limiter_state) .map_err(VirtioBlockError::RateLimiter)?; + rate_limiter.set_min_refill_delay(RATE_LIMITER_MIN_REFILL_DELAY); let disk_properties = DiskProperties::new( state.disk_path.clone(), @@ -243,5 +244,14 @@ mod tests { // Test that block specific fields are the same. assert_eq!(restored_block.disk.file_path, block.disk.file_path); + // The adaptive refill delay is device policy and must be reapplied on restore. + assert_eq!( + block.rate_limiter.min_refill_delay(), + Some(RATE_LIMITER_MIN_REFILL_DELAY) + ); + assert_eq!( + restored_block.rate_limiter.min_refill_delay(), + Some(RATE_LIMITER_MIN_REFILL_DELAY) + ); } } diff --git a/src/vmm/src/rate_limiter/mod.rs b/src/vmm/src/rate_limiter/mod.rs index 97ebad51fcc..d97543b6433 100644 --- a/src/vmm/src/rate_limiter/mod.rs +++ b/src/vmm/src/rate_limiter/mod.rs @@ -236,6 +236,24 @@ impl TokenBucket { self.budget = std::cmp::min(self.budget.saturating_add(tokens), self.size); } + /// Returns how long until the budget covers `tokens`, assuming no other consumption. + /// + /// Meant to be called right after `reduce()` failed, when the budget has just been + /// replenished up to `last_update`. + fn refill_delay(&self, tokens: u64) -> Duration { + let deficit = u128::from(tokens.saturating_sub(self.budget)); + if deficit == 0 { + return Duration::ZERO; + } + let processed_capacity = u128::from(self.processed_capacity); + let processed_refill_time = u128::from(self.processed_refill_time); + // Round up so the timer never fires before the tokens are available. + let needed_ns = (deficit * processed_refill_time).div_ceil(processed_capacity); + // `auto_replenish()` carries sub-token time in `last_update`; it counts towards the wait. + let needed_ns = needed_ns.saturating_sub(self.last_update.elapsed().as_nanos()); + Duration::from_nanos(u64::try_from(needed_ns).unwrap_or(u64::MAX)) + } + /// Returns the capacity of the token bucket. pub fn capacity(&self) -> u64 { self.size @@ -304,6 +322,9 @@ pub struct RateLimiter { timer_fd: TimerFd, // Internal flag that quickly determines timer state. timer_active: bool, + // When set, a depleted bucket arms the timer for the time its tokens take to refill, + // bounded below by this value. When unset, the timer always waits the fixed interval. + min_refill_delay: Option, } impl PartialEq for RateLimiter { @@ -374,9 +395,23 @@ impl RateLimiter { ops: ops_token_bucket, timer_fd, timer_active: false, + min_refill_delay: None, }) } + /// Arms the refill timer for the time a depleted bucket needs to cover the failed request, + /// but no less than `min_delay` and no more than the fixed refill interval. + /// + /// Without this, the limiter always waits the fixed refill interval once a bucket is empty. + pub fn set_min_refill_delay(&mut self, min_delay: Duration) { + self.min_refill_delay = Some(min_delay); + } + + /// Returns the minimum refill delay, if adaptive refill timing is enabled. + pub fn min_refill_delay(&self) -> Option { + self.min_refill_delay + } + // Arm the timer of the rate limiter with the provided `TimerState`. fn activate_timer(&mut self, timer_state: TimerState) { // Register the timer; don't care about its previous state @@ -407,7 +442,16 @@ impl RateLimiter { // make sure there is only one running timer for this limiter. BucketReduction::Failure => { if !self.timer_active { - self.activate_timer(TIMER_REFILL_STATE); + let timer_state = match self.min_refill_delay { + Some(min_delay) => TimerState::Oneshot( + bucket + .refill_delay(tokens) + .max(min_delay) + .min(Duration::from_millis(REFILL_TIMER_INTERVAL_MS)), + ), + None => TIMER_REFILL_STATE, + }; + self.activate_timer(timer_state); } false } @@ -996,6 +1040,104 @@ pub(crate) mod tests { } } + #[test] + fn test_token_bucket_refill_delay() { + // 1 token per millisecond. + let mut tb = TokenBucket::new(1000, 0, 1000).unwrap(); + assert_eq!(tb.refill_delay(1000), Duration::ZERO); + + assert_eq!(tb.reduce(1000), BucketReduction::Success); + tb.last_update = Instant::now(); + let delay = tb.refill_delay(10); + assert!(delay <= Duration::from_millis(10), "{delay:?}"); + assert!(delay > Duration::from_millis(9), "{delay:?}"); + + // Time already accrued towards the next tokens shortens the wait. + tb.last_update = Instant::now() - Duration::from_millis(4); + let delay = tb.refill_delay(10); + assert!(delay <= Duration::from_millis(6), "{delay:?}"); + assert!(delay > Duration::from_millis(5), "{delay:?}"); + + // Only the deficit over the current budget needs to refill. + tb.force_replenish(8); + tb.last_update = Instant::now(); + let delay = tb.refill_delay(10); + assert!(delay <= Duration::from_millis(2), "{delay:?}"); + assert!(delay > Duration::from_millis(1), "{delay:?}"); + + // Rounds up to the nanosecond the token becomes available: 3 tokens per millisecond. + let mut tb = TokenBucket::new(3, 0, 1).unwrap(); + assert_eq!(tb.reduce(3), BucketReduction::Success); + // A future `last_update` means no time has accrued yet. + tb.last_update = Instant::now() + Duration::from_secs(1); + assert_eq!(tb.refill_delay(1), Duration::from_nanos(333_334)); + } + + fn armed_delay(l: &RateLimiter) -> Duration { + match l.timer_fd.get_state() { + TimerState::Oneshot(delay) => delay, + state => panic!("expected an armed oneshot timer, got {state:?}"), + } + } + + #[test] + fn test_rate_limiter_adaptive_refill_timer() { + let min_delay = Duration::from_millis(5); + let max_delay = Duration::from_millis(REFILL_TIMER_INTERVAL_MS); + + // Bandwidth of 1 byte per millisecond: 20 missing bytes refill in 20ms. + let mut l = RateLimiter::new(1000, 0, 1000, 0, 0, 0).unwrap(); + assert_eq!(l.min_refill_delay(), None); + l.set_min_refill_delay(min_delay); + assert_eq!(l.min_refill_delay(), Some(min_delay)); + assert!(l.consume(1000, TokenType::Bytes)); + assert!(!l.consume(20, TokenType::Bytes)); + assert!(l.is_blocked()); + let delay = armed_delay(&l); + assert!(delay <= Duration::from_millis(20), "{delay:?}"); + assert!(delay > Duration::from_millis(15), "{delay:?}"); + // The limiter unblocks once those tokens are available, well before 100ms. + thread::sleep(Duration::from_millis(30)); + l.event_handler().unwrap(); + assert!(!l.is_blocked()); + assert!(l.consume(20, TokenType::Bytes)); + + // A tiny deficit still waits for the minimum delay. + let mut l = RateLimiter::new(1000, 0, 1000, 0, 0, 0).unwrap(); + l.set_min_refill_delay(min_delay); + assert!(l.consume(1000, TokenType::Bytes)); + assert!(!l.consume(1, TokenType::Bytes)); + let delay = armed_delay(&l); + assert!(delay <= min_delay, "{delay:?}"); + assert!(delay > min_delay - Duration::from_millis(2), "{delay:?}"); + + // A large deficit waits no longer than the fixed refill interval. + let mut l = RateLimiter::new(1000, 0, 1000, 0, 0, 0).unwrap(); + l.set_min_refill_delay(min_delay); + assert!(l.consume(1000, TokenType::Bytes)); + assert!(!l.consume(500, TokenType::Bytes)); + let delay = armed_delay(&l); + assert!(delay <= max_delay, "{delay:?}"); + assert!(delay > max_delay - Duration::from_millis(5), "{delay:?}"); + + // The ops bucket arms the timer from its own refill rate. + let mut l = RateLimiter::new(0, 0, 0, 1000, 0, 1000).unwrap(); + l.set_min_refill_delay(min_delay); + assert!(l.consume(1000, TokenType::Ops)); + assert!(!l.consume(20, TokenType::Ops)); + let delay = armed_delay(&l); + assert!(delay <= Duration::from_millis(20), "{delay:?}"); + assert!(delay > Duration::from_millis(15), "{delay:?}"); + + // Without a minimum refill delay, the fixed interval is kept. + let mut l = RateLimiter::new(1000, 0, 1000, 0, 0, 0).unwrap(); + assert!(l.consume(1000, TokenType::Bytes)); + assert!(!l.consume(1, TokenType::Bytes)); + let delay = armed_delay(&l); + assert!(delay <= max_delay, "{delay:?}"); + assert!(delay > max_delay - Duration::from_millis(5), "{delay:?}"); + } + #[test] fn test_rate_limiter_bandwidth() { // rate limiter with limit of 1000 bytes/s diff --git a/src/vmm/src/rate_limiter/persist.rs b/src/vmm/src/rate_limiter/persist.rs index 6c9e5052ecf..ee400a52850 100644 --- a/src/vmm/src/rate_limiter/persist.rs +++ b/src/vmm/src/rate_limiter/persist.rs @@ -84,6 +84,8 @@ impl Persist<'_> for RateLimiter { }, timer_fd: TimerFd::new_custom(ClockId::Monotonic, true, true)?, timer_active: false, + // Device policy, not limiter state: the owning device sets it after restore. + min_refill_delay: None, }; Ok(rate_limiter) From 366826857b6a94e6a9433c85990f8d4a2fb71743 Mon Sep 17 00:00:00 2001 From: Tom Date: Tue, 29 Sep 2026 14:15:21 -0700 Subject: [PATCH 3/3] seccomp: allow cloning threaded drive backing files (#19) Updating a Threaded drive clones its backing file on the VMM thread. The Rust clone path uses F_DUPFD_CLOEXEC, which the VMM seccomp policy did not allow and caused Firecracker to exit with BadSyscall. Allow that fcntl command in both architecture policies and exercise file rescan with the Threaded engine. Signed-off-by: Tom Ryan <299368098+tomr-fly@users.noreply.github.com> Co-authored-by: Tom Ryan <299368098+tomr-fly@users.noreply.github.com> --- resources/seccomp/aarch64-unknown-linux-musl.json | 13 +++++++++++++ resources/seccomp/x86_64-unknown-linux-musl.json | 13 +++++++++++++ .../functional/test_drive_virtio.py | 1 + 3 files changed, 27 insertions(+) diff --git a/resources/seccomp/aarch64-unknown-linux-musl.json b/resources/seccomp/aarch64-unknown-linux-musl.json index 0e105b0d9c9..2347b4754a1 100644 --- a/resources/seccomp/aarch64-unknown-linux-musl.json +++ b/resources/seccomp/aarch64-unknown-linux-musl.json @@ -146,6 +146,19 @@ } ] }, + { + "syscall": "fcntl", + "comment": "Used to duplicate the backing file when updating a Threaded drive", + "args": [ + { + "index": 1, + "type": "dword", + "op": "eq", + "val": 1030, + "comment": "FCNTL_F_DUPFD_CLOEXEC" + } + ] + }, { "syscall": "futex", "comment": "Used for synchronization (during thread teardown when joining multiple vcpu threads at once)", diff --git a/resources/seccomp/x86_64-unknown-linux-musl.json b/resources/seccomp/x86_64-unknown-linux-musl.json index c742f6e6f0c..b9bbbd79f6c 100644 --- a/resources/seccomp/x86_64-unknown-linux-musl.json +++ b/resources/seccomp/x86_64-unknown-linux-musl.json @@ -146,6 +146,19 @@ } ] }, + { + "syscall": "fcntl", + "comment": "Used to duplicate the backing file when updating a Threaded drive", + "args": [ + { + "index": 1, + "type": "dword", + "op": "eq", + "val": 1030, + "comment": "FCNTL_F_DUPFD_CLOEXEC" + } + ] + }, { "syscall": "futex", "comment": "Used for synchronization (during thread teardown when joining multiple vcpu threads at once)", diff --git a/tests/integration_tests/functional/test_drive_virtio.py b/tests/integration_tests/functional/test_drive_virtio.py index 9c61ead56a9..76796eef83e 100644 --- a/tests/integration_tests/functional/test_drive_virtio.py +++ b/tests/integration_tests/functional/test_drive_virtio.py @@ -23,6 +23,7 @@ def partuuid_and_disk_path_tmpfs(rootfs, tmp_path): disk_path.unlink() +@pytest.mark.parametrize("io_engine", ["Sync", "Async", "Threaded"], indirect=True) def test_rescan_file(uvm_plain_any, io_engine): """ Verify that rescan works with a file-backed virtio device.