diff --git a/resources/seccomp/aarch64-unknown-linux-musl.json b/resources/seccomp/aarch64-unknown-linux-musl.json index 26dd661e46b..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)", @@ -1094,5 +1107,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..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)", @@ -1226,5 +1239,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..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}; @@ -50,6 +53,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 +277,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; } @@ -291,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); @@ -560,6 +566,7 @@ impl VirtioBlock { match self.disk.file_engine { FileEngine::Sync(_) => FileEngineType::Sync, FileEngine::Async(_) => FileEngineType::Async, + FileEngine::Threaded(_) => FileEngineType::Threaded, } } @@ -579,6 +586,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..fc0985c4110 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; @@ -29,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 380fe1de0e8..cde43b0fecf 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, } } } @@ -87,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(), @@ -239,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/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(_))); + } +} 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) 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.