runtime: I/O API + simulator

crates/runtime-core/Cargo.toml

@@ -11,8 +11,16 @@ workspace = true
 
 [features]
 default = []
-sim = ["dep:async-task", "dep:spin"]
+alloc = []
+sim = ["alloc", "dep:async-task", "dep:futures-channel", "dep:slab", "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 }
+slab = { version = "0.4", default-features = false, optional = true }
 spin = { version = "0.9", default-features = false, features = ["mutex", "spin_mutex"], optional = true }
+thiserror = { version = "2.0", default-features = false }
+zerocopy = "0.8"
+
+[dev-dependencies]
+tokio.workspace = true

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

@@ -0,0 +1,186 @@
+use zerocopy::{FromBytes, Immutable, IntoBytes, KnownLayout};
+
+use crate::io::SECTOR_SIZE;
+
+/// 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 alignment.
+    ///
+    /// The type must also not rely on drop glue, i.e. `!core::mem::needs_drop()`.
+    ///
+    /// 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));
+        assert!(!core::mem::needs_drop::<Self>());
+    };
+
+    /// 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()
+    }
+}
+
+#[cfg(feature = "alloc")]
+mod boxed {
+    use alloc::boxed::Box;
+    use core::{alloc::Layout, any::TypeId, ptr::NonNull};
+
+    use crate::io::AlignedBytes;
+
+    /// A type-erased [AlignedBytes] heap allocation.
+    pub struct ErasedBox {
+        ptr: NonNull<u8>,
+        len: usize,
+        layout: Layout,
+        ty: TypeId,
+    }
+
+    impl ErasedBox {
+        /// Create an [ErasedBox] from `B` by allocating a new [Box].
+        pub fn from_aligned<B: AlignedBytes + 'static>(b: B) -> Self {
+            Self::from_aligned_box(Box::new(b))
+        }
+
+        /// Create an [ErasedBox] from an already-boxed `B`.
+        pub fn from_aligned_box<B: AlignedBytes + 'static>(b: Box<B>) -> Self {
+            let () = B::ASSERT_VALID_LAYOUT;
+
+            let ptr = Box::into_raw(b);
+            Self {
+                ptr: NonNull::new(ptr.cast()).unwrap(),
+                len: size_of::<B>(),
+                layout: Layout::from_size_align(size_of::<B>(), align_of::<B>()).unwrap(),
+                ty: TypeId::of::<B>(),
+            }
+        }
+
+        /// Reify `B` via casting.
+        pub fn into_aligned<B: AlignedBytes + 'static>(self) -> B {
+            *Self::into_aligned_box(self)
+        }
+
+        /// Reify `B` via casting, without unboxing.
+        pub fn into_aligned_box<B: AlignedBytes + 'static>(self) -> Box<B> {
+            assert_eq!(self.len, size_of::<B>());
+            assert_eq!(self.ty, TypeId::of::<B>());
+
+            let boxed = unsafe { Box::from_raw(self.ptr.as_ptr().cast::<B>()) };
+            // Prevent drop, which would deallocate.
+            core::mem::forget(self);
+
+            boxed
+        }
+
+        pub fn as_ptr(&self) -> ErasedBoxPtr {
+            ErasedBoxPtr {
+                ptr: self.ptr.as_ptr(),
+                len: self.len,
+            }
+        }
+
+        pub fn len(&self) -> usize {
+            self.len
+        }
+
+        pub fn is_empty(&self) -> bool {
+            self.len == 0
+        }
+    }
+
+    impl Drop for ErasedBox {
+        fn drop(&mut self) {
+            unsafe { alloc::alloc::dealloc(self.ptr.as_ptr(), self.layout) }
+        }
+    }
+
+    pub struct ErasedBoxPtr {
+        ptr: *mut u8,
+        len: usize,
+    }
+
+    impl ErasedBoxPtr {
+        pub fn as_bytes(&mut self) -> &[u8] {
+            unsafe { core::slice::from_raw_parts(self.ptr, self.len) }
+        }
+
+        pub fn as_bytes_mut(&mut self) -> &mut [u8] {
+            unsafe { core::slice::from_raw_parts_mut(self.ptr, self.len) }
+        }
+    }
+
+    #[cfg(test)]
+    mod tests {
+        use super::*;
+
+        #[repr(C, align(4096))]
+        #[derive(Clone, Copy, Debug, PartialEq, Eq)]
+        struct Trivial([u8; 4096]);
+
+        impl AlignedBytes for Trivial {
+            fn as_bytes(&self) -> &[u8] {
+                &self.0
+            }
+
+            fn as_bytes_mut(&mut self) -> &mut [u8] {
+                &mut self.0
+            }
+
+            fn from_bytes(b: &[u8]) -> Self {
+                assert_eq!(b.len(), size_of::<Self>());
+                let mut a = [0; 4096];
+                a.copy_from_slice(b);
+                Self(a)
+            }
+        }
+
+        #[test]
+        fn roundtrip_preserves_value() {
+            let t = Trivial([32; 4096]);
+
+            let erased = ErasedBox::from_aligned(t);
+            let reified = erased.into_aligned::<Trivial>();
+
+            assert_eq!(reified, t);
+        }
+    }
+}
+#[cfg(feature = "alloc")]
+pub use boxed::{ErasedBox, ErasedBoxPtr};

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

