From 1f58e74978a62a9187aef6933a5cb64a00021820 Mon Sep 17 00:00:00 2001 From: Rob Bradford Date: Fri, 29 May 2026 16:47:45 +0100 Subject: [PATCH] virtio-devices: Simplify epoll thread handling Rely on the vector of threads in VirtioCommon for storing all the handles of the spawned threads rather than storing them in the devices or in VhostUserCommon directly. Assisted-by: Claude:Opus-4.7 Signed-off-by: Rob Bradford --- virtio-devices/src/net.rs | 21 +++--------- virtio-devices/src/vhost_user/blk.rs | 7 +--- virtio-devices/src/vhost_user/fs.rs | 7 +--- .../src/vhost_user/generic_vhost_user.rs | 7 +--- virtio-devices/src/vhost_user/mod.rs | 30 ++++++---------- virtio-devices/src/vhost_user/net.rs | 34 ++++++------------- 6 files changed, 28 insertions(+), 78 deletions(-) diff --git a/virtio-devices/src/net.rs b/virtio-devices/src/net.rs index 6edc40894..c1df658f3 100644 --- a/virtio-devices/src/net.rs +++ b/virtio-devices/src/net.rs @@ -10,9 +10,9 @@ use std::net::IpAddr; use std::num::Wrapping; use std::ops::Deref; use std::os::unix::io::{AsRawFd, RawFd}; +use std::result; use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; use std::sync::{Arc, Barrier}; -use std::{result, thread}; use anyhow::anyhow; use event_monitor::event; @@ -400,7 +400,6 @@ pub struct Net { id: String, taps: Vec, config: VirtioNetConfig, - ctrl_queue_epoll_thread: Option>, counters: NetCounters, seccomp_action: SeccompAction, rate_limiter_config: Option, @@ -520,7 +519,6 @@ impl Net { id, taps, config, - ctrl_queue_epoll_thread: None, counters: NetCounters::default(), seccomp_action, rate_limiter_config, @@ -655,11 +653,6 @@ impl Drop for Net { } // Needed to ensure all references to tap FDs are dropped (#4868) self.common.wait_for_epoll_threads(); - if let Some(thread) = self.ctrl_queue_epoll_thread.take() - && let Err(e) = thread.join() - { - error!("Error joining thread: {e:?}"); - } } } @@ -704,6 +697,8 @@ impl VirtioDevice for Net { let qp_threads = (num_queues - ctrl_threads) / 2; self.common.paused_sync = Some(Arc::new(Barrier::new(1 + qp_threads + ctrl_threads))); + let mut epoll_threads = Vec::new(); + if has_ctrl_queue { let ctrl_queue_index = num_queues - 1; let (_, mut ctrl_queue, ctrl_queue_evt) = queues.remove(ctrl_queue_index); @@ -726,7 +721,6 @@ impl VirtioDevice for Net { let paused = self.common.paused.clone(); let paused_sync = self.common.paused_sync.clone(); - let mut epoll_threads = Vec::new(); spawn_virtio_thread( &format!("{}_ctrl", self.id), &self.seccomp_action, @@ -737,10 +731,8 @@ impl VirtioDevice for Net { interrupt_cb.clone(), move || ctrl_handler.run_ctrl(&paused, paused_sync.as_ref().unwrap()), )?; - self.ctrl_queue_epoll_thread = Some(epoll_threads.remove(0)); } - let mut epoll_threads = Vec::new(); let mut taps = self.taps.clone(); for i in 0..queues.len() / 2 { let rx = RxVirtio::new(); @@ -867,12 +859,7 @@ impl Pausable for Net { } fn resume(&mut self) -> result::Result<(), MigratableError> { - self.common.resume()?; - - if let Some(ctrl_queue_epoll_thread) = &self.ctrl_queue_epoll_thread { - ctrl_queue_epoll_thread.thread().unpark(); - } - Ok(()) + self.common.resume() } } diff --git a/virtio-devices/src/vhost_user/blk.rs b/virtio-devices/src/vhost_user/blk.rs index dbaab8b01..811680b0a 100644 --- a/virtio-devices/src/vhost_user/blk.rs +++ b/virtio-devices/src/vhost_user/blk.rs @@ -309,7 +309,7 @@ impl VirtioDevice for Blk { interrupt_cb.clone(), move || handler.run(&paused, paused_sync.as_ref().unwrap()), )?; - self.vu_common.epoll_thread = Some(epoll_threads.remove(0)); + self.vu_common.virtio_common.epoll_threads = Some(epoll_threads); Ok(()) } @@ -338,11 +338,6 @@ impl Pausable for Blk { fn resume(&mut self) -> result::Result<(), MigratableError> { self.vu_common.virtio_common.resume()?; - - if let Some(epoll_thread) = &self.vu_common.epoll_thread { - epoll_thread.thread().unpark(); - } - self.vu_common.resume() } } diff --git a/virtio-devices/src/vhost_user/fs.rs b/virtio-devices/src/vhost_user/fs.rs index 31738a1bb..0f1f1de72 100644 --- a/virtio-devices/src/vhost_user/fs.rs +++ b/virtio-devices/src/vhost_user/fs.rs @@ -282,7 +282,7 @@ impl VirtioDevice for Fs { interrupt_cb.clone(), move || handler.run(&paused, paused_sync.as_ref().unwrap()), )?; - self.vu_common.epoll_thread = Some(epoll_threads.remove(0)); + self.vu_common.virtio_common.epoll_threads = Some(epoll_threads); event!("virtio-device", "activated", "id", &self.id); Ok(()) @@ -342,11 +342,6 @@ impl Pausable for Fs { fn resume(&mut self) -> result::Result<(), MigratableError> { self.vu_common.virtio_common.resume()?; - - if let Some(epoll_thread) = &self.vu_common.epoll_thread { - epoll_thread.thread().unpark(); - } - self.vu_common.resume() } } diff --git a/virtio-devices/src/vhost_user/generic_vhost_user.rs b/virtio-devices/src/vhost_user/generic_vhost_user.rs index 5a302dc55..ce03db76e 100644 --- a/virtio-devices/src/vhost_user/generic_vhost_user.rs +++ b/virtio-devices/src/vhost_user/generic_vhost_user.rs @@ -344,7 +344,7 @@ impl VirtioDevice for GenericVhostUser { interrupt_cb.clone(), move || handler.run(&paused, paused_sync.as_ref().unwrap()), )?; - self.vu_common.epoll_thread = Some(epoll_threads.remove(0)); + self.vu_common.virtio_common.epoll_threads = Some(epoll_threads); event!("virtio-device", "activated", "id", &self.id); Ok(()) @@ -404,11 +404,6 @@ impl Pausable for GenericVhostUser { fn resume(&mut self) -> result::Result<(), MigratableError> { self.vu_common.virtio_common.resume()?; - - if let Some(epoll_thread) = &self.vu_common.epoll_thread { - epoll_thread.thread().unpark(); - } - self.vu_common.resume() } } diff --git a/virtio-devices/src/vhost_user/mod.rs b/virtio-devices/src/vhost_user/mod.rs index babc823ae..5a6cbdfe2 100644 --- a/virtio-devices/src/vhost_user/mod.rs +++ b/virtio-devices/src/vhost_user/mod.rs @@ -1,12 +1,12 @@ // Copyright 2019 Intel Corporation. All Rights Reserved. // SPDX-License-Identifier: Apache-2.0 +use std::io; use std::io::ErrorKind; use std::ops::Deref; use std::os::unix::io::AsRawFd; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Barrier, Mutex}; -use std::{io, thread}; use anyhow::anyhow; use event_monitor::event; @@ -463,7 +463,6 @@ pub struct VhostUserCommon { pub migration_started: bool, pub server: bool, pub vring_bases: Option>, - pub epoll_thread: Option>, /// Indicates that the backend is no longer reachable. Shared with EPollHandler. pub disconnected: Arc, } @@ -566,15 +565,9 @@ impl VhostUserCommon { } } - if let Some(kill_evt) = self.virtio_common.kill_evt.take() { - // Ignore the result because there is nothing we can do about it. - let _ = kill_evt.write(1); - } + self.virtio_common.reset(); event!("virtio-device", "reset", "id", id); - - // Drop the interrupt callback clone - self.virtio_common.interrupt_cb = None; } fn memory_update_error(&self, source: Error) -> crate::Error { @@ -587,21 +580,20 @@ impl VhostUserCommon { } pub fn shutdown(&mut self) { - // Signal the epoll thread to exit, unpause it (it may be parked - // if the VM was paused for migration), then wait for it to finish. - // This ensures the thread drops its Arc, fully - // closing the vhost-user socket so the backend can accept a new - // connection from the destination. + // 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(t) = self.epoll_thread.as_ref() { - t.thread().unpark(); - } - if let Some(t) = self.epoll_thread.take() { - let _ = t.join(); + if let Some(threads) = self.virtio_common.epoll_threads.as_ref() { + for t in threads { + t.thread().unpark(); + } } + self.virtio_common.wait_for_epoll_threads(); // Remove socket path if needed if self.server { diff --git a/virtio-devices/src/vhost_user/net.rs b/virtio-devices/src/vhost_user/net.rs index 620f876a7..7c42ce2f0 100644 --- a/virtio-devices/src/vhost_user/net.rs +++ b/virtio-devices/src/vhost_user/net.rs @@ -1,9 +1,9 @@ // Copyright 2019 Intel Corporation. All Rights Reserved. // SPDX-License-Identifier: Apache-2.0 +use std::result; use std::sync::atomic::AtomicBool; use std::sync::{Arc, Barrier, Mutex}; -use std::{result, thread}; use log::{error, info}; use net_util::{CtrlQueue, MacAddr, VirtioNetConfig, build_net_config_space}; @@ -44,7 +44,6 @@ pub struct Net { id: String, config: VirtioNetConfig, guest_memory: Option>, - ctrl_queue_epoll_thread: Option>, seccomp_action: SeccompAction, exit_evt: EventFd, access_platform_enabled: bool, @@ -221,7 +220,6 @@ impl Net { }, config, guest_memory: None, - ctrl_queue_epoll_thread: None, seccomp_action, exit_evt, access_platform_enabled, @@ -236,12 +234,6 @@ impl Net { impl Drop for Net { fn drop(&mut self) { self.vu_common.shutdown(); - - if let Some(thread) = self.ctrl_queue_epoll_thread.take() - && let Err(e) = thread.join() - { - error!("Error joining thread: {e:?}"); - } } } @@ -319,18 +311,18 @@ impl VirtioDevice for Net { self.vu_common.virtio_common.paused_sync = Some(Arc::new(Barrier::new(3))); let paused_sync = self.vu_common.virtio_common.paused_sync.clone(); - let mut epoll_threads = Vec::new(); + let mut ctrl_threads = Vec::new(); spawn_virtio_thread( &format!("{}_ctrl", self.id), &self.seccomp_action, Thread::VirtioVhostNetCtl, - &mut epoll_threads, + &mut ctrl_threads, &self.exit_evt, device_status.clone(), interrupt_cb.clone(), move || ctrl_handler.run_ctrl(&paused, paused_sync.as_ref().unwrap()), )?; - self.ctrl_queue_epoll_thread = Some(epoll_threads.remove(0)); + self.vu_common.virtio_common.epoll_threads = Some(ctrl_threads); } let backend_req_handler: Option> = None; @@ -357,18 +349,21 @@ impl VirtioDevice for Net { let paused = self.vu_common.virtio_common.paused.clone(); let paused_sync = self.vu_common.virtio_common.paused_sync.clone(); - let mut epoll_threads = Vec::new(); + let threads = self + .vu_common + .virtio_common + .epoll_threads + .get_or_insert_with(Vec::new); spawn_virtio_thread( &self.id, &self.seccomp_action, Thread::VirtioVhostNet, - &mut epoll_threads, + threads, &self.exit_evt, device_status.clone(), interrupt_cb.clone(), move || handler.run(&paused, paused_sync.as_ref().unwrap()), )?; - self.vu_common.epoll_thread = Some(epoll_threads.remove(0)); Ok(()) } @@ -397,15 +392,6 @@ impl Pausable for Net { fn resume(&mut self) -> result::Result<(), MigratableError> { self.vu_common.virtio_common.resume()?; - - if let Some(epoll_thread) = &self.vu_common.epoll_thread { - epoll_thread.thread().unpark(); - } - - if let Some(ctrl_queue_epoll_thread) = &self.ctrl_queue_epoll_thread { - ctrl_queue_epoll_thread.thread().unpark(); - } - self.vu_common.resume() } }