fix(vsock): pass correct index when triggering interrupts

We were confusing queue indexex with event indexes, when passing the
index of the queue that needed to be triggered after handling events.

Fix the logic to pass the correct index. This refactors a bit the code
to signal the queues in each event handler method. With MMIO we don't
need to signal each queue independently (one signal will cause the guest
to scan all queues). With PCI though, we are using MSI-X, so we need to
signal each queue independently.

Also, change vsock functional integration tests to also run for
PCI-enabled microVMs.

Signed-off-by: Babis Chalios <bchalios@amazon.es>

Author: bchalios

src/vmm/src/devices/virtio/vsock/device.rs

@@ -380,7 +380,7 @@ where
         // `TRANSPORT_RESET_EVENT` event we sent during snapshot creation.
         if self.is_activated() {
             info!("kick vsock {}.", self.id());
-            self.signal_used_queue(0).unwrap();
+            self.signal_used_queue(EVQ_INDEX).unwrap();
         }
     }
 }

src/vmm/src/devices/virtio/vsock/event_handler.rs

@@ -46,75 +46,80 @@ where
     const PROCESS_EVQ: u32 = 3;
     const PROCESS_NOTIFY_BACKEND: u32 = 4;
 
-    pub fn handle_rxq_event(&mut self, evset: EventSet) -> bool {
+    pub fn handle_rxq_event(&mut self, evset: EventSet) {
         if evset != EventSet::IN {
             warn!("vsock: rxq unexpected event {:?}", evset);
             METRICS.rx_queue_event_fails.inc();
-            return false;
+            return;
         }
 
-        let mut raise_irq = false;
         if let Err(err) = self.queue_events[RXQ_INDEX].read() {
             error!("Failed to get vsock rx queue event: {:?}", err);
             METRICS.rx_queue_event_fails.inc();
         } else if self.backend.has_pending_rx() {
-            raise_irq |= self.process_rx();
+            if self.process_rx() {
+                self.signal_used_queue(RXQ_INDEX)
+                    .expect("vsock: Could not trigger device interrupt or RX queue");
+            }
             METRICS.rx_queue_event_count.inc();
         }
-        raise_irq
     }
 
-    pub fn handle_txq_event(&mut self, evset: EventSet) -> bool {
+    pub fn handle_txq_event(&mut self, evset: EventSet) {
         if evset != EventSet::IN {
             warn!("vsock: txq unexpected event {:?}", evset);
             METRICS.tx_queue_event_fails.inc();
-            return false;
+            return;
         }
 
-        let mut raise_irq = false;
         if let Err(err) = self.queue_events[TXQ_INDEX].read() {
             error!("Failed to get vsock tx queue event: {:?}", err);
             METRICS.tx_queue_event_fails.inc();
         } else {
-            raise_irq |= self.process_tx();
+            if self.process_tx() {
+                self.signal_used_queue(TXQ_INDEX)
+                    .expect("vsock: Could not trigger device interrupt or TX queue");
+            }
             METRICS.tx_queue_event_count.inc();
             // The backend may have queued up responses to the packets we sent during
             // TX queue processing. If that happened, we need to fetch those responses
             // and place them into RX buffers.
-            if self.backend.has_pending_rx() {
-                raise_irq |= self.process_rx();
+            if self.backend.has_pending_rx() && self.process_rx() {
+                self.signal_used_queue(RXQ_INDEX)
+                    .expect("vsock: Could not trigger device interrupt or RX queue");
             }
         }
-        raise_irq
     }
 
-    pub fn handle_evq_event(&mut self, evset: EventSet) -> bool {
+    pub fn handle_evq_event(&mut self, evset: EventSet) {
         if evset != EventSet::IN {
             warn!("vsock: evq unexpected event {:?}", evset);
             METRICS.ev_queue_event_fails.inc();
-            return false;
+            return;
         }
 
         if let Err(err) = self.queue_events[EVQ_INDEX].read() {
             error!("Failed to consume vsock evq event: {:?}", err);
             METRICS.ev_queue_event_fails.inc();
         }
-        false
     }
 
     /// Notify backend of new events.
-    pub fn notify_backend(&mut self, evset: EventSet) -> bool {
+    pub fn notify_backend(&mut self, evset: EventSet) {
         self.backend.notify(evset);
         // After the backend has been kicked, it might've freed up some resources, so we
         // can attempt to send it more data to process.
         // In particular, if `self.backend.send_pkt()` halted the TX queue processing (by
         // returning an error) at some point in the past, now is the time to try walking the
         // TX queue again.
-        let mut raise_irq = self.process_tx();
-        if self.backend.has_pending_rx() {
-            raise_irq |= self.process_rx();
+        if self.process_tx() {
+            self.signal_used_queue(TXQ_INDEX)
+                .expect("vsock: Could not trigger device interrupt or TX queue");
+        }
+        if self.backend.has_pending_rx() && self.process_rx() {
+            self.signal_used_queue(RXQ_INDEX)
+                .expect("vsock: Could not trigger device interrupt or RX queue");
         }
-        raise_irq
     }
 
     fn register_runtime_events(&self, ops: &mut EventOps) {
@@ -182,19 +187,14 @@ where
         let evset = event.event_set();
 
         if self.is_activated() {
-            let mut raise_irq = false;
             match source {
                 Self::PROCESS_ACTIVATE => self.handle_activate_event(ops),
-                Self::PROCESS_RXQ => raise_irq = self.handle_rxq_event(evset),
-                Self::PROCESS_TXQ => raise_irq = self.handle_txq_event(evset),
-                Self::PROCESS_EVQ => raise_irq = self.handle_evq_event(evset),
-                Self::PROCESS_NOTIFY_BACKEND => raise_irq = self.notify_backend(evset),
+                Self::PROCESS_RXQ => self.handle_rxq_event(evset),
+                Self::PROCESS_TXQ => self.handle_txq_event(evset),
+                Self::PROCESS_EVQ => self.handle_evq_event(evset),
+                Self::PROCESS_NOTIFY_BACKEND => self.notify_backend(evset),
                 _ => warn!("Unexpected vsock event received: {:?}", source),
             };
-            if raise_irq {
-                self.signal_used_queue(source as usize)
-                    .expect("vsock: Could not trigger device interrupt");
-            }
         } else {
             warn!(
                 "Vsock: The device is not yet activated. Spurious event received: {:?}",
@@ -302,7 +302,9 @@ mod tests {
             let mut ctx = test_ctx.create_event_handler_context();
             ctx.mock_activate(test_ctx.mem.clone(), test_ctx.interrupt.clone());
 
-            assert!(!ctx.device.handle_txq_event(EventSet::IN));
+            let metric_before = METRICS.tx_queue_event_fails.count();
+            ctx.device.handle_txq_event(EventSet::IN);
+            assert_eq!(metric_before + 1, METRICS.tx_queue_event_fails.count());
         }
     }
 
@@ -363,7 +365,9 @@ mod tests {
             let mut ctx = test_ctx.create_event_handler_context();
             ctx.mock_activate(test_ctx.mem.clone(), test_ctx.interrupt.clone());
             ctx.device.backend.set_pending_rx(false);
-            assert!(!ctx.device.handle_rxq_event(EventSet::IN));
+            let metric_before = METRICS.rx_queue_event_fails.count();
+            ctx.device.handle_rxq_event(EventSet::IN);
+            assert_eq!(metric_before + 1, METRICS.rx_queue_event_fails.count());
         }
     }
 
@@ -374,7 +378,9 @@ mod tests {
             let test_ctx = TestContext::new();
             let mut ctx = test_ctx.create_event_handler_context();
             ctx.device.backend.set_pending_rx(false);
-            assert!(!ctx.device.handle_evq_event(EventSet::IN));
+            let metric_before = METRICS.ev_queue_event_fails.count();
+            ctx.device.handle_evq_event(EventSet::IN);
+            assert_eq!(metric_before + 1, METRICS.ev_queue_event_fails.count());
         }
     }
 

tests/integration_tests/functional/test_vsock.py

@@ -37,15 +37,15 @@
 TEST_WORKER_COUNT = 10
 
 
-def test_vsock(uvm_plain_any, bin_vsock_path, test_fc_session_root_path):
+def test_vsock(uvm_plain_any, pci_enabled, bin_vsock_path, test_fc_session_root_path):
     """
     Test guest and host vsock initiated connections.
 
     Check the module docstring for details on the setup.
     """
 
     vm = uvm_plain_any
-    vm.spawn()
+    vm.spawn(pci=pci_enabled)
 
     vm.basic_config()
     vm.add_net_iface()
@@ -102,12 +102,12 @@ def negative_test_host_connections(vm, blob_path, blob_hash):
     validate_fc_metrics(metrics)
 
 
-def test_vsock_epipe(uvm_plain, bin_vsock_path, test_fc_session_root_path):
+def test_vsock_epipe(uvm_plain, pci_enabled, bin_vsock_path, test_fc_session_root_path):
     """
     Vsock negative test to validate SIGPIPE/EPIPE handling.
     """
     vm = uvm_plain
-    vm.spawn()
+    vm.spawn(pci=pci_enabled)
     vm.basic_config()
     vm.add_net_iface()
     vm.api.vsock.put(vsock_id="vsock0", guest_cid=3, uds_path=f"/{VSOCK_UDS_PATH}")
@@ -129,7 +129,7 @@ def test_vsock_epipe(uvm_plain, bin_vsock_path, test_fc_session_root_path):
 
 
 def test_vsock_transport_reset_h2g(
-    uvm_nano, microvm_factory, bin_vsock_path, test_fc_session_root_path
+    uvm_plain, pci_enabled, microvm_factory, bin_vsock_path, test_fc_session_root_path
 ):
     """
     Vsock transport reset test.
@@ -146,7 +146,9 @@ def test_vsock_transport_reset_h2g(
     6. Close VM -> Load VM from Snapshot -> check that vsock
        device is still working.
     """
-    test_vm = uvm_nano
+    test_vm = uvm_plain
+    test_vm.spawn(pci=pci_enabled)
+    test_vm.basic_config(vcpu_count=2, mem_size_mib=256)
     test_vm.add_net_iface()
     test_vm.api.vsock.put(vsock_id="vsock0", guest_cid=3, uds_path=f"/{VSOCK_UDS_PATH}")
     test_vm.start()
@@ -213,11 +215,13 @@ def test_vsock_transport_reset_h2g(
     validate_fc_metrics(metrics)
 
 
-def test_vsock_transport_reset_g2h(uvm_nano, microvm_factory):
+def test_vsock_transport_reset_g2h(uvm_plain, pci_enabled, microvm_factory):
     """
     Vsock transport reset test.
     """
-    test_vm = uvm_nano
+    test_vm = uvm_plain
+    test_vm.spawn(pci=pci_enabled)
+    test_vm.basic_config(vcpu_count=2, mem_size_mib=256)
     test_vm.add_net_iface()
     test_vm.api.vsock.put(vsock_id="vsock0", guest_cid=3, uds_path=f"/{VSOCK_UDS_PATH}")
     test_vm.start()