diff --git a/Cargo.lock b/Cargo.lock index 476cf221e7..7f911a6daa 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -32,7 +32,7 @@ dependencies = [ "parking_lot 0.12.3", "plain", "redox-daemon", - "redox-scheme", + "redox-scheme 0.4.0", "redox_event", "redox_syscall", "ron", @@ -193,7 +193,7 @@ dependencies = [ "orbclient", "pcid", "redox-daemon", - "redox-scheme", + "redox-scheme 0.4.0", "redox_syscall", ] @@ -391,9 +391,11 @@ checksum = "22ec99545bb0ed0ea7bb9b8e1e9122ea386ff8a48c0922e43f36d45ab09e0e80" name = "driver-block" version = "0.1.0" dependencies = [ + "executor", "libredox", + "log", "partitionlib", - "redox-scheme", + "redox-scheme 0.5.0", "redox_syscall", ] @@ -406,7 +408,7 @@ dependencies = [ "inputd", "libredox", "log", - "redox-scheme", + "redox-scheme 0.4.0", "redox_syscall", ] @@ -415,7 +417,7 @@ name = "driver-network" version = "0.1.0" dependencies = [ "libredox", - "redox-scheme", + "redox-scheme 0.4.0", "redox_syscall", ] @@ -439,6 +441,15 @@ version = "1.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5443807d6dff69373d433ab9ef5378ad8df50ca6298caf15de6e52e24aaf54d5" +[[package]] +name = "executor" +version = "0.1.0" +dependencies = [ + "log", + "redox_event", + "slab", +] + [[package]] name = "fbbootlogd" version = "0.1.0" @@ -450,7 +461,7 @@ dependencies = [ "orbclient", "ransid", "redox-daemon", - "redox-scheme", + "redox-scheme 0.4.0", "redox_event", "redox_syscall", ] @@ -466,7 +477,7 @@ dependencies = [ "orbclient", "ransid", "redox-daemon", - "redox-scheme", + "redox-scheme 0.4.0", "redox_event", "redox_syscall", ] @@ -720,7 +731,7 @@ dependencies = [ "log", "orbclient", "redox-daemon", - "redox-scheme", + "redox-scheme 0.4.0", "redox_syscall", ] @@ -965,7 +976,7 @@ dependencies = [ "pico-args", "plain", "redox-daemon", - "redox-scheme", + "redox-scheme 0.4.0", "redox_syscall", "serde", ] @@ -1125,6 +1136,16 @@ dependencies = [ "redox_syscall", ] +[[package]] +name = "redox-scheme" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a96b9cfb034251dfb0aaa66a67059a7f0ea344039904d1d70cd36266af9c8a2f" +dependencies = [ + "libredox", + "redox_syscall", +] + [[package]] name = "redox_event" version = "0.4.1" @@ -1137,9 +1158,9 @@ dependencies = [ [[package]] name = "redox_syscall" -version = "0.5.9" +version = "0.5.10" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "82b568323e98e49e2a0899dcee453dd679fae22d69adf9b11dd508d1549b7e2f" +checksum = "0b8c0c260b63a8219631167be35e6a988e9554dbd323f8bd08439c8ed1302bd1" dependencies = [ "bitflags 2.6.0", ] @@ -1915,7 +1936,7 @@ dependencies = [ "pcid", "plain", "redox-daemon", - "redox-scheme", + "redox-scheme 0.4.0", "redox_event", "redox_syscall", "regex", diff --git a/Cargo.toml b/Cargo.toml index 14dd17de9d..d27f9e6e46 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,7 +1,9 @@ [workspace] members = [ - "acpid", "common", + "executor", + + "acpid", "hwd", "pcid", "pcid-spawner", diff --git a/executor/Cargo.toml b/executor/Cargo.toml new file mode 100644 index 0000000000..0d5b12f070 --- /dev/null +++ b/executor/Cargo.toml @@ -0,0 +1,12 @@ +[package] +name = "executor" +authors = ["4lDO2 <4lDO2@protonmail.com>"] +version = "0.1.0" +edition = "2021" +license = "MIT" +description = "Async framework for queue-based HW interfaces" + +[dependencies] +log = "0.4" +redox_event = "0.4.1" +slab = "0.4.9" diff --git a/executor/src/lib.rs b/executor/src/lib.rs new file mode 100644 index 0000000000..6b54e07bc1 --- /dev/null +++ b/executor/src/lib.rs @@ -0,0 +1,396 @@ +use std::cell::{Cell, RefCell}; +use std::collections::{HashMap, VecDeque}; +use std::fmt::Debug; +use std::fs::File; +use std::future::{Future, IntoFuture}; +use std::hash::Hash; +use std::io::{Read, Write}; +use std::marker::PhantomData; +use std::os::fd::AsRawFd; +use std::panic::AssertUnwindSafe; +use std::pin::Pin; +use std::ptr::NonNull; +use std::rc::Rc; +use std::task; + +use event::{EventFlags, RawEventQueue}; +use slab::Slab; + +type EventUserData = usize; + +type FutIdx = usize; + +pub trait Hardware: Sized { + type CmdId: Clone + Copy + Debug + Hash + Eq + PartialEq; + type CqId: Clone + Copy + Debug + Hash + Eq + PartialEq; + type SqId: Clone + Copy + Debug + Hash + Eq + PartialEq; + type Sqe: Debug + Clone + Copy; + type Cqe; + type Iv: Clone + Copy + Debug; + + type GlobalCtxt; + + // TODO: the kernel should also do this automatically before sending EOI messages to the IC + fn mask_vector(ctxt: &Self::GlobalCtxt, iv: Self::Iv); + fn unmask_vector(ctxt: &Self::GlobalCtxt, iv: Self::Iv); + + fn set_sqe_cmdid(sqe: &mut Self::Sqe, id: Self::CmdId); + fn get_cqe_cmdid(cqe: &Self::Cqe) -> Self::CmdId; + + // TODO: support multiple SQs per CQ or vice versa? + fn sq_cq(ctxt: &Self::GlobalCtxt, id: Self::CqId) -> Self::SqId; + + fn current() -> Rc>; + fn vtable() -> &'static task::RawWakerVTable; + + fn try_submit( + ctxt: &Self::GlobalCtxt, + sq_id: Self::SqId, + success: impl FnOnce(Self::CmdId) -> Self::Sqe, + fail: impl FnOnce(), + ) -> Option<(Self::CqId, Self::CmdId)>; + fn poll_cqes(ctxt: &Self::GlobalCtxt, handle: impl FnMut(Self::CqId, Self::Cqe)); +} + +/// Async executor, single IV, thread-per-core architecture +pub struct LocalExecutor { + global_ctxt: Hw::GlobalCtxt, + + queue: RawEventQueue, + vector: Hw::Iv, + irq_handle: File, + intx: bool, + + // TODO: One IV and SQ/CQ per core (where the admin queue can be managed by the main thread). + awaiting_submission: RefCell>>, + awaiting_completion: + RefCell>)>>>, + + external_event: RefCell)>>, + next_user_data: Cell, + + ready_futures: RefCell>, + futures: RefCell + 'static>>>>, + is_polling: Cell, +} + +impl LocalExecutor { + pub fn register_external_event( + &self, + fd: usize, + flags: event::EventFlags, + ) -> ExternalEventSource { + let user_data = self.next_user_data.get(); + self.next_user_data.set(user_data.checked_add(1).unwrap()); + + self.queue + .subscribe(fd, user_data, flags) + .expect("failed to subscribe to event"); + + ExternalEventSource { + flags: event::EventFlags::empty(), + user_data, + _not_send_or_unpin: PhantomData, + } + } + pub fn current() -> Rc { + Hw::current() + } + pub fn poll(&self) -> usize { + assert!(!self.is_polling.replace(true)); + + let mut finished = 0; + + for future_idx in self.ready_futures.borrow_mut().drain(..) { + let waker = waker::(future_idx); + + let mut futures = self.futures.borrow_mut(); + let res = match std::panic::catch_unwind(AssertUnwindSafe(|| { + futures[future_idx] + .as_mut() + .poll(&mut task::Context::from_waker(&waker)) + })) { + Ok(r) => r, + Err(_) => { + log::error!("Task panicked!"); + core::mem::forget(futures.remove(future_idx)); + continue; + } + }; + if res.is_ready() { + drop(futures.remove(future_idx)); + finished += 1; + } + } + self.is_polling.set(false); + + finished + } + pub fn spawn(&self, fut: impl IntoFuture + 'static) { + let idx = self + .futures + .borrow_mut() + .insert(Box::pin(fut.into_future())); + self.ready_futures.borrow_mut().push_back(idx); + } + pub fn block_on<'a, O: 'a>(&self, fut: impl IntoFuture + 'a) -> O { + let retval = Rc::new(RefCell::new(None)); + + let retval2 = Rc::clone(&retval); + let idx = self.futures.borrow_mut().insert({ + let t1: Pin + 'a>> = Box::pin(async move { + *retval2.borrow_mut() = Some(fut.await); + }); + // SAFETY: Apart from the lifetimes, the types are exactly the same. We also know + // block_on simply cannot return without having fully awaited and dropped the future, + // even if that future panics (cf. the catch_unwind invocation). + let t2: Pin + 'static>> = + unsafe { std::mem::transmute(t1) }; + + t2 + }); + + self.ready_futures.borrow_mut().push_front(idx); + + loop { + let finished = self.poll(); + if retval.borrow().is_some() { + break; + } + if finished == 0 { + self.react(); + } + } + + let o = retval.borrow_mut().take().unwrap(); + o + } + fn react(&self) { + let event = self.queue.next_event().expect("failed to get next event"); + + if event.user_data != 0 { + let Some((fut_idx, flags_ptr)) = + self.external_event.borrow_mut().remove(&event.user_data) + else { + // Spurious event + return; + }; + unsafe { + flags_ptr + .as_ptr() + .write(event::EventFlags::from_bits_retain(event.flags)); + } + self.ready_futures.borrow_mut().push_back(fut_idx); + return; + } + + if self.intx { + let mut buf = [0_u8; core::mem::size_of::()]; + if (&self.irq_handle).read(&mut buf).unwrap() != 0 { + (&self.irq_handle).write(&buf).unwrap(); + } + } + + // TODO: The kernel should probably do the masking (when using MSI/MSI-X at least), which + // should happen before EOI messages to the interrupt controller. + Hw::mask_vector(&self.global_ctxt, self.vector); + Hw::poll_cqes(&self.global_ctxt, |cq_id, cqe| { + if let Some((fut_idx, comp_ptr)) = self + .awaiting_completion + .borrow_mut() + .get_mut(&cq_id) + .and_then(|per_cmd| per_cmd.remove(&Hw::get_cqe_cmdid(&cqe))) + { + unsafe { + comp_ptr.as_ptr().write(Some(cqe)); + } + self.ready_futures.borrow_mut().push_back(fut_idx); + + if let Some(submitting) = self + .awaiting_submission + .borrow_mut() + .get_mut(&Hw::sq_cq(&self.global_ctxt, cq_id)) + .and_then(|q| q.pop_front()) + { + self.ready_futures.borrow_mut().push_back(submitting); + } + } + }); + Hw::unmask_vector(&self.global_ctxt, self.vector); + } + pub async fn submit(&self, sq_id: Hw::SqId, cmd: Hw::Sqe) -> Hw::Cqe { + CqeFuture:: { + state: State::::Submitting { sq_id, cmd }, + comp: None, + _not_send: PhantomData, + } + .await + } +} + +struct CqeFuture { + pub state: State, + pub comp: Option, + pub _not_send: PhantomData<*const ()>, +} +enum State { + Submitting { sq_id: Hw::SqId, cmd: Hw::Sqe }, + Completing { cq_id: Hw::CqId, cmd_id: Hw::CmdId }, +} + +fn current_executor_and_idx( + cx: &mut task::Context<'_>, +) -> (Rc>, FutIdx) { + let executor = LocalExecutor::current(); + + let idx = cx.waker().data() as FutIdx; + assert_eq!( + cx.waker().vtable() as *const _, + Hw::vtable(), + "incompatible executor for CqeFuture" + ); + + (executor, idx) +} + +impl Future for CqeFuture { + type Output = Hw::Cqe; + + fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> task::Poll { + let this = unsafe { self.get_unchecked_mut() }; + + let (executor, idx) = current_executor_and_idx::(cx); + + match this.state { + State::Submitting { sq_id, mut cmd } => { + let mut awaiting = executor.awaiting_submission.borrow_mut(); + + if let Some((cq_id, cmd_id)) = Hw::try_submit( + &executor.global_ctxt, + sq_id, + |cmd_id| { + Hw::set_sqe_cmdid(&mut cmd, cmd_id); + log::trace!("About to submit {cmd:?}"); + cmd + }, + || { + awaiting.entry(sq_id).or_default().push_back(idx); + }, + ) { + executor + .awaiting_completion + .borrow_mut() + .entry(cq_id) + .or_default() + .insert(cmd_id, (idx, (&mut this.comp).into())); + this.state = State::Completing { cq_id, cmd_id }; + } + task::Poll::Pending + } + State::Completing { cq_id, cmd_id } => match this.comp.take() { + Some(comp) => { + log::trace!("ready!"); + task::Poll::Ready(comp) + } + + // Shouldn't technically be possible + None => { + log::trace!("spurious poll"); + executor + .awaiting_completion + .borrow_mut() + .entry(cq_id) + .or_default() + .insert(cmd_id, (idx, (&mut this.comp).into())); + task::Poll::Pending + } + }, + } + } +} + +unsafe fn vt_clone(idx: *const ()) -> task::RawWaker { + task::RawWaker::new(idx, Hw::vtable()) +} +unsafe fn vt_drop(_idx: *const ()) {} +unsafe fn vt_wake(idx: *const ()) { + Hw::current() + .ready_futures + .borrow_mut() + .push_back(idx as FutIdx); +} + +fn waker(idx: FutIdx) -> task::Waker { + unsafe { task::Waker::from_raw(task::RawWaker::new(idx as *const (), Hw::vtable())) } +} +pub const fn vtable() -> task::RawWakerVTable { + task::RawWakerVTable::new(vt_clone::, vt_wake::, vt_wake::, vt_drop) +} + +pub struct ExternalEventSource { + flags: event::EventFlags, + user_data: EventUserData, + _not_send_or_unpin: PhantomData<(*const (), fn() -> Hw)>, +} +pub struct Event { + flags: event::EventFlags, + _not_send: PhantomData<*const ()>, +} +impl Event { + pub fn flags(&self) -> event::EventFlags { + self.flags + } +} +impl ExternalEventSource { + fn poll_next(self: Pin<&mut Self>, cx: &mut task::Context) -> task::Poll> { + let this = unsafe { self.get_unchecked_mut() }; + + let flags = std::mem::take(&mut this.flags); + + if flags.is_empty() { + let (executor, idx) = current_executor_and_idx::(cx); + executor + .external_event + .borrow_mut() + .insert(this.user_data, (idx, (&mut this.flags).into())); + return task::Poll::Pending; + } + task::Poll::Ready(Some(Event { + flags, + _not_send: PhantomData, + })) + } + pub async fn next(mut self: Pin<&mut Self>) -> Option { + core::future::poll_fn(|cx| self.as_mut().poll_next(cx)).await + } +} +pub fn init_raw( + global_ctxt: Hw::GlobalCtxt, + vector: Hw::Iv, + intx: bool, + irq_handle: File, +) -> LocalExecutor { + let queue = RawEventQueue::new().expect("failed to allocate event queue for local executor"); + + // TODO: Multiple CPUs + queue + .subscribe(irq_handle.as_raw_fd() as usize, 0, EventFlags::READ) + .expect("failed to subscribe to IRQ event"); + + LocalExecutor { + global_ctxt, + + queue, + vector, + intx, + irq_handle, + + awaiting_submission: RefCell::new(HashMap::new()), + awaiting_completion: RefCell::new(HashMap::new()), + external_event: RefCell::new(HashMap::new()), + next_user_data: Cell::new(1), + ready_futures: RefCell::new(VecDeque::new()), + futures: RefCell::new(Slab::with_capacity(16)), + is_polling: Cell::new(false), + } +} diff --git a/storage/driver-block/Cargo.toml b/storage/driver-block/Cargo.toml index 00e10150bb..1ae22f56e0 100644 --- a/storage/driver-block/Cargo.toml +++ b/storage/driver-block/Cargo.toml @@ -4,8 +4,10 @@ version = "0.1.0" edition = "2021" [dependencies] +executor = { path = "../../executor" } partitionlib = { path = "../partitionlib" } libredox = "0.1.3" -redox_syscall = "0.5" -redox-scheme = "0.4" +log = "0.4" +redox_syscall = { version = "0.5", features = ["std"] } +redox-scheme = "0.5" diff --git a/storage/driver-block/src/lib.rs b/storage/driver-block/src/lib.rs index 6efe0f5c2c..bd36c9a23b 100644 --- a/storage/driver-block/src/lib.rs +++ b/storage/driver-block/src/lib.rs @@ -1,4 +1,5 @@ use std::cmp; +use std::future::IntoFuture; use std::io::{self, Read, Seek, SeekFrom}; use std::collections::BTreeMap; @@ -6,15 +7,15 @@ use std::convert::TryFrom; use std::fmt::Write; use std::str; +use executor::LocalExecutor; use libredox::Fd; use partitionlib::{LogicalBlockSize, PartitionTable}; -use redox_scheme::{ - CallRequest, CallerCtx, OpenResult, RequestKind, SchemeBlock, SignalBehavior, Socket, -}; +use redox_scheme::scheme::SchemeAsync; +use redox_scheme::{CallerCtx, OpenResult, RequestKind, Response, SignalBehavior, Socket}; +use syscall::dirent::DirentBuf; use syscall::schemev2::NewFdFlags; use syscall::{ - Error, Result, Stat, EACCES, EAGAIN, EBADF, EINVAL, EISDIR, ENOENT, ENOLCK, EOVERFLOW, - MODE_DIR, MODE_FILE, O_DIRECTORY, O_STAT, + Error, Result, Stat, EACCES, EAGAIN, EBADF, EINTR, EINVAL, EISDIR, ENOENT, ENOLCK, EOPNOTSUPP, EOVERFLOW, EWOULDBLOCK, MODE_DIR, MODE_FILE, O_DIRECTORY, O_STAT }; /// Split the read operation into a series of block reads. @@ -71,8 +72,8 @@ pub trait Disk { // These operate on a whole multiple of the block size // FIXME maybe only operate on a single block worth of data? - fn read(&mut self, block: u64, buffer: &mut [u8]) -> syscall::Result>; - fn write(&mut self, block: u64, buffer: &[u8]) -> syscall::Result>; + async fn read(&mut self, block: u64, buffer: &mut [u8]) -> syscall::Result; + async fn write(&mut self, block: u64, buffer: &[u8]) -> syscall::Result; } impl Disk for Box { @@ -84,12 +85,12 @@ impl Disk for Box { (**self).size() } - fn read(&mut self, block: u64, buffer: &mut [u8]) -> syscall::Result> { - (**self).read(block, buffer) + async fn read(&mut self, block: u64, buffer: &mut [u8]) -> syscall::Result { + (**self).read(block, buffer).await } - fn write(&mut self, block: u64, buffer: &[u8]) -> syscall::Result> { - (**self).write(block, buffer) + async fn write(&mut self, block: u64, buffer: &[u8]) -> syscall::Result { + (**self).write(block, buffer).await } } @@ -99,18 +100,19 @@ pub struct DiskWrapper { } impl DiskWrapper { - pub fn pt(disk: &mut T) -> Option { + pub fn pt(disk: &mut T, executor: &impl ExecutorTrait) -> Option { let bs = match disk.block_size() { 512 => LogicalBlockSize::Lb512, 4096 => LogicalBlockSize::Lb4096, _ => return None, }; - struct Device<'a> { - disk: &'a mut dyn Disk, + struct Device<'a, D: Disk, E: ExecutorTrait> { + disk: &'a mut D, + executor: &'a E, offset: u64, } - impl<'a> Seek for Device<'a> { + impl<'a, D: Disk, E: ExecutorTrait> Seek for Device<'a, D, E> { fn seek(&mut self, from: SeekFrom) -> io::Result { let size = i64::try_from(self.disk.size()).or(Err(io::Error::new( io::ErrorKind::Other, @@ -129,7 +131,7 @@ impl DiskWrapper { } } // TODO: Perhaps this impl should be used in the rest of the scheme. - impl<'a> Read for Device<'a> { + impl<'a, D: Disk, E: ExecutorTrait> Read for Device<'a, D, E> { fn read(&mut self, buf: &mut [u8]) -> io::Result { let blksize = self.disk.block_size(); let size_in_blocks = self.disk.size() / u64::from(blksize); @@ -141,17 +143,9 @@ impl DiskWrapper { return Err(io::Error::from_raw_os_error(syscall::EOVERFLOW)); } loop { - match disk.read(block, block_bytes) { - Ok(Some(bytes)) => { - assert_eq!(bytes, block_bytes.len()); - return Ok(()); - } - Ok(None) => { - std::thread::yield_now(); - continue; - } - Err(err) => return Err(io::Error::from_raw_os_error(err.errno)), - } + let bytes = self.executor.block_on(disk.read(block, block_bytes))?; + assert_eq!(bytes, block_bytes.len()); + return Ok(()); } }; let bytes_read = block_read(self.offset, blksize, buf, read_block)?; @@ -161,14 +155,21 @@ impl DiskWrapper { } } - partitionlib::get_partitions(&mut Device { disk, offset: 0 }, bs) - .ok() - .flatten() + partitionlib::get_partitions( + &mut Device { + disk, + offset: 0, + executor, + }, + bs, + ) + .ok() + .flatten() } - pub fn new(mut disk: T) -> Self { + pub fn new(mut disk: T, executor: &impl ExecutorTrait) -> Self { Self { - pt: Self::pt(&mut disk), + pt: Self::pt(&mut disk, executor), disk, } } @@ -189,12 +190,12 @@ impl DiskWrapper { self.disk.size() } - pub fn read( + pub async fn read( &mut self, part_num: Option, block: u64, buf: &mut [u8], - ) -> syscall::Result> { + ) -> syscall::Result { if buf.len() as u64 % u64::from(self.disk.block_size()) != 0 { return Err(Error::new(EINVAL)); } @@ -214,18 +215,18 @@ impl DiskWrapper { let abs_block = part.start_lba + block; - self.disk.read(abs_block, buf) + self.disk.read(abs_block, buf).await } else { - self.disk.read(block, buf) + self.disk.read(block, buf).await } } - pub fn write( + pub async fn write( &mut self, part_num: Option, block: u64, buf: &[u8], - ) -> syscall::Result> { + ) -> syscall::Result { if buf.len() as u64 % u64::from(self.disk.block_size()) != 0 { return Err(Error::new(EINVAL)); } @@ -245,9 +246,9 @@ impl DiskWrapper { let abs_block = part.start_lba + block; - self.disk.write(abs_block, buf) + self.disk.write(abs_block, buf).await } else { - self.disk.write(block, buf) + self.disk.write(block, buf).await } } } @@ -264,11 +265,23 @@ pub struct DiskScheme { disks: BTreeMap>, handles: BTreeMap, next_id: usize, - blocked: Vec, +} + +pub trait ExecutorTrait { + fn block_on<'a, O: 'a>(&self, fut: impl IntoFuture + 'a) -> O; +} +impl ExecutorTrait for LocalExecutor { + fn block_on<'a, O: 'a>(&self, fut: impl IntoFuture + 'a) -> O { + LocalExecutor::block_on(self, fut) + } } impl DiskScheme { - pub fn new(scheme_name: String, disks: BTreeMap) -> Self { + pub fn new( + scheme_name: String, + disks: BTreeMap, + executor: &impl ExecutorTrait, + ) -> Self { assert!(scheme_name.starts_with("disk")); let socket = Socket::nonblock(&scheme_name).expect("failed to create disk scheme"); @@ -277,11 +290,10 @@ impl DiskScheme { socket, disks: disks .into_iter() - .map(|(k, disk)| (k, DiskWrapper::new(disk))) + .map(|(k, disk)| (k, DiskWrapper::new(disk, executor))) .collect(), next_id: 0, handles: BTreeMap::new(), - blocked: vec![], } } @@ -291,53 +303,45 @@ impl DiskScheme { /// Process pending and new requests. /// - /// This needs to be called each time there is a new event on the scheme - /// file and each time a read or write operation has completed. - // FIXME maybe split into one method for events on the scheme fd and one - // to call when an irq is received to indicate that blocked packets can - // be processed. - pub fn tick(&mut self) -> io::Result<()> { - // Handle any blocked requests - let mut i = 0; - while i < self.blocked.len() { - if let Some(resp) = self.blocked[i].handle_scheme_block(self) { - self.socket - .write_response(resp, SignalBehavior::Restart) - .expect("driver-block: failed to write scheme"); - self.blocked.remove(i); - } else { - i += 1; - } - } - + /// This needs to be called each time there is a new event on the scheme. + pub async fn tick(&mut self) -> io::Result<()> { // Handle new scheme requests loop { - let request = match self.socket.next_request(SignalBehavior::Restart) { + let request = match self.socket.next_request(SignalBehavior::Interrupt) { Ok(Some(request)) => request, Ok(None) => { // Scheme likely got unmounted + // TODO: return this to caller instead std::process::exit(0); } - Err(err) if err.errno == EAGAIN => break, + Err(error) if error.errno == EWOULDBLOCK || error.errno == EAGAIN => break, + Err(err) if err.errno == EINTR => continue, Err(err) => return Err(err.into()), }; - match request.kind() { + let response = match request.kind() { RequestKind::Call(call_request) => { - if let Some(resp) = call_request.handle_scheme_block(self) { - self.socket.write_response(resp, SignalBehavior::Restart)?; - } else { - self.blocked.push(call_request); - } + // TODO: Spawn a separate task for each scheme call. This would however require the + // use of a smarter buffer pool (or direct IO, or a buffer per fd) in order to do + // parallel IO. It might also require async-aware locks so that a close() is + // correctly ordered wrt IO on the same fd. + call_request.handle_async(self).await + } + RequestKind::SendFd(sendfd_request) => Response::err(EOPNOTSUPP, sendfd_request), + RequestKind::Cancellation(_cancellation_request) => { + // FIXME implement cancellation + continue; + } + RequestKind::MsyncMsg | RequestKind::MunmapMsg | RequestKind::MmapMsg => { + unreachable!() } RequestKind::OnClose { id } => { self.on_close(id); + continue; } - RequestKind::Cancellation(_cancellation_request) => { - // FIXME implement cancellation - } - _ => {} - } + }; + self.socket + .write_response(response, SignalBehavior::Restart)?; } Ok(()) @@ -373,13 +377,8 @@ impl DiskScheme { } } -impl SchemeBlock for DiskScheme { - fn xopen( - &mut self, - path_str: &str, - flags: usize, - ctx: &CallerCtx, - ) -> Result> { +impl SchemeAsync for DiskScheme { + async fn open(&mut self, path_str: &str, flags: usize, ctx: &CallerCtx) -> Result { if ctx.uid != 0 { return Err(Error::new(EACCES)); } @@ -446,18 +445,27 @@ impl SchemeBlock for DiskScheme { let id = self.next_id; self.next_id += 1; self.handles.insert(id, handle); - Ok(Some(OpenResult::ThisScheme { + Ok(OpenResult::ThisScheme { number: id, flags: NewFdFlags::POSITIONED, - })) + }) + } + async fn getdents<'buf>( + &mut self, + _id: usize, + _buf: DirentBuf<&'buf mut [u8]>, + _opaque_offset: u64, + ) -> Result> { + // TODO + Err(Error::new(EOPNOTSUPP)) } - fn fstat(&mut self, id: usize, stat: &mut Stat) -> Result> { + async fn fstat(&mut self, id: usize, stat: &mut Stat, _ctx: &CallerCtx) -> Result<()> { match *self.handles.get(&id).ok_or(Error::new(EBADF))? { Handle::List(ref data) => { stat.st_mode = MODE_DIR; stat.st_size = data.len() as u64; - Ok(Some(0)) + Ok(()) } Handle::Disk(number) => { let disk = self.disks.get_mut(&number).ok_or(Error::new(EBADF))?; @@ -465,7 +473,7 @@ impl SchemeBlock for DiskScheme { stat.st_blocks = disk.disk().size() / u64::from(disk.block_size()); stat.st_blksize = disk.block_size(); stat.st_size = disk.size(); - Ok(Some(0)) + Ok(()) } Handle::Partition(disk_num, part_num) => { let disk = self.disks.get_mut(&disk_num).ok_or(Error::new(EBADF))?; @@ -480,18 +488,19 @@ impl SchemeBlock for DiskScheme { stat.st_size = part.size * u64::from(disk.block_size()); stat.st_blocks = part.size; stat.st_blksize = disk.block_size(); - Ok(Some(0)) + Ok(()) } } } - fn fpath(&mut self, id: usize, buf: &mut [u8]) -> Result> { + async fn fpath(&mut self, id: usize, buf: &mut [u8], _ctx: &CallerCtx) -> Result { let handle = self.handles.get(&id).ok_or(Error::new(EBADF))?; let mut i = 0; let scheme_name = self.scheme_name.as_bytes(); let mut j = 0; + // TODO: copy_from_slice while i < buf.len() && j < scheme_name.len() { buf[i] = scheme_name[j]; i += 1; @@ -527,16 +536,17 @@ impl SchemeBlock for DiskScheme { } } - Ok(Some(i)) + Ok(i) } - fn read( + async fn read( &mut self, id: usize, buf: &mut [u8], offset: u64, _fcntl_flags: u32, - ) -> Result> { + _ctx: &CallerCtx, + ) -> Result { match *self.handles.get_mut(&id).ok_or(Error::new(EBADF))? { Handle::List(ref handle) => { let src = usize::try_from(offset) @@ -545,70 +555,69 @@ impl SchemeBlock for DiskScheme { .unwrap_or(&[]); let count = core::cmp::min(src.len(), buf.len()); buf[..count].copy_from_slice(&src[..count]); - Ok(Some(count)) + Ok(count) } Handle::Disk(number) => { let disk = self.disks.get_mut(&number).ok_or(Error::new(EBADF))?; let block = offset / u64::from(disk.block_size()); - disk.read(None, block, buf) + disk.read(None, block, buf).await } Handle::Partition(disk_num, part_num) => { let disk = self.disks.get_mut(&disk_num).ok_or(Error::new(EBADF))?; let block = offset / u64::from(disk.block_size()); - disk.read(Some(part_num as usize), block, buf) + disk.read(Some(part_num as usize), block, buf).await } } } - fn write( + async fn write( &mut self, id: usize, buf: &[u8], offset: u64, _fcntl_flags: u32, - ) -> Result> { + _ctx: &CallerCtx, + ) -> Result { match *self.handles.get_mut(&id).ok_or(Error::new(EBADF))? { Handle::List(_) => Err(Error::new(EBADF)), Handle::Disk(number) => { let disk = self.disks.get_mut(&number).ok_or(Error::new(EBADF))?; let block = offset / u64::from(disk.block_size()); - disk.write(None, block, buf) + disk.write(None, block, buf).await } Handle::Partition(disk_num, part_num) => { let disk = self.disks.get_mut(&disk_num).ok_or(Error::new(EBADF))?; let block = offset / u64::from(disk.block_size()); - disk.write(Some(part_num as usize), block, buf) + disk.write(Some(part_num as usize), block, buf).await } } } - fn fsize(&mut self, id: usize) -> Result> { - Ok(Some( - match *self.handles.get_mut(&id).ok_or(Error::new(EBADF))? { - Handle::List(ref handle) => handle.len() as u64, - Handle::Disk(number) => { - let disk = self.disks.get_mut(&number).ok_or(Error::new(EBADF))?; - disk.size() - } - Handle::Partition(disk_num, part_num) => { - let disk = self.disks.get_mut(&disk_num).ok_or(Error::new(EBADF))?; - let part = disk - .pt - .as_ref() - .ok_or(Error::new(EBADF))? - .partitions - .get(part_num as usize) - .ok_or(Error::new(EBADF))?; + async fn fsize(&mut self, id: usize, _ctx: &CallerCtx) -> Result { + Ok(match *self.handles.get_mut(&id).ok_or(Error::new(EBADF))? { + Handle::List(ref handle) => handle.len() as u64, + Handle::Disk(number) => { + let disk = self.disks.get_mut(&number).ok_or(Error::new(EBADF))?; + disk.size() + } + Handle::Partition(disk_num, part_num) => { + let disk = self.disks.get_mut(&disk_num).ok_or(Error::new(EBADF))?; + let part = disk + .pt + .as_ref() + .ok_or(Error::new(EBADF))? + .partitions + .get(part_num as usize) + .ok_or(Error::new(EBADF))?; part.size * u64::from(disk.block_size()) - } - }, - )) + } + }) } } -impl DiskScheme { - fn on_close(&mut self, id: usize) { - self.handles.remove(&id); +impl DiskScheme { + pub fn on_close(&mut self, id: usize) { + let _ = self.handles.remove(&id); } }