@@ -0,0 +1,43 @@
+/// 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.
+///
+/// [AlignedBytes]: crate::io::buf::AlignedBytes
+#[derive(Debug)]
+pub struct ErrorWith<E, T> {
+    pub error: E,
+    pub with: T,
+}
+
+impl<E, T> ErrorWith<E, T> {
+    /// Map a type-changing function over `self.error`.
+    pub fn map_err<F>(self, f: impl FnOnce(E) -> F) -> ErrorWith<F, T> {
+        ErrorWith {
+            error: f(self.error),
+            with: self.with,
+        }
+    }
+
+    /// Map a type-changing function over `self.with`.
+    pub fn map_with<U>(self, f: impl FnOnce(T) -> U) -> ErrorWith<E, U> {
+        ErrorWith {
+            error: self.error,
+            with: f(self.with),
+        }
+    }
+
+    /// Extract `self.error`, discarding `self.with`.
+    pub fn into_err(self) -> E {
+        self.error
+    }
+
+    /// Convert from `&ErrorWith<E, T>` to `ErrorWith<&E, &T>`.
+    pub fn as_ref(&self) -> ErrorWith<&E, &T> {
+        let Self { ref error, ref with } = *self;
+        ErrorWith { error, with }
+    }
+}

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

@@ -0,0 +1,112 @@
+mod buf;
+pub use buf::AlignedBytes;
+#[cfg(feature = "alloc")]
+pub use buf::{ErasedBox, ErasedBoxPtr};
+
+mod error;
+pub use error::ErrorWith;
+
+/// Size in bytes of a disk sector.
+pub const SECTOR_SIZE: usize = 4096;
+
+/// Subset of the `statx` metadata.
+#[derive(Debug)]
+#[non_exhaustive]
+pub struct Statx {
+    pub size: u64,
+}
+
+impl Statx {
+    pub fn from_size(size: u64) -> Self {
+        Self { size }
+    }
+}
+
+/// 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: core::error::Error;
+    /// The completion [Future] of all methods in this trait.
+    type Completion<T>: Future<Output = T> + Unpin;
+
+    /// Open the file at `path`.
+    fn open_file(&self, path: &str) -> Self::Completion<Result<Self::Fd, Self::Error>>;
+
+    /// Create the file at `path`.
+    ///
+    /// Returns an error if the file already exists.
+    fn create_file(&self, path: &str) -> Self::Completion<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,
+    ) -> Self::Completion<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,
+    ) -> Self::Completion<Result<B, ErrorWith<Self::Error, B>>>;
+
+    /// Call `fsync(2)` on `fd`.
+    fn fsync(&self, fd: Self::Fd) -> Self::Completion<Result<(), Self::Error>>;
+    /// Call `fdatasync(2)` on `fd`.
+    fn fdatasync(&self, fd: Self::Fd) -> Self::Completion<Result<(), Self::Error>>;
+
+    /// Allocate `total` bytes for the file `fd`.
+    ///
+    /// Implementations must ensure that attempts to shrink the file result in
+    /// an error. The operation should succeed if the file's size is already
+    /// `total`.
+    fn reserve(&self, fd: Self::Fd, total: u64) -> Self::Completion<Result<(), Self::Error>>;
+
+    /// Determine the length of the file `fd`.
+    ///
+    /// This should not depend on `fsync`, i.e. `statx`. See `std::io::Seek::stream_len`.
+    fn statx(&self, fd: Self::Fd) -> Self::Completion<Result<Statx, Self::Error>>;
+}

crates/runtime-core/src/lib.rs

@@ -1,9 +1,11 @@
 #![no_std]
 
-#[cfg(feature = "sim")]
+#[cfg(any(feature = "sim", feature = "alloc"))]
 extern crate alloc;
 #[cfg(test)]
 extern crate std;
 
 #[cfg(feature = "sim")]
 pub mod sim;
+
+pub mod io;