From d7f1a6b7c5c56bdd7f4d1e3db7b3e1d58825221a Mon Sep 17 00:00:00 2001 From: Kim Altintop Date: Mon, 3 Aug 2026 10:34:34 +0200 Subject: [PATCH 1/9] WIP: I/O API + simulator --- Cargo.lock | 4 + crates/runtime-core/Cargo.toml | 4 +- crates/runtime-core/src/io/mod.rs | 148 ++++++++++++ crates/runtime-core/src/lib.rs | 2 + crates/runtime-core/src/sim/io/fs.rs | 154 ++++++++++++ crates/runtime-core/src/sim/io/mod.rs | 261 +++++++++++++++++++++ crates/runtime-core/src/sim/io/op.rs | 322 ++++++++++++++++++++++++++ crates/runtime-core/src/sim/mod.rs | 1 + crates/runtime/Cargo.toml | 4 + crates/runtime/src/io.rs | 2 + crates/runtime/src/io/tokio.rs | 163 +++++++++++++ crates/runtime/src/lib.rs | 2 + 12 files changed, 1066 insertions(+), 1 deletion(-) create mode 100644 crates/runtime-core/src/io/mod.rs create mode 100644 crates/runtime-core/src/sim/io/fs.rs create mode 100644 crates/runtime-core/src/sim/io/mod.rs create mode 100644 crates/runtime-core/src/sim/io/op.rs create mode 100644 crates/runtime/src/io.rs create mode 100644 crates/runtime/src/io/tokio.rs diff --git a/Cargo.lock b/Cargo.lock index ef0f375ecec..3b085e5511e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8556,7 +8556,9 @@ dependencies = [ "futures", "libc", "spacetimedb-runtime-core", + "static_assertions", "tokio", + "windows-sys 0.61.2", ] [[package]] @@ -8564,7 +8566,9 @@ name = "spacetimedb-runtime-core" version = "2.7.1" dependencies = [ "async-task", + "futures-channel", "spin", + "zerocopy", ] [[package]] diff --git a/crates/runtime-core/Cargo.toml b/crates/runtime-core/Cargo.toml index a3369a69f89..6ac037162dd 100644 --- a/crates/runtime-core/Cargo.toml +++ b/crates/runtime-core/Cargo.toml @@ -11,8 +11,10 @@ workspace = true [features] default = [] -sim = ["dep:async-task", "dep:spin"] +sim = ["dep:async-task", "dep:futures-channel", "dep:spin"] [dependencies] async-task = { version = "4.4", default-features = false, optional = true } +futures-channel = { version = "0.3", default-features = false, features = ["alloc"], optional = true } spin = { version = "0.9", default-features = false, features = ["mutex", "spin_mutex"], optional = true } +zerocopy = "0.8" diff --git a/crates/runtime-core/src/io/mod.rs b/crates/runtime-core/src/io/mod.rs new file mode 100644 index 00000000000..b4ff3744916 --- /dev/null +++ b/crates/runtime-core/src/io/mod.rs @@ -0,0 +1,148 @@ +use zerocopy::{FromBytes, Immutable, IntoBytes, KnownLayout}; + +/// Size in bytes of a disk sector. +pub const SECTOR_SIZE: usize = 4096; + +/// Types that can be safely converted to and from sector-aligned byte slices. +pub trait AlignedBytes: Sized { + /// Assert that the type' size is a multiple of [SECTOR_SIZE] and has the + /// right aligment. + /// + /// NOTE: Associated constants are evaluated lazily -- add a free + /// + /// `const _: () = ::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)); + }; + + /// 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() + } +} + +/// An error `E`, along with auxiliary data `T`. +/// +/// `T` is usually a buffer of type [AlignedBytes], whose ownership is +/// transferred back to the caller when an error occurs. +/// +/// As this type signifies an error condition, the contents of `T` are +/// unspecified. +#[derive(Debug)] +pub struct ErrorWith { + pub error: E, + pub with: T, +} + +/// The canonical, low-level I/O API. +/// +/// Currently only supports file I/O, but eventually all I/O performed by +/// SpacetimeDB should go through this trait. +/// +/// Intended to support implementations based on `io-uring`, which means that +/// buffer ownership is transferred to the I/O engine while reading or writing. +/// +/// Implementations should be `!Send`, i.e. all I/O happens on a single thread. +/// +/// File operations should never be mutually exclusive, and therefore expose a +/// `pwrite`/`pread`-style API. It is assumed that direct I/O (`O_DIRECT`) is +/// used, i.e. the kernel page cache is bypassed. The [AlignedBytes] type +/// ensures that the alignment requirements for direct I/O are met. +pub trait SpacetimeIO { + /// An open file handle. + /// + /// Like [std::fs::File], the file shall be closed when the last reference + /// to the handle is dropped. + /// + /// Unlike [std::fs::File], the file handle must be clone-able. + type Fd: Clone; + /// The error returned by methods of this trait. + /// + /// This should always be instantiated to [std::io::Error]. However, pending + /// [alloc_io], this type is not in `core`, which would prevent this crate + /// from being `no_std`. + /// + /// [alloc_io]: https://github.com/rust-lang/rust/issues/154046 + type Error; + + /// Open the file at `path`. + fn open_file(&self, path: &str) -> impl Future>; + + /// Create the file at `path` and allocate `len` bytes. + /// + /// Returns an error if the file already exists. + fn create_file(&self, path: &str, len: u64) -> impl Future>; + + /// 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, + ) -> impl Future>>; + + /// 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, + ) -> impl Future>>; + + /// Call `fsync(2)` on `fd`. + fn fsync(&self, fd: Self::Fd) -> impl Future>; + /// Call `fdatasync(2)` on `fd`. + fn fdatasync(&self, fd: Self::Fd) -> impl Future>; + + /// Allocate `additional` bytes for the file `fd`. + fn reserve(&self, fd: Self::Fd, additional: u64) -> impl Future>; +} diff --git a/crates/runtime-core/src/lib.rs b/crates/runtime-core/src/lib.rs index f7590ada98b..e35d042ea9a 100644 --- a/crates/runtime-core/src/lib.rs +++ b/crates/runtime-core/src/lib.rs @@ -7,3 +7,5 @@ extern crate std; #[cfg(feature = "sim")] pub mod sim; + +pub mod io; 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..9c540f5e1b3 --- /dev/null +++ b/crates/runtime-core/src/sim/io/fs.rs @@ -0,0 +1,154 @@ +use alloc::{collections::BTreeMap, rc::Rc}; +use core::{ + cell::{Cell, RefCell}, + cmp, +}; + +pub const PAGE_SIZE: usize = 4096; +const PAGE_SIZE_U64: u64 = PAGE_SIZE as u64; + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum Error { + UnalignedOffset, + UnalignedBuffer, + 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: RefCell<[u8; PAGE_SIZE]>, +} + +impl Page { + fn zeroed() -> Self { + Self { + bytes: RefCell::new([0; PAGE_SIZE]), + } + } +} + +/// 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: RefCell>>, + len: Cell, +} + +impl File { + pub(super) const fn new() -> Self { + Self { + pages: RefCell::new(BTreeMap::new()), + len: Cell::new(0), + } + } + + pub(super) const fn len(&self) -> u64 { + self.len.get() + } + + #[allow(unused)] + pub(super) const fn is_empty(&self) -> bool { + self.len.get() == 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 fn set_len(&self, new_len: u64) -> Result<()> { + use cmp::Ordering::*; + + if !new_len.is_multiple_of(PAGE_SIZE_U64) { + return Err(Error::UnalignedOffset); + } + let old_len = self.len.get(); + + 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 { + self.get_or_allocate_page(PageIndex(index)); + } + + self.len.set(new_len); + } + Less => { + self.len.set(new_len); + + let first_removed = PageIndex::from_offset(new_len); + let removed = self.pages.borrow_mut().split_off(&first_removed); + drop(removed); + } + } + + Ok(()) + } + + /// Read one complete page. + pub 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.borrow()); + } + None => { + dst.fill(0); + } + } + + Ok(()) + } + + /// Write one complete page. + pub 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.borrow_mut().copy_from_slice(src); + + let end = index + .checked_add(1) + .and_then(|pages| pages.checked_mul(PAGE_SIZE_U64)) + .ok_or(Error::OffsetOverflow)?; + + self.len.set(cmp::max(self.len.get(), end)); + + Ok(()) + } + + fn get_page(&self, index: PageIndex) -> Option> { + self.pages.borrow().get(&index).cloned() + } + + fn get_or_allocate_page(&self, index: PageIndex) -> Rc { + let mut pages = self.pages.borrow_mut(); + Rc::clone(pages.entry(index).or_insert_with(|| Rc::new(Page::zeroed()))) + } +} 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..670b68d7aa3 --- /dev/null +++ b/crates/runtime-core/src/sim/io/mod.rs @@ -0,0 +1,261 @@ +use alloc::{ + boxed::Box, + collections::{BTreeMap, VecDeque}, + rc::Rc, +}; +use core::{ + cell::RefCell, + future::{poll_fn, Future}, + pin::Pin, + result::Result, + task::Poll, +}; +use futures_channel::oneshot; + +use crate::io::{AlignedBytes, ErrorWith, SpacetimeIO}; + +mod fs; +mod op; + +pub use crate::io::SECTOR_SIZE; +pub use fs::File; + +#[derive(Debug)] +pub enum Error { + FileNotFound { path: Box }, + FileAlreadyExists { path: Box }, + ShortWrite { expected: usize, written: usize }, + UnexpectedEof { expected: usize, read: usize }, + Fs(fs::Error), +} + +impl From for Error { + fn from(e: fs::Error) -> Self { + Self::Fs(e) + } +} + +#[derive(Default)] +pub struct SimulatorIO { + inner: Rc>, +} + +impl SimulatorIO { + pub fn tick(&self) { + self.inner.borrow_mut().tick(); + } + + fn submit(&self, op: impl FnOnce(oneshot::Sender) -> Box) -> oneshot::Receiver { + let (tx, rx) = oneshot::channel(); + self.inner.borrow_mut().submit(op(tx)); + rx + } + + // TODO: The sim runtime should be advancing I/O. Until it does, `tick()` + // whenever a result future is polled and returns pending. + async fn wait_for(&self, mut rx: oneshot::Receiver) -> Result { + poll_fn(|cx| match Pin::new(&mut rx).poll(cx) { + Poll::Ready(result) => Poll::Ready(result), + Poll::Pending => { + self.tick(); + cx.waker().wake_by_ref(); + Poll::Pending + } + }) + .await + } + + async fn submit_and_wait( + &self, + op: impl FnOnce(oneshot::Sender) -> Box, + ) -> Result { + let rx = self.submit(op); + self.wait_for(rx).await + } +} + +impl SpacetimeIO for SimulatorIO { + type Fd = fs::File; + type Error = Error; + + async fn open_file(&self, path: &str) -> Result { + self.submit_and_wait(|tx| op::open_file(path, tx)) + .await + .expect("`open_file` future cancelled") + } + + async fn create_file(&self, path: &str, len: u64) -> Result { + self.submit_and_wait(|tx| op::create_file(path, len, tx)) + .await + .expect("`create_file` future cancelled") + } + + async fn write_all_at( + &self, + fd: Self::Fd, + buf: B, + offset: u64, + ) -> Result> { + let () = B::ASSERT_VALID_LAYOUT; + + if !offset.is_multiple_of(SECTOR_SIZE as _) { + self.submit_and_wait(|tx| { + op::ready( + Err(ErrorWith { + error: fs::Error::UnalignedOffset.into(), + with: buf, + }), + tx, + ) + }) + .await + .expect("`write_all_at` future cancelled") + } else { + let (tx, rx) = oneshot::channel(); + for op in op::write_at(fd, buf, offset, tx) { + self.inner.borrow_mut().submit(op); + } + self.wait_for(rx).await.expect("`write_all_at` future cancelled") + } + } + + async fn read_exact_at( + &self, + fd: Self::Fd, + buf: B, + offset: u64, + ) -> Result> { + let () = B::ASSERT_VALID_LAYOUT; + + if !offset.is_multiple_of(SECTOR_SIZE as _) { + self.submit_and_wait(|tx| { + op::ready( + Err(ErrorWith { + error: fs::Error::UnalignedOffset.into(), + with: buf, + }), + tx, + ) + }) + .await + .expect("`read_exact_at` future cancelled") + } else { + let (tx, rx) = oneshot::channel(); + for op in op::read_at(fd, buf, offset, tx) { + self.inner.borrow_mut().submit(op); + } + self.wait_for(rx).await.expect("`read_exact_at` future cancelled") + } + } + + async fn fsync(&self, _fd: Self::Fd) -> Result<(), Self::Error> { + Ok(()) + } + + async fn fdatasync(&self, _fd: Self::Fd) -> Result<(), Self::Error> { + Ok(()) + } + + async fn reserve(&self, fd: Self::Fd, additional: u64) -> Result<(), Self::Error> { + let len = self + .submit_and_wait(|tx| op::get_len(fd.clone(), tx)) + .await + .expect("`get_len` future cancelled")?; + self.submit_and_wait(|tx| op::set_len(fd, len + additional, tx)) + .await + .expect("`set_len` future cancelled") + } +} + +#[derive(Default)] +struct SimulatorIOInner { + files: BTreeMap, fs::File>, + submissions: VecDeque>, + completions: VecDeque>, +} + +impl SimulatorIOInner { + fn tick(&mut self) { + if let Some(sqe) = self.submissions.pop_front() { + sqe.execute(&mut self.files, &mut self.completions); + } + if let Some(cqe) = self.completions.pop_front() { + cqe.complete(); + } + } + + fn submit(&mut self, op: Box) { + self.submissions.push_back(op); + } +} + +trait Submission { + fn execute( + self: Box, + files: &mut BTreeMap, fs::File>, + completions: &mut VecDeque>, + ); +} + +trait Completion { + fn complete(self: Box); +} + +#[cfg(test)] +mod tests { + use crate::sim::Runtime; + + use super::*; + + #[test] + fn create_file() { + let mut rt = Runtime::new(1); + let io = SimulatorIO::default(); + + let fd = rt + .block_on(io.create_file("/data/test", 2 * SECTOR_SIZE as u64)) + .unwrap(); + assert_eq!(fd.len(), 2 * SECTOR_SIZE as u64); + } + + #[repr(C, align(4096))] + struct Buf([u8; 2 * SECTOR_SIZE]); + + 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 mut rt = Runtime::new(1); + let io = SimulatorIO::default(); + + let fd = rt + .block_on(io.create_file("/data/test", 2 * SECTOR_SIZE as u64)) + .unwrap(); + let mut buf = rt + .block_on(io.write_all_at(fd.clone(), Buf([22; 2 * SECTOR_SIZE]), 0)) + .map_err(|ErrorWith { error, .. }| error) + .unwrap(); + buf.0.fill(0); + let buf = rt + .block_on(io.read_exact_at(fd, buf, 0)) + .map_err(|ErrorWith { error, .. }| error) + .unwrap(); + + assert!(buf.0.iter().all(|&b| b == 22)); + } +} diff --git a/crates/runtime-core/src/sim/io/op.rs b/crates/runtime-core/src/sim/io/op.rs new file mode 100644 index 00000000000..921129d2536 --- /dev/null +++ b/crates/runtime-core/src/sim/io/op.rs @@ -0,0 +1,322 @@ +use alloc::{ + boxed::Box, + collections::{btree_map, BTreeMap, VecDeque}, + rc::Rc, +}; +use core::cell::RefCell; +use futures_channel::oneshot; + +use super::{fs, Completion, Error, Submission}; +use crate::io::{AlignedBytes, ErrorWith, SECTOR_SIZE}; + +pub type WriteAtResult = Result>; +pub type ReadAtResult = Result>; + +pub fn write_at( + fd: fs::File, + buf: B, + offset: u64, + notify: oneshot::Sender>, +) -> impl Iterator> { + let first_page = (offset / SECTOR_SIZE as u64) as usize; + let page_count = buf.as_bytes().len() / SECTOR_SIZE; + + let state = Rc::new(RefCell::new(PagedOpState { + buf: Some(buf), + notify: Some(notify), + remaining: page_count, + first_error: None, + })); + + (0..page_count).map(move |buf_page| { + let op = WritePage { + fd: fd.clone(), + file_page: first_page + buf_page, + buf_page, + state: state.clone(), + }; + + Box::new(op) as Box + }) +} + +pub fn read_at( + fd: fs::File, + buf: B, + offset: u64, + notify: oneshot::Sender>, +) -> impl Iterator> { + let first_page = (offset / SECTOR_SIZE as u64) as usize; + let page_count = buf.as_bytes().len() / SECTOR_SIZE; + + let state = Rc::new(RefCell::new(PagedOpState { + buf: Some(buf), + notify: Some(notify), + remaining: page_count, + first_error: None, + })); + + (0..page_count).map(move |buf_page| { + let op = ReadPage { + fd: fd.clone(), + file_page: first_page + buf_page, + buf_page, + state: state.clone(), + }; + + Box::new(op) as Box + }) +} + +pub fn open_file(path: &str, notify: oneshot::Sender>) -> Box { + Box::new(OpenFile { + path: path.into(), + notify, + }) +} + +pub fn create_file(path: &str, len: u64, notify: oneshot::Sender>) -> Box { + Box::new(CreateFile { + path: path.into(), + len, + notify, + }) +} + +pub fn get_len(fd: fs::File, notify: oneshot::Sender>) -> Box { + Box::new(GetLen { fd, notify }) +} + +pub fn set_len(fd: fs::File, len: u64, notify: oneshot::Sender>) -> Box { + Box::new(SetLen { fd, len, notify }) +} + +struct GenericCompletion { + result: T, + notify: oneshot::Sender, +} + +fn completion(result: T, notify: oneshot::Sender) -> Box { + Box::new(GenericCompletion { result, notify }) +} + +impl Completion for GenericCompletion { + fn complete(self: Box) { + let Self { result, notify } = *self; + let _ = notify.send(result); + } +} + +struct Ready(Box); + +impl Submission for Ready { + fn execute( + self: Box, + _files: &mut BTreeMap, fs::File>, + completions: &mut VecDeque>, + ) { + let Self(completion) = *self; + completions.push_back(completion); + } +} + +pub fn ready(result: T, notify: oneshot::Sender) -> Box { + Box::new(Ready(completion(result, notify))) +} + +struct OpenFile { + path: Box, + notify: oneshot::Sender>, +} + +impl Submission for OpenFile { + fn execute( + self: Box, + files: &mut BTreeMap, fs::File>, + completions: &mut VecDeque>, + ) { + let Self { path, notify } = *self; + let result = files.get(&path).cloned().ok_or(Error::FileNotFound { path }); + completions.push_back(completion(result, notify)); + } +} + +struct CreateFile { + path: Box, + len: u64, + notify: oneshot::Sender>, +} + +impl Submission for CreateFile { + fn execute( + self: Box, + files: &mut BTreeMap, fs::File>, + completions: &mut VecDeque>, + ) { + let Self { path, len, notify } = *self; + let result = (|| { + let file = match files.entry(path.clone()) { + btree_map::Entry::Vacant(entry) => Ok(entry.insert(fs::File::new()).clone()), + btree_map::Entry::Occupied(_) => Err(Error::FileAlreadyExists { path }), + }?; + file.set_len(len)?; + Ok(file) + })(); + completions.push_back(completion(result, notify)); + } +} + +struct GetLen { + fd: fs::File, + notify: oneshot::Sender>, +} + +impl Submission for GetLen { + fn execute( + self: Box, + _files: &mut BTreeMap, fs::File>, + completions: &mut VecDeque>, + ) { + let Self { fd, notify } = *self; + let result = Ok(fd.len()); + completions.push_back(completion(result, notify)); + } +} + +struct SetLen { + fd: fs::File, + len: u64, + notify: oneshot::Sender>, +} + +impl Submission for SetLen { + fn execute( + self: Box, + _files: &mut BTreeMap, fs::File>, + completions: &mut VecDeque>, + ) { + let Self { fd, len, notify } = *self; + let result = fd.set_len(len).map_err(Error::from); + completions.push_back(completion(result, notify)); + } +} + +struct PagedOpState { + buf: Option, + notify: Option>>>, + remaining: usize, + first_error: Option, +} + +fn complete_page_op( + state: &Rc>>, + result: Result<(), fs::Error>, + completions: &mut VecDeque>, +) { + let complete = { + let mut state = state.borrow_mut(); + if let Err(e) = result + && state.first_error.is_none() + { + state.first_error.replace(e.into()); + } + assert!(state.remaining > 0); + state.remaining -= 1; + + state.remaining == 0 + }; + + if complete { + completions.push_back(Box::new(WriteCompletion { state: state.clone() })); + } +} + +struct WriteCompletion { + state: Rc>>, +} + +impl Completion for WriteCompletion { + fn complete(self: Box) { + let (notify, result) = { + let mut state = self.state.borrow_mut(); + + assert_eq!(state.remaining, 0); + + let buf = state.buf.take().expect("write completed more than once"); + let notify = state.notify.take().expect("write completed more than once"); + + let result = match state.first_error.take() { + None => Ok(buf), + Some(error) => Err(ErrorWith { error, with: buf }), + }; + + (notify, result) + }; + + let _ = notify.send(result); + } +} + +struct WritePage { + fd: fs::File, + file_page: usize, + buf_page: usize, + state: Rc>>, +} + +impl Submission for WritePage { + fn execute( + self: Box, + _files: &mut BTreeMap, fs::File>, + completions: &mut VecDeque>, + ) { + let Self { + fd, + file_page, + buf_page, + state, + } = *self; + + let result = { + let state_ref = state.borrow(); + let buf = state_ref.buf.as_ref().expect("buffer went away"); + + let start = buf_page * SECTOR_SIZE; + let end = start + SECTOR_SIZE; + fd.write_page(&buf.as_bytes()[start..end], file_page as _) + }; + complete_page_op(&state, result, completions); + } +} + +struct ReadPage { + fd: fs::File, + file_page: usize, + buf_page: usize, + state: Rc>>, +} + +impl Submission for ReadPage { + fn execute( + self: Box, + _files: &mut BTreeMap, fs::File>, + completions: &mut VecDeque>, + ) { + let Self { + fd, + file_page, + buf_page, + state, + } = *self; + + let result = { + let mut state_ref = state.borrow_mut(); + let buf = state_ref.buf.as_mut().expect("buffer went away"); + + let start = buf_page * SECTOR_SIZE; + let end = start + SECTOR_SIZE; + fd.read_page(&mut buf.as_bytes_mut()[start..end], file_page as _) + }; + complete_page_op(&state, result, completions); + } +} 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..2b2cffc5317 100644 --- a/crates/runtime/Cargo.toml +++ b/crates/runtime/Cargo.toml @@ -13,6 +13,10 @@ workspace = true tokio.workspace = true spacetimedb-runtime-core = { workspace = true, optional = true } libc = { version = "0.2", optional = true } +static_assertions = "1.1" + +[target.'cfg(windows)'.dependencies] +windows-sys = { version = "0.61", features = ["Win32_Storage_FileSystem"] } [dev-dependencies] futures.workspace = true 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..eb948ed7e1c --- /dev/null +++ b/crates/runtime/src/io/tokio.rs @@ -0,0 +1,163 @@ +use std::{io, marker::PhantomData, rc::Rc, sync::Arc}; + +#[cfg(unix)] +use std::os::unix::fs::FileExt as _; +#[cfg(windows)] +use std::os::windows::fs::FileExt as _; + +use spacetimedb_runtime_core::io::{AlignedBytes, ErrorWith, SpacetimeIO}; +use static_assertions::assert_not_impl_any; +use tokio::fs::OpenOptions; +use tokio::{runtime, task::spawn_blocking}; + +/// Implementation of [SpacetimeIO] that runs on a tokio runtime. +pub struct TokioIO { + // TODO: Should this be [runtime::Runtime]? + 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); + +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; + + async fn open_file(&self, path: &str) -> Result { + let _rt = self.rt.enter(); + + let mut open_options = tokio::fs::File::options(); + open_options.read(true).write(true); + let file = open_with_direct_io(open_options, path).await?; + + Ok(Arc::new(file.into_std().await)) + } + + async fn create_file(&self, path: &str, len: u64) -> Result { + let _rt = self.rt.enter(); + + let mut open_options = tokio::fs::File::options(); + open_options.read(true).write(true).create_new(true); + let file = open_with_direct_io(open_options, path).await?; + file.set_len(len).await?; + + Ok(Arc::new(file.into_std().await)) + } + + async fn write_all_at( + &self, + fd: Self::Fd, + buf: B, + offset: u64, + ) -> Result> { + let _rt = self.rt.enter(); + asyncify(move || { + #[cfg(unix)] + let res = fd.write_all_at(buf.as_bytes(), offset); + #[cfg(windows)] + let res = fd.seek_write(buf.as_bytes(), offset); + + match res { + Ok(()) => Ok(buf), + Err(error) => Err(ErrorWith { error, with: buf }), + } + }) + .await + } + + async fn read_exact_at( + &self, + fd: Self::Fd, + mut buf: B, + offset: u64, + ) -> Result> { + let _rt = self.rt.enter(); + asyncify(move || { + #[cfg(unix)] + let res = fd.read_exact_at(buf.as_bytes_mut(), offset); + #[cfg(windows)] + let res = fd.seek_read(buf.as_bytes_mut(), offset); + + match res { + Ok(()) => Ok(buf), + Err(error) => Err(ErrorWith { error, with: buf }), + } + }) + .await + } + + async fn fsync(&self, fd: Self::Fd) -> Result<(), Self::Error> { + let _rt = self.rt.enter(); + asyncify(move || fd.sync_all()).await + } + + async fn fdatasync(&self, fd: Self::Fd) -> Result<(), Self::Error> { + let _rt = self.rt.enter(); + asyncify(move || fd.sync_data()).await + } + + async fn reserve(&self, fd: Self::Fd, additional: u64) -> Result<(), Self::Error> { + let _rt = self.rt.enter(); + asyncify(move || { + let len = fd.metadata()?.len(); + fd.set_len(len + additional)?; + + Ok(()) + }) + .await + } +} + +async fn asyncify(f: F) -> R +where + F: FnOnce() -> R + Send + 'static, + R: Send + 'static, +{ + spawn_blocking(f).await.unwrap_or_else(|e| match e.try_into_panic() { + Ok(panic_payload) => std::panic::resume_unwind(panic_payload), + // A cancellation should not be possible, because we await the task. + Err(e) => panic!("unexpected error joining blocking task: {e}"), + }) +} + +#[cfg(all(unix, not(target_os = "macos")))] +async fn open_with_direct_io(mut options: OpenOptions, path: &str) -> io::Result { + options.custom_flags(libc::O_DIRECT).open(path).await +} + +#[cfg(target_os = "macos")] +async fn open_with_direct_io(options: OpenOptions, path: &str) -> io::Result { + let file = options.open(path).await?; + asyncify(move || { + 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) + } + }) + .await +} + +#[cfg(target_os = "windows")] +async fn open_with_direct_io(options: OpenOptions, path: &str) -> io::Result { + options + .custom_flags(windows_sys::Win32::Storage::FileSystem::FILE_FLAG_NO_BUFFERING) + .open(path) + .await +} 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, } From 0f320311de8382b825a3028185197efdf2221610 Mon Sep 17 00:00:00 2001 From: Kim Altintop Date: Tue, 4 Aug 2026 15:00:51 +0200 Subject: [PATCH 2/9] Use SimulatorIO as the "I/O driver" in the executor Entails making it Send + Sync, which may or may not be what we want. --- crates/runtime-core/src/sim/executor/mod.rs | 26 +++++- crates/runtime-core/src/sim/io/fs.rs | 50 +++++------ crates/runtime-core/src/sim/io/mod.rs | 92 ++++++++++----------- crates/runtime-core/src/sim/io/op.rs | 41 +++++---- 4 files changed, 114 insertions(+), 95 deletions(-) diff --git a/crates/runtime-core/src/sim/executor/mod.rs b/crates/runtime-core/src/sim/executor/mod.rs index fbb7f7c0cf2..eee88b4a88a 100644 --- a/crates/runtime-core/src/sim/executor/mod.rs +++ b/crates/runtime-core/src/sim/executor/mod.rs @@ -10,6 +10,8 @@ use core::{ use spin::Mutex; +use crate::sim::io::SimulatorIO; + use super::{time::TimeHandle, Rng}; mod task; @@ -21,11 +23,12 @@ type Runnable = async_task::Runnable; #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub struct RuntimeConfig { pub seed: u64, + pub enable_io: bool, } impl RuntimeConfig { pub const fn new(seed: u64) -> Self { - Self { seed } + Self { seed, enable_io: false } } } @@ -145,6 +148,12 @@ impl Runtime { } } + // TODO: This is a stopgap to allow submission of I/O tasks. We probably + // want the user-facing API to hide this. + pub fn io(&self) -> &Option { + &self.executor.io + } + /// Drive a top-level future to completion on the simulation executor. /// /// While the future runs, spawned tasks share the same deterministic @@ -360,6 +369,7 @@ struct Executor { next_node: AtomicU64, rng: Rng, time: TimeHandle, + io: Option, } impl Executor { @@ -375,6 +385,7 @@ impl Executor { next_node: AtomicU64::new(1), rng: Rng::new(config.seed), time: TimeHandle::new(), + io: config.enable_io.then(SimulatorIO::default), } } @@ -499,6 +510,10 @@ impl Executor { }; } + if self.run_pending_io() { + continue; + } + if self.time.wake_next_timer() { continue; } @@ -527,6 +542,15 @@ impl Executor { } } + fn run_pending_io(&self) -> bool { + // TODO: Inject faults (reorder, delay, drop, ..) when buggify is enabled. + // Also, should this run more than one queue entry? + match &self.io { + Some(io) => io.tick(), + None => false, + } + } + /// Look up the record for a node, panicking if the node is unknown. fn node_record(&self, node: NodeId) -> Arc { self.nodes diff --git a/crates/runtime-core/src/sim/io/fs.rs b/crates/runtime-core/src/sim/io/fs.rs index 9c540f5e1b3..bdc64b87657 100644 --- a/crates/runtime-core/src/sim/io/fs.rs +++ b/crates/runtime-core/src/sim/io/fs.rs @@ -1,7 +1,7 @@ -use alloc::{collections::BTreeMap, rc::Rc}; +use alloc::{collections::BTreeMap, sync::Arc}; use core::{ - cell::{Cell, RefCell}, cmp, + sync::atomic::{AtomicU64, Ordering}, }; pub const PAGE_SIZE: usize = 4096; @@ -27,13 +27,13 @@ impl PageIndex { } struct Page { - bytes: RefCell<[u8; PAGE_SIZE]>, + bytes: spin::Mutex<[u8; PAGE_SIZE]>, } impl Page { fn zeroed() -> Self { Self { - bytes: RefCell::new([0; PAGE_SIZE]), + bytes: spin::Mutex::new([0; PAGE_SIZE]), } } } @@ -47,25 +47,25 @@ impl Page { /// or written. Writing a page is atomic. #[derive(Clone)] pub struct File { - pages: RefCell>>, - len: Cell, + pages: Arc>>>, + len: Arc, } impl File { - pub(super) const fn new() -> Self { + pub(super) fn new() -> Self { Self { - pages: RefCell::new(BTreeMap::new()), - len: Cell::new(0), + pages: Arc::new(spin::Mutex::new(BTreeMap::new())), + len: Arc::new(AtomicU64::new(0)), } } - pub(super) const fn len(&self) -> u64 { - self.len.get() + pub(super) fn len(&self) -> u64 { + self.len.load(Ordering::Relaxed) } #[allow(unused)] - pub(super) const fn is_empty(&self) -> bool { - self.len.get() == 0 + pub(super) fn is_empty(&self) -> bool { + self.len() == 0 } /// Change the file length. @@ -80,7 +80,7 @@ impl File { if !new_len.is_multiple_of(PAGE_SIZE_U64) { return Err(Error::UnalignedOffset); } - let old_len = self.len.get(); + let old_len = self.len(); match new_len.cmp(&old_len) { Equal => {} @@ -92,13 +92,13 @@ impl File { self.get_or_allocate_page(PageIndex(index)); } - self.len.set(new_len); + self.len.store(new_len, Ordering::Relaxed); } Less => { - self.len.set(new_len); + self.len.store(new_len, Ordering::Relaxed); let first_removed = PageIndex::from_offset(new_len); - let removed = self.pages.borrow_mut().split_off(&first_removed); + let removed = self.pages.lock().split_off(&first_removed); drop(removed); } } @@ -114,7 +114,7 @@ impl File { match self.get_page(PageIndex(index)) { Some(page) => { - dst.copy_from_slice(&*page.bytes.borrow()); + dst.copy_from_slice(&*page.bytes.lock()); } None => { dst.fill(0); @@ -131,24 +131,24 @@ impl File { } let page = self.get_or_allocate_page(PageIndex(index)); - page.bytes.borrow_mut().copy_from_slice(src); + 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.len.set(cmp::max(self.len.get(), end)); + self.len.fetch_max(end, Ordering::Relaxed); Ok(()) } - fn get_page(&self, index: PageIndex) -> Option> { - self.pages.borrow().get(&index).cloned() + fn get_page(&self, index: PageIndex) -> Option> { + self.pages.lock().get(&index).cloned() } - fn get_or_allocate_page(&self, index: PageIndex) -> Rc { - let mut pages = self.pages.borrow_mut(); - Rc::clone(pages.entry(index).or_insert_with(|| Rc::new(Page::zeroed()))) + fn get_or_allocate_page(&self, index: PageIndex) -> Arc { + let mut pages = self.pages.lock(); + Arc::clone(pages.entry(index).or_insert_with(|| Arc::new(Page::zeroed()))) } } diff --git a/crates/runtime-core/src/sim/io/mod.rs b/crates/runtime-core/src/sim/io/mod.rs index 670b68d7aa3..6f8f9af8569 100644 --- a/crates/runtime-core/src/sim/io/mod.rs +++ b/crates/runtime-core/src/sim/io/mod.rs @@ -1,15 +1,9 @@ use alloc::{ boxed::Box, collections::{BTreeMap, VecDeque}, - rc::Rc, -}; -use core::{ - cell::RefCell, - future::{poll_fn, Future}, - pin::Pin, - result::Result, - task::Poll, + sync::Arc, }; +use core::result::Result; use futures_channel::oneshot; use crate::io::{AlignedBytes, ErrorWith, SpacetimeIO}; @@ -35,42 +29,23 @@ impl From for Error { } } -#[derive(Default)] +#[derive(Clone, Default)] pub struct SimulatorIO { - inner: Rc>, + inner: Arc>, } impl SimulatorIO { - pub fn tick(&self) { - self.inner.borrow_mut().tick(); - } - - fn submit(&self, op: impl FnOnce(oneshot::Sender) -> Box) -> oneshot::Receiver { - let (tx, rx) = oneshot::channel(); - self.inner.borrow_mut().submit(op(tx)); - rx - } - - // TODO: The sim runtime should be advancing I/O. Until it does, `tick()` - // whenever a result future is polled and returns pending. - async fn wait_for(&self, mut rx: oneshot::Receiver) -> Result { - poll_fn(|cx| match Pin::new(&mut rx).poll(cx) { - Poll::Ready(result) => Poll::Ready(result), - Poll::Pending => { - self.tick(); - cx.waker().wake_by_ref(); - Poll::Pending - } - }) - .await + pub fn tick(&self) -> bool { + self.inner.lock().tick() } async fn submit_and_wait( &self, op: impl FnOnce(oneshot::Sender) -> Box, ) -> Result { - let rx = self.submit(op); - self.wait_for(rx).await + let (tx, rx) = oneshot::channel(); + self.inner.lock().submit(op(tx)); + rx.await } } @@ -90,7 +65,7 @@ impl SpacetimeIO for SimulatorIO { .expect("`create_file` future cancelled") } - async fn write_all_at( + async fn write_all_at( &self, fd: Self::Fd, buf: B, @@ -113,13 +88,13 @@ impl SpacetimeIO for SimulatorIO { } else { let (tx, rx) = oneshot::channel(); for op in op::write_at(fd, buf, offset, tx) { - self.inner.borrow_mut().submit(op); + self.inner.lock().submit(op); } - self.wait_for(rx).await.expect("`write_all_at` future cancelled") + rx.await.expect("`write_all_at` future cancelled") } } - async fn read_exact_at( + async fn read_exact_at( &self, fd: Self::Fd, buf: B, @@ -142,9 +117,9 @@ impl SpacetimeIO for SimulatorIO { } else { let (tx, rx) = oneshot::channel(); for op in op::read_at(fd, buf, offset, tx) { - self.inner.borrow_mut().submit(op); + self.inner.lock().submit(op); } - self.wait_for(rx).await.expect("`read_exact_at` future cancelled") + rx.await.expect("`read_exact_at` future cancelled") } } @@ -175,13 +150,28 @@ struct SimulatorIOInner { } impl SimulatorIOInner { - fn tick(&mut self) { + // TODO: Allow runtime to inject faults via: + // + // - pick random entries from the submission queue + // - drop queue entries + // - delay `execute` (somehow) + // - delay `complete` + // - make a submission fail without performing its effect + // - execute an arbitrary number of (random) SQEs + // - complete an arbitrary number of CQEs + + fn tick(&mut self) -> bool { + let mut progress = false; if let Some(sqe) = self.submissions.pop_front() { sqe.execute(&mut self.files, &mut self.completions); + progress = true; } if let Some(cqe) = self.completions.pop_front() { cqe.complete(); + progress = true; } + + progress } fn submit(&mut self, op: Box) { @@ -189,7 +179,7 @@ impl SimulatorIOInner { } } -trait Submission { +trait Submission: Send { fn execute( self: Box, files: &mut BTreeMap, fs::File>, @@ -197,20 +187,23 @@ trait Submission { ); } -trait Completion { +trait Completion: Send { fn complete(self: Box); } #[cfg(test)] mod tests { - use crate::sim::Runtime; + use crate::sim::{Runtime, RuntimeConfig}; use super::*; #[test] fn create_file() { - let mut rt = Runtime::new(1); - let io = SimulatorIO::default(); + let mut rt = Runtime::with_config(RuntimeConfig { + enable_io: true, + ..<_>::default() + }); + let io = rt.io().clone().unwrap(); let fd = rt .block_on(io.create_file("/data/test", 2 * SECTOR_SIZE as u64)) @@ -240,8 +233,11 @@ mod tests { #[test] fn write_read_roundtrip() { - let mut rt = Runtime::new(1); - let io = SimulatorIO::default(); + let mut rt = Runtime::with_config(RuntimeConfig { + enable_io: true, + ..<_>::default() + }); + let io = rt.io().clone().unwrap(); let fd = rt .block_on(io.create_file("/data/test", 2 * SECTOR_SIZE as u64)) diff --git a/crates/runtime-core/src/sim/io/op.rs b/crates/runtime-core/src/sim/io/op.rs index 921129d2536..963f26dc522 100644 --- a/crates/runtime-core/src/sim/io/op.rs +++ b/crates/runtime-core/src/sim/io/op.rs @@ -1,9 +1,8 @@ use alloc::{ boxed::Box, collections::{btree_map, BTreeMap, VecDeque}, - rc::Rc, + sync::Arc, }; -use core::cell::RefCell; use futures_channel::oneshot; use super::{fs, Completion, Error, Submission}; @@ -12,7 +11,7 @@ use crate::io::{AlignedBytes, ErrorWith, SECTOR_SIZE}; pub type WriteAtResult = Result>; pub type ReadAtResult = Result>; -pub fn write_at( +pub fn write_at( fd: fs::File, buf: B, offset: u64, @@ -21,7 +20,7 @@ pub fn write_at( let first_page = (offset / SECTOR_SIZE as u64) as usize; let page_count = buf.as_bytes().len() / SECTOR_SIZE; - let state = Rc::new(RefCell::new(PagedOpState { + let state = Arc::new(spin::Mutex::new(PagedOpState { buf: Some(buf), notify: Some(notify), remaining: page_count, @@ -40,7 +39,7 @@ pub fn write_at( }) } -pub fn read_at( +pub fn read_at( fd: fs::File, buf: B, offset: u64, @@ -49,7 +48,7 @@ pub fn read_at( let first_page = (offset / SECTOR_SIZE as u64) as usize; let page_count = buf.as_bytes().len() / SECTOR_SIZE; - let state = Rc::new(RefCell::new(PagedOpState { + let state = Arc::new(spin::Mutex::new(PagedOpState { buf: Some(buf), notify: Some(notify), remaining: page_count, @@ -96,11 +95,11 @@ struct GenericCompletion { notify: oneshot::Sender, } -fn completion(result: T, notify: oneshot::Sender) -> Box { +fn completion(result: T, notify: oneshot::Sender) -> Box { Box::new(GenericCompletion { result, notify }) } -impl Completion for GenericCompletion { +impl Completion for GenericCompletion { fn complete(self: Box) { let Self { result, notify } = *self; let _ = notify.send(result); @@ -120,7 +119,7 @@ impl Submission for Ready { } } -pub fn ready(result: T, notify: oneshot::Sender) -> Box { +pub fn ready(result: T, notify: oneshot::Sender) -> Box { Box::new(Ready(completion(result, notify))) } @@ -208,13 +207,13 @@ struct PagedOpState { first_error: Option, } -fn complete_page_op( - state: &Rc>>, +fn complete_page_op( + state: &Arc>>, result: Result<(), fs::Error>, completions: &mut VecDeque>, ) { let complete = { - let mut state = state.borrow_mut(); + let mut state = state.lock(); if let Err(e) = result && state.first_error.is_none() { @@ -232,13 +231,13 @@ fn complete_page_op( } struct WriteCompletion { - state: Rc>>, + state: Arc>>, } -impl Completion for WriteCompletion { +impl Completion for WriteCompletion { fn complete(self: Box) { let (notify, result) = { - let mut state = self.state.borrow_mut(); + let mut state = self.state.lock(); assert_eq!(state.remaining, 0); @@ -261,10 +260,10 @@ struct WritePage { fd: fs::File, file_page: usize, buf_page: usize, - state: Rc>>, + state: Arc>>, } -impl Submission for WritePage { +impl Submission for WritePage { fn execute( self: Box, _files: &mut BTreeMap, fs::File>, @@ -278,7 +277,7 @@ impl Submission for WritePage { } = *self; let result = { - let state_ref = state.borrow(); + let state_ref = state.lock(); let buf = state_ref.buf.as_ref().expect("buffer went away"); let start = buf_page * SECTOR_SIZE; @@ -293,10 +292,10 @@ struct ReadPage { fd: fs::File, file_page: usize, buf_page: usize, - state: Rc>>, + state: Arc>>, } -impl Submission for ReadPage { +impl Submission for ReadPage { fn execute( self: Box, _files: &mut BTreeMap, fs::File>, @@ -310,7 +309,7 @@ impl Submission for ReadPage { } = *self; let result = { - let mut state_ref = state.borrow_mut(); + let mut state_ref = state.lock(); let buf = state_ref.buf.as_mut().expect("buffer went away"); let start = buf_page * SECTOR_SIZE; From 98d7d2342a38b7747018a77500ad4d1cdee295bc Mon Sep 17 00:00:00 2001 From: Kim Altintop Date: Tue, 4 Aug 2026 16:29:02 +0200 Subject: [PATCH 3/9] Make it clearer that we're clearing --- crates/runtime-core/src/sim/io/mod.rs | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/crates/runtime-core/src/sim/io/mod.rs b/crates/runtime-core/src/sim/io/mod.rs index 6f8f9af8569..91ac80527b4 100644 --- a/crates/runtime-core/src/sim/io/mod.rs +++ b/crates/runtime-core/src/sim/io/mod.rs @@ -214,6 +214,12 @@ mod tests { #[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 @@ -246,7 +252,7 @@ mod tests { .block_on(io.write_all_at(fd.clone(), Buf([22; 2 * SECTOR_SIZE]), 0)) .map_err(|ErrorWith { error, .. }| error) .unwrap(); - buf.0.fill(0); + buf.clear(); let buf = rt .block_on(io.read_exact_at(fd, buf, 0)) .map_err(|ErrorWith { error, .. }| error) From 16a1fc7c2e433fb65a865a0bb1575096e874b81f Mon Sep 17 00:00:00 2001 From: Kim Altintop Date: Thu, 6 Aug 2026 18:27:58 +0200 Subject: [PATCH 4/9] Expose ways for the runtime to inject failures. --- crates/runtime-core/src/sim/io/mod.rs | 78 +++++++-- crates/runtime-core/src/sim/io/op.rs | 234 ++++++++++++++++---------- 2 files changed, 205 insertions(+), 107 deletions(-) diff --git a/crates/runtime-core/src/sim/io/mod.rs b/crates/runtime-core/src/sim/io/mod.rs index 91ac80527b4..e50c77f558a 100644 --- a/crates/runtime-core/src/sim/io/mod.rs +++ b/crates/runtime-core/src/sim/io/mod.rs @@ -2,14 +2,19 @@ use alloc::{ boxed::Box, collections::{BTreeMap, VecDeque}, sync::Arc, + vec::Vec, }; -use core::result::Result; +use core::{ops::RangeBounds, result::Result}; use futures_channel::oneshot; -use crate::io::{AlignedBytes, ErrorWith, SpacetimeIO}; +use crate::{ + io::{AlignedBytes, ErrorWith, SpacetimeIO}, + sim::Rng, +}; mod fs; -mod op; +pub mod op; +use op::{Completion, Submission}; pub use crate::io::SECTOR_SIZE; pub use fs::File; @@ -31,14 +36,46 @@ impl From for Error { #[derive(Clone, Default)] pub struct SimulatorIO { + // TODO: We make `SimulatorIO` `Send + Sync` for now, because + // [crate::sim::executor::Handle] is just `Arc`. This means that a + // future carrying a handle can't be `spawn`ed, because spawning requires + // the future to be `Send`. + // + // We should fix this at some point, so below can become `Rc>`. inner: Arc>, } impl SimulatorIO { + /// Run the submission at the front of the queue (if any), and complete the + /// completion at the front of the queue (if any). pub fn tick(&self) -> bool { self.inner.lock().tick() } + /// Execute `sqe`. + pub fn execute(&self, sqe: Box) { + self.inner.lock().execute(sqe); + } + + /// Remove and return the submission at the fron of the queue, if any. + pub fn next_submission(&self) -> Option> { + self.inner.lock().next() + } + + /// Remove and return a random submission, or `None` if the queue is empty. + pub fn random_submission(&self, rng: &Rng) -> Option> { + self.inner.lock().next_random(rng) + } + + /// Remove `range` from the completion queue. + pub fn completions(&self, range: impl RangeBounds) -> impl Iterator> { + self.inner + .lock() + .drain_completions(range) + .collect::>() + .into_iter() + } + async fn submit_and_wait( &self, op: impl FnOnce(oneshot::Sender) -> Box, @@ -163,7 +200,9 @@ impl SimulatorIOInner { fn tick(&mut self) -> bool { let mut progress = false; if let Some(sqe) = self.submissions.pop_front() { - sqe.execute(&mut self.files, &mut self.completions); + if let Some(cqe) = sqe.execute(&mut self.files) { + self.completions.push_back(cqe); + } progress = true; } if let Some(cqe) = self.completions.pop_front() { @@ -174,21 +213,28 @@ impl SimulatorIOInner { progress } - fn submit(&mut self, op: Box) { - self.submissions.push_back(op); + fn execute(&mut self, sqe: Box) { + if let Some(cqe) = sqe.execute(&mut self.files) { + self.completions.push_back(cqe); + } } -} -trait Submission: Send { - fn execute( - self: Box, - files: &mut BTreeMap, fs::File>, - completions: &mut VecDeque>, - ); -} + fn next(&mut self) -> Option> { + self.submissions.pop_front() + } + + fn next_random(&mut self, rng: &Rng) -> Option> { + let i = rng.next_u64() % self.submissions.len() as u64; + self.submissions.remove(i as usize) + } + + fn drain_completions(&mut self, range: impl RangeBounds) -> impl Iterator> { + self.completions.drain(range) + } -trait Completion: Send { - fn complete(self: Box); + fn submit(&mut self, op: Box) { + self.submissions.push_back(op); + } } #[cfg(test)] diff --git a/crates/runtime-core/src/sim/io/op.rs b/crates/runtime-core/src/sim/io/op.rs index 963f26dc522..a4e95d71fb6 100644 --- a/crates/runtime-core/src/sim/io/op.rs +++ b/crates/runtime-core/src/sim/io/op.rs @@ -1,28 +1,60 @@ +use core::any::Any; + use alloc::{ boxed::Box, - collections::{btree_map, BTreeMap, VecDeque}, + collections::{btree_map, BTreeMap}, sync::Arc, }; use futures_channel::oneshot; -use super::{fs, Completion, Error, Submission}; +use super::{fs, Error}; use crate::io::{AlignedBytes, ErrorWith, SECTOR_SIZE}; +/// An operation that can be submitted to the [super::SimulatorIO] driver. +pub trait Submission: Send + Any { + /// Run the operations with mutable access to the currently registered + /// [fs::File]s. + /// + /// If the operation is done, a [Completion] is returned in a `Some`. + /// `None` may be returned if: + /// + /// - The submission is a sub-operation, such as [WritePage] or [ReadPage]. + /// - The submission is a [Noop]. + /// + fn execute(self: Box, files: &mut BTreeMap, fs::File>) -> Option>; +} + +/// An object containing the result of executing a [Submission], as well as a +/// handle to resolve a future waiting on the outcome of the operation. +pub trait Completion: Send { + /// Resolve the future waiting on the outcome of the operation. + fn complete(self: Box); +} + +/// A channel to resolve a future waiting on the outcome of a submitted +/// operation. +pub type OnComplete = oneshot::Sender; + pub type WriteAtResult = Result>; -pub type ReadAtResult = Result>; +/// Write the contents of `buf` to `fd` at `offset`. +/// +/// This operation is split into multiple writes to individual pages. The +/// `on_complete` future resolves only after all page writes completed. +/// +/// Ownership of `buf` is transferred back when the operation completes. pub fn write_at( fd: fs::File, buf: B, offset: u64, - notify: oneshot::Sender>, + on_complete: OnComplete>, ) -> impl Iterator> { let first_page = (offset / SECTOR_SIZE as u64) as usize; let page_count = buf.as_bytes().len() / SECTOR_SIZE; let state = Arc::new(spin::Mutex::new(PagedOpState { buf: Some(buf), - notify: Some(notify), + on_complete: Some(on_complete), remaining: page_count, first_error: None, })); @@ -39,18 +71,27 @@ pub fn write_at( }) } +pub type ReadAtResult = Result>; + +/// Fill `buf` by reading from `fd` at `offset`. +/// +/// This operation is split into multple reads from the individual pages needed +/// to fill `buf`. The `on_complete` future resolves only after all page reads +/// completed. +/// +/// Ownership of `buf` is transferred back when the operation completes. pub fn read_at( fd: fs::File, buf: B, offset: u64, - notify: oneshot::Sender>, + on_complete: OnComplete>, ) -> impl Iterator> { let first_page = (offset / SECTOR_SIZE as u64) as usize; let page_count = buf.as_bytes().len() / SECTOR_SIZE; let state = Arc::new(spin::Mutex::new(PagedOpState { buf: Some(buf), - notify: Some(notify), + on_complete: Some(on_complete), remaining: page_count, first_error: None, })); @@ -67,92 +108,107 @@ pub fn read_at( }) } -pub fn open_file(path: &str, notify: oneshot::Sender>) -> Box { +/// Open file at `path`. +pub fn open_file(path: &str, on_complete: OnComplete>) -> Box { Box::new(OpenFile { path: path.into(), - notify, + on_complete, }) } -pub fn create_file(path: &str, len: u64, notify: oneshot::Sender>) -> Box { +/// Create a new file at `path` and allocate `len` space for it. +pub fn create_file(path: &str, len: u64, on_complete: OnComplete>) -> Box { Box::new(CreateFile { path: path.into(), len, - notify, + on_complete, }) } -pub fn get_len(fd: fs::File, notify: oneshot::Sender>) -> Box { - Box::new(GetLen { fd, notify }) +/// Get the length of the file `fd`. +pub fn get_len(fd: fs::File, on_complete: OnComplete>) -> Box { + Box::new(GetLen { fd, on_complete }) } -pub fn set_len(fd: fs::File, len: u64, notify: oneshot::Sender>) -> Box { - Box::new(SetLen { fd, len, notify }) +/// Set the length of the file `fd`. +pub fn set_len(fd: fs::File, len: u64, on_complete: OnComplete>) -> Box { + Box::new(SetLen { fd, len, on_complete }) } struct GenericCompletion { result: T, - notify: oneshot::Sender, + on_complete: OnComplete, } -fn completion(result: T, notify: oneshot::Sender) -> Box { - Box::new(GenericCompletion { result, notify }) +fn completion(result: T, on_complete: OnComplete) -> Box { + Box::new(GenericCompletion { result, on_complete }) } impl Completion for GenericCompletion { fn complete(self: Box) { - let Self { result, notify } = *self; - let _ = notify.send(result); + let Self { + result, on_complete, .. + } = *self; + let _ = on_complete.send(result); } } -struct Ready(Box); +/// [Submission] created by [noop]. +pub(crate) struct Noop; + +impl Submission for Noop { + fn execute(self: Box, _files: &mut BTreeMap, fs::File>) -> Option> { + None + } +} + +/// An operation that does nothing. +/// +/// Note that no completion is associated with a noop, but the submission still +/// occupies a slot in the submission queue. +pub fn noop() -> Box { + Box::new(Noop) +} + +/// [Submission] created by [ready]. +pub(crate) struct Ready(Box); impl Submission for Ready { - fn execute( - self: Box, - _files: &mut BTreeMap, fs::File>, - completions: &mut VecDeque>, - ) { + fn execute(self: Box, _files: &mut BTreeMap, fs::File>) -> Option> { let Self(completion) = *self; - completions.push_back(completion); + Some(completion) } } -pub fn ready(result: T, notify: oneshot::Sender) -> Box { - Box::new(Ready(completion(result, notify))) +/// An operation that is already complete with `result`. +pub fn ready(result: T, on_complete: OnComplete) -> Box { + Box::new(Ready(completion(result, on_complete))) } -struct OpenFile { +/// [Submission] created by [open_file]. +pub(crate) struct OpenFile { path: Box, - notify: oneshot::Sender>, + on_complete: OnComplete>, } impl Submission for OpenFile { - fn execute( - self: Box, - files: &mut BTreeMap, fs::File>, - completions: &mut VecDeque>, - ) { - let Self { path, notify } = *self; + fn execute(self: Box, files: &mut BTreeMap, fs::File>) -> Option> { + let Self { path, on_complete } = *self; let result = files.get(&path).cloned().ok_or(Error::FileNotFound { path }); - completions.push_back(completion(result, notify)); + Some(completion(result, on_complete)) } } -struct CreateFile { +/// [Submission] created by [create_file]. +pub(crate) struct CreateFile { path: Box, len: u64, - notify: oneshot::Sender>, + on_complete: OnComplete>, } impl Submission for CreateFile { - fn execute( - self: Box, - files: &mut BTreeMap, fs::File>, - completions: &mut VecDeque>, - ) { - let Self { path, len, notify } = *self; + fn execute(self: Box, files: &mut BTreeMap, fs::File>) -> Option> { + let Self { path, len, on_complete } = *self; let result = (|| { let file = match files.entry(path.clone()) { btree_map::Entry::Vacant(entry) => Ok(entry.insert(fs::File::new()).clone()), @@ -161,48 +217,42 @@ impl Submission for CreateFile { file.set_len(len)?; Ok(file) })(); - completions.push_back(completion(result, notify)); + Some(completion(result, on_complete)) } } -struct GetLen { +/// [Submission] created by [get_len]. +pub(crate) struct GetLen { fd: fs::File, - notify: oneshot::Sender>, + on_complete: OnComplete>, } impl Submission for GetLen { - fn execute( - self: Box, - _files: &mut BTreeMap, fs::File>, - completions: &mut VecDeque>, - ) { - let Self { fd, notify } = *self; + fn execute(self: Box, _files: &mut BTreeMap, fs::File>) -> Option> { + let Self { fd, on_complete } = *self; let result = Ok(fd.len()); - completions.push_back(completion(result, notify)); + Some(completion(result, on_complete)) } } -struct SetLen { +/// [Submission] created by [set_len]. +pub(crate) struct SetLen { fd: fs::File, len: u64, - notify: oneshot::Sender>, + on_complete: OnComplete>, } impl Submission for SetLen { - fn execute( - self: Box, - _files: &mut BTreeMap, fs::File>, - completions: &mut VecDeque>, - ) { - let Self { fd, len, notify } = *self; + fn execute(self: Box, _files: &mut BTreeMap, fs::File>) -> Option> { + let Self { fd, len, on_complete } = *self; let result = fd.set_len(len).map_err(Error::from); - completions.push_back(completion(result, notify)); + Some(completion(result, on_complete)) } } struct PagedOpState { buf: Option, - notify: Option>>>, + on_complete: Option>>>, remaining: usize, first_error: Option, } @@ -210,8 +260,7 @@ struct PagedOpState { fn complete_page_op( state: &Arc>>, result: Result<(), fs::Error>, - completions: &mut VecDeque>, -) { +) -> Option>> { let complete = { let mut state = state.lock(); if let Err(e) = result @@ -225,38 +274,36 @@ fn complete_page_op( state.remaining == 0 }; - if complete { - completions.push_back(Box::new(WriteCompletion { state: state.clone() })); - } + complete.then(|| Box::new(PageOpCompletion { state: state.clone() })) } -struct WriteCompletion { +struct PageOpCompletion { state: Arc>>, } -impl Completion for WriteCompletion { +impl Completion for PageOpCompletion { fn complete(self: Box) { - let (notify, result) = { + let (on_complete, result) = { let mut state = self.state.lock(); assert_eq!(state.remaining, 0); let buf = state.buf.take().expect("write completed more than once"); - let notify = state.notify.take().expect("write completed more than once"); + let on_complete = state.on_complete.take().expect("write completed more than once"); let result = match state.first_error.take() { None => Ok(buf), Some(error) => Err(ErrorWith { error, with: buf }), }; - (notify, result) + (on_complete, result) }; - let _ = notify.send(result); + let _ = on_complete.send(result); } } -struct WritePage { +pub(crate) struct WritePage { fd: fs::File, file_page: usize, buf_page: usize, @@ -264,11 +311,7 @@ struct WritePage { } impl Submission for WritePage { - fn execute( - self: Box, - _files: &mut BTreeMap, fs::File>, - completions: &mut VecDeque>, - ) { + fn execute(self: Box, _files: &mut BTreeMap, fs::File>) -> Option> { let Self { fd, file_page, @@ -284,11 +327,11 @@ impl Submission for WritePage { let end = start + SECTOR_SIZE; fd.write_page(&buf.as_bytes()[start..end], file_page as _) }; - complete_page_op(&state, result, completions); + complete_page_op(&state, result).map(|c| c as Box) } } -struct ReadPage { +pub(crate) struct ReadPage { fd: fs::File, file_page: usize, buf_page: usize, @@ -296,11 +339,7 @@ struct ReadPage { } impl Submission for ReadPage { - fn execute( - self: Box, - _files: &mut BTreeMap, fs::File>, - completions: &mut VecDeque>, - ) { + fn execute(self: Box, _files: &mut BTreeMap, fs::File>) -> Option> { let Self { fd, file_page, @@ -316,6 +355,19 @@ impl Submission for ReadPage { let end = start + SECTOR_SIZE; fd.read_page(&mut buf.as_bytes_mut()[start..end], file_page as _) }; - complete_page_op(&state, result, completions); + complete_page_op(&state, result).map(|c| c as Box) + } +} + +#[cfg(test)] +mod tests { + use core::any::Any; + + use super::*; + + #[test] + fn downcast() { + let sqe: Box = noop(); + sqe.downcast::().unwrap(); } } From 99638a6f4427bafe528c9cd17288bc0d2750b9f7 Mon Sep 17 00:00:00 2001 From: Kim Altintop Date: Fri, 7 Aug 2026 14:01:11 +0200 Subject: [PATCH 5/9] Encapsulate SimulatorIO in an I/O "driver" that can inject failures. --- crates/runtime-core/src/sim/executor/io.rs | 101 ++++++++++++++++++++ crates/runtime-core/src/sim/executor/mod.rs | 45 +++++---- crates/runtime-core/src/sim/io/fs.rs | 6 +- crates/runtime-core/src/sim/io/mod.rs | 92 +++++++++++------- crates/runtime-core/src/sim/io/op.rs | 75 ++++++++++++++- 5 files changed, 262 insertions(+), 57 deletions(-) create mode 100644 crates/runtime-core/src/sim/executor/io.rs diff --git a/crates/runtime-core/src/sim/executor/io.rs b/crates/runtime-core/src/sim/executor/io.rs new file mode 100644 index 00000000000..ca1831978bf --- /dev/null +++ b/crates/runtime-core/src/sim/executor/io.rs @@ -0,0 +1,101 @@ +use crate::sim::{io::SimulatorIO, Rng}; + +#[derive(Clone, Copy, Debug, PartialEq)] +pub struct Config { + /// The max number of submissions to run per [Driver::tick]. + pub max_submissions_per_tick: usize, + /// The max number of completions to finish per [Driver::tick]. + pub max_completions_per_tick: usize, + /// Submission reordering probability. + /// + /// Describes the probability by which to select the next submission queue + /// entry randomly, as opposed to the oldest entry in the queue. + pub prob_reorder_submissions: f64, + /// Completion reordering probability. + /// + /// Describes the probability by which to select the next completion queue + /// entry randomly, as opposed to the oldest entry in the queue. + pub prob_reorder_completions: f64, + /// Probability by which to skip one submission queue entry. + /// + /// If skipped, the entry still counts towards `max_submissions_per_tick`. + pub prob_skip: f64, + /// Probability by which to cancel a submission queue entry. + /// + /// [crate::sim::io::op::Submission::cancel()] is called on the entry, which + /// may generate a completion. + pub prob_cancel: f64, +} + +impl Default for Config { + fn default() -> Self { + Self { + max_submissions_per_tick: 1, + max_completions_per_tick: 1, + prob_reorder_submissions: 0.0, + prob_reorder_completions: 0.0, + prob_skip: 0.0, + prob_cancel: 0.0, + } + } +} + +pub struct Driver { + io: SimulatorIO, + config: Config, +} + +impl Driver { + pub fn new(config: Config) -> Self { + Self { + io: <_>::default(), + config, + } + } + + /// Advance the I/O simulator according the [Config]. + /// + /// Returns `true` if progress has been made, or there are pending entries + /// in either the submission or completion queue. + pub fn tick(&self, rng: &Rng) -> bool { + let mut progress = false; + for _ in 0..self.config.max_submissions_per_tick { + if !rng.buggify_with_prob(self.config.prob_skip) { + let sqe = if rng.buggify_with_prob(self.config.prob_reorder_submissions) { + self.io.random_submission(rng) + } else { + self.io.next_submission() + }; + + if let Some(sqe) = sqe { + if rng.buggify_with_prob(self.config.prob_cancel) { + sqe.cancel(); + } else { + self.io.execute(sqe); + } + progress = true; + } + } + } + + for _ in 0..self.config.max_completions_per_tick { + let cqe = if rng.buggify_with_prob(self.config.prob_reorder_completions) { + self.io.random_completion(rng) + } else { + self.io.next_completion() + }; + + if let Some(cqe) = cqe { + cqe.complete(); + progress = true + } + } + + progress |= self.io.pending(); + progress + } + + pub fn io(&self) -> &SimulatorIO { + &self.io + } +} diff --git a/crates/runtime-core/src/sim/executor/mod.rs b/crates/runtime-core/src/sim/executor/mod.rs index eee88b4a88a..1913329b2e2 100644 --- a/crates/runtime-core/src/sim/executor/mod.rs +++ b/crates/runtime-core/src/sim/executor/mod.rs @@ -14,21 +14,34 @@ use crate::sim::io::SimulatorIO; use super::{time::TimeHandle, Rng}; +mod io; + mod task; use task::Abortable; pub use task::{AbortHandle, JoinError, JoinHandle}; type Runnable = async_task::Runnable; -#[derive(Clone, Copy, Debug, Eq, PartialEq)] +#[derive(Clone, Copy, Debug, PartialEq)] pub struct RuntimeConfig { pub seed: u64, - pub enable_io: bool, + pub io: Option, } impl RuntimeConfig { pub const fn new(seed: u64) -> Self { - Self { seed, enable_io: false } + Self { seed, io: None } + } + + pub fn enable_io(self) -> Self { + Self { + io: Some(self.io.unwrap_or_default()), + ..self + } + } + + pub fn with_io_config(self, io: Option) -> Self { + Self { io, ..self } } } @@ -150,8 +163,8 @@ impl Runtime { // TODO: This is a stopgap to allow submission of I/O tasks. We probably // want the user-facing API to hide this. - pub fn io(&self) -> &Option { - &self.executor.io + pub fn io(&self) -> Option<&SimulatorIO> { + self.executor.io.as_ref().map(|driver| driver.io()) } /// Drive a top-level future to completion on the simulation executor. @@ -369,7 +382,7 @@ struct Executor { next_node: AtomicU64, rng: Rng, time: TimeHandle, - io: Option, + io: Option, } impl Executor { @@ -385,7 +398,7 @@ impl Executor { next_node: AtomicU64::new(1), rng: Rng::new(config.seed), time: TimeHandle::new(), - io: config.enable_io.then(SimulatorIO::default), + io: config.io.map(io::Driver::new), } } @@ -502,6 +515,7 @@ impl Executor { loop { self.run_all_ready(); + let pending_io = self.drive_io(); if task.is_finished() { let waker = Waker::noop(); return match Pin::new(&mut task).poll(&mut Context::from_waker(waker)) { @@ -510,11 +524,7 @@ impl Executor { }; } - if self.run_pending_io() { - continue; - } - - if self.time.wake_next_timer() { + if self.time.wake_next_timer() || pending_io { continue; } @@ -542,12 +552,11 @@ impl Executor { } } - fn run_pending_io(&self) -> bool { - // TODO: Inject faults (reorder, delay, drop, ..) when buggify is enabled. - // Also, should this run more than one queue entry? - match &self.io { - Some(io) => io.tick(), - None => false, + fn drive_io(&self) -> bool { + if let Some(io) = &self.io { + io.tick(&self.rng) + } else { + false } } diff --git a/crates/runtime-core/src/sim/io/fs.rs b/crates/runtime-core/src/sim/io/fs.rs index bdc64b87657..908c8b7ba83 100644 --- a/crates/runtime-core/src/sim/io/fs.rs +++ b/crates/runtime-core/src/sim/io/fs.rs @@ -74,7 +74,7 @@ impl File { /// /// Extending allocates pages eagerly as needed. Shrinking drops all pages /// at or beyond the new EOF. - pub fn set_len(&self, new_len: u64) -> Result<()> { + pub(super) fn set_len(&self, new_len: u64) -> Result<()> { use cmp::Ordering::*; if !new_len.is_multiple_of(PAGE_SIZE_U64) { @@ -107,7 +107,7 @@ impl File { } /// Read one complete page. - pub fn read_page(&self, dst: &mut [u8], index: u64) -> Result<()> { + pub(super) fn read_page(&self, dst: &mut [u8], index: u64) -> Result<()> { if dst.len() != PAGE_SIZE { return Err(Error::UnalignedBuffer); } @@ -125,7 +125,7 @@ impl File { } /// Write one complete page. - pub fn write_page(&self, src: &[u8], index: u64) -> Result<()> { + pub(super) fn write_page(&self, src: &[u8], index: u64) -> Result<()> { if src.len() != PAGE_SIZE { return Err(Error::UnalignedBuffer); } diff --git a/crates/runtime-core/src/sim/io/mod.rs b/crates/runtime-core/src/sim/io/mod.rs index e50c77f558a..c9a69fcb8ba 100644 --- a/crates/runtime-core/src/sim/io/mod.rs +++ b/crates/runtime-core/src/sim/io/mod.rs @@ -2,9 +2,8 @@ use alloc::{ boxed::Box, collections::{BTreeMap, VecDeque}, sync::Arc, - vec::Vec, }; -use core::{ops::RangeBounds, result::Result}; +use core::{num::NonZeroUsize, result::Result}; use futures_channel::oneshot; use crate::{ @@ -21,11 +20,23 @@ pub use fs::File; #[derive(Debug)] pub enum Error { - FileNotFound { path: Box }, - FileAlreadyExists { path: Box }, - ShortWrite { expected: usize, written: usize }, - UnexpectedEof { expected: usize, read: usize }, + FileNotFound { + path: Box, + }, + FileAlreadyExists { + path: Box, + }, + ShortWrite { + expected: usize, + written: usize, + }, + UnexpectedEof { + expected: usize, + read: usize, + }, Fs(fs::Error), + /// Injected by the I/O driver. + Cancelled, } impl From for Error { @@ -46,6 +57,23 @@ pub struct SimulatorIO { } impl SimulatorIO { + /// Returns `true` if there are entries in either the submission or + /// completion queues. + pub fn pending(&self) -> bool { + let inner = self.inner.lock(); + inner.submissions.len() + inner.completions.len() > 0 + } + + /// Number of entries in the submission queue. + pub fn pending_submissions(&self) -> usize { + self.inner.lock().submissions.len() + } + + /// Number of entries in the completion queue. + pub fn pending_completions(&self) -> usize { + self.inner.lock().completions.len() + } + /// Run the submission at the front of the queue (if any), and complete the /// completion at the front of the queue (if any). pub fn tick(&self) -> bool { @@ -57,23 +85,24 @@ impl SimulatorIO { self.inner.lock().execute(sqe); } - /// Remove and return the submission at the fron of the queue, if any. + /// Remove and return the submission at the front of the queue, if any. pub fn next_submission(&self) -> Option> { - self.inner.lock().next() + self.inner.lock().next_submission() } /// Remove and return a random submission, or `None` if the queue is empty. pub fn random_submission(&self, rng: &Rng) -> Option> { - self.inner.lock().next_random(rng) + self.inner.lock().random_submission(rng) + } + + /// Remove and return the completion at the front of the queue, if any. + pub fn next_completion(&self) -> Option> { + self.inner.lock().next_completion() } - /// Remove `range` from the completion queue. - pub fn completions(&self, range: impl RangeBounds) -> impl Iterator> { - self.inner - .lock() - .drain_completions(range) - .collect::>() - .into_iter() + /// Remove and return a random completion, or `None` if the queue is empty. + pub fn random_completion(&self, rng: &Rng) -> Option> { + self.inner.lock().random_completion(rng) } async fn submit_and_wait( @@ -219,17 +248,22 @@ impl SimulatorIOInner { } } - fn next(&mut self) -> Option> { + fn next_submission(&mut self) -> Option> { self.submissions.pop_front() } - fn next_random(&mut self, rng: &Rng) -> Option> { - let i = rng.next_u64() % self.submissions.len() as u64; - self.submissions.remove(i as usize) + fn random_submission(&mut self, rng: &Rng) -> Option> { + let len = NonZeroUsize::new(self.submissions.len())?; + self.submissions.remove(rng.index(len.get())) + } + + fn next_completion(&mut self) -> Option> { + self.completions.pop_front() } - fn drain_completions(&mut self, range: impl RangeBounds) -> impl Iterator> { - self.completions.drain(range) + fn random_completion(&mut self, rng: &Rng) -> Option> { + let len = NonZeroUsize::new(self.completions.len())?; + self.completions.remove(rng.index(len.get())) } fn submit(&mut self, op: Box) { @@ -245,11 +279,8 @@ mod tests { #[test] fn create_file() { - let mut rt = Runtime::with_config(RuntimeConfig { - enable_io: true, - ..<_>::default() - }); - let io = rt.io().clone().unwrap(); + let mut rt = Runtime::with_config(RuntimeConfig::default().enable_io()); + let io = rt.io().cloned().unwrap(); let fd = rt .block_on(io.create_file("/data/test", 2 * SECTOR_SIZE as u64)) @@ -285,11 +316,8 @@ mod tests { #[test] fn write_read_roundtrip() { - let mut rt = Runtime::with_config(RuntimeConfig { - enable_io: true, - ..<_>::default() - }); - let io = rt.io().clone().unwrap(); + let mut rt = Runtime::with_config(RuntimeConfig::default().enable_io()); + let io = rt.io().cloned().unwrap(); let fd = rt .block_on(io.create_file("/data/test", 2 * SECTOR_SIZE as u64)) diff --git a/crates/runtime-core/src/sim/io/op.rs b/crates/runtime-core/src/sim/io/op.rs index a4e95d71fb6..3c2b1f90086 100644 --- a/crates/runtime-core/src/sim/io/op.rs +++ b/crates/runtime-core/src/sim/io/op.rs @@ -22,6 +22,16 @@ pub trait Submission: Send + Any { /// - The submission is a [Noop]. /// fn execute(self: Box, files: &mut BTreeMap, fs::File>) -> Option>; + + /// Cancel the operation instead of executing it. + /// + /// This will generate a [Completion] with the result [Error::Cancelled], + /// unless: + /// + /// - The submission is a sub-operation, such as [WritePage] or [ReadPage]. + /// - The submission is a [Noop]. + /// + fn cancel(self: Box) -> Option>; } /// An object containing the result of executing a [Submission], as well as a @@ -160,6 +170,10 @@ impl Submission for Noop { fn execute(self: Box, _files: &mut BTreeMap, fs::File>) -> Option> { None } + + fn cancel(self: Box) -> Option> { + None + } } /// An operation that does nothing. @@ -178,6 +192,11 @@ impl Submission for Ready { let Self(completion) = *self; Some(completion) } + + fn cancel(self: Box) -> Option> { + let Self(completion) = *self; + Some(completion) + } } /// An operation that is already complete with `result`. @@ -197,6 +216,11 @@ impl Submission for OpenFile { let result = files.get(&path).cloned().ok_or(Error::FileNotFound { path }); Some(completion(result, on_complete)) } + + fn cancel(self: Box) -> Option> { + let Self { path: _, on_complete } = *self; + Some(completion(Err(Error::Cancelled), on_complete)) + } } /// [Submission] created by [create_file]. @@ -219,6 +243,15 @@ impl Submission for CreateFile { })(); Some(completion(result, on_complete)) } + + fn cancel(self: Box) -> Option> { + let Self { + path: _, + len: _, + on_complete, + } = *self; + Some(completion(Err(Error::Cancelled), on_complete)) + } } /// [Submission] created by [get_len]. @@ -233,6 +266,11 @@ impl Submission for GetLen { let result = Ok(fd.len()); Some(completion(result, on_complete)) } + + fn cancel(self: Box) -> Option> { + let Self { fd: _, on_complete } = *self; + Some(completion(Err(Error::Cancelled), on_complete)) + } } /// [Submission] created by [set_len]. @@ -248,6 +286,15 @@ impl Submission for SetLen { let result = fd.set_len(len).map_err(Error::from); Some(completion(result, on_complete)) } + + fn cancel(self: Box) -> Option> { + let Self { + fd: _, + len: _, + on_complete, + } = *self; + Some(completion(Err(Error::Cancelled), on_complete)) + } } struct PagedOpState { @@ -259,14 +306,14 @@ struct PagedOpState { fn complete_page_op( state: &Arc>>, - result: Result<(), fs::Error>, + result: Result<(), Error>, ) -> Option>> { let complete = { let mut state = state.lock(); if let Err(e) = result && state.first_error.is_none() { - state.first_error.replace(e.into()); + state.first_error.replace(e); } assert!(state.remaining > 0); state.remaining -= 1; @@ -327,7 +374,17 @@ impl Submission for WritePage { let end = start + SECTOR_SIZE; fd.write_page(&buf.as_bytes()[start..end], file_page as _) }; - complete_page_op(&state, result).map(|c| c as Box) + complete_page_op(&state, result.map_err(Into::into)).map(|c| c as Box) + } + + fn cancel(self: Box) -> Option> { + let Self { + fd: _, + file_page: _, + buf_page: _, + state, + } = *self; + complete_page_op(&state, Err(Error::Cancelled)).map(|c| c as Box) } } @@ -355,7 +412,17 @@ impl Submission for ReadPage { let end = start + SECTOR_SIZE; fd.read_page(&mut buf.as_bytes_mut()[start..end], file_page as _) }; - complete_page_op(&state, result).map(|c| c as Box) + complete_page_op(&state, result.map_err(Into::into)).map(|c| c as Box) + } + + fn cancel(self: Box) -> Option> { + let Self { + fd: _, + file_page: _, + buf_page: _, + state, + } = *self; + complete_page_op(&state, Err(Error::Cancelled)).map(|c| c as Box) } } From b3284d8c16eaf01adb4954a4157da4bea096adb3 Mon Sep 17 00:00:00 2001 From: Kim Altintop Date: Fri, 7 Aug 2026 14:08:19 +0200 Subject: [PATCH 6/9] Remove TODO --- crates/runtime-core/src/sim/io/mod.rs | 10 ---------- 1 file changed, 10 deletions(-) diff --git a/crates/runtime-core/src/sim/io/mod.rs b/crates/runtime-core/src/sim/io/mod.rs index c9a69fcb8ba..6ea3a2eaa1d 100644 --- a/crates/runtime-core/src/sim/io/mod.rs +++ b/crates/runtime-core/src/sim/io/mod.rs @@ -216,16 +216,6 @@ struct SimulatorIOInner { } impl SimulatorIOInner { - // TODO: Allow runtime to inject faults via: - // - // - pick random entries from the submission queue - // - drop queue entries - // - delay `execute` (somehow) - // - delay `complete` - // - make a submission fail without performing its effect - // - execute an arbitrary number of (random) SQEs - // - complete an arbitrary number of CQEs - fn tick(&mut self) -> bool { let mut progress = false; if let Some(sqe) = self.submissions.pop_front() { From e9378c334054d9dbab5ab93cc45616cb28ef9be9 Mon Sep 17 00:00:00 2001 From: Kim Altintop Date: Fri, 7 Aug 2026 15:30:52 +0200 Subject: [PATCH 7/9] Fix optional dependencies --- crates/runtime/Cargo.toml | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/crates/runtime/Cargo.toml b/crates/runtime/Cargo.toml index 2b2cffc5317..d23741ce139 100644 --- a/crates/runtime/Cargo.toml +++ b/crates/runtime/Cargo.toml @@ -11,10 +11,12 @@ 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"] } @@ -22,4 +24,4 @@ windows-sys = { version = "0.61", features = ["Win32_Storage_FileSystem"] } futures.workspace = true [features] -simulation = ["dep:spacetimedb-runtime-core", "spacetimedb-runtime-core/sim", "dep:libc"] +simulation = ["spacetimedb-runtime-core/sim"] From f868c37829ce116ddbf7c6fb11b96d5415913b3e Mon Sep 17 00:00:00 2001 From: Kim Altintop Date: Fri, 7 Aug 2026 17:02:48 +0200 Subject: [PATCH 8/9] Fix windows --- crates/runtime/src/io/tokio.rs | 74 ++++++++++++++++++++++++---------- 1 file changed, 53 insertions(+), 21 deletions(-) diff --git a/crates/runtime/src/io/tokio.rs b/crates/runtime/src/io/tokio.rs index eb948ed7e1c..7907a3f7cab 100644 --- a/crates/runtime/src/io/tokio.rs +++ b/crates/runtime/src/io/tokio.rs @@ -66,16 +66,9 @@ impl SpacetimeIO for TokioIO { offset: u64, ) -> Result> { let _rt = self.rt.enter(); - asyncify(move || { - #[cfg(unix)] - let res = fd.write_all_at(buf.as_bytes(), offset); - #[cfg(windows)] - let res = fd.seek_write(buf.as_bytes(), offset); - - match res { - Ok(()) => Ok(buf), - Err(error) => Err(ErrorWith { error, with: buf }), - } + asyncify(move || match write_all_at(&fd, buf.as_bytes(), offset) { + Ok(()) => Ok(buf), + Err(error) => Err(ErrorWith { error, with: buf }), }) .await } @@ -87,16 +80,9 @@ impl SpacetimeIO for TokioIO { offset: u64, ) -> Result> { let _rt = self.rt.enter(); - asyncify(move || { - #[cfg(unix)] - let res = fd.read_exact_at(buf.as_bytes_mut(), offset); - #[cfg(windows)] - let res = fd.seek_read(buf.as_bytes_mut(), offset); - - match res { - Ok(()) => Ok(buf), - Err(error) => Err(ErrorWith { error, with: buf }), - } + asyncify(move || match read_exact_at(&fd, buf.as_bytes_mut(), offset) { + Ok(()) => Ok(buf), + Err(error) => Err(ErrorWith { error, with: buf }), }) .await } @@ -154,10 +140,56 @@ async fn open_with_direct_io(options: OpenOptions, path: &str) -> io::Result io::Result { options .custom_flags(windows_sys::Win32::Storage::FileSystem::FILE_FLAG_NO_BUFFERING) .open(path) .await } + +#[cfg(unix)] +#[inline] +fn read_exact_at(fd: &std::fs::File, buf: &mut [u8], offset: u64) -> io::Result<()> { + fd.read_exact_at(buf, offset) +} + +#[cfg(windows)] +fn read_exact_at(fd: &std::fs::File, buf: &mut [u8], mut offset: u64) -> io::Result<()> { + while !buf.is_empty() { + match file.seek_read(buf, offset) { + Ok(0) => return Err(ErrorKind::UnexpectedEof.into()), + Ok(n) => { + offset += n as u64; + buf = &mut buf[n..]; + } + Err(ref e) if e.kind() == ErrorKind::Interrupted => {} + Err(e) => return Err(e), + } + } + + Ok(()) +} + +#[cfg(unix)] +#[inline] +fn write_all_at(fd: &std::fs::File, buf: &[u8], offset: u64) -> io::Result<()> { + fd.write_all_at(buf, offset) +} + +#[cfg(windows)] +fn write_all_at(fd: &std::fd::File, buf: &[u8], offset: u64) -> io::Result<()> { + while !buf.is_empty() { + match file.seek_write(buf, offset) { + Ok(0) => return Err(ErrorKind::WriteZero.into()), + Ok(n) => { + offset += n as u64; + buf = &buf[n..]; + } + Err(ref e) if e.kind() == ErrorKind::Interrupted => {} + Err(e) => return Err(e), + } + } + + Ok(()) +} From 031d709990178f2fba1072b8693fe4d810adc391 Mon Sep 17 00:00:00 2001 From: Kim Altintop Date: Sat, 8 Aug 2026 12:24:23 +0200 Subject: [PATCH 9/9] Fix fix windows --- crates/runtime/src/io/tokio.rs | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/crates/runtime/src/io/tokio.rs b/crates/runtime/src/io/tokio.rs index 7907a3f7cab..dcc77dbc5b4 100644 --- a/crates/runtime/src/io/tokio.rs +++ b/crates/runtime/src/io/tokio.rs @@ -141,7 +141,7 @@ async fn open_with_direct_io(options: OpenOptions, path: &str) -> io::Result io::Result { +async fn open_with_direct_io(mut options: OpenOptions, path: &str) -> io::Result { options .custom_flags(windows_sys::Win32::Storage::FileSystem::FILE_FLAG_NO_BUFFERING) .open(path) @@ -155,15 +155,15 @@ fn read_exact_at(fd: &std::fs::File, buf: &mut [u8], offset: u64) -> io::Result< } #[cfg(windows)] -fn read_exact_at(fd: &std::fs::File, buf: &mut [u8], mut offset: u64) -> io::Result<()> { +fn read_exact_at(fd: &std::fs::File, mut buf: &mut [u8], mut offset: u64) -> io::Result<()> { while !buf.is_empty() { - match file.seek_read(buf, offset) { - Ok(0) => return Err(ErrorKind::UnexpectedEof.into()), + 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() == ErrorKind::Interrupted => {} + Err(ref e) if e.kind() == io::ErrorKind::Interrupted => {} Err(e) => return Err(e), } } @@ -178,15 +178,15 @@ fn write_all_at(fd: &std::fs::File, buf: &[u8], offset: u64) -> io::Result<()> { } #[cfg(windows)] -fn write_all_at(fd: &std::fd::File, buf: &[u8], offset: u64) -> io::Result<()> { +fn write_all_at(fd: &std::fs::File, mut buf: &[u8], mut offset: u64) -> io::Result<()> { while !buf.is_empty() { - match file.seek_write(buf, offset) { - Ok(0) => return Err(ErrorKind::WriteZero.into()), + 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() == ErrorKind::Interrupted => {} + Err(ref e) if e.kind() == io::ErrorKind::Interrupted => {} Err(e) => return Err(e), } }