diff --git a/Cargo.lock b/Cargo.lock index 6f9455400ec..cdfcdb19b23 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8532,7 +8532,9 @@ dependencies = [ "futures", "libc", "spacetimedb-runtime-core", + "static_assertions", "tokio", + "windows-sys 0.61.2", ] [[package]] @@ -8540,7 +8542,12 @@ name = "spacetimedb-runtime-core" version = "2.10.0" dependencies = [ "async-task", + "futures-channel", + "slab", "spin", + "thiserror 2.0.17", + "tokio", + "zerocopy", ] [[package]] diff --git a/crates/runtime-core/Cargo.toml b/crates/runtime-core/Cargo.toml index a3369a69f89..5b2b8571106 100644 --- a/crates/runtime-core/Cargo.toml +++ b/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 diff --git a/crates/runtime-core/src/io/buf.rs b/crates/runtime-core/src/io/buf.rs new file mode 100644 index 00000000000..14b88f05c9e --- /dev/null +++ b/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 _: () = ::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() + } +} + +#[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, + len: usize, + layout: Layout, + ty: TypeId, + } + + impl ErasedBox { + /// Create an [ErasedBox] from `B` by allocating a new [Box]. + pub fn from_aligned(b: B) -> Self { + Self::from_aligned_box(Box::new(b)) + } + + /// Create an [ErasedBox] from an already-boxed `B`. + pub fn from_aligned_box(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) -> B { + *Self::into_aligned_box(self) + } + + /// Reify `B` via casting, without unboxing. + pub fn into_aligned_box(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_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::()); + 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::(); + + assert_eq!(reified, t); + } + } +} +#[cfg(feature = "alloc")] +pub use boxed::{ErasedBox, ErasedBoxPtr}; diff --git a/crates/runtime-core/src/io/error.rs b/crates/runtime-core/src/io/error.rs new file mode 100644 index 00000000000..0a06c1ada98 --- /dev/null +++ b/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 { + 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-core/src/io/mod.rs b/crates/runtime-core/src/io/mod.rs new file mode 100644 index 00000000000..d5a946b14ca --- /dev/null +++ b/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: Future + Unpin; + + /// Open the file at `path`. + fn open_file(&self, path: &str) -> Self::Completion>; + + /// Create the file at `path`. + /// + /// Returns an error if the file already exists. + fn create_file(&self, path: &str) -> 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: B, + 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: B, + 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-core/src/lib.rs b/crates/runtime-core/src/lib.rs index f7590ada98b..8a841c22036 100644 --- a/crates/runtime-core/src/lib.rs +++ b/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; diff --git a/crates/runtime-core/src/sim/executor/mod.rs b/crates/runtime-core/src/sim/executor/mod.rs index fbb7f7c0cf2..a0fbca1bf7c 100644 --- a/crates/runtime-core/src/sim/executor/mod.rs +++ b/crates/runtime-core/src/sim/executor/mod.rs @@ -18,7 +18,7 @@ pub use task::{AbortHandle, JoinError, JoinHandle}; type Runnable = async_task::Runnable; -#[derive(Clone, Copy, Debug, Eq, PartialEq)] +#[derive(Clone, Copy, Debug, PartialEq)] pub struct RuntimeConfig { pub seed: u64, } diff --git a/crates/runtime-core/src/sim/io/executor.rs b/crates/runtime-core/src/sim/io/executor.rs new file mode 100644 index 00000000000..9d4237a760b --- /dev/null +++ b/crates/runtime-core/src/sim/io/executor.rs @@ -0,0 +1,862 @@ +#![allow(unused)] + +use alloc::{ + boxed::Box, + collections::{btree_map, BTreeMap, VecDeque}, + vec::Vec, +}; +use core::{mem, num::NonZeroUsize, result::Result}; +use slab::Slab; + +use crate::{ + io::{ErasedBoxPtr, Statx, SECTOR_SIZE}, + sim::{ + io::{fs, Error, Instant}, + Rng, + }, +}; + +pub use crate::sim::io::fs::Datasync; + +mod sqe; +use sqe::SqeInner; +pub use sqe::{LinkKind, Sqe, SqeId}; + +// TODO: There is no difference between fsync and fdatasync as long as we don't +// have an API to fsync the directory of a file after it was created. +#[derive(Clone, Copy)] +pub enum FsyncEffect { + Datasync(Datasync), +} + +#[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 page_offset: usize, + pub buf_offset: usize, +} + +#[derive(Clone, Copy)] +pub struct ReadSector { + pub page_offset: usize, + pub buf_offset: usize, +} + +#[derive(Debug)] +pub enum Cqe { + Write { + result: Result, + user_data: Option, + }, + Read { + result: Result, + user_data: Option, + }, + Open { + result: Result, + user_data: Option, + }, + Create { + result: Result, + user_data: Option, + }, + Stat { + result: Result, + user_data: Option, + }, + Fallocate { + result: Result<(), Error>, + user_data: Option, + }, + Fsync { + result: Result<(), Error>, + user_data: Option, + }, + Fdatasync { + result: Result<(), Error>, + user_data: Option, + }, + Noop { + result: Result<(), Error>, + user_data: Option, + }, +} + +impl Cqe { + pub fn user_data(&self) -> &Option { + match self { + Self::Write { user_data, .. } + | Self::Read { user_data, .. } + | Self::Open { user_data, .. } + | Self::Create { user_data, .. } + | Self::Stat { user_data, .. } + | Self::Fallocate { user_data, .. } + | Self::Fsync { user_data, .. } + | Self::Fdatasync { user_data, .. } + | Self::Noop { user_data, .. } => user_data, + } + } +} + +pub struct Blocked { + pub link: LinkKind, + pub sqe: SqeInner, + pub user_data: Option, +} + +pub struct InFlight { + pub inner: InFlightInner, + pub blocked: VecDeque>, + pub user_data: Option, +} + +pub enum InFlightInner { + Write { + sqe: sqe::Write, + op_count: usize, + results: Vec>, + }, + Read { + sqe: sqe::Read, + op_count: usize, + results: Vec>, + }, + Open { + sqe: sqe::Open, + }, + Create { + sqe: sqe::Create, + }, + Stat { + sqe: sqe::Stat, + }, + Fallocate { + sqe: sqe::Fallocate, + }, + Fsync { + sqe: sqe::Fsync, + op_count: usize, + results: Vec>, + }, + Fdatasync { + sqe: sqe::Fdatasync, + op_count: usize, + results: Vec>, + }, + Noop, +} + +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 }) + } +} + +pub enum Fault { + /// Drop the operation entirely. + Skip, + /// Put the operation back onto the queue for later execution. + Delay(T), + /// Execute a visible effect. + Visible(Effect), +} + +impl Fault { + 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)), + } + } +} + +enum EitherOrBoth { + Left(T), + Right(U), + Both(T, U), +} + +impl EitherOrBoth { + 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) + } + } + } +} + +pub trait FaultInjector { + fn inject_write_sector_fault(&mut self, sqe: &InFlight, op: WriteSector) -> Fault { + Fault::Visible(Effect::Run(op)) + } + + fn inject_read_sector_fault(&mut self, sqe: &InFlight, op: ReadSector) -> Fault { + Fault::Visible(Effect::Run(op)) + } + + fn inject_open_fault(&mut self, sqe: &InFlight) -> Fault<()> { + Fault::Visible(Effect::Run(())) + } + + fn inject_create_fault(&mut self, sqe: &InFlight) -> Fault<()> { + Fault::Visible(Effect::Run(())) + } + + fn inject_stat_fault(&mut self, sqe: &InFlight) -> Fault<()> { + Fault::Visible(Effect::Run(())) + } + + fn inject_fallocate_fault(&mut self, sqe: &InFlight) -> Fault<()> { + Fault::Visible(Effect::Run(())) + } + + fn inject_fsync_fault(&mut self, sqe: &InFlight, op: FsyncEffect) -> Fault { + Fault::Visible(Effect::Run(op)) + } + + fn inject_fdatasync_fault(&mut self, sqe: &InFlight, op: Datasync) -> Fault { + Fault::Visible(Effect::Run(op)) + } + + fn inject_noop_fault(&mut self, sqe: &InFlight) -> Fault<()> { + Fault::Visible(Effect::Run(())) + } +} + +pub struct NoFaults; +impl FaultInjector for NoFaults {} + +/// 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 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 rounder up to the next power of + /// two. + pub cq_capacity: Option, + /// What to do if the completion queue overflows. + pub cq_overflow: OnCqOverflow, +} + +impl Default for Options { + fn default() -> Self { + Self { + capacity: NonZeroUsize::new(8).unwrap(), + cq_capacity: None, + cq_overflow: OnCqOverflow::default(), + } + } +} + +pub struct Executor { + submissions: VecDeque>, + completions: VecDeque>, + + in_flight: Slab>, + executing: VecDeque, + + fstree: BTreeMap, fs::File>, + + cq_overflow: OnCqOverflow, + cq_dropped: usize, +} + +impl Executor { + pub fn new( + Options { + capacity, + cq_capacity, + cq_overflow, + }: Options, + ) -> Self { + let sq_capacity = capacity.get().next_power_of_two(); + let cq_capacity = cq_capacity + .map(|c| c.get().next_power_of_two()) + .unwrap_or_else(|| 2 * sq_capacity); + Self { + submissions: VecDeque::with_capacity(sq_capacity), + completions: VecDeque::with_capacity(cq_capacity), + in_flight: Slab::new(), + executing: VecDeque::new(), + fstree: BTreeMap::new(), + 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 crash(&mut self) { + self.submissions.clear(); + self.completions.clear(); + self.in_flight.clear(); + self.executing.clear(); + self.cq_dropped = 0; + + for file in self.fstree.values_mut() { + file.crash(); + } + } + + /// Restart the executor, simulating a process crash. + /// + /// Unlike [Self::crash], this will drive the currently executing operations + /// to completion. 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. + /// + /// After this method returns, the completion queue is empty. + pub fn restart(&mut self, faults: &mut impl FaultInjector) { + self.submissions.clear(); + let cq_overflow_orig = self.cq_overflow; + self.cq_overflow = OnCqOverflow::Drop; + let executing = mem::take(&mut self.executing); + for op in executing { + self.execute(op, faults); + } + self.completions.clear(); + self.cq_overflow = cq_overflow_orig; + self.cq_dropped = 0; + } + + /// 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]. + 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 randomly using `rng`. + /// The operation is subject to `faults`. + pub fn tick(&mut self, rng: &Rng, faults: &mut impl FaultInjector) -> bool { + let mut progress = self.schedule(); + progress |= self.execute_random(rng, faults); + progress + } + + fn schedule(&mut self) -> bool { + let mut progress = false; + + while let Some(sqe) = self.submissions.pop_front() { + // If the sqe is linked, pop the whole chain. + // Links of sqes not submitted in the same batch are ignored. + let mut successors = VecDeque::new(); + if let Some(link) = sqe.link { + let mut link_kind = link; + while let Some(Sqe { inner, link, user_data }) = self.submissions.pop_front() { + successors.push_back(Blocked { + link: link_kind, + sqe: inner, + user_data, + }); + match link { + Some(kind) => link_kind = kind, + None => break, + } + } + } + let slot = self.in_flight.vacant_entry(); + let (in_flight, ops) = sqe.inner.schedule(SqeId(slot.key())); + self.executing.extend(ops); + slot.insert(InFlight { + inner: in_flight, + blocked: successors, + user_data: sqe.user_data, + }); + + progress = true + } + + progress + } + + fn execute_random(&mut self, rng: &Rng, faults: &mut impl FaultInjector) -> bool { + if self.executing.is_empty() { + return false; + } + if let Some(op) = self.executing.remove(rng.index(self.executing.len())) { + if let Some(delay) = self.execute(op, faults) { + self.executing.push_back(delay); + } + true + } else { + false + } + } + + fn execute(&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"); + 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 { + inner: + InFlightInner::Write { + sqe: sqe::Write { fd, buf, .. }, + op_count, + results, + }, + .. + } = self.in_flight.get_mut(sqe.key()).expect("invalid sqe id") + else { + unreachable!("invalid sqe: expected write") + }; + let mut run = |WriteSector { + page_offset, + buf_offset, + }| { + let bytes = buf.as_bytes(); + let end = (buf_offset + SECTOR_SIZE).min(bytes.len()); + + let buf = &buf.as_bytes()[buf_offset..end]; + fd.write_page(buf, page_offset as _).map_err(Into::into) + }; + results.push(eff.traverse(run, Err)); + + results.len() == *op_count + }; + + if is_complete { + let InFlight { + inner: + InFlightInner::Write { + sqe: sqe::Write { mut buf, .. }, + op_count, + results, + }, + blocked, + user_data, + } = self.in_flight.remove(sqe.key()) + else { + unreachable!("invalid sqe: expected write") + }; + assert!(results.len() == op_count); + let bytes_written = results.iter().filter(|r| r.is_ok()).count() * SECTOR_SIZE; + // TODO: Propagate all errors? + let result = match results.into_iter().find_map(Result::err) { + Some(error) => Err(error), + None => Ok(bytes_written), + }; + let is_success = result.is_ok(); + self.complete(Cqe::Write { result, user_data }); + self.schedule_linked(sqe, is_success, blocked); + } + } + + fn execute_read_sector(&mut self, sqe: SqeId, eff: EitherOrBoth) { + let is_complete = { + let InFlight { + inner: + InFlightInner::Read { + sqe: sqe::Read { fd, buf, .. }, + op_count, + results, + }, + .. + } = self.in_flight.get_mut(sqe.key()).expect("invalid sqe id") + else { + unreachable!("invalid sqe: expected read") + }; + let mut run = |ReadSector { + page_offset, + buf_offset, + }| { + let bytes = buf.as_bytes_mut(); + let end = (buf_offset + SECTOR_SIZE).min(bytes.len()); + + let buf = &mut buf.as_bytes_mut()[buf_offset..end]; + fd.read_page(buf, page_offset as _).map_err(Into::into) + }; + results.push(eff.traverse(run, Err)); + + results.len() == *op_count + }; + + if is_complete { + let InFlight { + inner: + InFlightInner::Read { + sqe: sqe::Read { mut buf, .. }, + op_count, + results, + }, + blocked, + user_data, + } = self.in_flight.remove(sqe.key()) + else { + unreachable!("invalid sqe: expected read") + }; + assert!(results.len() == op_count); + let bytes_read = results.iter().filter(|r| r.is_ok()).count() * SECTOR_SIZE; + // TODO: Propagate all errors? + let result = match results.into_iter().find_map(Result::err) { + Some(error) => Err(error), + None => Ok(bytes_read), + }; + let is_success = result.is_ok(); + self.complete(Cqe::Read { result, user_data }); + self.schedule_linked(sqe, is_success, blocked); + } + } + + fn execute_open(&mut self, sqe: SqeId, eff: EitherOrBoth<(), Error>) { + let InFlight { + inner: InFlightInner::Open { + sqe: sqe::Open { path }, + }, + blocked, + user_data, + } = self.in_flight.remove(sqe.key()) + else { + unreachable!("invalid sqe: expected open") + }; + let result = eff.traverse( + |()| self.fstree.get(&path).cloned().ok_or(Error::FileNotFound { path }), + Err, + ); + + let is_success = result.is_ok(); + self.complete(Cqe::Open { result, user_data }); + self.schedule_linked(sqe, is_success, blocked); + } + + fn execute_create(&mut self, sqe: SqeId, eff: EitherOrBoth<(), Error>) { + let InFlight { + inner: InFlightInner::Create { + sqe: sqe::Create { path }, + }, + blocked, + user_data, + } = self.in_flight.remove(sqe.key()) + else { + unreachable!("invalid sqe: expected create") + }; + let run = |()| match self.fstree.entry(path) { + btree_map::Entry::Vacant(entry) => Ok(entry.insert(fs::File::new()).clone()), + btree_map::Entry::Occupied(entry) => Err(Error::FileAlreadyExists { + path: entry.key().clone(), + }), + }; + let result = eff.traverse(run, Err); + let is_success = result.is_ok(); + self.complete(Cqe::Create { result, user_data }); + self.schedule_linked(sqe, is_success, blocked); + } + + fn execute_stat(&mut self, sqe: SqeId, eff: EitherOrBoth<(), Error>) { + let InFlight { + inner: InFlightInner::Stat { sqe: sqe::Stat { fd } }, + blocked, + 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::Stat { result, user_data }); + self.schedule_linked(sqe, is_success, blocked); + } + + fn execute_fallocate(&mut self, sqe: SqeId, eff: EitherOrBoth<(), Error>) { + let InFlight { + inner: InFlightInner::Fallocate { + sqe: sqe::Fallocate { fd, total_len }, + }, + blocked, + user_data, + } = self.in_flight.remove(sqe.key()) + else { + unreachable!("invalid sqe: expected fallocate") + }; + let result = eff.traverse(|()| fd.set_len(total_len).map_err(Into::into), Err); + let is_success = result.is_ok(); + self.complete(Cqe::Fallocate { result, user_data }); + self.schedule_linked(sqe, is_success, blocked); + } + + fn execute_fsync(&mut self, sqe: SqeId, eff: EitherOrBoth) { + let is_complete = { + let InFlight { + inner: + InFlightInner::Fsync { + sqe: sqe::Fsync { fd }, + op_count, + 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.fdatasync([effect]); + Ok(()) + }, + Err, + ); + results.push(result); + + results.len() == *op_count + }; + + if is_complete { + let InFlight { + inner: InFlightInner::Fsync { results, .. }, + blocked, + user_data, + } = self.in_flight.remove(sqe.key()) + else { + unreachable!("invalid sqe: expected fsync") + }; + // TODO: Propagate all errors? + let result = results.into_iter().find_map(Result::err).map(Err).unwrap_or(Ok(())); + let is_success = result.is_ok(); + self.complete(Cqe::Fsync { result, user_data }); + self.schedule_linked(sqe, is_success, blocked); + } + } + + fn execute_fdatasync(&mut self, sqe: SqeId, eff: EitherOrBoth) { + let is_complete = { + let InFlight { + inner: + InFlightInner::Fdatasync { + sqe: sqe::Fdatasync { fd }, + op_count, + results, + }, + .. + } = self.in_flight.get_mut(sqe.key()).expect("invalid sqe id") + else { + unreachable!("invalid sqe: expected fdatasync") + }; + let result = eff.traverse( + |effect| { + fd.fdatasync([effect]); + Ok(()) + }, + Err, + ); + results.push(result); + + results.len() == *op_count + }; + + if is_complete { + let InFlight { + inner: InFlightInner::Fdatasync { results, .. }, + blocked, + user_data, + } = self.in_flight.remove(sqe.key()) + else { + unreachable!("invalid sqe: expected fdatasync") + }; + // TODO: Propagate all errors? + let result = results.into_iter().find_map(Result::err).map(Err).unwrap_or(Ok(())); + let is_success = result.is_ok(); + self.complete(Cqe::Fdatasync { result, user_data }); + self.schedule_linked(sqe, is_success, blocked); + } + } + + fn execute_noop(&mut self, sqe: SqeId, eff: EitherOrBoth<(), Error>) { + let InFlight { + inner: InFlightInner::Noop, + blocked, + 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::Noop { result, user_data }); + self.schedule_linked(sqe, is_success, blocked); + } + + fn schedule_linked(&mut self, sqe: SqeId, prev_succeeded: bool, mut blocked: VecDeque>) { + if let Some(Blocked { + link, + sqe: next, + user_data, + }) = blocked.pop_front() + { + match (link, prev_succeeded) { + (LinkKind::Soft, false) => { + self.complete(next.cancel(user_data)); + for Blocked { + link: _, + sqe: next, + user_data, + } in blocked + { + self.complete(next.cancel(user_data)); + } + } + (LinkKind::Soft, true) | (LinkKind::Hard, _) => { + let (inner, ops) = next.schedule(sqe); + self.executing.extend(ops); + let slot = self.in_flight.get_mut(sqe.key()).expect("invalid sqe id"); + *slot = InFlight { + inner, + blocked, + user_data, + }; + } + } + } + } +} diff --git a/crates/runtime-core/src/sim/io/executor/sqe.rs b/crates/runtime-core/src/sim/io/executor/sqe.rs new file mode 100644 index 00000000000..5b702b4e227 --- /dev/null +++ b/crates/runtime-core/src/sim/io/executor/sqe.rs @@ -0,0 +1,391 @@ +use alloc::{boxed::Box, vec::Vec}; + +use crate::{ + io::{ErasedBoxPtr, SECTOR_SIZE}, + sim::io::{ + executor::{Cqe, Executing, FsyncEffect, InFlightInner, Operation, ReadSector, WriteSector}, + fs::{self, Datasync}, + Error, + }, +}; + +/// 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(super) inner: SqeInner, + pub(super) link: Option, + pub(super) user_data: Option, +} + +impl Sqe { + 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: ErasedBoxPtr, offset: u64) -> Self { + Write { fd, buf, offset }.into() + } + + pub fn read(fd: fs::File, buf: ErasedBoxPtr, offset: u64) -> Self { + Read { fd, buf, offset }.into() + } + + pub fn open(path: impl AsRef) -> Self { + Open { + path: path.as_ref().into(), + } + .into() + } + + pub fn create(path: impl AsRef) -> Self { + Create { + path: path.as_ref().into(), + } + .into() + } + + pub fn stat(fd: fs::File) -> Self { + Stat { fd }.into() + } + + pub fn fallocate(fd: fs::File, len: u64) -> Self { + Fallocate { fd, total_len: len }.into() + } + + pub fn fsync(fd: fs::File) -> Self { + Fsync { fd }.into() + } + + pub fn fdatasync(fd: fs::File) -> Self { + Fdatasync { fd }.into() + } + + pub fn noop() -> Self { + SqeInner::Noop.into() + } +} + +impl> From for Sqe { + fn from(inner: U) -> Self { + Self { + inner: inner.into(), + link: None, + user_data: None, + } + } +} + +pub enum SqeInner { + Write(Write), + Read(Read), + Open(Open), + Create(Create), + Stat(Stat), + Fallocate(Fallocate), + Fsync(Fsync), + Fdatasync(Fdatasync), + Noop, +} + +impl SqeInner { + pub(super) fn cancel(self, user_data: Option) -> Cqe { + match self { + SqeInner::Write(..) => Cqe::Write { + result: Err(Error::Cancelled), + user_data, + }, + SqeInner::Read(..) => Cqe::Read { + result: Err(Error::Cancelled), + user_data, + }, + SqeInner::Open(..) => Cqe::Open { + result: Err(Error::Cancelled), + user_data, + }, + SqeInner::Create(..) => Cqe::Create { + result: Err(Error::Cancelled), + user_data, + }, + SqeInner::Stat(..) => Cqe::Stat { + result: Err(Error::Cancelled), + user_data, + }, + SqeInner::Fallocate(..) => Cqe::Fallocate { + result: Err(Error::Cancelled), + user_data, + }, + SqeInner::Fsync(..) => Cqe::Fsync { + result: Err(Error::Cancelled), + user_data, + }, + SqeInner::Fdatasync(..) => Cqe::Fdatasync { + result: Err(Error::Cancelled), + user_data, + }, + SqeInner::Noop => Cqe::Noop { + result: Err(Error::Cancelled), + user_data, + }, + } + } + + pub(super) fn schedule(self, sqe_id: SqeId) -> (InFlightInner, Vec) { + match self { + SqeInner::Write(mut sqe) => { + let Write { buf, offset, .. } = &mut sqe; + let buf_len = buf.as_bytes().len(); + let first_sector = (*offset / SECTOR_SIZE as u64) as usize; + let page_count = buf_len / SECTOR_SIZE; + + let ops = (0..page_count) + .map(|page| Executing { + sqe: sqe_id, + inner: Operation::WriteSector(WriteSector { + page_offset: first_sector + page, + buf_offset: *offset as usize + (page * SECTOR_SIZE), + }), + }) + .collect::>(); + let op_count = ops.len(); + let write = InFlightInner::Write { + sqe, + op_count, + results: Vec::with_capacity(op_count), + }; + + (write, ops) + } + SqeInner::Read(mut sqe) => { + let Read { buf, offset, .. } = &mut sqe; + let buf_len = buf.as_bytes().len(); + let first_sector = (*offset / SECTOR_SIZE as u64) as usize; + let page_count = buf_len / SECTOR_SIZE; + + let ops = (0..page_count) + .map(|page| Executing { + sqe: sqe_id, + inner: Operation::ReadSector(ReadSector { + page_offset: first_sector + page, + buf_offset: *offset as usize + (page * SECTOR_SIZE), + }), + }) + .collect::>(); + let op_count = ops.len(); + let read = InFlightInner::Read { + sqe, + op_count, + results: Vec::with_capacity(op_count), + }; + + (read, ops) + } + SqeInner::Open(sqe) => ( + InFlightInner::Open { sqe }, + alloc::vec![Executing { + sqe: sqe_id, + inner: Operation::Open + }], + ), + SqeInner::Create(sqe) => ( + InFlightInner::Create { sqe }, + alloc::vec![Executing { + sqe: sqe_id, + inner: Operation::Create + }], + ), + SqeInner::Stat(sqe) => ( + InFlightInner::Stat { sqe }, + alloc::vec![Executing { + sqe: sqe_id, + inner: Operation::Stat + }], + ), + SqeInner::Fallocate(sqe) => ( + InFlightInner::Fallocate { sqe }, + alloc::vec![Executing { + sqe: sqe_id, + inner: Operation::Fallocate + }], + ), + SqeInner::Fsync(sqe) => { + let Fsync { fd } = &sqe; + + let sector_count = fd.len() / SECTOR_SIZE as u64; + let ops = (0..sector_count) + .map(|offset| Executing { + sqe: sqe_id, + inner: Operation::Fsync { + effect: FsyncEffect::Datasync(Datasync::Sector(offset)), + }, + }) + .chain([Executing { + sqe: sqe_id, + inner: Operation::Fsync { + effect: FsyncEffect::Datasync(Datasync::Length), + }, + }]) + .collect::>(); + let op_count = ops.len(); + let in_flight = InFlightInner::Fsync { + sqe, + op_count, + results: Vec::with_capacity(op_count), + }; + + (in_flight, ops) + } + SqeInner::Fdatasync(sqe) => { + let Fdatasync { fd } = &sqe; + + let sector_count = fd.len() / SECTOR_SIZE as u64; + let ops = (0..sector_count) + .map(|offset| Executing { + sqe: sqe_id, + inner: Operation::Fdatasync { + effect: Datasync::Sector(offset), + }, + }) + .chain([Executing { + sqe: sqe_id, + inner: Operation::Fdatasync { + effect: Datasync::Length, + }, + }]) + .collect::>(); + let op_count = ops.len(); + let in_flight = InFlightInner::Fdatasync { + sqe, + op_count, + results: Vec::with_capacity(op_count), + }; + + (in_flight, ops) + } + SqeInner::Noop => ( + InFlightInner::Noop, + alloc::vec![Executing { + sqe: sqe_id, + inner: Operation::Noop + }], + ), + } + } +} + +impl From for SqeInner { + fn from(inner: Write) -> Self { + Self::Write(inner) + } +} + +impl From for SqeInner { + fn from(inner: Read) -> Self { + Self::Read(inner) + } +} + +impl From for SqeInner { + fn from(inner: Open) -> Self { + Self::Open(inner) + } +} + +impl From for SqeInner { + fn from(inner: Create) -> Self { + Self::Create(inner) + } +} + +impl From for SqeInner { + fn from(inner: Stat) -> Self { + Self::Stat(inner) + } +} + +impl From for SqeInner { + fn from(inner: Fallocate) -> Self { + Self::Fallocate(inner) + } +} + +impl From for SqeInner { + fn from(inner: Fsync) -> Self { + Self::Fsync(inner) + } +} + +impl From for SqeInner { + fn from(inner: Fdatasync) -> Self { + Self::Fdatasync(inner) + } +} + +pub struct Write { + pub fd: fs::File, + pub buf: ErasedBoxPtr, + pub offset: u64, +} + +pub struct Read { + pub fd: fs::File, + pub buf: ErasedBoxPtr, + pub offset: u64, +} + +pub struct Open { + pub path: Box, +} + +pub struct Create { + pub path: Box, +} + +pub struct Stat { + pub fd: fs::File, +} + +pub struct Fallocate { + pub fd: fs::File, + pub total_len: u64, +} + +pub struct Fsync { + pub fd: fs::File, +} + +pub struct Fdatasync { + pub fd: fs::File, +} diff --git a/crates/runtime-core/src/sim/io/fs.rs b/crates/runtime-core/src/sim/io/fs.rs new file mode 100644 index 00000000000..fd793300e34 --- /dev/null +++ b/crates/runtime-core/src/sim/io/fs.rs @@ -0,0 +1,246 @@ +use alloc::{collections::BTreeMap, sync::Arc}; +use core::{ + fmt, + sync::atomic::{AtomicU64, Ordering}, +}; + +pub const PAGE_SIZE: usize = 4096; +const PAGE_SIZE_U64: u64 = PAGE_SIZE as u64; + +#[derive(Clone, Copy, Debug, Eq, PartialEq, thiserror::Error)] +pub enum Error { + #[error("unaligned offset")] + UnalignedOffset, + #[error("unaligned buffer")] + UnalignedBuffer, + #[error("offset overflow")] + OffsetOverflow, +} + +pub type Result = core::result::Result; + +#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)] +struct PageIndex(u64); + +impl PageIndex { + fn from_offset(offset: u64) -> Self { + assert!(offset.is_multiple_of(PAGE_SIZE_U64)); + Self(offset / PAGE_SIZE_U64) + } +} + +struct Page { + bytes: spin::Mutex<[u8; PAGE_SIZE]>, +} + +impl Page { + fn zeroed() -> Self { + Self { + bytes: spin::Mutex::new([0; PAGE_SIZE]), + } + } +} + +#[derive(Default)] +struct PageMap { + volatile: BTreeMap>, + durable: BTreeMap>, +} + +impl PageMap { + /// Reset the volatile to the durable state. + fn crash(&mut self) { + self.volatile = self.durable.clone(); + } + + /// Move the page at `index` from the volatile to the durable state. + fn sync(&mut self, index: PageIndex) { + self.durable.insert(index, self.volatile.get(&index).cloned().unwrap()); + } + + /// Get the page at `index` for reading. Uses the durable state. + fn get_page(&self, index: PageIndex) -> Option> { + self.durable.get(&index).cloned() + } + + /// Get the page at `index` for writing, or allocate a new page. + /// Uses the volatile state. + fn get_or_allocate_page(&mut self, index: PageIndex) -> Arc { + Arc::clone(self.volatile.entry(index).or_insert_with(|| Arc::new(Page::zeroed()))) + } + + /// Change the allocated space, allocating or deallocating pages as needed. + /// Changes the volatile state only. + fn set_len_volatile(&mut self, new_len: u64) { + Self::set_len(&mut self.volatile, new_len); + } + + /// Like [Self::set_len], but operate on the durable state only. + fn set_len_durable(&mut self, new_len: u64) { + Self::set_len(&mut self.durable, new_len); + } + + fn set_len(page_map: &mut BTreeMap>, new_len: u64) { + use core::cmp::Ordering::*; + + let old_len = page_map.len() as u64; + match new_len.cmp(&old_len) { + Equal => {} + Greater => { + let first_new_page = old_len / PAGE_SIZE_U64; + let end_page = new_len / PAGE_SIZE_U64; + + for index in first_new_page..end_page { + page_map + .entry(PageIndex(index)) + .or_insert_with(|| Arc::new(Page::zeroed())); + } + } + Less => { + let first_removed = PageIndex::from_offset(new_len); + let removed = page_map.split_off(&first_removed); + drop(removed); + } + } + } +} + +#[derive(Clone, Copy)] +pub enum Datasync { + Sector(u64), + Length, +} + +/// A memory-backed file. +/// +/// A [File] is backed by a sparse array of [Page]s. Missing pages are read as +/// zeroes. +/// +/// Read and write operations must be page-aligned. Only full pages can be read +/// or written. Writing a page is atomic. +#[derive(Clone)] +pub struct File { + pages: Arc>, + + volatile_len: Arc, + durable_len: Arc, +} + +impl fmt::Debug for File { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("File") + .field("volatile_len", &self.volatile_len) + .field("durable_len", &self.durable_len) + .finish() + } +} + +impl File { + pub(super) fn new() -> Self { + Self { + pages: <_>::default(), + volatile_len: <_>::default(), + durable_len: <_>::default(), + } + } + + /// Simulate a crash by resetting to the durable state. + pub(super) fn crash(&self) { + self.volatile_len + .store(self.durable_len.load(Ordering::Relaxed), Ordering::Relaxed); + self.pages.lock().crash(); + } + + pub(super) fn len(&self) -> u64 { + self.volatile_len.load(Ordering::Relaxed) + } + + #[allow(unused)] + pub(super) fn is_empty(&self) -> bool { + self.len() == 0 + } + + /// Change the file length. + /// + /// The new length must be page-aligned. + /// + /// Extending allocates pages eagerly as needed. Shrinking drops all pages + /// at or beyond the new EOF. + pub(super) fn set_len(&self, new_len: u64) -> Result<()> { + if !new_len.is_multiple_of(PAGE_SIZE_U64) { + return Err(Error::UnalignedOffset); + } + self.pages.lock().set_len_volatile(new_len); + self.volatile_len.store(new_len, Ordering::Relaxed); + + Ok(()) + } + + /// Read one complete page. + pub(super) fn read_page(&self, dst: &mut [u8], index: u64) -> Result<()> { + if dst.len() != PAGE_SIZE { + return Err(Error::UnalignedBuffer); + } + + match self.get_page(PageIndex(index)) { + Some(page) => { + dst.copy_from_slice(&*page.bytes.lock()); + } + None => { + dst.fill(0); + } + } + + Ok(()) + } + + /// Write one complete page. + pub(super) fn write_page(&self, src: &[u8], index: u64) -> Result<()> { + if src.len() != PAGE_SIZE { + return Err(Error::UnalignedBuffer); + } + + let page = self.get_or_allocate_page(PageIndex(index)); + page.bytes.lock().copy_from_slice(src); + + let end = index + .checked_add(1) + .and_then(|pages| pages.checked_mul(PAGE_SIZE_U64)) + .ok_or(Error::OffsetOverflow)?; + + self.volatile_len.fetch_max(end, Ordering::Relaxed); + + Ok(()) + } + + /// Execute an `fdatasync(2)` operation as a series of [Datasync] effects. + /// + /// The result may or may not leave the durable state in the same state as + /// the volatile state at the time the operation started. + /// + /// It is the caller's responsibility to decide whether the operation is + /// considered successful - a partial operation may report success, or a + /// complete operation may report failure. + pub(super) fn fdatasync(&self, ops: impl IntoIterator) { + for op in ops { + match op { + Datasync::Sector(offset) => { + self.pages.lock().sync(PageIndex(offset)); + } + Datasync::Length => { + let new_durable_len = self.volatile_len.load(Ordering::Relaxed); + self.durable_len.store(new_durable_len, Ordering::Relaxed); + self.pages.lock().set_len_durable(new_durable_len); + } + } + } + } + + fn get_page(&self, index: PageIndex) -> Option> { + self.pages.lock().get_page(index) + } + + fn get_or_allocate_page(&self, index: PageIndex) -> Arc { + self.pages.lock().get_or_allocate_page(index) + } +} diff --git a/crates/runtime-core/src/sim/io/mod.rs b/crates/runtime-core/src/sim/io/mod.rs new file mode 100644 index 00000000000..ce01d559fcb --- /dev/null +++ b/crates/runtime-core/src/sim/io/mod.rs @@ -0,0 +1,467 @@ +use alloc::{boxed::Box, sync::Arc}; +use core::{ + pin::Pin, + result::Result, + task::{Context, Poll}, + time::Duration, +}; +use futures_channel::oneshot; +use slab::Slab; + +use crate::{ + io::{AlignedBytes, ErasedBox, ErrorWith, SpacetimeIO, Statx}, + sim::{io::executor::FaultInjector, Rng}, +}; + +mod executor; +use executor::{Cqe, Executor, Sqe}; + +mod fs; +pub use fs::File; + +pub use crate::io::SECTOR_SIZE; + +/// Simulated clock measurement. +/// +/// In simulated time, an instant is actually a [Duration] since the time +/// instance was instantiated. To avoid confusion, we use the name "instant" to +/// convey that its semantics are that of the standard library type of the same +/// name. +pub type Instant = Duration; + +#[derive(Debug, thiserror::Error)] +pub enum Error { + #[error("file not found")] + FileNotFound { path: Box }, + #[error("file already exists")] + FileAlreadyExists { path: Box }, + #[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, +} + +impl From for Error { + fn from(e: fs::Error) -> Self { + Self::Fs(e) + } +} + +#[derive(Clone, Default)] +pub struct SimulatorIO { + inner: Arc, +} + +impl SimulatorIO { + pub fn tick(&self, rng: &Rng, faults: &mut impl FaultInjector) -> bool { + let mut executor = self.inner.executor.lock(); + let mut pending = self.inner.pending.lock(); + let mut buffers = self.inner.buffers.lock(); + + let mut progress = executor.tick(rng, faults); + for cqe in executor.completed() { + let completion = pending.remove(cqe.user_data().unwrap()); + match cqe { + Cqe::Write { result, .. } => { + let CompletionHandle::Write { tx, buf_key } = completion else { + unreachable!("invalid cqe / completion pairing") + }; + let erased_buf = buffers.remove(buf_key); + let result = match result { + Ok(written) if written == erased_buf.len() => Ok(erased_buf), + Ok(written) => Err(ErrorWith { + error: Error::ShortWrite { + expected: erased_buf.len(), + written, + }, + with: erased_buf, + }), + Err(error) => Err(ErrorWith { + error, + with: erased_buf, + }), + }; + let _ = tx.send(result); + } + Cqe::Read { result, .. } => { + let CompletionHandle::Read { tx, buf_key } = completion else { + unreachable!("invalid cqe / completion pairing") + }; + let erased_buf = buffers.remove(buf_key); + let result = match result { + Ok(read) if read == erased_buf.len() => Ok(erased_buf), + Ok(read) => Err(ErrorWith { + error: Error::UnexpectedEof { + expected: erased_buf.len(), + read, + }, + with: erased_buf, + }), + Err(error) => Err(ErrorWith { + error, + with: erased_buf, + }), + }; + let _ = tx.send(result); + } + Cqe::Open { result, .. } => { + let CompletionHandle::Open { tx } = completion else { + unreachable!("invalid cqe / completion pairing") + }; + let _ = tx.send(result); + } + Cqe::Create { result, .. } => { + let CompletionHandle::Create { tx } = completion else { + unreachable!("invalid cqe / completion pairing") + }; + let _ = tx.send(result); + } + Cqe::Stat { result, .. } => { + let CompletionHandle::Stat { tx } = completion else { + unreachable!("invalid cqe / completion pairing") + }; + let _ = tx.send(result); + } + Cqe::Fallocate { result, .. } => { + let CompletionHandle::Fallocate { tx } = completion else { + unreachable!("invalid cqe / completion pairing") + }; + let _ = tx.send(result); + } + Cqe::Fsync { result, .. } => { + let CompletionHandle::Fsync { tx } = completion else { + unreachable!("invalid cqe / completion pairing") + }; + let _ = tx.send(result); + } + Cqe::Fdatasync { result, .. } => { + let CompletionHandle::Fdatasync { tx } = completion else { + unreachable!("invalid cqe / completion pairing") + }; + let _ = tx.send(result); + } + Cqe::Noop { result, .. } => { + let CompletionHandle::Noop { tx } = completion else { + unreachable!("invalid cqe / completion pairing") + }; + let _ = tx.send(result); + } + } + + progress |= true; + } + + progress + } + + fn submit( + &self, + sqe: Sqe, + completion_handle: impl FnOnce(CompletionSender) -> CompletionHandle, + ) -> Completion> { + let (tx, rx) = oneshot::channel(); + + let mut executor = self.inner.executor.lock(); + let mut pending = self.inner.pending.lock(); + let pending_entry = pending.vacant_entry(); + + match executor.submit([sqe.attach(pending_entry.key())]) { + Err(_sqe) => tx + .send(Err(Error::SubmissionQueueOverflow)) + .unwrap_or_else(|_| unreachable!("rx is alive")), + Ok(()) => { + pending_entry.insert(completion_handle(tx)); + } + } + + rx.into() + } + + fn submit_with( + &self, + sqe: Sqe, + buf: ErasedBox, + completion_handle: impl FnOnce(CompletionSender>, usize) -> CompletionHandle, + ) -> Completion>> { + let (tx, rx) = oneshot::channel(); + + let mut executor = self.inner.executor.lock(); + let mut pending = self.inner.pending.lock(); + let pending_entry = pending.vacant_entry(); + + match executor.submit([sqe.attach(pending_entry.key())]) { + Err(_sqe) => tx + .send(Err(ErrorWith { + error: Error::SubmissionQueueOverflow, + with: buf, + })) + .unwrap_or_else(|_| unreachable!("rx is alive")), + Ok(()) => { + let buf_key = self.inner.buffers.lock().insert(buf); + pending_entry.insert(completion_handle(tx, buf_key)); + } + } + + Completion::mapped(rx, reify) + } +} + +struct SimulatorInner { + executor: spin::Mutex>, + pending: spin::Mutex>, + buffers: Arc>>, +} + +impl Default for SimulatorInner { + fn default() -> Self { + Self { + executor: spin::Mutex::new(Executor::new(<_>::default())), + pending: <_>::default(), + buffers: <_>::default(), + } + } +} + +pub type CompletionReceiver = oneshot::Receiver>; + +#[must_use = "completions must be polled to completion"] +pub struct Completion(CompletionInner); + +impl Completion { + pub fn mapped( + rx: CompletionReceiver>, + map: fn(Result>) -> T, + ) -> Self { + Self(CompletionInner::Mapped { rx, map }) + } +} + +impl From> for Completion { + fn from(rx: oneshot::Receiver) -> Self { + Self(CompletionInner::Direct { rx }) + } +} + +impl Future for Completion { + type Output = T; + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + let this = self.get_mut(); + Pin::new(&mut this.0).poll(cx) + } +} + +enum CompletionInner { + Direct { + rx: oneshot::Receiver, + }, + Mapped { + rx: CompletionReceiver>, + map: fn(Result>) -> T, + }, +} + +impl Future for CompletionInner { + type Output = T; + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + let this = self.get_mut(); + match this { + Self::Direct { rx } => Pin::new(rx) + .poll(cx) + .map(|result| result.expect("lost completion sender")), + Self::Mapped { rx, map } => Pin::new(rx).poll(cx).map(|result| { + let result = result.expect("lost completion sender"); + map(result) + }), + } + } +} + +type CompletionSender = oneshot::Sender>; +enum CompletionHandle { + Write { + tx: CompletionSender>, + buf_key: usize, + }, + Read { + tx: CompletionSender>, + buf_key: usize, + }, + Open { + tx: CompletionSender, + }, + Create { + tx: CompletionSender, + }, + Stat { + tx: CompletionSender, + }, + Fallocate { + tx: CompletionSender<(), Error>, + }, + Fsync { + tx: CompletionSender<(), Error>, + }, + Fdatasync { + tx: CompletionSender<(), Error>, + }, + // TODO: We may use this for timeouts. + #[allow(unused)] + Noop { + tx: CompletionSender<(), Error>, + }, +} + +impl SpacetimeIO for SimulatorIO { + type Fd = fs::File; + type Error = Error; + type Completion = Completion; + + fn open_file(&self, path: &str) -> Self::Completion> { + self.submit(Sqe::open(path), |tx| CompletionHandle::Open { tx }) + } + + fn create_file(&self, path: &str) -> Self::Completion> { + self.submit(Sqe::create(path), |tx| CompletionHandle::Create { tx }) + } + + fn write_all_at( + &self, + fd: Self::Fd, + buf: B, + offset: u64, + ) -> Self::Completion>> { + let erased_buf = ErasedBox::from_aligned(buf); + let buf_ptr = erased_buf.as_ptr(); + self.submit_with(Sqe::write(fd, buf_ptr, offset), erased_buf, |tx, buf_key| { + CompletionHandle::Write { tx, buf_key } + }) + } + + fn read_exact_at( + &self, + fd: Self::Fd, + buf: B, + offset: u64, + ) -> Self::Completion>> { + let erased_buf = ErasedBox::from_aligned(buf); + let buf_ptr = erased_buf.as_ptr(); + self.submit_with(Sqe::read(fd, buf_ptr, offset), erased_buf, |tx, buf_key| { + CompletionHandle::Read { tx, buf_key } + }) + } + + fn fsync(&self, fd: Self::Fd) -> Self::Completion> { + self.submit(Sqe::fsync(fd), |tx| CompletionHandle::Fsync { tx }) + } + + fn fdatasync(&self, fd: Self::Fd) -> Self::Completion> { + self.submit(Sqe::fdatasync(fd), |tx| CompletionHandle::Fdatasync { tx }) + } + + fn reserve(&self, fd: Self::Fd, total_size: u64) -> Self::Completion> { + self.submit(Sqe::fallocate(fd, total_size), |tx| CompletionHandle::Fallocate { tx }) + } + + fn statx(&self, fd: Self::Fd) -> Self::Completion> { + self.submit(Sqe::stat(fd), |tx| CompletionHandle::Stat { tx }) + } +} + +fn reify( + result: Result>, +) -> Result> { + match result { + Ok(erased) => Ok(erased.into_aligned::()), + Err(ErrorWith { error, with }) => Err(ErrorWith { + error, + with: with.into_aligned::(), + }), + } +} + +#[cfg(test)] +mod tests { + use crate::sim::{io::executor::NoFaults, GlobalRng}; + + use super::*; + + 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: GlobalRng::new(0), + } + } + + fn run(&self, f: impl FnOnce(&SimulatorIO) -> Completion) -> T { + let fut = self.rt.spawn_local(f(&self.io)); + while self.io.tick(&self.rng, &mut NoFaults) {} + self.rt.block_on(fut).unwrap() + } + } + + #[test] + fn create_file() { + let rt = Runtime::new(); + rt.run(|io| io.create_file("/data/test")).unwrap(); + } + + #[derive(Debug)] + #[repr(C, align(4096))] + struct Buf([u8; 2 * SECTOR_SIZE]); + + 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(), 2 * SECTOR_SIZE); + let mut buf = [0; 2 * SECTOR_SIZE]; + 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")).unwrap(); + let mut buf = rt + .run(|io| io.write_all_at(fd.clone(), Buf([22; 2 * SECTOR_SIZE]), 0)) + .unwrap(); + buf.clear(); + let buf = rt.run(|io| io.read_exact_at(fd, buf, 0)).unwrap(); + + assert!(buf.0.iter().all(|&b| b == 22)); + } +} diff --git a/crates/runtime-core/src/sim/mod.rs b/crates/runtime-core/src/sim/mod.rs index e2c231828a1..1a5a53a29bf 100644 --- a/crates/runtime-core/src/sim/mod.rs +++ b/crates/runtime-core/src/sim/mod.rs @@ -1,5 +1,6 @@ pub mod buggify; mod executor; +pub mod io; mod rng; pub mod time; diff --git a/crates/runtime/Cargo.toml b/crates/runtime/Cargo.toml index c8affea0f48..d23741ce139 100644 --- a/crates/runtime/Cargo.toml +++ b/crates/runtime/Cargo.toml @@ -11,11 +11,17 @@ workspace = true [dependencies] tokio.workspace = true -spacetimedb-runtime-core = { workspace = true, optional = true } -libc = { version = "0.2", optional = true } +spacetimedb-runtime-core = { 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..c49f9f690b6 --- /dev/null +++ b/crates/runtime/src/io/tokio.rs @@ -0,0 +1,254 @@ +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_core::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!("I/O task unexpectedly cancelled"); + } 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: &str) -> 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: &str) -> 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: B, + offset: u64, + ) -> Self::Completion>> { + 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, + mut buf: B, + offset: u64, + ) -> Self::Completion>> { + self.rt + .spawn_blocking(move || 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, }