fix: more compliant VIRTIO notifications

VIRTIO spec states:
```
After the device writes a descriptor index into the used ring:
  If the idx field in the used ring (which determined where that
  descriptor index was placed) was equal to used_event, the device
  MUST send a notification.
```
The current implementation does not follow this very precisely. It
bumps used ring index when new descriptors are added to the used
ring. But the check if the notification is needed is postponed to
later processing stage.
To be more VIRTIO spec compliant simply move the logic for updating
the used ring index into the later processing stage as well, just
before the check if the notification should be send.

Signed-off-by: Egor Lazarchuk <[email protected]>

Author: ShadowCurse

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

@@ -362,6 +362,7 @@ impl Balloon {
                 }
             }
         }
+        queue.advance_used_ring_idx();
 
         if needs_interrupt {
             self.signal_used_queue()?;
@@ -380,6 +381,7 @@ impl Balloon {
             queue.add_used(head.index, 0)?;
             needs_interrupt = true;
         }
+        queue.advance_used_ring_idx();
 
         if needs_interrupt {
             self.signal_used_queue()
@@ -453,6 +455,7 @@ impl Balloon {
         // and sending a used buffer notification
         if let Some(index) = self.stats_desc_index.take() {
             self.queues[STATS_INDEX].add_used(index, 0)?;
+            self.queues[STATS_INDEX].advance_used_ring_idx();
             self.signal_used_queue()
         } else {
             error!("Failed to update balloon stats, missing descriptor.");
@@ -606,8 +609,7 @@ impl VirtioDevice for Balloon {
 
     fn activate(&mut self, mem: GuestMemoryMmap) -> Result<(), ActivateError> {
         for q in self.queues.iter_mut() {
-            q.initialize(&mem)
-                .map_err(ActivateError::QueueError)?;
+            q.initialize(&mem).map_err(ActivateError::QueueError)?;
         }
 
         self.device_state = DeviceState::Activated(mem);

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

@@ -436,6 +436,7 @@ impl VirtioBlock {
                 }
             }
         }
+        queue.advance_used_ring_idx();
 
         if queue.prepare_kick() {
             self.irq_trigger
@@ -497,6 +498,7 @@ impl VirtioBlock {
                 }
             }
         }
+        queue.advance_used_ring_idx();
 
         if queue.prepare_kick() {
             self.irq_trigger

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

@@ -207,7 +207,7 @@ impl RxBuffers {
     /// This will let the guest know that about all the `DescriptorChain` object that has been
     /// used to receive a frame from the TAP.
     fn finish_frame(&mut self, rx_queue: &mut Queue) {
-        rx_queue.advance_used_ring(self.used_descriptors);
+        rx_queue.advance_next_used(self.used_descriptors);
         self.used_descriptors = 0;
         self.used_bytes = 0;
     }
@@ -396,6 +396,7 @@ impl Net {
             NetQueue::Rx => &mut self.queues[RX_INDEX],
             NetQueue::Tx => &mut self.queues[TX_INDEX],
         };
+        queue.advance_used_ring_idx();
 
         if queue.prepare_kick() {
             self.irq_trigger
@@ -1000,8 +1001,7 @@ impl VirtioDevice for Net {
 
     fn activate(&mut self, mem: GuestMemoryMmap) -> Result<(), ActivateError> {
         for q in self.queues.iter_mut() {
-            q.initialize(&mem)
-                .map_err(ActivateError::QueueError)?;
+            q.initialize(&mem).map_err(ActivateError::QueueError)?;
         }
 
         let event_idx = self.has_feature(u64::from(VIRTIO_RING_F_EVENT_IDX));

src/vmm/src/devices/virtio/queue.rs

@@ -653,20 +653,23 @@ impl Queue {
     }
 
     /// Advance queue and used ring by `n` elements.
-    pub fn advance_used_ring(&mut self, n: u16) {
+    pub fn advance_next_used(&mut self, n: u16) {
         self.num_added += Wrapping(n);
         self.next_used += Wrapping(n);
+    }
 
+    /// Set the used ring index to the current `next_used` value.
+    /// Should be called once after number of `add_used` calls.
+    pub fn advance_used_ring_idx(&mut self) {
         // This fence ensures all descriptor writes are visible before the index update is.
         fence(Ordering::Release);
-
         self.used_ring_idx_set(self.next_used.0);
     }
 
     /// Puts an available descriptor head into the used ring for use by the guest.
     pub fn add_used(&mut self, desc_index: u16, len: u32) -> Result<(), QueueError> {
         self.write_used_element(0, desc_index, len)?;
-        self.advance_used_ring(1);
+        self.advance_next_used(1);
         Ok(())
     }
 

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

@@ -185,6 +185,7 @@ impl Entropy {
                 }
             }
         }
+        self.queues[RNG_QUEUE].advance_used_ring_idx();
 
         if used_any {
             self.signal_used_queue().unwrap_or_else(|err| {
@@ -294,8 +295,7 @@ impl VirtioDevice for Entropy {
 
     fn activate(&mut self, mem: GuestMemoryMmap) -> Result<(), ActivateError> {
         for q in self.queues.iter_mut() {
-            q.initialize(&mem)
-                .map_err(ActivateError::QueueError)?;
+            q.initialize(&mem).map_err(ActivateError::QueueError)?;
         }
 
         self.activate_event.write(1).map_err(|_| {

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

@@ -149,9 +149,10 @@ where
         // This is safe since we checked in the event handler that the device is activated.
         let mem = self.device_state.mem().unwrap();
 
+        let queue = &mut self.queues[RXQ_INDEX];
         let mut have_used = false;
 
-        while let Some(head) = self.queues[RXQ_INDEX].pop()? {
+        while let Some(head) = queue.pop()? {
             let index = head.index;
             let used_len = match self.rx_packet.parse(mem, head) {
                 Ok(()) => {
@@ -174,7 +175,7 @@ where
                         // We are using a consuming iterator over the virtio buffers, so, if we
                         // can't fill in this buffer, we'll need to undo the
                         // last iterator step.
-                        self.queues[RXQ_INDEX].undo_pop();
+                        queue.undo_pop();
                         break;
                     }
                 }
@@ -185,12 +186,11 @@ where
             };
 
             have_used = true;
-            self.queues[RXQ_INDEX]
-                .add_used(index, used_len)
-                .unwrap_or_else(|err| {
-                    error!("Failed to add available descriptor {}: {}", index, err)
-                });
+            queue.add_used(index, used_len).unwrap_or_else(|err| {
+                error!("Failed to add available descriptor {}: {}", index, err)
+            });
         }
+        queue.advance_used_ring_idx();
 
         Ok(have_used)
     }
@@ -202,37 +202,35 @@ where
         // This is safe since we checked in the event handler that the device is activated.
         let mem = self.device_state.mem().unwrap();
 
+        let queue = &mut self.queues[TXQ_INDEX];
         let mut have_used = false;
 
-        while let Some(head) = self.queues[TXQ_INDEX].pop()? {
+        while let Some(head) = queue.pop()? {
             let index = head.index;
             // let pkt = match VsockPacket::from_tx_virtq_head(mem, head) {
             match self.tx_packet.parse(mem, head) {
                 Ok(()) => (),
                 Err(err) => {
                     error!("vsock: error reading TX packet: {:?}", err);
                     have_used = true;
-                    self.queues[TXQ_INDEX]
-                        .add_used(index, 0)
-                        .unwrap_or_else(|err| {
-                            error!("Failed to add available descriptor {}: {}", index, err);
-                        });
+                    queue.add_used(index, 0).unwrap_or_else(|err| {
+                        error!("Failed to add available descriptor {}: {}", index, err);
+                    });
                     continue;
                 }
             };
 
             if self.backend.send_pkt(&self.tx_packet).is_err() {
-                self.queues[TXQ_INDEX].undo_pop();
+                queue.undo_pop();
                 break;
             }
 
             have_used = true;
-            self.queues[TXQ_INDEX]
-                .add_used(index, 0)
-                .unwrap_or_else(|err| {
-                    error!("Failed to add available descriptor {}: {}", index, err);
-                });
+            queue.add_used(index, 0).unwrap_or_else(|err| {
+                error!("Failed to add available descriptor {}: {}", index, err);
+            });
         }
+        queue.advance_used_ring_idx();
 
         Ok(have_used)
     }
@@ -331,8 +329,7 @@ where
 
     fn activate(&mut self, mem: GuestMemoryMmap) -> Result<(), ActivateError> {
         for q in self.queues.iter_mut() {
-            q.initialize(&mem)
-                .map_err(ActivateError::QueueError)?;
+            q.initialize(&mem).map_err(ActivateError::QueueError)?;
         }
 
         if self.queues.len() != defs::VSOCK_NUM_QUEUES {