diff --git a/src/vmm/src/devices/virtio/block/virtio/device.rs b/src/vmm/src/devices/virtio/block/virtio/device.rs index 5486b873ae7..690483007f9 100644 --- a/src/vmm/src/devices/virtio/block/virtio/device.rs +++ b/src/vmm/src/devices/virtio/block/virtio/device.rs @@ -19,7 +19,7 @@ use vm_memory::ByteValued; use vmm_sys_util::eventfd::EventFd; use super::io::FileEngine; -use super::worker::{BlockWorker, FlushMode}; +use super::worker::{BlockWorker, FlushMode, WorkerHandle}; use super::{BLOCK_QUEUE_SIZES, SECTOR_SHIFT, SECTOR_SIZE, VirtioBlockError}; use crate::devices::virtio::ActivateError; use crate::devices::virtio::block::CacheType; @@ -256,13 +256,32 @@ pub struct VirtioBlock { #[derive(Debug)] pub(crate) enum BlockState { // Vmm owns the runtime resources before activation - Configuring(BlockResources), + // In threaded mode, it also owns a parked worker thread + Configuring(BlockResources, Option), // Active state, worker owns data-path resources - Active(BlockWorker), + Active(ActiveBlock), // Placeholder to hold state when activating Placeholder, } +/// Active block data path. +#[derive(Debug)] +#[allow(clippy::large_enum_variant)] +pub(crate) enum ActiveBlock { + // Data path on VMM thread (single thread) + Inline(BlockWorker), + // Data path on a dedicated worker thread (multi thread) + Threaded(ThreadedActive), +} + +/// VMM-side state when data path runs on a worker thread +#[derive(Debug)] +pub(crate) struct ThreadedActive { + worker_handle: WorkerHandle, + interrupt: Arc, + queue_config: Vec, +} + /// Runtime resources used by the block data path. #[derive(Debug)] pub(crate) struct BlockResources { @@ -321,12 +340,17 @@ impl VirtioBlock { config, rate_limiter: Arc::new(Mutex::new(rate_limiter)), - state: BlockState::Configuring(BlockResources { - queues: BLOCK_QUEUE_SIZES.iter().map(|&s| Queue::new(s)).collect(), - queue_evts: [EventFd::new(libc::EFD_NONBLOCK).map_err(VirtioBlockError::EventFd)?], - disk: disk_properties, - is_io_engine_throttled: false, - }), + state: BlockState::Configuring( + BlockResources { + queues: BLOCK_QUEUE_SIZES.iter().map(|&s| Queue::new(s)).collect(), + queue_evts: [ + EventFd::new(libc::EFD_NONBLOCK).map_err(VirtioBlockError::EventFd)? + ], + disk: disk_properties, + is_io_engine_throttled: false, + }, + None, + ), metrics, }) } @@ -338,16 +362,22 @@ impl VirtioBlock { pub(crate) fn resources(&self) -> &BlockResources { match &self.state { - BlockState::Configuring(resources) => resources, - BlockState::Active(worker) => &worker.resources, + BlockState::Configuring(resources, _) => resources, + BlockState::Active(ActiveBlock::Inline(worker)) => &worker.resources, + BlockState::Active(ActiveBlock::Threaded(_)) => { + unreachable!("worker thread owns the runtime resources") + } BlockState::Placeholder => unreachable!("not a runtime state"), } } pub(crate) fn resources_mut(&mut self) -> &mut BlockResources { match &mut self.state { - BlockState::Configuring(resources) => resources, - BlockState::Active(worker) => &mut worker.resources, + BlockState::Configuring(resources, _) => resources, + BlockState::Active(ActiveBlock::Inline(worker)) => &mut worker.resources, + BlockState::Active(ActiveBlock::Threaded(_)) => { + unreachable!("worker thread owns the runtime resources") + } BlockState::Placeholder => unreachable!("not a runtime state"), } } @@ -362,12 +392,8 @@ impl VirtioBlock { .expect("Poisoned block rate limiter lock") } - fn worker_mut(&mut self) -> &mut BlockWorker { - match &mut self.state { - BlockState::Active(worker) => worker, - BlockState::Configuring(_) => panic!("Device is not activated"), - BlockState::Placeholder => unreachable!("not a runtime state"), - } + pub(crate) fn is_threaded_active(&self) -> bool { + matches!(self.state, BlockState::Active(ActiveBlock::Threaded(_))) } /// Process a single event in the Virtio queue. @@ -375,12 +401,18 @@ impl VirtioBlock { /// This function is called by the event manager when the guest notifies us /// about new buffers in the queue. pub(crate) fn process_queue_event(&mut self) { - self.worker_mut().process_queue_event(); + if let BlockState::Active(ActiveBlock::Inline(worker)) = &mut self.state { + worker.process_queue_event(); + } } /// Process device virtio queue(s). pub fn process_virtio_queues(&mut self) -> Result<(), InvalidAvailIdx> { - self.worker_mut().process_virtio_queues() + if let BlockState::Active(ActiveBlock::Inline(worker)) = &mut self.state { + worker.process_virtio_queues() + } else { + Ok(()) + } } pub(crate) fn process_rate_limiter_event(&mut self) { @@ -392,16 +424,21 @@ impl VirtioBlock { // Upon rate limiter event, call the rate limiter handler // and restart processing the queue. match &mut self.state { - BlockState::Active(worker) => { + BlockState::Active(ActiveBlock::Inline(worker)) => { worker.process_virtio_queues().unwrap(); } - BlockState::Configuring(_) => {} + BlockState::Active(ActiveBlock::Threaded(_)) => { + unreachable!("worker control messages are not connected yet") + } + BlockState::Configuring(_, _) => {} BlockState::Placeholder => unreachable!("not a runtime state"), } } pub fn process_async_completion_event(&mut self) { - self.worker_mut().process_async_completion_event(); + if let BlockState::Active(ActiveBlock::Inline(worker)) = &mut self.state { + worker.process_async_completion_event(); + } } /// Update the backing file and the config space of the block device. @@ -409,11 +446,16 @@ impl VirtioBlock { let config_path = disk_image_path.clone(); let read_only = self.config.is_read_only; let nsectors = match &mut self.state { - BlockState::Configuring(resources) => { + BlockState::Configuring(resources, _) => { resources.disk.update(disk_image_path, read_only)?; resources.disk.nsectors } - BlockState::Active(worker) => worker.update_disk_image(disk_image_path, read_only)?, + BlockState::Active(ActiveBlock::Inline(worker)) => { + worker.update_disk_image(disk_image_path, read_only)? + } + BlockState::Active(ActiveBlock::Threaded(_)) => { + unreachable!("worker control messages are not connected yet") + } BlockState::Placeholder => unreachable!("not a runtime state"), }; self.config_space.capacity = nsectors.to_le(); @@ -447,10 +489,30 @@ impl VirtioBlock { /// Prepare device for being snapshotted. pub fn prepare_save(&mut self) { - if let BlockState::Active(worker) = &mut self.state { + if let BlockState::Active(ActiveBlock::Inline(worker)) = &mut self.state { worker.prepare_save(); } } + + /// Spawn a parked worker thread for the next activation. + // Currently unused because threaded mode is not exposed through device configuration yet. + #[allow(dead_code)] + pub(crate) fn spawn_worker(&mut self) -> Result<(), VirtioBlockError> { + if let BlockState::Configuring(resources, worker_handle @ None) = &mut self.state { + let queue_evts = resources + .queue_evts + .iter() + .map(EventFd::try_clone) + .collect::, _>>() + .map_err(VirtioBlockError::EventFd)?; + + let name = format!("fc_{}", self.config.drive_id); + + *worker_handle = + Some(WorkerHandle::spawn(queue_evts, name).map_err(VirtioBlockError::ThreadSpawn)?); + } + Ok(()) + } } impl VirtioDevice for VirtioBlock { @@ -473,31 +535,65 @@ impl VirtioDevice for VirtioBlock { } fn num_queues(&self) -> usize { - self.resources().queues.len() + match &self.state { + BlockState::Configuring(resources, _) => resources.queues.len(), + BlockState::Active(ActiveBlock::Inline(worker)) => worker.resources.queues.len(), + BlockState::Active(ActiveBlock::Threaded(active)) => active.queue_config.len(), + BlockState::Placeholder => unreachable!("not a runtime state"), + } } fn queue_config(&self, index: usize) -> Option<&QueueConfig> { - self.resources() - .queues - .get(index) - .map(|queue| &queue.config) + match &self.state { + BlockState::Configuring(resources, _) => { + resources.queues.get(index).map(|queue| &queue.config) + } + BlockState::Active(ActiveBlock::Inline(worker)) => worker + .resources + .queues + .get(index) + .map(|queue| &queue.config), + BlockState::Active(ActiveBlock::Threaded(active)) => active.queue_config.get(index), + BlockState::Placeholder => unreachable!("not a runtime state"), + } } fn queue_config_mut(&mut self, index: usize) -> Option<&mut QueueConfig> { - self.resources_mut() - .queues - .get_mut(index) - .map(|queue| &mut queue.config) + match &mut self.state { + BlockState::Configuring(resources, _) => resources + .queues + .get_mut(index) + .map(|queue| &mut queue.config), + BlockState::Active(ActiveBlock::Inline(worker)) => worker + .resources + .queues + .get_mut(index) + .map(|queue| &mut queue.config), + BlockState::Active(ActiveBlock::Threaded(active)) => active.queue_config.get_mut(index), + BlockState::Placeholder => unreachable!("not a runtime state"), + } } fn queue_event(&self, index: usize) -> Option<&EventFd> { - self.resources().queue_evts.get(index) + match &self.state { + BlockState::Configuring(resources, _) => resources.queue_evts.get(index), + BlockState::Active(ActiveBlock::Inline(worker)) => { + worker.resources.queue_evts.get(index) + } + BlockState::Active(ActiveBlock::Threaded(active)) => { + active.worker_handle.queue_events().get(index) + } + BlockState::Placeholder => unreachable!("not a runtime state"), + } } fn interrupt_trigger(&self) -> &dyn VirtioInterrupt { match &self.state { - BlockState::Active(worker) => worker.active_state.interrupt.deref(), - BlockState::Configuring(_) => panic!("Device is not initialized"), + BlockState::Active(ActiveBlock::Inline(worker)) => { + worker.active_state.interrupt.deref() + } + BlockState::Active(ActiveBlock::Threaded(active)) => active.interrupt.deref(), + BlockState::Configuring(_, _) => panic!("Device is not initialized"), BlockState::Placeholder => unreachable!("not a runtime state"), } } @@ -524,7 +620,7 @@ impl VirtioDevice for VirtioBlock { let event_idx = self.has_feature(u64::from(VIRTIO_RING_F_EVENT_IDX)); - let BlockState::Configuring(resources) = &mut self.state else { + let BlockState::Configuring(resources, _) = &mut self.state else { unreachable!("inactive device is not configurable"); }; @@ -544,19 +640,40 @@ impl VirtioDevice for VirtioBlock { return Err(ActivateError::EventFd); } - let BlockState::Configuring(resources) = + let BlockState::Configuring(resources, worker_handle) = std::mem::replace(&mut self.state, BlockState::Placeholder) else { unreachable!("state checked before activation"); }; + + let queue_config = resources + .queues + .iter() + .map(|queue| queue.config.clone()) + .collect(); + let is_blocked = self.rate_limiter().clone_blocked_flag(); - self.state = BlockState::Active(BlockWorker { + let worker = BlockWorker { resources, rate_limiter: self.rate_limiter.clone(), is_blocked, - active_state: ActiveState { mem, interrupt }, + active_state: ActiveState { + mem, + interrupt: interrupt.clone(), + }, metrics: self.metrics.clone(), - }); + }; + + self.state = if let Some(worker_handle) = worker_handle { + worker_handle.start(worker); + BlockState::Active(ActiveBlock::Threaded(ThreadedActive { + worker_handle, + interrupt, + queue_config, + })) + } else { + BlockState::Active(ActiveBlock::Inline(worker)) + }; Ok(()) } @@ -567,15 +684,18 @@ impl VirtioDevice for VirtioBlock { fn deactivate(&mut self) { let state = std::mem::replace(&mut self.state, BlockState::Placeholder); self.state = match state { - BlockState::Active(worker) => BlockState::Configuring(worker.resources), + BlockState::Active(ActiveBlock::Inline(worker)) => { + BlockState::Configuring(worker.resources, None) + } state => state, }; } fn _reset(&mut self) -> bool { match &mut self.state { - BlockState::Active(worker) => worker.reset(), - BlockState::Configuring(resources) => { + BlockState::Active(ActiveBlock::Inline(worker)) => worker.reset(), + BlockState::Active(ActiveBlock::Threaded(_)) => false, + BlockState::Configuring(resources, _) => { if let Err(err) = resources.disk.file_engine.drain(true) { error!("Failed to reset block IO engine: {:?}", err); return false; @@ -588,15 +708,32 @@ impl VirtioDevice for VirtioBlock { } fn reset_queues(&mut self) { - self.resources_mut() - .queues - .iter_mut() - .for_each(Queue::reset); + match &mut self.state { + BlockState::Configuring(resources, _) => { + resources.queues.iter_mut().for_each(Queue::reset); + } + BlockState::Active(ActiveBlock::Inline(worker)) => { + worker.resources.queues.iter_mut().for_each(Queue::reset); + } + BlockState::Active(ActiveBlock::Threaded(_)) => {} + BlockState::Placeholder => unreachable!("not a runtime state"), + } } fn mark_queue_memory_dirty(&mut self, mem: &GuestMemoryMmap) -> Result<(), QueueError> { - for queue in &mut self.resources_mut().queues { - queue.initialize(mem)?; + match &mut self.state { + BlockState::Configuring(resources, _) => { + for queue in &mut resources.queues { + queue.initialize(mem)?; + } + } + BlockState::Active(ActiveBlock::Inline(worker)) => { + for queue in &mut worker.resources.queues { + queue.initialize(mem)?; + } + } + BlockState::Active(ActiveBlock::Threaded(_)) => {} + BlockState::Placeholder => unreachable!("not a runtime state"), } Ok(()) } @@ -610,22 +747,30 @@ impl Drop for VirtioBlock { }; match std::mem::replace(&mut self.state, BlockState::Placeholder) { - BlockState::Active(mut worker) => match flush_mode { + BlockState::Active(ActiveBlock::Inline(mut worker)) => match flush_mode { FlushMode::Drain => worker.drain(true), FlushMode::DrainAndFlush => worker.drain_and_flush(true), }, - BlockState::Configuring(mut resources) => match flush_mode { - FlushMode::Drain => { - if let Err(err) = resources.disk.file_engine.drain(true) { - error!("Failed to drain ops on drop: {:?}", err); + // Worker teardown is connected in the follow-up lifecycle commit. + BlockState::Active(ActiveBlock::Threaded(_)) => {} + BlockState::Configuring(mut resources, worker_handle) => { + match flush_mode { + FlushMode::Drain => { + if let Err(err) = resources.disk.file_engine.drain(true) { + error!("Failed to drain ops on drop: {:?}", err); + } } - } - FlushMode::DrainAndFlush => { - if let Err(err) = resources.disk.file_engine.drain_and_flush(true) { - error!("Failed to drain ops and flush block data: {:?}", err); + FlushMode::DrainAndFlush => { + if let Err(err) = resources.disk.file_engine.drain_and_flush(true) { + error!("Failed to drain ops and flush block data: {:?}", err); + } } } - }, + if let Some(worker_handle) = worker_handle { + // a parked worker owns no runtime resources, it just joins the thread + worker_handle.finish(FlushMode::Drain); + } + } BlockState::Placeholder => {} }; } 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 30810318f63..e11e92ec55b 100644 --- a/src/vmm/src/devices/virtio/block/virtio/event_handler.rs +++ b/src/vmm/src/devices/virtio/block/virtio/event_handler.rs @@ -57,8 +57,12 @@ impl VirtioBlock { if let Err(err) = self.activate_evt.read() { error!("Failed to consume block activate event: {:?}", err); } + self.register_rate_limiter_event(ops); - self.register_runtime_events(ops); + if !self.is_threaded_active() { + self.register_runtime_events(ops); + } + if let Err(err) = ops.remove(Events::with_data( &self.activate_evt, Self::PROCESS_ACTIVATE, @@ -121,7 +125,9 @@ impl MutEventSubscriber for VirtioBlock { // - on device restore from snapshot. if self.is_activated() { self.register_rate_limiter_event(ops); - self.register_runtime_events(ops); + if !self.is_threaded_active() { + self.register_runtime_events(ops); + } } else { self.register_activate_event(ops); } diff --git a/src/vmm/src/devices/virtio/block/virtio/mod.rs b/src/vmm/src/devices/virtio/block/virtio/mod.rs index b7d95f8c321..863e202b1af 100644 --- a/src/vmm/src/devices/virtio/block/virtio/mod.rs +++ b/src/vmm/src/devices/virtio/block/virtio/mod.rs @@ -64,4 +64,6 @@ pub enum VirtioBlockError { RateLimiter(std::io::Error), /// Persistence error: {0} Persist(crate::devices::virtio::persist::PersistError), + /// Error spawning the block worker thread: {0} + ThreadSpawn(std::io::Error), } diff --git a/src/vmm/src/devices/virtio/block/virtio/persist.rs b/src/vmm/src/devices/virtio/block/virtio/persist.rs index 5a31ec04ed3..2c581238c6e 100644 --- a/src/vmm/src/devices/virtio/block/virtio/persist.rs +++ b/src/vmm/src/devices/virtio/block/virtio/persist.rs @@ -141,7 +141,7 @@ impl Persist<'_> for VirtioBlock { config, rate_limiter: Arc::new(Mutex::new(rate_limiter)), - state: BlockState::Configuring(resources), + state: BlockState::Configuring(resources, None), metrics: BlockMetricsPerDevice::alloc(state.id.clone()), }) } diff --git a/src/vmm/src/devices/virtio/block/virtio/worker.rs b/src/vmm/src/devices/virtio/block/virtio/worker.rs index 8995c22b9eb..aed95be5c85 100644 --- a/src/vmm/src/devices/virtio/block/virtio/worker.rs +++ b/src/vmm/src/devices/virtio/block/virtio/worker.rs @@ -54,8 +54,6 @@ enum ControlMsg { } /// VMM-side handle for controlling and joining a block worker thread. -// Handle gets connected to VirtioBlock in follow-up commits -#[allow(dead_code)] #[derive(Debug)] pub(crate) struct WorkerHandle { to_worker: Sender, @@ -295,8 +293,6 @@ impl BlockWorker { } } -// Handle gets connected to VirtioBlock in follow-up commits -#[allow(dead_code)] impl WorkerHandle { /// Spawn a parked block worker thread. pub(crate) fn spawn(queue_evts: Vec, name: String) -> Result {