runtime: I/O API + simulator

crates/runtime-core/Cargo.toml

@@ -11,8 +11,10 @@ workspace = true
 
 [features]
 default = []
-sim = ["dep:async-task", "dep:spin"]
+sim = ["dep:async-task", "dep:futures-channel", "dep:spin"]
 
 [dependencies]
 async-task = { version = "4.4", default-features = false, optional = true }
+futures-channel = { version = "0.3", default-features = false, features = ["alloc"], optional = true }
 spin = { version = "0.9", default-features = false, features = ["mutex", "spin_mutex"], optional = true }
+zerocopy = "0.8"

crates/runtime-core/src/io/mod.rs

@@ -0,0 +1,148 @@
+use zerocopy::{FromBytes, Immutable, IntoBytes, KnownLayout};
+
+/// Size in bytes of a disk sector.
+pub const SECTOR_SIZE: usize = 4096;
+
+/// Types that can be safely converted to and from sector-aligned byte slices.
+pub trait AlignedBytes: Sized {
+    /// Assert that the type' size is a multiple of [SECTOR_SIZE] and has the
+    /// right aligment.
+    ///
+    /// NOTE: Associated constants are evaluated lazily -- add a free
+    ///
+    /// `const _: () = <T as AlignedBytes>::ASSERT_VALID_LAYOUT;`
+    ///
+    /// for each `T` that is supposed to be used as an `AlignedBytes`.
+    const ASSERT_VALID_LAYOUT: () = {
+        assert!(align_of::<Self>() == SECTOR_SIZE);
+        assert!(size_of::<Self>().is_multiple_of(SECTOR_SIZE));
+    };
+
+    /// Reinterpret `self` as a byte slice.
+    ///
+    /// The returned slice will be of length `size_of::<Self>()`.
+    fn as_bytes(&self) -> &[u8];
+
+    /// Reinterpret `self` as a mutable byte slice.
+    ///
+    /// The returned slice will be of length `size_of::<Self>()`.
+    fn as_bytes_mut(&mut self) -> &mut [u8];
+
+    /// Reinterpret a byte slice as `Self`.
+    ///
+    /// The slice must be of length `size_of::<Self>()`.
+    ///
+    /// NOTE: Any slice of the right size, but consisting of only `0` (zero)
+    /// bytes can be converted to `Self`. It is the caller's responsibility to
+    /// validate the returned type as per the application's invariants.
+    ///
+    /// # Panics
+    ///
+    /// Panics if `b.len() != size_of::<Self>()`.
+    fn from_bytes(b: &[u8]) -> Self;
+}
+
+impl<T: FromBytes + IntoBytes + KnownLayout + Immutable> AlignedBytes for T {
+    fn as_bytes(&self) -> &[u8] {
+        <T as IntoBytes>::as_bytes(self)
+    }
+
+    fn as_bytes_mut(&mut self) -> &mut [u8] {
+        <T as IntoBytes>::as_mut_bytes(self)
+    }
+
+    fn from_bytes(b: &[u8]) -> Self {
+        Self::read_from_bytes(b).unwrap()
+    }
+}
+
+/// An error `E`, along with auxiliary data `T`.
+///
+/// `T` is usually a buffer of type [AlignedBytes], whose ownership is
+/// transferred back to the caller when an error occurs.
+///
+/// As this type signifies an error condition, the contents of `T` are
+/// unspecified.
+#[derive(Debug)]
+pub struct ErrorWith<E, T> {
+    pub error: E,
+    pub with: T,
+}
+
+/// The canonical, low-level I/O API.
+///
+/// Currently only supports file I/O, but eventually all I/O performed by
+/// SpacetimeDB should go through this trait.
+///
+/// Intended to support implementations based on `io-uring`, which means that
+/// buffer ownership is transferred to the I/O engine while reading or writing.
+///
+/// Implementations should be `!Send`, i.e. all I/O happens on a single thread.
+///
+/// File operations should never be mutually exclusive, and therefore expose a
+/// `pwrite`/`pread`-style API. It is assumed that direct I/O (`O_DIRECT`) is
+/// used, i.e. the kernel page cache is bypassed. The [AlignedBytes] type
+/// ensures that the alignment requirements for direct I/O are met.
+pub trait SpacetimeIO {
+    /// An open file handle.
+    ///
+    /// Like [std::fs::File], the file shall be closed when the last reference
+    /// to the handle is dropped.
+    ///
+    /// Unlike [std::fs::File], the file handle must be clone-able.
+    type Fd: Clone;
+    /// The error returned by methods of this trait.
+    ///
+    /// This should always be instantiated to [std::io::Error]. However, pending
+    /// [alloc_io], this type is not in `core`, which would prevent this crate
+    /// from being `no_std`.
+    ///
+    /// [alloc_io]: https://github.com/rust-lang/rust/issues/154046
+    type Error;
+
+    /// Open the file at `path`.
+    fn open_file(&self, path: &str) -> impl Future<Output = Result<Self::Fd, Self::Error>>;
+
+    /// Create the file at `path` and allocate `len` bytes.
+    ///
+    /// Returns an error if the file already exists.
+    fn create_file(&self, path: &str, len: u64) -> impl Future<Output = Result<Self::Fd, Self::Error>>;
+
+    /// Write `buf` to `fd` at `offset`.
+    ///
+    /// `offset` must be a multiple of [SECTOR_SIZE].
+    ///
+    /// Behaves like `FileExt::write_all_at`, i.e. tries to write all bytes in
+    /// `buf`, potentially retrying on errors of kind interrupted, and returns
+    /// an error if that fails.
+    fn write_all_at<B: AlignedBytes + Send + 'static>(
+        &self,
+        fd: Self::Fd,
+        buf: B,
+        offset: u64,
+    ) -> impl Future<Output = Result<B, ErrorWith<Self::Error, B>>>;
+
+    /// Read `size_of::<B>()` bytes from `fd` at `offset` and interpret them at
+    /// type `B`.
+    ///
+    /// `offset` must be a multiple of [SECTOR_SIZE].
+    ///
+    /// Behaves like `FileExt::read_exact_at`, i.e. attempts to read
+    /// `size_of::<B>()` bytes, potentially retrying on errors of kind
+    /// interrupted, and returns an error if less than the required bytes could
+    /// be read.
+    fn read_exact_at<B: AlignedBytes + Send + 'static>(
+        &self,
+        fd: Self::Fd,
+        buf: B,
+        offset: u64,
+    ) -> impl Future<Output = Result<B, ErrorWith<Self::Error, B>>>;
+
+    /// Call `fsync(2)` on `fd`.
+    fn fsync(&self, fd: Self::Fd) -> impl Future<Output = Result<(), Self::Error>>;
+    /// Call `fdatasync(2)` on `fd`.
+    fn fdatasync(&self, fd: Self::Fd) -> impl Future<Output = Result<(), Self::Error>>;
+
+    /// Allocate `additional` bytes for the file `fd`.
+    fn reserve(&self, fd: Self::Fd, additional: u64) -> impl Future<Output = Result<(), Self::Error>>;
+}

