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() } }