diff --git a/Cargo.lock b/Cargo.lock index bb5abb64fdd..a3060155446 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.8.0" 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/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 fbb7f7c0cf2..1913329b2e2 100644 --- a/crates/runtime-core/src/sim/executor/mod.rs +++ b/crates/runtime-core/src/sim/executor/mod.rs @@ -10,22 +10,38 @@ use core::{ use spin::Mutex; +use crate::sim::io::SimulatorIO; + use super::{time::TimeHandle, Rng}; +mod io; + mod task; use task::Abortable; pub use task::{AbortHandle, JoinError, JoinHandle}; type Runnable = async_task::Runnable; -#[derive(Clone, Copy, Debug, Eq, PartialEq)] +#[derive(Clone, Copy, Debug, PartialEq)] pub struct RuntimeConfig { pub seed: u64, + pub io: Option, } impl RuntimeConfig { pub const fn new(seed: u64) -> Self { - Self { seed } + Self { seed, io: None } + } + + pub fn enable_io(self) -> Self { + Self { + io: Some(self.io.unwrap_or_default()), + ..self + } + } + + pub fn with_io_config(self, io: Option) -> Self { + Self { io, ..self } } } @@ -145,6 +161,12 @@ impl Runtime { } } + // TODO: This is a stopgap to allow submission of I/O tasks. We probably + // want the user-facing API to hide this. + pub fn io(&self) -> Option<&SimulatorIO> { + self.executor.io.as_ref().map(|driver| driver.io()) + } + /// Drive a top-level future to completion on the simulation executor. /// /// While the future runs, spawned tasks share the same deterministic @@ -360,6 +382,7 @@ struct Executor { next_node: AtomicU64, rng: Rng, time: TimeHandle, + io: Option, } impl Executor { @@ -375,6 +398,7 @@ impl Executor { next_node: AtomicU64::new(1), rng: Rng::new(config.seed), time: TimeHandle::new(), + io: config.io.map(io::Driver::new), } } @@ -491,6 +515,7 @@ impl Executor { loop { self.run_all_ready(); + let pending_io = self.drive_io(); if task.is_finished() { let waker = Waker::noop(); return match Pin::new(&mut task).poll(&mut Context::from_waker(waker)) { @@ -499,7 +524,7 @@ impl Executor { }; } - if self.time.wake_next_timer() { + if self.time.wake_next_timer() || pending_io { continue; } @@ -527,6 +552,14 @@ impl Executor { } } + fn drive_io(&self) -> bool { + if let Some(io) = &self.io { + io.tick(&self.rng) + } else { + false + } + } + /// Look up the record for a node, panicking if the node is unknown. fn node_record(&self, node: NodeId) -> Arc { self.nodes 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..908c8b7ba83 --- /dev/null +++ b/crates/runtime-core/src/sim/io/fs.rs @@ -0,0 +1,154 @@ +use alloc::{collections::BTreeMap, sync::Arc}; +use core::{ + cmp, + 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)] +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: spin::Mutex<[u8; PAGE_SIZE]>, +} + +impl Page { + fn zeroed() -> Self { + Self { + bytes: spin::Mutex::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: Arc>>>, + len: Arc, +} + +impl File { + pub(super) fn new() -> Self { + Self { + pages: Arc::new(spin::Mutex::new(BTreeMap::new())), + len: Arc::new(AtomicU64::new(0)), + } + } + + pub(super) fn len(&self) -> u64 { + self.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<()> { + use cmp::Ordering::*; + + if !new_len.is_multiple_of(PAGE_SIZE_U64) { + return Err(Error::UnalignedOffset); + } + let old_len = self.len(); + + 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.store(new_len, Ordering::Relaxed); + } + Less => { + self.len.store(new_len, Ordering::Relaxed); + + let first_removed = PageIndex::from_offset(new_len); + let removed = self.pages.lock().split_off(&first_removed); + drop(removed); + } + } + + 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.len.fetch_max(end, Ordering::Relaxed); + + Ok(()) + } + + fn get_page(&self, index: PageIndex) -> Option> { + self.pages.lock().get(&index).cloned() + } + + 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 new file mode 100644 index 00000000000..6ea3a2eaa1d --- /dev/null +++ b/crates/runtime-core/src/sim/io/mod.rs @@ -0,0 +1,327 @@ +use alloc::{ + boxed::Box, + collections::{BTreeMap, VecDeque}, + sync::Arc, +}; +use core::{num::NonZeroUsize, result::Result}; +use futures_channel::oneshot; + +use crate::{ + io::{AlignedBytes, ErrorWith, SpacetimeIO}, + sim::Rng, +}; + +mod fs; +pub mod op; +use op::{Completion, Submission}; + +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), + /// Injected by the I/O driver. + Cancelled, +} + +impl From for Error { + fn from(e: fs::Error) -> Self { + Self::Fs(e) + } +} + +#[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 { + /// 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 { + self.inner.lock().tick() + } + + /// Execute `sqe`. + pub fn execute(&self, sqe: Box) { + self.inner.lock().execute(sqe); + } + + /// Remove and return the submission at the front of the queue, if any. + pub fn next_submission(&self) -> Option> { + 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().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 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( + &self, + op: impl FnOnce(oneshot::Sender) -> Box, + ) -> Result { + let (tx, rx) = oneshot::channel(); + self.inner.lock().submit(op(tx)); + 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.lock().submit(op); + } + 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.lock().submit(op); + } + 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) -> bool { + let mut progress = false; + if let Some(sqe) = self.submissions.pop_front() { + if let Some(cqe) = sqe.execute(&mut self.files) { + self.completions.push_back(cqe); + } + progress = true; + } + if let Some(cqe) = self.completions.pop_front() { + cqe.complete(); + progress = true; + } + + progress + } + + fn execute(&mut self, sqe: Box) { + if let Some(cqe) = sqe.execute(&mut self.files) { + self.completions.push_back(cqe); + } + } + + fn next_submission(&mut self) -> Option> { + self.submissions.pop_front() + } + + 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 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) { + self.submissions.push_back(op); + } +} + +#[cfg(test)] +mod tests { + use crate::sim::{Runtime, RuntimeConfig}; + + use super::*; + + #[test] + fn create_file() { + 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)) + .unwrap(); + assert_eq!(fd.len(), 2 * SECTOR_SIZE as u64); + } + + #[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 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)) + .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.clear(); + 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..3c2b1f90086 --- /dev/null +++ b/crates/runtime-core/src/sim/io/op.rs @@ -0,0 +1,440 @@ +use core::any::Any; + +use alloc::{ + boxed::Box, + collections::{btree_map, BTreeMap}, + sync::Arc, +}; +use futures_channel::oneshot; + +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>; + + /// 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 +/// 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>; + +/// 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, + 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), + on_complete: Some(on_complete), + 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 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, + 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), + on_complete: Some(on_complete), + 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 + }) +} + +/// Open file at `path`. +pub fn open_file(path: &str, on_complete: OnComplete>) -> Box { + Box::new(OpenFile { + path: path.into(), + on_complete, + }) +} + +/// 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, + on_complete, + }) +} + +/// Get the length of the file `fd`. +pub fn get_len(fd: fs::File, on_complete: OnComplete>) -> Box { + Box::new(GetLen { fd, on_complete }) +} + +/// 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, + on_complete: OnComplete, +} + +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, on_complete, .. + } = *self; + let _ = on_complete.send(result); + } +} + +/// [Submission] created by [noop]. +pub(crate) struct Noop; + +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. +/// +/// 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>) -> Option> { + 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`. +pub fn ready(result: T, on_complete: OnComplete) -> Box { + Box::new(Ready(completion(result, on_complete))) +} + +/// [Submission] created by [open_file]. +pub(crate) struct OpenFile { + path: Box, + on_complete: OnComplete>, +} + +impl Submission for OpenFile { + 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 }); + 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]. +pub(crate) struct CreateFile { + path: Box, + len: u64, + on_complete: OnComplete>, +} + +impl Submission for CreateFile { + 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()), + btree_map::Entry::Occupied(_) => Err(Error::FileAlreadyExists { path }), + }?; + file.set_len(len)?; + Ok(file) + })(); + 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]. +pub(crate) struct GetLen { + fd: fs::File, + on_complete: OnComplete>, +} + +impl Submission for GetLen { + fn execute(self: Box, _files: &mut BTreeMap, fs::File>) -> Option> { + let Self { fd, on_complete } = *self; + 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]. +pub(crate) struct SetLen { + fd: fs::File, + len: u64, + on_complete: OnComplete>, +} + +impl Submission for SetLen { + 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); + 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 { + buf: Option, + on_complete: Option>>>, + remaining: usize, + first_error: Option, +} + +fn complete_page_op( + state: &Arc>>, + 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); + } + assert!(state.remaining > 0); + state.remaining -= 1; + + state.remaining == 0 + }; + + complete.then(|| Box::new(PageOpCompletion { state: state.clone() })) +} + +struct PageOpCompletion { + state: Arc>>, +} + +impl Completion for PageOpCompletion { + fn complete(self: Box) { + 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 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 }), + }; + + (on_complete, result) + }; + + let _ = on_complete.send(result); + } +} + +pub(crate) struct WritePage { + fd: fs::File, + file_page: usize, + buf_page: usize, + state: Arc>>, +} + +impl Submission for WritePage { + fn execute(self: Box, _files: &mut BTreeMap, fs::File>) -> Option> { + let Self { + fd, + file_page, + buf_page, + state, + } = *self; + + let result = { + let state_ref = state.lock(); + 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.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) + } +} + +pub(crate) struct ReadPage { + fd: fs::File, + file_page: usize, + buf_page: usize, + state: Arc>>, +} + +impl Submission for ReadPage { + fn execute(self: Box, _files: &mut BTreeMap, fs::File>) -> Option> { + let Self { + fd, + file_page, + buf_page, + state, + } = *self; + + let result = { + let mut state_ref = state.lock(); + 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.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) + } +} + +#[cfg(test)] +mod tests { + use core::any::Any; + + use super::*; + + #[test] + fn downcast() { + let sqe: Box = noop(); + sqe.downcast::().unwrap(); + } +} 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..dcc77dbc5b4 --- /dev/null +++ b/crates/runtime/src/io/tokio.rs @@ -0,0 +1,195 @@ +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 || match write_all_at(&fd, buf.as_bytes(), offset) { + 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 || match read_exact_at(&fd, buf.as_bytes_mut(), offset) { + 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(windows)] +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) + .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, 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(()) +} + +#[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::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(()) +} 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, }