diff --git a/virtio-devices/src/balloon.rs b/virtio-devices/src/balloon.rs index c11c00ab9..8b40a2f1a 100644 --- a/virtio-devices/src/balloon.rs +++ b/virtio-devices/src/balloon.rs @@ -583,16 +583,6 @@ impl Balloon { } } -impl Drop for Balloon { - fn drop(&mut self) { - if let Some(kill_evt) = self.common.kill_evt.take() { - // Ignore the result because there is nothing we can do about it. - let _ = kill_evt.write(1); - } - self.common.wait_for_epoll_threads(); - } -} - impl VirtioDevice for Balloon { fn device_type(&self) -> u32 { self.common.device_type diff --git a/virtio-devices/src/block.rs b/virtio-devices/src/block.rs index 18c75843b..bbc031bdb 100644 --- a/virtio-devices/src/block.rs +++ b/virtio-devices/src/block.rs @@ -1035,16 +1035,6 @@ impl Block { } } -impl Drop for Block { - fn drop(&mut self) { - if let Some(kill_evt) = self.common.kill_evt.take() { - // Ignore the result because there is nothing we can do about it. - let _ = kill_evt.write(1); - } - self.common.wait_for_epoll_threads(); - } -} - impl VirtioDevice for Block { fn device_type(&self) -> u32 { self.common.device_type diff --git a/virtio-devices/src/console.rs b/virtio-devices/src/console.rs index 91ba3a1a5..53e3f02ad 100644 --- a/virtio-devices/src/console.rs +++ b/virtio-devices/src/console.rs @@ -690,16 +690,6 @@ impl Console { } } -impl Drop for Console { - fn drop(&mut self) { - if let Some(kill_evt) = self.common.kill_evt.take() { - // Ignore the result because there is nothing we can do about it. - let _ = kill_evt.write(1); - } - self.common.wait_for_epoll_threads(); - } -} - impl VirtioDevice for Console { fn device_type(&self) -> u32 { self.common.device_type diff --git a/virtio-devices/src/device.rs b/virtio-devices/src/device.rs index 53554d313..9e5dfb138 100644 --- a/virtio-devices/src/device.rs +++ b/virtio-devices/src/device.rs @@ -225,17 +225,70 @@ pub trait DmaRemapping { ) -> std::result::Result; } +/// Owns a device's worker threads plus the kill event that stops them. +/// +/// Dropping signals every worker to exit, unparks any that are parked, and joins them. +pub struct WorkerThreads { + /// shared kill event, a single write wakes all of them. + kill_evt: EventFd, + // true if the device is paused. + paused: Arc, + // The running worker thread's handles. + threads: Vec>, +} + +impl WorkerThreads { + fn new(kill_evt: EventFd, paused: Arc) -> Self { + WorkerThreads { + kill_evt, + paused, + threads: Vec::new(), + } + } + + /// Borrow access to the kill eventfd + fn kill_evt(&self) -> &EventFd { + &self.kill_evt + } + + /// Signal the workers to exit without joining; they are joined later when + /// this is dropped. + pub(crate) fn signal_exit(&self) -> std::io::Result<()> { + self.kill_evt.write(1) + } + + /// Unpark every worker so threads parked while paused resume their loop. + fn unpark(&self) { + for t in &self.threads { + t.thread().unpark(); + } + } +} + +impl Drop for WorkerThreads { + fn drop(&mut self) { + // Signal the workers to exit, wake any parked so they observe it, then join. + let _ = self.kill_evt.write(1); + self.paused.store(false, Ordering::SeqCst); + self.unpark(); + for t in self.threads.drain(..) { + if let Err(e) = t.join() { + error!("Error joining thread: {e:?}"); + } + } + } +} + /// Structure to handle device state common to all devices #[derive(Default)] pub struct VirtioCommon { pub avail_features: u64, pub acked_features: u64, - pub kill_evt: Option, pub interrupt_cb: Option>, pub pause_evt: Option, pub paused: Arc, pub paused_sync: Option>, - pub epoll_threads: Option>>, + pub workers: Option, pub queue_sizes: Vec, pub queue_evts: Vec, pub device_type: u32, @@ -289,7 +342,9 @@ impl VirtioCommon { error!("failed creating kill EventFd: {e}"); ActivateError::BadActivate })?; - self.kill_evt = Some(kill_evt); + // Create the worker collection up front so it owns the kill event; + // handlers clone it via dup_eventfds() before any worker is spawned. + self.workers = Some(WorkerThreads::new(kill_evt, self.paused.clone())); let pause_evt = EventFd::new(EFD_NONBLOCK).map_err(|e| { error!("failed creating pause EventFd: {e}"); @@ -306,35 +361,21 @@ impl VirtioCommon { pub fn reset(&mut self) { self.queue_evts.clear(); + self.pause_evt = None; - // Resume the virtio thread if it was paused. Reset must always - // converge to fresh state, so a resume failure is logged but doesn't - // skip the rest of the teardown. - if self.pause_evt.take().is_some() - && let Err(e) = self.resume() - { - error!("Failed to resume paused device during reset: {e:?}"); - } + // Clear paused explicitly; the workers' Drop does so too, but only + // when they exist, and reset may run before activate(). + self.paused.store(false, Ordering::SeqCst); - if let Some(kill_evt) = self.kill_evt.take() { - // Ignore the result because there is nothing we can do about it. - let _ = kill_evt.write(1); - } - - if let Some(mut threads) = self.epoll_threads.take() { - for t in threads.drain(..) { - if let Err(e) = t.join() { - error!("Error joining thread: {e:?}"); - } - } - } + // Dropping the workers signals kill_evt, unparks any thread parked + // for migration, and joins them. + self.workers = None; // Drop the interrupt callback clone self.interrupt_cb = None; } - /// Spawn a worker, push its handle into `self.epoll_threads`, and on - /// spawn failure run `self.reset()` to join prior workers. + /// Spawn a worker; on failure, reset the device to join prior workers. #[expect(clippy::too_many_arguments)] pub fn spawn_worker( &mut self, @@ -349,18 +390,23 @@ impl VirtioCommon { where F: FnOnce() -> Result<(), EpollHelperError> + Send + 'static, { - let mut threads = self.epoll_threads.take().unwrap_or_default(); - let res = spawn_virtio_thread( - name, - seccomp_action, - thread_type, - &mut threads, - exit_evt, - device_status, - interrupt_cb, - f, - ); - self.epoll_threads = Some(threads); + // Scope the borrow of `workers` so it ends before the reset() below. + let res = { + let Some(workers) = self.workers.as_mut() else { + error!("spawn_worker called before activate()"); + return Err(ActivateError::BadActivate); + }; + spawn_virtio_thread( + name, + seccomp_action, + thread_type, + &mut workers.threads, + exit_evt, + device_status, + interrupt_cb, + f, + ) + }; if let Err(e) = res { self.reset(); return Err(e); @@ -376,20 +422,19 @@ impl VirtioCommon { } } - // Wait for the worker thread to finish and return + // Dropping the workers signals, unparks, and joins them. Idempotent. pub fn wait_for_epoll_threads(&mut self) { - if let Some(mut threads) = self.epoll_threads.take() { - for t in threads.drain(..) { - if let Err(e) = t.join() { - error!("Error joining thread: {e:?}"); - } - } - } + self.workers = None; } pub fn dup_eventfds(&self) -> (EventFd, EventFd) { ( - self.kill_evt.as_ref().unwrap().try_clone().unwrap(), + self.workers + .as_ref() + .unwrap() + .kill_evt() + .try_clone() + .unwrap(), self.pause_evt.as_ref().unwrap().try_clone().unwrap(), ) } @@ -446,10 +491,8 @@ impl Pausable for VirtioCommon { VirtioDeviceType::from(self.device_type) ); self.paused.store(false, Ordering::SeqCst); - if let Some(epoll_threads) = &self.epoll_threads { - for t in epoll_threads.iter() { - t.thread().unpark(); - } + if let Some(workers) = &self.workers { + workers.unpark(); } // Signal each activated queue eventfd so workers process restored queues @@ -495,19 +538,23 @@ mod unit_tests { } } - fn make_common_with_kill_evt() -> (VirtioCommon, EventFd) { + /// VirtioCommon with its worker collection created (as activate() does) + /// and a kill_evt clone for the spawned worker to watch. + fn make_common_with_workers() -> (VirtioCommon, EventFd) { let kill_evt = EventFd::new(EFD_NONBLOCK).unwrap(); let kill_evt_clone = kill_evt.try_clone().unwrap(); + let common = VirtioCommon::default(); + let workers = WorkerThreads::new(kill_evt, common.paused.clone()); let common = VirtioCommon { - kill_evt: Some(kill_evt), - ..Default::default() + workers: Some(workers), + ..common }; (common, kill_evt_clone) } #[test] - fn spawn_worker_appends_to_epoll_threads() { - let (mut common, kill_evt_clone) = make_common_with_kill_evt(); + fn spawn_worker_appends_to_workers() { + let (mut common, kill_evt_clone) = make_common_with_workers(); let started = Arc::new(AtomicUsize::new(0)); let started_clone = started.clone(); @@ -530,15 +577,60 @@ mod unit_tests { ) .unwrap(); - let threads = common.epoll_threads.as_ref().expect("epoll_threads set"); - assert_eq!(threads.len(), 1); + let workers = common.workers.as_ref().expect("workers set"); + assert_eq!(workers.threads.len(), 1); - // reset() joins the worker; this confirms the spawn_worker - // -> reset() chain (used on spawn failure) actually drains - // a real, running worker rather than dropping it detached. + // reset() drops the WorkerThreads and joins the worker, exercising + // the spawn-failure cleanup path on a real running thread. common.reset(); - assert!(common.epoll_threads.is_none()); - assert!(common.kill_evt.is_none()); + assert!(common.workers.is_none()); assert_eq!(started.load(Ordering::SeqCst), 1); } + + #[test] + fn dropping_common_joins_workers() { + let (mut common, kill_evt_clone) = make_common_with_workers(); + let started = Arc::new(AtomicUsize::new(0)); + let started_clone = started.clone(); + + let exit_evt = EventFd::new(EFD_NONBLOCK).unwrap(); + let status = Arc::new(AtomicU8::new(0)); + + common + .spawn_worker( + "test", + &SeccompAction::Allow, + Thread::VirtioBlock, + &exit_evt, + status, + Arc::new(NoopInterrupt), + move || { + started_clone.fetch_add(1, Ordering::SeqCst); + let _ = kill_evt_clone.read(); + Ok(()) + }, + ) + .unwrap(); + + // Dropping `common` alone must join the worker via WorkerThreads' Drop. + drop(common); + assert_eq!(started.load(Ordering::SeqCst), 1); + } + + #[test] + fn reset_clears_paused_without_workers() { + // reset() before any worker was spawned must still clear paused, or + // the next activation's workers would park immediately and never run. + let mut common = VirtioCommon { + pause_evt: Some(EventFd::new(EFD_NONBLOCK).unwrap()), + ..Default::default() + }; + common.paused.store(true, Ordering::SeqCst); + assert!(common.workers.is_none()); + + common.reset(); + + assert!(!common.paused.load(Ordering::SeqCst)); + assert!(common.pause_evt.is_none()); + } } diff --git a/virtio-devices/src/iommu.rs b/virtio-devices/src/iommu.rs index 2e132ead2..191d823cf 100644 --- a/virtio-devices/src/iommu.rs +++ b/virtio-devices/src/iommu.rs @@ -1252,16 +1252,6 @@ impl Iommu { } } -impl Drop for Iommu { - fn drop(&mut self) { - if let Some(kill_evt) = self.common.kill_evt.take() { - // Ignore the result because there is nothing we can do about it. - let _ = kill_evt.write(1); - } - self.common.wait_for_epoll_threads(); - } -} - impl VirtioDevice for Iommu { fn device_type(&self) -> u32 { self.common.device_type diff --git a/virtio-devices/src/mem.rs b/virtio-devices/src/mem.rs index d383d6ee0..10cb7ee48 100644 --- a/virtio-devices/src/mem.rs +++ b/virtio-devices/src/mem.rs @@ -922,16 +922,6 @@ impl Mem { } } -impl Drop for Mem { - fn drop(&mut self) { - if let Some(kill_evt) = self.common.kill_evt.take() { - // Ignore the result because there is nothing we can do about it. - let _ = kill_evt.write(1); - } - self.common.wait_for_epoll_threads(); - } -} - impl VirtioDevice for Mem { fn device_type(&self) -> u32 { self.common.device_type diff --git a/virtio-devices/src/net.rs b/virtio-devices/src/net.rs index 8e87ba85e..5207f0564 100644 --- a/virtio-devices/src/net.rs +++ b/virtio-devices/src/net.rs @@ -644,17 +644,6 @@ impl Net { } } -impl Drop for Net { - fn drop(&mut self) { - if let Some(kill_evt) = self.common.kill_evt.take() { - // Ignore the result because there is nothing we can do about it. - let _ = kill_evt.write(1); - } - // Needed to ensure all references to tap FDs are dropped (#4868) - self.common.wait_for_epoll_threads(); - } -} - impl VirtioDevice for Net { fn device_type(&self) -> u32 { self.common.device_type diff --git a/virtio-devices/src/pmem.rs b/virtio-devices/src/pmem.rs index ecdad9815..790512f83 100644 --- a/virtio-devices/src/pmem.rs +++ b/virtio-devices/src/pmem.rs @@ -349,16 +349,6 @@ impl Pmem { } } -impl Drop for Pmem { - fn drop(&mut self) { - if let Some(kill_evt) = self.common.kill_evt.take() { - // Ignore the result because there is nothing we can do about it. - let _ = kill_evt.write(1); - } - self.common.wait_for_epoll_threads(); - } -} - impl VirtioDevice for Pmem { fn device_type(&self) -> u32 { self.common.device_type diff --git a/virtio-devices/src/rng.rs b/virtio-devices/src/rng.rs index 1814f3c35..792860091 100644 --- a/virtio-devices/src/rng.rs +++ b/virtio-devices/src/rng.rs @@ -228,16 +228,6 @@ impl Rng { } } -impl Drop for Rng { - fn drop(&mut self) { - if let Some(kill_evt) = self.common.kill_evt.take() { - // Ignore the result because there is nothing we can do about it. - let _ = kill_evt.write(1); - } - self.common.wait_for_epoll_threads(); - } -} - impl VirtioDevice for Rng { fn device_type(&self) -> u32 { self.common.device_type diff --git a/virtio-devices/src/rtc.rs b/virtio-devices/src/rtc.rs index e56031f3d..6c8de79cf 100644 --- a/virtio-devices/src/rtc.rs +++ b/virtio-devices/src/rtc.rs @@ -599,16 +599,6 @@ impl Rtc { } } -impl Drop for Rtc { - fn drop(&mut self) { - if let Some(kill_evt) = self.common.kill_evt.take() { - // Ignore the result because there is nothing we can do about it. - let _ = kill_evt.write(1); - } - self.common.wait_for_epoll_threads(); - } -} - impl VirtioDevice for Rtc { fn device_type(&self) -> u32 { self.common.device_type diff --git a/virtio-devices/src/vhost_user/mod.rs b/virtio-devices/src/vhost_user/mod.rs index b7cd5b814..c9cceac93 100644 --- a/virtio-devices/src/vhost_user/mod.rs +++ b/virtio-devices/src/vhost_user/mod.rs @@ -613,19 +613,8 @@ impl VhostUserCommon { } pub fn shutdown(&mut self) { - // Signal workers to exit, unpause them (they may be parked - // if the VM was paused for migration), then wait for them - // to finish so they drop their Arc and the - // socket fully closes for the destination to reconnect. - if let Some(kill_evt) = self.virtio_common.kill_evt.take() { - let _ = kill_evt.write(1); - } - self.virtio_common.paused.store(false, Ordering::SeqCst); - if let Some(threads) = self.virtio_common.epoll_threads.as_ref() { - for t in threads { - t.thread().unpark(); - } - } + // Join the workers so they drop their Arc and the + // socket closes, letting the migration destination reconnect. self.virtio_common.wait_for_epoll_threads(); // Remove socket path if needed @@ -862,8 +851,8 @@ impl VhostUserCommon { // Make sure the device thread is killed in order to prevent from // reconnections to the socket. - if let Some(kill_evt) = self.virtio_common.kill_evt.take() { - kill_evt.write(1).map_err(|e| { + if let Some(workers) = self.virtio_common.workers.as_ref() { + workers.signal_exit().map_err(|e| { MigratableError::CompleteMigration(anyhow!( "Error killing vhost-user thread: {e:?}" )) diff --git a/virtio-devices/src/vsock/device.rs b/virtio-devices/src/vsock/device.rs index c7e181e75..047a46da6 100644 --- a/virtio-devices/src/vsock/device.rs +++ b/virtio-devices/src/vsock/device.rs @@ -426,19 +426,6 @@ where } } -impl Drop for Vsock -where - B: VsockBackend, -{ - fn drop(&mut self) { - if let Some(kill_evt) = self.common.kill_evt.take() { - // Ignore the result because there is nothing we can do about it. - let _ = kill_evt.write(1); - } - self.common.wait_for_epoll_threads(); - } -} - impl VirtioDevice for Vsock where B: VsockBackend + Sync + 'static, diff --git a/virtio-devices/src/watchdog.rs b/virtio-devices/src/watchdog.rs index f394c09df..eef571e51 100644 --- a/virtio-devices/src/watchdog.rs +++ b/virtio-devices/src/watchdog.rs @@ -270,16 +270,6 @@ impl Watchdog { } } -impl Drop for Watchdog { - fn drop(&mut self) { - if let Some(kill_evt) = self.common.kill_evt.take() { - // Ignore the result because there is nothing we can do about it. - let _ = kill_evt.write(1); - } - self.common.wait_for_epoll_threads(); - } -} - fn timerfd_create() -> Result { // SAFETY: FFI call, trivially safe let res = unsafe { libc::timerfd_create(libc::CLOCK_MONOTONIC, 0) };