diff --git a/Cargo.lock b/Cargo.lock index 685bbac8eb9..d7f858ebf5e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8540,7 +8540,10 @@ dependencies = [ "futures", "libc", "spacetimedb-runtime-core", + "spacetimedb-runtime-io", + "static_assertions", "tokio", + "windows-sys 0.61.2", ] [[package]] @@ -8551,6 +8554,19 @@ dependencies = [ "spin", ] +[[package]] +name = "spacetimedb-runtime-io" +version = "2.10.1" +dependencies = [ + "slab", + "spacetimedb-runtime-core", + "spacetimedb-runtime-io", + "spin", + "thiserror 2.0.17", + "tokio", + "zerocopy", +] + [[package]] name = "spacetimedb-sats" version = "2.10.1" diff --git a/Cargo.toml b/Cargo.toml index 621a99774b5..0487e682378 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -29,6 +29,7 @@ members = [ "crates/query", "crates/runtime", "crates/runtime-core", + "crates/runtime-io", "crates/sats", "crates/schema", "crates/smoketests", @@ -176,6 +177,7 @@ spacetimedb-query = { path = "crates/query", version = "=2.10.1" } spacetimedb-query-builder = { path = "crates/query-builder", version = "=2.10.1" } spacetimedb-runtime = { path = "crates/runtime", version = "=2.10.1" } spacetimedb-runtime-core = { path = "crates/runtime-core", version = "=2.10.1" } +spacetimedb-runtime-io = { path = "crates/runtime-io", version = "=2.10.1" } spacetimedb-sats = { path = "crates/sats", version = "=2.10.1" } spacetimedb-schema = { path = "crates/schema", version = "=2.10.1" } spacetimedb-snapshot = { path = "crates/snapshot", version = "=2.10.1" } diff --git a/crates/runtime-io/Cargo.toml b/crates/runtime-io/Cargo.toml new file mode 100644 index 00000000000..316d30bbdd1 --- /dev/null +++ b/crates/runtime-io/Cargo.toml @@ -0,0 +1,22 @@ +[package] +name = "spacetimedb-runtime-io" +version.workspace = true +edition.workspace = true +rust-version.workspace = true + +[lints] +workspace = true + +[features] +sim = ["dep:slab", "dep:spin"] + +[dependencies] +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] +spacetimedb-runtime-core = { workspace = true, features = ["sim"] } +spacetimedb-runtime-io = { path = ".", features = ["sim"] } +tokio.workspace = true diff --git a/crates/runtime-io/src/buf.rs b/crates/runtime-io/src/buf.rs new file mode 100644 index 00000000000..636ff51b227 --- /dev/null +++ b/crates/runtime-io/src/buf.rs @@ -0,0 +1,163 @@ +use alloc::boxed::Box; +use core::{alloc::Layout, any::TypeId, ptr::NonNull}; +use zerocopy::{FromBytes, Immutable, IntoBytes, KnownLayout}; + +use crate::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 _: () = ::ASSERT_VALID_LAYOUT;` + /// + /// for each `T` that is supposed to be used as an `AlignedBytes`. + const ASSERT_VALID_LAYOUT: () = { + assert!(align_of::() == SECTOR_SIZE); + assert!(size_of::().is_multiple_of(SECTOR_SIZE)); + assert!(!core::mem::needs_drop::()); + }; + + /// Reinterpret `self` as a byte slice. + /// + /// The returned slice will be of length `size_of::()`. + fn as_bytes(&self) -> &[u8]; + + /// Reinterpret `self` as a mutable byte slice. + /// + /// The returned slice will be of length `size_of::()`. + fn as_bytes_mut(&mut self) -> &mut [u8]; + + /// Reinterpret a byte slice as `Self`. + /// + /// The slice must be of length `size_of::()`. + /// + /// 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::()`. + fn from_bytes(b: &[u8]) -> Self; +} + +impl AlignedBytes for T { + fn as_bytes(&self) -> &[u8] { + ::as_bytes(self) + } + + fn as_bytes_mut(&mut self) -> &mut [u8] { + ::as_mut_bytes(self) + } + + fn from_bytes(b: &[u8]) -> Self { + Self::read_from_bytes(b).unwrap() + } +} + +/// A type-erased [AlignedBytes] heap allocation. +#[derive(Debug)] +pub struct ErasedBox { + ptr: NonNull, + len: usize, + layout: Layout, + ty: TypeId, +} + +impl ErasedBox { + /// Create an [ErasedBox] from boxed [AlignedBytes].. + pub fn from_aligned(b: Box) -> Self { + let () = B::ASSERT_VALID_LAYOUT; + + let ptr = Box::into_raw(b); + Self { + ptr: NonNull::new(ptr.cast()).unwrap(), + len: size_of::(), + layout: Layout::from_size_align(size_of::(), align_of::()).unwrap(), + ty: TypeId::of::(), + } + } + + /// Reify `B` via casting. + pub fn into_aligned(self) -> Box { + assert_eq!(self.len, size_of::()); + assert_eq!(self.ty, TypeId::of::()); + + let boxed = unsafe { Box::from_raw(self.ptr.as_ptr().cast::()) }; + // Prevent drop, which would deallocate. + core::mem::forget(self); + + boxed + } + + pub fn as_mut_ptr(&self) -> *mut u8 { + self.ptr.as_ptr() + } + + pub fn len(&self) -> usize { + self.len + } + + pub fn is_empty(&self) -> bool { + self.len == 0 + } + + pub fn as_bytes(&self) -> &[u8] { + unsafe { core::slice::from_raw_parts(self.ptr.as_ptr(), self.len) } + } + + pub fn as_bytes_mut(&mut self) -> &mut [u8] { + unsafe { core::slice::from_raw_parts_mut(self.ptr.as_ptr(), self.len) } + } +} + +impl Drop for ErasedBox { + fn drop(&mut self) { + unsafe { alloc::alloc::dealloc(self.ptr.as_ptr(), self.layout) } + } +} + +// SAFETY: [ErasedBox] is `Send` because it can only be constructed from a +// `Send` [AlignedBuffer]. +unsafe impl Send for ErasedBox {} + +#[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::()); + 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(Box::new(t)); + let reified = erased.into_aligned::(); + + assert_eq!(reified, Box::new(t)); + } +} diff --git a/crates/runtime-io/src/error.rs b/crates/runtime-io/src/error.rs new file mode 100644 index 00000000000..0a06c1ada98 --- /dev/null +++ b/crates/runtime-io/src/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 { + pub error: E, + pub with: T, +} + +impl ErrorWith { + /// Map a type-changing function over `self.error`. + pub fn map_err(self, f: impl FnOnce(E) -> F) -> ErrorWith { + ErrorWith { + error: f(self.error), + with: self.with, + } + } + + /// Map a type-changing function over `self.with`. + pub fn map_with(self, f: impl FnOnce(T) -> U) -> ErrorWith { + 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` to `ErrorWith<&E, &T>`. + pub fn as_ref(&self) -> ErrorWith<&E, &T> { + let Self { ref error, ref with } = *self; + ErrorWith { error, with } + } +} diff --git a/crates/runtime-io/src/lib.rs b/crates/runtime-io/src/lib.rs new file mode 100644 index 00000000000..4b13070fdc0 --- /dev/null +++ b/crates/runtime-io/src/lib.rs @@ -0,0 +1,122 @@ +#![no_std] + +extern crate alloc; + +use alloc::boxed::Box; + +mod buf; +pub use buf::{AlignedBytes, ErasedBox}; + +mod error; +pub use error::ErrorWith; + +#[cfg(feature = "sim")] +pub mod sim; + +/// Size in bytes of a disk sector. +pub const SECTOR_SIZE: usize = 4096; +const SECTOR_SIZE64: u64 = SECTOR_SIZE as u64; + +/// 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 } + } +} + +pub type ReadWriteResult = Result, ErrorWith>>; + +/// 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: Future + Unpin; + + /// Open the file at `path`. + fn open_file(&self, path: Box) -> Self::Completion>; + + /// Create the file at `path`. + /// + /// Returns an error if the file already exists. + fn create_file(&self, path: Box) -> Self::Completion>; + + /// 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( + &self, + fd: Self::Fd, + buf: Box, + offset: u64, + ) -> Self::Completion>; + + /// Read `size_of::()` 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::()` 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( + &self, + fd: Self::Fd, + buf: Box, + offset: u64, + ) -> Self::Completion>; + + /// Call `fsync(2)` on `fd`. + fn fsync(&self, fd: Self::Fd) -> Self::Completion>; + /// Call `fdatasync(2)` on `fd`. + fn fdatasync(&self, fd: Self::Fd) -> Self::Completion>; + + /// 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>; + + /// 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>; +} diff --git a/crates/runtime-io/src/sim/completion.rs b/crates/runtime-io/src/sim/completion.rs new file mode 100644 index 00000000000..7064a5d3f26 --- /dev/null +++ b/crates/runtime-io/src/sim/completion.rs @@ -0,0 +1,514 @@ +use core::{ + convert::identity, + pin::Pin, + task::{Context, Poll, Waker}, +}; + +use alloc::{boxed::Box, sync::Arc}; +use slab::Slab; + +use crate::{ + sim::{fs, Error}, + AlignedBytes, ErasedBox, ErrorWith, Statx, +}; + +pub use slab::VacantEntry; + +pub(crate) enum CompletionState { + Pending(Option), + Ready(T), +} + +impl CompletionState { + fn complete(&mut self, v: T) -> Option { + match self { + Self::Pending(waker) => { + let waker = waker.take(); + *self = CompletionState::Ready(v); + waker + } + Self::Ready(_) => unreachable!("completion completed twice"), + } + } +} + +pub(super) struct PendingCompletions { + inner: Slab, +} + +impl PendingCompletions { + pub(super) fn with_capacity(cap: usize) -> Self { + Self { + inner: Slab::with_capacity(cap), + } + } + + pub(super) fn get_mut(&mut self, key: usize) -> Option<&mut CompletionHandle> { + self.inner.get_mut(key) + } + + pub(super) fn clear(&mut self) { + self.inner.clear(); + } + + pub(super) fn vacant_entry(&mut self) -> Option> { + if self.inner.capacity() == self.inner.len() { + None + } else { + Some(self.inner.vacant_entry()) + } + } + + fn remove(&mut self, key: usize) -> CompletionHandle { + self.inner.remove(key) + } + + fn try_remove(&mut self, key: usize) -> Option { + self.inner.try_remove(key) + } +} + +pub(crate) enum CompletionHandle { + Write(CompletionState>>), + Read(CompletionState>>), + Open(CompletionState>), + Create(CompletionState>), + Stat(CompletionState>), + Fallocate(CompletionState>), + Fsync(CompletionState>), + Fdatasync(CompletionState>), + #[allow(unused)] + Noop(CompletionState>), +} + +impl CompletionHandle { + fn write_state_mut(&mut self) -> &mut CompletionState>> { + match self { + Self::Write(state) => state, + _ => unreachable!(), + } + } + + fn into_write_state(self) -> CompletionState>> { + match self { + Self::Write(state) => state, + _ => unreachable!(), + } + } + + pub(crate) fn complete_write(&mut self, result: Result>) -> Option { + self.write_state_mut().complete(result) + } + + fn read_state_mut(&mut self) -> &mut CompletionState>> { + match self { + Self::Read(state) => state, + _ => unreachable!(), + } + } + + fn into_read_state(self) -> CompletionState>> { + match self { + Self::Read(state) => state, + _ => unreachable!(), + } + } + + pub(crate) fn complete_read(&mut self, result: Result>) -> Option { + self.read_state_mut().complete(result) + } + + fn open_state_mut(&mut self) -> &mut CompletionState> { + match self { + Self::Open(state) => state, + _ => unreachable!(), + } + } + + fn into_open_state(self) -> CompletionState> { + match self { + Self::Open(state) => state, + _ => unreachable!(), + } + } + + pub(crate) fn complete_open(&mut self, result: Result) -> Option { + self.open_state_mut().complete(result) + } + + fn create_state_mut(&mut self) -> &mut CompletionState> { + match self { + Self::Create(state) => state, + _ => unreachable!(), + } + } + + fn into_create_state(self) -> CompletionState> { + match self { + Self::Create(state) => state, + _ => unreachable!(), + } + } + + pub(crate) fn complete_create(&mut self, result: Result) -> Option { + self.create_state_mut().complete(result) + } + + fn stat_state_mut(&mut self) -> &mut CompletionState> { + match self { + Self::Stat(state) => state, + _ => unreachable!(), + } + } + + fn into_stat_state(self) -> CompletionState> { + match self { + Self::Stat(state) => state, + _ => unreachable!(), + } + } + + pub(crate) fn complete_stat(&mut self, result: Result) -> Option { + self.stat_state_mut().complete(result) + } + + fn fallocate_state_mut(&mut self) -> &mut CompletionState> { + match self { + Self::Fallocate(state) => state, + _ => unreachable!(), + } + } + + fn into_fallocate_state(self) -> CompletionState> { + match self { + Self::Fallocate(state) => state, + _ => unreachable!(), + } + } + + pub(crate) fn complete_fallocate(&mut self, result: Result<(), Error>) -> Option { + self.fallocate_state_mut().complete(result) + } + + fn fsync_state_mut(&mut self) -> &mut CompletionState> { + match self { + Self::Fsync(state) => state, + _ => unreachable!(), + } + } + + fn into_fsync_state(self) -> CompletionState> { + match self { + Self::Fsync(state) => state, + _ => unreachable!(), + } + } + + pub(crate) fn complete_fsync(&mut self, result: Result<(), Error>) -> Option { + self.fsync_state_mut().complete(result) + } + + fn fdatasync_state_mut(&mut self) -> &mut CompletionState> { + match self { + Self::Fdatasync(state) => state, + _ => unreachable!(), + } + } + + fn into_fdatasync_state(self) -> CompletionState> { + match self { + Self::Fdatasync(state) => state, + _ => unreachable!(), + } + } + + pub(crate) fn complete_fdatasync(&mut self, result: Result<(), Error>) -> Option { + self.fdatasync_state_mut().complete(result) + } + + fn noop_state_mut(&mut self) -> &mut CompletionState> { + match self { + Self::Noop(state) => state, + _ => unreachable!(), + } + } + + fn into_noop_state(self) -> CompletionState> { + match self { + Self::Noop(state) => state, + _ => unreachable!(), + } + } + + pub(crate) fn complete_noop(&mut self, result: Result<(), Error>) -> Option { + self.noop_state_mut().complete(result) + } +} + +pub struct Completion { + inner: CompletionInner, +} + +impl Completion { + pub(super) fn ready(val: T) -> Self { + CompletionInner::Ready(Some(val)).into() + } +} + +impl Completion, ErrorWith>>> { + pub(super) fn write(pending: Arc>, key: usize) -> Self { + CompletionInner::Poll { + pending, + key, + poll: |pending, key, cx| { + poll_completion( + pending, + key, + CompletionHandle::write_state_mut, + CompletionHandle::into_write_state, + reify, + cx, + ) + }, + } + .into() + } + + pub(super) fn read(pending: Arc>, key: usize) -> Self { + CompletionInner::Poll { + pending, + key, + poll: |pending, key, cx| { + poll_completion( + pending, + key, + CompletionHandle::read_state_mut, + CompletionHandle::into_read_state, + reify, + cx, + ) + }, + } + .into() + } +} + +impl Completion> { + pub(super) fn open(pending: Arc>, key: usize) -> Self { + CompletionInner::Poll { + pending, + key, + poll: |pending, key, cx| { + poll_completion( + pending, + key, + CompletionHandle::open_state_mut, + CompletionHandle::into_open_state, + identity, + cx, + ) + }, + } + .into() + } + + pub(super) fn create(pending: Arc>, key: usize) -> Self { + CompletionInner::Poll { + pending, + key, + poll: |pending, key, cx| { + poll_completion( + pending, + key, + CompletionHandle::create_state_mut, + CompletionHandle::into_create_state, + identity, + cx, + ) + }, + } + .into() + } +} + +impl Completion> { + pub(super) fn stat(pending: Arc>, key: usize) -> Self { + CompletionInner::Poll { + pending, + key, + poll: |pending, key, cx| { + poll_completion( + pending, + key, + CompletionHandle::stat_state_mut, + CompletionHandle::into_stat_state, + identity, + cx, + ) + }, + } + .into() + } +} + +impl Completion> { + pub(super) fn fallocate(pending: Arc>, key: usize) -> Self { + CompletionInner::Poll { + pending, + key, + poll: |pending, key, cx| { + poll_completion( + pending, + key, + CompletionHandle::fallocate_state_mut, + CompletionHandle::into_fallocate_state, + identity, + cx, + ) + }, + } + .into() + } + + pub(super) fn fsync(pending: Arc>, key: usize) -> Self { + CompletionInner::Poll { + pending, + key, + poll: |pending, key, cx| { + poll_completion( + pending, + key, + CompletionHandle::fsync_state_mut, + CompletionHandle::into_fsync_state, + identity, + cx, + ) + }, + } + .into() + } + + pub(super) fn fdatasync(pending: Arc>, key: usize) -> Self { + CompletionInner::Poll { + pending, + key, + poll: |pending, key, cx| { + poll_completion( + pending, + key, + CompletionHandle::fdatasync_state_mut, + CompletionHandle::into_fdatasync_state, + identity, + cx, + ) + }, + } + .into() + } + + #[allow(unused)] + pub(super) fn noop(pending: Arc>, key: usize) -> Self { + CompletionInner::Poll { + pending, + key, + poll: |pending, key, cx| { + poll_completion( + pending, + key, + CompletionHandle::noop_state_mut, + CompletionHandle::into_noop_state, + identity, + cx, + ) + }, + } + .into() + } +} + +impl From> for Completion { + fn from(inner: CompletionInner) -> Self { + Self { inner } + } +} + +/// Dropping a [Completion] future removes the [CompletionHandle] from the +/// pending list, if it is present. +/// +/// If it is not present, then the future was polled to completion already. +enum CompletionInner { + Poll { + pending: Arc>, + key: usize, + poll: fn(spin::MutexGuard<'_, PendingCompletions>, usize, &mut Context<'_>) -> Poll, + }, + Ready(Option), +} + +impl Drop for CompletionInner { + fn drop(&mut self) { + let Self::Poll { pending, key, .. } = self else { + return; + }; + pending.lock().try_remove(*key); + } +} + +fn poll_completion( + mut pending: spin::MutexGuard<'_, PendingCompletions>, + key: usize, + state_mut: fn(&mut CompletionHandle) -> &mut CompletionState, + into_state: fn(CompletionHandle) -> CompletionState, + map: fn(S) -> T, + cx: &mut Context<'_>, +) -> Poll { + match pending.get_mut(key) { + None => unreachable!("completion polled after already complete"), + Some(handle) => { + if let CompletionState::Pending(maybe_waker) = (state_mut)(handle) { + if !maybe_waker.as_ref().is_some_and(|waker| waker.will_wake(cx.waker())) { + *maybe_waker = Some(cx.waker().clone()); + } + + return Poll::Pending; + } + + let handle = pending.remove(key); + let state = (into_state)(handle); + + match state { + CompletionState::Ready(result) => Poll::Ready((map)(result)), + CompletionState::Pending(_) => unreachable!("pending case already handled"), + } + } + } +} + +impl Unpin for Completion {} + +impl Future for Completion { + type Output = T; + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + let this = self.get_mut(); + match &mut this.inner { + CompletionInner::Poll { pending, key, poll } => poll(pending.lock(), *key, cx), + CompletionInner::Ready(val) => match val.take() { + Some(val) => Poll::Ready(val), + None => Poll::Pending, + }, + } + } +} + +fn reify( + result: Result>, +) -> Result, ErrorWith>> { + match result { + Ok(erased) => Ok(erased.into_aligned::()), + Err(ErrorWith { error, with }) => Err(ErrorWith { + error, + with: with.into_aligned::(), + }), + } +} diff --git a/crates/runtime-io/src/sim/executor.rs b/crates/runtime-io/src/sim/executor.rs new file mode 100644 index 00000000000..165bfd8cb03 --- /dev/null +++ b/crates/runtime-io/src/sim/executor.rs @@ -0,0 +1,947 @@ +use alloc::collections::VecDeque; +use core::{ + iter::{Map, Scan}, + num::NonZeroUsize, + ops::Range, + result::Result, + task::Waker, +}; +use slab::Slab; + +use crate::{ + sim::{ + completion::CompletionHandle, + faults::{EitherOrBoth, IndexSelector, TaskSelection, TaskSelector}, + fs::{self, Datasync}, + Error, FaultInjector, + }, + ErasedBox, ErrorWith, Statx, SECTOR_SIZE, SECTOR_SIZE64, +}; + +mod sqe; +use sqe::SqeInner; +pub use sqe::{LinkKind, Sqe, SqeId}; + +// TODO: There is no difference between fsync and fdatasync until we extend +// [Statx] with additional fields. +#[derive(Clone, Copy)] +pub enum FsyncEffect { + Datasync(Datasync), +} + +impl From for FsyncEffect { + fn from(value: Datasync) -> Self { + Self::Datasync(value) + } +} + +#[derive(Clone, Copy)] +pub enum Operation { + WriteSector(WriteSector), + ReadSector(ReadSector), + Open, + Create, + Stat, + Fallocate, + Fsync { effect: FsyncEffect }, + Fdatasync { effect: Datasync }, + Noop, +} + +#[derive(Clone, Copy)] +pub struct WriteSector { + pub sector: usize, + pub buf_offset: usize, +} + +#[derive(Clone, Copy)] +pub struct ReadSector { + pub sector: usize, + pub buf_offset: usize, +} + +#[derive(Debug)] +pub struct Cqe { + inner: CqeInner, + user_data: Option, +} +impl Cqe { + pub fn user_data(&self) -> &Option { + &self.user_data + } + + pub(crate) fn complete(self, completion: &mut CompletionHandle) -> Option { + self.inner.complete(completion) + } +} + +#[derive(Debug)] +pub enum CqeInner { + Write { + result: Result, + buf: ErasedBox, + }, + Read { + result: Result, + buf: ErasedBox, + }, + Open { + result: Result, + }, + Create { + result: Result, + }, + Stat { + result: Result, + }, + Fallocate { + result: Result<(), Error>, + }, + Fsync { + result: Result<(), Error>, + }, + Fdatasync { + result: Result<(), Error>, + }, + Noop { + result: Result<(), Error>, + }, +} + +impl CqeInner { + fn complete(self, completion: &mut CompletionHandle) -> Option { + match self { + Self::Write { result, buf, .. } => { + let result = match result { + Ok(written) if written == buf.len() => Ok(buf), + Ok(written) => Err(ErrorWith { + error: Error::ShortWrite { + expected: buf.len(), + written, + }, + with: buf, + }), + Err(error) => Err(ErrorWith { error, with: buf }), + }; + completion.complete_write(result) + } + Self::Read { result, buf, .. } => { + let result = match result { + Ok(read) if read == buf.len() => Ok(buf), + Ok(read) => Err(ErrorWith { + error: Error::UnexpectedEof { + expected: buf.len(), + read, + }, + with: buf, + }), + Err(error) => Err(ErrorWith { error, with: buf }), + }; + completion.complete_read(result) + } + Self::Open { result, .. } => completion.complete_open(result), + Self::Create { result, .. } => completion.complete_create(result), + Self::Stat { result, .. } => completion.complete_stat(result), + Self::Fallocate { result, .. } => completion.complete_fallocate(result), + Self::Fsync { result, .. } => completion.complete_fsync(result), + Self::Fdatasync { result, .. } => completion.complete_fdatasync(result), + Self::Noop { result, .. } => completion.complete_noop(result), + } + } +} + +pub struct InFlight { + sqe: SqeInner, + state: InFlightState, + next: Option, + pub user_data: Option, +} + +impl InFlight { + fn is_blocked(&self) -> bool { + self.state.is_blocked() + } + + fn is_active(&self) -> bool { + self.state.is_active() + } + + fn cancel(self) -> Cqe { + self.sqe.cancel(self.user_data) + } +} + +enum InFlightState { + Blocked, + Active(Pending), +} + +impl InFlightState { + fn is_blocked(&self) -> bool { + matches!(self, Self::Blocked) + } + + fn is_active(&self) -> bool { + matches!(self, Self::Active(_)) + } +} + +struct Link { + kind: LinkKind, + next: SqeId, +} + +pub trait ResultAcc { + fn empty() -> Self; + fn add(&mut self, other: Self); +} + +impl ResultAcc for usize { + fn empty() -> Self { + 0 + } + + fn add(&mut self, other: Self) { + *self += other + } +} + +impl ResultAcc for () { + fn empty() -> Self {} + fn add(&mut self, _: Self) {} +} + +pub struct Results { + remaining: usize, + value: T, + error: Option, +} + +impl Results { + fn new(op_count: usize) -> Self { + Self { + remaining: op_count, + value: T::empty(), + error: None, + } + } + + fn push(&mut self, res: Result) { + match res { + Ok(value) => self.value.add(value), + Err(e) => { + self.error.get_or_insert(e); + } + } + self.remaining -= 1; + } + + fn into_result(self) -> Result { + assert_eq!(self.remaining, 0); + self.error.map_or(Ok(self.value), Err) + } + + fn is_complete(&self) -> bool { + self.remaining == 0 + } +} + +type ReadWriteOps = Scan, usize, fn(&mut usize, usize) -> Option>; +type SyncOps = Map Operation>; + +enum Pending { + ReadWrite { ops: ReadWriteOps, results: Results }, + Sync { ops: SyncOps, results: Results<()> }, + Unit { op: Option }, +} + +impl Iterator for Pending { + type Item = Operation; + + fn next(&mut self) -> Option { + match self { + Self::ReadWrite { ops, .. } => ops.next(), + Self::Sync { ops, .. } => ops.next(), + Self::Unit { op } => op.take(), + } + } +} + +struct Executing { + sqe: SqeId, + inner: Operation, +} + +impl Executing { + fn traverse(self, f: impl FnOnce(SqeId, Operation) -> Option) -> Option { + let Self { sqe, inner } = self; + f(sqe, inner).map(|inner| Self { sqe, inner }) + } +} + +/// Completion queue overflow policy. +/// +/// Note that we do **not** model `IORING_FEAT_NODROP`, because we never want +/// the application to rely on dynamic memory allocation in the kernel. +/// +/// The default is to panic, which should prompt the user to adjust queue size +/// configuration. However, sometimes it may be useful to see how the +/// application behaves when completions are dropped. +#[derive(Clone, Copy, Default)] +pub enum OnCqOverflow { + #[default] + Panic, + Drop, +} + +pub struct Options { + /// Capacity of the submission queue. + /// + /// This basically limits how many [Sqe]s can be submitted in one batch. + /// Should be a power of 2, or is otherwise rounded up to the next power of + /// 2. + pub sq_capacity: NonZeroUsize, + /// Override the completion queue capacity. + /// + /// By default, the completion queue's capacity is twice the submission + /// queue's. This can be insufficient for some workloads, so this setting + /// can be used to override the default. + /// + /// Should be a power of 2, or is otherwise rounded up to the next power of + /// two. + pub cq_capacity: Option, + /// What to do if the completion queue overflows. + pub cq_overflow: OnCqOverflow, + /// Bound on the number of concurrently executing tasks. + /// + /// Note that one [Sqe] can result in many operations to be scheduled. + /// Should be a power of 2, or is otherwise rounded up to the next power of + /// two. + pub max_concurrency: NonZeroUsize, + /// Size in bytes of the virtual disk / filesystem. + /// + /// Should be a multiple of [SECTOR_SIZE] and is rounded if it isn't. + pub disk_space_bytes: u64, +} + +impl Options { + pub(crate) fn sq_capacity(&self) -> usize { + self.sq_capacity.get().next_power_of_two() + } + + pub(crate) fn cq_capacity(&self) -> usize { + self.cq_capacity + .map(|c| c.get().next_power_of_two()) + .unwrap_or_else(|| 2 * self.sq_capacity()) + } + + pub(crate) fn max_concurrency(&self) -> usize { + self.max_concurrency.get().next_power_of_two() + } +} + +impl Default for Options { + fn default() -> Self { + Self { + sq_capacity: NonZeroUsize::new(8).unwrap(), + cq_capacity: None, + cq_overflow: OnCqOverflow::default(), + max_concurrency: NonZeroUsize::new(32).unwrap(), + disk_space_bytes: 2 * 4096, + } + } +} + +pub struct Executor { + submissions: VecDeque>, + completions: VecDeque>, + + in_flight: Slab>, + executing: VecDeque, + + fs: fs::Filesystem, + + cq_overflow: OnCqOverflow, + cq_dropped: usize, +} + +impl Executor { + pub fn new(options: Options) -> Self { + let sq_capacity = options.sq_capacity(); + let cq_capacity = options.cq_capacity(); + let fs_capacity = options.disk_space_bytes.next_multiple_of(SECTOR_SIZE64) as usize; + Self { + submissions: VecDeque::with_capacity(sq_capacity), + completions: VecDeque::with_capacity(cq_capacity), + in_flight: Slab::with_capacity(2 * sq_capacity), + executing: VecDeque::with_capacity(options.max_concurrency()), + fs: fs::Filesystem::new(fs_capacity), + cq_overflow: options.cq_overflow, + cq_dropped: 0, + } + } + + /// Simulate a power-loss crash. + /// + /// All submitted and executing operations are cancelled, and files reset to + /// their durable state. After this method returns, the completion queue is + /// empty. + pub fn power_loss(&mut self) { + self.submissions.clear(); + self.completions.clear(); + self.in_flight.clear(); + self.executing.clear(); + self.cq_dropped = 0; + self.fs.power_loss(); + } + + /// Restart the executor, simulating a process crash. + /// + /// Unlike [Self::crash], this will allow the currently executing operations + /// to complete. It will, however, cancel all in-flight submissions, even if + /// they were partly completed. + /// + /// Submissions that were not yet scheduled are dropped. The file state + /// remains unchanged. + /// + /// Execution is subject to `faults`. If a fault evaluates to [Fault::Skip], + /// that operation is dropped. + /// + /// All outstanding [Cqe]s are fed to `complete`, including completions that + /// happen as part of the shutdown. The completion queue is empty after this + /// method returns. + pub fn restart(&mut self, faults: &mut impl FaultInjector, mut complete: impl FnMut(Cqe)) { + self.submissions.clear(); + self.cq_dropped = 0; + + self.completed().for_each(&mut complete); + while let Some(op) = self.executing.pop_front() { + if let Some(in_flight) = self.in_flight.get(op.sqe.key()) + && !in_flight.is_active() + { + continue; + } + self.execute_op(op, faults); + self.completed().for_each(&mut complete); + } + self.executing.clear(); + for in_flight in self.in_flight.drain() { + complete(in_flight.cancel()); + } + } + + /// Submit a batch of [Sqe]s for later execution. + pub fn submit(&mut self, sqes: Batch) -> Result<(), Batch::IntoIter> + where + Batch: IntoIterator>, + Batch::IntoIter: ExactSizeIterator, + { + let sqes = sqes.into_iter(); + if self.submissions.len() + sqes.len() > self.submissions.capacity() { + Err(sqes) + } else { + self.submissions.extend(sqes); + Ok(()) + } + } + + fn complete(&mut self, cqe: Cqe) { + if self.completions.len() == self.completions.capacity() { + match self.cq_overflow { + OnCqOverflow::Panic => panic!("completion queue overflow"), + OnCqOverflow::Drop => { + self.cq_dropped += 1; + return; + } + } + } + self.completions.push_back(cqe); + } + + /// Number of completions that were dropped due to completion queue overflow + /// over the lifetime of this executor. + /// + /// Always zero if the executor was configured with [OnCqOverflow::Panic]. + #[allow(unused)] + pub fn dropped_completions(&self) -> usize { + self.cq_dropped + } + + /// Drain the completion queue. + pub fn completed(&mut self) -> impl Iterator> { + self.completions.drain(..) + } + + /// Drain the submission queue and advance one scheduled operation. + /// + /// The operation to advance is chosen by `task_selector`. + /// The operation is subject to `faults`. + pub fn tick(&mut self, task_selector: &impl TaskSelector, faults: &mut impl FaultInjector) -> bool { + let mut progress = self.schedule(); + progress |= self.execute(task_selector, faults); + progress + } + + fn schedule_blocked(&mut self, sqe: SqeInner, user_data: Option) -> SqeId { + let entry = self.in_flight.vacant_entry(); + let sqe_id = SqeId(entry.key()); + + entry.insert(InFlight { + sqe, + state: InFlightState::Blocked, + next: None, + user_data, + }); + + sqe_id + } + + fn unblock(&mut self, sqe: SqeId) { + let Self { in_flight, .. } = self; + + let in_flight = in_flight.get_mut(sqe.key()).expect("invalid sqe id"); + assert!(in_flight.is_blocked(), "in-flight sqe unblocked more than once"); + in_flight.state = InFlightState::Active(in_flight.sqe.prepare()); + } + + fn have_in_flight_capacity(&self) -> bool { + // The batch size is the prefix of linked SQEs, plus the first unlinked + // one. They all need to be scheduled together to preserve linking + // semantics. + let batch_size = self + .submissions + .iter() + .position(|sqe| !sqe.is_linked()) + .map_or(self.submissions.len(), |i| i + 1); + + batch_size <= self.in_flight.capacity() - self.in_flight.len() + } + + fn schedule(&mut self) -> bool { + let mut progress = false; + + if self.have_in_flight_capacity() { + while let Some(Sqe { inner, link, user_data }) = self.submissions.pop_front() { + let head = self.schedule_blocked(inner, user_data); + + let mut prev = head; + let mut prev_link = link; + while let Some(kind) = prev_link { + let Some(Sqe { inner, link, user_data }) = self.submissions.pop_front() else { + break; + }; + + let next = self.schedule_blocked(inner, user_data); + let prev_in_flight = self.in_flight.get_mut(prev.key()).expect("invalid sqe id"); + prev_in_flight.next = Some(Link { kind, next }); + + prev = next; + prev_link = link; + } + + self.unblock(head); + progress |= true; + + if !self.have_in_flight_capacity() { + break; + } + } + } + + for (sqe_id, in_flight) in self.in_flight.iter_mut() { + if self.executing.capacity() == self.executing.len() { + break; + } + + let InFlightState::Active(pending) = &mut in_flight.state else { + continue; + }; + if let Some(op) = pending.next() { + self.executing.push_back(Executing { + sqe: SqeId(sqe_id), + inner: op, + }); + progress |= true; + } + } + + progress + } + + fn execute(&mut self, task_selector: &impl TaskSelector, faults: &mut impl FaultInjector) -> bool { + let mut progress = false; + + let mut run = |this: &mut Executor, op| { + if let Some(delay) = this.execute_op(op, faults) { + this.executing.push_back(delay); + } + progress |= true; + }; + + let Some(remaining) = NonZeroUsize::new(self.executing.len()) else { + return progress; + }; + match task_selector.select_tasks(remaining) { + TaskSelection::Fifo { count } => { + let mut count = count.get(); + while count > 0 { + let Some(op) = self.executing.pop_front() else { + break; + }; + run(self, op); + count -= 1; + } + } + TaskSelection::Any { count, mut select } => { + let mut count = count.get(); + while count > 0 { + let Some(op) = (|| -> Option { + let remaining = NonZeroUsize::new(self.executing.len())?; + let index = select.select_index(remaining); + self.executing.swap_remove_front(index) + })() else { + break; + }; + run(self, op); + count -= 1; + } + } + } + + progress + } + + fn execute_op(&mut self, op: Executing, faults: &mut impl FaultInjector) -> Option { + op.traverse(|sqe, op| { + let in_flight = self.in_flight.get(sqe.key()).expect("invalid sqe id"); + assert!(in_flight.is_active()); + + match op { + Operation::WriteSector(effect) => faults + .inject_write_sector_fault(in_flight, effect) + .exec_visible(|eff| self.execute_write_sector(sqe, eff)) + .map(Operation::WriteSector), + Operation::ReadSector(effect) => faults + .inject_read_sector_fault(in_flight, effect) + .exec_visible(|eff| self.execute_read_sector(sqe, eff)) + .map(Operation::ReadSector), + Operation::Open => faults + .inject_open_fault(in_flight) + .exec_visible(|eff| self.execute_open(sqe, eff)) + .map(|()| Operation::Open), + Operation::Create => faults + .inject_create_fault(in_flight) + .exec_visible(|eff| self.execute_create(sqe, eff)) + .map(|()| Operation::Create), + Operation::Stat => faults + .inject_stat_fault(in_flight) + .exec_visible(|eff| self.execute_stat(sqe, eff)) + .map(|()| Operation::Stat), + Operation::Fallocate => faults + .inject_fallocate_fault(in_flight) + .exec_visible(|eff| self.execute_fallocate(sqe, eff)) + .map(|()| Operation::Fallocate), + Operation::Fsync { effect } => faults + .inject_fsync_fault(in_flight, effect) + .exec_visible(|eff| self.execute_fsync(sqe, eff)) + .map(|effect| Operation::Fsync { effect }), + Operation::Fdatasync { effect } => faults + .inject_fdatasync_fault(in_flight, effect) + .exec_visible(|eff| self.execute_fdatasync(sqe, eff)) + .map(|effect| Operation::Fdatasync { effect }), + Operation::Noop => faults + .inject_noop_fault(in_flight) + .exec_visible(|eff| self.execute_noop(sqe, eff)) + .map(|()| Operation::Noop), + } + }) + } + + fn execute_write_sector(&mut self, sqe: SqeId, eff: EitherOrBoth) { + let is_complete = { + let InFlight { + sqe: SqeInner::Write { fd, buf, .. }, + state: InFlightState::Active(Pending::ReadWrite { results, .. }), + .. + } = self.in_flight.get_mut(sqe.key()).expect("invalid sqe id") + else { + unreachable!("invalid sqe: expected write") + }; + let run = |WriteSector { sector, buf_offset }| { + let bytes = buf.as_bytes(); + let end = (buf_offset + SECTOR_SIZE).min(bytes.len()); + + let buf: &[u8; SECTOR_SIZE] = buf.as_bytes()[buf_offset..end] + .try_into() + .expect("buffer must be sector aligned"); + fd.write_sector(buf, sector as _).map_err(Into::into) + }; + results.push(eff.traverse(run, Err)); + results.is_complete() + }; + + if is_complete { + let InFlight { + sqe: SqeInner::Write { buf, .. }, + state: InFlightState::Active(Pending::ReadWrite { results, .. }), + next, + user_data, + } = self.in_flight.remove(sqe.key()) + else { + unreachable!("invalid sqe: expected write") + }; + let result = results.into_result(); + let is_success = result.is_ok(); + self.complete(Cqe { + inner: CqeInner::Write { result, buf }, + user_data, + }); + self.schedule_linked(is_success, next); + } + } + + fn execute_read_sector(&mut self, sqe: SqeId, eff: EitherOrBoth) { + let is_complete = { + let InFlight { + sqe: SqeInner::Read { fd, buf, .. }, + state: InFlightState::Active(Pending::ReadWrite { results, .. }), + .. + } = self.in_flight.get_mut(sqe.key()).expect("invalid sqe id") + else { + unreachable!("invalid sqe: expected read") + }; + let run = |ReadSector { sector, buf_offset }| { + let bytes = buf.as_bytes_mut(); + let end = (buf_offset + SECTOR_SIZE).min(bytes.len()); + + let buf: &mut [u8; SECTOR_SIZE] = (&mut buf.as_bytes_mut()[buf_offset..end]) + .try_into() + .expect("buffer must be sector aligned"); + fd.read_sector(buf, sector as _).map_err(Into::into) + }; + results.push(eff.traverse(run, Err)); + results.is_complete() + }; + + if is_complete { + let InFlight { + sqe: SqeInner::Read { buf, .. }, + state: InFlightState::Active(Pending::ReadWrite { results, .. }), + next, + user_data, + } = self.in_flight.remove(sqe.key()) + else { + unreachable!("invalid sqe: expected read") + }; + let result = results.into_result(); + let is_success = result.is_ok(); + self.complete(Cqe { + inner: CqeInner::Read { result, buf }, + user_data, + }); + self.schedule_linked(is_success, next); + } + } + + fn execute_open(&mut self, sqe: SqeId, eff: EitherOrBoth<(), Error>) { + let InFlight { + sqe: SqeInner::Open { path }, + state: InFlightState::Active(Pending::Unit { .. }), + next, + user_data, + } = self.in_flight.remove(sqe.key()) + else { + unreachable!("invalid sqe: expected open") + }; + let result = eff.traverse(|()| self.fs.open(&path).map_err(Into::into), Err); + + let is_success = result.is_ok(); + self.complete(Cqe { + inner: CqeInner::Open { result }, + user_data, + }); + self.schedule_linked(is_success, next); + } + + fn execute_create(&mut self, sqe: SqeId, eff: EitherOrBoth<(), Error>) { + let InFlight { + sqe: SqeInner::Create { path }, + state: InFlightState::Active(Pending::Unit { .. }), + next, + user_data, + } = self.in_flight.remove(sqe.key()) + else { + unreachable!("invalid sqe: expected create") + }; + let run = |()| self.fs.create(path).map_err(Into::into); + let result = eff.traverse(run, Err); + let is_success = result.is_ok(); + self.complete(Cqe { + inner: CqeInner::Create { result }, + user_data, + }); + self.schedule_linked(is_success, next); + } + + fn execute_stat(&mut self, sqe: SqeId, eff: EitherOrBoth<(), Error>) { + let InFlight { + sqe: SqeInner::Stat { fd }, + state: InFlightState::Active(Pending::Unit { .. }), + next, + user_data, + } = self.in_flight.remove(sqe.key()) + else { + unreachable!("invalid sqe: expected stat") + }; + let result = eff.traverse(|()| Ok(Statx { size: fd.len() }), Err); + let is_success = result.is_ok(); + self.complete(Cqe { + inner: CqeInner::Stat { result }, + user_data, + }); + self.schedule_linked(is_success, next); + } + + fn execute_fallocate(&mut self, sqe: SqeId, eff: EitherOrBoth<(), Error>) { + let InFlight { + sqe: SqeInner::Fallocate { fd, total_len }, + state: InFlightState::Active(Pending::Unit { .. }), + next, + user_data, + } = self.in_flight.remove(sqe.key()) + else { + unreachable!("invalid sqe: expected fallocate") + }; + let result = eff.traverse(|()| fd.reserve(total_len).map_err(Into::into), Err); + let is_success = result.is_ok(); + self.complete(Cqe { + inner: CqeInner::Fallocate { result }, + user_data, + }); + self.schedule_linked(is_success, next); + } + + fn execute_fsync(&mut self, sqe: SqeId, eff: EitherOrBoth) { + let is_complete = { + let InFlight { + sqe: SqeInner::Fsync { fd }, + state: InFlightState::Active(Pending::Sync { results, .. }), + .. + } = self.in_flight.get_mut(sqe.key()).expect("invalid sqe id") + else { + unreachable!("invalid sqe: expected fsync") + }; + let result = eff.traverse( + |FsyncEffect::Datasync(effect)| fd.apply_datasync(effect).map_err(Into::into), + Err, + ); + results.push(result); + results.is_complete() + }; + + if is_complete { + let InFlight { + state: InFlightState::Active(Pending::Sync { results, .. }), + next, + user_data, + .. + } = self.in_flight.remove(sqe.key()) + else { + unreachable!("invalid sqe: expected fsync") + }; + let result = results.into_result(); + let is_success = result.is_ok(); + self.complete(Cqe { + inner: CqeInner::Fsync { result }, + user_data, + }); + self.schedule_linked(is_success, next); + } + } + + fn execute_fdatasync(&mut self, sqe: SqeId, eff: EitherOrBoth) { + let is_complete = { + let InFlight { + sqe: SqeInner::Fdatasync { fd }, + state: InFlightState::Active(Pending::Sync { results, .. }), + .. + } = self.in_flight.get_mut(sqe.key()).expect("invalid sqe id") + else { + unreachable!("invalid sqe: expected fdatasync") + }; + let result = eff.traverse(|effect| fd.apply_datasync(effect).map_err(Into::into), Err); + results.push(result); + results.is_complete() + }; + + if is_complete { + let InFlight { + state: InFlightState::Active(Pending::Sync { results, .. }), + next, + user_data, + .. + } = self.in_flight.remove(sqe.key()) + else { + unreachable!("invalid sqe: expected fdatasync") + }; + let result = results.into_result(); + let is_success = result.is_ok(); + self.complete(Cqe { + inner: CqeInner::Fdatasync { result }, + user_data, + }); + self.schedule_linked(is_success, next); + } + } + + fn execute_noop(&mut self, sqe: SqeId, eff: EitherOrBoth<(), Error>) { + let InFlight { + sqe: SqeInner::Noop, + state: InFlightState::Active(Pending::Unit { .. }), + next, + user_data, + } = self.in_flight.remove(sqe.key()) + else { + unreachable!("invalid sqe: expected noop") + }; + let result = eff.traverse(Ok, Err); + let is_success = result.is_ok(); + self.complete(Cqe { + inner: CqeInner::Noop { result }, + user_data, + }); + self.schedule_linked(is_success, next); + } + + fn schedule_linked(&mut self, prev_succeeded: bool, next: Option) { + let Some(Link { kind, next }) = next else { + return; + }; + match (kind, prev_succeeded) { + (LinkKind::Hard, _) | (LinkKind::Soft, true) => { + self.unblock(next); + } + (LinkKind::Soft, false) => { + self.cancel_chain(next); + } + } + } + + fn cancel_chain(&mut self, mut head: SqeId) { + loop { + let InFlight { + sqe, + state, + next, + user_data, + } = self.in_flight.remove(head.key()); + assert!(state.is_blocked(), "attempted to cancel an active linked sqe"); + + self.complete(sqe.cancel(user_data)); + let Some(Link { next, .. }) = next else { + break; + }; + head = next; + } + } +} diff --git a/crates/runtime-io/src/sim/executor/sqe.rs b/crates/runtime-io/src/sim/executor/sqe.rs new file mode 100644 index 00000000000..a5c130ac08a --- /dev/null +++ b/crates/runtime-io/src/sim/executor/sqe.rs @@ -0,0 +1,262 @@ +use alloc::boxed::Box; + +use crate::{ + sim::{ + executor::{Cqe, CqeInner, FsyncEffect, Operation, Pending, ReadSector, Results, WriteSector}, + fs::{self, Datasync}, + Error, + }, + ErasedBox, SECTOR_SIZE, SECTOR_SIZE64, +}; + +/// Opaque identifier of a scheduled [Sqe]. +#[derive(Clone, Copy)] +pub struct SqeId(pub(super) usize); + +impl SqeId { + pub(super) fn key(&self) -> usize { + self.0 + } +} + +/// Dependency on the previous [Sqe]. +/// +/// A link imposes an ordering constraint: the [Sqe] carrying the link will not +/// be executed before the preceding one completed. Note that a link is only +/// meaningful within a batch of SQEs submitted together. +#[derive(Clone, Copy)] +pub enum LinkKind { + /// If the preceding SQE failed, cancel this SQE with [Error::Cancelled]. + /// Analogous to `IOSQE_IO_LINK`. + Soft, + /// Run the SQE regardless of the preceding SQE's result. + /// Analoguous to `IOSQE_IO_HARDLINK`. + Hard, +} + +pub struct Sqe { + pub(crate) inner: SqeInner, + pub(super) link: Option, + pub(super) user_data: Option, +} + +impl Sqe { + fn new(inner: SqeInner) -> Self { + Self { + inner, + link: None, + user_data: None, + } + } + + #[allow(unused)] + pub fn link(mut self, kind: Option) -> Self { + self.link = kind; + self + } + + pub fn is_linked(&self) -> bool { + self.link.is_some() + } + + pub fn attach(mut self, user_data: T) -> Self { + self.user_data.replace(user_data); + self + } + + pub fn write(fd: fs::File, buf: ErasedBox, offset: u64) -> Self { + assert!(offset.is_multiple_of(SECTOR_SIZE64)); + Self::new(SqeInner::Write { fd, buf, offset }) + } + + pub fn read(fd: fs::File, buf: ErasedBox, offset: u64) -> Self { + assert!(offset.is_multiple_of(SECTOR_SIZE64)); + Self::new(SqeInner::Read { fd, buf, offset }) + } + + pub fn open(path: Box) -> Self { + Self::new(SqeInner::Open { path }) + } + + pub fn create(path: Box) -> Self { + Self::new(SqeInner::Create { path }) + } + + pub fn stat(fd: fs::File) -> Self { + Self::new(SqeInner::Stat { fd }) + } + + pub fn fallocate(fd: fs::File, len: u64) -> Self { + Self::new(SqeInner::Fallocate { fd, total_len: len }) + } + + pub fn fsync(fd: fs::File) -> Self { + Self::new(SqeInner::Fsync { fd }) + } + + pub fn fdatasync(fd: fs::File) -> Self { + Self::new(SqeInner::Fdatasync { fd }) + } + + #[allow(unused)] + pub fn noop() -> Self { + Self::new(SqeInner::Noop) + } + + /// Extract the [ErasedBox] buffer if the [Sqe] carries one. + pub(crate) fn into_buf(self) -> Option { + match self.inner { + SqeInner::Write { buf, .. } | SqeInner::Read { buf, .. } => Some(buf), + SqeInner::Open { .. } + | SqeInner::Create { .. } + | SqeInner::Stat { .. } + | SqeInner::Fallocate { .. } + | SqeInner::Fsync { .. } + | SqeInner::Fdatasync { .. } + | SqeInner::Noop => None, + } + } +} + +pub enum SqeInner { + Write { fd: fs::File, buf: ErasedBox, offset: u64 }, + Read { fd: fs::File, buf: ErasedBox, offset: u64 }, + Open { path: Box }, + Create { path: Box }, + Stat { fd: fs::File }, + Fallocate { fd: fs::File, total_len: u64 }, + Fsync { fd: fs::File }, + Fdatasync { fd: fs::File }, + Noop, +} + +impl SqeInner { + pub(super) fn cancel(self, user_data: Option) -> Cqe { + match self { + SqeInner::Write { buf, .. } => Cqe { + inner: CqeInner::Write { + result: Err(Error::Cancelled), + buf, + }, + user_data, + }, + SqeInner::Read { buf, .. } => Cqe { + inner: CqeInner::Read { + result: Err(Error::Cancelled), + buf, + }, + user_data, + }, + SqeInner::Open { .. } => Cqe { + inner: CqeInner::Open { + result: Err(Error::Cancelled), + }, + user_data, + }, + SqeInner::Create { .. } => Cqe { + inner: CqeInner::Create { + result: Err(Error::Cancelled), + }, + user_data, + }, + SqeInner::Stat { .. } => Cqe { + inner: CqeInner::Stat { + result: Err(Error::Cancelled), + }, + user_data, + }, + SqeInner::Fallocate { .. } => Cqe { + inner: CqeInner::Fallocate { + result: Err(Error::Cancelled), + }, + user_data, + }, + SqeInner::Fsync { .. } => Cqe { + inner: CqeInner::Fsync { + result: Err(Error::Cancelled), + }, + user_data, + }, + SqeInner::Fdatasync { .. } => Cqe { + inner: CqeInner::Fdatasync { + result: Err(Error::Cancelled), + }, + user_data, + }, + SqeInner::Noop => Cqe { + inner: CqeInner::Noop { + result: Err(Error::Cancelled), + }, + user_data, + }, + } + } + + pub(super) fn prepare(&self) -> Pending { + match self { + SqeInner::Write { buf, offset, .. } => { + let buf_len = buf.as_bytes().len(); + let first_sector = (*offset / SECTOR_SIZE64) as usize; + let sector_count = buf_len / SECTOR_SIZE; + + Pending::ReadWrite { + ops: (0..sector_count).scan(first_sector, |first_sector, sector| { + Some(Operation::WriteSector(WriteSector { + sector: *first_sector + sector, + buf_offset: sector * SECTOR_SIZE, + })) + }), + results: Results::new(sector_count), + } + } + SqeInner::Read { buf, offset, .. } => { + let buf_len = buf.as_bytes().len(); + let first_sector = (*offset / SECTOR_SIZE64) as usize; + let sector_count = buf_len / SECTOR_SIZE; + + Pending::ReadWrite { + ops: (0..sector_count).scan(first_sector, |first_sector, sector| { + Some(Operation::ReadSector(ReadSector { + sector: *first_sector + sector, + buf_offset: sector * SECTOR_SIZE, + })) + }), + results: Results::new(sector_count), + } + } + SqeInner::Fsync { fd } => { + let sector_count = fd.len() / SECTOR_SIZE64; + Pending::Sync { + ops: fd.prepare_datasync().map( + (|effect| Operation::Fsync { + effect: FsyncEffect::Datasync(effect), + }) as fn(Datasync) -> Operation, + ), + results: Results::new(1 + sector_count as usize), + } + } + SqeInner::Fdatasync { fd } => { + let sector_count = fd.len() / SECTOR_SIZE64; + Pending::Sync { + ops: fd.prepare_datasync().map(|effect| Operation::Fdatasync { effect }), + results: Results::new(1 + sector_count as usize), + } + } + SqeInner::Open { .. } => Pending::Unit { + op: Some(Operation::Open), + }, + SqeInner::Create { .. } => Pending::Unit { + op: Some(Operation::Create), + }, + SqeInner::Stat { .. } => Pending::Unit { + op: Some(Operation::Stat), + }, + SqeInner::Fallocate { .. } => Pending::Unit { + op: Some(Operation::Fallocate), + }, + SqeInner::Noop => Pending::Unit { + op: Some(Operation::Noop), + }, + } + } +} diff --git a/crates/runtime-io/src/sim/faults.rs b/crates/runtime-io/src/sim/faults.rs new file mode 100644 index 00000000000..5428f3ef5f9 --- /dev/null +++ b/crates/runtime-io/src/sim/faults.rs @@ -0,0 +1,198 @@ +use core::{convert::Infallible, num::NonZeroUsize}; + +use crate::sim::{ + executor::{FsyncEffect, InFlight, ReadSector, WriteSector}, + fs::Datasync, + Error, +}; + +/// Interface to inject [Fault]s into the simulator. +/// +/// Submissions are split into one or more operations, or effects, each +/// executing independently and in any order (see [TaskSelector]). For example, +/// a write that spans multiple sectors will yield one atomic write per sector. +/// +/// Before executing an effect, the simulator calls the corresponding `inject_*` +/// method on the fault injector. This allows the implementor to skip, delay or +/// fail the effect. Where applicable, the effect can also be modified: for +/// example, a sector write can be made misdirected by returning a [WriteSector] +/// with a different `sector` from [FaultInjector::inject_write_sector_fault]. +/// +/// A [FaultInjector] can be stateful, which is why the methods take a mutable +/// `self` reference. If desired, effects can be correlated with submissions via +/// the [InFlight] reference passed to the `inject_*` methods. +pub trait FaultInjector { + fn inject_write_sector_fault(&mut self, _: &InFlight, op: WriteSector) -> Fault { + Fault::Visible(Effect::Run(op)) + } + + fn inject_read_sector_fault(&mut self, _: &InFlight, op: ReadSector) -> Fault { + Fault::Visible(Effect::Run(op)) + } + + fn inject_open_fault(&mut self, _: &InFlight) -> Fault<()> { + Fault::Visible(Effect::Run(())) + } + + fn inject_create_fault(&mut self, _: &InFlight) -> Fault<()> { + Fault::Visible(Effect::Run(())) + } + + fn inject_stat_fault(&mut self, _: &InFlight) -> Fault<()> { + Fault::Visible(Effect::Run(())) + } + + fn inject_fallocate_fault(&mut self, _: &InFlight) -> Fault<()> { + Fault::Visible(Effect::Run(())) + } + + fn inject_fsync_fault(&mut self, _: &InFlight, op: FsyncEffect) -> Fault { + Fault::Visible(Effect::Run(op)) + } + + fn inject_fdatasync_fault(&mut self, _: &InFlight, op: Datasync) -> Fault { + Fault::Visible(Effect::Run(op)) + } + + fn inject_noop_fault(&mut self, _: &InFlight) -> Fault<()> { + Fault::Visible(Effect::Run(())) + } +} + +/// The unit [FaultInjector] injects no faults. +impl FaultInjector for () {} + +pub enum Fault { + /// Drop the operation entirely. + /// + /// Note that this is not generally possible in `io-uring`: an SQE always + /// yields a CQE, even if it was cancelled or returned an error. It may be + /// useful occasionally to construct "byzantine" failures. + Skip, + /// Put the operation back onto the queue for later execution. + Delay(T), + /// Execute a visible effect. + Visible(Effect), +} + +impl Fault { + pub(super) fn exec_visible(self, f: impl FnOnce(EitherOrBoth)) -> Option { + match self { + Fault::Skip => None, + Fault::Delay(effect) => Some(effect), + Fault::Visible(visible) => { + visible.exec(f); + None + } + } + } +} + +pub enum Effect { + /// Run the operation as normal. + Run(T), + /// Run the effect, but report an injected error. + RunThenError { effect: T, error: Error }, + /// Skip the effect, but report an injected error. + SkipThenError { error: Error }, +} + +impl Effect { + fn exec(self, f: impl FnOnce(EitherOrBoth)) { + use EitherOrBoth::*; + match self { + Effect::Run(effect) => f(Left(effect)), + Effect::RunThenError { effect, error } => f(Both(effect, error)), + Effect::SkipThenError { error } => f(Right(error)), + } + } +} + +pub(super) enum EitherOrBoth { + Left(T), + Right(U), + Both(T, U), +} + +impl EitherOrBoth { + pub(super) fn traverse(self, f: impl FnOnce(T) -> V, g: impl FnOnce(U) -> V) -> V { + match self { + Self::Left(t) => f(t), + Self::Right(u) => g(u), + Self::Both(t, u) => { + f(t); + g(u) + } + } + } +} + +/// Interface to perturb execution ordering. +/// +/// Submissions are split into one or more operations, or effects, each +/// executing independently. For example, a write that spans multiple sectors +/// will yield one atomic write per sector. +/// +/// Effects could be executed by the kernel in any order, and this trait allows +/// to inject this ordering deterministically (e.g. using a deterministic random +/// number generator). +pub trait TaskSelector { + type IndexSelector<'a>: IndexSelector + where + Self: 'a; + + /// Of `task_count` queued tasks, select the ones to run. + /// + /// Called on each [crate::sim::Executor::tick]. + fn select_tasks(&self, task_count: NonZeroUsize) -> TaskSelection>; +} + +/// Task selection. +pub enum TaskSelection { + /// Run `count` tasks in FIFO order. + Fifo { count: NonZeroUsize }, + /// Select `count` tasks by calling `select` `count` times. + Any { count: NonZeroUsize, select: S }, +} + +/// Select an index in a range. +pub trait IndexSelector { + /// Select a number in the range `0..range_upper`. + fn select_index(&mut self, range_upper: NonZeroUsize) -> usize; +} + +impl IndexSelector for Infallible { + fn select_index(&mut self, _: NonZeroUsize) -> usize { + match *self {} + } +} + +/// [TaskSelector] that runs all queued tasks in FIFO order. +pub struct FifoAll; + +impl TaskSelector for FifoAll { + type IndexSelector<'a> + = Infallible + where + Self: 'a; + + fn select_tasks(&self, count: NonZeroUsize) -> TaskSelection> { + TaskSelection::Fifo { count } + } +} + +/// [TaskSelector] that runs one task at a time, in FIFO order. +pub struct FifoOne; + +impl TaskSelector for FifoOne { + type IndexSelector<'a> + = Infallible + where + Self: 'a; + + fn select_tasks(&self, _: NonZeroUsize) -> TaskSelection> { + TaskSelection::Fifo { + count: NonZeroUsize::MIN, + } + } +} diff --git a/crates/runtime-io/src/sim/fs.rs b/crates/runtime-io/src/sim/fs.rs new file mode 100644 index 00000000000..5d1c6094b20 --- /dev/null +++ b/crates/runtime-io/src/sim/fs.rs @@ -0,0 +1,378 @@ +use alloc::{boxed::Box, collections::BTreeMap, sync::Arc, vec::Vec}; +use core::mem; + +use spin::Mutex; + +use crate::{SECTOR_SIZE, SECTOR_SIZE64}; + +type SectorId = usize; +type FileId = usize; + +#[derive(Clone, Copy, Debug, Eq, PartialEq, thiserror::Error)] +pub enum Error { + #[error("file already exists")] + FileAlreadyExists, + #[error("file not found")] + FileNotFound, + #[error("no space left on device")] + NoSpace, + #[error("invalid argument")] + InvalidArgument, +} + +pub type Result = core::result::Result; + +#[derive(Debug)] +struct Sector { + volatile: Box<[u8; SECTOR_SIZE]>, + durable: Box<[u8; SECTOR_SIZE]>, +} + +#[derive(Debug)] +struct FileState { + // Logical sector index -> physical sector. + sectors: Vec, + + volatile_len: u64, + durable_len: u64, + + // Incremented whenever a datasync snapshot is created. + dirty_generation: u64, + // Logical sector -> generation it was last dirtied. + dirty_sectors: BTreeMap, + // Generation in which the length was last dirtied. + dirty_len: Option, +} + +/// A virtual filesystem that keeps track of files, allocated space and +/// name->file mappings. +#[derive(Clone, Debug)] +pub struct Filesystem { + inner: Arc>, +} + +#[derive(Debug)] +struct FsInner { + sectors: Vec, + free: Vec, + + files: Vec, + paths: BTreeMap, FileId>, +} + +/// A virtual file in the [Filesystem]. +#[derive(Clone, Debug)] +pub struct File { + fs: Filesystem, + id: FileId, +} + +/// Individual effects produced by [File::prepare_datasync]. +/// +/// `fsync` / `fdatasync` is modelled as a series of sector effects and +/// potentially the file length. This allows to inject failures, in particular +/// partial failure of a sync operation. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Datasync { + Sector { sector: usize, generation: u64 }, + Length { generation: u64 }, +} + +impl Filesystem { + pub fn new(capacity: usize) -> Self { + assert!(capacity.is_multiple_of(SECTOR_SIZE)); + + let sector_count = capacity / SECTOR_SIZE; + let sectors = (0..sector_count) + .map(|_| Sector { + volatile: Box::new([0; SECTOR_SIZE]), + durable: Box::new([0; SECTOR_SIZE]), + }) + .collect(); + + Self { + inner: Arc::new(Mutex::new(FsInner { + sectors, + free: (0..sector_count).rev().collect(), + files: Vec::new(), + paths: BTreeMap::new(), + })), + } + } + + pub fn create(&self, path: Box) -> Result { + let mut fs = self.inner.lock(); + + if fs.paths.contains_key(&path) { + return Err(Error::FileAlreadyExists); + } + + let id = fs.files.len(); + + fs.files.push(FileState { + sectors: Vec::new(), + volatile_len: 0, + durable_len: 0, + dirty_generation: 0, + dirty_sectors: BTreeMap::new(), + dirty_len: None, + }); + + fs.paths.insert(path, id); + + Ok(File { fs: self.clone(), id }) + } + + pub fn open(&self, path: &str) -> Result { + let fs = self.inner.lock(); + + let id = *fs.paths.get(path).ok_or(Error::FileNotFound)?; + + Ok(File { fs: self.clone(), id }) + } + + /// Simulate power loss. + /// + /// All unsynced data and metadata are discarded. + pub fn power_loss(&self) { + let mut fs = self.inner.lock(); + + for file_id in 0..fs.files.len() { + let durable_len = fs.files[file_id].durable_len; + let durable_sectors = sector_count(durable_len); + + // Allocations beyond the durable file length were not durably + // reachable, so they become free again. + while fs.files[file_id].sectors.len() > durable_sectors { + let sector_id = fs.files[file_id].sectors.pop().unwrap(); + fs.free.push(sector_id); + } + + let sector_ids = fs.files[file_id].sectors.clone(); + + for sector_id in sector_ids { + let sector = &mut fs.sectors[sector_id]; + sector.volatile.copy_from_slice(&*sector.durable); + } + + let file = &mut fs.files[file_id]; + file.volatile_len = file.durable_len; + file.dirty_generation = 0; + file.dirty_sectors.clear(); + file.dirty_len = None; + } + } +} + +impl File { + pub(super) fn len(&self) -> u64 { + self.fs.inner.lock().files[self.id].volatile_len + } + + /// Preallocate enough physical sectors to cover `new_len`. + /// + /// Growing fails with [Error::NoSpace] if the fixed filesystem pool cannot + /// satisfy the allocation. + /// + /// Shrinking is deliberately not handled by this operation. + /// + /// The result of this operation is not durable until [Self::apply_datasync] + /// is called. + pub(super) fn reserve(&self, new_len: u64) -> Result<()> { + let mut fs = self.fs.inner.lock(); + grow(&mut fs, self.id, new_len) + } + + /// Read one complete sector. + /// + /// Returns 0 at or beyond EOF. Reading one sector is atomic. + pub(super) fn read_sector(&self, dst: &mut [u8; SECTOR_SIZE], sector: usize) -> Result { + let fs = self.fs.inner.lock(); + let file = &fs.files[self.id]; + + let offset = sector as u64 * SECTOR_SIZE64; + if offset >= file.volatile_len { + return Ok(0); + } + + let sector_id = file.sectors[sector]; + dst.copy_from_slice(&fs.sectors[sector_id].volatile[..]); + + Ok(SECTOR_SIZE) + } + + /// Write one complete sector. + /// + /// Like `pwrite`, this may extend the file. Extending can fail with + /// [Error::NoSpace]. Writing one sector is atomic. + pub(super) fn write_sector(&self, src: &[u8; SECTOR_SIZE], sector: usize) -> Result { + let mut fs = self.fs.inner.lock(); + + let end = (sector + 1).checked_mul(SECTOR_SIZE).ok_or(Error::InvalidArgument)? as u64; + grow(&mut fs, self.id, end)?; + + let sector_id = fs.files[self.id].sectors[sector]; + + fs.sectors[sector_id].volatile.copy_from_slice(src); + let generation = fs.files[self.id].dirty_generation; + fs.files[self.id].dirty_sectors.insert(sector, generation); + + Ok(SECTOR_SIZE) + } + + /// Produce a series of [Datasync] effects to model `fdatasync`. + /// + /// The iterator captures sectors that are marked dirty at the time this + /// method is called, and ignores sectors dirtied later (while traversing + /// the iterator). Similarly for the file length. + /// + /// This is not an accurate model of how `fdatasync` works, but allows + /// interleaving of effects and injection of faults to produce + /// partially-durable states. + /// + /// [Datasync] effects are executed via [Self::apply_datasync]. + pub(super) fn prepare_datasync(&self) -> IterDatasync { + let mut fs = self.fs.inner.lock(); + let file = &mut fs.files[self.id]; + + let generation = file.dirty_generation; + file.dirty_generation += 1; + + IterDatasync { + file: self.clone(), + generation, + next_sector: 0, + length_pending: true, + } + } + + /// Execute one [Datasync] effect produced by [Self::prepare_datasync]. + /// + /// A sector is removed from the dirty set only after it succeeds. + pub(super) fn apply_datasync(&self, effect: Datasync) -> Result<()> { + let mut fs = self.fs.inner.lock(); + + match effect { + Datasync::Sector { sector, generation } => { + // The sector may cease to exist if the file was concurrently + // truncated (not currently supported). Treat such an obsolete + // sync effect as already satisfied. + let Some(§or_id) = fs.files[self.id].sectors.get(sector) else { + fs.files[self.id].dirty_sectors.remove(§or); + return Ok(()); + }; + + // Persist what is visible when this effect executes. + let volatile = fs.sectors[sector_id].volatile.clone(); + fs.sectors[sector_id].durable.copy_from_slice(&*volatile); + + // Clear dirty flag only if the sector hasn't been modified + // since. + let file = &mut fs.files[self.id]; + if file + .dirty_sectors + .get(§or) + .is_some_and(|&dirty_generation| dirty_generation <= generation) + { + file.dirty_sectors.remove(§or); + } + } + + Datasync::Length { generation } => { + let volatile_len = fs.files[self.id].volatile_len; + let file = &mut fs.files[self.id]; + // Persist the length visible when this effect executes. + file.durable_len = volatile_len; + // Clear dirty flag only if no resize happened since. + if file + .dirty_len + .is_some_and(|dirty_generation| dirty_generation <= generation) + { + file.dirty_len = None; + } + } + } + + Ok(()) + } +} + +pub(super) struct IterDatasync { + file: File, + generation: u64, + next_sector: usize, + length_pending: bool, +} + +impl Iterator for IterDatasync { + type Item = Datasync; + + fn next(&mut self) -> Option { + let fs = self.file.fs.inner.lock(); + let file = &fs.files[self.file.id]; + + if let Some((§or, _)) = file + .dirty_sectors + .range(self.next_sector..) + .find(|(_, generation)| **generation <= self.generation) + { + self.next_sector = sector + 1; + return Some(Datasync::Sector { + sector, + generation: self.generation, + }); + } + + if mem::take(&mut self.length_pending) && file.dirty_len.is_some_and(|generation| generation <= self.generation) + { + return Some(Datasync::Length { + generation: self.generation, + }); + } + + None + } +} + +fn sector_count(len: u64) -> usize { + len.div_ceil(SECTOR_SIZE64) as usize +} + +fn grow(fs: &mut FsInner, file_id: FileId, new_len: u64) -> Result<()> { + let old_len = fs.files[file_id].volatile_len; + + if new_len < old_len { + return Err(Error::InvalidArgument); + } + + if new_len == old_len { + return Ok(()); + } + + let old_sector_count = sector_count(old_len); + let new_sector_count = sector_count(new_len); + let needed = new_sector_count - old_sector_count; + + // Make growth atomic with respect to ENOSPC: either all required sectors + // are allocated, or nothing changes. + if fs.free.len() < needed { + return Err(Error::NoSpace); + } + + for _ in 0..needed { + let sector_id = fs.free.pop().unwrap(); + + // Newly allocated filesystem space reads as zero. Clearing both copies + // also prevents data from a previous owner from becoming observable. + fs.sectors[sector_id].volatile.fill(0); + fs.sectors[sector_id].durable.fill(0); + + fs.files[file_id].sectors.push(sector_id); + } + + let file = &mut fs.files[file_id]; + file.volatile_len = new_len; + file.dirty_len = Some(file.dirty_generation); + + Ok(()) +} diff --git a/crates/runtime-io/src/sim/mod.rs b/crates/runtime-io/src/sim/mod.rs new file mode 100644 index 00000000000..e7d63662771 --- /dev/null +++ b/crates/runtime-io/src/sim/mod.rs @@ -0,0 +1,473 @@ +use alloc::{boxed::Box, rc::Rc, sync::Arc}; +use core::{result::Result, task::Waker}; + +use crate::{ + sim::{ + completion::{CompletionState, PendingCompletions}, + executor::Cqe, + }, + AlignedBytes, ErasedBox, ErrorWith, ReadWriteResult, SpacetimeIO, Statx, +}; + +mod completion; +pub use completion::Completion; +use completion::CompletionHandle; + +mod executor; +use executor::{Executor, Sqe}; + +mod faults; +pub use faults::{FaultInjector, FifoAll, FifoOne, IndexSelector, TaskSelector}; + +mod fs; +pub use fs::File; + +pub use crate::{ + sim::executor::{LinkKind, Options}, + SECTOR_SIZE, +}; + +#[derive(Debug, thiserror::Error)] +pub enum Error { + #[error("failed to write expected number of bytes")] + ShortWrite { expected: usize, written: usize }, + #[error("unexpected eof")] + UnexpectedEof { expected: usize, read: usize }, + #[error(transparent)] + Fs(fs::Error), + /// Injected by the I/O driver. + #[error("operation cancelled")] + Cancelled, + #[error("submission queue overflow")] + SubmissionQueueOverflow, + #[error("too many pending completion futures")] + TooManyCompletions, +} + +impl From for Error { + fn from(e: fs::Error) -> Self { + Self::Fs(e) + } +} + +#[derive(Clone, Default)] +pub struct SimulatorIO { + inner: Rc, +} + +impl SimulatorIO { + pub fn with_options(options: Options) -> Self { + Self { + inner: Rc::new(SimulatorInner::with_options(options)), + } + } + + pub fn tick(&self, task_selector: &impl TaskSelector, faults: &mut impl FaultInjector) -> bool { + let mut executor = self.inner.executor.lock(); + + let mut progress = executor.tick(task_selector, faults); + executor.completed().for_each(|cqe| { + self.process_cqe(cqe); + progress |= true; + }); + + progress + } + + fn process_cqe(&self, cqe: Cqe) { + let key = cqe.user_data().expect("user data must be set"); + let maybe_waker = (|| -> Option { + let mut pending = self.inner.pending.lock(); + // If the handle is no longer present in `pending`, the + // completion future was dropped. + let handle = pending.get_mut(key)?; + cqe.complete(handle) + })(); + if let Some(waker) = maybe_waker { + waker.wake(); + } + } + + /// Simulate a power loss event. + /// + /// All submitted and executing operations are cancelled, and files reset to + /// their durable state. + /// + /// The caller must uphold that no pending [Completion] futures are live (as + /// would happen during an actual power loss event). Polling a [Completion] + /// future after calling this method will panic. + pub fn power_loss(&self) { + self.inner.executor.lock().power_loss(); + self.inner.pending.lock().clear(); + } + + /// Simulate a restart event, i.e. process crash. + /// + /// Unlike [Self::power_loss], this will drive the currently executing + /// operations to completion, subject to fault injection. + /// + /// Submissions that were not yet scheduled are dropped. The file state + /// remains unchanged. + /// + /// Pending [Completion]s will resolve to [Error::Cancelled]. + /// + /// Note that the simulator owns the storage for [Completion]s until they + /// are either dropped or completed. So they count toward the completion + /// queue capacity. To simulate an actual process crash, the caller should + /// prefer to drop pending [Completion]s. + pub fn restart(&self, faults: &mut impl FaultInjector) { + self.inner.executor.lock().restart(faults, |cqe| self.process_cqe(cqe)) + } + + fn submit( + &self, + sqe: Sqe, + completion: impl FnOnce(Arc>, usize) -> Completion>, + completion_handle: impl FnOnce(CompletionState>) -> CompletionHandle, + ) -> Completion> { + let mut executor = self.inner.executor.lock(); + let mut pending = self.inner.pending.lock(); + match pending.vacant_entry() { + Some(pending_entry) => match executor.submit([sqe.attach(pending_entry.key())]) { + Err(_sqe) => Completion::ready(Err(Error::SubmissionQueueOverflow)), + Ok(()) => { + let key = pending_entry.key(); + pending_entry.insert(completion_handle(CompletionState::Pending(None))); + + completion(self.inner.pending.clone(), key) + } + }, + None => Completion::ready(Err(Error::TooManyCompletions)), + } + } + + fn submit_with( + &self, + sqe: Sqe, + completion: impl FnOnce(Arc>, usize) -> Completion>, + completion_handle: impl FnOnce(CompletionState>>) -> CompletionHandle, + ) -> Completion> { + let mut executor = self.inner.executor.lock(); + let mut pending = self.inner.pending.lock(); + + let reify = |sqe: Sqe| { + sqe.into_buf() + .map(|erased| erased.into_aligned()) + .expect("sqe must have been buffer-carrying") + }; + + match pending.vacant_entry() { + Some(pending_entry) => match executor.submit([sqe.attach(pending_entry.key())]) { + Err(mut sqe) => Completion::ready(Err(ErrorWith { + error: Error::SubmissionQueueOverflow, + with: reify( + sqe.next() + .expect("submitted one sqe therefore one must be returned on overflow"), + ), + })), + Ok(()) => { + let key = pending_entry.key(); + pending_entry.insert(completion_handle(CompletionState::Pending(None))); + + completion(self.inner.pending.clone(), key) + } + }, + None => Completion::ready(Err(ErrorWith { + error: Error::TooManyCompletions, + with: reify(sqe), + })), + } + } +} + +struct SimulatorInner { + executor: spin::Mutex>, + pending: Arc>, +} + +impl SimulatorInner { + fn with_options(options: Options) -> Self { + let pending = PendingCompletions::with_capacity(options.cq_capacity()); + let executor = Executor::new(options); + Self { + pending: Arc::new(spin::Mutex::new(pending)), + executor: spin::Mutex::new(executor), + } + } +} + +impl Default for SimulatorInner { + fn default() -> Self { + Self::with_options(<_>::default()) + } +} + +impl SpacetimeIO for SimulatorIO { + type Fd = fs::File; + type Error = Error; + type Completion = Completion; + + fn open_file(&self, path: Box) -> Self::Completion> { + self.submit(Sqe::open(path), Completion::open, CompletionHandle::Open) + } + + fn create_file(&self, path: Box) -> Self::Completion> { + self.submit(Sqe::create(path), Completion::create, CompletionHandle::Create) + } + + fn write_all_at( + &self, + fd: Self::Fd, + buf: Box, + offset: u64, + ) -> Self::Completion, ErrorWith>>> { + self.submit_with( + Sqe::write(fd, ErasedBox::from_aligned(buf), offset), + Completion::write, + CompletionHandle::Write, + ) + } + + fn read_exact_at( + &self, + fd: Self::Fd, + buf: Box, + offset: u64, + ) -> Self::Completion, ErrorWith>>> { + self.submit_with( + Sqe::read(fd, ErasedBox::from_aligned(buf), offset), + Completion::read, + CompletionHandle::Read, + ) + } + + fn fsync(&self, fd: Self::Fd) -> Self::Completion> { + self.submit(Sqe::fsync(fd), Completion::fsync, CompletionHandle::Fsync) + } + + fn fdatasync(&self, fd: Self::Fd) -> Self::Completion> { + self.submit(Sqe::fdatasync(fd), Completion::fdatasync, CompletionHandle::Fdatasync) + } + + fn reserve(&self, fd: Self::Fd, total_size: u64) -> Self::Completion> { + self.submit( + Sqe::fallocate(fd, total_size), + Completion::fallocate, + CompletionHandle::Fallocate, + ) + } + + fn statx(&self, fd: Self::Fd) -> Self::Completion> { + self.submit(Sqe::stat(fd), Completion::stat, CompletionHandle::Stat) + } +} + +#[cfg(test)] +mod tests { + use core::num::NonZeroUsize; + + use spacetimedb_runtime_core::sim::Rng; + + use super::*; + use crate::{sim::faults::TaskSelection, SECTOR_SIZE64}; + + struct RandomTaskSelector<'a> { + rng: &'a Rng, + } + + impl TaskSelector for RandomTaskSelector<'_> { + type IndexSelector<'a> + = &'a Rng + where + Self: 'a; + + fn select_tasks(&self, task_count: NonZeroUsize) -> TaskSelection> { + let count = NonZeroUsize::new(self.rng.index(task_count.get())).unwrap_or(NonZeroUsize::MIN); + if self.rng.sample_probability(0.5) { + TaskSelection::Fifo { count } + } else { + TaskSelection::Any { + count, + select: self.rng, + } + } + } + } + + impl IndexSelector for &Rng { + fn select_index(&mut self, range_upper: NonZeroUsize) -> usize { + self.index(range_upper.get()) + } + } + + struct Runtime { + rt: tokio::runtime::LocalRuntime, + io: SimulatorIO, + rng: Rng, + } + + impl Runtime { + fn new() -> Self { + Self { + rt: tokio::runtime::Builder::new_current_thread() + .build_local(<_>::default()) + .unwrap(), + io: SimulatorIO::default(), + rng: Rng::new(0), + } + } + + fn run(&self, f: impl FnOnce(&SimulatorIO) -> Completion) -> T { + let fut = self.rt.spawn_local(f(&self.io)); + while self.io.tick(&RandomTaskSelector { rng: &self.rng }, &mut ()) {} + self.rt.block_on(fut).unwrap() + } + + fn power_loss(&self) { + self.io.power_loss(); + } + } + + #[test] + fn create_file() { + let rt = Runtime::new(); + rt.run(|io| io.create_file("/data/test".into())).unwrap(); + } + + #[derive(Debug)] + #[repr(C, align(4096))] + struct Buf([u8; N]); + + impl Buf { + fn clear(&mut self) { + self.0.fill(0); + } + } + + impl AlignedBytes for Buf { + 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(), N); + let mut buf = [0; N]; + buf.copy_from_slice(b); + Self(buf) + } + } + + #[test] + fn write_read_roundtrip() { + let rt = Runtime::new(); + + let fd = rt.run(|io| io.create_file("/data/test".into())).unwrap(); + let buf = Box::new(Buf([22; 2 * SECTOR_SIZE])); + let mut buf = rt + .run(|io| io.write_all_at(fd.clone(), buf, 0)) + .map_err(ErrorWith::into_err) + .unwrap(); + buf.clear(); + let buf = rt.run(|io| io.read_exact_at(fd, buf, 0)).unwrap(); + + assert_eq!(buf.0, [22; 2 * SECTOR_SIZE]); + } + + #[test] + fn write_read_at_offset() { + let rt = Runtime::new(); + + let fd = rt.run(|io| io.create_file("/data/test".into())).unwrap(); + let buf: Box> = { + let mut buf = Box::new(Buf([0; SECTOR_SIZE])); + for i in 0usize..2 { + buf.0.fill((i + 1) as u8 * 2); + let offset = (i * SECTOR_SIZE) as u64; + buf = rt + .run(|io| io.write_all_at(fd.clone(), buf, offset)) + .map_err(ErrorWith::into_err) + .unwrap(); + } + + buf.clear(); + buf + }; + let buf = rt.run(|io| io.read_exact_at(fd, buf, SECTOR_SIZE64)).unwrap(); + + assert_eq!(buf.0, [4; SECTOR_SIZE]); + } + + #[test] + fn preallocate() { + let rt = Runtime::new(); + + let fd = rt.run(|io| io.create_file("/data/test".into())).unwrap(); + rt.run(|io| io.reserve(fd.clone(), 2 * SECTOR_SIZE64)).unwrap(); + + // Check that reserved space reads as zeroes. + let buf = rt + .run(|io| io.read_exact_at(fd.clone(), Box::new(Buf([1; 2 * SECTOR_SIZE])), 0)) + .unwrap(); + assert_eq!(buf.0, [0; 2 * SECTOR_SIZE]); + + // The length is reported as the preallocated length. + let stat = rt.run(|io| io.statx(fd.clone())).unwrap(); + assert_eq!(stat.size, 2 * SECTOR_SIZE64); + + // Overwriting the second sector works. + let buf = rt + .run(|io| io.write_all_at(fd.clone(), Box::new(Buf([42; SECTOR_SIZE])), SECTOR_SIZE64)) + .unwrap(); + let buf = rt.run(|io| io.read_exact_at(fd.clone(), buf, SECTOR_SIZE64)).unwrap(); + assert_eq!(buf.0, [42; SECTOR_SIZE]); + // The first sector still reads as zeroes. + let buf = rt.run(|io| io.read_exact_at(fd, buf, 0)).unwrap(); + assert_eq!(buf.0, [0; SECTOR_SIZE]); + } + + #[test] + fn open_succeeds_after_create() { + let rt = Runtime::new(); + + matches!( + rt.run(|io| io.open_file("/data/test".into())), + Err(Error::Fs(fs::Error::FileNotFound)) + ); + rt.run(|io| io.create_file("/data/test".into())).unwrap(); + assert!(rt.run(|io| io.open_file("/data/test".into())).is_ok()); + } + + #[test] + fn unsynced_data_is_lost_after_power_loss() { + let rt = Runtime::new(); + + let fd = rt.run(|io| io.create_file("/data/test".into())).unwrap(); + let mut buf = rt + .run(|io| io.write_all_at(fd.clone(), Box::new(Buf([1; SECTOR_SIZE])), 0)) + .map_err(ErrorWith::into_err) + .unwrap(); + buf.clear(); + + rt.run(|io| io.fdatasync(fd.clone())).unwrap(); + + let mut buf = rt + .run(|io| io.write_all_at(fd.clone(), Box::new(Buf([2; SECTOR_SIZE])), SECTOR_SIZE64)) + .map_err(ErrorWith::into_err) + .unwrap(); + buf.clear(); + + rt.power_loss(); + + let buf = rt.run(|io| io.read_exact_at(fd.clone(), buf, 0)).unwrap(); + assert_eq!(buf.0, [1; SECTOR_SIZE]); + matches!( + rt.run(|io| io.read_exact_at(fd.clone(), buf, SECTOR_SIZE as u64)) + .map_err(ErrorWith::into_err), + Err(Error::UnexpectedEof { .. }) + ); + } +} diff --git a/crates/runtime/Cargo.toml b/crates/runtime/Cargo.toml index c8affea0f48..10c6c6fbd28 100644 --- a/crates/runtime/Cargo.toml +++ b/crates/runtime/Cargo.toml @@ -11,11 +11,18 @@ workspace = true [dependencies] tokio.workspace = true -spacetimedb-runtime-core = { workspace = true, optional = true } -libc = { version = "0.2", optional = true } +spacetimedb-runtime-core = { workspace = true } +spacetimedb-runtime-io = { workspace = true } +static_assertions = "1.1" + +[target.'cfg(unix)'.dependencies] +libc = "0.2" + +[target.'cfg(windows)'.dependencies] +windows-sys = { version = "0.61", features = ["Win32_Storage_FileSystem"] } [dev-dependencies] futures.workspace = true [features] -simulation = ["dep:spacetimedb-runtime-core", "spacetimedb-runtime-core/sim", "dep:libc"] +simulation = ["spacetimedb-runtime-core/sim"] diff --git a/crates/runtime/src/io.rs b/crates/runtime/src/io.rs new file mode 100644 index 00000000000..f7c24fe029f --- /dev/null +++ b/crates/runtime/src/io.rs @@ -0,0 +1,2 @@ +mod tokio; +pub use tokio::TokioIO as Tokio; diff --git a/crates/runtime/src/io/tokio.rs b/crates/runtime/src/io/tokio.rs new file mode 100644 index 00000000000..1ae1e36fddd --- /dev/null +++ b/crates/runtime/src/io/tokio.rs @@ -0,0 +1,257 @@ +use std::io::{Seek, SeekFrom}; +use std::panic; +use std::path::PathBuf; +use std::pin::Pin; +use std::task::{Context, Poll}; +use std::{io, marker::PhantomData, rc::Rc, sync::Arc}; + +use spacetimedb_runtime_io::{AlignedBytes, ErrorWith, SpacetimeIO, Statx}; +use static_assertions::assert_not_impl_any; +use tokio::runtime; + +/// Implementation of [SpacetimeIO] that runs on a tokio runtime. +pub struct TokioIO { + rt: runtime::Handle, + // Ensure I/O stays on a single thread. + _not_send: PhantomData>, +} + +impl TokioIO { + pub fn new(rt: runtime::Handle) -> Self { + Self { + rt, + _not_send: PhantomData, + } + } +} + +assert_not_impl_any!(TokioIO: Send); + +#[must_use = "completions must be polled to completion"] +pub struct Completion(tokio::task::JoinHandle); + +impl Future for Completion { + type Output = T; + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + let this = self.get_mut(); + match Pin::new(&mut this.0).poll(cx) { + Poll::Pending => Poll::Pending, + Poll::Ready(result) => match result { + Ok(output) => Poll::Ready(output), + Err(error) => { + if error.is_panic() { + panic::resume_unwind(error.into_panic()) + } else if error.is_cancelled() { + panic!("completion cancelled unexpectedly") + } else { + unreachable!("unexpected I/O task error") + } + } + }, + } + } +} + +impl From> for Completion { + fn from(handle: tokio::task::JoinHandle) -> Self { + Self(handle) + } +} + +impl SpacetimeIO for TokioIO { + // NOTE: This operates on a [std::fs::File] handle instead of + // [tokio::fs::File] because `pwrite`/`pread`-style APIs are not available + // from tokio proper. As a consequence, operations on an open `Fd` use + // [spawn_blocking]. This is what [tokio::fs::File] does internally, while + // here we can avoid some locking. + type Fd = Arc; + type Error = io::Error; + type Completion = Completion; + + fn open_file(&self, path: Box) -> Self::Completion> { + let path = PathBuf::from(&*path); + self.rt + .spawn_blocking(move || { + let mut open_options = std::fs::File::options(); + open_options.read(true).write(true); + platform::open_with_direct_io(open_options, path).map(Arc::new) + }) + .into() + } + + fn create_file(&self, path: Box) -> Self::Completion> { + let path = PathBuf::from(&*path); + self.rt + .spawn_blocking(move || { + let mut open_options = std::fs::File::options(); + open_options.read(true).write(true).create_new(true); + platform::open_with_direct_io(open_options, path).map(Arc::new) + }) + .into() + } + + fn write_all_at( + &self, + fd: Self::Fd, + buf: Box, + offset: u64, + ) -> Self::Completion, ErrorWith>>> { + self.rt + .spawn_blocking(move || match platform::write_all_at(&fd, buf.as_bytes(), offset) { + Ok(()) => Ok(buf), + Err(error) => Err(ErrorWith { error, with: buf }), + }) + .into() + } + + fn read_exact_at( + &self, + fd: Self::Fd, + buf: Box, + offset: u64, + ) -> Self::Completion, ErrorWith>>> { + self.rt + .spawn_blocking(move || { + let mut buf = buf; + match platform::read_exact_at(&fd, buf.as_bytes_mut(), offset) { + Ok(()) => Ok(buf), + Err(error) => Err(ErrorWith { error, with: buf }), + } + }) + .into() + } + + fn fsync(&self, fd: Self::Fd) -> Self::Completion> { + self.rt.spawn_blocking(move || fd.sync_all()).into() + } + + fn fdatasync(&self, fd: Self::Fd) -> Self::Completion> { + self.rt.spawn_blocking(move || fd.sync_data()).into() + } + + fn reserve(&self, fd: Self::Fd, total: u64) -> Self::Completion> { + self.rt + .spawn_blocking(move || { + let mut fd = fd.try_clone()?; + let len = file_length(&mut fd)?; + assert!(total >= len); + fd.set_len(total) + }) + .into() + } + + fn statx(&self, fd: Self::Fd) -> Self::Completion> { + self.rt + .spawn_blocking(move || { + let mut fd = fd.try_clone()?; + file_length(&mut fd).map(Statx::from_size) + }) + .into() + } +} + +fn file_length(fd: &mut std::fs::File) -> io::Result { + let pos = fd.stream_position()?; + let len = fd.seek(SeekFrom::End(0))?; + + if pos != len { + fd.seek(SeekFrom::Start(pos))?; + } + + Ok(len) +} + +mod platform { + #[cfg(unix)] + pub use super::unix::*; + + #[cfg(windows)] + pub use super::windows::*; +} + +#[cfg(unix)] +mod unix { + use std::{ + io, + os::unix::fs::{FileExt as _, OpenOptionsExt as _}, + path::Path, + }; + + #[inline] + pub fn read_exact_at(fd: &std::fs::File, buf: &mut [u8], offset: u64) -> io::Result<()> { + fd.read_exact_at(buf, offset) + } + + #[inline] + pub fn write_all_at(fd: &std::fs::File, buf: &[u8], offset: u64) -> io::Result<()> { + fd.write_all_at(buf, offset) + } + + #[cfg(not(target_os = "macos"))] + pub fn open_with_direct_io(mut options: std::fs::OpenOptions, path: impl AsRef) -> io::Result { + options.custom_flags(libc::O_DIRECT).open(path) + } + + #[cfg(target_os = "macos")] + pub async fn open_with_direct_io( + options: std::fs::OpenOptions, + path: impl AsRef, + ) -> io::Result { + let file = options.open(path)?; + let res = unsafe { libc::fcntl(file.as_raw_fd(), libc::F_NOCACHE, 1) }; + if res == -1 { + Err(io::Error::last_os_error()) + } else { + Ok(file) + } + } +} + +#[cfg(windows)] +mod windows { + use std::io; + + pub fn write_all_at(fd: &std::fs::File, mut buf: &[u8], mut offset: u64) -> io::Result<()> { + while !buf.is_empty() { + match fd.seek_write(buf, offset) { + Ok(0) => return Err(io::ErrorKind::WriteZero.into()), + Ok(n) => { + offset += n as u64; + buf = &buf[n..]; + } + Err(ref e) if e.kind() == io::ErrorKind::Interrupted => {} + Err(e) => return Err(e), + } + } + + Ok(()) + } + + pub fn read_exact_at(fd: &std::fs::File, mut buf: &mut [u8], mut offset: u64) -> io::Result<()> { + while !buf.is_empty() { + match fd.seek_read(buf, offset) { + Ok(0) => return Err(io::ErrorKind::UnexpectedEof.into()), + Ok(n) => { + offset += n as u64; + buf = &mut buf[n..]; + } + Err(ref e) if e.kind() == io::ErrorKind::Interrupted => {} + Err(e) => return Err(e), + } + } + + Ok(()) + } + + pub async fn open_with_direct_io( + mut options: std::fs::OpenOptions, + path: impl AsRef, + ) -> io::Result { + use std::os::windows::fs::OpenOptionsExt as _; + + options + .custom_flags(windows_sys::Win32::Storage::FileSystem::FILE_FLAG_NO_BUFFERING) + .open(path) + } +} diff --git a/crates/runtime/src/lib.rs b/crates/runtime/src/lib.rs index c6192e1b738..48400676009 100644 --- a/crates/runtime/src/lib.rs +++ b/crates/runtime/src/lib.rs @@ -53,6 +53,8 @@ pub enum Handle { Simulation(sim::Handle), } +pub mod io; + pub struct JoinHandle { inner: JoinHandleInner, }