diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ffa08db42..111ef2fa8 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -257,7 +257,8 @@ jobs: # spawned-concurrency's new sibling versions are published in dependency # order by release.yaml; its package is also validated before release. cargo package --package spawned-macros --no-verify - cargo package --package spawned-rt --no-verify + # spawned-sim-libc (the simulator's interposers) ships with spawned-rt. + cargo package --package spawned-sim-libc --package spawned-rt --no-verify - name: Verify spawned-test publication against unreleased siblings run: | diff --git a/.github/workflows/release.yaml b/.github/workflows/release.yaml index d1f1ec6e7..100b7a0c7 100644 --- a/.github/workflows/release.yaml +++ b/.github/workflows/release.yaml @@ -67,6 +67,22 @@ jobs: - name: Wait for crates.io indexing run: sleep 30 + - name: Publish spawned-sim-libc + env: + CARGO_REGISTRY_TOKEN: ${{ secrets.CRATES_IO_TOKEN }} + run: | + OUTPUT=$(cargo publish --locked --package spawned-sim-libc 2>&1) || { + if echo "$OUTPUT" | grep -Eq "already (exists|uploaded)"; then + echo "spawned-sim-libc already published, continuing" + else + echo "$OUTPUT" + exit 1 + fi + } + + - name: Wait for crates.io indexing + run: sleep 30 + - name: Publish spawned-rt env: CARGO_REGISTRY_TOKEN: ${{ secrets.CRATES_IO_TOKEN }} diff --git a/CHANGELOG.md b/CHANGELOG.md index b40287dad..39853b19b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -186,6 +186,23 @@ not returning to its baseline after quiesce, and has a `--inject-leak` negative control. Run it before each release (`docs/TESTING.md`). +- Determinism D1-D3 (priority tier 3). Under `--cfg spawned_sim`: + wall-clock and OS-entropy reads made inside a simulated run return virtual + time and seeded bytes, so `Instant::now`, `SystemTime::now`, + `rand::thread_rng`, `OsRng` and `HashMap` iteration order replay per seed + with no call-site change (D1; madsim's technique, in the new sim-only crate + `spawned-sim-libc`, which holds all the unsafe interposers so the other + crates stay `forbid(unsafe_code)`; Linux/glibc and macOS). Each harness run + gets a fresh thread, so bodies passed to `sim_test`/`run_once`/`explore` + must be `Sync`. `SPAWNED_SIM_STRICT=1` (`sim::set_strict`) fails the run on + such a read, naming the caller. `spawn_blocking` runs inline as one + simulator step instead of panicking (D2; not concurrent under simulation). + `spawned_rt::tasks` gains `sleep_until`, `interval`/`interval_at` + (`Interval`, `MissedTickBehavior`) and `JoinSet` (`AbortHandle`, + `JoinError`), virtual under the simulator and Tokio's otherwise, and a + direct `tokio::time` timer fails the run with a message naming the + replacement (D3). See `docs/TESTING.md` and `docs/MIGRATION.md`. + ## 0.7.3 — 2026-11-10 Patch release with one supervision behaviour change: a failed restart (a diff --git a/Cargo.lock b/Cargo.lock index 7185cf38c..e960fe242 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -221,7 +221,7 @@ checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06" dependencies = [ "cfg-if", "cpufeatures", - "rand_core", + "rand_core 0.10.1", ] [[package]] @@ -392,7 +392,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -617,7 +617,7 @@ dependencies = [ "cfg-if", "libc", "r-efi 6.0.0", - "rand_core", + "rand_core 0.10.1", ] [[package]] @@ -934,7 +934,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" dependencies = [ "hermit-abi", "libc", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -1346,6 +1346,15 @@ dependencies = [ "thiserror", ] +[[package]] +name = "ppv-lite86" +version = "0.2.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85eae3c4ed2f50dcfe72643da4befc30deadb458a9b590d720cde2f2b1e97da9" +dependencies = [ + "zerocopy", +] + [[package]] name = "proc-macro2" version = "1.0.103" @@ -1365,7 +1374,7 @@ dependencies = [ "chacha20", "core_detect", "num-traits", - "rand", + "rand 0.10.3", "rand_xorshift", "regex-syntax", "unarray", @@ -1401,6 +1410,17 @@ version = "6.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" +[[package]] +name = "rand" +version = "0.8.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e058c7de0b26af77780c769414d6257830bb240f3c38477dbc2c16e5f54d6d4c" +dependencies = [ + "libc", + "rand_chacha", + "rand_core 0.6.4", +] + [[package]] name = "rand" version = "0.10.3" @@ -1408,7 +1428,26 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "65c9fb96cbc91e3478eaae79a69fcd3f1ae4ad052e471fe6732fff548984b4af" dependencies = [ "getrandom 0.4.3", - "rand_core", + "rand_core 0.10.1", +] + +[[package]] +name = "rand_chacha" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6c10a63a0fa32252be49d21e7709d4d4baf8d231c2dbce1eaa8141b9b127d88" +dependencies = [ + "ppv-lite86", + "rand_core 0.6.4", +] + +[[package]] +name = "rand_core" +version = "0.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c" +dependencies = [ + "getrandom 0.2.17", ] [[package]] @@ -1423,7 +1462,7 @@ version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "60aa6af80be32871323012e02e6e65f8a7cc7890931ae421d217ad8fe0df2ccf" dependencies = [ - "rand_core", + "rand_core 0.10.1", ] [[package]] @@ -1533,7 +1572,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -1830,6 +1869,8 @@ dependencies = [ "ctrlc", "loom", "nix 0.30.1", + "rand 0.8.8", + "spawned-sim-libc", "tokio", "tokio-stream", "tokio-util", @@ -1837,6 +1878,13 @@ dependencies = [ "tracing-subscriber", ] +[[package]] +name = "spawned-sim-libc" +version = "0.7.3" +dependencies = [ + "libc", +] + [[package]] name = "spawned-test" version = "0.7.3" @@ -1986,7 +2034,7 @@ dependencies = [ "getrandom 0.3.4", "once_cell", "rustix", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -2484,7 +2532,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.48.0", + "windows-sys 0.61.2", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 65eb2b2aa..cae21f712 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -6,6 +6,7 @@ members = [ "concurrency", "macros", "testing", + "sim-libc", "examples/bank", "examples/bank_threads", "examples/name_server", diff --git a/ROADMAP.md b/ROADMAP.md index 669df9f32..9845b9be7 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -74,9 +74,9 @@ From the [determinism audit](docs/research/determinism-audit.md) (2026-11-11) of | # | Item | Effort | |---|---|---| -| D1 | Interpose `clock_gettime`/`gettimeofday` and `getrandom`/`getentropy` under `spawned_sim` (madsim's technique): virtual time and seeded bytes on a thread that owns a world, so `Instant::now`, `SystemTime::now`, `thread_rng`, `OsRng` and `HashMap` keys become deterministic with no call-site change; `SPAWNED_SIM_STRICT=1` panics with the caller instead (the `tokio::spawn`-style check). Covers ~270 clock and ~50 rand sites in the clients. | M | -| D2 | `spawn_blocking` runs inline under sim as one atomic step (today it panics): synchronous DB and compute offload (64 client sites) becomes simulable. | S | -| D3 | Complete `spawned_rt::tasks` (`interval`, `sleep_until`, a `JoinSet`) and turn Tokio's "timers are disabled" panic into a message naming the `spawned_rt` replacement. | S | +| D1 | **Implemented (#267).** Interpose `clock_gettime`/`gettimeofday` and `getrandom`/`getentropy` under `spawned_sim` (madsim's technique): virtual time and seeded bytes on a thread that owns a world, so `Instant::now`, `SystemTime::now`, `thread_rng`, `OsRng` and `HashMap` keys become deterministic with no call-site change; `SPAWNED_SIM_STRICT=1` panics with the caller instead (the `tokio::spawn`-style check). Covers ~270 clock and ~50 rand sites in the clients. Done in the sim-only `spawned-sim-libc` crate (Linux/glibc and macOS; each harness run on a fresh thread so per-thread seeding is per run); left out: `getrandom` 0.3+ on Linux, which finds the symbol with `dlsym`. | M | +| D2 | **Implemented (#267).** `spawn_blocking` runs inline under sim as one atomic step (today it panics): synchronous DB and compute offload (64 client sites) becomes simulable. | S | +| D3 | **Implemented (#267).** Complete `spawned_rt::tasks` (`interval`, `sleep_until`, a `JoinSet`) and turn Tokio's "timers are disabled" panic into a message naming the `spawned_rt` replacement. | S | | D4 | Route the clients' direct Tokio calls: Ethrex 34 `tokio::spawn`/`JoinSet` and 48 `tokio::time`, Ethlambda 8 and 8 (lists in the audit). | M | | D5 | Library fixes: per-world id counters under sim (byte-identical traces, so S2 can run in CI); `PendingReplies::drain` in insertion order (0.8); `FailScenario::setup()` seeded from the world. | S | | D6 | First client `#[sim_test]`s (none exist today): Ethlambda's blockchain actor, Ethrex `tx_broadcaster` and `peer_table`, after D1. | S-M | diff --git a/concurrency/tests/sim.rs b/concurrency/tests/sim.rs index cac1ec64c..8f77566f4 100644 --- a/concurrency/tests/sim.rs +++ b/concurrency/tests/sim.rs @@ -323,7 +323,7 @@ fn same_seed_same_trace_different_seeds_differ() { #[test] fn virtual_time_advances_only_when_idle() { let rt = Runtime::with_seed(1); - let wall = Instant::now(); + let wall = sim::wall_now(); rt.block_on(async { spawned_rt::tasks::sleep(Duration::from_secs(3600)).await; let late = spawned_rt::tasks::timeout( @@ -337,13 +337,37 @@ fn virtual_time_advances_only_when_idle() { assert!(wall.elapsed() < Duration::from_secs(1)); } +/// Determinism D2: `spawn_blocking` runs inline as one simulator step. #[test] -#[should_panic(expected = "spawned_sim: spawn_blocking is not supported")] -fn spawn_blocking_is_rejected() { +fn spawn_blocking_runs_inline() { let rt = Runtime::with_seed(1); - rt.block_on(async { - drop(spawned_rt::tasks::spawn_blocking(|| ())); + let thread = std::thread::current().id(); + let ran_on = rt.block_on(async { + spawned_rt::tasks::spawn_blocking(|| std::thread::current().id()) + .await + .unwrap() + }); + assert_eq!(ran_on, thread); +} + +/// A Blocking-backend actor drives its loop with a nested `block_on`, which +/// cannot run on the simulation thread: the run fails naming it instead of +/// the actor silently never starting. +#[test] +fn blocking_backend_actor_fails_the_run() { + let outcome = sim::run_once(1, None, &|_| async { + let log = Log::default(); + let node = Node { + id: 1, + crash_after: None, + log, + } + .start_with_backend(spawned_concurrency::tasks::Backend::Blocking); + spawned_rt::tasks::sleep(Duration::from_millis(10)).await; + drop(node); }); + let failure = outcome.failure.expect("a Blocking actor must fail the run"); + assert!(failure.contains("Backend::Blocking"), "{failure}"); } #[test] diff --git a/concurrency/tests/sim_explore.rs b/concurrency/tests/sim_explore.rs index f6af8020c..7f060e6ac 100644 --- a/concurrency/tests/sim_explore.rs +++ b/concurrency/tests/sim_explore.rs @@ -262,7 +262,7 @@ async fn unstarted_child_never_ends(wanted: &str) { fn run(strategy: &str, budget: usize, body: &F) -> Report where - F: Fn(u64) -> Fut, + F: Fn(u64) -> Fut + Sync, Fut: std::future::Future, { let strategy = Strategy::parse(strategy).unwrap(); @@ -311,7 +311,7 @@ fn assert_found(report: &Report, bug: &str) { /// through the existing tooling. fn assert_replays(report: &Report, body: &F) where - F: Fn(u64) -> Fut, + F: Fn(u64) -> Fut + Sync, Fut: std::future::Future, { let found = report.found.as_ref().expect("bug not found"); @@ -543,7 +543,7 @@ fn explore_budget_table() { fn table(title: &str, budget: usize, bug: &str, body: F) where - F: Fn(u64) -> Fut, + F: Fn(u64) -> Fut + Sync, Fut: std::future::Future, { println!("{title}:"); diff --git a/concurrency/tests/sim_turn_budget.rs b/concurrency/tests/sim_turn_budget.rs index 339438f4e..7e3f5c757 100644 --- a/concurrency/tests/sim_turn_budget.rs +++ b/concurrency/tests/sim_turn_budget.rs @@ -12,7 +12,7 @@ use spawned_concurrency::{ tasks::{Actor, ActorStart, Context, Handler}, }; use spawned_rt::tasks::{sim, SimTest}; -use std::time::{Duration, Instant}; +use std::time::Duration; /// Budget used by the in-process tests; the blocking sections run well past /// it and the yielding ones stay well under it. @@ -40,9 +40,11 @@ impl Message for Spin { } impl Handler for Worker { async fn handle(&mut self, _msg: Spin, _ctx: &Context) -> u64 { - let start = Instant::now(); + // The real clock: `std::time::Instant` is virtual inside a + // simulated poll and does not move while a handler spins. + let start = spawned_rt::tasks::sim::wall_now(); let mut n = 0u64; - while start.elapsed() < OVER { + while spawned_rt::tasks::sim::wall_now() - start < OVER { n = std::hint::black_box(n.wrapping_add(1)); } n diff --git a/docs/GUIDE.md b/docs/GUIDE.md index e2d5bc9e4..ce2c54860 100644 --- a/docs/GUIDE.md +++ b/docs/GUIDE.md @@ -600,16 +600,22 @@ pub enum ActorError { ### tasks module (`spawned_rt::tasks`) -Wraps tokio primitives: +Wraps tokio primitives. Under `--cfg spawned_sim` every timer and task here +runs on the deterministic simulator (virtual time), so code that uses these +instead of `tokio::time` / `tokio::spawn` directly is testable with +`#[sim_test]` ([docs/TESTING.md](TESTING.md)): | Item | Description | |------|-------------| | `run(future)` | Create a tokio runtime, initialize tracing, and block on the future | | `block_on(future)` | Block on a future using the current tokio runtime handle | | `spawn(future)` | Spawn an async task | -| `spawn_blocking(f)` | Spawn a blocking closure on tokio's blocking pool | +| `spawn_blocking(f)` | Spawn a blocking closure on tokio's blocking pool (inline, as one step, under the simulator) | | `sleep(duration)` | Async sleep | +| `sleep_until(deadline)` | Async sleep until a `spawned_rt::time::Instant` | | `timeout(duration, future)` | Wrap a future with a timeout | +| `interval(period)` / `interval_at(start, period)` | Periodic ticks (`Interval`, `MissedTickBehavior`), Tokio's semantics | +| `JoinSet` / `AbortHandle` / `JoinError` | A set of spawned tasks whose results are collected as they finish | | `Runtime` | Tokio runtime (re-export) | | `JoinHandle` | Handle to a spawned task | | `CancellationToken` | Cooperative cancellation (from tokio-util) | diff --git a/docs/MIGRATION.md b/docs/MIGRATION.md index 6f3e81021..f753e32b3 100644 --- a/docs/MIGRATION.md +++ b/docs/MIGRATION.md @@ -3,6 +3,26 @@ Upgrade notes per release, newest first. The [CHANGELOG](../CHANGELOG.md) lists every change; this guide covers what you need to edit when upgrading. +## Unreleased + +**Simulation builds only (`--cfg spawned_sim`; determinism D1-D3).** + +- `sim_test` / `SimTest::run`, `sim::run_once`, `run_with`, `shrink` and + `explore` bodies must be `Sync` (each simulated run gets a fresh thread). A + closure that captures an `Rc` or `Cell` needs an `Arc`/`Mutex` instead; the + bound applies in normal builds too, so a test compiles the same way in both. + State a body leaves in a `thread_local!` is on the run's thread, not the + test's: collect results through an `Arc>` or a static keyed by + `std::thread::current().name()` (the run thread is named like the test). +- `std::time::Instant::now()` and `SystemTime::now()` inside a simulated run + read virtual time, and OS entropy (`HashMap` keys, `thread_rng`, `OsRng`) is + seeded by the run. A test that measured real time inside a run must use + `sim::wall_now()`; a loop that spins on `Instant::elapsed()` with no + `.await` never ends (the watchdog aborts it). +- `spawn_blocking` no longer panics: it runs inline as one step. + `Backend::Blocking` actors and nested `block_on` still fail, now by failing + the run. + ## 0.7.3 **Behaviour change (supervision).** A Transient child's replacement (any generation after the child was first diff --git a/docs/TESTING.md b/docs/TESTING.md index 729688af3..ff883f0c9 100644 --- a/docs/TESTING.md +++ b/docs/TESTING.md @@ -492,6 +492,99 @@ Other `#[test]`s in a simulated binary that build a runtime with `spawned_rt::tasks::Runtime::new()` run on the simulator too, but only with seed 0 (or `SPAWNED_SIM_SEED`); they are not seed sweeps. +**Wall clock, entropy, blocking work and timers (determinism D1-D3).** +Real code reads the clock and the OS entropy source directly, offloads +synchronous work with `spawn_blocking`, and uses Tokio's timers. Under the cfg: + +- *D1, clock and entropy.* The `spawned-sim-libc` crate (compiled only under + `--cfg spawned_sim`; the one place with unsafe code, so `spawned-rt` and + the other crates stay `#![forbid(unsafe_code)]` in every build) defines + `clock_gettime`, `gettimeofday`, `getrandom`, `getentropy`, on Linux the + `syscall` entry for `SYS_getrandom` and on macOS `CCRandomGenerateBytes`, + madsim's technique. The test binary's references (std, `rand`, `chrono`) + bind to them at link time; each asks the simulator and otherwise calls libc + through `dlsym(RTLD_NEXT)`. While a world runs on the thread (inside its + `block_on`), `std::time::Instant::now()` reads `sim::VIRTUAL_BOOT` (one day) + plus virtual time and `SystemTime::now()` reads `sim::VIRTUAL_EPOCH` + (2000-01-01T00:00:00Z) plus virtual time. On any thread that has a world, + entropy reads (`HashMap`/`HashSet` keys, `rand::thread_rng`, `OsRng`, + `getrandom`) return bytes from a stream seeded by the world's seed, separate + from the scheduling generator. Every other thread, and the simulator's own + watchdog and turn budget (`sim::wall_now()`), read the real clock. Each + harness run gets a fresh OS thread, so state std and `rand` seed once per + thread is seeded inside the run: a seed in a sweep, the same seed alone, a + tape replay and a separate process observe the same values + (`rt/tests/sim_determinism.rs`). This is why `sim_test`, `run_once` and + `explore` bodies must be `Sync`. Interposition works on Linux with glibc + (the CI target; the `syscall` entry on x86_64 and aarch64) and macOS; + `sim::INTERPOSED` says whether it is active. +- *Strict mode.* `SPAWNED_SIM_STRICT=1` (or `sim::set_strict(true)`) turns a + direct clock or entropy read on the simulated thread into a failure, the + way a direct `tokio::spawn` fails today, with the caller's frames: + + ```text + spawned_sim: determinism leak: clock_gettime(CLOCK_REALTIME) on the simulated + thread (SPAWNED_SIM_STRICT=1): a direct wall-clock or OS-entropy read + (Instant::now, SystemTime::now, thread_rng, OsRng, getrandom); read time through + spawned_rt::time::Instant and draw randomness from a generator seeded by the + test. Caller: + 13: ::now + 14: my_crate::reads_the_wall_clock + ``` + + `HashMap` key generation (std asks with `GRND_INSECURE`) is never + reported, nor are Tokio's own runtime-metrics reads. Log timestamps + (`tracing-subscriber`'s default formatter) are wall-clock reads; use + `without_time()` under strict mode. Strict mode needs the interposers. +- *D2, blocking work.* `spawned_rt::tasks::spawn_blocking` (and + `Handle::spawn_blocking`, `JoinSet::spawn_blocking`, actor + `spawn_blocking_helper`s) runs its closure inline as the single poll of a + new simulated task: the seed decides when it runs, and nothing else in the + world runs while it does. It is **not concurrent** under the simulator. A + closure that waits for async work in the same world cannot finish on one + thread; a nested `block_on` fails the run (so does a `Backend::Blocking` + actor), and a closure that blocks otherwise trips the poll watchdog. + `block_in_place` calls its closure. +- *D3, timers and task sets.* `spawned_rt::tasks` has `sleep_until`, + `interval` / `interval_at` (`Interval`, `MissedTickBehavior`; Tokio's + semantics: the first tick is immediate, `Burst` by default) and `JoinSet` + (`spawn`, `spawn_blocking`, `join_next`, `try_join_next`, `abort_all`, + `shutdown`, `join_all`, `AbortHandle`), virtual under the simulator and + Tokio's otherwise. A simulated `JoinSet::join_next` returns, among tasks + that finished since the last call, the one with the smallest task id. A + direct `tokio::time` timer (Tokio's "timers are disabled" panic) fails the + run, even from a task nobody joins, with: + + ```text + spawned_sim: direct tokio::time call (sleep, sleep_until, timeout or interval) + under the simulator: Tokio's timers do not run on virtual time; use + spawned_rt::tasks::{sleep, sleep_until, timeout, interval} (virtual under + spawned_sim, Tokio otherwise) + ``` + +The simulated `JoinSet` covers `spawn`, `spawn_blocking`, `join_next`, +`join_next_with_id`, `try_join_next`, `abort_all`, `detach_all`, `shutdown`, +`join_all` and a cloneable `AbortHandle`; Tokio's `spawn_local`/`spawn_on` and +`JoinError::id` have no simulated version, and the simulated `Sleep` is +`Unpin` where Tokio's is not, so pin a sleep before selecting on it by +reference. Code that sticks to the common surface compiles in both builds. + +Simulator environment variables (all read under the cfg only): + +| Variable | Effect | +|---|---| +| `SPAWNED_SIM_SEED=n` | run only seed `n` | +| `SPAWNED_SIM_SEEDS=n` | run seeds `0..n` (default 8) | +| `SPAWNED_SIM_TAPE=path` | replay a decision tape | +| `SPAWNED_SIM_TAPE_DIR=dir` | where failing tapes go (default `$TMPDIR/spawned-sim`) | +| `SPAWNED_SIM_SHRINK=n` | shrinker run budget (default 200, 0 = off) | +| `SPAWNED_SIM_STRATEGY=s`, `SPAWNED_SIM_ITERATIONS=n`, `SPAWNED_SIM_FAIR_AFTER=n` | exploration (below) | +| `SPAWNED_SIM_SPIN=n` | busy polls before the clock creeps 1 ms (default 10,000) | +| `SPAWNED_SIM_MAX_STEPS=n` | livelock guard (default 10 million polls) | +| `SPAWNED_SIM_POLL_TIMEOUT=s` | wall-clock poll watchdog (default 60, 0 = off) | +| `SPAWNED_SIM_TURN_BUDGET=ms` | wall-clock budget of one actor turn (default 250, 0 = off) | +| `SPAWNED_SIM_STRICT=1` | a direct clock or entropy read fails the run (D1) | + **What the simulator does not model.** Determinism and replay hold only within these limits: @@ -501,19 +594,29 @@ these limits: - Tokio's cooperative budget is emulated (each poll runs inside a per-run Tokio context, so a task is forced to yield after 128 ready channel operations, as on Tokio), but Tokio's timer, I/O and blocking pools are not: - a direct `tokio::spawn` fails the run, and direct `tokio::time` use panics. + a direct `tokio::spawn` or `tokio::task::spawn_blocking` fails the run, and + a direct `tokio::time` timer fails it with a message naming the + `spawned_rt::tasks` replacement. - Timers fire only when no task is ready, or after `SPAWNED_SIM_SPIN` busy polls (default 10,000) when the clock creeps by 1 ms; every 100 creeps without an idle moment that threshold halves, down to one poll per millisecond. A real clock would fire timers at wall-clock moments in the middle of busy work, at a rate unrelated to poll counts. -- One thread: Thread, Blocking, native and pinned placements, dedicated - runtimes and `spawn_blocking` do not run. A poll with no await point that +- One thread: Thread, Blocking, native and pinned placements and dedicated + runtimes do not run (a Blocking-backend actor fails the run), and + `spawned_rt::tasks::spawn_blocking` runs inline as one atomic step, never + concurrently. A poll with no await point that never returns trips a wall-clock watchdog (`SPAWNED_SIM_POLL_TIMEOUT`, default 60 s) that aborts the process; one that returns after more than `SPAWNED_SIM_TURN_BUDGET` (250 ms) inside an actor turn fails the run. -- Direct `std::time`/`SystemTime` reads, `std::thread`, Tokio I/O and - direct Tokio timers are outside its control. +- `std::thread`, Tokio I/O, and wall-clock and entropy reads the + interposers cannot see (targets other than Linux/glibc and macOS, + `getrandom` found with `dlsym` as the `getrandom` crate 0.3+ does on Linux, + raw syscalls) are outside its control. A blocking wait with a timeout on + the simulated thread (`Condvar::wait_timeout`, `park_timeout`, + `recv_timeout`) computes its kernel deadline from virtual time: it can + return at once or stall until the poll watchdog fires. Blocking the + simulated thread is outside the model either way. - Remaining nondeterminism between runs in one process: actor, monitor and registration ids come from process-global counters (their order within a run is fixed, their values depend on earlier tests), and the name diff --git a/docs/research/determinism-audit.md b/docs/research/determinism-audit.md index 52bf4a021..65019ffe2 100644 --- a/docs/research/determinism-audit.md +++ b/docs/research/determinism-audit.md @@ -99,6 +99,12 @@ per thread, so maps created before the world starts keep OS keys (madsim documents the same caveat); the simulator's own watchdog must call the real symbol; Linux first. +Implemented in #267 (D1-D3): `spawned-sim-libc` holds the interposers, and +the harness runs each seed on a fresh thread, which removes the `HashMap` +caveat above for `sim_test`/`run_once` bodies (keys and `thread_rng` are +seeded inside the run). See [TESTING.md](../TESTING.md) "Wall clock, entropy, +blocking work and timers". + ## Sources - Simulator: `rt/src/tasks/sim.rs`, `rt/src/tasks/sim_test.rs`, diff --git a/examples/capabilities/tests/deterministic_simulation.rs b/examples/capabilities/tests/deterministic_simulation.rs index 9b7f93934..46b160a4d 100644 --- a/examples/capabilities/tests/deterministic_simulation.rs +++ b/examples/capabilities/tests/deterministic_simulation.rs @@ -120,7 +120,9 @@ mod virtual_time { #[spawned_concurrency::sim_test] async fn an_hour_long_timer_fires_in_virtual_time() { - let wall = std::time::Instant::now(); + // The real clock: inside a simulated run `std::time::Instant` reads + // virtual time too (determinism D1). + let wall = spawned_rt::tasks::sim::wall_now(); let virtual_start = spawned_rt::time::Instant::now(); let reminder = Reminder::default().start(); @@ -130,6 +132,6 @@ mod virtual_time { assert!(reminder.rung().await.unwrap()); assert!(virtual_start.elapsed() >= HOUR); - assert!(wall.elapsed() < Duration::from_secs(10)); + assert!(spawned_rt::tasks::sim::wall_now() - wall < Duration::from_secs(10)); } } diff --git a/examples/thread_only/Cargo.lock b/examples/thread_only/Cargo.lock index 49448f12a..ed15bba9e 100644 --- a/examples/thread_only/Cargo.lock +++ b/examples/thread_only/Cargo.lock @@ -382,10 +382,18 @@ version = "0.7.3" dependencies = [ "ctrlc", "loom", + "spawned-sim-libc", "tracing", "tracing-subscriber", ] +[[package]] +name = "spawned-sim-libc" +version = "0.7.3" +dependencies = [ + "libc", +] + [[package]] name = "spawned-thread-only-example" version = "0.1.0" diff --git a/fuzz/Cargo.lock b/fuzz/Cargo.lock index 0306c9099..460fe7438 100644 --- a/fuzz/Cargo.lock +++ b/fuzz/Cargo.lock @@ -551,6 +551,7 @@ dependencies = [ "ctrlc", "loom", "nix 0.30.1", + "spawned-sim-libc", "tokio", "tokio-stream", "tokio-util", @@ -558,6 +559,13 @@ dependencies = [ "tracing-subscriber", ] +[[package]] +name = "spawned-sim-libc" +version = "0.7.3" +dependencies = [ + "libc", +] + [[package]] name = "syn" version = "2.0.119" diff --git a/rt/Cargo.toml b/rt/Cargo.toml index 2e700026a..7da7522ca 100644 --- a/rt/Cargo.toml +++ b/rt/Cargo.toml @@ -20,6 +20,14 @@ async-task = { version = "=4.7.1", optional = true } [lib] path = "./src/lib.rs" +# Clock and entropy interposers (determinism D1): simulation builds only, so +# this crate keeps `forbid(unsafe_code)` and no unsafe code is linked otherwise. +[target.'cfg(spawned_sim)'.dependencies] +spawned-sim-libc = { path = "../sim-libc", version = "0.7.3" } + +[target.'cfg(spawned_sim)'.dev-dependencies] +rand = "0.8" + [target.'cfg(spawned_loom)'.dependencies] loom = "0.7" diff --git a/rt/src/tasks/mod.rs b/rt/src/tasks/mod.rs index d6a1f8989..5094e1018 100644 --- a/rt/src/tasks/mod.rs +++ b/rt/src/tasks/mod.rs @@ -55,6 +55,17 @@ pub use crate::tasks::tokio::{ Runtime, RuntimeFlavor, }; pub use crate::tasks::tokio::{BroadcastStream, ReceiverStream}; +// Timers and task sets (determinism D3): virtual under the simulator. +#[cfg(spawned_sim)] +pub use crate::tasks::sim::{ + interval, interval_at, sleep_until, AbortHandle, Interval, JoinError, JoinSet, + MissedTickBehavior, +}; +#[cfg(not(spawned_sim))] +pub use crate::tasks::tokio::{ + interval, interval_at, sleep_until, AbortHandle, Interval, JoinError, JoinSet, + MissedTickBehavior, +}; use std::future::Future; /// Build an explicitly sized async runtime. Runtime owners must destroy it diff --git a/rt/src/tasks/sim.rs b/rt/src/tasks/sim.rs index c0967a102..14a0c090b 100644 --- a/rt/src/tasks/sim.rs +++ b/rt/src/tasks/sim.rs @@ -33,9 +33,29 @@ //! about two million polls, well before the livelock guard //! (`SPAWNED_SIM_MAX_STEPS`, default 10 million), even if other timers keep //! firing. -//! `spawned_rt::time::Instant` reads this clock. -//! - **Detected limits.** `spawn_blocking`, `block_in_place` and a nested -//! `block_on` panic with a message naming the call; a run with no ready task +//! `spawned_rt::time::Instant` reads this clock, and so do +//! `sleep_until`, `interval` and `JoinSet` (determinism D3). A direct +//! `tokio::time` timer fails the run with a message naming the +//! `spawned_rt` replacement. +//! - **Wall clock and OS entropy (D1).** On Linux (glibc) and macOS, +//! `spawned-sim-libc` interposes `clock_gettime`, `gettimeofday`, +//! `getrandom`, `getentropy` (and `syscall(SYS_getrandom)` on Linux, +//! `CCRandomGenerateBytes` on macOS). While a world runs on a thread +//! (inside its `block_on`), `std::time::Instant::now` reads +//! [`VIRTUAL_BOOT`] plus virtual time and `SystemTime::now` reads +//! [`VIRTUAL_EPOCH`] plus virtual time; on any thread that has a world, +//! `getrandom`-style reads (`HashMap` keys, `rand::thread_rng`, `OsRng`) +//! return bytes from a stream seeded by the world's seed. Other threads, +//! and the simulator's own watchdog and turn budget ([`wall_now`]), read +//! the real clock. Each run of the harness gets a fresh thread, so +//! per-thread seeding (`HashMap` keys, `thread_rng`) happens inside the +//! run. `SPAWNED_SIM_STRICT=1` ([`set_strict`]) turns such a read into a +//! failure naming the caller instead. +//! - **Blocking work (D2).** `spawn_blocking` runs its closure inline as the +//! single poll of a new task: one atomic step, never concurrent with +//! anything else in the world. `block_in_place` calls its closure. A +//! nested `block_on` (a Blocking-backend actor, `Handle::block_on` inside +//! a task or a `spawn_blocking` closure) fails the run; a run with no ready task //! and no pending timer panics as a deadlock instead of hanging. A single //! poll that runs longer than `SPAWNED_SIM_POLL_TIMEOUT` wall-clock seconds //! (default 60; `SPAWNED_SIM_POLL_TIMEOUT=0` disables it, and the per-poll @@ -66,8 +86,10 @@ //! Library code reads time and jitter entropy through `spawned_rt::time`, //! which this module backs under the cfg. The one scheduling-relevant //! `HashMap` walk in the library (waking monitor waiters on exit) is done in -//! monitor-creation order under the cfg. Still outside its control: direct -//! `std::time` / `SystemTime` reads, `std::thread`, Tokio I/O, and +//! monitor-creation order under the cfg. Still outside its control: +//! `std::time` and entropy reads on targets without interposers, +//! `getrandom` found through `dlsym` (the `getrandom` crate 0.3+ on Linux), +//! `std::thread`, Tokio I/O, and //! process-global id counters (ids differ between runs in one process, but //! their relative order within a run is fixed). //! @@ -110,6 +132,18 @@ const DEFAULT_POLL_TIMEOUT_SECS: u64 = 60; const DEFAULT_TURN_BUDGET_MS: u64 = 250; pub mod explore; +mod interval; +mod join_set; +pub use interval::{interval, interval_at, sleep_until, Interval, MissedTickBehavior}; +pub use join_set::{AbortHandle, JoinSet}; + +/// The real monotonic clock, also inside a simulated poll, where +/// `std::time::Instant::now` reads virtual time (D1). The simulator's own +/// wall-clock verdicts (poll watchdog, turn budget) use it; so can a test that +/// needs to measure or spend real time. +pub fn wall_now() -> std::time::Instant { + spawned_sim_libc::real_clock(std::time::Instant::now) +} fn lock(m: &Mutex) -> MutexGuard<'_, T> { m.lock().unwrap_or_else(|p| p.into_inner()) @@ -153,7 +187,15 @@ mod watchdog { fn base() -> Instant { static BASE: OnceLock = OnceLock::new(); - *BASE.get_or_init(Instant::now) + *BASE.get_or_init(super::wall_now) + } + + /// Wall-clock milliseconds since `base()` (the real clock, also on a + /// simulated thread). + fn elapsed_ms() -> u64 { + super::wall_now() + .saturating_duration_since(base()) + .as_millis() as u64 } fn limit() -> u64 { @@ -192,7 +234,7 @@ mod watchdog { pub(super) fn busy(&self, task: u64) -> Busy<'_> { if self.limit > 0 { self.task.store(task, Ordering::Relaxed); - let now = base().elapsed().as_millis() as u64; + let now = elapsed_ms(); self.since.store(now.max(1), Ordering::Release); } Busy(self) @@ -211,7 +253,7 @@ mod watchdog { fn run() { loop { std::thread::sleep(Duration::from_millis(200)); - let now = base().elapsed().as_millis() as u64; + let now = elapsed_ms(); for watch in lock(&WATCHED).iter().filter_map(Weak::upgrade) { let since = watch.since.load(Ordering::Acquire); let task = watch.task.load(Ordering::Relaxed); @@ -247,6 +289,7 @@ mod watchdog { /// deterministic. What it measures is still a property of the code, so the /// default is generous and a failure replays with the same seed or tape. mod turn { + use super::wall_now; use std::cell::RefCell; use std::time::{Duration, Instant}; @@ -281,7 +324,7 @@ mod turn { impl Poll { /// Charge the stretch since the last boundary to the current turn. fn close(&mut self) { - let now = Instant::now(); + let now = wall_now(); if let Some(label) = self.turn { let took = now.duration_since(self.since); if took > self.budget && self.overrun.is_none() { @@ -299,7 +342,7 @@ mod turn { let poll = Poll { budget, turn, - since: Instant::now(), + since: wall_now(), overrun: None, }; POLL.with(|p| p.borrow_mut().replace(poll)) @@ -577,7 +620,19 @@ fn choose(state: &mut State) -> usize { struct Shared { owner: ThreadId, max_steps: u64, + /// The world's seed (also `State::seed`), readable without the lock. + seed: u64, state: Mutex, + /// `State::now` in nanoseconds, mirrored so an interposed clock read + /// (D1, `interpose`) never needs the state lock, which the reading + /// thread may hold. + now_nanos: AtomicU64, + /// Draws from the OS-entropy stream (interposed `getrandom` and kin). + os_entropy: AtomicU64, + /// `SPAWNED_SIM_STRICT=1`: a direct clock or entropy read fails the run. + strict: std::sync::atomic::AtomicBool, + /// The first such read under strict mode: the call and its caller. + strict_leak: Mutex>, /// Driverless Tokio runtime every poll runs inside (see the module docs), /// built on the first poll so a replay can still choose its seed. tokio: Mutex>>, @@ -596,9 +651,15 @@ impl Shared { .ok() .and_then(|v| v.parse().ok()) .unwrap_or(DEFAULT_MAX_STEPS); + interpose::install(); Arc::new(Self { owner: std::thread::current().id(), max_steps, + seed, + now_nanos: AtomicU64::new(0), + os_entropy: AtomicU64::new(0), + strict: std::sync::atomic::AtomicBool::new(strict_mode()), + strict_leak: Mutex::new(None), tokio: Mutex::new(None), watch: watchdog::Watch::register(seed), state: Mutex::new(State { @@ -643,7 +704,8 @@ impl Shared { /// after `f` to let Tokio deliver them to our wakers before `block_on` /// returns. fn run_poll(&self, id: u64, f: impl FnOnce() -> R) -> R { - let runtime = self.tokio_runtime(); + // Built on the first poll: Tokio's own clock reads stay real. + let runtime = spawned_sim_libc::real_clock(|| self.tokio_runtime()); let spawned = runtime.metrics().spawned_tasks_count(); // Work handed to Tokio directly runs outside the simulated schedule: // a `tokio::spawn` (async task) or `tokio::task::spawn_blocking` (an OS @@ -684,19 +746,23 @@ impl Shared { let mut f = Some(f); let mut out = None; let _busy = self.watch.busy(id); - runtime.block_on(std::future::poll_fn(|cx| match f.take() { - Some(f) => { - out = Some(f()); - if escaped(&runtime) { - // Return before Tokio's scheduler tick could run it. - Poll::Ready(()) - } else { - cx.waker().wake_by_ref(); - Poll::Pending + // Tokio's own clock reads around the poll (runtime metrics) stay + // real; the poll itself, the program's code, is intercepted (D1). + spawned_sim_libc::real_clock(|| { + runtime.block_on(std::future::poll_fn(|cx| match f.take() { + Some(f) => { + out = Some(spawned_sim_libc::intercepted(f)); + if escaped(&runtime) { + // Return before Tokio's scheduler tick could run it. + Poll::Ready(()) + } else { + cx.waker().wake_by_ref(); + Poll::Pending + } } - } - None => Poll::Ready(()), - })); + None => Poll::Ready(()), + })) + }); out.expect("spawned_sim: poll did not run") } @@ -747,6 +813,18 @@ impl Shared { /// Fail the run if a determinism leak was recorded. fn check_leaks(&self) { + let strict = lock(&self.strict_leak).take(); + if let Some((call, backtrace)) = strict { + self.shutdown_tokio(); + panic!( + "spawned_sim: determinism leak: {call} on the simulated thread \ + (SPAWNED_SIM_STRICT=1): a direct wall-clock or OS-entropy read \ + (Instant::now, SystemTime::now, thread_rng, OsRng, getrandom); read time \ + through spawned_rt::time::Instant and draw randomness from a generator \ + seeded by the test. Caller:\n{}", + caller_frames(call, &backtrace) + ); + } let foreign = lock(&self.state).foreign.take(); if let Some(foreign) = foreign { self.shutdown_tokio(); @@ -858,7 +936,11 @@ impl Shared { } Action::Fire(deadline) => { state.spin = 0; + // The only write to `now`: keep its lock-free mirror + // (read by interposed clock calls) in step. state.now = state.now.max(deadline); + self.now_nanos + .store(state.now.as_nanos() as u64, Ordering::Relaxed); if let Some(tape) = &mut state.tape { tape.push(Decision::Fire(deadline)); } @@ -983,11 +1065,18 @@ impl Shared { ); { let mut state = lock(&self.state); - assert!( - !state.running, - "spawned_sim: nested block_on (for example ChildHandle::wait_exit_blocking \ - or Handle::block_on inside a task) cannot run on the single simulation thread" - ); + if state.running { + let message = "nested block_on (for example ChildHandle::wait_exit_blocking, \ + Handle::block_on inside a task or a spawn_blocking closure, or a \ + Backend::Blocking actor) cannot run on the single simulation thread"; + // Fails the run even if the panic is caught (a task's + // JoinError nobody reads). + if state.foreign.is_none() { + state.foreign = Some(message.to_string()); + } + drop(state); + panic!("spawned_sim: {message}"); + } state.running = true; } struct Running<'a>(&'a Shared); @@ -1017,15 +1106,27 @@ impl Shared { ); }; if id == ROOT { - let polled = self.run_poll(ROOT, || { - let root = future.as_mut().expect("root polled after completion"); - match root.as_mut().poll(&mut Context::from_waker(&waker)) { - Poll::Ready(output) => { - future = None; - Poll::Ready(output) + let polled = catch_unwind(AssertUnwindSafe(|| { + self.run_poll(ROOT, || { + let root = future.as_mut().expect("root polled after completion"); + match root.as_mut().poll(&mut Context::from_waker(&waker)) { + Poll::Ready(output) => { + future = None; + Poll::Ready(output) + } + Poll::Pending => Poll::Pending, } - Poll::Pending => Poll::Pending, + }) + })) + .unwrap_or_else(|payload| { + let payload = tokio_time_panic(payload); + // The run fails with the renamed panic, not as a leak. + let mut state = lock(&self.state); + if state.foreign.as_deref() == Some(DIRECT_TOKIO_TIME) { + state.foreign = None; } + drop(state); + std::panic::resume_unwind(payload) }); if let Poll::Ready(output) = polled { self.retire(ROOT); @@ -1170,12 +1271,14 @@ impl Handle { { spawn_on(&self.shared, future) } - pub fn spawn_blocking(&self, _f: F) -> JoinHandle + /// Runs `f` inline as one step of a new simulated task; see + /// [`spawn_blocking`]. + pub fn spawn_blocking(&self, f: F) -> JoinHandle where F: FnOnce() -> R + Send + 'static, R: Send + 'static, { - unsupported("Handle::spawn_blocking") + spawn_on(&self.shared, async move { f() }) } pub fn block_on(&self, future: F) -> F::Output { self.shared.block_on(future) @@ -1383,18 +1486,20 @@ pub fn set_turn_budget(budget: Option) { lock(¤t_or_ambient().state).turn_budget = budget.filter(|b| !b.is_zero()); } +/// Strict mode (`SPAWNED_SIM_STRICT=1`) for the running simulation: while +/// on, a direct wall-clock or OS-entropy read on its thread (`Instant::now`, +/// `SystemTime::now`, `thread_rng`, `OsRng`, `getrandom`) fails the run with +/// the caller's backtrace instead of reading virtual time or seeded bytes. +/// `HashMap` key generation is never reported. +pub fn set_strict(on: bool) { + current_or_ambient().strict.store(on, Ordering::Relaxed); +} + /// Virtual time elapsed in this thread's simulation. pub fn now() -> Duration { lock(¤t_or_ambient().state).now } -fn unsupported(call: &str) -> ! { - panic!( - "spawned_sim: {call} is not supported under the simulation backend \ - (it needs a real OS thread); use Backend::Async and async groups" - ) -} - // --------------------------------------------------------------------------- // Tasks // --------------------------------------------------------------------------- @@ -1425,19 +1530,26 @@ where spawn_on(¤t_or_ambient(), future) } -pub fn spawn_blocking(_f: F) -> JoinHandle +/// Runs `f` under the simulator: inline, as the single poll of a new +/// simulated task, so it is one atomic scheduling step and the seed decides +/// when it runs (not concurrently with anything: while `f` runs, nothing else +/// in the world does). A closure that waits for async work in the same world +/// (a channel receive, `block_on`) never returns; a nested `block_on` fails +/// the run, and a stalled poll trips the watchdog. +pub fn spawn_blocking(f: F) -> JoinHandle where F: FnOnce() -> R + Send + 'static, R: Send + 'static, { - unsupported("spawn_blocking") + spawn_on(¤t_or_ambient(), async move { f() }) } -pub fn block_in_place(_f: F) -> R +/// Runs `f` directly, inside the current poll (one atomic step). +pub fn block_in_place(f: F) -> R where F: FnOnce() -> R, { - unsupported("block_in_place") + f() } fn spawn_on(shared: &Arc, future: F) -> JoinHandle @@ -1581,16 +1693,7 @@ impl fmt::Debug for JoinHandle { } impl JoinHandle { pub fn abort(&self) { - let finished = { - let mut inner = lock(&self.join.inner); - inner.aborted = true; - inner.finished - }; - if !finished { - if let Some(shared) = self.shared.upgrade() { - shared.schedule(self.id); - } - } + abort_task(&self.join, &self.shared, self.id); } pub fn is_finished(&self) -> bool { lock(&self.join.inner).finished @@ -1599,6 +1702,20 @@ impl JoinHandle { Id(self.id) } } +/// Mark task `id` aborted and wake it so it is cancelled at its next turn. +fn abort_task(join: &JoinState, shared: &Weak, id: u64) { + let finished = { + let mut inner = lock(&join.inner); + inner.aborted = true; + inner.finished + }; + if !finished { + if let Some(shared) = shared.upgrade() { + shared.schedule(id); + } + } +} + impl Drop for JoinHandle { fn drop(&mut self) { let result = { @@ -1635,6 +1752,31 @@ struct TaskFuture { future: Option>>, join: Arc>, } +/// Tokio's panic for a timer used without a time driver. +const TOKIO_TIMERS_DISABLED: &str = "timers are disabled"; + +/// What a direct `tokio::time` call under the simulator fails with. +const DIRECT_TOKIO_TIME: &str = "direct tokio::time call (sleep, sleep_until, timeout or \ + interval) under the simulator: Tokio's timers do not run on virtual time; use \ + spawned_rt::tasks::{sleep, sleep_until, timeout, interval} (virtual under spawned_sim, \ + Tokio otherwise)"; + +/// Replace Tokio's "timers are disabled" panic with one naming the +/// `spawned_rt` replacement, and fail the run with it (a task's panic may go +/// unread). +fn tokio_time_panic(payload: Box) -> Box { + if !panic_message(&*payload).contains(TOKIO_TIMERS_DISABLED) { + return payload; + } + if let Some(shared) = current() { + let mut state = lock(&shared.state); + if state.foreign.is_none() { + state.foreign = Some(DIRECT_TOKIO_TIME.to_string()); + } + } + Box::new(format!("spawned_sim: {DIRECT_TOKIO_TIME}")) +} + fn panic_error(id: u64, payload: Box) -> JoinError { let message = payload .downcast_ref::<&str>() @@ -1703,7 +1845,7 @@ impl Future for TaskFuture { Err(payload) => { // The poll's panic wins over a second one from the destructor. let _ = self.drop_future(); - Err(panic_error(id, payload)) + Err(panic_error(id, tokio_time_panic(payload))) } }; self.join.complete(result); @@ -1975,7 +2117,7 @@ fn panic_message(payload: &(dyn std::any::Any + Send)) -> String { /// scheduling from the seed's generator. pub fn run_once(seed: u64, script: Option<&Tape>, body: &F) -> Outcome where - F: Fn(u64) -> Fut, + F: Fn(u64) -> Fut + Sync, Fut: Future, { run_with(seed, script.map(|tape| (tape, ReplayMode::Strict)), body) @@ -1984,7 +2126,7 @@ where /// [`run_once`] with the replay mode chosen. pub fn run_with(seed: u64, script: Option<(&Tape, ReplayMode)>, body: &F) -> Outcome where - F: Fn(u64) -> Fut, + F: Fn(u64) -> Fut + Sync, Fut: Future, { run_planned(seed, script, None, body).0 @@ -1992,17 +2134,60 @@ where /// [`run_with`] with an exploration plan choosing the tasks (a script, if /// given, still wins). Returns the plan back so its record can be read. +/// +/// Every run gets a fresh OS thread (named like the caller, with the +/// caller's panic-output setting), so per-thread state that std and crates +/// seed once per thread from the clock or OS entropy (`HashMap` keys, +/// `rand::thread_rng`) is seeded inside this run's world: a seed in a sweep +/// and the same seed replayed alone see the same values. fn run_planned( seed: u64, script: Option<(&Tape, ReplayMode)>, plan: Option, body: &F, ) -> (Outcome, Option) +where + F: Fn(u64) -> Fut + Sync, + Fut: Future, +{ + let quiet = QUIET.with(|q| q.get()); + // At least 8 MiB (a `harness = false` main thread's size; libtest's test + // threads have 2 MiB), or `RUST_MIN_STACK` if larger. + let stack = std::env::var("RUST_MIN_STACK") + .ok() + .and_then(|v| v.trim().parse().ok()) + .unwrap_or(0usize) + .max(8 << 20); + let mut builder = std::thread::Builder::new().stack_size(stack); + if let Some(name) = std::thread::current().name() { + builder = builder.name(name.to_owned()); + } + std::thread::scope(|scope| { + let run = builder + .spawn_scoped(scope, move || { + QUIET.with(|q| q.set(quiet)); + run_on_this_thread(seed, script, plan, body) + }) + .expect("spawned_sim: cannot start a thread for the run"); + run.join() + .unwrap_or_else(|panic| std::panic::resume_unwind(panic)) + }) +} + +fn run_on_this_thread( + seed: u64, + script: Option<(&Tape, ReplayMode)>, + plan: Option, + body: &F, +) -> (Outcome, Option) where F: Fn(u64) -> Fut, Fut: Future, { let rt = Runtime::with_seed(seed); + // std seeds this thread's `HashMap` keys at the first `RandomState`: take + // them now, from the world just registered, before anything else could. + drop(std::collections::hash_map::RandomState::new()); rt.record_tape(); let strict = matches!(script, Some((_, ReplayMode::Strict))); if let Some((tape, mode)) = script { @@ -2093,7 +2278,7 @@ fn script_tape(seed: u64, spin: Option, picks: &[Option]) -> Tape { /// smallest failing tape found and the runs spent. pub fn shrink(failing: &Tape, failure: &str, body: &F, budget: usize) -> (Tape, usize) where - F: Fn(u64) -> Fut, + F: Fn(u64) -> Fut + Sync, Fut: Future, { let seed = failing.seed; @@ -2196,7 +2381,8 @@ impl Instant { Some(shared) => Self(lock(&shared.state).now), None => { static BASE: std::sync::OnceLock = std::sync::OnceLock::new(); - Self(BASE.get_or_init(std::time::Instant::now).elapsed()) + let base = *BASE.get_or_init(wall_now); + Self(wall_now().saturating_duration_since(base)) } } } @@ -2267,6 +2453,145 @@ pub fn entropy() -> u64 { } } +// --------------------------------------------------------------------------- +// Interposed libc clock and entropy (D1) +// --------------------------------------------------------------------------- + +/// Where the virtual monotonic clock starts: `std::time::Instant` on a +/// simulated thread reads this plus the virtual time elapsed. One day, so +/// code that subtracts a margin from `Instant::now()` does not underflow. +pub const VIRTUAL_BOOT: Duration = Duration::from_secs(86_400); + +/// Where the virtual wall clock starts: `SystemTime::now()` on a simulated +/// thread reads 2000-01-01T00:00:00Z plus the virtual time elapsed (an +/// obviously virtual date in logs and timestamps). +pub const VIRTUAL_EPOCH: Duration = Duration::from_secs(946_684_800); + +/// Whether this target interposes the libc clock and entropy calls (Linux +/// with glibc, macOS). Elsewhere `std::time` and OS entropy stay real under +/// the simulator, and `SPAWNED_SIM_STRICT` cannot see them. +pub const INTERPOSED: bool = spawned_sim_libc::SUPPORTED; + +/// The frames of `backtrace` below the interposed libc function named by +/// `call` (the interposer and hook frames are noise). +fn caller_frames(call: &str, backtrace: &std::backtrace::Backtrace) -> String { + let symbol = call.split('(').next().unwrap_or(call); + let text = backtrace.to_string(); + let lines: Vec<&str> = text.lines().collect(); + let is_frame = |line: &str, name: &str| { + line.trim_start() + .split_once(": ") + .is_some_and(|(n, rest)| n.chars().all(|c| c.is_ascii_digit()) && rest.trim() == name) + }; + match lines.iter().position(|line| is_frame(line, symbol)) { + Some(at) => { + let mut rest = &lines[at + 1..]; + // Skip the interposed frame's own `at` line, if any. + while rest + .first() + .is_some_and(|l| l.trim_start().starts_with("at ")) + { + rest = &rest[1..]; + } + rest.join("\n") + } + None => text, + } +} + +fn strict_mode() -> bool { + matches!( + std::env::var("SPAWNED_SIM_STRICT") + .ok() + .as_deref() + .map(str::trim), + Some("1" | "true" | "yes") + ) +} + +/// The hooks behind `spawned-sim-libc`'s interposers: on a thread that runs +/// a world, virtual time and bytes from the world's seeded entropy stream; +/// elsewhere `None`, which makes the interposer call libc. +mod interpose { + use super::{lock, splitmix, Shared, AMBIENT, CURRENT, VIRTUAL_BOOT, VIRTUAL_EPOCH}; + use spawned_sim_libc::{Clock, Hooks}; + use std::sync::atomic::Ordering; + use std::sync::Arc; + use std::time::Duration; + + static HOOKS: Hooks = Hooks { clock, random }; + + pub(super) fn install() { + spawned_sim_libc::install(&HOOKS); + } + + /// The world running on this thread (inside its `block_on` or an + /// `enter` guard), without panicking during thread teardown or while the + /// thread-locals are borrowed. Clock reads are virtual only there: a + /// real Tokio runtime driven on the same thread outside the simulation + /// keeps real time. + fn running() -> Option> { + CURRENT + .try_with(|c| c.try_borrow().ok().and_then(|c| c.last().cloned())) + .ok() + .flatten() + } + + /// The running world, else the thread's ambient one: entropy is seeded + /// whenever this thread has a world, because per-thread seeding (`HashMap` + /// keys, `thread_rng`) happens once and must not take OS bytes. + fn world() -> Option> { + running().or_else(|| { + AMBIENT + .try_with(|a| a.try_borrow().ok().and_then(|a| a.upgrade())) + .ok() + .flatten() + }) + } + + impl Shared { + /// Under strict mode, remember the first direct read and its caller; + /// the next scheduling step fails the run with it. + fn strict_read(&self, call: &'static str) { + if self.strict.load(Ordering::Relaxed) { + let mut leak = lock(&self.strict_leak); + if leak.is_none() { + *leak = Some((call, std::backtrace::Backtrace::force_capture())); + } + } + } + } + + fn clock(kind: Clock, call: &'static str) -> Option { + let shared = running()?; + shared.strict_read(call); + let now = Duration::from_nanos(shared.now_nanos.load(Ordering::Relaxed)); + Some(match kind { + Clock::Monotonic => VIRTUAL_BOOT + now, + Clock::Realtime => VIRTUAL_EPOCH + now, + }) + } + + fn random(buf: &mut [u8], call: &'static str, quiet: bool) -> bool { + let Some(shared) = world() else { + return false; + }; + if !quiet { + shared.strict_read(call); + } + for chunk in buf.chunks_mut(8) { + let draw = shared.os_entropy.fetch_add(1, Ordering::Relaxed) + 1; + // A stream of its own: separate from scheduling and from + // `spawned_rt::time::entropy`, so drawing does not shift either. + let mut x = + shared.seed ^ 0x6A09_E667_F3BC_C909 ^ draw.wrapping_mul(0xD1B5_4A32_D192_ED03); + let bytes = splitmix(&mut x).to_le_bytes(); + chunk.copy_from_slice(&bytes[..chunk.len()]); + } + true + } +} + #[cfg(test)] mod turn_budget_tests { use super::{parse_turn_budget, Duration, DEFAULT_TURN_BUDGET_MS}; diff --git a/rt/src/tasks/sim/explore.rs b/rt/src/tasks/sim/explore.rs index fef9a4c71..047a480f7 100644 --- a/rt/src/tasks/sim/explore.rs +++ b/rt/src/tasks/sim/explore.rs @@ -440,7 +440,7 @@ impl Report { /// suppressed; the failure is in the report. pub fn explore(strategy: &Strategy, first_seed: u64, runs: usize, body: &F) -> Report where - F: Fn(u64) -> Fut, + F: Fn(u64) -> Fut + Sync, Fut: Future, { run_strategy(strategy, first_seed, runs, true, body) @@ -450,7 +450,7 @@ where /// strategy hits a bug ([`Report::failures`] out of [`Report::runs`]). pub fn sweep(strategy: &Strategy, first_seed: u64, runs: usize, body: &F) -> Report where - F: Fn(u64) -> Fut, + F: Fn(u64) -> Fut + Sync, Fut: Future, { run_strategy(strategy, first_seed, runs, false, body) @@ -464,7 +464,7 @@ fn run_strategy( body: &F, ) -> Report where - F: Fn(u64) -> Fut, + F: Fn(u64) -> Fut + Sync, Fut: Future, { let mut report = Report { diff --git a/rt/src/tasks/sim/interval.rs b/rt/src/tasks/sim/interval.rs new file mode 100644 index 000000000..1bf84a1da --- /dev/null +++ b/rt/src/tasks/sim/interval.rs @@ -0,0 +1,136 @@ +//! `sleep_until` and `interval` on virtual time (determinism D3), with +//! Tokio's semantics: the first tick completes at once, and a late tick +//! follows the [`MissedTickBehavior`]. + +use super::{current_or_ambient, lock, Instant, Sleep}; +use std::future::Future; +use std::pin::Pin; +use std::task::{Context, Poll}; +use std::time::Duration; + +/// Completes once virtual time reaches `deadline` (yields once if it already +/// has, as a Tokio timer does). +pub fn sleep_until(deadline: Instant) -> Sleep { + Sleep { + shared: current_or_ambient(), + deadline: deadline.0, + key: None, + yielded: false, + } +} + +/// What an [`Interval`] does with ticks it could not deliver on time +/// (Tokio's enum, same default). +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum MissedTickBehavior { + /// Deliver the missed ticks back to back, then keep the original + /// schedule. + #[default] + Burst, + /// Schedule the next tick one period after the late one. + Delay, + /// Skip the missed ticks and keep to multiples of the period from the + /// original start. + Skip, +} + +/// How late a tick may be before [`MissedTickBehavior`] applies (Tokio's). +const LATE: Duration = Duration::from_millis(5); + +/// Ticks every `period` of virtual time; see [`interval`]. +#[derive(Debug)] +pub struct Interval { + next: Duration, + period: Duration, + behavior: MissedTickBehavior, + sleep: Option, +} + +/// An [`Interval`] whose first tick completes immediately. +/// +/// # Panics +/// If `period` is zero, as Tokio's does. +pub fn interval(period: Duration) -> Interval { + interval_at(Instant::now(), period) +} + +/// An [`Interval`] whose first tick completes at `start`. +/// +/// # Panics +/// If `period` is zero, as Tokio's does. +pub fn interval_at(start: Instant, period: Duration) -> Interval { + assert!(period > Duration::ZERO, "`period` must be non-zero."); + Interval { + next: start.0, + period, + behavior: MissedTickBehavior::default(), + sleep: None, + } +} + +impl Interval { + /// Completes at the next tick and returns its scheduled instant. + pub async fn tick(&mut self) -> Instant { + std::future::poll_fn(|cx| self.poll_tick(cx)).await + } + + /// Poll for the next tick. + pub fn poll_tick(&mut self, cx: &mut Context<'_>) -> Poll { + let next = self.next; + let sleep = self.sleep.get_or_insert_with(|| sleep_until(Instant(next))); + if Pin::new(&mut *sleep).poll(cx).is_pending() { + return Poll::Pending; + } + let now = lock(&sleep.shared.state).now; + self.sleep = None; + let fired = self.next; + // As Tokio: a tick up to 5 ms late keeps the schedule; later than + // that, the missed-tick behavior decides. + self.next = if now.saturating_sub(fired) > LATE { + match self.behavior { + MissedTickBehavior::Burst => fired + self.period, + MissedTickBehavior::Delay => now + self.period, + MissedTickBehavior::Skip => { + let late = now.saturating_sub(fired).as_nanos(); + let period = self.period.as_nanos(); + now + Duration::from_nanos( + u64::try_from(period - late % period).unwrap_or(u64::MAX), + ) + } + } + } else { + fired + self.period + }; + Poll::Ready(Instant(fired)) + } + + /// Restart the schedule: the next tick is one period from now. + pub fn reset(&mut self) { + self.next = Instant::now().0 + self.period; + self.sleep = None; + } + + /// Restart the schedule with the next tick right away. + pub fn reset_immediately(&mut self) { + self.next = Instant::now().0; + self.sleep = None; + } + + /// Restart the schedule with the next tick at `deadline`. + pub fn reset_at(&mut self, deadline: Instant) { + self.next = deadline.0; + self.sleep = None; + } + + pub fn period(&self) -> Duration { + self.period + } + + pub fn missed_tick_behavior(&self) -> MissedTickBehavior { + self.behavior + } + + pub fn set_missed_tick_behavior(&mut self, behavior: MissedTickBehavior) { + self.behavior = behavior; + } +} diff --git a/rt/src/tasks/sim/join_set.rs b/rt/src/tasks/sim/join_set.rs new file mode 100644 index 000000000..ecb545f29 --- /dev/null +++ b/rt/src/tasks/sim/join_set.rs @@ -0,0 +1,216 @@ +//! A `JoinSet` over simulated tasks (determinism D3), with Tokio's API. +//! +//! Tasks are spawned on the simulator, so their schedule is the seed's. +//! [`JoinSet::join_next`] returns a finished task's result; when several have +//! finished since the last call it returns the one with the smallest task id +//! (Tokio returns them in completion order), so the order is deterministic. + +use super::{spawn, spawn_blocking, Id, JoinError, JoinHandle}; +use std::collections::BTreeMap; +use std::fmt; +use std::future::Future; +use std::pin::Pin; +use std::sync::Arc; +use std::task::{Context, Poll, Waker}; + +/// A set of simulated tasks whose results are collected as they finish. +/// Dropping the set aborts every task still in it. +pub struct JoinSet { + tasks: BTreeMap>, +} + +/// Aborts one task of a [`JoinSet`] without removing it. +#[derive(Clone)] +pub struct AbortHandle { + abort: Arc, + finished: Arc bool + Send + Sync>, + id: Id, +} + +impl AbortHandle { + pub fn abort(&self) { + (self.abort)(); + } + pub fn is_finished(&self) -> bool { + (self.finished)() + } + pub fn id(&self) -> Id { + self.id + } +} + +impl fmt::Debug for AbortHandle { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("sim::AbortHandle") + .field("id", &self.id) + .finish() + } +} + +impl fmt::Debug for JoinSet { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("sim::JoinSet") + .field("len", &self.tasks.len()) + .finish() + } +} + +impl Default for JoinSet { + fn default() -> Self { + Self { + tasks: BTreeMap::new(), + } + } +} + +impl JoinSet { + pub fn new() -> Self { + Self::default() + } + + pub fn len(&self) -> usize { + self.tasks.len() + } + + pub fn is_empty(&self) -> bool { + self.tasks.is_empty() + } + + /// Abort every task; their results (`Cancelled`, or a value if they + /// had already finished) are still returned by `join_next`. + pub fn abort_all(&mut self) { + for handle in self.tasks.values() { + handle.abort(); + } + } + + /// Remove every task without aborting it. + pub fn detach_all(&mut self) { + self.tasks.clear(); + } +} + +impl JoinSet { + fn insert(&mut self, handle: JoinHandle) -> AbortHandle { + let (id, join, shared) = (handle.id, handle.join.clone(), handle.shared.clone()); + let finished = join.clone(); + self.tasks.insert(id, handle); + AbortHandle { + abort: Arc::new(move || super::abort_task(&join, &shared, id)), + finished: Arc::new(move || super::lock(&finished.inner).finished), + id: Id(id), + } + } + + /// Spawn `future` on the simulator as a member of the set. + pub fn spawn(&mut self, future: F) -> AbortHandle + where + F: Future + Send + 'static, + { + self.insert(spawn(future)) + } + + /// Run `f` as a member of the set: inline, as one simulator step (see + /// [`spawn_blocking`]). + pub fn spawn_blocking(&mut self, f: F) -> AbortHandle + where + F: FnOnce() -> T + Send + 'static, + { + self.insert(spawn_blocking(f)) + } + + /// The next finished task's result, or `None` if the set is empty. + pub async fn join_next(&mut self) -> Option> { + std::future::poll_fn(|cx| self.poll_join_next(cx)).await + } + + /// [`JoinSet::join_next`] with the task's id. + pub async fn join_next_with_id(&mut self) -> Option> { + std::future::poll_fn(|cx| { + self.poll_next_entry(cx) + .map(|entry| entry.map(|(id, result)| result.map(|value| (id, value)))) + }) + .await + } + + /// Poll for the next finished task. + pub fn poll_join_next(&mut self, cx: &mut Context<'_>) -> Poll>> { + self.poll_next_entry(cx) + .map(|entry| entry.map(|(_, result)| result)) + } + + /// Polls every member in task-id order (O(n) per wake; sets in tests + /// are small). + fn poll_next_entry( + &mut self, + cx: &mut Context<'_>, + ) -> Poll)>> { + if self.tasks.is_empty() { + return Poll::Ready(None); + } + let mut done = None; + for (id, handle) in self.tasks.iter_mut() { + if let Poll::Ready(result) = Pin::new(handle).poll(cx) { + done = Some((*id, result)); + break; + } + } + match done { + Some((id, result)) => { + self.tasks.remove(&id); + Poll::Ready(Some((Id(id), result))) + } + None => Poll::Pending, + } + } + + /// A finished task's result without waiting, if there is one. Polls + /// only a finished handle, so the wakers `join_next` registered on the + /// others stay in place. + pub fn try_join_next(&mut self) -> Option> { + let id = *self.tasks.iter().find(|(_, h)| h.is_finished())?.0; + let mut handle = self.tasks.remove(&id)?; + match Pin::new(&mut handle).poll(&mut Context::from_waker(Waker::noop())) { + Poll::Ready(result) => Some(result), + // Finished means its result is stored (or it was cancelled). + Poll::Pending => None, + } + } + + /// Abort every task and wait for all of them to stop. + pub async fn shutdown(&mut self) { + self.abort_all(); + while self.join_next().await.is_some() {} + } + + /// Wait for every task and return their values (in the order + /// `join_next` yields them). A task that panicked resumes the panic, as + /// Tokio does. + pub async fn join_all(mut self) -> Vec { + let mut values = Vec::with_capacity(self.len()); + while let Some(result) = self.join_next().await { + match result { + Ok(value) => values.push(value), + Err(error) if error.is_panic() => std::panic::resume_unwind(error.into_panic()), + Err(error) => panic!("{error}"), + } + } + values + } +} + +impl Drop for JoinSet { + fn drop(&mut self) { + self.abort_all(); + } +} + +impl + Send + 'static> FromIterator for JoinSet { + fn from_iter>(iter: I) -> Self { + let mut set = Self::new(); + for future in iter { + set.spawn(future); + } + set + } +} diff --git a/rt/src/tasks/sim_test.rs b/rt/src/tasks/sim_test.rs index f6cb18bf5..9b81da054 100644 --- a/rt/src/tasks/sim_test.rs +++ b/rt/src/tasks/sim_test.rs @@ -21,6 +21,7 @@ //! | `SPAWNED_SIM_ITERATIONS=n` | runs for that strategy (default 8 for random, 100 otherwise); overrides the test's `iterations` | //! | `SPAWNED_SIM_FAIR_AFTER=n` | polls after which a PCT/DFS run switches to a fair scheduler (default 20,000) | //! | `SPAWNED_SIM_TURN_BUDGET=ms` | wall-clock budget of one actor turn inside a poll (default 250, 0 = off) | +//! | `SPAWNED_SIM_STRICT=1` | a direct wall-clock or OS-entropy read on the simulated thread (`Instant::now`, `SystemTime::now`, `thread_rng`, `OsRng`) fails the run naming the caller, instead of reading virtual time or seeded bytes | use std::future::Future; use std::ops::Range; @@ -40,7 +41,7 @@ pub struct SimTest { /// Run `body` as a seeded test with the default settings. pub fn sim_test(body: F) where - F: Fn(u64) -> Fut, + F: Fn(u64) -> Fut + Sync, Fut: Future, { SimTest::new().run(body) @@ -142,7 +143,7 @@ impl SimTest { #[cfg(not(spawned_sim))] pub fn run(self, body: F) where - F: Fn(u64) -> Fut, + F: Fn(u64) -> Fut + Sync, Fut: Future, { let seed = env_u64("SPAWNED_SIM_SEED").unwrap_or(0); @@ -162,7 +163,7 @@ impl SimTest { #[cfg(spawned_sim)] pub fn run(self, body: F) where - F: Fn(u64) -> Fut, + F: Fn(u64) -> Fut + Sync, Fut: Future, { use super::sim::explore::{explore, Strategy}; @@ -175,6 +176,7 @@ impl SimTest { "SPAWNED_SIM_MAX_STEPS", "SPAWNED_SIM_FAIR_AFTER", "SPAWNED_SIM_TURN_BUDGET", + "SPAWNED_SIM_STRICT", ] .iter() .filter_map(|key| std::env::var(key).ok().map(|v| format!("{key}={v} "))) diff --git a/rt/src/tasks/tokio/mod.rs b/rt/src/tasks/tokio/mod.rs index 1705d0054..2f12c0e2b 100644 --- a/rt/src/tasks/tokio/mod.rs +++ b/rt/src/tasks/tokio/mod.rs @@ -3,6 +3,8 @@ pub mod mpsc; pub mod oneshot; pub use tokio::sync::watch; +pub use tokio::task::{AbortHandle, JoinError, JoinSet}; +pub use tokio::time::MissedTickBehavior; pub use tokio::{ runtime::{Handle, Runtime, RuntimeFlavor}, task::{ @@ -27,3 +29,57 @@ pub async fn ctrl_c() { .await .expect("Failed to listen for Ctrl+C"); } + +/// Completes at `deadline` (Tokio's `sleep_until`, taking the +/// `spawned_rt::time::Instant` that is virtual under simulation). +pub fn sleep_until(deadline: std::time::Instant) -> tokio::time::Sleep { + tokio::time::sleep_until(deadline.into()) +} + +/// Tokio's `Interval`, with instants as `spawned_rt::time::Instant` so the +/// same code runs on virtual time under `--cfg spawned_sim`. +#[derive(Debug)] +pub struct Interval(tokio::time::Interval); + +/// An [`Interval`] whose first tick completes immediately (Tokio's +/// `interval`). +pub fn interval(period: std::time::Duration) -> Interval { + Interval(tokio::time::interval(period)) +} + +/// An [`Interval`] whose first tick completes at `start` (Tokio's +/// `interval_at`). +pub fn interval_at(start: std::time::Instant, period: std::time::Duration) -> Interval { + Interval(tokio::time::interval_at(start.into(), period)) +} + +impl Interval { + /// Completes at the next tick and returns its scheduled instant. + pub async fn tick(&mut self) -> std::time::Instant { + self.0.tick().await.into_std() + } + pub fn poll_tick( + &mut self, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll { + self.0.poll_tick(cx).map(tokio::time::Instant::into_std) + } + pub fn reset(&mut self) { + self.0.reset() + } + pub fn reset_immediately(&mut self) { + self.0.reset_immediately() + } + pub fn reset_at(&mut self, deadline: std::time::Instant) { + self.0.reset_at(deadline.into()) + } + pub fn period(&self) -> std::time::Duration { + self.0.period() + } + pub fn missed_tick_behavior(&self) -> MissedTickBehavior { + self.0.missed_tick_behavior() + } + pub fn set_missed_tick_behavior(&mut self, behavior: MissedTickBehavior) { + self.0.set_missed_tick_behavior(behavior) + } +} diff --git a/rt/tests/blocking_and_timers.rs b/rt/tests/blocking_and_timers.rs new file mode 100644 index 000000000..54bc73539 --- /dev/null +++ b/rt/tests/blocking_and_timers.rs @@ -0,0 +1,235 @@ +//! Determinism D2 and D3: `spawn_blocking`, `block_in_place`, `sleep_until`, +//! `interval` and `JoinSet` through `spawned_rt::tasks`, on Tokio in normal +//! builds and on the simulator (virtual time, one thread) under +//! `--cfg spawned_sim`. The `sim` module checks what only the simulator +//! guarantees: exact virtual instants, inline blocking work, and the failure +//! a direct `tokio::time` call produces. +#![cfg(feature = "async")] + +use spawned_rt::tasks::{self as rt, MissedTickBehavior, SimTest}; +use spawned_rt::time::Instant; +use std::time::Duration; + +const MS: Duration = Duration::from_millis(1); + +#[test] +fn spawn_blocking_returns_its_value() { + rt::sim_test(|seed| async move { + let doubled = rt::spawn_blocking(move || seed * 2).await.unwrap(); + assert_eq!(doubled, seed * 2); + let panicked = rt::spawn_blocking(|| -> u8 { panic!("blocking work failed") }).await; + assert!(panicked.unwrap_err().is_panic()); + }); +} + +#[test] +fn block_in_place_returns_its_value() { + // Tokio needs a multi-thread runtime for block_in_place. + SimTest::new() + .multi_thread(2) + .run(|_| async { assert_eq!(rt::block_in_place(|| 21 * 2), 42) }); +} + +#[test] +fn sleep_until_waits_for_the_deadline() { + rt::sim_test(|_| async { + let start = Instant::now(); + rt::sleep_until(start + 20 * MS).await; + assert!(start.elapsed() >= 20 * MS); + // A deadline in the past completes. + rt::sleep_until(start).await; + }); +} + +#[test] +fn interval_ticks_every_period() { + rt::sim_test(|_| async { + let mut ticks = rt::interval(10 * MS); + assert_eq!(ticks.period(), 10 * MS); + let first = ticks.tick().await; + let second = ticks.tick().await; + let third = ticks.tick().await; + assert_eq!(second - first, 10 * MS); + assert_eq!(third - first, 20 * MS); + assert_eq!(ticks.missed_tick_behavior(), MissedTickBehavior::Burst); + }); +} + +#[test] +fn join_set_collects_every_result() { + rt::sim_test(|_| async { + let mut set = rt::JoinSet::new(); + for n in 0..5u64 { + set.spawn(async move { + rt::sleep(Duration::from_millis(n)).await; + n + }); + } + set.spawn_blocking(|| 100); + assert_eq!(set.len(), 6); + let mut seen = Vec::new(); + while let Some(result) = set.join_next().await { + seen.push(result.unwrap()); + } + seen.sort_unstable(); + assert_eq!(seen, [0, 1, 2, 3, 4, 100]); + assert!(set.is_empty()); + assert!(set.join_next().await.is_none()); + + let mut set = rt::JoinSet::new(); + let handle = set.spawn(async { + rt::sleep(Duration::from_secs(3600)).await; + 1u8 + }); + handle.abort(); + assert!(set.join_next().await.unwrap().unwrap_err().is_cancelled()); + + let set: rt::JoinSet = (1..=3u32).map(|n| async move { n * 10 }).collect(); + let mut all = set.join_all().await; + all.sort_unstable(); + assert_eq!(all, [10, 20, 30]); + }); +} + +#[cfg(spawned_sim)] +mod sim { + use super::MS; + use spawned_rt::tasks::{self as rt, sim, MissedTickBehavior}; + use spawned_rt::time::Instant; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::{Arc, Mutex}; + use std::time::Duration; + + /// D2: the closure runs on the simulated thread as one step of a new + /// task: after the spawner yields, in virtual time, never concurrently. + #[test] + fn spawn_blocking_runs_inline_as_one_step() { + let outcome = sim::run_once(3, None, &|_| async { + let thread = std::thread::current().id(); + let ran = Arc::new(AtomicBool::new(false)); + let flag = ran.clone(); + let handle = rt::spawn_blocking(move || { + flag.store(true, Ordering::SeqCst); + (std::thread::current().id(), Instant::now()) + }); + // Not run at the spawn: it is a task the scheduler polls. + assert!(!ran.load(Ordering::SeqCst)); + rt::sleep(5 * MS).await; + let (ran_on, at) = handle.await.unwrap(); + assert_eq!(ran_on, thread); + assert!(at < Instant::now()); + }); + assert_eq!(outcome.failure, None); + } + + /// A closure that blocks on async work in the same world cannot finish + /// on one thread: a nested `block_on` fails the run even though the + /// panic lands in a task nobody joins. + #[test] + fn nested_block_on_in_spawn_blocking_fails_the_run() { + let outcome = sim::run_once(1, None, &|_| async { + drop(rt::spawn_blocking(|| rt::block_on(rt::sleep(MS)))); + rt::sleep(10 * MS).await; + }); + let failure = outcome.failure.expect("must fail"); + assert!(failure.contains("nested block_on"), "{failure}"); + } + + #[test] + fn timers_are_exact_in_virtual_time() { + let outcome = sim::run_once(2, None, &|_| async { + let start = Instant::now(); + rt::sleep_until(start + 7 * MS).await; + assert_eq!(start.elapsed(), 7 * MS); + + let base = Instant::now(); + let mut burst = rt::interval(10 * MS); + assert_eq!(burst.tick().await, base); + rt::sleep(35 * MS).await; + // Burst: the missed ticks come back to back, then the schedule. + assert_eq!(burst.tick().await, base + 10 * MS); + assert_eq!(burst.tick().await, base + 20 * MS); + assert_eq!(burst.tick().await, base + 30 * MS); + assert_eq!(burst.tick().await, base + 40 * MS); + assert_eq!(Instant::now(), base + 40 * MS); + + // Up to 5 ms late keeps the schedule, as on Tokio. + let base = Instant::now(); + let mut slightly_late = rt::interval_at(base, 10 * MS); + slightly_late.set_missed_tick_behavior(MissedTickBehavior::Delay); + slightly_late.tick().await; + rt::sleep(13 * MS).await; + assert_eq!(slightly_late.tick().await, base + 10 * MS); + assert_eq!(slightly_late.tick().await, base + 20 * MS); + + for (behavior, next) in [ + (MissedTickBehavior::Delay, 45), + (MissedTickBehavior::Skip, 40), + ] { + let base = Instant::now(); + let mut ticks = rt::interval_at(base, 10 * MS); + ticks.set_missed_tick_behavior(behavior); + ticks.tick().await; + rt::sleep(35 * MS).await; + assert_eq!(ticks.tick().await, base + 10 * MS); + assert_eq!(ticks.tick().await, base + next * MS, "{behavior:?}"); + } + }); + assert_eq!(outcome.failure, None); + } + + /// `join_next` returns tasks in completion order in virtual time, and + /// tasks that finish together by task id: the same for every run. + #[test] + fn join_set_order_is_deterministic() { + let order = |seed| { + let seen = Arc::new(Mutex::new(Vec::new())); + let out = seen.clone(); + let outcome = sim::run_once(seed, None, &move |_| { + let out = out.clone(); + async move { + let mut set = rt::JoinSet::new(); + for n in [30u64, 10, 20, 10, 0] { + set.spawn(async move { + rt::sleep(Duration::from_millis(n)).await; + n + }); + } + while let Some(n) = set.join_next().await { + out.lock().unwrap().push(n.unwrap()); + } + } + }); + assert_eq!(outcome.failure, None); + let seen = seen.lock().unwrap().clone(); + seen + }; + let first = order(4); + assert_eq!(first, [0, 10, 10, 20, 30]); + assert_eq!(order(4), first); + } + + /// D3: a direct Tokio timer fails the run with a message naming the + /// `spawned_rt` replacement, from the root and from a detached task. + #[test] + fn direct_tokio_time_names_the_replacement() { + let root = sim::run_once(1, None, &|_| async { + tokio::time::sleep(MS).await; + }); + let failure = root.failure.expect("must fail"); + assert!( + failure.contains("direct tokio::time call") + && failure.contains("spawned_rt::tasks::{sleep, sleep_until, timeout, interval}"), + "{failure}" + ); + + let detached = sim::run_once(1, None, &|_| async { + drop(rt::spawn(async { + let _ = tokio::time::timeout(MS, std::future::pending::<()>()).await; + })); + rt::sleep(10 * MS).await; + }); + let failure = detached.failure.expect("must fail"); + assert!(failure.contains("direct tokio::time call"), "{failure}"); + } +} diff --git a/rt/tests/sim_determinism.rs b/rt/tests/sim_determinism.rs new file mode 100644 index 000000000..d59b245a1 --- /dev/null +++ b/rt/tests/sim_determinism.rs @@ -0,0 +1,205 @@ +//! Determinism D1: under `--cfg spawned_sim`, wall-clock and OS-entropy reads +//! on the simulated thread come from the world (virtual time, seeded bytes) +//! through `spawned-sim-libc`'s interposers; other threads read the real +//! clock; strict mode turns such a read into a failure naming the caller. +#![cfg(all(spawned_sim, feature = "async"))] + +use rand::{Rng, RngCore}; +use spawned_rt::tasks::{self as rt, sim}; +use std::collections::{HashMap, HashSet}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +/// Everything a run observed through std, rand and the OS. +#[derive(Debug, Clone, PartialEq, Eq)] +struct Observed { + instant_debug: String, + instant_steps: Vec, + system_time: Vec, + thread_rng: Vec, + os_rng: u64, + getrandom: [u8; 16], + map_order: Vec, + set_order: Vec, + /// A thread that does not run the world read a real date. + other_thread_real: bool, +} + +async fn observe(out: Arc>>) { + let start = std::time::Instant::now(); + let mut instant_steps = Vec::new(); + let mut system_time = Vec::new(); + for step in [0u64, 250, 1_500] { + rt::sleep(Duration::from_millis(step)).await; + instant_steps.push(start.elapsed()); + system_time.push(SystemTime::now().duration_since(UNIX_EPOCH).unwrap()); + } + let mut rng = rand::thread_rng(); + let thread_rng = (0..4).map(|_| rng.gen::()).collect(); + let os_rng = rand::rngs::OsRng.next_u64(); + let mut getrandom = [0u8; 16]; + rand::rngs::OsRng.fill_bytes(&mut getrandom); + let map: HashMap = (0..64).map(|k| (k, ())).collect(); + let set: HashSet = (0..64).collect(); + // A thread that does not run the world reads the real clock. + let other_thread_wall = + std::thread::spawn(|| SystemTime::now().duration_since(UNIX_EPOCH).unwrap()) + .join() + .unwrap(); + out.lock().unwrap().push(Observed { + instant_debug: format!("{start:?}"), + instant_steps, + system_time, + thread_rng, + os_rng, + getrandom, + map_order: map.keys().copied().collect(), + set_order: set.iter().copied().collect(), + other_thread_real: other_thread_wall > Duration::from_secs(1_700_000_000), + }); +} + +fn run(seed: u64) -> Observed { + let out = Arc::new(Mutex::new(Vec::new())); + let outcome = sim::run_once(seed, None, &|_| observe(out.clone())); + assert_eq!(outcome.failure, None); + let observed = out.lock().unwrap().pop().unwrap(); + observed +} + +#[test] +fn clock_and_entropy_are_virtual_and_seeded() { + if !sim::INTERPOSED { + eprintln!("this target does not interpose libc; skipped"); + return; + } + let a = run(7); + // Virtual time: exact, starting at the virtual epoch. + assert_eq!( + a.instant_steps, + [ + Duration::ZERO, + Duration::from_millis(250), + Duration::from_millis(1_750) + ] + ); + assert_eq!( + a.system_time, + [ + sim::VIRTUAL_EPOCH, + sim::VIRTUAL_EPOCH + Duration::from_millis(250), + sim::VIRTUAL_EPOCH + Duration::from_millis(1_750), + ] + ); + // Another thread: real time (well after the virtual epoch of 2000). + assert!(a.other_thread_real); + + // Same seed, a later run in this process: identical observations, + // including the absolute `Instant`, `HashMap` order and `thread_rng`. + assert_eq!(run(7), a); + + // A different seed draws different bytes and orders maps differently. + let b = run(8); + assert_ne!(b.thread_rng, a.thread_rng); + assert_ne!(b.os_rng, a.os_rng); + assert_ne!(b.getrandom, a.getrandom); + assert_ne!(b.map_order, a.map_order); + // Time does not depend on the seed. + assert_eq!(b.system_time, a.system_time); +} + +/// A seed observed in the middle of a sweep matches the same seed run alone +/// (each run has a fresh thread, so per-thread seeding happens inside it). +#[test] +fn a_seed_in_a_sweep_matches_the_seed_alone() { + if !sim::INTERPOSED { + return; + } + let sweep: Vec = (0..4).map(run).collect(); + assert_eq!(run(2), sweep[2]); +} + +/// The same body replayed from its recorded tape sees the same values. +#[test] +fn a_tape_replay_sees_the_same_values() { + if !sim::INTERPOSED { + return; + } + let out = Arc::new(Mutex::new(Vec::new())); + let body = |_| observe(out.clone()); + let recorded = sim::run_once(5, None, &body); + let replayed = sim::run_once(5, Some(&recorded.tape), &body); + assert_eq!(replayed.failure, None); + let seen = out.lock().unwrap().clone(); + assert_eq!(seen[0], seen[1]); +} + +#[test] +fn strict_mode_fails_the_run_naming_the_caller() { + if !sim::INTERPOSED { + return; + } + #[inline(never)] + fn reads_the_wall_clock() -> SystemTime { + SystemTime::now() + } + let outcome = sim::run_once(1, None, &|_| async { + sim::set_strict(true); + // Not reported: `HashMap` keys, and the routed clock. + let map: HashMap = HashMap::new(); + drop(map); + let _ = spawned_rt::time::Instant::now(); + rt::sleep(Duration::from_millis(1)).await; + let _ = reads_the_wall_clock(); + rt::sleep(Duration::from_millis(1)).await; + }); + let failure = outcome.failure.expect("strict mode must fail the run"); + assert!( + failure.contains("determinism leak: clock_gettime(CLOCK_REALTIME)") + && failure.contains("SPAWNED_SIM_STRICT=1") + && failure.contains("reads_the_wall_clock"), + "{failure}" + ); + + let outcome = sim::run_once(1, None, &|_| async { + sim::set_strict(true); + let _: u64 = rand::thread_rng().gen(); + rt::yield_now().await; + }); + let failure = outcome.failure.expect("strict mode must fail the run"); + assert!(failure.contains("determinism leak"), "{failure}"); +} + +/// Run by `separate_processes_observe_the_same_values` in child processes. +#[test] +#[ignore = "run in a child process by separate_processes_observe_the_same_values"] +fn observe_child() { + let seed = std::env::var("OBSERVE_SEED").unwrap().parse().unwrap(); + println!("OBSERVED {:?}", run(seed)); +} + +/// The same seed in two separate processes (fresh process-wide state, a +/// different OS entropy pool) observes identical values. +#[test] +fn separate_processes_observe_the_same_values() { + if !sim::INTERPOSED { + return; + } + let observe = |seed: u64| { + let output = std::process::Command::new(std::env::current_exe().unwrap()) + .args(["--ignored", "--exact", "observe_child", "--nocapture"]) + .env("OBSERVE_SEED", seed.to_string()) + .output() + .unwrap(); + let stdout = String::from_utf8_lossy(&output.stdout).into_owned(); + assert!(output.status.success(), "{stdout}"); + stdout + .lines() + .find(|line| line.starts_with("OBSERVED ")) + .unwrap_or_else(|| panic!("no observation in {stdout}")) + .to_owned() + }; + let first = observe(11); + assert_eq!(observe(11), first); + assert_ne!(observe(12), first); +} diff --git a/sim-libc/Cargo.toml b/sim-libc/Cargo.toml new file mode 100644 index 000000000..4cfd9bc27 --- /dev/null +++ b/sim-libc/Cargo.toml @@ -0,0 +1,17 @@ +[package] +name = "spawned-sim-libc" +description = "Clock and entropy interposers for Spawned's deterministic simulator (compiled only under --cfg spawned_sim)" +version.workspace = true +edition.workspace = true +license.workspace = true +repository.workspace = true + +# The whole crate is `#![cfg(spawned_sim)]`: in every other build it is empty, +# so no unsafe code is compiled outside simulation builds. spawned-rt depends +# on it only under `cfg(spawned_sim)`. + +[dependencies] +libc = "0.2" + +[lints.rust] +unexpected_cfgs = { level = "warn", check-cfg = ['cfg(spawned_sim)'] } diff --git a/sim-libc/src/lib.rs b/sim-libc/src/lib.rs new file mode 100644 index 000000000..f9362326f --- /dev/null +++ b/sim-libc/src/lib.rs @@ -0,0 +1,444 @@ +//! Clock and entropy interposers for Spawned's deterministic simulator. +//! +//! **Simulation builds only.** The whole crate is `#![cfg(spawned_sim)]`: in +//! any other build it compiles to nothing, so the unsafe code below never +//! exists outside `--cfg spawned_sim`. `spawned-rt` depends on it only under +//! that cfg and keeps `#![forbid(unsafe_code)]` in every build; this crate is +//! where the unsafe FFI the technique needs is contained. +//! +//! The technique is madsim's: the test binary defines the libc symbols that +//! read the clock and the OS entropy source (`clock_gettime`, +//! `gettimeofday`, `getrandom`, `getentropy`, and on Linux the `syscall` +//! entry for `SYS_getrandom`; `CCRandomGenerateBytes` on macOS). References +//! from Rust code (std's `Instant::now`, `SystemTime::now`, `HashMap`'s +//! `RandomState` keys, the `getrandom` 0.2 crate behind `rand::thread_rng` +//! and `OsRng`, `chrono::Utc::now`) bind to these definitions at link time. +//! Each one asks the [`Hooks`] that `spawned-rt` installed: on a thread that +//! runs a simulated world, the hook answers with virtual time or seeded bytes; +//! on every other thread (and for the clock inside [`real_clock`]) the call goes to the +//! real libc function (found with `dlsym(RTLD_NEXT)`). +//! +//! Supported targets: Linux with glibc (x86_64 and aarch64 for the `syscall` +//! entry) and macOS. Elsewhere [`SUPPORTED`] is false and nothing is +//! interposed. Not covered: code that finds `getrandom` with +//! `dlsym(RTLD_DEFAULT)` at run time (the `getrandom` 0.3+ crate on Linux, +//! that is `rand` 0.9+), unless the binary exports the symbol +//! (`-C link-arg=-Wl,--export-dynamic-symbol=getrandom`); raw `syscall` +//! instructions; and the vDSO called directly. +//! +//! Caveats: the interposers are not async-signal-safe (they read +//! thread-locals, clone an `Arc`, and in strict mode lock a mutex and capture +//! a backtrace), so a signal handler that reads the clock on a simulated +//! thread can deadlock. The `syscall` entry sits in front of every +//! `libc::syscall` call in the binary (std's futex and `statx` among them); +//! numbers other than `SYS_getrandom` go straight to libc. Only one copy of +//! this crate (and no other interposer such as madsim's) may be linked: two +//! would define the same symbols. +#![cfg(spawned_sim)] +#![deny(unsafe_op_in_unsafe_fn)] +#![warn(clippy::undocumented_unsafe_blocks)] +// Without interposers (unsupported targets) the helpers below go unused. +#![cfg_attr( + not(any(all(target_os = "linux", target_env = "gnu"), target_os = "macos")), + allow(dead_code) +)] + +use std::cell::Cell; +use std::panic::{catch_unwind, AssertUnwindSafe}; +use std::sync::atomic::{AtomicPtr, Ordering}; +use std::time::Duration; + +/// Which clock a read asked for. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Clock { + /// `CLOCK_MONOTONIC` and its variants (`Instant::now`): time since boot. + Monotonic, + /// `CLOCK_REALTIME` and `gettimeofday` (`SystemTime::now`): time since + /// the Unix epoch. + Realtime, +} + +/// Callbacks `spawned-rt` installs. They run inside a libc call, so they must +/// not panic (a panic is caught and the real function used) and must not hold +/// a lock that a clock read on the same thread could need; reads they make +/// themselves go to the real functions. +#[derive(Debug, Clone, Copy)] +pub struct Hooks { + /// A clock read: the time to report (since boot for + /// [`Clock::Monotonic`], since the Unix epoch for [`Clock::Realtime`]), + /// or `None` for the real clock. The `&str` names the call. + pub clock: fn(Clock, &'static str) -> Option, + /// An entropy read: fill the buffer and return `true`, or `false` for + /// real bytes. The `&str` names the call; the flag marks reads std makes + /// for `HashMap` keys (`GRND_INSECURE`, `CCRandomGenerateBytes`), which a + /// strict mode should not report. + pub random: fn(&mut [u8], &'static str, bool) -> bool, +} + +/// Whether this target has interposers. Where it is false, [`install`] does +/// nothing and every read is real. +pub const SUPPORTED: bool = cfg!(any( + all(target_os = "linux", target_env = "gnu"), + target_os = "macos" +)); + +static HOOKS: AtomicPtr = AtomicPtr::new(std::ptr::null_mut()); + +/// Route intercepted reads through `hooks` from now on (process-wide; +/// idempotent). +pub fn install(hooks: &'static Hooks) { + HOOKS.store(hooks as *const Hooks as *mut Hooks, Ordering::Release); + imp::ensure_linked(); +} + +thread_local! { + /// Set while this thread runs a hook, so the hook's own reads (a lock, + /// a backtrace) go to libc instead of recursing. A const `Cell` has no + /// destructor, so it is readable during thread teardown. + static IN_HOOK: Cell = const { Cell::new(false) }; + /// Nonzero inside [`real_clock`]: clock reads go to libc. + static REAL_CLOCK: Cell = const { Cell::new(0) }; +} + +struct InHook; +impl InHook { + /// `None` if this thread is already inside a hook (or tearing down). + fn enter() -> Option { + let entered = IN_HOOK.try_with(|h| !h.replace(true)); + // `then`, not `then_some`: a guard built and dropped on the refusing + // path would clear the flag. + entered.ok()?.then(|| Self) + } +} +impl Drop for InHook { + fn drop(&mut self) { + let _ = IN_HOOK.try_with(|h| h.set(false)); + } +} + +/// Run `f` with the real clock on this thread (the simulator's own +/// wall-clock watchdog and turn budget, and Tokio's runtime metrics around a +/// simulated poll, use it). Entropy reads are still seeded: a `HashMap` +/// created here must not take this thread's keys from the OS. +pub fn real_clock(f: impl FnOnce() -> R) -> R { + struct Guard; + impl Drop for Guard { + fn drop(&mut self) { + let _ = REAL_CLOCK.try_with(|d| d.set(d.get().saturating_sub(1))); + } + } + let _ = REAL_CLOCK.try_with(|d| d.set(d.get() + 1)); + let _guard = Guard; + f() +} + +/// Run `f` with the clock interposed again, inside a [`real_clock`] (the +/// simulator polls a task, whose reads are the program's, from inside Tokio +/// machinery whose own reads are not). +pub fn intercepted(f: impl FnOnce() -> R) -> R { + struct Restore(u32); + impl Drop for Restore { + fn drop(&mut self) { + let depth = self.0; + let _ = REAL_CLOCK.try_with(|d| d.set(depth)); + } + } + let depth = REAL_CLOCK.try_with(|d| d.replace(0)).unwrap_or(0); + let _restore = Restore(depth); + f() +} + +/// Ask the installed hooks, unless none are installed or this thread is +/// already inside one. +fn ask(f: impl FnOnce(&Hooks) -> Option) -> Option { + let hooks = HOOKS.load(Ordering::Acquire); + if hooks.is_null() { + return None; + } + let _in_hook = InHook::enter()?; + // SAFETY: `HOOKS` only ever holds a `&'static Hooks` (see `install`). + let hooks = unsafe { &*hooks }; + catch_unwind(AssertUnwindSafe(|| f(hooks))).ok().flatten() +} + +fn clock(kind: Clock, call: &'static str) -> Option { + if REAL_CLOCK.try_with(|d| d.get() > 0).unwrap_or(true) { + return None; + } + ask(|hooks| (hooks.clock)(kind, call)) +} + +/// Fill `len` bytes at `buf` from the hooks; `false` means use the real call. +/// +/// # Safety +/// `buf` must be valid for writes of `len` bytes. +unsafe fn random(buf: *mut u8, len: usize, call: &'static str, quiet: bool) -> bool { + // A slice may not exceed `isize::MAX` bytes: forward such a request. + if (buf.is_null() && len > 0) || len > isize::MAX as usize { + return false; + } + ask(|hooks| { + let bytes: &mut [u8] = if len == 0 { + &mut [] + } else { + // SAFETY: the caller guarantees `buf` is valid for `len` writes; + // zeroing it first makes every byte initialized before the slice + // is formed. + unsafe { + std::ptr::write_bytes(buf, 0, len); + std::slice::from_raw_parts_mut(buf, len) + } + }; + (hooks.random)(bytes, call, quiet).then_some(()) + }) + .is_some() +} + +fn to_timespec(t: Duration) -> libc::timespec { + libc::timespec { + tv_sec: t.as_secs() as libc::time_t, + tv_nsec: t.subsec_nanos() as _, + } +} + +#[cfg(any(all(target_os = "linux", target_env = "gnu"), target_os = "macos"))] +mod imp { + use super::{clock, random, to_timespec, Clock}; + use libc::{c_int, c_void, clockid_t, size_t, timespec, timeval}; + use std::ffi::CStr; + use std::sync::atomic::{AtomicUsize, Ordering}; + + /// Address of the real `name` (the next definition after this binary's, + /// in libc), cached in `cache`. Aborts if there is none: every interposed + /// symbol exists in the platform libc. + fn real(cache: &AtomicUsize, name: &CStr) -> usize { + let mut addr = cache.load(Ordering::Relaxed); + if addr == 0 { + // SAFETY: `name` is a valid C string; RTLD_NEXT is a valid handle. + addr = unsafe { libc::dlsym(libc::RTLD_NEXT, name.as_ptr()) } as usize; + if addr == 0 { + // SAFETY: abort has no preconditions. + unsafe { libc::abort() }; + } + cache.store(addr, Ordering::Relaxed); + } + addr + } + + fn classify(clk: clockid_t) -> Option<(Clock, &'static str)> { + #[cfg(target_os = "linux")] + let kind = match clk { + libc::CLOCK_REALTIME | libc::CLOCK_REALTIME_COARSE | libc::CLOCK_TAI => Clock::Realtime, + libc::CLOCK_MONOTONIC + | libc::CLOCK_MONOTONIC_RAW + | libc::CLOCK_MONOTONIC_COARSE + | libc::CLOCK_BOOTTIME => Clock::Monotonic, + _ => return None, + }; + #[cfg(target_os = "macos")] + let kind = match clk { + libc::CLOCK_REALTIME => Clock::Realtime, + libc::CLOCK_MONOTONIC | libc::CLOCK_MONOTONIC_RAW | libc::CLOCK_UPTIME_RAW => { + Clock::Monotonic + } + _ => return None, + }; + Some(match kind { + Clock::Realtime => (kind, "clock_gettime(CLOCK_REALTIME)"), + Clock::Monotonic => (kind, "clock_gettime(CLOCK_MONOTONIC)"), + }) + } + + /// Interposed `clock_gettime`: virtual time for the monotonic and + /// realtime clocks on a simulated thread. + /// + /// # Safety + /// Same contract as libc's `clock_gettime`. + #[no_mangle] + pub unsafe extern "C" fn clock_gettime(clk: clockid_t, tp: *mut timespec) -> c_int { + if !tp.is_null() { + if let Some((kind, call)) = classify(clk) { + if let Some(t) = clock(kind, call) { + // SAFETY: `tp` is non-null and, per the libc contract, + // valid for writes. + unsafe { tp.write(to_timespec(t)) }; + return 0; + } + } + } + static REAL: AtomicUsize = AtomicUsize::new(0); + let addr = real(&REAL, c"clock_gettime"); + // SAFETY: `addr` is libc's `clock_gettime`, which has this signature. + let f: unsafe extern "C" fn(clockid_t, *mut timespec) -> c_int = + unsafe { std::mem::transmute(addr) }; + // SAFETY: forwarding the caller's arguments unchanged. + unsafe { f(clk, tp) } + } + + /// Interposed `gettimeofday`: virtual realtime on a simulated thread. + /// + /// # Safety + /// Same contract as libc's `gettimeofday`. + #[no_mangle] + pub unsafe extern "C" fn gettimeofday(tv: *mut timeval, tz: *mut c_void) -> c_int { + // A caller asking for the (obsolete) timezone gets libc's answer. + if !tv.is_null() && tz.is_null() { + if let Some(t) = clock(Clock::Realtime, "gettimeofday") { + // SAFETY: `tv` is non-null and valid for writes per the + // libc contract. + unsafe { + tv.write(timeval { + tv_sec: t.as_secs() as libc::time_t, + tv_usec: t.subsec_micros() as _, + }) + }; + return 0; + } + } + static REAL: AtomicUsize = AtomicUsize::new(0); + let addr = real(&REAL, c"gettimeofday"); + // SAFETY: `addr` is libc's `gettimeofday`, which has this signature. + let f: unsafe extern "C" fn(*mut timeval, *mut c_void) -> c_int = + unsafe { std::mem::transmute(addr) }; + // SAFETY: forwarding the caller's arguments unchanged. + unsafe { f(tv, tz) } + } + + /// Interposed `getentropy`: seeded bytes on a simulated thread. + /// + /// # Safety + /// Same contract as libc's `getentropy`. + #[no_mangle] + pub unsafe extern "C" fn getentropy(buf: *mut c_void, len: size_t) -> c_int { + // SAFETY: the libc contract makes `buf` valid for `len` writes. + if len <= 256 && unsafe { random(buf.cast(), len, "getentropy", false) } { + return 0; + } + static REAL: AtomicUsize = AtomicUsize::new(0); + let addr = real(&REAL, c"getentropy"); + // SAFETY: `addr` is libc's `getentropy`, which has this signature. + let f: unsafe extern "C" fn(*mut c_void, size_t) -> c_int = + unsafe { std::mem::transmute(addr) }; + // SAFETY: forwarding the caller's arguments unchanged. + unsafe { f(buf, len) } + } + + /// `GRND_INSECURE`: std asks for `HashMap` keys with it. + #[cfg(target_os = "linux")] + const GRND_INSECURE: libc::c_uint = 0x0004; + + /// Interposed `getrandom` (std's `HashMap` keys and `std::random` bind + /// to it through a weak symbol): seeded bytes on a simulated thread. + /// + /// # Safety + /// Same contract as libc's `getrandom`. + #[cfg(target_os = "linux")] + #[no_mangle] + pub unsafe extern "C" fn getrandom( + buf: *mut c_void, + len: size_t, + flags: libc::c_uint, + ) -> libc::ssize_t { + let quiet = flags & GRND_INSECURE != 0; + // SAFETY: the libc contract makes `buf` valid for `len` writes. + if unsafe { random(buf.cast(), len, "getrandom", quiet) } { + return len as libc::ssize_t; + } + static REAL: AtomicUsize = AtomicUsize::new(0); + let addr = real(&REAL, c"getrandom"); + // SAFETY: `addr` is libc's `getrandom`, which has this signature. + let f: unsafe extern "C" fn(*mut c_void, size_t, libc::c_uint) -> libc::ssize_t = + unsafe { std::mem::transmute(addr) }; + // SAFETY: forwarding the caller's arguments unchanged. + unsafe { f(buf, len, flags) } + } + + /// Interposed `syscall`, for the `getrandom` crate 0.2 (behind + /// `rand::thread_rng` and `OsRng` in rand 0.8), which calls + /// `syscall(SYS_getrandom, ..)`. Every other number goes straight to + /// libc. Defined with six fixed arguments: on x86_64 and aarch64 Linux a + /// variadic call passes integer arguments exactly like a fixed one, and + /// reading an argument the caller did not pass yields an unused value + /// (libc's own `syscall` reads six too). + /// + /// # Safety + /// Same contract as libc's `syscall`. + #[cfg(all( + target_os = "linux", + any(target_arch = "x86_64", target_arch = "aarch64") + ))] + #[no_mangle] + pub unsafe extern "C" fn syscall( + n: libc::c_long, + a1: libc::c_long, + a2: libc::c_long, + a3: libc::c_long, + a4: libc::c_long, + a5: libc::c_long, + a6: libc::c_long, + ) -> libc::c_long { + if n == libc::SYS_getrandom { + let quiet = (a3 as libc::c_uint) & GRND_INSECURE != 0; + // SAFETY: for SYS_getrandom, `a1` is a buffer valid for `a2` + // writes per the syscall's contract. + if a2 >= 0 + && unsafe { random(a1 as *mut u8, a2 as usize, "syscall(SYS_getrandom)", quiet) } + { + return a2; + } + } + static REAL: AtomicUsize = AtomicUsize::new(0); + let addr = real(&REAL, c"syscall"); + // SAFETY: `addr` is libc's variadic `syscall`. + let f: unsafe extern "C" fn(libc::c_long, ...) -> libc::c_long = + unsafe { std::mem::transmute(addr) }; + // SAFETY: forwarding the caller's arguments (and at worst unused + // values in the trailing slots) unchanged. + unsafe { f(n, a1, a2, a3, a4, a5, a6) } + } + + /// Interposed `CCRandomGenerateBytes` (std's `HashMap` keys on macOS). + /// + /// # Safety + /// Same contract as CommonCrypto's `CCRandomGenerateBytes`. + #[cfg(target_os = "macos")] + #[no_mangle] + pub unsafe extern "C" fn CCRandomGenerateBytes(buf: *mut c_void, len: size_t) -> i32 { + // SAFETY: the contract makes `buf` valid for `len` writes. + if unsafe { random(buf.cast(), len, "CCRandomGenerateBytes", true) } { + return 0; + } + static REAL: AtomicUsize = AtomicUsize::new(0); + let addr = real(&REAL, c"CCRandomGenerateBytes"); + // SAFETY: `addr` is CommonCrypto's function, which has this signature. + let f: unsafe extern "C" fn(*mut c_void, size_t) -> i32 = + unsafe { std::mem::transmute(addr) }; + // SAFETY: forwarding the caller's arguments unchanged. + unsafe { f(buf, len) } + } + + /// Reference every interposer so the linker keeps the object that + /// defines them (and so their definitions win over libc's). + #[inline(never)] + pub(super) fn ensure_linked() { + std::hint::black_box([ + clock_gettime as *const (), + gettimeofday as *const (), + getentropy as *const (), + ]); + #[cfg(target_os = "linux")] + std::hint::black_box(getrandom as *const ()); + #[cfg(all( + target_os = "linux", + any(target_arch = "x86_64", target_arch = "aarch64") + ))] + std::hint::black_box(syscall as *const ()); + #[cfg(target_os = "macos")] + std::hint::black_box(CCRandomGenerateBytes as *const ()); + } +} + +#[cfg(not(any(all(target_os = "linux", target_env = "gnu"), target_os = "macos")))] +mod imp { + pub(super) fn ensure_linked() {} +}