virtio-core: make send async

This allows for us to do cool stuff such as
`join!(queue.send(read_command), queue.send(write_command),
queue.send(read_command_2))`

Signed-off-by: Anhad Singh <andypython@protonmail.com>

Cargo.lock

@@ -354,6 +354,16 @@ dependencies = [
  "crossbeam-utils 0.8.15",
 ]
 
+[[package]]
+name = "crossbeam-queue"
+version = "0.3.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d1cfb3ea8a53f37c40dea2c7bedcbd88bdfae54f5e2175d6ecaff1c988353add"
+dependencies = [
+ "cfg-if 1.0.0",
+ "crossbeam-utils 0.8.15",
+]
+
 [[package]]
 name = "crossbeam-utils"
 version = "0.7.2"
@@ -1910,12 +1920,12 @@ version = "0.1.0"
 dependencies = [
  "anyhow",
  "block-io-wrapper",
+ "futures",
  "log",
  "partitionlib",
  "pcid",
  "redox-daemon",
  "redox-log",
- "redox_event",
  "redox_syscall 0.3.5",
  "static_assertions",
  "thiserror",
@@ -1927,9 +1937,12 @@ name = "virtio-core"
 version = "0.1.0"
 dependencies = [
  "bitflags 2.3.2",
+ "crossbeam-queue",
+ "futures",
  "log",
  "pcid",
  "redox-log",
+ "redox_event",
  "redox_syscall 0.3.5",
  "static_assertions",
  "thiserror",
@@ -1939,6 +1952,7 @@ dependencies = [
 name = "virtio-netd"
 version = "0.1.0"
 dependencies = [
+ "futures",
  "log",
  "netutils",
  "pcid",

README.md

@@ -24,7 +24,7 @@ These are the currently implemented devices/hardware interfaces.
 - usbctl - USB control (incomplete).
 - usbhidd - USB HID (incomplete).
 - usbscsid - USB SCSI (incomplete).
-- virtiod - VirtIO (incomplete) (`virtio-blk`).
+- virtio-* - VirtIO (incomplete) (`virtio-blk`, `virtio-net`).
 
 ## Contributing to Drivers
 

virtio-blkd/Cargo.toml

@@ -9,11 +9,11 @@ anyhow = "1.0.71"
 log = "0.4"
 thiserror = "1.0.40"
 static_assertions = "1.1.0"
+futures = { version = "0.3.28", features = ["executor"] }
 
 redox-daemon = "0.1"
 redox-log = "0.1"
 redox_syscall = "0.3"
-redox_event = { git = "https://gitlab.redox-os.org/redox-os/event.git" }
 partitionlib = { git = "https://gitlab.redox-os.org/redox-os/partitionlib.git" }
 
 block-io-wrapper = { path = "../block-io-wrapper" }

virtio-blkd/src/main.rs

@@ -1,17 +1,15 @@
 #![deny(trivial_numeric_casts, unused_allocation)]
-#![feature(int_roundings)]
+#![feature(int_roundings, async_fn_in_trait)]
 
 use std::fs::File;
-use std::io;
 use std::io::{Read, Write};
-use std::os::fd::{AsRawFd, FromRawFd, RawFd};
+use std::os::fd::{FromRawFd, RawFd};
 
 use static_assertions::const_assert_eq;
 
 use pcid_interface::*;
 use virtio_core::spec::*;
 
-use event::EventQueue;
 use syscall::{Packet, SchemeBlockMut};
 
 use virtio_core::utils::VolatileCell;
@@ -93,67 +91,15 @@ fn deamon(deamon: redox_daemon::Daemon) -> anyhow::Result<()> {
     assert_eq!(pci_config.func.devid, 0x1001);
     log::info!("virtio-blk: initiating startup sequence :^)");
 
-    let mut device = virtio_core::probe_device(&mut pcid_handle)?;
+    let device = virtio_core::probe_device(&mut pcid_handle)?;
     device.transport.finalize_features();
 
     let queue = device
         .transport
-        .setup_queue(virtio_core::MSIX_PRIMARY_VECTOR)?;
-    let queue_copy = queue.clone();
+        .setup_queue(virtio_core::MSIX_PRIMARY_VECTOR, &device.irq_handle)?;
 
     let device_space = unsafe { &mut *(device.device_space as *mut BlockDeviceConfig) };
 
-    std::thread::spawn(move || {
-        let mut event_queue = EventQueue::<usize>::new().unwrap();
-        let mut progress_head = 0;
-
-        event_queue
-            .add(
-                device.irq_handle.as_raw_fd(),
-                move |_| -> Result<Option<usize>, io::Error> {
-                    // Read from ISR to acknowledge the interrupt.
-                    let _isr = device.isr.get() as usize;
-
-                    let mut inner = queue_copy.inner.lock().unwrap();
-                    let used_head = inner.used.head_index();
-
-                    if progress_head == used_head {
-                        return Ok(None);
-                    }
-
-                    for i in progress_head..used_head {
-                        let used = inner.used.get_element_at(i as usize);
-                        let mut desc_idx = used.table_index.get();
-                        inner.descriptor_stack.push_back(desc_idx as u16);
-
-                        loop {
-                            let desc = &inner.descriptor[desc_idx as usize];
-                            if !desc.flags.contains(DescriptorFlags::NEXT) {
-                                break;
-                            }
-
-                            desc_idx = desc.next.into();
-                            inner.descriptor_stack.push_back(desc_idx as u16);
-                        }
-                    }
-
-                    progress_head = used_head;
-                    drop(inner);
-
-                    let mut buf = [0u8; 8];
-                    device.irq_handle.read(&mut buf)?;
-                    // Acknowledge the interrupt.
-                    // irq_handle.write(&buf)?;
-                    Ok(None)
-                },
-            )
-            .unwrap();
-
-        loop {
-            event_queue.run().unwrap();
-        }
-    });
-
     // At this point the device is alive!
     device.transport.run_device();
 

virtio-blkd/src/scheme.rs

@@ -21,33 +21,12 @@ use crate::BlockVirtRequest;
 const BLK_SIZE: u64 = 512;
 
 trait BlkExtension {
-    /// XXX: Reads only one block despite the size of the output buffer. Use [`BlkExtension::read`] instead.
-    fn read_block(&self, block: u64, block_bytes: &mut [u8]) -> usize;
-
-    /// XXX: Reads only one block despite the size of the output buffer. Use [`BlkExtension::write`] instead.
-    fn write_block(&self, block: u64, block_bytes: &[u8]) -> usize;
-
-    // FIXME(andypython): The following two methods can be done asynchronously (i.e, send all of the commands at once
-    //                    and wait for the result in the end).
-    fn read(&self, block: u64, block_bytes: &mut [u8]) -> usize {
-        let sectors = block_bytes.len() / BLK_SIZE as usize;
-
-        (0..sectors)
-            .map(|i| self.read_block(block + i as u64, &mut block_bytes[i * BLK_SIZE as usize..]))
-            .sum()
-    }
-
-    fn write(&self, block: u64, block_bytes: &[u8]) -> usize {
-        let sectors = block_bytes.len() / BLK_SIZE as usize;
-
-        (0..sectors)
-            .map(|i| self.write_block(block + i as u64, &block_bytes[i * BLK_SIZE as usize..]))
-            .sum()
-    }
+    async fn read(&self, block: u64, target: &mut [u8]) -> usize;
+    async fn write(&self, block: u64, target: &[u8]) -> usize;
 }
 
 impl BlkExtension for Queue<'_> {
-    fn read_block(&self, block: u64, block_bytes: &mut [u8]) -> usize {
+    async fn read(&self, block: u64, target: &mut [u8]) -> usize {
         let req = syscall::Dma::new(BlockVirtRequest {
             ty: BlockRequestTy::In,
             reserved: 0,
@@ -55,28 +34,24 @@ impl BlkExtension for Queue<'_> {
         })
         .unwrap();
 
-        let result = syscall::Dma::new([0u8; 512]).unwrap();
+        let result = unsafe { syscall::Dma::<[u8]>::zeroed_unsized(target.len()) }.unwrap();
         let status = syscall::Dma::new(u8::MAX).unwrap();
 
         let chain = ChainBuilder::new()
             .chain(Buffer::new(&req))
-            .chain(Buffer::new(&result).flags(DescriptorFlags::WRITE_ONLY))
+            .chain(Buffer::new_unsized(&result).flags(DescriptorFlags::WRITE_ONLY))
             .chain(Buffer::new(&status).flags(DescriptorFlags::WRITE_ONLY))
             .build();
 
-        self.send(chain);
+        // XXX: Subtract 1 because the of status byte.
+        let written = self.send(chain).await as usize - 1;
+        assert_eq!(*status, 0);
 
-        // FIXME: interrupts are for a reason
-        while *status != 0 {
-            core::hint::spin_loop();
-        }
-
-        let size = core::cmp::min(block_bytes.len(), result.len());
-        block_bytes[..size].copy_from_slice(&result.as_slice()[..size]);
-        size
+        target[..written].copy_from_slice(&result);
+        written
     }
 
-    fn write_block(&self, block: u64, block_bytes: &[u8]) -> usize {
+    async fn write(&self, block: u64, target: &[u8]) -> usize {
         let req = syscall::Dma::new(BlockVirtRequest {
             ty: BlockRequestTy::Out,
             reserved: 0,
@@ -84,25 +59,21 @@ impl BlkExtension for Queue<'_> {
         })
         .unwrap();
 
-        let mut result = syscall::Dma::new([0u8; 512]).unwrap();
-        result.copy_from_slice(&block_bytes[..512]);
+        let mut result = unsafe { syscall::Dma::<[u8]>::zeroed_unsized(target.len()) }.unwrap();
+        result.copy_from_slice(target.as_ref());
 
         let status = syscall::Dma::new(u8::MAX).unwrap();
 
         let chain = ChainBuilder::new()
             .chain(Buffer::new(&req))
-            .chain(Buffer::new(&result))
+            .chain(Buffer::new_sized(&result, target.len()))
             .chain(Buffer::new(&status).flags(DescriptorFlags::WRITE_ONLY))
             .build();
 
-        self.send(chain);
+        self.send(chain).await as usize;
+        assert_eq!(*status, 0);
 
-        // FIXME: interrupts are for a reason
-        while *status != 0 {
-            core::hint::spin_loop();
-        }
-
-        block_bytes.len()
+        target.len()
     }
 }
 
@@ -167,18 +138,11 @@ impl<'a> DiskScheme<'a> {
                         .chain(Buffer::new(&status).flags(DescriptorFlags::WRITE_ONLY))
                         .build();
 
-                    self.scheme.queue.send(chain);
-
-                    // FIXME: interrupts are for a reason
-                    while *status != 0 {
-                        core::hint::spin_loop();
-                    }
+                    futures::executor::block_on(self.scheme.queue.send(chain));
+                    assert_eq!(*status, 0);
 
                     let size = core::cmp::min(block_bytes.len(), result.len());
                     block_bytes[..size].copy_from_slice(&result.as_slice()[..size]);
-
-                    std::thread::yield_now();
-
                     Ok(())
                 };
 
@@ -328,7 +292,7 @@ impl<'a> SchemeBlockMut for DiskScheme<'a> {
 
                 let abs_block = part.start_lba + rel_block;
 
-                let count = self.queue.read(abs_block, buf);
+                let count = futures::executor::block_on(self.queue.read(abs_block, buf));
                 *offset += count;
                 Ok(Some(count))
             }
@@ -336,7 +300,9 @@ impl<'a> SchemeBlockMut for DiskScheme<'a> {
             Handle::Disk { ref mut offset } => {
                 let block_size = self.cfg.block_size();
 
-                let count = self.queue.read((*offset as u64) / block_size as u64, buf);
+                let count = futures::executor::block_on(
+                    self.queue.read((*offset as u64) / block_size as u64, buf),
+                );
                 *offset += count;
                 Ok(Some(count))
             }
@@ -347,7 +313,9 @@ impl<'a> SchemeBlockMut for DiskScheme<'a> {
         match *self.handles.get_mut(&id).ok_or(Error::new(EBADF))? {
             Handle::Disk { ref mut offset } => {
                 let block_size = self.cfg.block_size();
-                let count = self.queue.write((*offset as u64) / block_size as u64, buf);
+                let count = futures::executor::block_on(
+                    self.queue.write((*offset as u64) / block_size as u64, buf),
+                );
 
                 *offset += count;
                 Ok(Some(count))

virtio-core/Cargo.toml

@@ -10,7 +10,10 @@ bitflags = "2.3.2"
 redox_syscall = "0.3"
 log = "0.4"
 thiserror = "1.0.40"
+futures = { version = "0.3.28", features = ["executor"] }
+crossbeam-queue = "0.3.8"
 
 redox-log = "0.1"
+redox_event = { git = "https://gitlab.redox-os.org/redox-os/event.git" }
 
 pcid = { path = "../pcid" }

virtio-core/src/probe.rs

@@ -23,7 +23,6 @@ pub struct Device<'a> {
 
 struct MsixInfo {
     pub virt_table_base: NonNull<MsixTableEntry>,
-    pub virt_pba_base: NonNull<u64>,
     pub capability: MsixCapability,
 }
 
@@ -84,11 +83,9 @@ fn enable_msix(pcid_handle: &mut PcidServerHandle) -> Result<File, Error> {
     }
 
     let virt_table_base = ((table_base - bar_ptr as usize) + address) as *mut MsixTableEntry;
-    let virt_pba_base = ((pba_base - bar_ptr as usize) + address) as *mut u64;
 
     let mut info = MsixInfo {
         virt_table_base: NonNull::new(virt_table_base).unwrap(),
-        virt_pba_base: NonNull::new(virt_pba_base).unwrap(),
         capability,
     };
 

virtio-core/src/spec.rs

@@ -1,5 +1,7 @@
 //! https://docs.oasis-open.org/virtio/virtio/v1.1/virtio-v1.1.html
 
+use std::sync::atomic::{AtomicU16, AtomicU32, AtomicU64, Ordering};
+
 use crate::utils::{IncompleteArrayField, VolatileCell};
 use static_assertions::const_assert_eq;
 
@@ -116,16 +118,42 @@ bitflags::bitflags! {
 #[repr(C)]
 pub struct Descriptor {
     /// Address (guest-physical).
-    pub address: u64,
+    pub address: AtomicU64,
     /// Size of the descriptor.
-    pub size: u32,
-    pub flags: DescriptorFlags,
+    pub size: AtomicU32,
+    flags: AtomicU16,
     /// Index of next desciptor in chain.
-    pub next: u16,
+    pub next: AtomicU16,
 }
 
 const_assert_eq!(core::mem::size_of::<Descriptor>(), 16);
 
+impl Descriptor {
+    pub fn set_addr(&self, addr: u64) {
+        self.address.store(addr, Ordering::SeqCst)
+    }
+
+    pub fn set_size(&self, size: u32) {
+        self.size.store(size, Ordering::SeqCst)
+    }
+
+    pub fn set_next(&self, next: Option<u16>) {
+        self.next.store(next.unwrap_or_default(), Ordering::SeqCst)
+    }
+
+    pub fn set_flags(&self, flags: DescriptorFlags) {
+        self.flags.store(flags.bits(), Ordering::SeqCst)
+    }
+
+    pub fn next(&self) -> u16 {
+        self.next.load(Ordering::SeqCst)
+    }
+
+    pub fn flags(&self) -> DescriptorFlags {
+        DescriptorFlags::from_bits_truncate(self.flags.load(Ordering::SeqCst))
+    }
+}
+
 /// This indicates compliance with the version 1 VirtIO specification.
 ///
 /// See `6.1 Driver Requirements: Reserved Feature Bits` section of the VirtIO
@@ -140,7 +168,13 @@ pub const VIRTIO_NET_F_MAC: u32 = 5;
 //      chain.
 #[repr(C)]
 pub struct AvailableRingElement {
-    pub table_index: VolatileCell<u16>,
+    pub table_index: AtomicU16,
+}
+
+impl AvailableRingElement {
+    pub fn set_table_index(&self, index: u16) {
+        self.table_index.store(index, Ordering::SeqCst)
+    }
 }
 
 const_assert_eq!(core::mem::size_of::<AvailableRingElement>(), 2);
@@ -148,7 +182,7 @@ const_assert_eq!(core::mem::size_of::<AvailableRingElement>(), 2);
 #[repr(C)]
 pub struct AvailableRing {
     pub flags: VolatileCell<u16>,
-    pub head_index: VolatileCell<u16>,
+    pub head_index: AtomicU16,
     pub elements: IncompleteArrayField<AvailableRingElement>,
 }
 
@@ -158,7 +192,7 @@ impl Default for AvailableRing {
     fn default() -> Self {
         Self {
             flags: VolatileCell::new(0),
-            head_index: VolatileCell::new(0),
+            head_index: AtomicU16::new(0),
             elements: IncompleteArrayField::new(),
         }
     }
@@ -228,6 +262,14 @@ impl Buffer {
         }
     }
 
+    pub fn new_sized<T>(val: &syscall::Dma<[T]>, size: usize) -> Self {
+        Self {
+            buffer: val.physical(),
+            size,
+            flags: DescriptorFlags::empty(),
+        }
+    }
+
     pub fn flags(mut self, flags: DescriptorFlags) -> Self {
         self.flags = flags;
         self

virtio-core/src/transport.rs

@@ -1,13 +1,17 @@
 use crate::spec::*;
-use crate::utils::{align, VolatileCell};
+use crate::utils::align;
 
+use event::EventQueue;
 use syscall::{Dma, PHYSMAP_WRITE};
 
 use core::mem::size_of;
-use core::sync::atomic::{fence, AtomicU16, Ordering};
+use core::sync::atomic::{AtomicU16, Ordering};
 
-use std::collections::VecDeque;
+use std::fs::File;
+use std::future::Future;
+use std::os::fd::AsRawFd;
 use std::sync::{Arc, Mutex, Weak};
+use std::task::{Poll, Waker};
 
 #[derive(thiserror::Error, Debug)]
 pub enum Error {
@@ -64,24 +68,63 @@ const fn queue_part_sizes(queue_size: usize) -> (usize, usize, usize) {
     )
 }
 
-pub struct QueueInner<'a> {
-    pub descriptor: Dma<[Descriptor]>,
-    pub available: Available<'a>,
-    pub used: Used<'a>,
-
-    /// Keeps track of unused descriptor indicies.
-    pub descriptor_stack: VecDeque<u16>,
-
-    notification_bell: &'a mut VolatileCell<u16>,
-    head_index: u16,
+pub struct PendingRequest<'a> {
+    queue: Arc<Queue<'a>>,
+    first_descriptor: u32,
 }
 
-unsafe impl Sync for QueueInner<'_> {}
-unsafe impl Send for QueueInner<'_> {}
+impl<'a> Future for PendingRequest<'a> {
+    type Output = u32;
+
+    fn poll(self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll<Self::Output> {
+        // XXX: Register the waker before checking the queue to avoid the race condition
+        //      where you lose a notification.
+        self.queue
+            .waker
+            .lock()
+            .unwrap()
+            .insert(self.first_descriptor, cx.waker().clone());
+
+        let used_head = self.queue.used.head_index();
+        let used_element = self.queue.used.get_element_at((used_head - 1) as usize);
+        let written = used_element.written.get();
+
+        let mut table_index = used_element.table_index.get();
+
+        if table_index == self.first_descriptor {
+            // The request has been completed; recycle the descriptors used.
+            while self.queue.descriptor[table_index as usize]
+                .flags()
+                .contains(DescriptorFlags::NEXT)
+            {
+                let next_index = self.queue.descriptor[table_index as usize].next();
+                self.queue.descriptor_stack.push(table_index as u16);
+                table_index = next_index.into();
+            }
+
+            // Push the last descriptor.
+            self.queue.descriptor_stack.push(table_index as u16);
+            self.queue
+                .waker
+                .lock()
+                .unwrap()
+                .remove(&self.first_descriptor);
+            return Poll::Ready(written);
+        } else {
+            return Poll::Pending;
+        }
+    }
+}
 
 pub struct Queue<'a> {
-    pub inner: Mutex<QueueInner<'a>>,
     pub queue_index: u16,
+    pub waker: Mutex<std::collections::HashMap<u32, Waker>>,
+    pub used: Used<'a>,
+    pub descriptor: Dma<[Descriptor]>,
+    pub available: Available<'a>,
+
+    notification_bell: &'a mut AtomicU16,
+    descriptor_stack: crossbeam_queue::SegQueue<u16>,
     sref: Weak<Self>,
 }
 
@@ -91,101 +134,79 @@ impl<'a> Queue<'a> {
         available: Available<'a>,
         used: Used<'a>,
 
-        notification_bell: &'a mut VolatileCell<u16>,
+        notification_bell: &'a mut AtomicU16,
         queue_index: u16,
     ) -> Arc<Self> {
+        let descriptor_stack = crossbeam_queue::SegQueue::new();
+        (0..descriptor.len() as u16).for_each(|i| descriptor_stack.push(i));
+
         Arc::new_cyclic(|sref| Self {
-            inner: Mutex::new(QueueInner {
-                head_index: 0,
-                descriptor_stack: (0..descriptor.len() as u16).collect(),
-
-                descriptor,
-                available,
-                used,
-
-                notification_bell,
-            }),
-
+            notification_bell,
+            available,
+            descriptor,
+            used,
+            waker: Mutex::new(std::collections::HashMap::new()),
             queue_index,
+            descriptor_stack,
             sref: sref.clone(),
         })
     }
 
-    pub fn send(&self, chain: Vec<Buffer>) {
+    #[must_use = "The function returns a future that must be awaited to ensure the sent request is completed."]
+    pub fn send(&self, chain: Vec<Buffer>) -> PendingRequest<'a> {
         let mut first_descriptor: Option<usize> = None;
         let mut last_descriptor: Option<usize> = None;
 
         for buffer in chain.iter() {
-            let descriptor = self.alloc_descriptor();
-
-            let mut inner = self.inner.lock().unwrap();
+            let descriptor = self.descriptor_stack.pop().unwrap() as usize;
 
             if first_descriptor.is_none() {
                 first_descriptor = Some(descriptor);
             }
 
-            inner.descriptor[descriptor].address = buffer.buffer as u64;
-            inner.descriptor[descriptor].flags = buffer.flags;
-            inner.descriptor[descriptor].size = buffer.size as u32;
+            self.descriptor[descriptor].set_addr(buffer.buffer as u64);
+            self.descriptor[descriptor].set_flags(buffer.flags);
+            self.descriptor[descriptor].set_size(buffer.size as u32);
 
             if let Some(index) = last_descriptor {
-                inner.descriptor[index].next = descriptor as u16;
+                self.descriptor[index].set_next(Some(descriptor as u16));
             }
 
             last_descriptor = Some(descriptor);
         }
 
-        let mut inner = self.inner.lock().unwrap();
-
         let last_descriptor = last_descriptor.unwrap();
         let first_descriptor = first_descriptor.unwrap();
 
-        inner.descriptor[last_descriptor].next = 0;
+        self.descriptor[last_descriptor].set_next(None);
 
-        fence(Ordering::SeqCst);
-        let index = inner.head_index as usize;
+        let index = self.available.head_index() as usize;
 
-        inner
-            .available
+        self.available
             .get_element_at(index)
-            .table_index
-            .set(first_descriptor as u16);
+            .set_table_index(first_descriptor as u16);
 
-        fence(Ordering::SeqCst);
-        inner.available.set_head_idx(index as u16 + 1);
-        inner.head_index += 1;
+        self.available.set_head_idx(index as u16 + 1);
+        self.notification_bell
+            .store(self.queue_index, Ordering::SeqCst);
 
-        assert_eq!(inner.used.flags(), 0);
-        inner.notification_bell.set(self.queue_index);
-    }
+        assert_eq!(self.used.flags(), 0);
 
-    fn alloc_descriptor(&self) -> usize {
-        let mut inner = self.inner.lock().unwrap();
-
-        if let Some(index) = inner.descriptor_stack.pop_front() {
-            index as usize
-        } else {
-            log::warn!("virtiod-core: descriptors exhausted, waiting on garabage collector");
-            drop(inner);
-
-            // Wait for the garbage collector thread to release some descriptors.
-            //
-            // TODO(andypython): Instead of just yielding, we should have a proper notificiation
-            //                   mechanism. I am not aware whats the standard way redox applications
-            //                   or drivers implement basically a WaitQueue which you can use to wake
-            //                   up a thread. The descripts really should NEVER run out, but if they
-            //                   do, have a proper way to handle them.
-            std::thread::yield_now();
-            self.alloc_descriptor()
+        PendingRequest {
+            queue: self.sref.upgrade().unwrap(),
+            first_descriptor: first_descriptor as u32,
         }
     }
 
     /// Returns the number of descriptors in the descriptor table of this queue.
     pub fn descriptor_len(&self) -> usize {
-        self.inner.lock().unwrap().descriptor.len()
+        self.descriptor.len()
     }
 }
 
+unsafe impl Sync for Queue<'_> {}
+unsafe impl Send for Queue<'_> {}
+
 pub struct Available<'a> {
     addr: usize,
     size: usize,
@@ -216,20 +237,24 @@ impl<'a> Available<'a> {
 
     /// ## Panics
     /// This function panics if the index is out of bounds.
-    pub fn get_element_at(&mut self, index: usize) -> &mut AvailableRingElement {
+    pub fn get_element_at(&self, index: usize) -> &AvailableRingElement {
         // SAFETY: We have exclusive access to the elements and the number of elements
         //         is correct; same as the queue size.
         unsafe {
             self.ring
                 .elements
-                .as_mut_slice(self.queue_size)
-                .get_mut(index % 256)
+                .as_slice(self.queue_size)
+                .get(index % 256)
                 .expect("virtio-core::available: index out of bounds")
         }
     }
 
-    pub fn set_head_idx(&mut self, index: u16) {
-        self.ring.head_index.set(index);
+    pub fn head_index(&self) -> u16 {
+        self.ring.head_index.load(Ordering::SeqCst)
+    }
+
+    pub fn set_head_idx(&self, index: u16) {
+        self.ring.head_index.store(index, Ordering::SeqCst);
     }
 
     pub fn phys_addr(&self) -> usize {
@@ -278,7 +303,21 @@ impl<'a> Used<'a> {
 
     /// ## Panics
     /// This function panics if the index is out of bounds.
-    pub fn get_element_at(&mut self, index: usize) -> &mut UsedRingElement {
+    pub fn get_element_at(&self, index: usize) -> &UsedRingElement {
+        // SAFETY: We have exclusive access to the elements and the number of elements
+        //         is correct; same as the queue size.
+        unsafe {
+            self.ring
+                .elements
+                .as_slice(self.queue_size)
+                .get(index % 256)
+                .expect("virtio-core::used: index out of bounds")
+        }
+    }
+
+    /// ## Panics
+    /// This function panics if the index is out of bounds.
+    pub fn get_mut_element_at(&mut self, index: usize) -> &mut UsedRingElement {
         // SAFETY: We have exclusive access to the elements and the number of elements
         //         is correct; same as the queue size.
         unsafe {
@@ -320,27 +359,19 @@ pub struct StandardTransport<'a> {
     notify_mul: u32,
 
     queue_index: AtomicU16,
-    sref: Weak<Self>,
 }
 
 impl<'a> StandardTransport<'a> {
     pub fn new(common: &'a mut CommonCfg, notify: *const u8, notify_mul: u32) -> Arc<Self> {
-        Arc::new_cyclic(|sref| Self {
+        Arc::new(Self {
             common: Mutex::new(common),
             notify,
             notify_mul,
 
             queue_index: AtomicU16::new(0),
-            sref: sref.clone(),
         })
     }
 
-    pub fn sref(&self) -> Arc<Self> {
-        // UNWRAP: The constructor ensures that we are wrapped in our own `Arc`. So this
-        //         unwrap is going to be unreachable.
-        self.sref.upgrade().unwrap()
-    }
-
     pub fn check_device_feature(&self, feature: u32) -> bool {
         let mut common = self.common.lock().unwrap();
 
@@ -384,7 +415,7 @@ impl<'a> StandardTransport<'a> {
             .set(status | DeviceStatusFlags::DRIVER_OK);
     }
 
-    pub fn setup_queue(&self, vector: u16) -> Result<Arc<Queue<'a>>, Error> {
+    pub fn setup_queue(&self, vector: u16, irq_handle: &File) -> Result<Arc<Queue<'a>>, Error> {
         let mut common = self.common.lock().unwrap();
 
         let queue_index = self.queue_index.fetch_add(1, Ordering::SeqCst);
@@ -398,14 +429,17 @@ impl<'a> StandardTransport<'a> {
             Dma::<[Descriptor]>::zeroed_unsized(queue_size).map_err(Error::SyscallError)?
         };
 
-        let mut avail = Available::new(queue_size)?;
+        let avail = Available::new(queue_size)?;
         let mut used = Used::new(queue_size)?;
 
         for i in 0..queue_size {
             // XXX: Fill the `table_index` of the elements with `T::MAX` to help with
             //      debugging since qemu reports them as illegal values.
-            avail.get_element_at(i).table_index.set(u16::MAX);
-            used.get_element_at(i).table_index.set(u32::MAX);
+            avail
+                .get_element_at(i)
+                .table_index
+                .store(u16::MAX, Ordering::Relaxed);
+            used.get_mut_element_at(i).table_index.set(u32::MAX);
         }
 
         common.queue_desc.set(descriptor.physical() as u64);
@@ -421,16 +455,33 @@ impl<'a> StandardTransport<'a> {
 
         let notification_bell = unsafe {
             let offset = self.notify_mul * queue_notify_idx as u32;
-            &mut *(self.notify.add(offset as usize) as *mut VolatileCell<u16>)
+            &mut *(self.notify.add(offset as usize) as *mut AtomicU16)
         };
 
         log::info!("virtio-core: enabled queue #{queue_index} (size={queue_size})");
-        Ok(Queue::new(
-            descriptor,
-            avail,
-            used,
-            notification_bell,
-            queue_index,
-        ))
+
+        let queue = Queue::new(descriptor, avail, used, notification_bell, queue_index);
+
+        let queue_copy = queue.clone();
+        let irq_fd = irq_handle.as_raw_fd();
+
+        std::thread::spawn(move || {
+            let mut event_queue = EventQueue::<usize>::new().unwrap();
+
+            event_queue
+                .add(irq_fd, move |_| -> Result<Option<usize>, std::io::Error> {
+                    for (_, task) in queue_copy.waker.lock().unwrap().iter() {
+                        task.wake_by_ref();
+                    }
+                    Ok(None)
+                })
+                .unwrap();
+
+            loop {
+                event_queue.run().unwrap();
+            }
+        });
+
+        Ok(queue)
     }
 }

virtio-netd/Cargo.toml

@@ -6,6 +6,7 @@ edition = "2021"
 [dependencies]
 log = "0.4"
 static_assertions = "1.1.0"
+futures = { version = "0.3.28", features = ["executor"] }
 
 virtio-core = { path = "../virtio-core" }
 pcid = { path = "../pcid" }

virtio-netd/src/main.rs

@@ -73,11 +73,11 @@ fn deamon(deamon: redox_daemon::Daemon) -> Result<(), Error> {
     // TODO(andypython): Should we use the same IRQ vector for both?
     let rx_queue = device
         .transport
-        .setup_queue(virtio_core::MSIX_PRIMARY_VECTOR)?;
+        .setup_queue(virtio_core::MSIX_PRIMARY_VECTOR, &device.irq_handle)?;
 
     let tx_queue = device
         .transport
-        .setup_queue(virtio_core::MSIX_PRIMARY_VECTOR)?;
+        .setup_queue(virtio_core::MSIX_PRIMARY_VECTOR, &device.irq_handle)?;
 
     device.transport.run_device();
 

virtio-netd/src/scheme.rs

@@ -35,7 +35,7 @@ impl<'a> NetworkScheme<'a> {
                 .chain(Buffer::new_unsized(&rx_buffers[i]).flags(DescriptorFlags::WRITE_ONLY))
                 .build();
 
-            rx.send(chain);
+            let _ = rx.send(chain);
         }
 
         Self {
@@ -54,15 +54,13 @@ impl<'a> NetworkScheme<'a> {
     fn try_recv(&mut self, target: &mut [u8]) -> usize {
         let header_size = core::mem::size_of::<VirtHeader>();
 
-        let mut queue = self.rx.inner.lock().unwrap();
-
-        if self.recv_head == queue.used.head_index() {
+        if self.recv_head == self.rx.used.head_index() {
             // The read would block.
             return 0;
         }
 
-        let idx = queue.used.head_index() as usize;
-        let element = queue.used.get_element_at(idx - 1);
+        let idx = self.rx.used.head_index() as usize;
+        let element = self.rx.used.get_element_at(idx - 1);
 
         let descriptor_idx = element.table_index.get();
         let payload_size = element.written.get() as usize - header_size;
@@ -77,7 +75,7 @@ impl<'a> NetworkScheme<'a> {
         // Copy the packet into the buffer.
         target[..payload_size].copy_from_slice(&packet);
 
-        self.recv_head = queue.used.head_index();
+        self.recv_head = self.rx.used.head_index();
         payload_size
     }
 }
@@ -123,7 +121,6 @@ impl<'a> SchemeBlockMut for NetworkScheme<'a> {
 
         let header = unsafe { Dma::<VirtHeader>::zeroed()?.assume_init() };
 
-        // TODO: Does the payload actually need to be a DMA buffer?
         let mut payload = unsafe { Dma::<[u8]>::zeroed_unsized(buffer.len())? };
         payload.copy_from_slice(buffer);
 
@@ -132,9 +129,7 @@ impl<'a> SchemeBlockMut for NetworkScheme<'a> {
             .chain(Buffer::new_unsized(&payload))
             .build();
 
-        self.tx.send(chain);
-        core::mem::forget(payload);
-
+        futures::executor::block_on(self.tx.send(chain));
         Ok(Some(buffer.len()))
     }
 
@@ -147,7 +142,10 @@ impl<'a> SchemeBlockMut for NetworkScheme<'a> {
         id: usize,
         _flags: syscall::EventFlags,
     ) -> syscall::Result<Option<syscall::EventFlags>> {
-        let _flags = self.handles.get(&id).ok_or(SysError::new(EBADF))?;
+        if self.handles.get(&id).is_none() {
+            return Err(SysError::new(EBADF));
+        }
+
         Ok(Some(syscall::EventFlags::empty()))
     }