vasilito/RedBear-OS
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implement asynchronous command __submission__.

Browse Source 
Note that by writing submission, I'm referring to blocking until a
submission queue has more entries available. The command completion
handling is already async.
This commit is contained in:
4lDO2
2020-05-01 13:14:31 +02:00
parent c61e886108
commit 64e9eea9b0
3 changed files with 184 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nvmed/src/nvme/cq_reactor.rs
+117 -65
View File
@@ -19,7 +19,7 @@ use syscall::Result;
use crossbeam_channel::{Receiver, Sender};
use crate::nvme::{CmdId, CqId, InterruptSources, Nvme, NvmeComp, NvmeCompQueue, SqId};
use super::{CmdId, CqId, InterruptSources, Nvme, NvmeComp, NvmeCmd, 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.
@@ -35,14 +35,24 @@ pub enum NotifReq {
// TODO: Maybe the `remem` crate.
message: Arc<Mutex<Option<CompletionMessage>>>,
},
RequestAvailSubmission {
sq_id: SqId,
waker: task::Waker,
}
}
struct PendingReq {
waker: task::Waker,
message: Arc<Mutex<Option<CompletionMessage>>>,
cq_id: u16,
sq_id: u16,
cmd_id: u16,
enum PendingReq {
PendingCompletion {
waker: task::Waker,
message: Arc<Mutex<Option<CompletionMessage>>>,
cq_id: CqId,
sq_id: SqId,
cmd_id: CmdId,
},
PendingAvailSubmission {
waker: task::Waker,
sq_id: SqId,
},
}
struct CqReactor {
int_sources: InterruptSources,
@@ -94,13 +104,14 @@ impl CqReactor {
cmd_id,
waker,
message,
} => self.pending_reqs.push(PendingReq {
} => self.pending_reqs.push(PendingReq::PendingCompletion {
sq_id,
cq_id,
cmd_id,
message,
waker,
}),
NotifReq::RequestAvailSubmission { sq_id, waker } => self.pending_reqs.push(PendingReq::PendingAvailSubmission { sq_id, waker, }),
}
}
}
@@ -119,7 +130,9 @@ impl CqReactor {
None => continue,
};
self.nvme.completion_queue_head(cq_id, head as u16);
self.nvme.completion_queue_head(cq_id, head);
self.nvme.submission_queues.read().unwrap().get(&entry.sq_id).expect("nvmed: internal error: queue returned from controller doesn't exist").lock().unwrap().head = entry.sq_head;
self.try_notify_futures(cq_id, &entry);
@@ -129,27 +142,54 @@ impl CqReactor {
Some(())
}
fn finish_pending_completion(&mut self, req_cq_id: CqId, cq_id: CqId, sq_id: SqId, cmd_id: CmdId, entry: &NvmeComp, i: usize) -> bool {
if req_cq_id == cq_id
&& sq_id == entry.sq_id
&& cmd_id == entry.cid
{
let (waker, message) = match self.pending_reqs.remove(i) {
PendingReq::PendingCompletion { waker, message, .. } => (waker, message),
_ => unreachable!(),
};
*message.lock().unwrap() = Some(CompletionMessage { cq_entry: *entry });
waker.wake();
true
} else {
false
}
}
fn finish_pending_avail_submission(&mut self, sq_id: SqId, entry: &NvmeComp, i: usize) -> bool {
if sq_id == entry.sq_id {
let waker = match self.pending_reqs.remove(i) {
PendingReq::PendingAvailSubmission { waker, .. } => waker,
_ => unreachable!(),
};
waker.wake();
true
} else {
false
}
}
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.cmd_id == entry.cid
{
let pending_req_owned = self.pending_reqs.remove(i);
*pending_req_owned.message.lock().unwrap() =
Some(CompletionMessage { cq_entry: *entry });
pending_req_owned.waker.wake();
futures_notified += 1;
} else {
i += 1;
match &self.pending_reqs[i] {
&PendingReq::PendingCompletion { cq_id: req_cq_id, sq_id, cmd_id, .. } => if self.finish_pending_completion(req_cq_id, cq_id, sq_id, cmd_id, entry, i) {
futures_notified += 1;
} else {
i += 1;
}
&PendingReq::PendingAvailSubmission { sq_id, .. } => if self.finish_pending_avail_submission(sq_id, entry, i) {
futures_notified += 1;
} else {
i += 1;
}
}
}
if futures_notified == 0 {}
@@ -284,48 +324,60 @@ impl Nvme {
/// Returns a future representing a submission queue becoming non-full. Make sure that the
/// queue doesn't have any additional free entries first though, so that the reactor doesn't
/// have to interfere.
pub fn wait_for_available_submission(&self, sq_id: SqId) -> AvailableSqEntryFuture {
todo!()
}
}
struct AvailMessage {
cmd_id: CmdId,
}
enum AvailableSqEntryFutureState {
Pending {
sq_id: SqId,
message: Option<Arc<Mutex<Option<AvailMessage>>>>,
},
Finished,
}
pub struct AvailableSqEntryFuture {
state: AvailableSqEntryFutureState,
}
impl Unpin for AvailableSqEntryFuture {}
impl Future for AvailableSqEntryFuture {
type Output = CmdId;
fn poll(self: Pin<&mut Self>, context: &mut task::Context<'_>) -> task::Poll<Self::Output> {
let this = &mut self.get_mut().state;
match this {
&mut AvailableSqEntryFutureState::Pending {
pub fn wait_for_available_submission<'a, F: FnOnce(CmdId) -> NvmeCmd>(&'a self, sq_id: SqId, f: F) -> SubmissionFuture<'a, F> {
SubmissionFuture {
state: SubmissionFutureState::Pending {
sq_id,
ref mut message,
} => {
if let Some(message) = message.lock().unwrap().take() {
} else {
task::Poll::Pending
}
}
&mut AvailableSqEntryFutureState::Finished => {
panic!("calling poll() on an already finished AvailableSqEntryFuture")
}
cmd_init: f,
nvme: &self,
},
}
}
}
pub(crate) enum SubmissionFutureState<'a, F> {
// the queue was known to be full when checked, thus the reactor is asked
Pending {
sq_id: SqId,
cmd_init: F,
nvme: &'a Nvme,
},
// returned when there was an available submission entry from the beginning
Ready(CmdId),
Finished,
}
/// A future representing a submission queue eventually becoming non-full. In most cases this
/// future will finish directly, since all entries in the queue have to be occupied for it to block.
pub struct SubmissionFuture<'a, F> {
pub(crate) state: SubmissionFutureState<'a, F>,
}
impl<F> Unpin for SubmissionFuture<'_, F> {}
impl<F: FnOnce(CmdId) -> NvmeCmd> Future for SubmissionFuture<'_, F> {
type Output = CmdId;
fn poll(self: Pin<&mut Self>, context: &mut task::Context<'_>) -> task::Poll<Self::Output> {
let state = &mut self.get_mut().state;
match state {
&mut SubmissionFutureState::Pending { sq_id, cmd_init, nvme } => match nvme.try_submit_command(sq_id, cmd_init) {
Ok(cmd_id) => {
*state = SubmissionFutureState::Finished;
task::Poll::Ready(cmd_id)
}
Err(closure) => {
nvme.reactor_sender.send(NotifReq::RequestAvailSubmission { sq_id, waker: context.waker().clone() });
*state = SubmissionFutureState::Pending { sq_id, cmd_init: closure, nvme };
task::Poll::Pending
}
}
&mut SubmissionFutureState::Ready(value) => {
*state = SubmissionFutureState::Finished;
task::Poll::Ready(value)
}
&mut SubmissionFutureState::Finished => panic!("calling poll() on an already finished SubmissionFuture"),
}
}
}
nvmed/src/nvme/mod.rs
+37 -16
View File
@@ -186,9 +186,9 @@ pub enum FullSqHandling {
Wait,
}
pub enum Submission {
Nonblocking(Option<CmdId>), // TODO: Add full error
MaybeBlocking(),
pub enum SubmissionBehavior<'a, F: FnOnce(CmdId) -> NvmeCmd> {
Nonblocking(Result<CmdId, F>), // TODO: Add full error
Future(self::cq_reactor::SubmissionFuture<'a, F>),
}
impl Nvme {
@@ -396,30 +396,51 @@ impl Nvme {
self.set_vectors_masked(std::iter::once((vector, masked)))
}
/// Try submitting a new entry to the specified submission queue, or return None if the queue
/// was full.
pub fn try_submit_command<F: FnOnce(CmdId) -> NvmeCmd>(
&self,
sq_id: SqId,
full_sq_handling: FullSqHandling,
f: F,
) -> Option<CmdId> {
pub fn submit_command_generic<'a, F: FnOnce(CmdId) -> NvmeCmd>(&'a self, sq_id: SqId, full_sq_handling: FullSqHandling, cmd_init: F) -> SubmissionBehavior<'a, F> {
let sqs_read_guard = self.submission_queues.read().unwrap();
let sq_lock = sqs_read_guard
.get(&sq_id)
.expect("nvmed: internal error: given SQ for SQ ID not there")
.lock()
.unwrap();
if sq_lock.is_full() {
match full_sq_handling {
FullSqHandling::ErrorDirectly => return SubmissionBehavior::Nonblocking(Err(cmd_init)),
FullSqHandling::Wait => return SubmissionBehavior::Future(self.wait_for_available_submission(sq_id, cmd_init)),
}
}
let cmd_id =
u16::try_from(sq_lock.i).expect("nvmed: internal error: CQ has more than 2^16 entries");
let tail = sq_lock.submit(f(cmd_id))?;
u16::try_from(sq_lock.tail).expect("nvmed: internal error: CQ has more than 2^16 entries");
let tail = sq_lock.submit_unchecked(cmd_init(cmd_id));
let tail = u16::try_from(tail).unwrap();
self.submission_queue_tail(sq_id, tail);
Some(cmd_id)
match full_sq_handling {
FullSqHandling::ErrorDirectly => SubmissionBehavior::Nonblocking(Ok(cmd_id)),
FullSqHandling::Wait => SubmissionBehavior::Future(self::cq_reactor::SubmissionFuture { state: self::cq_reactor::SubmissionFutureState::Ready(cmd_id) })
}
}
/// Try submitting a new entry to the specified submission queue, or return None if the queue
/// was full.
pub fn try_submit_command<F: FnOnce(CmdId) -> NvmeCmd>(
&self,
sq_id: SqId,
f: F,
) -> Result<CmdId, F> {
match self.submit_command_generic(sq_id, FullSqHandling::ErrorDirectly, f) {
SubmissionBehavior::Nonblocking(opt) => opt,
_ => unreachable!(),
}
}
pub async fn submit_command_async<F: FnOnce(CmdId) -> NvmeCmd>(&self, sq_id: SqId, f: F) -> CmdId {
match self.submit_command_generic(sq_id, FullSqHandling::Wait, f) {
SubmissionBehavior::Future(future) => future.await,
_ => unreachable!(),
}
}
pub async fn submit_admin_command<F: FnOnce(CmdId) -> NvmeCmd>(&self, f: F) -> CmdId {
self.try_submit_command(0, FullSqHandling::Wait, f);
todo!()
self.submit_command_async(0, f).await
}
pub async fn admin_queue_completion(&self, cmd_id: CmdId) -> NvmeComp {
self.completion(0, cmd_id, 0).await
nvmed/src/nvme/queues.rs
+30 -26
View File
@@ -47,37 +47,37 @@ pub struct NvmeComp {
/// Completion queue
pub struct NvmeCompQueue {
pub data: Dma<[NvmeComp; 256]>,
pub i: usize,
pub data: Dma<[NvmeComp]>,
pub head: u16,
pub phase: bool,
}
impl NvmeCompQueue {
pub fn new() -> Result<Self> {
Ok(Self {
data: Dma::zeroed()?,
i: 0,
data: Dma::zeroed_unsized(256)?,
head: 0,
phase: true,
})
}
/// Get a new completion queue entry, or return None if no entry is available yet.
pub(crate) fn complete(&mut self) -> Option<(usize, NvmeComp)> {
let entry = unsafe { ptr::read_volatile(self.data.as_ptr().add(self.i)) };
pub(crate) fn complete(&mut self) -> Option<(u16, NvmeComp)> {
let entry = unsafe { ptr::read_volatile(self.data.as_ptr().add(self.head as usize)) };
// println!("{:?}", entry);
if ((entry.status & 1) == 1) == self.phase {
self.i = (self.i + 1) % self.data.len();
if self.i == 0 {
self.head = (self.head + 1) % (self.data.len() as u16);
if self.head == 0 {
self.phase = !self.phase;
}
Some((self.i, entry))
Some((self.head, entry))
} else {
None
}
}
/// Get a new CQ entry, busy waiting until an entry appears.
fn complete_spin(&mut self) -> (usize, NvmeComp) {
fn complete_spin(&mut self) -> (u16, NvmeComp) {
loop {
if let Some(some) = self.complete() {
return some;
@@ -90,28 +90,32 @@ impl NvmeCompQueue {
/// Submission queue
pub struct NvmeCmdQueue {
pub data: Dma<[NvmeCmd; 64]>,
pub i: usize,
pub data: Dma<[NvmeCmd]>,
pub tail: u16,
pub head: u16,
}
impl NvmeCmdQueue {
pub(crate) fn new() -> Result<Self> {
pub fn new() -> Result<Self> {
Ok(Self {
data: Dma::zeroed()?,
i: 0,
data: Dma::zeroed_unsized(64)?,
tail: 0,
head: 0,
})
}
/// Add a new submission command entry to the queue. Returns Some(tail) when a vacant entry was
/// found, or None if the queue was full.
pub(crate) fn submit(&mut self, entry: NvmeCmd) -> Option<usize> {
// FIXME: Check for full conditions
if true {
self.data[self.i] = entry;
self.i = (self.i + 1) % self.data.len();
Some(self.i)
} else {
None
}
pub fn is_empty(&self) -> bool {
self.head == self.tail
}
pub fn is_full(&self) -> bool {
self.head + 1 == self.tail
}
/// Add a new submission command entry to the queue. The caller must ensure that the queue have free
/// entries; this can be checked using `is_full`.
pub fn submit_unchecked(&mut self, entry: NvmeCmd) -> u16 {
self.data[self.tail as usize] = entry;
self.tail = (self.tail + 1) % (self.data.len() as u16);
self.tail
}
}