diff --git a/src/vmm/src/devices/virtio/block/virtio/device.rs b/src/vmm/src/devices/virtio/block/virtio/device.rs index 690483007f9..eb255164e0c 100644 --- a/src/vmm/src/devices/virtio/block/virtio/device.rs +++ b/src/vmm/src/devices/virtio/block/virtio/device.rs @@ -682,41 +682,45 @@ impl VirtioDevice for VirtioBlock { } fn deactivate(&mut self) { - let state = std::mem::replace(&mut self.state, BlockState::Placeholder); - self.state = match state { - BlockState::Active(ActiveBlock::Inline(worker)) => { - BlockState::Configuring(worker.resources, None) - } - state => state, - }; + // `_reset` moves data-path resources back into the configuring state. } fn _reset(&mut self) -> bool { - match &mut self.state { - BlockState::Active(ActiveBlock::Inline(worker)) => worker.reset(), - BlockState::Active(ActiveBlock::Threaded(_)) => false, - BlockState::Configuring(resources, _) => { + let state = std::mem::replace(&mut self.state, BlockState::Placeholder); + let (resources, worker_handle) = match state { + BlockState::Active(ActiveBlock::Threaded(active)) => { + let Some(resources) = active.worker_handle.reset() else { + self.state = BlockState::Active(ActiveBlock::Threaded(active)); + return false; + }; + (resources, Some(active.worker_handle)) + } + BlockState::Active(ActiveBlock::Inline(mut worker)) => { + if !worker.reset() { + self.state = BlockState::Active(ActiveBlock::Inline(worker)); + return false; + } + (worker.resources, None) + } + BlockState::Configuring(mut resources, worker_handle) => { if let Err(err) = resources.disk.file_engine.drain(true) { error!("Failed to reset block IO engine: {:?}", err); + self.state = BlockState::Configuring(resources, worker_handle); return false; } resources.is_io_engine_throttled = false; - true + (resources, worker_handle) } BlockState::Placeholder => unreachable!("not a runtime state"), - } + }; + + self.state = BlockState::Configuring(resources, worker_handle); + true } fn reset_queues(&mut self) { - 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"), + if let BlockState::Configuring(resources, _) = &mut self.state { + resources.queues.iter_mut().for_each(Queue::reset); } } @@ -751,8 +755,9 @@ impl Drop for VirtioBlock { FlushMode::Drain => worker.drain(true), FlushMode::DrainAndFlush => worker.drain_and_flush(true), }, - // Worker teardown is connected in the follow-up lifecycle commit. - BlockState::Active(ActiveBlock::Threaded(_)) => {} + BlockState::Active(ActiveBlock::Threaded(active)) => { + active.worker_handle.finish(flush_mode); + } BlockState::Configuring(mut resources, worker_handle) => { match flush_mode { FlushMode::Drain => { @@ -1940,4 +1945,38 @@ mod tests { assert_eq!(block.disk().image_id, id.as_slice()); } } + + #[test] + fn test_reset_and_reactivation() { + for engine in [FileEngineType::Sync, FileEngineType::Async] { + for threaded in [false, true] { + let mut block = default_block(engine); + if threaded { + block.spawn_worker().unwrap(); + } + + let mem = default_mem(); + let vq = VirtQueue::new(GuestAddress(0), &mem, 16); + set_queue(&mut block, 0, vq.create_queue()); + block.set_acked_features(1); + block.activate(mem.clone(), default_interrupt()).unwrap(); + + assert!(block.is_activated()); + assert_eq!(block.is_threaded_active(), threaded); + assert!(block.reset()); + assert!(!block.is_activated()); + assert_eq!(block.acked_features(), 0); + assert!(!block.queue_config(0).unwrap().ready); + let BlockState::Configuring(_, worker_handle) = &block.state else { + panic!("reset must leave the block device configuring"); + }; + assert_eq!(worker_handle.is_some(), threaded); + + set_queue(&mut block, 0, vq.create_queue()); + block.activate(mem, default_interrupt()).unwrap(); + assert!(block.is_activated()); + assert_eq!(block.is_threaded_active(), threaded); + } + } + } } diff --git a/src/vmm/src/devices/virtio/block/virtio/worker.rs b/src/vmm/src/devices/virtio/block/virtio/worker.rs index aed95be5c85..db1075fcd9f 100644 --- a/src/vmm/src/devices/virtio/block/virtio/worker.rs +++ b/src/vmm/src/devices/virtio/block/virtio/worker.rs @@ -36,6 +36,7 @@ struct ThreadedWorker { state: WorkerState, control_evt: EventFd, from_vmm: Receiver, + to_vmm: Sender, } /// Data-path ownership state of the worker thread. @@ -50,13 +51,20 @@ enum WorkerState { #[allow(clippy::large_enum_variant)] enum ControlMsg { Start(BlockWorker), + Reset, Finish(FlushMode), } +#[allow(clippy::large_enum_variant)] +enum ControlResponse { + Reset(Option), +} + /// VMM-side handle for controlling and joining a block worker thread. #[derive(Debug)] pub(crate) struct WorkerHandle { to_worker: Sender, + from_worker: Receiver, control_evt: EventFd, join: JoinHandle<()>, queue_evts: Vec, @@ -299,16 +307,19 @@ impl WorkerHandle { // handle writes and worker reads the control eventfd let control_evt = EventFd::new(libc::EFD_NONBLOCK)?; let handle_evt = control_evt.try_clone()?; + let (to_worker, from_vmm) = channel::(); + let (to_vmm, from_worker) = channel::(); let join = thread::Builder::new().name(name).spawn(move || { let event_manager = EventManager::new().expect("Failed to create block worker EventManager"); - run_worker_loop(event_manager, control_evt, from_vmm); + run_worker_loop(event_manager, control_evt, from_vmm, to_vmm); })?; Ok(Self { to_worker, + from_worker, control_evt: handle_evt, join, queue_evts, @@ -331,11 +342,40 @@ impl WorkerHandle { } } + /// Stop processing and return the data-path resources to the VMM thread. + pub(crate) fn reset(&self) -> Option { + if let Err(err) = self.to_worker.send(ControlMsg::Reset) { + error!( + "Block worker receiver already dropped during reset: {:?}", + err + ); + return None; + } + + if let Err(err) = self.control_evt.write(1) { + error!( + "Block worker control event is closed during reset: {:?}", + err + ); + return None; + } + + match self.from_worker.recv() { + Ok(ControlResponse::Reset(resources)) => resources, + Err(err) => { + error!("Block worker failed to acknowledge reset: {:?}", err); + None + } + } + } + /// Stop the worker and wait for its thread to exit. pub(crate) fn finish(self, flush_mode: FlushMode) { if let Err(err) = self.to_worker.send(ControlMsg::Finish(flush_mode)) { error!("Block worker receiver already dropped: {:?}", err); - } else if let Err(err) = self.control_evt.write(1) { + } + + if let Err(err) = self.control_evt.write(1) { error!("Block worker control event is closed: {:?}", err); } @@ -410,6 +450,7 @@ impl ThreadedWorker { while let Ok(msg) = self.from_vmm.try_recv() { match msg { ControlMsg::Start(worker) => self.start_worker(worker, ops), + ControlMsg::Reset => self.reset_worker(ops), ControlMsg::Finish(flush_mode) => self.finish_worker(flush_mode, ops), } @@ -429,6 +470,33 @@ impl ThreadedWorker { self.state = WorkerState::Running(worker); } + fn reset_worker(&mut self, ops: &mut EventOps) { + let resources = match std::mem::replace(&mut self.state, WorkerState::Parked) { + WorkerState::Running(mut worker) => { + Self::unregister_runtime_events(&worker.resources, ops); + if worker.reset() { + Some(worker.resources) + } else { + Self::register_runtime_events(&worker.resources, ops); + self.state = WorkerState::Running(worker); + None + } + } + WorkerState::Parked => { + warn!("Reset requested while block worker is parked"); + None + } + WorkerState::Finished => { + self.state = WorkerState::Finished; + None + } + }; + + if let Err(err) = self.to_vmm.send(ControlResponse::Reset(resources)) { + error!("Failed to send block worker reset response: {:?}", err); + } + } + fn finish_worker(&mut self, flush_mode: FlushMode, ops: &mut EventOps) { if let WorkerState::Running(mut worker) = std::mem::replace(&mut self.state, WorkerState::Finished) @@ -451,11 +519,13 @@ fn run_worker_loop( mut event_manager: EventManager, control_evt: EventFd, from_vmm: Receiver, + to_vmm: Sender, ) { let worker = Arc::new(Mutex::new(ThreadedWorker { state: WorkerState::Parked, control_evt, from_vmm, + to_vmm, })); let subscriber: Arc> = worker.clone(); event_manager.add_subscriber(subscriber);