[blk-threaded - 6/12] block: add worker thread reset and teardown

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);
+            }
+        }
+    }
 }

src/vmm/src/devices/virtio/block/virtio/worker.rs

@@ -36,6 +36,7 @@ struct ThreadedWorker {
     state: WorkerState,
     control_evt: EventFd,
     from_vmm: Receiver<ControlMsg>,
+    to_vmm: Sender<ControlResponse>,
 }
 
 /// 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<BlockResources>),
+}
+
 /// VMM-side handle for controlling and joining a block worker thread.
 #[derive(Debug)]
 pub(crate) struct WorkerHandle {
     to_worker: Sender<ControlMsg>,
+    from_worker: Receiver<ControlResponse>,
     control_evt: EventFd,
     join: JoinHandle<()>,
     queue_evts: Vec<EventFd>,
@@ -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::<ControlMsg>();
+        let (to_vmm, from_worker) = channel::<ControlResponse>();
 
         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<BlockResources> {
+        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<ControlMsg>,
+    to_vmm: Sender<ControlResponse>,
 ) {
     let worker = Arc::new(Mutex::new(ThreadedWorker {
         state: WorkerState::Parked,
         control_evt,
         from_vmm,
+        to_vmm,
     }));
     let subscriber: Arc<Mutex<dyn MutEventSubscriber>> = worker.clone();
     event_manager.add_subscriber(subscriber);