nvme,mana: ensure completions are read and written in order (#506)

The NVMe completion entry is a 16-byte value, and it contains a phase
bit that the NVMe driver reads to determine when a completion in the
queue is valid. To ensure the driver observes a consistent completion
entry, the phase bit must be written by the device _after_ the rest of
the entry, and it must be read by the driver _before_ the rest.

The memory access machinery in OpenVMM does not provide many ordering
guarantees, especially on ARM64. Ensure accesses are performed in the
right order via two atomic 8-byte accesses, plus proper fences, in the
NVMe emulator and driver.

There's a similar bug in MANA. Fix this, too.

Author: jstarks

vm/devices/net/gdma/src/queues.rs

@@ -24,6 +24,7 @@ use parking_lot::Mutex;
 use parking_lot::MutexGuard;
 use pci_core::capabilities::msix::MsixEmulator;
 use std::marker::PhantomData;
+use std::sync::atomic::Ordering::Release;
 use std::task::Context;
 use std::task::Poll;
 use std::task::Waker;
@@ -95,9 +96,19 @@ impl<T: AsBytes> CqEq<T> {
         let offset = (self.tail & (self.cap - 1)) as usize * size_of::<T>();
         let mut range = self.region.range();
         range.skip(offset);
-        if let Err(err) = range.writer(gm).write(entry.as_bytes()) {
+        let mut writer = range.writer(gm);
+        let (entry, last) = entry.as_bytes().split_at(entry.as_bytes().len() - 1);
+        if let Err(err) = writer.write(entry) {
             tracing::warn!(err = &err as &dyn std::error::Error, "failed to write");
         }
+        // Write the final byte last after a release fence to ensure that the
+        // guest sees the entire entry before the owner count is updated.
+        std::sync::atomic::fence(Release);
+        if let Err(err) = writer.write(last) {
+            tracing::warn!(err = &err as &dyn std::error::Error, "failed to write");
+        }
+        // Ensure the write is flushed before sending the interrupt.
+        std::sync::atomic::fence(Release);
         let new_tail = self.tail.wrapping_add(1);
         self.tail = new_tail;
         std::mem::take(&mut self.armed)

vm/devices/net/mana_driver/src/queues.rs

@@ -23,6 +23,7 @@ use gdma_defs::OWNER_MASK;
 use gdma_defs::WQE_ALIGNMENT;
 use inspect::Inspect;
 use std::marker::PhantomData;
+use std::sync::atomic::Ordering::Acquire;
 use std::sync::Arc;
 use user_driver::memory::MemoryBlock;
 use zerocopy::AsBytes;
@@ -140,7 +141,11 @@ impl<T: AsBytes + FromBytes> CqEq<T> {
 
     /// Pops an event queue entry.
     pub fn pop(&mut self) -> Option<T> {
-        let b = self.read_next::<u8>(size_of::<T>() as u32 - 1);
+        // Perform an acquire load to ensure that the read of the queue entry is
+        // not reordered before the read of the owner count.
+        let b = self.mem.as_slice()
+            [(self.next.wrapping_add(size_of::<T>() as u32 - 1) & (self.size - 1)) as usize]
+            .load(Acquire);
         let owner_count = b >> 5;
         let cur_owner_count = (self.next >> self.shift) as u8;
         if owner_count == (cur_owner_count.wrapping_sub(1)) & OWNER_MASK as u8 {

vm/devices/storage/disk_nvme/nvme_driver/src/queues.rs

@@ -8,6 +8,10 @@ use crate::driver::save_restore::CompletionQueueSavedState;
 use crate::driver::save_restore::SubmissionQueueSavedState;
 use crate::registers::DeviceRegisters;
 use inspect::Inspect;
+use safeatomic::AtomicSliceOps;
+use std::sync::atomic::AtomicU64;
+use std::sync::atomic::Ordering::Acquire;
+use std::sync::atomic::Ordering::Relaxed;
 use user_driver::memory::MemoryBlock;
 use user_driver::DeviceBacking;
 
@@ -133,12 +137,21 @@ impl CompletionQueue {
     }
 
     pub fn read(&mut self) -> Option<spec::Completion> {
-        let completion = self
-            .mem
-            .read_obj::<spec::Completion>(self.head as usize * size_of::<spec::Completion>());
-        if completion.status.phase() != self.phase {
+        let completion_mem = self.mem.as_slice()
+            [self.head as usize * size_of::<spec::Completion>()..]
+            [..size_of::<spec::Completion>() * 2]
+            .as_atomic_slice::<AtomicU64>()
+            .unwrap();
+
+        // Check the phase bit, using an acquire read to ensure the rest of the
+        // completion is read with or after the phase bit.
+        let high = completion_mem[1].load(Acquire);
+        let status = spec::CompletionStatus::from((high >> 48) as u16);
+        if status.phase() != self.phase {
             return None;
         }
+        let low = completion_mem[0].load(Relaxed);
+        let completion: spec::Completion = zerocopy::transmute!([low, high]);
         self.head += 1;
         if self.head == self.len {
             self.head = 0;

vm/devices/storage/nvme/src/queue.rs

@@ -266,8 +266,20 @@ impl CompletionQueue {
             return Ok(false);
         }
         data.status.set_phase(self.phase);
-        mem.write_plain(self.gpa.wrapping_add(self.tail as u64 * 16), &data)
+
+        // Atomically write the low part of the completion entry first, then the
+        // high part, using release fences to ensure ordering.
+        //
+        // This is necessary to ensure the guest can observe the full completion
+        // once it observes the phase bit change (which is in the high part).
+        let [low, high]: [u64; 2] = zerocopy::transmute!(data);
+        let gpa = self.gpa.wrapping_add(self.tail as u64 * 16);
+        mem.write_plain(gpa, &low).map_err(QueueError::Memory)?;
+        std::sync::atomic::fence(Ordering::Release);
+        mem.write_plain(gpa + 8, &high)
             .map_err(QueueError::Memory)?;
+        std::sync::atomic::fence(Ordering::Release);
+
         if let Some(interrupt) = &self.interrupt {
             interrupt.deliver();
         }