diff --git a/Cargo.lock b/Cargo.lock index 0a006c6d13..c46545bf26 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -740,7 +740,8 @@ dependencies = [ "partitionlib 0.1.0 (git+https://gitlab.redox-os.org/redox-os/partitionlib.git)", "pcid 0.1.0", "redox-log 0.1.0 (git+https://gitlab.redox-os.org/redox-os/redox-log.git?tag=v0.1.0)", - "redox_syscall 0.1.56 (registry+https://github.com/rust-lang/crates.io-index)", + "redox_syscall 0.1.56 (git+https://gitlab.redox-os.org/redox-os/syscall.git)", + "smallvec 1.3.0 (registry+https://github.com/rust-lang/crates.io-index)", ] [[package]] diff --git a/nvmed/Cargo.toml b/nvmed/Cargo.toml index ec8152f9e9..002d88c6e4 100644 --- a/nvmed/Cargo.toml +++ b/nvmed/Cargo.toml @@ -9,8 +9,9 @@ bitflags = "0.7" crossbeam-channel = "0.4" log = "0.4" redox-log = { git = "https://gitlab.redox-os.org/redox-os/redox-log.git", tag = "v0.1.0" } -redox_syscall = "0.1" +redox_syscall = { git = "https://gitlab.redox-os.org/redox-os/syscall.git" } partitionlib = { git = "https://gitlab.redox-os.org/redox-os/partitionlib.git" } +smallvec = "1" block-io-wrapper = { path = "../block-io-wrapper" } pcid = { path = "../pcid" } diff --git a/nvmed/src/nvme/cq_reactor.rs b/nvmed/src/nvme/cq_reactor.rs index 59532aa6bf..2e16b62ce7 100644 --- a/nvmed/src/nvme/cq_reactor.rs +++ b/nvmed/src/nvme/cq_reactor.rs @@ -1,26 +1,34 @@ +//! The Completion Queue Reactor. Functions like any other async/await reactor, but are driven by +//! IRQs triggering wakeups in order to poll NVME completion queues. + use std::collections::BTreeMap; use std::fs::File; use std::future::Future; use std::io::prelude::*; use std::pin::Pin; use std::sync::{Arc, Mutex}; -use std::{io, task, thread}; +use std::{io, mem, task, thread}; use std::os::unix::io::{AsRawFd, FromRawFd, RawFd}; use syscall::Result; use syscall::data::Event; +use syscall::flag::EVENT_READ; use crossbeam_channel::{Sender, Receiver}; -use crate::nvme::{InterruptMethod, InterruptSources, Nvme, NvmeComp}; +use crate::nvme::{CqId, CmdId, InterruptMethod, InterruptSources, Nvme, NvmeComp, NvmeCompQueue, SqId}; /// A notification request, sent by the future in order to tell the completion thread that the /// current task wants a notification when a matching completion queue entry has been seen. pub enum NotifReq { RequestCompletion { - queue_id: usize, + cq_id: CqId, + sq_id: SqId, + cmd_id: CmdId, + waker: task::Waker, - // TODO: Get rid of this allocation + + // TODO: Get rid of this allocation, or maybe a thread-local vec for reusing. message: Arc>>, }, } @@ -28,7 +36,9 @@ pub enum NotifReq { struct PendingReq { waker: task::Waker, message: Arc>>, - queue_id: usize, + cq_id: u16, + sq_id: u16, + cmd_id: u16, } struct CqReactor { int_sources: InterruptSources, @@ -43,25 +53,7 @@ impl CqReactor { let fd = syscall::open("event:", O_CLOEXEC | O_RDWR)?; let mut file = unsafe { File::from_raw_fd(fd as RawFd) }; - let mut msix_iter; - let mut msi_iter; - let mut intx_iter; - - let iter: &mut dyn Iterator = match int_sources { - InterruptSources::MsiX(ref btree) => { - msix_iter = btree.iter().map(|(&n, f)| (n, f)); - &mut msix_iter - } - InterruptSources::Msi(ref btree) => { - msi_iter = btree.iter().map(|(&n, f)| (u16::from(n), f)); - &mut msi_iter - } - InterruptSources::Intx(ref file) => { - intx_iter = std::iter::once((0, file)); - &mut intx_iter - } - }; - for (num, irq_handle) in iter { + for (num, irq_handle) in int_sources.iter_mut() { if file.write(&Event { id: irq_handle.as_raw_fd() as usize, flags: syscall::EVENT_READ, @@ -84,23 +76,95 @@ impl CqReactor { fn handle_notif_reqs(&mut self) { for req in self.receiver.try_iter() { match req { - NotifReq::RequestCompletion { queue_id, waker, message } => self.pending_reqs.push(PendingReq { - queue_id, + NotifReq::RequestCompletion { sq_id, cq_id, cmd_id, waker, message } => self.pending_reqs.push(PendingReq { + sq_id, + cq_id, + cmd_id, message, waker, }), } } } - fn block_on_new_irq(&mut self) -> Event { - let mut event = Event::default(); - self.event_queue.read(&mut event); - event + fn poll_completion_queues(&mut self, iv: u16) -> Option<()> { + let ivs_read_guard = self.nvme.cqs_for_ivs.read().unwrap(); + let cqs_read_guard = self.nvme.completion_queues.read().unwrap(); + + let mut entry_count = 0; + + for cq_id in ivs_read_guard.get(&iv)?.iter() { + let completion_queue_guard = cqs_read_guard.get(cq_id)?.lock().unwrap(); + let completion_queue: &mut NvmeCompQueue = &mut *completion_queue_guard; + + let entry = match completion_queue.complete() { + Some((_index, entry)) => entry, + None => continue, + }; + + self.try_notify_futures(cq_id, &entry); + + entry_count += 1; + } + if entry_count == 0 { + + } + + Some(()) } - fn run(mut self) -> ! { + fn try_notify_futures(&mut self, cq_id: CqId, entry: &NvmeComp) -> Option<()> { + let mut i = 0usize; + + let mut futures_notified = 0; + + while i < self.pending_reqs.len() { + let pending_req = &self.pending_reqs[i]; + + if pending_req.cq_id == cq_id && pending_req.sq_id == entry.sq_id && pending_req.cid == entry.cmd_id { + let pending_req_owned = self.pending_reqs.remove(i); + + *pending_req_owned.message.lock().unwrap() = Some(*entry); + pending_req_owned.waker.wake(); + + futures_notified += 1; + } else { + i += 1; + } + } + if futures_notified == 0 { + } + } + + fn run(mut self) { + let mut event = Event::default(); + let mut irq_word = [0u8; 8]; // stores the IRQ count + + const WORD_SIZE: usize = mem::size_of::(); + loop { self.handle_notif_reqs(); - let event = self.block_on_new_irq(); + + // block on getting the next event + if self.event_queue.read(&mut event) == 0 { + // event queue has been destroyed + break; + } + if event.flags & EVENT_READ == 0 { + continue; + } + + let (vector, irq_handle) = match self.int_sources.get_mut().nth(event.id) { + Some(s) => s, + None => continue, + }; + if irq_handle.read(&mut irq_word[..WORD_SIZE]) == 0 { + continue; + } + // acknowledge the interrupt (only necessary for level-triggered INTx# interrups) + if irq_handle.write(&irq_word[..WORD_SIZE]) == 0 { + continue; + } + + self.poll_completion_queues(vector); } } } @@ -108,9 +172,10 @@ impl CqReactor { pub fn start_cq_reactor_thread(nvme: Arc, interrupt_sources: InterruptSources, receiver: Receiver) -> thread::JoinHandle<()> { // Actually, nothing prevents us from spawning additional threads. the channel is MPMC and // everything is properly synchronized. I'm not saying this is strictly required, but with - // multiple completion queues it might actually be worth considering. The IRQ subsystem might - // be improved to lower the latency, but MSI-X allows multiple vectors to point to different - // CPUs, so that the load is balanced across the logical processors. + // multiple completion queues it might actually be worth considering. The (in-kernel) IRQ + // subsystem can have some room for improvement regarding lowering the latency, but MSI-X allows + // multiple vectors to point to different CPUs, so that the load can be balanced across the + // logical processors. thread::spawn(move || { CqReactor::new(nvme, interrupt_sources, receiver) .expect("nvmed: failed to setup CQ reactor") @@ -118,17 +183,17 @@ pub fn start_cq_reactor_thread(nvme: Arc, interrupt_sources: InterruptSour }) } -pub struct CompletionMessage { +struct CompletionMessage { cq_entry: NvmeComp, } enum CompletionFuture { // not really required, but makes futures inert - Init { - sender: Sender, - queue_id: usize, - }, Pending { + sender: Sender, + cq_id: CqId, + cmd_id: CmdId, + sq_id: SqId, message: Arc>>, }, Finished, @@ -144,24 +209,17 @@ impl Future for CompletionFuture { let this = self.get_mut(); match this { - &mut Self::Init { sender, queue_id } => { - let message = Arc::new(Mutex::new(None)); - sender.send(NotifReq::RequestCompletion { - queue_id, - waker: context.waker().clone(), - message: Arc::clone(&message), - }); - *this = CompletionFuture::Pending { - message, - }; - task::Poll::Pending - } - &mut Self::Pending { message } => if let Some(value) = message.lock().unwrap().take() { + &mut Self::Pending { message, cq_id, cmd_id, sq_id, sender } => if let Some(value) = message.lock().unwrap().take() { *this = Self::Finished; task::Poll::Ready(value.cq_entry) } else { - // woken up but the reactor hadn't sent the message. - // this is ideally unreachable + sender.send(NotifReq::RequestCompletion { + cq_id, + sq_id, + cmd_id, + waker: context.waker().clone(), + message: Arc::clone(&message), + }); task::Poll::Pending } &mut Self::Finished => panic!("calling poll() on an already finished CompletionFuture"), @@ -171,10 +229,15 @@ impl Future for CompletionFuture { impl Nvme { - pub fn completion(&self, cq_id: usize) -> impl Future + '_ { - CompletionFuture::Init { + /// Returns a future representing an eventual completion queue event, in `cq_id`, from `sq_id`, + /// with the individual command identified by `cmd_id`. + pub fn completion(&self, sq_id: usize, cmd_id: usize, cq_id: usize) -> impl Future + '_ { + CompletionFuture::Pending { sender: self.reactor_sender.clone(), - queue_id: cq_id, + cq_id, + cmd_id, + sq_id, + message: Arc::new(Mutex::new(None)), } } } diff --git a/nvmed/src/nvme/mod.rs b/nvmed/src/nvme/mod.rs index ac56cb9b78..0420020022 100644 --- a/nvmed/src/nvme/mod.rs +++ b/nvmed/src/nvme/mod.rs @@ -1,10 +1,11 @@ use std::collections::BTreeMap; use std::fs::File; use std::sync::{Mutex, RwLock}; -use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicUsize, AtomicU16, Ordering}; use std::ptr; use crossbeam_channel::Sender; +use smallvec::{smallvec, SmallVec}; use syscall::io::{Dma, Io, Mmio}; use syscall::error::{Error, Result, EINVAL}; @@ -21,9 +22,47 @@ pub enum InterruptSources { Msi(BTreeMap), Intx(File), } +impl InterruptSources { + pub fn iter_mut(&mut self) -> impl Iterator { + use std::collections::btree_map::IterMut as BTreeIterMut; + use std::iter::Once; + enum IterMut<'a> { + Msi(BTreeIterMut<'a, u16, File>), + MsiX(BTreeIterMut<'a, u8, File>), + Intx(Once<&'a mut File>), + } + impl<'a> Iterator for IterMut<'a> { + type Item = (u16, &'a mut File); + + fn next(&mut self) -> Option { + match self { + &mut Self::Msi(ref mut iter) => iter.next().map(|&mut (vector, ref mut handle)| (u16::from(vector), handle)), + &mut Self::MsiX(ref mut iter) => iter.next(), + &mut Self::Intx(ref mut iter) => (0, iter.next()), + } + } + fn size_hint(&self) -> (usize, Option) { + match self { + &Self::Msi(mut iter) => iter.size_hint(), + &Self::MsiX(mut iter) => iter.size_hint(), + &Self::Intx(mut iter) => iter.size_hint(), + } + } + } + + match self { + &mut Self::MsiX(ref mut map) => IterMut::MsiX(map.iter_mut()), + &mut Self::Msi(ref mut map) => IterMut::Msi(map.iter_mut()), + &mut Self::Intx(ref mut single) => IterMut::Intx(false, single), + } + } +} + +/// The way interrupts are sent. Unlike other PCI-based interfaces, like XHCI, it doesn't seem like +/// NVME supports operating with interrupts completely disabled. pub enum InterruptMethod { - /// INTx# interrupt pins + /// Traditional level-triggered, INTx# interrupt pins. Intx, /// Message signaled interrupts Msi(MsiCapability), @@ -85,6 +124,9 @@ pub struct NvmeCmd { cdw15: u32, } +pub struct IoCompletionQueueCreateInfo { +} + impl NvmeCmd { pub fn create_io_completion_queue(cid: u16, qid: u16, ptr: usize, size: u16) -> Self { Self { @@ -108,7 +150,7 @@ impl NvmeCmd { Self { opcode: 1, flags: 0, - cid: cid, + cid, nsid: 0, _rsvd: 0, mptr: 0, @@ -126,8 +168,8 @@ impl NvmeCmd { Self { opcode: 6, flags: 0, - cid: cid, - nsid: nsid, + cid, + nsid, _rsvd: 0, mptr: 0, dptr: [ptr as u64, 0], @@ -316,22 +358,37 @@ impl NvmeCompQueue { } } +#[derive(Debug)] pub struct NvmeNamespace { pub id: u32, pub blocks: u64, pub block_size: u64, } +pub type CqId = u16; +pub type SqId = u16; +pub type CmdId = u16; +pub type AtomicCqId = AtomicU16; +pub type AtomicSqId = AtomicU16; +pub type AtomicCmdId = AtomicU16; + pub struct Nvme { interrupt_method: Mutex, pcid_interface: Mutex, - regs: Mutex<&'static mut NvmeRegs>, - submission_queues: RwLock>>, - pub (crate) completion_queues: RwLock>>, + regs: RwLock<&'static mut NvmeRegs>, + + pub(crate) submission_queues: RwLock>>, + pub(crate) completion_queues: RwLock)>>>, + + // maps interrupt vectors with the completion queues they have + cqs_for_ivs: RwLock>>, + buffer: Mutex>, // 2MB of buffer buffer_prp: Mutex>, // 4KB of PRP for the buffer reactor_sender: Sender, - next_cid: AtomicUsize, + + next_sqid: AtomicSqId, + next_cqid: AtomicCqId, } unsafe impl Send for Nvme {} unsafe impl Sync for Nvme {} @@ -342,12 +399,17 @@ impl Nvme { regs: Mutex::new(unsafe { &mut *(address as *mut NvmeRegs) }), submission_queues: RwLock::new(vec! [Mutex::new(NvmeCmdQueue::new()?), Mutex::new(NvmeCmdQueue::new()?)]), completion_queues: RwLock::new(vec! [Mutex::new(NvmeCompQueue::new()?), Mutex::new(NvmeCompQueue::new()?)]), + // map the zero interrupt vector (which according to the spec shall always point to the + // admin completion queue) to CQID 0 (admin completion queue) + cqs_for_ivs: RwLock::new(std::iter::once((0, smallvec!(0))).collect()), buffer: Mutex::new(Dma::zeroed()?), buffer_prp: Mutex::new(Dma::zeroed()?), - next_cid: AtomicUsize::new(0), interrupt_method: Mutex::new(interrupt_method), pcid_interface: Mutex::new(pcid_interface), reactor_sender, + + next_sqid: AtomicSqId::new(0), + next_cqid: AtomicCqId::new(0), }) } unsafe fn doorbell_write(&self, index: usize, value: u32) { @@ -443,7 +505,7 @@ impl Nvme { } } } - pub fn init_with_queues(&self) -> BTreeMap { + pub async fn init_with_queues(&self) -> BTreeMap { { //TODO: Use buffer let data: Dma<[u8; 4096]> = Dma::zeroed().unwrap(); diff --git a/xhcid/src/main.rs b/xhcid/src/main.rs index 6e8befb546..66faca6e95 100644 --- a/xhcid/src/main.rs +++ b/xhcid/src/main.rs @@ -135,7 +135,7 @@ fn main() { let destination_id = read_bsp_apic_id().expect("xhcid: failed to read BSP apic id"); let lapic_id = u8::try_from(destination_id).expect("CPU id didn't fit inside u8"); - let msg_addr = x86_64_msix::message_address(lapic_id, false, false, 0b00); + let msg_addr = x86_64_msix::message_address(lapic_id, false, false); let (vector, interrupt_handle) = allocate_single_interrupt_vector(destination_id).expect("xhcid: failed to allocate interrupt vector").expect("xhcid: no interrupt vectors left"); let msg_data = x86_64_msix::message_data_edge_triggered(DeliveryMode::Fixed, vector); @@ -196,7 +196,7 @@ fn main() { let lapic_id = u8::try_from(destination_id).expect("xhcid: CPU id couldn't fit inside u8"); let rh = false; let dm = false; - let addr = x86_64_msix::message_address(lapic_id, rh, dm, 0b00); + let addr = x86_64_msix::message_address(lapic_id, rh, dm); let (vector, interrupt_handle) = allocate_single_interrupt_vector(destination_id).expect("xhcid: failed to allocate interrupt vector").expect("xhcid: no interrupt vectors left"); let msg_data = x86_64_msix::message_data_edge_triggered(DeliveryMode::Fixed, vector);