diff --git a/src/devices/src/virtio/balloon/device.rs b/src/devices/src/virtio/balloon/device.rs index 96d21e1445..eb7e47ea06 100644 --- a/src/devices/src/virtio/balloon/device.rs +++ b/src/devices/src/virtio/balloon/device.rs @@ -240,4 +240,28 @@ impl VirtioDevice for Balloon { fn is_activated(&self) -> bool { self.device_state.is_activated() } + + fn reset(&mut self) -> bool { + self.queues = None; + self.stats_desc_index = None; + self.device_state = DeviceState::Inactive; + true + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn reset_drops_guest_queue_state_for_reactivation() { + let mut balloon = Balloon::new(MetricsWriter::default(), None).unwrap(); + balloon.queues = Some(Vec::new()); + balloon.stats_desc_index = Some(7); + + assert!(balloon.reset()); + assert!(balloon.queues.is_none()); + assert_eq!(balloon.stats_desc_index, None); + assert!(!balloon.is_activated()); + } } diff --git a/src/devices/src/virtio/balloon/event_handler.rs b/src/devices/src/virtio/balloon/event_handler.rs index 259eeaa7e1..9a117adc81 100644 --- a/src/devices/src/virtio/balloon/event_handler.rs +++ b/src/devices/src/virtio/balloon/event_handler.rs @@ -237,6 +237,13 @@ impl Balloon { let frq = eventfd_pollable(self.queue_event(FRQ_INDEX)); let stats_timer = timerfd_pollable(&self.stats_timer); + // Reset keeps event-manager subscriptions alive while dropping the + // guest queues. Remove stale registrations before reusing the stable + // queue eventfds for a new activation. + for source in [ifq, dfq, stq, phq, frq, stats_timer] { + let _ = event_manager.unregister(source); + } + event_manager .register(ifq, pollable_event(ifq), self_subscriber.clone()) .unwrap_or_else(|e| { @@ -279,40 +286,41 @@ impl Balloon { error!("Failed to register balloon frq with event manager: {e:?}"); }); - event_manager.unregister(activate_evt).unwrap_or_else(|e| { - error!("Failed to unregister balloon activate evt: {e:?}"); - }) + // Keep the activation event registered. A virtio reset drops the + // queues and a later guest activation reuses this eventfd. } } impl Subscriber for Balloon { fn process(&mut self, event: &EpollEvent, event_manager: &mut EventManager) { let source = event.fd(); + let activate_evt = eventfd_pollable(&self.activate_evt); + if source == activate_evt { + self.handle_activate_event(event_manager); + return; + } + if !self.is_activated() { + warn!("balloon: The device is not yet activated. Spurious event received: {source:?}"); + return; + } + let ifq = eventfd_pollable(self.queue_event(IFQ_INDEX)); let dfq = eventfd_pollable(self.queue_event(DFQ_INDEX)); let stq = eventfd_pollable(self.queue_event(STQ_INDEX)); let phq = eventfd_pollable(self.queue_event(PHQ_INDEX)); let frq = eventfd_pollable(self.queue_event(FRQ_INDEX)); - let activate_evt = eventfd_pollable(&self.activate_evt); let stats_timer = timerfd_pollable(&self.stats_timer); - if self.is_activated() { - match source { - _ if source == ifq => self.handle_ifq_event(event), - _ if source == dfq => self.handle_dfq_event(event), - _ if source == stq => self.handle_stq_event(event), - _ if self.stats_enabled() && source == stats_timer => { - self.handle_stats_timer_event(event) - } - _ if source == phq => self.handle_phq_event(event), - _ if source == frq => self.handle_frq_event(event), - _ if source == activate_evt => { - self.handle_activate_event(event_manager); - } - _ => warn!("Unexpected balloon event received: {source:?}"), + match source { + _ if source == ifq => self.handle_ifq_event(event), + _ if source == dfq => self.handle_dfq_event(event), + _ if source == stq => self.handle_stq_event(event), + _ if self.stats_enabled() && source == stats_timer => { + self.handle_stats_timer_event(event) } - } else { - warn!("balloon: The device is not yet activated. Spurious event received: {source:?}"); + _ if source == phq => self.handle_phq_event(event), + _ if source == frq => self.handle_frq_event(event), + _ => warn!("Unexpected balloon event received: {source:?}"), } } diff --git a/src/devices/src/virtio/console/event_handler.rs b/src/devices/src/virtio/console/event_handler.rs index af9f4c551c..65f1048064 100644 --- a/src/devices/src/virtio/console/event_handler.rs +++ b/src/devices/src/virtio/console/event_handler.rs @@ -53,6 +53,10 @@ impl Console { for queue_index in 0..self.queues.len() { let queue_evt = eventfd_pollable(&self.queue_events[queue_index]); + // A virtio reset leaves the event-manager subscription alive even + // though the guest will activate the same queue eventfd again. + // Remove that stale subscription before re-registering it. + let _ = event_manager.unregister(queue_evt); event_manager .register( queue_evt, @@ -66,9 +70,8 @@ impl Console { }); } - event_manager.unregister(activate_evt).unwrap_or_else(|e| { - error!("Failed to unregister fs activate evt: {e:?}"); - }) + // Keep the activation event registered. Virtio device reset is a valid + // lifecycle transition and a later activation reuses this eventfd. } fn handle_sigwinch_event(&mut self, event: &EpollEvent) { @@ -106,46 +109,53 @@ impl Console { impl Subscriber for Console { fn process(&mut self, event: &EpollEvent, event_manager: &mut EventManager) { let source = event.fd(); + let activate_evt = eventfd_pollable(&self.activate_evt); + let sigwinch_evt = eventfd_pollable(&self.sigwinch_evt); + + if source == activate_evt { + self.handle_activate_event(event_manager); + return; + } + if source == sigwinch_evt { + self.handle_sigwinch_event(event); + return; + } + if !self.is_activated() { + warn!("console: The device is not yet activated. Spurious event received: {source:?}"); + return; + } + if self.queue_events.len() <= CONTROL_RXQ_INDEX.max(CONTROL_TXQ_INDEX) { + warn!("console: queue event received while queues are reset: {source:?}"); + return; + } let control_rxq = eventfd_pollable(&self.queue_events[CONTROL_RXQ_INDEX]); let control_txq = eventfd_pollable(&self.queue_events[CONTROL_TXQ_INDEX]); let control_rxq_control = eventfd_pollable(self.control.queue_evt()); - - let activate_evt = eventfd_pollable(&self.activate_evt); - let sigwinch_evt = eventfd_pollable(&self.sigwinch_evt); - - if self.is_activated() { - let mut raise_irq = false; - - if source == control_txq { - raise_irq |= - self.read_queue_event(CONTROL_TXQ_INDEX, event) && self.process_control_tx() - } else if source == control_rxq_control { - self.read_control_queue_event(event); - raise_irq |= self.process_control_rx(); - } else if source == control_rxq { - raise_irq |= self.read_queue_event(CONTROL_RXQ_INDEX, event) - } - /* Guest signaled input/output on port */ - else if let Some(queue_index) = self - .queue_events - .iter() - .position(|fd| eventfd_pollable(fd) == source) - { - raise_irq |= self.read_queue_event(queue_index, event); - self.notify_port_queue_event(queue_index); - } else if source == activate_evt { - self.handle_activate_event(event_manager); - } else if source == sigwinch_evt { - self.handle_sigwinch_event(event); - } else { - log::warn!("Unexpected console event received: {source:?}") - } - if raise_irq { - self.device_state.signal_used_queue(); - } + let mut raise_irq = false; + + if source == control_txq { + raise_irq |= + self.read_queue_event(CONTROL_TXQ_INDEX, event) && self.process_control_tx() + } else if source == control_rxq_control { + self.read_control_queue_event(event); + raise_irq |= self.process_control_rx(); + } else if source == control_rxq { + raise_irq |= self.read_queue_event(CONTROL_RXQ_INDEX, event) + } + /* Guest signaled input/output on port */ + else if let Some(queue_index) = self + .queue_events + .iter() + .position(|fd| eventfd_pollable(fd) == source) + { + raise_irq |= self.read_queue_event(queue_index, event); + self.notify_port_queue_event(queue_index); } else { - warn!("console: The device is not yet activated. Spurious event received: {source:?}"); + log::warn!("Unexpected console event received: {source:?}") + } + if raise_irq { + self.device_state.signal_used_queue(); } } @@ -161,3 +171,72 @@ impl Subscriber for Console { ] } } + +#[cfg(test)] +mod tests { + use std::sync::{Arc, Mutex}; + + use super::*; + use crate::virtio::PortDescription; + + #[test] + fn activation_event_remains_registered_for_device_reactivation() { + let mut console = Console::new(vec![PortDescription { + name: "agent".into(), + input: None, + output: None, + terminal: None, + queue_size: 32, + }]) + .unwrap(); + console.queues = (0..2).map(|_| None).collect(); + console.queue_events = (0..2) + .map(|_| Arc::new(utils::eventfd::EventFd::new(utils::eventfd::EFD_NONBLOCK).unwrap())) + .collect(); + let console = Arc::new(Mutex::new(console)); + let mut event_manager = EventManager::new().unwrap(); + event_manager.add_subscriber(console.clone()).unwrap(); + let activate_evt = eventfd_pollable(&console.lock().unwrap().activate_evt); + + console + .lock() + .unwrap() + .handle_activate_event(&mut event_manager); + + console + .lock() + .unwrap() + .handle_activate_event(&mut event_manager); + + assert!(event_manager.subscriber(activate_evt).is_ok()); + for queue_evt in &console.lock().unwrap().queue_events { + assert!(event_manager + .subscriber(eventfd_pollable(queue_evt)) + .is_ok()); + } + } + + #[test] + fn activation_event_is_safe_while_queue_vectors_are_empty() { + let console = Arc::new(Mutex::new( + Console::new(vec![PortDescription { + name: "agent".into(), + input: None, + output: None, + terminal: None, + queue_size: 32, + }]) + .unwrap(), + )); + let mut event_manager = EventManager::new().unwrap(); + event_manager.add_subscriber(console.clone()).unwrap(); + let activate_evt = eventfd_pollable(&console.lock().unwrap().activate_evt); + + console.lock().unwrap().process( + &EpollEvent::new(EventSet::IN, pollable_token(activate_evt)), + &mut event_manager, + ); + + assert!(event_manager.subscriber(activate_evt).is_ok()); + } +} diff --git a/src/devices/src/virtio/console/port.rs b/src/devices/src/virtio/console/port.rs index 3ef969906e..664ff0b166 100644 --- a/src/devices/src/virtio/console/port.rs +++ b/src/devices/src/virtio/console/port.rs @@ -179,33 +179,64 @@ impl Port { } pub fn shutdown(&mut self) { - if let PortState::Active { + let PortState::Active { stopfd, stop, tx_thread, rx_thread, - } = &mut self.state - { - stop.store(true, Ordering::Release); - if let Some(tx_thread) = mem::take(tx_thread) { - tx_thread.thread().unpark(); - if let Err(e) = tx_thread.join() { - log::error!( - "Failed to flush tx for port {port_id}, thread panicked: {e:?}", - port_id = self.port_id - ) - } + } = mem::replace(&mut self.state, PortState::Inactive) + else { + return; + }; + + stop.store(true, Ordering::Release); + if let Some(tx_thread) = tx_thread { + tx_thread.thread().unpark(); + if let Err(e) = tx_thread.join() { + log::error!( + "Failed to flush tx for port {port_id}, thread panicked: {e:?}", + port_id = self.port_id + ) } - stopfd.write(1).unwrap(); - if let Some(rx_thread) = mem::take(rx_thread) { - rx_thread.thread().unpark(); - if let Err(e) = rx_thread.join() { - log::error!( - "Failed to flush tx for port {port_id}, thread panicked: {e:?}", - port_id = self.port_id - ) - } + } + stopfd.write(1).unwrap(); + if let Some(rx_thread) = rx_thread { + rx_thread.thread().unpark(); + if let Err(e) = rx_thread.join() { + log::error!( + "Failed to flush tx for port {port_id}, thread panicked: {e:?}", + port_id = self.port_id + ) } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn shutdown_marks_an_active_port_inactive() { + let mut port = Port::new( + 0, + PortDescription { + name: "agent".into(), + input: None, + output: None, + terminal: None, + queue_size: DEFAULT_QUEUE_SIZE, + }, + ); + port.state = PortState::Active { + stopfd: utils::eventfd::EventFd::new(utils::eventfd::EFD_NONBLOCK).unwrap(), + stop: Arc::new(AtomicBool::new(false)), + rx_thread: None, + tx_thread: None, }; + + port.shutdown(); + + assert!(!port.is_active()); } } diff --git a/src/devices/src/virtio/mmio.rs b/src/devices/src/virtio/mmio.rs index d3387c70db..2d37472f8d 100644 --- a/src/devices/src/virtio/mmio.rs +++ b/src/devices/src/virtio/mmio.rs @@ -274,9 +274,10 @@ impl MmioTransport { // FIXME: check if activated! self.with_queue_mut(f); } else { + let device_name = self.locked_device().device_name().to_string(); warn!( - "update virtio queue in invalid state 0x{:x}", - self.device_status + "update virtio queue for {device_name} in invalid state 0x{:x}", + self.device_status, ); } } @@ -330,6 +331,11 @@ impl MmioTransport { #[allow(unused_assignments)] fn set_device_status(&mut self, status: u32) { use device_status::*; + let device_name = self.locked_device().device_name().to_string(); + debug!( + "virtio {device_name} status transition request: 0x{:x} -> 0x{status:x}", + self.device_status, + ); // match changed bits match !self.device_status & status { ACKNOWLEDGE if self.device_status == INIT => { @@ -365,8 +371,8 @@ impl MmioTransport { } _ => { warn!( - "invalid virtio driver status transition: 0x{:x} -> 0x{:x}", - self.device_status, status + "invalid virtio {device_name} driver status transition: 0x{:x} -> 0x{:x}", + self.device_status, status, ); } } diff --git a/src/devices/src/virtio/net/device.rs b/src/devices/src/virtio/net/device.rs index 7a77989ad6..90586f81e6 100644 --- a/src/devices/src/virtio/net/device.rs +++ b/src/devices/src/virtio/net/device.rs @@ -23,6 +23,8 @@ use std::io::Write; use std::os::fd::RawFd; #[cfg(unix)] use std::path::PathBuf; +use std::thread::JoinHandle; +use utils::eventfd::{EventFd, EFD_NONBLOCK}; use virtio_bindings::virtio_net::VIRTIO_NET_F_MAC; use virtio_bindings::virtio_ring::VIRTIO_RING_F_EVENT_IDX; use vm_memory::{ByteValued, GuestMemoryError, GuestMemoryMmap}; @@ -88,6 +90,9 @@ pub struct Net { pub(crate) device_state: DeviceState, + worker_stopfd: EventFd, + worker_thread: Option>, + config: VirtioNetConfig, } @@ -141,6 +146,8 @@ impl Net { acked_features: 0u64, device_state: DeviceState::Inactive, + worker_stopfd: EventFd::new(EFD_NONBLOCK).map_err(super::Error::EventFd)?, + worker_thread: None, config, }) } @@ -221,9 +228,13 @@ impl VirtioDevice for Net { self.acked_features, cfg_backend, &self.rate_limiters, + self.worker_stopfd.try_clone().map_err(|error| { + error!("Cannot clone virtio-net reset event: {error}"); + ActivateError::BadActivate + })?, ) { Ok(worker) => { - worker.run(); + self.worker_thread = Some(worker.run()); self.device_state = DeviceState::Activated(mem, interrupt); Ok(()) } @@ -240,4 +251,27 @@ impl VirtioDevice for Net { fn is_activated(&self) -> bool { self.device_state.is_activated() } + + fn reset(&mut self) -> bool { + let Some(worker) = self.worker_thread.take() else { + self.device_state = DeviceState::Inactive; + return true; + }; + if let Err(error) = self.worker_stopfd.write(1) { + error!("Cannot signal virtio-net worker reset: {error}"); + self.worker_thread = Some(worker); + return false; + } + match worker.join() { + Ok(backend) => { + self.cfg_backend = Some(backend); + self.device_state = DeviceState::Inactive; + true + } + Err(error) => { + error!("Cannot join virtio-net worker during reset: {error:?}"); + false + } + } + } } diff --git a/src/devices/src/virtio/net/worker.rs b/src/devices/src/virtio/net/worker.rs index b2bf58e0ff..c09216bee9 100644 --- a/src/devices/src/virtio/net/worker.rs +++ b/src/devices/src/virtio/net/worker.rs @@ -21,6 +21,7 @@ use std::os::fd::{AsRawFd, FromRawFd, OwnedFd}; #[cfg(windows)] use std::os::windows::io::{AsRawHandle, RawHandle}; use std::thread; +use std::thread::JoinHandle; use std::time::{Duration, Instant}; use std::{cmp, result}; use utils::epoll::{ControlOperation, Epoll, EpollEvent, EventSet}; @@ -38,6 +39,7 @@ const RX_QUEUE_EVENT: u64 = 0; const TX_QUEUE_EVENT: u64 = 1; const BACKEND_EVENT: u64 = 2; const RATE_LIMIT_TIMER_EVENT: u64 = 3; +const STOP_EVENT: u64 = 4; pub struct NetWorker { rx_q: DeviceQueue, @@ -63,6 +65,7 @@ pub struct NetWorker { rate_limit_timer: Option, armed_rate_limit_deadline: Option, + stop_evt: EventFd, } impl NetWorker { @@ -74,6 +77,7 @@ impl NetWorker { _vnet_features: u64, cfg_backend: VirtioNetBackend, rate_limiters: &RateLimiters, + stop_evt: EventFd, ) -> Result { let backend = match cfg_backend { #[cfg(unix)] @@ -148,17 +152,18 @@ impl NetWorker { rate_limit_timer, armed_rate_limit_deadline: None, + stop_evt, }) } - pub fn run(self) { + pub fn run(self) -> JoinHandle { thread::Builder::new() .name("virtio-net worker".into()) .spawn(|| self.work()) - .unwrap(); + .expect("failed to spawn virtio-net worker") } - fn work(mut self) { + fn work(mut self) -> VirtioNetBackend { let virtq_rx_ev = eventfd_pollable(&self.rx_q.event); let virtq_tx_ev = eventfd_pollable(&self.tx_q.event); let backend_source = self.backend.event_source(BACKEND_EVENT); @@ -166,9 +171,10 @@ impl NetWorker { Ok(pollable) => pollable, Err(err) => { log::error!("virtio-net backend event source is unsupported: {err}"); - return; + return VirtioNetBackend::Custom(self.backend); } }; + let stop_pollable = eventfd_pollable(&self.stop_evt); let epoll = Epoll::new().unwrap(); @@ -197,6 +203,11 @@ impl NetWorker { &EpollEvent::new(EventSet::IN, RATE_LIMIT_TIMER_EVENT), ); } + let _ = epoll.ctl( + ControlOperation::Add, + stop_pollable, + &EpollEvent::new(EventSet::IN, STOP_EVENT), + ); loop { let mut epoll_events = vec![EpollEvent::new(EventSet::empty(), 0); 32]; @@ -231,6 +242,14 @@ impl NetWorker { RATE_LIMIT_TIMER_EVENT if event_set.contains(EventSet::IN) => { self.process_rate_limit_timer_event(); } + STOP_EVENT if event_set.contains(EventSet::IN) => { + if let Err(error) = self.stop_evt.read() { + log::error!( + "Failed to consume virtio-net reset event: {error:?}" + ); + } + return VirtioNetBackend::Custom(self.backend); + } _ => { log::warn!( "Received unknown virtio-net event: {event_set:?} token={source}" @@ -825,9 +844,33 @@ mod tests { tx_resume_at: None, rate_limit_timer: None, armed_rate_limit_deadline: None, + stop_evt: EventFd::new(0).unwrap(), } } + #[test] + fn worker_stop_returns_backend_for_reactivation() { + let mem = GuestMemoryMmap::from_ranges(&[(GuestAddress(0), QUEUE_MEMORY_SIZE)]).unwrap(); + let rx_vq = VirtQueue::new(RX_QUEUE_ADDR, &mem, 8); + let tx_vq = VirtQueue::new(TX_QUEUE_ADDR, &mem, 8); + let worker = worker( + mem.clone(), + device_queue(rx_vq.create_queue()), + device_queue(tx_vq.create_queue()), + Arc::new(Mutex::new(BackendState::default())), + RateLimiters::default(), + Instant::now(), + ); + let stop_evt = worker.stop_evt.try_clone().unwrap(); + let handle = worker.run(); + + stop_evt.write(1).unwrap(); + assert!(matches!( + handle.join().unwrap(), + VirtioNetBackend::Custom(_) + )); + } + #[test] fn tx_retries_backpressure_without_double_charging_and_resumes_at_deadline() { let mem = GuestMemoryMmap::from_ranges(&[(GuestAddress(0), QUEUE_MEMORY_SIZE)]).unwrap();