crates/runtime-core/src/lib.rs

@@ -7,3 +7,5 @@ extern crate std;
 
 #[cfg(feature = "sim")]
 pub mod sim;
+
+pub mod io;

crates/runtime-core/src/sim/executor/io.rs

@@ -0,0 +1,101 @@
+use crate::sim::{io::SimulatorIO, Rng};
+
+#[derive(Clone, Copy, Debug, PartialEq)]
+pub struct Config {
+    /// The max number of submissions to run per [Driver::tick].
+    pub max_submissions_per_tick: usize,
+    /// The max number of completions to finish per [Driver::tick].
+    pub max_completions_per_tick: usize,
+    /// Submission reordering probability.
+    ///
+    /// Describes the probability by which to select the next submission queue
+    /// entry randomly, as opposed to the oldest entry in the queue.
+    pub prob_reorder_submissions: f64,
+    /// Completion reordering probability.
+    ///
+    /// Describes the probability by which to select the next completion queue
+    /// entry randomly, as opposed to the oldest entry in the queue.
+    pub prob_reorder_completions: f64,
+    /// Probability by which to skip one submission queue entry.
+    ///
+    /// If skipped, the entry still counts towards `max_submissions_per_tick`.
+    pub prob_skip: f64,
+    /// Probability by which to cancel a submission queue entry.
+    ///
+    /// [crate::sim::io::op::Submission::cancel()] is called on the entry, which
+    /// may generate a completion.
+    pub prob_cancel: f64,
+}
+
+impl Default for Config {
+    fn default() -> Self {
+        Self {
+            max_submissions_per_tick: 1,
+            max_completions_per_tick: 1,
+            prob_reorder_submissions: 0.0,
+            prob_reorder_completions: 0.0,
+            prob_skip: 0.0,
+            prob_cancel: 0.0,
+        }
+    }
+}
+
+pub struct Driver {
+    io: SimulatorIO,
+    config: Config,
+}
+
+impl Driver {
+    pub fn new(config: Config) -> Self {
+        Self {
+            io: <_>::default(),
+            config,
+        }
+    }
+
+    /// Advance the I/O simulator according the [Config].
+    ///
+    /// Returns `true` if progress has been made, or there are pending entries
+    /// in either the submission or completion queue.
+    pub fn tick(&self, rng: &Rng) -> bool {
+        let mut progress = false;
+        for _ in 0..self.config.max_submissions_per_tick {
+            if !rng.buggify_with_prob(self.config.prob_skip) {
+                let sqe = if rng.buggify_with_prob(self.config.prob_reorder_submissions) {
+                    self.io.random_submission(rng)
+                } else {
+                    self.io.next_submission()
+                };
+
+                if let Some(sqe) = sqe {
+                    if rng.buggify_with_prob(self.config.prob_cancel) {
+                        sqe.cancel();
+                    } else {
+                        self.io.execute(sqe);
+                    }
+                    progress = true;
+                }
+            }
+        }
+
+        for _ in 0..self.config.max_completions_per_tick {
+            let cqe = if rng.buggify_with_prob(self.config.prob_reorder_completions) {
+                self.io.random_completion(rng)
+            } else {
+                self.io.next_completion()
+            };
+
+            if let Some(cqe) = cqe {
+                cqe.complete();
+                progress = true
+            }
+        }
+
+        progress |= self.io.pending();
+        progress
+    }
+
+    pub fn io(&self) -> &SimulatorIO {
+        &self.io
+    }
+}

