diff --git a/Cargo.lock b/Cargo.lock index 0ed00573d8..cc75943e6c 100644 --- a/Cargo.lock +++ b/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", diff --git a/README.md b/README.md index 6023ec3f00..0262ed8d1b 100644 --- a/README.md +++ b/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 diff --git a/virtio-blkd/Cargo.toml b/virtio-blkd/Cargo.toml index 0d9e12e5d0..461a0d679a 100644 --- a/virtio-blkd/Cargo.toml +++ b/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" } diff --git a/virtio-blkd/src/main.rs b/virtio-blkd/src/main.rs index 64497da127..963156a37b 100644 --- a/virtio-blkd/src/main.rs +++ b/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::::new().unwrap(); - let mut progress_head = 0; - - event_queue - .add( - device.irq_handle.as_raw_fd(), - move |_| -> Result, 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(); diff --git a/virtio-blkd/src/scheme.rs b/virtio-blkd/src/scheme.rs index c71e29d463..8a3f6154cb 100644 --- a/virtio-blkd/src/scheme.rs +++ b/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)) diff --git a/virtio-core/Cargo.toml b/virtio-core/Cargo.toml index 87467d6bfb..9a8c197b5d 100644 --- a/virtio-core/Cargo.toml +++ b/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" } diff --git a/virtio-core/src/probe.rs b/virtio-core/src/probe.rs index 34fbc00bed..a68ed54541 100644 --- a/virtio-core/src/probe.rs +++ b/virtio-core/src/probe.rs @@ -23,7 +23,6 @@ pub struct Device<'a> { struct MsixInfo { pub virt_table_base: NonNull, - pub virt_pba_base: NonNull, pub capability: MsixCapability, } @@ -84,11 +83,9 @@ fn enable_msix(pcid_handle: &mut PcidServerHandle) -> Result { } 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, }; diff --git a/virtio-core/src/spec.rs b/virtio-core/src/spec.rs index 072314de2b..eb8f12f829 100644 --- a/virtio-core/src/spec.rs +++ b/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::(), 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) { + 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, + 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::(), 2); @@ -148,7 +182,7 @@ const_assert_eq!(core::mem::size_of::(), 2); #[repr(C)] pub struct AvailableRing { pub flags: VolatileCell, - pub head_index: VolatileCell, + pub head_index: AtomicU16, pub elements: IncompleteArrayField, } @@ -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(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 diff --git a/virtio-core/src/transport.rs b/virtio-core/src/transport.rs index cbcb7f860d..ae94c70316 100644 --- a/virtio-core/src/transport.rs +++ b/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, - - notification_bell: &'a mut VolatileCell, - head_index: u16, +pub struct PendingRequest<'a> { + queue: Arc>, + 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 { + // 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>, pub queue_index: u16, + pub waker: Mutex>, + pub used: Used<'a>, + pub descriptor: Dma<[Descriptor]>, + pub available: Available<'a>, + + notification_bell: &'a mut AtomicU16, + descriptor_stack: crossbeam_queue::SegQueue, sref: Weak, } @@ -91,101 +134,79 @@ impl<'a> Queue<'a> { available: Available<'a>, used: Used<'a>, - notification_bell: &'a mut VolatileCell, + notification_bell: &'a mut AtomicU16, queue_index: u16, ) -> Arc { + 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) { + #[must_use = "The function returns a future that must be awaited to ensure the sent request is completed."] + pub fn send(&self, chain: Vec) -> PendingRequest<'a> { let mut first_descriptor: Option = None; let mut last_descriptor: Option = 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, } impl<'a> StandardTransport<'a> { pub fn new(common: &'a mut CommonCfg, notify: *const u8, notify_mul: u32) -> Arc { - 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 { - // 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>, Error> { + pub fn setup_queue(&self, vector: u16, irq_handle: &File) -> Result>, 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) + &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::::new().unwrap(); + + event_queue + .add(irq_fd, move |_| -> Result, 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) } } diff --git a/virtio-netd/Cargo.toml b/virtio-netd/Cargo.toml index 8a905d18e8..f4a116f10c 100644 --- a/virtio-netd/Cargo.toml +++ b/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" } diff --git a/virtio-netd/src/main.rs b/virtio-netd/src/main.rs index 172034c2e2..e6fb3ee842 100644 --- a/virtio-netd/src/main.rs +++ b/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(); diff --git a/virtio-netd/src/scheme.rs b/virtio-netd/src/scheme.rs index c3eb8fe4a7..251a501ce0 100644 --- a/virtio-netd/src/scheme.rs +++ b/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::(); - 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::::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> { - 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())) }