diff --git a/src/vmm/src/devices/virtio/block/virtio/device.rs b/src/vmm/src/devices/virtio/block/virtio/device.rs index eb255164e0c..6583a32dec8 100644 --- a/src/vmm/src/devices/virtio/block/virtio/device.rs +++ b/src/vmm/src/devices/virtio/block/virtio/device.rs @@ -427,9 +427,7 @@ impl VirtioBlock { BlockState::Active(ActiveBlock::Inline(worker)) => { worker.process_virtio_queues().unwrap(); } - BlockState::Active(ActiveBlock::Threaded(_)) => { - unreachable!("worker control messages are not connected yet") - } + BlockState::Active(ActiveBlock::Threaded(active)) => active.worker_handle.kick(), BlockState::Configuring(_, _) => {} BlockState::Placeholder => unreachable!("not a runtime state"), } @@ -453,9 +451,9 @@ impl VirtioBlock { 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::Active(ActiveBlock::Threaded(active)) => active + .worker_handle + .update_disk_image(disk_image_path, read_only)?, BlockState::Placeholder => unreachable!("not a runtime state"), }; self.config_space.capacity = nsectors.to_le(); diff --git a/src/vmm/src/devices/virtio/block/virtio/mod.rs b/src/vmm/src/devices/virtio/block/virtio/mod.rs index 863e202b1af..3be79dbac24 100644 --- a/src/vmm/src/devices/virtio/block/virtio/mod.rs +++ b/src/vmm/src/devices/virtio/block/virtio/mod.rs @@ -66,4 +66,6 @@ pub enum VirtioBlockError { Persist(crate::devices::virtio::persist::PersistError), /// Error spawning the block worker thread: {0} ThreadSpawn(std::io::Error), + /// Error communicating with the block worker thread: {0} + WorkerControl(String), } diff --git a/src/vmm/src/devices/virtio/block/virtio/worker.rs b/src/vmm/src/devices/virtio/block/virtio/worker.rs index db1075fcd9f..af254f4e0c6 100644 --- a/src/vmm/src/devices/virtio/block/virtio/worker.rs +++ b/src/vmm/src/devices/virtio/block/virtio/worker.rs @@ -51,12 +51,15 @@ enum WorkerState { #[allow(clippy::large_enum_variant)] enum ControlMsg { Start(BlockWorker), + UpdateDiskImage { path: String, read_only: bool }, + Kick, Reset, Finish(FlushMode), } #[allow(clippy::large_enum_variant)] enum ControlResponse { + DiskUpdated(Result), // returns nsectors on success Reset(Option), } @@ -360,15 +363,70 @@ impl WorkerHandle { return None; } - match self.from_worker.recv() { - Ok(ControlResponse::Reset(resources)) => resources, - Err(err) => { - error!("Block worker failed to acknowledge reset: {:?}", err); - None + loop { + match self.from_worker.recv() { + Ok(ControlResponse::Reset(resources)) => return resources, + Ok(ControlResponse::DiskUpdated(_)) => { + warn!("Ignoring disk update response while waiting for reset"); + } + Err(err) => { + error!("Block worker failed to acknowledge reset: {:?}", err); + return None; + } + } + } + } + + /// Replace the worker's backing file and return its new sector count. + pub(crate) fn update_disk_image( + &self, + disk_image_path: String, + read_only: bool, + ) -> Result { + if let Err(err) = self.to_worker.send(ControlMsg::UpdateDiskImage { + path: disk_image_path, + read_only, + }) { + error!("Failed to send block worker disk update: {:?}", err); + return Err(VirtioBlockError::WorkerControl(format!( + "failed to send disk update: {err}" + ))); + } + + if let Err(err) = self.control_evt.write(1) { + error!("Failed to notify block worker of disk update: {:?}", err); + return Err(VirtioBlockError::WorkerControl(format!( + "failed to notify worker of disk update: {err}" + ))); + } + + loop { + match self.from_worker.recv() { + Ok(ControlResponse::DiskUpdated(result)) => return result, + Ok(ControlResponse::Reset(_)) => { + warn!("Ignoring reset response while waiting for disk update"); + } + Err(err) => { + error!("Block worker failed to acknowledge disk update: {:?}", err); + return Err(VirtioBlockError::WorkerControl(format!( + "failed to receive disk update response: {err}" + ))); + } } } } + pub(crate) fn kick(&self) { + if let Err(err) = self.to_worker.send(ControlMsg::Kick) { + error!("Failed to send block worker kick: {:?}", err); + return; + } + + if let Err(err) = self.control_evt.write(1) { + error!("Failed to notify block worker of kick: {:?}", err); + } + } + /// 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)) { @@ -450,6 +508,10 @@ impl ThreadedWorker { while let Ok(msg) = self.from_vmm.try_recv() { match msg { ControlMsg::Start(worker) => self.start_worker(worker, ops), + ControlMsg::UpdateDiskImage { path, read_only } => { + self.update_disk_image(path, read_only) + } + ControlMsg::Kick => self.kick_worker(), ControlMsg::Reset => self.reset_worker(ops), ControlMsg::Finish(flush_mode) => self.finish_worker(flush_mode, ops), } @@ -470,6 +532,51 @@ impl ThreadedWorker { self.state = WorkerState::Running(worker); } + fn update_disk_image(&mut self, path: String, read_only: bool) { + let result = match &mut self.state { + WorkerState::Running(worker) => worker.update_disk_image(path, read_only), + WorkerState::Parked => { + warn!("Disk image update requested while block worker is parked"); + Err(VirtioBlockError::WorkerControl( + "disk update requested while worker is parked".to_string(), + )) + } + WorkerState::Finished => { + warn!("Disk image update requested after block worker finished"); + Err(VirtioBlockError::WorkerControl( + "disk update requested after worker finished".to_string(), + )) + } + }; + + if let Err(err) = self.to_vmm.send(ControlResponse::DiskUpdated(result)) { + error!( + "Failed to send block worker disk update response: {:?}", + err + ); + } + } + + fn kick_worker(&mut self) { + match std::mem::replace(&mut self.state, WorkerState::Parked) { + WorkerState::Running(worker) => { + self.state = WorkerState::Running(worker); + } + state => { + warn!("Kick requested while block worker is not active"); + self.state = state; + return; + } + } + + // process directly instead of going through epoll + if let WorkerState::Running(worker) = &mut self.state { + worker + .process_virtio_queues() + .unwrap_or_else(|err| error!("Failed to kick block worker queue: {:?}", err)); + } + } + fn reset_worker(&mut self, ops: &mut EventOps) { let resources = match std::mem::replace(&mut self.state, WorkerState::Parked) { WorkerState::Running(mut worker) => {