WIP: Complete the NVME CQ reactor (doesn't compile yet)

2020-04-29 19:35:48 +02:00
parent d485176bd6
commit 6c2f103841
5 changed files with 200 additions and 73 deletions
@@ -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]]
@@ -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" }
@@ -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<Mutex<Option<CompletionMessage>>>,
    },
 }
@@ -28,7 +36,9 @@ pub enum NotifReq {
 struct PendingReq {
    waker: task::Waker,
    message: Arc<Mutex<Option<CompletionMessage>>>,
-    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<Item = (u16, &File)> = 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::<usize>();
+
        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<Nvme>, interrupt_sources: InterruptSources, receiver: Receiver<NotifReq>) -> 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<Nvme>, interrupt_sources: InterruptSour
    })
 }

-pub struct CompletionMessage {
+struct CompletionMessage {
    cq_entry: NvmeComp,
 }

 enum CompletionFuture {
    // not really required, but makes futures inert
-    Init {
-        sender: Sender<NotifReq>,
-        queue_id: usize,
-    },
    Pending {
+        sender: Sender<NotifReq>,
+        cq_id: CqId,
+        cmd_id: CmdId,
+        sq_id: SqId,
        message: Arc<Mutex<Option<CompletionMessage>>>,
    },
    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<Output = NvmeComp> + '_ {
-        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<Output = NvmeComp> + '_ {
+        CompletionFuture::Pending {
            sender: self.reactor_sender.clone(),
-            queue_id: cq_id,
+            cq_id,
+            cmd_id,
+            sq_id,
+            message: Arc::new(Mutex::new(None)),
        }
    }
 }
@@ -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<u8, File>),
    Intx(File),
 }
+impl InterruptSources {
+    pub fn iter_mut(&mut self) -> impl Iterator<Item = (u16, &mut File)> {
+        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<Self::Item> {
+                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<usize>) {
+                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<InterruptMethod>,
    pcid_interface: Mutex<PcidServerHandle>,
-    regs: Mutex<&'static mut NvmeRegs>,
-    submission_queues: RwLock<Vec<Mutex<NvmeCmdQueue>>>,
-    pub (crate) completion_queues: RwLock<Vec<Mutex<NvmeCompQueue>>>,
+    regs: RwLock<&'static mut NvmeRegs>,
+
+    pub(crate) submission_queues: RwLock<BTreeMap<SqId, Mutex<NvmeCmdQueue>>>,
+    pub(crate) completion_queues: RwLock<BTreeMap<CqId, Mutex<(NvmeCompQueue, SmallVec<[SqId; 16]>)>>>,
+
+    // maps interrupt vectors with the completion queues they have
+    cqs_for_ivs: RwLock<BTreeMap<u16, SmallVec<[CqId; 4]>>>,
+
    buffer: Mutex<Dma<[u8; 512 * 4096]>>, // 2MB of buffer
    buffer_prp: Mutex<Dma<[u64; 512]>>, // 4KB of PRP for the buffer
    reactor_sender: Sender<cq_reactor::NotifReq>,
-    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<u32, NvmeNamespace> {
+    pub async fn init_with_queues(&self) -> BTreeMap<u32, NvmeNamespace> {
        {
            //TODO: Use buffer
            let data: Dma<[u8; 4096]> = Dma::zeroed().unwrap();
@@ -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);