Add a hw-based async framework for block drivers.

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",

Cargo.toml

@@ -1,7 +1,9 @@
 [workspace]
 members = [
-    "acpid",
     "common",
+    "executor",
+
+    "acpid",
     "hwd",
     "pcid",
     "pcid-spawner",

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"

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<LocalExecutor<Self>>;
+    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<Hw: Hardware> {
+    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<HashMap<Hw::SqId, VecDeque<FutIdx>>>,
+    awaiting_completion:
+        RefCell<HashMap<Hw::CqId, HashMap<Hw::CmdId, (FutIdx, NonNull<Option<Hw::Cqe>>)>>>,
+
+    external_event: RefCell<HashMap<EventUserData, (FutIdx, NonNull<EventFlags>)>>,
+    next_user_data: Cell<usize>,
+
+    ready_futures: RefCell<VecDeque<FutIdx>>,
+    futures: RefCell<Slab<Pin<Box<dyn Future<Output = ()> + 'static>>>>,
+    is_polling: Cell<bool>,
+}
+
+impl<Hw: Hardware> LocalExecutor<Hw> {
+    pub fn register_external_event(
+        &self,
+        fd: usize,
+        flags: event::EventFlags,
+    ) -> ExternalEventSource<Hw> {
+        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<Self> {
+        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::<Hw>(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<Output = ()> + '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<Output = O> + 'a) -> O {
+        let retval = Rc::new(RefCell::new(None));
+
+        let retval2 = Rc::clone(&retval);
+        let idx = self.futures.borrow_mut().insert({
+            let t1: Pin<Box<dyn Future<Output = ()> + '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<Box<dyn Future<Output = ()> + '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::<usize>()];
+            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::<Hw> {
+            state: State::<Hw>::Submitting { sq_id, cmd },
+            comp: None,
+            _not_send: PhantomData,
+        }
+        .await
+    }
+}
+
+struct CqeFuture<Hw: Hardware> {
+    pub state: State<Hw>,
+    pub comp: Option<Hw::Cqe>,
+    pub _not_send: PhantomData<*const ()>,
+}
+enum State<Hw: Hardware> {
+    Submitting { sq_id: Hw::SqId, cmd: Hw::Sqe },
+    Completing { cq_id: Hw::CqId, cmd_id: Hw::CmdId },
+}
+
+fn current_executor_and_idx<Hw: Hardware>(
+    cx: &mut task::Context<'_>,
+) -> (Rc<LocalExecutor<Hw>>, 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<Hw: Hardware> Future for CqeFuture<Hw> {
+    type Output = Hw::Cqe;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> task::Poll<Self::Output> {
+        let this = unsafe { self.get_unchecked_mut() };
+
+        let (executor, idx) = current_executor_and_idx::<Hw>(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<Hw: Hardware>(idx: *const ()) -> task::RawWaker {
+    task::RawWaker::new(idx, Hw::vtable())
+}
+unsafe fn vt_drop(_idx: *const ()) {}
+unsafe fn vt_wake<Hw: Hardware>(idx: *const ()) {
+    Hw::current()
+        .ready_futures
+        .borrow_mut()
+        .push_back(idx as FutIdx);
+}
+
+fn waker<Hw: Hardware>(idx: FutIdx) -> task::Waker {
+    unsafe { task::Waker::from_raw(task::RawWaker::new(idx as *const (), Hw::vtable())) }
+}
+pub const fn vtable<Hw: Hardware>() -> task::RawWakerVTable {
+    task::RawWakerVTable::new(vt_clone::<Hw>, vt_wake::<Hw>, vt_wake::<Hw>, vt_drop)
+}
+
+pub struct ExternalEventSource<Hw: Hardware> {
+    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<Hw: Hardware> ExternalEventSource<Hw> {
+    fn poll_next(self: Pin<&mut Self>, cx: &mut task::Context) -> task::Poll<Option<Event>> {
+        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::<Hw>(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<Event> {
+        core::future::poll_fn(|cx| self.as_mut().poll_next(cx)).await
+    }
+}
+pub fn init_raw<Hw: Hardware>(
+    global_ctxt: Hw::GlobalCtxt,
+    vector: Hw::Iv,
+    intx: bool,
+    irq_handle: File,
+) -> LocalExecutor<Hw> {
+    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),
+    }
+}

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"

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<Option<usize>>;
-    fn write(&mut self, block: u64, buffer: &[u8]) -> syscall::Result<Option<usize>>;
+    async fn read(&mut self, block: u64, buffer: &mut [u8]) -> syscall::Result<usize>;
+    async fn write(&mut self, block: u64, buffer: &[u8]) -> syscall::Result<usize>;
 }
 
 impl<T: Disk + ?Sized> Disk for Box<T> {
@@ -84,12 +85,12 @@ impl<T: Disk + ?Sized> Disk for Box<T> {
         (**self).size()
     }
 
-    fn read(&mut self, block: u64, buffer: &mut [u8]) -> syscall::Result<Option<usize>> {
-        (**self).read(block, buffer)
+    async fn read(&mut self, block: u64, buffer: &mut [u8]) -> syscall::Result<usize> {
+        (**self).read(block, buffer).await
     }
 
-    fn write(&mut self, block: u64, buffer: &[u8]) -> syscall::Result<Option<usize>> {
-        (**self).write(block, buffer)
+    async fn write(&mut self, block: u64, buffer: &[u8]) -> syscall::Result<usize> {
+        (**self).write(block, buffer).await
     }
 }
 
@@ -99,18 +100,19 @@ pub struct DiskWrapper<T> {
 }
 
 impl<T: Disk> DiskWrapper<T> {
-    pub fn pt(disk: &mut T) -> Option<PartitionTable> {
+    pub fn pt(disk: &mut T, executor: &impl ExecutorTrait) -> Option<PartitionTable> {
         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<u64> {
                 let size = i64::try_from(self.disk.size()).or(Err(io::Error::new(
                     io::ErrorKind::Other,
@@ -129,7 +131,7 @@ impl<T: Disk> DiskWrapper<T> {
             }
         }
         // 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<usize> {
                 let blksize = self.disk.block_size();
                 let size_in_blocks = self.disk.size() / u64::from(blksize);
@@ -141,17 +143,9 @@ impl<T: Disk> DiskWrapper<T> {
                         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<T: Disk> DiskWrapper<T> {
             }
         }
 
-        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<T: Disk> DiskWrapper<T> {
         self.disk.size()
     }
 
-    pub fn read(
+    pub async fn read(
         &mut self,
         part_num: Option<usize>,
         block: u64,
         buf: &mut [u8],
-    ) -> syscall::Result<Option<usize>> {
+    ) -> syscall::Result<usize> {
         if buf.len() as u64 % u64::from(self.disk.block_size()) != 0 {
             return Err(Error::new(EINVAL));
         }
@@ -214,18 +215,18 @@ impl<T: Disk> DiskWrapper<T> {
 
             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<usize>,
         block: u64,
         buf: &[u8],
-    ) -> syscall::Result<Option<usize>> {
+    ) -> syscall::Result<usize> {
         if buf.len() as u64 % u64::from(self.disk.block_size()) != 0 {
             return Err(Error::new(EINVAL));
         }
@@ -245,9 +246,9 @@ impl<T: Disk> DiskWrapper<T> {
 
             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<T> {
     disks: BTreeMap<u32, DiskWrapper<T>>,
     handles: BTreeMap<usize, Handle>,
     next_id: usize,
-    blocked: Vec<CallRequest>,
+}
+
+pub trait ExecutorTrait {
+    fn block_on<'a, O: 'a>(&self, fut: impl IntoFuture<Output = O> + 'a) -> O;
+}
+impl<Hw: executor::Hardware> ExecutorTrait for LocalExecutor<Hw> {
+    fn block_on<'a, O: 'a>(&self, fut: impl IntoFuture<Output = O> + 'a) -> O {
+        LocalExecutor::block_on(self, fut)
+    }
 }
 
 impl<T: Disk> DiskScheme<T> {
-    pub fn new(scheme_name: String, disks: BTreeMap<u32, T>) -> Self {
+    pub fn new(
+        scheme_name: String,
+        disks: BTreeMap<u32, T>,
+        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<T: Disk> DiskScheme<T> {
             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<T: Disk> DiskScheme<T> {
 
     /// 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<T: Disk> DiskScheme<T> {
     }
 }
 
-impl<T: Disk> SchemeBlock for DiskScheme<T> {
-    fn xopen(
-        &mut self,
-        path_str: &str,
-        flags: usize,
-        ctx: &CallerCtx,
-    ) -> Result<Option<OpenResult>> {
+impl<T: Disk> SchemeAsync for DiskScheme<T> {
+    async fn open(&mut self, path_str: &str, flags: usize, ctx: &CallerCtx) -> Result<OpenResult> {
         if ctx.uid != 0 {
             return Err(Error::new(EACCES));
         }
@@ -446,18 +445,27 @@ impl<T: Disk> SchemeBlock for DiskScheme<T> {
         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<DirentBuf<&'buf mut [u8]>> {
+        // TODO
+        Err(Error::new(EOPNOTSUPP))
     }
 
-    fn fstat(&mut self, id: usize, stat: &mut Stat) -> Result<Option<usize>> {
+    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<T: Disk> SchemeBlock for DiskScheme<T> {
                 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<T: Disk> SchemeBlock for DiskScheme<T> {
                 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<Option<usize>> {
+    async fn fpath(&mut self, id: usize, buf: &mut [u8], _ctx: &CallerCtx) -> Result<usize> {
         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<T: Disk> SchemeBlock for DiskScheme<T> {
             }
         }
 
-        Ok(Some(i))
+        Ok(i)
     }
 
-    fn read(
+    async fn read(
         &mut self,
         id: usize,
         buf: &mut [u8],
         offset: u64,
         _fcntl_flags: u32,
-    ) -> Result<Option<usize>> {
+        _ctx: &CallerCtx,
+    ) -> Result<usize> {
         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<T: Disk> SchemeBlock for DiskScheme<T> {
                     .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<Option<usize>> {
+        _ctx: &CallerCtx,
+    ) -> Result<usize> {
         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<Option<u64>> {
-        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<u64> {
+        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<T: Disk> DiskScheme<T> {
-    fn on_close(&mut self, id: usize) {
-        self.handles.remove(&id);
+impl<D: Disk> DiskScheme<D> {
+    pub fn on_close(&mut self, id: usize) {
+        let _ = self.handles.remove(&id);
     }
 }