diff --git a/nvmed/Cargo.toml b/nvmed/Cargo.toml index af7288f810..f219f0100d 100644 --- a/nvmed/Cargo.toml +++ b/nvmed/Cargo.toml @@ -17,3 +17,7 @@ smallvec = "1" block-io-wrapper = { path = "../block-io-wrapper" } pcid = { path = "../pcid" } + +[features] +default = [] +async = [] diff --git a/nvmed/src/main.rs b/nvmed/src/main.rs index f1b37f3c48..6d11e209bb 100644 --- a/nvmed/src/main.rs +++ b/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); diff --git a/nvmed/src/nvme/cq_reactor.rs b/nvmed/src/nvme/cq_reactor.rs index 0e817ff4c0..5926684974 100644 --- a/nvmed/src/nvme/cq_reactor.rs +++ b/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); diff --git a/nvmed/src/nvme/mod.rs b/nvmed/src/nvme/mod.rs index 5d3aae1e20..f21f478545 100644 --- a/nvmed/src/nvme/mod.rs +++ b/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 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 NvmeCmd>(&self, sq_id: SqId, cmd_init: F) -> NvmeComp { use crate::nvme::cq_reactor::{CompletionFuture, CompletionFutureState}; futures::executor::block_on( diff --git a/nvmed/src/nvme/queues.rs b/nvmed/src/nvme/queues.rs index dfd0e6871f..7e06373efa 100644 --- a/nvmed/src/nvme/queues.rs +++ b/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(); }