diff --git a/virtio-devices/src/balloon.rs b/virtio-devices/src/balloon.rs index 9cf06c2f8..476160340 100644 --- a/virtio-devices/src/balloon.rs +++ b/virtio-devices/src/balloon.rs @@ -25,7 +25,7 @@ use std::os::unix::io::AsRawFd; use std::result; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::mpsc; -use std::sync::{Arc, Mutex}; +use std::sync::{Arc, Barrier, Mutex}; use std::thread; use vm_memory::{ Address, ByteValued, Bytes, GuestAddress, GuestAddressSpace, GuestMemoryAtomic, @@ -229,12 +229,16 @@ impl BalloonEpollHandler { Ok(()) } - fn run(&mut self, paused: Arc) -> result::Result<(), EpollHelperError> { + fn run( + &mut self, + paused: Arc, + paused_sync: Arc, + ) -> result::Result<(), EpollHelperError> { let mut helper = EpollHelper::new(&self.kill_evt, &self.pause_evt)?; helper.add_event(self.resize_receiver.evt.as_raw_fd(), RESIZE_EVENT)?; helper.add_event(self.inflate_queue_evt.as_raw_fd(), INFLATE_QUEUE_EVENT)?; helper.add_event(self.deflate_queue_evt.as_raw_fd(), DEFLATE_QUEUE_EVENT)?; - helper.run(paused, self)?; + helper.run(paused, paused_sync, self)?; Ok(()) } @@ -313,6 +317,7 @@ pub struct Balloon { interrupt_cb: Option>, epoll_threads: Option>>>, paused: Arc, + paused_sync: Arc, } impl Balloon { @@ -335,6 +340,7 @@ impl Balloon { interrupt_cb: None, epoll_threads: None, paused: Arc::new(AtomicBool::new(false)), + paused_sync: Arc::new(Barrier::new(2)), }) } @@ -443,10 +449,11 @@ impl VirtioDevice for Balloon { }; let paused = self.paused.clone(); + let paused_sync = self.paused_sync.clone(); let mut epoll_threads = Vec::new(); thread::Builder::new() .name("virtio_balloon".to_string()) - .spawn(move || handler.run(paused)) + .spawn(move || handler.run(paused, paused_sync)) .map(|thread| epoll_threads.push(thread)) .map_err(|e| { error!("failed to clone virtio-balloon epoll thread: {}", e); diff --git a/virtio-devices/src/block.rs b/virtio-devices/src/block.rs index 7669711e8..fd09c2ff7 100644 --- a/virtio-devices/src/block.rs +++ b/virtio-devices/src/block.rs @@ -27,7 +27,7 @@ use std::os::unix::io::AsRawFd; use std::path::PathBuf; use std::result; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; -use std::sync::{Arc, Mutex}; +use std::sync::{Arc, Barrier, Mutex}; use std::thread; use virtio_bindings::bindings::virtio_blk::*; use virtio_bindings::bindings::virtio_ring::VIRTIO_RING_F_EVENT_IDX; @@ -198,10 +198,14 @@ impl BlockEpollHandler { Ok(()) } - fn run(&mut self, paused: Arc) -> result::Result<(), EpollHelperError> { + fn run( + &mut self, + paused: Arc, + paused_sync: Arc, + ) -> result::Result<(), EpollHelperError> { let mut helper = EpollHelper::new(&self.kill_evt, &self.pause_evt)?; helper.add_event(self.queue_evt.as_raw_fd(), QUEUE_AVAIL_EVENT)?; - helper.run(paused, self)?; + helper.run(paused, paused_sync, self)?; Ok(()) } @@ -268,6 +272,7 @@ pub struct Block { epoll_threads: Option>>, pause_evt: Option, paused: Arc, + paused_sync: Arc, queue_size: Vec, writeback: Arc, counters: BlockCounters, @@ -346,6 +351,7 @@ impl Block { epoll_threads: None, pause_evt: None, paused: Arc::new(AtomicBool::new(false)), + paused_sync: Arc::new(Barrier::new(num_queues + 1)), queue_size: vec![queue_size; num_queues], writeback: Arc::new(AtomicBool::new(true)), counters: BlockCounters::default(), @@ -534,6 +540,7 @@ impl VirtioDevice for Block { handler.queue.set_event_idx(event_idx); let paused = self.paused.clone(); + let paused_sync = self.paused_sync.clone(); // Retrieve seccomp filter for virtio_blk thread let virtio_blk_seccomp_filter = @@ -545,7 +552,7 @@ impl VirtioDevice for Block { .spawn(move || { if let Err(e) = SeccompFilter::apply(virtio_blk_seccomp_filter) { error!("Error applying seccomp filter: {:?}", e); - } else if let Err(e) = handler.run(paused) { + } else if let Err(e) = handler.run(paused, paused_sync) { error!("Error running worker: {:?}", e); } }) diff --git a/virtio-devices/src/block_io_uring.rs b/virtio-devices/src/block_io_uring.rs index 97cb603ef..c3253c57e 100644 --- a/virtio-devices/src/block_io_uring.rs +++ b/virtio-devices/src/block_io_uring.rs @@ -26,7 +26,7 @@ use std::os::unix::io::{AsRawFd, RawFd}; use std::path::PathBuf; use std::result; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; -use std::sync::Arc; +use std::sync::{Arc, Barrier}; use std::thread; use virtio_bindings::bindings::virtio_blk::*; use vm_memory::{ @@ -229,11 +229,15 @@ impl BlockIoUringEpollHandler { }) } - fn run(&mut self, paused: Arc) -> result::Result<(), EpollHelperError> { + fn run( + &mut self, + paused: Arc, + paused_sync: Arc, + ) -> result::Result<(), EpollHelperError> { let mut helper = EpollHelper::new(&self.kill_evt, &self.pause_evt)?; helper.add_event(self.queue_evt.as_raw_fd(), QUEUE_AVAIL_EVENT)?; helper.add_event(self.io_uring_evt.as_raw_fd(), IO_URING_EVENT)?; - helper.run(paused, self)?; + helper.run(paused, paused_sync, self)?; Ok(()) } @@ -309,6 +313,7 @@ pub struct BlockIoUring { epoll_threads: Option>>>, pause_evt: Option, paused: Arc, + paused_sync: Arc, queue_size: Vec, writeback: Arc, counters: BlockCounters, @@ -381,6 +386,7 @@ impl BlockIoUring { epoll_threads: None, pause_evt: None, paused: Arc::new(AtomicBool::new(false)), + paused_sync: Arc::new(Barrier::new(num_queues + 1)), queue_size: vec![queue_size; num_queues], writeback: Arc::new(AtomicBool::new(true)), counters: BlockCounters::default(), @@ -580,6 +586,7 @@ impl VirtioDevice for BlockIoUring { }; let paused = self.paused.clone(); + let paused_sync = self.paused_sync.clone(); // Register the io_uring eventfd that will notify the epoll loop // when something in the completion queue is ready. @@ -594,7 +601,7 @@ impl VirtioDevice for BlockIoUring { thread::Builder::new() .name("virtio_blk".to_string()) - .spawn(move || handler.run(paused)) + .spawn(move || handler.run(paused, paused_sync)) .map(|thread| epoll_threads.push(thread)) .map_err(|e| { error!("failed to clone the virtio-blk epoll thread: {}", e); diff --git a/virtio-devices/src/console.rs b/virtio-devices/src/console.rs index c4acf6cf7..68d9f13ed 100644 --- a/virtio-devices/src/console.rs +++ b/virtio-devices/src/console.rs @@ -21,7 +21,7 @@ use std::ops::DerefMut; use std::os::unix::io::AsRawFd; use std::result; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; -use std::sync::{Arc, Mutex}; +use std::sync::{Arc, Barrier, Mutex}; use std::thread; use vm_memory::{ByteValued, Bytes, GuestAddressSpace, GuestMemoryAtomic, GuestMemoryMmap}; use vm_migration::{ @@ -180,13 +180,17 @@ impl ConsoleEpollHandler { }) } - fn run(&mut self, paused: Arc) -> result::Result<(), EpollHelperError> { + fn run( + &mut self, + paused: Arc, + paused_sync: Arc, + ) -> result::Result<(), EpollHelperError> { let mut helper = EpollHelper::new(&self.kill_evt, &self.pause_evt)?; helper.add_event(self.input_queue_evt.as_raw_fd(), INPUT_QUEUE_EVENT)?; helper.add_event(self.output_queue_evt.as_raw_fd(), OUTPUT_QUEUE_EVENT)?; helper.add_event(self.input_evt.as_raw_fd(), INPUT_EVENT)?; helper.add_event(self.config_evt.as_raw_fd(), CONFIG_EVENT)?; - helper.run(paused, self)?; + helper.run(paused, paused_sync, self)?; Ok(()) } @@ -306,6 +310,7 @@ pub struct Console { interrupt_cb: Option>, epoll_threads: Option>>, paused: Arc, + paused_sync: Arc, seccomp_action: SeccompAction, } @@ -358,6 +363,7 @@ impl Console { interrupt_cb: None, epoll_threads: None, paused: Arc::new(AtomicBool::new(false)), + paused_sync: Arc::new(Barrier::new(2)), seccomp_action, }, console_input, @@ -495,6 +501,7 @@ impl VirtioDevice for Console { }; let paused = self.paused.clone(); + let paused_sync = self.paused_sync.clone(); let mut epoll_threads = Vec::new(); // Retrieve seccomp filter for virtio_console thread let virtio_console_seccomp_filter = @@ -505,7 +512,7 @@ impl VirtioDevice for Console { .spawn(move || { if let Err(e) = SeccompFilter::apply(virtio_console_seccomp_filter) { error!("Error applying seccomp filter: {:?}", e); - } else if let Err(e) = handler.run(paused) { + } else if let Err(e) = handler.run(paused, paused_sync) { error!("Error running worker: {:?}", e); } }) diff --git a/virtio-devices/src/device.rs b/virtio-devices/src/device.rs index 1e35e4b53..91dec082b 100644 --- a/virtio-devices/src/device.rs +++ b/virtio-devices/src/device.rs @@ -205,6 +205,13 @@ macro_rules! virtio_pausable_trait_inner { pause_evt .write(1) .map_err(|e| MigratableError::Pause(e.into()))?; + + // Wait for all threads to acknowledge the pause before going + // any further. This is exclusively performed when pause_evt + // eventfd is Some(), as this means the virtio device has been + // activated. One specific case where the device can be paused + // while it hasn't been yet activated is snapshot/restore. + self.paused_sync.wait(); } Ok(()) diff --git a/virtio-devices/src/epoll_helper.rs b/virtio-devices/src/epoll_helper.rs index 849ca5ed9..a0eb04519 100644 --- a/virtio-devices/src/epoll_helper.rs +++ b/virtio-devices/src/epoll_helper.rs @@ -11,7 +11,7 @@ use std::fs::File; use std::os::unix::io::{AsRawFd, FromRawFd, RawFd}; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::Arc; +use std::sync::{Arc, Barrier}; use std::thread; use vmm_sys_util::eventfd::EventFd; @@ -69,6 +69,7 @@ impl EpollHelper { pub fn run( &mut self, paused: Arc, + paused_sync: Arc, handler: &mut dyn EpollHelperHandler, ) -> std::result::Result<(), EpollHelperError> { const EPOLL_EVENTS_LEN: usize = 100; @@ -110,6 +111,11 @@ impl EpollHelper { } EPOLL_HELPER_EVENT_PAUSE => { debug!("PAUSE_EVENT received, pausing epoll loop"); + + // Acknowledge the pause is effective by using the + // paused_sync barrier. + paused_sync.wait(); + // We loop here to handle spurious park() returns. // Until we have not resumed, the paused boolean will // be true. diff --git a/virtio-devices/src/iommu.rs b/virtio-devices/src/iommu.rs index 598225c8d..8be20d234 100644 --- a/virtio-devices/src/iommu.rs +++ b/virtio-devices/src/iommu.rs @@ -19,7 +19,7 @@ use std::ops::Bound::Included; use std::os::unix::io::AsRawFd; use std::result; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::{Arc, RwLock}; +use std::sync::{Arc, Barrier, RwLock}; use std::thread; use vfio_ioctls::ExternalDmaMapping; use vm_memory::{ @@ -646,11 +646,15 @@ impl IommuEpollHandler { }) } - fn run(&mut self, paused: Arc) -> result::Result<(), EpollHelperError> { + fn run( + &mut self, + paused: Arc, + paused_sync: Arc, + ) -> result::Result<(), EpollHelperError> { let mut helper = EpollHelper::new(&self.kill_evt, &self.pause_evt)?; helper.add_event(self.queue_evts[0].as_raw_fd(), REQUEST_Q_EVENT)?; helper.add_event(self.queue_evts[1].as_raw_fd(), EVENT_Q_EVENT)?; - helper.run(paused, self)?; + helper.run(paused, paused_sync, self)?; Ok(()) } @@ -743,6 +747,7 @@ pub struct Iommu { interrupt_cb: Option>, epoll_threads: Option>>>, paused: Arc, + paused_sync: Arc, } #[derive(Serialize, Deserialize)] @@ -783,6 +788,7 @@ impl Iommu { interrupt_cb: None, epoll_threads: None, paused: Arc::new(AtomicBool::new(false)), + paused_sync: Arc::new(Barrier::new(2)), }, mapping, )) @@ -955,10 +961,11 @@ impl VirtioDevice for Iommu { }; let paused = self.paused.clone(); + let paused_sync = self.paused_sync.clone(); let mut epoll_threads = Vec::new(); thread::Builder::new() .name("virtio_iommu".to_string()) - .spawn(move || handler.run(paused)) + .spawn(move || handler.run(paused, paused_sync)) .map(|thread| epoll_threads.push(thread)) .map_err(|e| { error!("failed to clone the virtio-iommu epoll thread: {}", e); diff --git a/virtio-devices/src/mem.rs b/virtio-devices/src/mem.rs index 057705705..593dd3cec 100644 --- a/virtio-devices/src/mem.rs +++ b/virtio-devices/src/mem.rs @@ -28,7 +28,7 @@ use std::os::unix::io::{AsRawFd, RawFd}; use std::result; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::mpsc; -use std::sync::{Arc, Mutex}; +use std::sync::{Arc, Barrier, Mutex}; use std::thread; use vm_memory::{ Address, ByteValued, Bytes, GuestAddress, GuestAddressSpace, GuestMemoryAtomic, @@ -589,11 +589,15 @@ impl MemEpollHandler { used_count > 0 } - fn run(&mut self, paused: Arc) -> result::Result<(), EpollHelperError> { + fn run( + &mut self, + paused: Arc, + paused_sync: Arc, + ) -> result::Result<(), EpollHelperError> { let mut helper = EpollHelper::new(&self.kill_evt, &self.pause_evt)?; helper.add_event(self.resize.evt.as_raw_fd(), RESIZE_EVENT)?; helper.add_event(self.queue_evt.as_raw_fd(), QUEUE_AVAIL_EVENT)?; - helper.run(paused, self)?; + helper.run(paused, paused_sync, self)?; Ok(()) } @@ -687,6 +691,7 @@ pub struct Mem { interrupt_cb: Option>, epoll_threads: Option>>>, paused: Arc, + paused_sync: Arc, } impl Mem { @@ -737,6 +742,7 @@ impl Mem { interrupt_cb: None, epoll_threads: None, paused: Arc::new(AtomicBool::new(false)), + paused_sync: Arc::new(Barrier::new(2)), }) } } @@ -844,10 +850,11 @@ impl VirtioDevice for Mem { }; let paused = self.paused.clone(); + let paused_sync = self.paused_sync.clone(); let mut epoll_threads = Vec::new(); thread::Builder::new() .name("virtio_mem".to_string()) - .spawn(move || handler.run(paused)) + .spawn(move || handler.run(paused, paused_sync)) .map(|thread| epoll_threads.push(thread)) .map_err(|e| { error!("failed to clone virtio-mem epoll thread: {}", e); diff --git a/virtio-devices/src/net.rs b/virtio-devices/src/net.rs index 705343be8..5cca455a3 100644 --- a/virtio-devices/src/net.rs +++ b/virtio-devices/src/net.rs @@ -28,7 +28,7 @@ use std::num::Wrapping; use std::os::unix::io::AsRawFd; use std::result; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::Arc; +use std::sync::{Arc, Barrier}; use std::thread; use std::vec::Vec; use virtio_bindings::bindings::virtio_net::*; @@ -134,7 +134,11 @@ impl NetEpollHandler { Ok(()) } - fn run(&mut self, paused: Arc) -> result::Result<(), EpollHelperError> { + fn run( + &mut self, + paused: Arc, + paused_sync: Arc, + ) -> result::Result<(), EpollHelperError> { let mut helper = EpollHelper::new(&self.kill_evt, &self.pause_evt)?; helper.add_event(self.queue_evt_pair[0].as_raw_fd(), RX_QUEUE_EVENT)?; helper.add_event(self.queue_evt_pair[1].as_raw_fd(), TX_QUEUE_EVENT)?; @@ -153,7 +157,7 @@ impl NetEpollHandler { // The NetQueuePair needs the epoll fd. self.net.epoll_fd = Some(helper.as_raw_fd()); - helper.run(paused, self)?; + helper.run(paused, paused_sync, self)?; Ok(()) } @@ -205,6 +209,7 @@ pub struct Net { epoll_threads: Option>>>, ctrl_queue_epoll_thread: Option>, paused: Arc, + paused_sync: Arc, queue_size: Vec, counters: NetCounters, seccomp_action: SeccompAction, @@ -265,6 +270,7 @@ impl Net { epoll_threads: None, ctrl_queue_epoll_thread: None, paused: Arc::new(AtomicBool::new(false)), + paused_sync: Arc::new(Barrier::new((num_queues / 2) + 1)), queue_size: vec![queue_size; queue_num], counters: NetCounters::default(), seccomp_action, @@ -419,6 +425,12 @@ impl VirtioDevice for Net { }; let paused = self.paused.clone(); + // Let's update the barrier as we need 1 for each RX/TX pair + + // 1 for the control queue + 1 for the main thread signalling + // the pause. + self.paused_sync = Arc::new(Barrier::new(taps.len() + 2)); + let paused_sync = self.paused_sync.clone(); + // Retrieve seccomp filter for virtio_net thread let virtio_net_seccomp_filter = get_seccomp_filter(&self.seccomp_action, Thread::VirtioNet) @@ -428,7 +440,7 @@ impl VirtioDevice for Net { .spawn(move || { if let Err(e) = SeccompFilter::apply(virtio_net_seccomp_filter) { error!("Error applying seccomp filter: {:?}", e); - } else if let Err(e) = ctrl_handler.run_ctrl(paused) { + } else if let Err(e) = ctrl_handler.run_ctrl(paused, paused_sync) { error!("Error running worker: {:?}", e); } }) @@ -477,9 +489,10 @@ impl VirtioDevice for Net { }; let paused = self.paused.clone(); + let paused_sync = self.paused_sync.clone(); thread::Builder::new() .name("virtio_net".to_string()) - .spawn(move || handler.run(paused)) + .spawn(move || handler.run(paused, paused_sync)) .map(|thread| epoll_threads.push(thread)) .map_err(|e| { error!("failed to clone queue EventFd: {}", e); diff --git a/virtio-devices/src/net_util.rs b/virtio-devices/src/net_util.rs index c87a70704..18b47fbdf 100644 --- a/virtio-devices/src/net_util.rs +++ b/virtio-devices/src/net_util.rs @@ -10,7 +10,7 @@ use net_util::MacAddr; use serde::ser::{Serialize, SerializeStruct, Serializer}; use std::os::unix::io::{AsRawFd, RawFd}; use std::sync::atomic::AtomicBool; -use std::sync::Arc; +use std::sync::{Arc, Barrier}; use virtio_bindings::bindings::virtio_net::*; use vm_memory::{ ByteValued, Bytes, GuestAddressSpace, GuestMemoryAtomic, GuestMemoryError, GuestMemoryMmap, @@ -177,10 +177,11 @@ impl NetCtrlEpollHandler { pub fn run_ctrl( &mut self, paused: Arc, + paused_sync: Arc, ) -> std::result::Result<(), EpollHelperError> { let mut helper = EpollHelper::new(&self.kill_evt, &self.pause_evt)?; helper.add_event(self.ctrl_q.queue_evt.as_raw_fd(), CTRL_QUEUE_EVENT)?; - helper.run(paused, self)?; + helper.run(paused, paused_sync, self)?; Ok(()) } diff --git a/virtio-devices/src/pmem.rs b/virtio-devices/src/pmem.rs index f771cde66..5515f99e9 100644 --- a/virtio-devices/src/pmem.rs +++ b/virtio-devices/src/pmem.rs @@ -25,7 +25,7 @@ use std::mem::size_of; use std::os::unix::io::AsRawFd; use std::result; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::Arc; +use std::sync::{Arc, Barrier}; use std::thread; use vm_memory::{ Address, ByteValued, Bytes, GuestAddress, GuestAddressSpace, GuestMemoryAtomic, @@ -243,10 +243,14 @@ impl PmemEpollHandler { }) } - fn run(&mut self, paused: Arc) -> result::Result<(), EpollHelperError> { + fn run( + &mut self, + paused: Arc, + paused_sync: Arc, + ) -> result::Result<(), EpollHelperError> { let mut helper = EpollHelper::new(&self.kill_evt, &self.pause_evt)?; helper.add_event(self.queue_evt.as_raw_fd(), QUEUE_AVAIL_EVENT)?; - helper.run(paused, self)?; + helper.run(paused, paused_sync, self)?; Ok(()) } @@ -288,6 +292,7 @@ pub struct Pmem { interrupt_cb: Option>, epoll_threads: Option>>, paused: Arc, + paused_sync: Arc, mapping: UserspaceMapping, seccomp_action: SeccompAction, @@ -336,6 +341,7 @@ impl Pmem { interrupt_cb: None, epoll_threads: None, paused: Arc::new(AtomicBool::new(false)), + paused_sync: Arc::new(Barrier::new(2)), mapping, seccomp_action, _region, @@ -461,6 +467,7 @@ impl VirtioDevice for Pmem { }; let paused = self.paused.clone(); + let paused_sync = self.paused_sync.clone(); let mut epoll_threads = Vec::new(); // Retrieve seccomp filter for virtio_pmem thread let virtio_pmem_seccomp_filter = @@ -471,7 +478,7 @@ impl VirtioDevice for Pmem { .spawn(move || { if let Err(e) = SeccompFilter::apply(virtio_pmem_seccomp_filter) { error!("Error applying seccomp filter: {:?}", e); - } else if let Err(e) = handler.run(paused) { + } else if let Err(e) = handler.run(paused, paused_sync) { error!("Error running worker: {:?}", e); } }) diff --git a/virtio-devices/src/rng.rs b/virtio-devices/src/rng.rs index a6f03717c..3bd9d6ead 100644 --- a/virtio-devices/src/rng.rs +++ b/virtio-devices/src/rng.rs @@ -18,7 +18,7 @@ use std::io; use std::os::unix::io::AsRawFd; use std::result; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::Arc; +use std::sync::{Arc, Barrier}; use std::thread; use vm_memory::{Bytes, GuestAddressSpace, GuestMemoryAtomic, GuestMemoryMmap}; use vm_migration::{ @@ -88,10 +88,14 @@ impl RngEpollHandler { }) } - fn run(&mut self, paused: Arc) -> result::Result<(), EpollHelperError> { + fn run( + &mut self, + paused: Arc, + paused_sync: Arc, + ) -> result::Result<(), EpollHelperError> { let mut helper = EpollHelper::new(&self.kill_evt, &self.pause_evt)?; helper.add_event(self.queue_evt.as_raw_fd(), QUEUE_AVAIL_EVENT)?; - helper.run(paused, self)?; + helper.run(paused, paused_sync, self)?; Ok(()) } @@ -133,6 +137,7 @@ pub struct Rng { interrupt_cb: Option>, epoll_threads: Option>>, paused: Arc, + paused_sync: Arc, seccomp_action: SeccompAction, } @@ -169,6 +174,7 @@ impl Rng { interrupt_cb: None, epoll_threads: None, paused: Arc::new(AtomicBool::new(false)), + paused_sync: Arc::new(Barrier::new(2)), seccomp_action, }) } @@ -288,6 +294,7 @@ impl VirtioDevice for Rng { }; let paused = self.paused.clone(); + let paused_sync = self.paused_sync.clone(); let mut epoll_threads = Vec::new(); // Retrieve seccomp filter for virtio_rng thread let virtio_rng_seccomp_filter = @@ -298,7 +305,7 @@ impl VirtioDevice for Rng { .spawn(move || { if let Err(e) = SeccompFilter::apply(virtio_rng_seccomp_filter) { error!("Error applying seccomp filter: {:?}", e); - } else if let Err(e) = handler.run(paused) { + } else if let Err(e) = handler.run(paused, paused_sync) { error!("Error running worker: {:?}", e); } }) diff --git a/virtio-devices/src/vhost_user/blk.rs b/virtio-devices/src/vhost_user/blk.rs index 5f417b7bc..2ce73ce18 100644 --- a/virtio-devices/src/vhost_user/blk.rs +++ b/virtio-devices/src/vhost_user/blk.rs @@ -14,7 +14,7 @@ use std::mem; use std::os::unix::io::AsRawFd; use std::result; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::Arc; +use std::sync::{Arc, Barrier}; use std::thread; use std::vec::Vec; use vhost_rs::vhost_user::message::VhostUserConfigFlags; @@ -44,6 +44,7 @@ pub struct Blk { interrupt_cb: Option>, epoll_threads: Option>>>, paused: Arc, + paused_sync: Arc, } impl Blk { @@ -148,6 +149,7 @@ impl Blk { interrupt_cb: None, epoll_threads: None, paused: Arc::new(AtomicBool::new(false)), + paused_sync: Arc::new(Barrier::new(vu_cfg.num_queues + 1)), }) } } @@ -273,9 +275,10 @@ impl VirtioDevice for Blk { }); let paused = self.paused.clone(); + let paused_sync = self.paused_sync.clone(); thread::Builder::new() .name("vhost_user_blk".to_string()) - .spawn(move || handler.run(paused)) + .spawn(move || handler.run(paused, paused_sync)) .map(|thread| epoll_threads.push(thread)) .map_err(|e| { error!("failed to clone virtio epoll thread: {}", e); diff --git a/virtio-devices/src/vhost_user/fs.rs b/virtio-devices/src/vhost_user/fs.rs index 8f03b35ce..968a1baff 100644 --- a/virtio-devices/src/vhost_user/fs.rs +++ b/virtio-devices/src/vhost_user/fs.rs @@ -13,7 +13,7 @@ use std::io; use std::os::unix::io::{AsRawFd, RawFd}; use std::result; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::{Arc, Mutex}; +use std::sync::{Arc, Barrier, Mutex}; use std::thread; use vhost_rs::vhost_user::message::{ VhostUserFSSlaveMsg, VhostUserFSSlaveMsgFlags, VhostUserProtocolFeatures, @@ -280,6 +280,7 @@ pub struct Fs { interrupt_cb: Option>, epoll_threads: Option>>>, paused: Arc, + paused_sync: Arc, } impl Fs { @@ -365,6 +366,7 @@ impl Fs { interrupt_cb: None, epoll_threads: None, paused: Arc::new(AtomicBool::new(false)), + paused_sync: Arc::new(Barrier::new(2)), }) } } @@ -500,10 +502,11 @@ impl VirtioDevice for Fs { }); let paused = self.paused.clone(); + let paused_sync = self.paused_sync.clone(); let mut epoll_threads = Vec::new(); thread::Builder::new() .name("virtio_fs".to_string()) - .spawn(move || handler.run(paused)) + .spawn(move || handler.run(paused, paused_sync)) .map(|thread| epoll_threads.push(thread)) .map_err(|e| { error!("failed to clone queue EventFd: {}", e); diff --git a/virtio-devices/src/vhost_user/handler.rs b/virtio-devices/src/vhost_user/handler.rs index 33f007ff6..cc2bb7c8c 100644 --- a/virtio-devices/src/vhost_user/handler.rs +++ b/virtio-devices/src/vhost_user/handler.rs @@ -17,7 +17,7 @@ use vmm_sys_util::eventfd::EventFd; use crate::VirtioInterrupt; use std::os::unix::io::AsRawFd; use std::sync::atomic::AtomicBool; -use std::sync::Arc; +use std::sync::{Arc, Barrier}; use vhost_rs::vhost_user::{MasterReqHandler, VhostUserMasterReqHandler}; /// Collection of common parameters required by vhost-user devices while @@ -67,7 +67,11 @@ impl VhostUserEpollHandler { .map_err(Error::FailedSignalingUsedQueue) } - pub fn run(&mut self, paused: Arc) -> std::result::Result<(), EpollHelperError> { + pub fn run( + &mut self, + paused: Arc, + paused_sync: Arc, + ) -> std::result::Result<(), EpollHelperError> { let mut helper = EpollHelper::new(&self.vu_epoll_cfg.kill_evt, &self.vu_epoll_cfg.pause_evt)?; @@ -81,7 +85,7 @@ impl VhostUserEpollHandler { helper.add_event(self_req_handler.as_raw_fd(), self.slave_evt_idx)?; } - helper.run(paused, self)?; + helper.run(paused, paused_sync, self)?; Ok(()) } diff --git a/virtio-devices/src/vhost_user/net.rs b/virtio-devices/src/vhost_user/net.rs index 4b7234471..77667c081 100644 --- a/virtio-devices/src/vhost_user/net.rs +++ b/virtio-devices/src/vhost_user/net.rs @@ -16,7 +16,7 @@ use net_util::MacAddr; use std::os::unix::io::AsRawFd; use std::result; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::Arc; +use std::sync::{Arc, Barrier}; use std::thread; use std::vec::Vec; use vhost_rs::vhost_user::message::{VhostUserProtocolFeatures, VhostUserVirtioFeatures}; @@ -48,6 +48,7 @@ pub struct Net { epoll_threads: Option>>>, ctrl_queue_epoll_thread: Option>>, paused: Arc, + paused_sync: Arc, } impl Net { @@ -153,6 +154,7 @@ impl Net { epoll_threads: None, ctrl_queue_epoll_thread: None, paused: Arc::new(AtomicBool::new(false)), + paused_sync: Arc::new(Barrier::new((vu_cfg.num_queues / 2) + 1)), }) } } @@ -259,9 +261,14 @@ impl VirtioDevice for Net { }; let paused = self.paused.clone(); + // Let's update the barrier as we need 1 for each RX/TX pair + + // 1 for the control queue + 1 for the main thread signalling + // the pause. + self.paused_sync = Arc::new(Barrier::new((queue_num / 2) + 2)); + let paused_sync = self.paused_sync.clone(); thread::Builder::new() .name("virtio_net".to_string()) - .spawn(move || ctrl_handler.run_ctrl(paused)) + .spawn(move || ctrl_handler.run_ctrl(paused, paused_sync)) .map(|thread| self.ctrl_queue_epoll_thread = Some(thread)) .map_err(|e| { error!("failed to clone queue EventFd: {}", e); @@ -294,9 +301,10 @@ impl VirtioDevice for Net { }); let paused = self.paused.clone(); + let paused_sync = self.paused_sync.clone(); thread::Builder::new() .name("vhost_user_net".to_string()) - .spawn(move || handler.run(paused)) + .spawn(move || handler.run(paused, paused_sync)) .map(|thread| epoll_threads.push(thread)) .map_err(|e| { error!("failed to clone queue EventFd: {}", e); diff --git a/virtio-devices/src/vsock/device.rs b/virtio-devices/src/vsock/device.rs index 5ff505932..a43b25c80 100644 --- a/virtio-devices/src/vsock/device.rs +++ b/virtio-devices/src/vsock/device.rs @@ -43,7 +43,7 @@ use std::os::unix::io::AsRawFd; use std::path::PathBuf; use std::result; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::{Arc, RwLock}; +use std::sync::{Arc, Barrier, RwLock}; use std::thread; use vm_memory::{GuestAddressSpace, GuestMemoryAtomic, GuestMemoryMmap}; use vm_migration::{ @@ -193,13 +193,17 @@ where } } - fn run(&mut self, paused: Arc) -> result::Result<(), EpollHelperError> { + fn run( + &mut self, + paused: Arc, + paused_sync: Arc, + ) -> result::Result<(), EpollHelperError> { let mut helper = EpollHelper::new(&self.kill_evt, &self.pause_evt)?; helper.add_event(self.queue_evts[0].as_raw_fd(), RX_QUEUE_EVENT)?; helper.add_event(self.queue_evts[1].as_raw_fd(), TX_QUEUE_EVENT)?; helper.add_event(self.queue_evts[2].as_raw_fd(), EVT_QUEUE_EVENT)?; helper.add_event(self.backend.read().unwrap().get_polled_fd(), BACKEND_EVENT)?; - helper.run(paused, self)?; + helper.run(paused, paused_sync, self)?; Ok(()) } @@ -303,6 +307,7 @@ pub struct Vsock { interrupt_cb: Option>, epoll_threads: Option>>>, paused: Arc, + paused_sync: Arc, path: PathBuf, } @@ -343,6 +348,7 @@ where interrupt_cb: None, epoll_threads: None, paused: Arc::new(AtomicBool::new(false)), + paused_sync: Arc::new(Barrier::new(2)), path, }) } @@ -476,10 +482,11 @@ where }; let paused = self.paused.clone(); + let paused_sync = self.paused_sync.clone(); let mut epoll_threads = Vec::new(); thread::Builder::new() .name("virtio_vsock".to_string()) - .spawn(move || handler.run(paused)) + .spawn(move || handler.run(paused, paused_sync)) .map(|thread| epoll_threads.push(thread)) .map_err(|e| { error!("failed to clone the vsock epoll thread: {}", e);