Add feature for nvmed to not use reactor thread

nvmed/Cargo.toml

@@ -17,3 +17,7 @@ smallvec = "1"
 block-io-wrapper = { path = "../block-io-wrapper" }
 
 pcid = { path = "../pcid" }
+
+[features]
+default = []
+async = []

nvmed/src/main.rs

@@ -349,6 +349,7 @@ fn daemon(daemon: redox_daemon::Daemon) -> ! {
     unsafe { nvme.init() }
     log::debug!("Finished base initialization");
     let nvme = Arc::new(nvme);
+    #[cfg(feature = "async")]
     let reactor_thread = nvme::cq_reactor::start_cq_reactor_thread(Arc::clone(&nvme), interrupt_sources, reactor_receiver);
     let namespaces = nvme.init_with_queues();
 
@@ -386,6 +387,7 @@ fn daemon(daemon: redox_daemon::Daemon) -> ! {
     }
 
     //TODO: destroy NVMe stuff
+    #[cfg(feature = "async")]
     reactor_thread.join().expect("nvmed: failed to join reactor thread");
 
     std::process::exit(0);

nvmed/src/nvme/cq_reactor.rs

@@ -141,7 +141,7 @@ impl CqReactor {
             let mut completion_queue_guard = cqs_read_guard.get(&cq_id)?.lock().unwrap();
             let &mut (ref mut completion_queue, _) = &mut *completion_queue_guard;
 
-            while let Some((head, entry)) = completion_queue.complete() {
+            while let Some((head, entry)) = completion_queue.complete(None) {
                 unsafe { self.nvme.completion_queue_head(cq_id, head) };
 
                 log::trace!("Got completion queue entry (CQID {}): {:?} at {}", cq_id, entry, head);

nvmed/src/nvme/mod.rs

@@ -412,6 +412,70 @@ impl Nvme {
         self.set_vectors_masked(std::iter::once((vector, masked)))
     }
 
+    #[cfg(not(feature = "async"))]
+    pub fn submit_and_complete_command<F: FnOnce(CmdId) -> NvmeCmd>(&self, sq_id: SqId, cmd_init: F) -> NvmeComp {
+        // Submit command
+        let cmd = {
+            let sqs_read_guard = self.submission_queues.read().unwrap();
+            let &(ref sq_lock, cq_id) = sqs_read_guard
+                .get(&sq_id)
+                .expect("nvmed: internal error: given SQ for SQ ID not there");
+            let mut sq_guard = sq_lock.lock().unwrap();
+            let sq = &mut *sq_guard;
+
+            assert!(!sq.is_full());
+
+            let cmd_id =
+                u16::try_from(sq.tail).expect("nvmed: internal error: CQ has more than 2^16 entries");
+            let cmd = cmd_init(cmd_id);
+            log::trace!("Sent submission queue entry (SQID {}): {:?} at {}", sq_id, cmd, cmd_id);
+            let tail = sq.submit_unchecked(cmd);
+            let tail = u16::try_from(tail).unwrap();
+
+            // make sure that we register interest before the reactor can get notified
+            unsafe { self.submission_queue_tail(sq_id, tail) };
+
+            cmd
+        };
+
+        // Read completion
+        loop {
+            for (cq_id, completion_queue_lock) in self.completion_queues.read().unwrap().iter() {
+                if *cq_id != sq_id {
+                    // Currently, CQ and SQ IDs have to match
+                    continue;
+                }
+
+                let mut completion_queue_guard = completion_queue_lock.lock().unwrap();
+                let &mut (ref mut completion_queue, _) = &mut *completion_queue_guard;
+
+                while let Some((head, entry)) = completion_queue.complete(Some((sq_id, cmd))) {
+                    unsafe { self.completion_queue_head(*cq_id, head) };
+
+                    log::trace!("Got completion queue entry (CQID {}): {:?} at {}", cq_id, entry, head);
+
+                    assert_eq!(sq_id, entry.sq_id);
+                    assert_eq!(cmd.cid, entry.cid);
+
+                    {
+                        let submission_queues_read_lock = self.submission_queues.read().unwrap();
+                        // this lock is actually important, since it will block during submission from other
+                        // threads. the lock won't be held for long by the submitters, but it still prevents
+                        // the entry being lost before this reactor is actually able to respond:
+                        let &(ref sq_lock, corresponding_cq_id) = submission_queues_read_lock.get(&{entry.sq_id}).expect("nvmed: internal error: queue returned from controller doesn't exist");
+                        assert_eq!(*cq_id, corresponding_cq_id);
+                        let mut sq_guard = sq_lock.lock().unwrap();
+                        sq_guard.head = entry.sq_head;
+                    }
+
+                    return entry;
+                }
+            }
+            unsafe { pause(); }
+        }
+    }
+
+    #[cfg(feature = "async")]
     pub fn submit_and_complete_command<F: FnOnce(CmdId) -> NvmeCmd>(&self, sq_id: SqId, cmd_init: F) -> NvmeComp {
         use crate::nvme::cq_reactor::{CompletionFuture, CompletionFutureState};
         futures::executor::block_on(

nvmed/src/nvme/queues.rs

@@ -62,8 +62,18 @@ impl NvmeCompQueue {
     }
 
     /// Get a new completion queue entry, or return None if no entry is available yet.
-    pub(crate) fn complete(&mut self) -> Option<(u16, NvmeComp)> {
+    pub(crate) fn complete(&mut self, cmd_opt: Option<(u16, NvmeCmd)>) -> Option<(u16, NvmeComp)> {
         let entry = unsafe { ptr::read_volatile(self.data.as_ptr().add(self.head as usize)) };
+
+        //HACK FOR SOMETIMES RETURNING INVALID DATA ON QEMU!
+        if let Some((sq_id, cmd)) = cmd_opt {
+            if entry.sq_id != sq_id
+            || entry.cid != cmd.cid
+            {
+                return None;
+            }
+        }
+
         if ((entry.status & 1) == 1) == self.phase {
             self.head = (self.head + 1) % (self.data.len() as u16);
             if self.head == 0 {
@@ -76,10 +86,10 @@ impl NvmeCompQueue {
     }
 
     /// Get a new CQ entry, busy waiting until an entry appears.
-    fn complete_spin(&mut self) -> (u16, NvmeComp) {
+    fn complete_spin(&mut self, cmd_opt: Option<(u16, NvmeCmd)>) -> (u16, NvmeComp) {
         log::debug!("Waiting for new CQ entry");
         loop {
-            if let Some(some) = self.complete() {
+            if let Some(some) = self.complete(cmd_opt) {
                 return some;
             } else {
                 unsafe { super::pause(); }