fix virtio console reactivation after reset

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

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:?}"),
         }
     }
 

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

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