crates/runtime-core/src/sim/executor/mod.rs

@@ -10,22 +10,38 @@ use core::{
 
 use spin::Mutex;
 
+use crate::sim::io::SimulatorIO;
+
 use super::{time::TimeHandle, Rng};
 
+mod io;
+
 mod task;
 use task::Abortable;
 pub use task::{AbortHandle, JoinError, JoinHandle};
 
 type Runnable = async_task::Runnable<NodeId>;
 
-#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+#[derive(Clone, Copy, Debug, PartialEq)]
 pub struct RuntimeConfig {
     pub seed: u64,
+    pub io: Option<io::Config>,
 }
 
 impl RuntimeConfig {
     pub const fn new(seed: u64) -> Self {
-        Self { seed }
+        Self { seed, io: None }
+    }
+
+    pub fn enable_io(self) -> Self {
+        Self {
+            io: Some(self.io.unwrap_or_default()),
+            ..self
+        }
+    }
+
+    pub fn with_io_config(self, io: Option<io::Config>) -> Self {
+        Self { io, ..self }
     }
 }
 
@@ -145,6 +161,12 @@ impl Runtime {
         }
     }
 
+    // TODO: This is a stopgap to allow submission of I/O tasks. We probably
+    // want the user-facing API to hide this.
+    pub fn io(&self) -> Option<&SimulatorIO> {
+        self.executor.io.as_ref().map(|driver| driver.io())
+    }
+
     /// Drive a top-level future to completion on the simulation executor.
     ///
     /// While the future runs, spawned tasks share the same deterministic
@@ -360,6 +382,7 @@ struct Executor {
     next_node: AtomicU64,
     rng: Rng,
     time: TimeHandle,
+    io: Option<io::Driver>,
 }
 
 impl Executor {
@@ -375,6 +398,7 @@ impl Executor {
             next_node: AtomicU64::new(1),
             rng: Rng::new(config.seed),
             time: TimeHandle::new(),
+            io: config.io.map(io::Driver::new),
         }
     }
 
@@ -491,6 +515,7 @@ impl Executor {
 
         loop {
             self.run_all_ready();
+            let pending_io = self.drive_io();
             if task.is_finished() {
                 let waker = Waker::noop();
                 return match Pin::new(&mut task).poll(&mut Context::from_waker(waker)) {
@@ -499,7 +524,7 @@ impl Executor {
                 };
             }
 
-            if self.time.wake_next_timer() {
+            if self.time.wake_next_timer() || pending_io {
                 continue;
             }
 
@@ -527,6 +552,14 @@ impl Executor {
         }
     }
 
+    fn drive_io(&self) -> bool {
+        if let Some(io) = &self.io {
+            io.tick(&self.rng)
+        } else {
+            false
+        }
+    }
+
     /// Look up the record for a node, panicking if the node is unknown.
     fn node_record(&self, node: NodeId) -> Arc<NodeRecord> {
         self.nodes