Compare commits
5
Commits
b0c9685c89
...
7fcc9f2ec5
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
7fcc9f2ec5 | ||
|
|
7ec453920b | ||
|
|
87416f903a | ||
|
|
2ab82ac032 | ||
|
|
d8b5c8db02 |
@@ -2,3 +2,5 @@ target
|
|||||||
Cargo.lock
|
Cargo.lock
|
||||||
smarm_trace.json
|
smarm_trace.json
|
||||||
/bench_results/
|
/bench_results/
|
||||||
|
__pycache__/
|
||||||
|
*.pyc
|
||||||
|
|||||||
@@ -61,3 +61,7 @@ harness = false
|
|||||||
[[bench]]
|
[[bench]]
|
||||||
name = "rq_runtime"
|
name = "rq_runtime"
|
||||||
harness = false
|
harness = false
|
||||||
|
|
||||||
|
[[bench]]
|
||||||
|
name = "spin_sweep"
|
||||||
|
harness = false
|
||||||
|
|||||||
@@ -21,6 +21,7 @@ cargo bench # all of them (slow; rarely what you want)
|
|||||||
| `tokio_favored.rs` | Workloads tokio's model is built for. Expect to lose; the value is knowing *by how much* and catching the gap widening. |
|
| `tokio_favored.rs` | Workloads tokio's model is built for. Expect to lose; the value is knowing *by how much* and catching the gap widening. |
|
||||||
| `rq_micro.rs` | Run-queue **structures** in isolation (no runtime, no actors): push/pop throughput sweeping thread count × producer:consumer ratio. Covers all three queue types in one binary — the types compile in every build; only the runtime's alias is feature-selected. |
|
| `rq_micro.rs` | Run-queue **structures** in isolation (no runtime, no actors): push/pop throughput sweeping thread count × producer:consumer ratio. Covers all three queue types in one binary — the types compile in every build; only the runtime's alias is feature-selected. |
|
||||||
| `rq_runtime.rs` | The **whole scheduler** with the compile-time-selected queue: yield-storm (pure queue churn), ping-pong-pairs (park/unpark latency), spawn-storm (slab + free list + queue churn), sweeping scheduler count. Comparing variants requires rebuilding per `rq-*` feature. |
|
| `rq_runtime.rs` | The **whole scheduler** with the compile-time-selected queue: yield-storm (pure queue churn), ping-pong-pairs (park/unpark latency), spawn-storm (slab + free list + queue churn), sweeping scheduler count. Comparing variants requires rebuilding per `rq-*` feature. |
|
||||||
|
| `spin_sweep.rs` | RFC 004 spinning workers: wake **latency vs idle-CPU** over `spin_budget_cycles`. Four workloads (remote-wake, fork-join, half-load, pure-idle) × budget × `max_spinners` × scheduler count; latency percentiles plus cores-busy from `getrusage`. The data that picks the budget default and on/off posture. |
|
||||||
|
|
||||||
## The run-queue shootout
|
## The run-queue shootout
|
||||||
|
|
||||||
@@ -55,6 +56,70 @@ cargo bench --bench rq_runtime --no-default-features --features rq-striped
|
|||||||
| `SMARM_BENCH_PAIRS` / `_ROUNDTRIPS` | `32` / `1000` | `rq_runtime` ping-pong |
|
| `SMARM_BENCH_PAIRS` / `_ROUNDTRIPS` | `32` / `1000` | `rq_runtime` ping-pong |
|
||||||
| `SMARM_BENCH_SPAWNS` | `5000` | `rq_runtime` spawn-storm |
|
| `SMARM_BENCH_SPAWNS` | `5000` | `rq_runtime` spawn-storm |
|
||||||
|
|
||||||
|
## The spin sweep (RFC 004)
|
||||||
|
|
||||||
|
Spinning workers trade idle CPU for wake latency: an idle worker polls
|
||||||
|
`queue_len` for up to `spin_budget_cycles` (at most `max_spinners` at once)
|
||||||
|
before falling back to a futex park, so work that lands during the spin is
|
||||||
|
claimed with no syscall. `spin_budget_cycles = 0` is the feature fully off —
|
||||||
|
the historical `thread::sleep` park, exactly preserved. `spin_sweep` measures
|
||||||
|
both sides of that trade across the budget so the curve can pick the default.
|
||||||
|
|
||||||
|
```
|
||||||
|
cargo bench --bench spin_sweep
|
||||||
|
# on a big box (scale the work up from the sub-second sandbox defaults):
|
||||||
|
SMARM_BENCH_THREADS="1 2 4 8 16" SMARM_BENCH_RUNS=9 \
|
||||||
|
SMARM_SPIN_BUDGET="0 1000 5000 20000 100000 500000 2000000" \
|
||||||
|
SMARM_SPIN_ROUNDS=2000 SMARM_SPIN_BURSTS=500 SMARM_SPIN_ITEMS=5000 \
|
||||||
|
cargo bench --bench spin_sweep | tee bench_results/spin.txt
|
||||||
|
grep '^SPINCSV' bench_results/spin.txt > bench_results/spin.csv
|
||||||
|
```
|
||||||
|
|
||||||
|
Two axes per config: **latency** (p50/p90/p99/min/max, pooled across runs) and
|
||||||
|
**cost** (cores-busy = process CPU-seconds via `getrusage(RUSAGE_SELF)` over
|
||||||
|
wall-seconds; `1.0` = one core pegged, up to `max_spinners` when every spinner
|
||||||
|
is hot). The four workloads:
|
||||||
|
|
||||||
|
- **remote-wake** — the latency target. A consumer parks on `recv`; a producer
|
||||||
|
on another scheduler timestamps just before `send`. Latency is send→receipt.
|
||||||
|
- **fork-join** — fan-out wake latency: root forks `FANOUT` actors, joins them,
|
||||||
|
gap, repeat.
|
||||||
|
- **half-load** — the cost target. Work dispatched every `PERIOD_US`, so workers
|
||||||
|
idle a fraction of each period; cores-busy climbs with budget as spin fills
|
||||||
|
the gaps. Spin cost scales with **park frequency** × budget, so this (not
|
||||||
|
pure-idle) is where the cost curve lives.
|
||||||
|
- **pure-idle** — a floor check: one actor sleeps, nothing else runs. Confirms
|
||||||
|
budget 0 stays parked and a large budget doesn't leak *continuous* CPU (a
|
||||||
|
spinner spends its budget once per park, so a truly idle window barely moves).
|
||||||
|
|
||||||
|
The crossover the sweep reveals: spinning only wins when work arrives while a
|
||||||
|
worker is still inside its budget. With a fixed inter-arrival gap, budgets whose
|
||||||
|
spin-time (`budget_cycles / TSC_freq`) exceeds the gap start catching work
|
||||||
|
before the park — so `SMARM_SPIN_GAP_US` sets the knee; tune it against your
|
||||||
|
box's TSC frequency.
|
||||||
|
|
||||||
|
### Knobs (env vars, all optional)
|
||||||
|
|
||||||
|
| var | default | used by |
|
||||||
|
|---|---|---|
|
||||||
|
| `SMARM_BENCH_THREADS` | `"1 2"` | scheduler-count sweep |
|
||||||
|
| `SMARM_BENCH_RUNS` | `3` | runs per config (latency pooled, cores median) |
|
||||||
|
| `SMARM_SPIN_BUDGET` | `"0 1000 10000 100000 1000000"` | `spin_budget_cycles` sweep |
|
||||||
|
| `SMARM_MAX_SPINNERS` | `"-"` | `max_spinners` sweep; `-` = runtime default (N/2) |
|
||||||
|
| `SMARM_SPIN_GAP_US` | `150` | idle gap between rounds/bursts (sets the latency knee) |
|
||||||
|
| `SMARM_SPIN_ROUNDS` | `100` | remote-wake rounds |
|
||||||
|
| `SMARM_SPIN_FANOUT` / `_BURSTS` | `64` / `30` | fork-join |
|
||||||
|
| `SMARM_SPIN_ITEMS` / `_PERIOD_US` | `150` / `300` | half-load |
|
||||||
|
| `SMARM_SPIN_WINDOW_MS` | `25` | pure-idle window |
|
||||||
|
|
||||||
|
Output is the house table plus a greppable `SPINCSV,...` line per config.
|
||||||
|
|
||||||
|
**On 1 core spinning is pathological** — a hot spinner starves the very thread
|
||||||
|
that would enqueue work, so the optimal budget there is 0 and you only ever see
|
||||||
|
the cost, never the benefit (the sweep will show remote-wake latency getting
|
||||||
|
*worse* with budget on one core). One core validates the harness; the real curve
|
||||||
|
comes from the many-core box.
|
||||||
|
|
||||||
## Reading the numbers honestly
|
## Reading the numbers honestly
|
||||||
|
|
||||||
- **Core count is the experiment.** On a 1-core machine (CI, sandboxes) the
|
- **Core count is the experiment.** On a 1-core machine (CI, sandboxes) the
|
||||||
|
|||||||
@@ -0,0 +1,435 @@
|
|||||||
|
//! RFC 004 spinning-worker measurement tool — the latency-vs-idle-CPU curve
|
||||||
|
//! over `spin_budget_cycles`.
|
||||||
|
//!
|
||||||
|
//! Spinning workers trade idle CPU for wake latency: an idle worker burns
|
||||||
|
//! cycles polling `queue_len` (up to `spin_budget_cycles`, capped at
|
||||||
|
//! `max_spinners` concurrent spinners) so that incoming work is claimed with
|
||||||
|
//! no futex syscall. `spin_budget_cycles = 0` is the feature fully off — the
|
||||||
|
//! historical `thread::sleep` park, exactly preserved. This binary measures
|
||||||
|
//! both sides of that trade so the sweep can pick the budget default (RFC 004
|
||||||
|
//! Q3) and the on/off posture (Q4).
|
||||||
|
//!
|
||||||
|
//! Workloads — each one is a fresh runtime per sample (no cross-contamination):
|
||||||
|
//!
|
||||||
|
//! remote-wake — a consumer blocks on recv (its worker idles); a producer
|
||||||
|
//! on another scheduler timestamps just before send. Latency
|
||||||
|
//! = send -> receipt, the quantity spinning is meant to cut.
|
||||||
|
//! A fixed gap (SMARM_SPIN_GAP_US) between rounds lets the
|
||||||
|
//! consumer re-idle each round. THE LATENCY TARGET.
|
||||||
|
//! fork-join — root forks FANOUT trivial actors and joins them (one
|
||||||
|
//! burst), gap, repeat. Latency = fork -> all-joined; stresses
|
||||||
|
//! the fan-out wake. Secondary latency target.
|
||||||
|
//! half-load — work dispatched every PERIOD_US for ITEMS items, so
|
||||||
|
//! workers idle a fraction of each period. THE COST TARGET:
|
||||||
|
//! cores-busy climbs with budget as spin fills the idle gaps.
|
||||||
|
//! pure-idle — one actor sleeps WINDOW_MS, nothing else runs. The cleanest
|
||||||
|
//! single picture of spin burn: cores-busy is ~0 at budget 0
|
||||||
|
//! and climbs toward max_spinners as the budget grows.
|
||||||
|
//!
|
||||||
|
//! Two axes, reported together per config:
|
||||||
|
//! latency — p50 / p90 / p99 / min / max over the POOLED rounds of every run
|
||||||
|
//! (pooling gives a stabler tail than a median-of-percentiles).
|
||||||
|
//! cost — cores-busy = process CPU-seconds (getrusage RUSAGE_SELF, all
|
||||||
|
//! runtime threads) / wall-seconds over the measured window. 1.0 =
|
||||||
|
//! one core pegged; approaches max_spinners when every spinner is
|
||||||
|
//! hot. Reported as the median over runs.
|
||||||
|
//!
|
||||||
|
//! The crossover the sweep reveals: spinning only wins when work arrives while
|
||||||
|
//! a worker is still inside its budget. With a fixed inter-arrival gap, that
|
||||||
|
//! means budgets whose spin-time (budget_cycles / TSC_freq) exceeds the gap
|
||||||
|
//! start catching work before the park; smaller budgets park and pay the full
|
||||||
|
//! wake. So GAP_US sets the knee — tune it against your box's TSC frequency.
|
||||||
|
//!
|
||||||
|
//! Knobs (env):
|
||||||
|
//! SMARM_BENCH_THREADS scheduler-count sweep default "1 2"
|
||||||
|
//! SMARM_BENCH_RUNS runs per config (pooled/median) default 3
|
||||||
|
//! SMARM_SPIN_BUDGET spin_budget_cycles sweep default "0 1000 10000 100000 1000000"
|
||||||
|
//! SMARM_MAX_SPINNERS max_spinners sweep; "-" = runtime default (N/2) default "-"
|
||||||
|
//! SMARM_SPIN_GAP_US idle gap between rounds/bursts default 150
|
||||||
|
//! SMARM_SPIN_ROUNDS remote-wake rounds default 100
|
||||||
|
//! SMARM_SPIN_FANOUT fork-join actors per burst default 64
|
||||||
|
//! SMARM_SPIN_BURSTS fork-join bursts default 30
|
||||||
|
//! SMARM_SPIN_ITEMS half-load dispatch count default 150
|
||||||
|
//! SMARM_SPIN_PERIOD_US half-load inter-dispatch period default 300
|
||||||
|
//! SMARM_SPIN_WINDOW_MS pure-idle window default 25
|
||||||
|
//!
|
||||||
|
//! Output: house table + one greppable line per config:
|
||||||
|
//! SPINCSV,<variant>,<workload>,<threads>,<budget>,<spinners>,<gap_us>,
|
||||||
|
//! <n>,<p50_us>,<p90_us>,<p99_us>,<min_us>,<max_us>,<cores_busy>,<ops_s>
|
||||||
|
//! ("spinners" is the literal "default" when SMARM_MAX_SPINNERS is "-".)
|
||||||
|
//!
|
||||||
|
//! NOTE: on a 1-core box (CI, sandboxes) spinning is *pathological* — a hot
|
||||||
|
//! spinner starves the very thread that would enqueue work, so the optimal
|
||||||
|
//! budget there is 0 and you only ever see the cost, never the benefit. This
|
||||||
|
//! binary on one core validates the harness and catches gross pathologies; the
|
||||||
|
//! real latency-vs-idle-CPU curve comes from the many-core box.
|
||||||
|
|
||||||
|
use smarm::runtime::{init, Config};
|
||||||
|
use smarm::{channel, sleep, spawn};
|
||||||
|
use std::sync::{Arc, Mutex};
|
||||||
|
use std::time::{Duration, Instant};
|
||||||
|
|
||||||
|
// --------------------------------------------------------------------------
|
||||||
|
// env helpers (house style, matching rq_runtime.rs)
|
||||||
|
// --------------------------------------------------------------------------
|
||||||
|
|
||||||
|
fn variant() -> &'static str {
|
||||||
|
if cfg!(feature = "rq-mpmc") {
|
||||||
|
"rq-mpmc"
|
||||||
|
} else if cfg!(feature = "rq-striped") {
|
||||||
|
"rq-striped"
|
||||||
|
} else {
|
||||||
|
"rq-mutex"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn env_usize(key: &str, default: usize) -> usize {
|
||||||
|
std::env::var(key).ok().and_then(|v| v.parse().ok()).unwrap_or(default)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn env_u64(key: &str, default: u64) -> u64 {
|
||||||
|
std::env::var(key).ok().and_then(|v| v.parse().ok()).unwrap_or(default)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn env_threads() -> Vec<usize> {
|
||||||
|
std::env::var("SMARM_BENCH_THREADS")
|
||||||
|
.map(|v| v.split_whitespace().filter_map(|t| t.parse().ok()).collect())
|
||||||
|
.unwrap_or_else(|_| vec![1, 2])
|
||||||
|
}
|
||||||
|
|
||||||
|
fn env_budgets() -> Vec<u64> {
|
||||||
|
std::env::var("SMARM_SPIN_BUDGET")
|
||||||
|
.map(|v| v.split_whitespace().filter_map(|t| t.parse().ok()).collect())
|
||||||
|
.unwrap_or_else(|_| vec![0, 1_000, 10_000, 100_000, 1_000_000])
|
||||||
|
}
|
||||||
|
|
||||||
|
/// `None` means "leave the runtime default (N/2)"; encoded in env as "-".
|
||||||
|
fn env_spinners() -> Vec<Option<u32>> {
|
||||||
|
std::env::var("SMARM_MAX_SPINNERS")
|
||||||
|
.map(|v| {
|
||||||
|
v.split_whitespace()
|
||||||
|
.map(|t| if t == "-" { None } else { t.parse().ok() })
|
||||||
|
.collect()
|
||||||
|
})
|
||||||
|
.unwrap_or_else(|_| vec![None])
|
||||||
|
}
|
||||||
|
|
||||||
|
fn make_config(threads: usize, budget: u64, spinners: Option<u32>) -> Config {
|
||||||
|
let cfg = Config::exact(threads).spin_budget_cycles(budget);
|
||||||
|
match spinners {
|
||||||
|
Some(n) => cfg.max_spinners(n),
|
||||||
|
None => cfg,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// --------------------------------------------------------------------------
|
||||||
|
// CPU accounting — process-wide (all runtime threads) via getrusage
|
||||||
|
// --------------------------------------------------------------------------
|
||||||
|
|
||||||
|
/// Total CPU seconds (user + system) consumed by the whole process so far.
|
||||||
|
/// RUSAGE_SELF aggregates every thread, which is exactly what "how much did
|
||||||
|
/// the runtime burn" needs — the spinners are separate threads from the one
|
||||||
|
/// reading this.
|
||||||
|
fn cpu_seconds() -> f64 {
|
||||||
|
// SAFETY: getrusage with RUSAGE_SELF and a valid out-pointer; we check the
|
||||||
|
// return code and only read the (now-initialised) timeval fields.
|
||||||
|
let mut ru: libc::rusage = unsafe { std::mem::zeroed() };
|
||||||
|
let rc = unsafe { libc::getrusage(libc::RUSAGE_SELF, &mut ru) };
|
||||||
|
assert_eq!(rc, 0, "getrusage(RUSAGE_SELF) failed");
|
||||||
|
let tv = |t: libc::timeval| t.tv_sec as f64 + t.tv_usec as f64 * 1e-6;
|
||||||
|
tv(ru.ru_utime) + tv(ru.ru_stime)
|
||||||
|
}
|
||||||
|
|
||||||
|
// --------------------------------------------------------------------------
|
||||||
|
// One sample = one fresh runtime run
|
||||||
|
// --------------------------------------------------------------------------
|
||||||
|
|
||||||
|
struct Sample {
|
||||||
|
/// Per-event latencies in nanoseconds (empty for pure-idle).
|
||||||
|
lat_ns: Vec<u64>,
|
||||||
|
/// CPU-seconds / wall-seconds over the measured window.
|
||||||
|
cores_busy: f64,
|
||||||
|
/// Completed work units (for ops/s); 0 where not meaningful.
|
||||||
|
ops: u64,
|
||||||
|
/// Wall-seconds of the measured window (for ops/s).
|
||||||
|
wall_s: f64,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Shared scratch the root closure writes and the caller reads after `run()`
|
||||||
|
/// returns. (`run` takes a `FnOnce` and yields no value, so we hand results
|
||||||
|
/// out through an Arc rather than a return.)
|
||||||
|
#[derive(Default)]
|
||||||
|
struct Scratch {
|
||||||
|
lat_ns: Vec<u64>,
|
||||||
|
cpu_delta: f64,
|
||||||
|
wall_s: f64,
|
||||||
|
ops: u64,
|
||||||
|
}
|
||||||
|
|
||||||
|
fn finish(scratch: Arc<Mutex<Scratch>>) -> Sample {
|
||||||
|
let mut s = scratch.lock().unwrap();
|
||||||
|
let cores_busy = if s.wall_s > 0.0 { s.cpu_delta / s.wall_s } else { 0.0 };
|
||||||
|
Sample {
|
||||||
|
lat_ns: std::mem::take(&mut s.lat_ns),
|
||||||
|
cores_busy,
|
||||||
|
ops: s.ops,
|
||||||
|
wall_s: s.wall_s,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn remote_wake(threads: usize, budget: u64, spinners: Option<u32>, rounds: usize, gap: Duration) -> Sample {
|
||||||
|
let rt = init(make_config(threads, budget, spinners));
|
||||||
|
let scratch = Arc::new(Mutex::new(Scratch::default()));
|
||||||
|
let out = scratch.clone();
|
||||||
|
rt.run(move || {
|
||||||
|
let (tx, rx) = channel::channel::<Instant>();
|
||||||
|
let lat_sink = out.clone();
|
||||||
|
// Consumer parks on recv; its worker idles (spins or parks per budget)
|
||||||
|
// until each send wakes it. It records the wake latency itself.
|
||||||
|
let consumer = spawn(move || {
|
||||||
|
let mut lats = Vec::with_capacity(rounds);
|
||||||
|
for _ in 0..rounds {
|
||||||
|
let sent = rx.recv().expect("remote-wake recv");
|
||||||
|
lats.push(sent.elapsed().as_nanos() as u64);
|
||||||
|
}
|
||||||
|
lat_sink.lock().unwrap().lat_ns = lats;
|
||||||
|
});
|
||||||
|
// Producer: sleep the gap (so the consumer's worker re-idles), then
|
||||||
|
// timestamp and send. The send is the wake event under measurement.
|
||||||
|
let cpu0 = cpu_seconds();
|
||||||
|
let w0 = Instant::now();
|
||||||
|
for _ in 0..rounds {
|
||||||
|
sleep(gap);
|
||||||
|
tx.send(Instant::now()).expect("remote-wake send");
|
||||||
|
}
|
||||||
|
let wall = w0.elapsed().as_secs_f64();
|
||||||
|
let cpu = cpu_seconds() - cpu0;
|
||||||
|
let _ = consumer.join();
|
||||||
|
let mut s = out.lock().unwrap();
|
||||||
|
s.cpu_delta = cpu;
|
||||||
|
s.wall_s = wall;
|
||||||
|
s.ops = rounds as u64;
|
||||||
|
});
|
||||||
|
finish(scratch)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn fork_join(threads: usize, budget: u64, spinners: Option<u32>, fanout: usize, bursts: usize, gap: Duration) -> Sample {
|
||||||
|
let rt = init(make_config(threads, budget, spinners));
|
||||||
|
let scratch = Arc::new(Mutex::new(Scratch::default()));
|
||||||
|
let out = scratch.clone();
|
||||||
|
rt.run(move || {
|
||||||
|
let mut lats = Vec::with_capacity(bursts);
|
||||||
|
let cpu0 = cpu_seconds();
|
||||||
|
let w0 = Instant::now();
|
||||||
|
for _ in 0..bursts {
|
||||||
|
// Gap lets the workers park between bursts, so the fan-out has real
|
||||||
|
// wakes to do rather than landing on already-hot workers.
|
||||||
|
sleep(gap);
|
||||||
|
let t = Instant::now();
|
||||||
|
let handles: Vec<_> = (0..fanout).map(|_| spawn(|| {})).collect();
|
||||||
|
for h in handles {
|
||||||
|
let _ = h.join();
|
||||||
|
}
|
||||||
|
lats.push(t.elapsed().as_nanos() as u64);
|
||||||
|
}
|
||||||
|
let wall = w0.elapsed().as_secs_f64();
|
||||||
|
let cpu = cpu_seconds() - cpu0;
|
||||||
|
let mut s = out.lock().unwrap();
|
||||||
|
s.lat_ns = lats;
|
||||||
|
s.cpu_delta = cpu;
|
||||||
|
s.wall_s = wall;
|
||||||
|
s.ops = (bursts * fanout) as u64;
|
||||||
|
});
|
||||||
|
finish(scratch)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn half_load(threads: usize, budget: u64, spinners: Option<u32>, items: usize, period: Duration) -> Sample {
|
||||||
|
let rt = init(make_config(threads, budget, spinners));
|
||||||
|
let scratch = Arc::new(Mutex::new(Scratch::default()));
|
||||||
|
let out = scratch.clone();
|
||||||
|
rt.run(move || {
|
||||||
|
let cpu0 = cpu_seconds();
|
||||||
|
let w0 = Instant::now();
|
||||||
|
// Dispatch one trivial actor every `period`. Between dispatches the
|
||||||
|
// workers have nothing to do, so they idle for a fraction of each
|
||||||
|
// period — that idle fraction is what spinning fills (or doesn't).
|
||||||
|
let mut handles = Vec::with_capacity(items);
|
||||||
|
for i in 0..items {
|
||||||
|
let target = period * i as u32;
|
||||||
|
let elapsed = w0.elapsed();
|
||||||
|
if target > elapsed {
|
||||||
|
sleep(target - elapsed);
|
||||||
|
}
|
||||||
|
handles.push(spawn(|| {}));
|
||||||
|
}
|
||||||
|
for h in handles {
|
||||||
|
let _ = h.join();
|
||||||
|
}
|
||||||
|
let wall = w0.elapsed().as_secs_f64();
|
||||||
|
let cpu = cpu_seconds() - cpu0;
|
||||||
|
let mut s = out.lock().unwrap();
|
||||||
|
s.cpu_delta = cpu;
|
||||||
|
s.wall_s = wall;
|
||||||
|
s.ops = items as u64;
|
||||||
|
});
|
||||||
|
finish(scratch)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn pure_idle(threads: usize, budget: u64, spinners: Option<u32>, window: Duration) -> Sample {
|
||||||
|
let rt = init(make_config(threads, budget, spinners));
|
||||||
|
let scratch = Arc::new(Mutex::new(Scratch::default()));
|
||||||
|
let out = scratch.clone();
|
||||||
|
rt.run(move || {
|
||||||
|
// One actor sleeps; every other worker thread has nothing to do and
|
||||||
|
// spins-or-parks per budget for the whole window.
|
||||||
|
let cpu0 = cpu_seconds();
|
||||||
|
let w0 = Instant::now();
|
||||||
|
sleep(window);
|
||||||
|
let wall = w0.elapsed().as_secs_f64();
|
||||||
|
let cpu = cpu_seconds() - cpu0;
|
||||||
|
let mut s = out.lock().unwrap();
|
||||||
|
s.cpu_delta = cpu;
|
||||||
|
s.wall_s = wall;
|
||||||
|
s.ops = 0;
|
||||||
|
});
|
||||||
|
finish(scratch)
|
||||||
|
}
|
||||||
|
|
||||||
|
// --------------------------------------------------------------------------
|
||||||
|
// stats
|
||||||
|
// --------------------------------------------------------------------------
|
||||||
|
|
||||||
|
/// Nearest-rank percentile over an already-sorted slice. `p` in [0, 100].
|
||||||
|
fn pct(sorted: &[u64], p: f64) -> u64 {
|
||||||
|
if sorted.is_empty() {
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
let idx = ((p / 100.0) * (sorted.len() - 1) as f64).round() as usize;
|
||||||
|
sorted[idx.min(sorted.len() - 1)]
|
||||||
|
}
|
||||||
|
|
||||||
|
fn median_f64(xs: &mut [f64]) -> f64 {
|
||||||
|
if xs.is_empty() {
|
||||||
|
return 0.0;
|
||||||
|
}
|
||||||
|
xs.sort_by(|a, b| a.partial_cmp(b).unwrap());
|
||||||
|
xs[xs.len() / 2]
|
||||||
|
}
|
||||||
|
|
||||||
|
struct Aggregate {
|
||||||
|
n: usize,
|
||||||
|
p50_us: f64,
|
||||||
|
p90_us: f64,
|
||||||
|
p99_us: f64,
|
||||||
|
min_us: f64,
|
||||||
|
max_us: f64,
|
||||||
|
cores_busy: f64,
|
||||||
|
ops_s: f64,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Pool latencies across all runs (stabler tail); take the median cores-busy
|
||||||
|
/// and median ops/s across runs.
|
||||||
|
fn aggregate(samples: Vec<Sample>) -> Aggregate {
|
||||||
|
let mut pooled: Vec<u64> = Vec::new();
|
||||||
|
let mut cores: Vec<f64> = Vec::new();
|
||||||
|
let mut ops_s: Vec<f64> = Vec::new();
|
||||||
|
for s in &samples {
|
||||||
|
pooled.extend_from_slice(&s.lat_ns);
|
||||||
|
cores.push(s.cores_busy);
|
||||||
|
if s.wall_s > 0.0 && s.ops > 0 {
|
||||||
|
ops_s.push(s.ops as f64 / s.wall_s);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
pooled.sort_unstable();
|
||||||
|
let ns_to_us = |ns: u64| ns as f64 / 1000.0;
|
||||||
|
Aggregate {
|
||||||
|
n: pooled.len(),
|
||||||
|
p50_us: ns_to_us(pct(&pooled, 50.0)),
|
||||||
|
p90_us: ns_to_us(pct(&pooled, 90.0)),
|
||||||
|
p99_us: ns_to_us(pct(&pooled, 99.0)),
|
||||||
|
min_us: pooled.first().map(|&v| ns_to_us(v)).unwrap_or(0.0),
|
||||||
|
max_us: pooled.last().map(|&v| ns_to_us(v)).unwrap_or(0.0),
|
||||||
|
cores_busy: median_f64(&mut cores),
|
||||||
|
ops_s: median_f64(&mut ops_s),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// --------------------------------------------------------------------------
|
||||||
|
// driver
|
||||||
|
// --------------------------------------------------------------------------
|
||||||
|
|
||||||
|
fn main() {
|
||||||
|
let threads_sweep = env_threads();
|
||||||
|
let budgets = env_budgets();
|
||||||
|
let spinners_sweep = env_spinners();
|
||||||
|
let runs = env_usize("SMARM_BENCH_RUNS", 3);
|
||||||
|
let gap = Duration::from_micros(env_u64("SMARM_SPIN_GAP_US", 150));
|
||||||
|
let rounds = env_usize("SMARM_SPIN_ROUNDS", 100);
|
||||||
|
let fanout = env_usize("SMARM_SPIN_FANOUT", 64);
|
||||||
|
let bursts = env_usize("SMARM_SPIN_BURSTS", 30);
|
||||||
|
let items = env_usize("SMARM_SPIN_ITEMS", 150);
|
||||||
|
let period = Duration::from_micros(env_u64("SMARM_SPIN_PERIOD_US", 300));
|
||||||
|
let window = Duration::from_millis(env_u64("SMARM_SPIN_WINDOW_MS", 25));
|
||||||
|
|
||||||
|
// (name, closure producing one Sample for a (threads,budget,spinners) point)
|
||||||
|
type Workload = (&'static str, Box<dyn Fn(usize, u64, Option<u32>) -> Sample>);
|
||||||
|
let workloads: Vec<Workload> = vec![
|
||||||
|
("remote-wake", Box::new(move |t, b, s| remote_wake(t, b, s, rounds, gap))),
|
||||||
|
("fork-join", Box::new(move |t, b, s| fork_join(t, b, s, fanout, bursts, gap))),
|
||||||
|
("half-load", Box::new(move |t, b, s| half_load(t, b, s, items, period))),
|
||||||
|
("pure-idle", Box::new(move |t, b, s| pure_idle(t, b, s, window))),
|
||||||
|
];
|
||||||
|
|
||||||
|
let width = 122;
|
||||||
|
println!("\n{}", "=".repeat(width));
|
||||||
|
println!(
|
||||||
|
" spin sweep — variant={}, runs={runs} (latency pooled, cores median), gap={}µs",
|
||||||
|
variant(),
|
||||||
|
gap.as_micros()
|
||||||
|
);
|
||||||
|
println!("{}", "=".repeat(width));
|
||||||
|
println!(
|
||||||
|
"{:>11} | {:>3} | {:>9} | {:>8} | {:>6} | {:>9} | {:>9} | {:>9} | {:>9} | {:>9} | {:>7} | {:>10}",
|
||||||
|
"workload", "thr", "budget", "spinners", "n", "p50 µs", "p90 µs", "p99 µs", "min µs", "max µs", "cores", "ops/s"
|
||||||
|
);
|
||||||
|
println!("{}", "-".repeat(width));
|
||||||
|
|
||||||
|
for (name, run_one) in &workloads {
|
||||||
|
for &threads in &threads_sweep {
|
||||||
|
for &budget in &budgets {
|
||||||
|
for &spinners in &spinners_sweep {
|
||||||
|
let samples: Vec<Sample> =
|
||||||
|
(0..runs).map(|_| run_one(threads, budget, spinners)).collect();
|
||||||
|
let a = aggregate(samples);
|
||||||
|
let sp_str = match spinners {
|
||||||
|
Some(n) => n.to_string(),
|
||||||
|
None => "default".to_string(),
|
||||||
|
};
|
||||||
|
// Latency columns are blank for the no-latency probe.
|
||||||
|
let has_lat = a.n > 0;
|
||||||
|
let lat = |v: f64| if has_lat { format!("{v:.2}") } else { "-".to_string() };
|
||||||
|
println!(
|
||||||
|
"{:>11} | {:>3} | {:>9} | {:>8} | {:>6} | {:>9} | {:>9} | {:>9} | {:>9} | {:>9} | {:>7.3} | {:>10}",
|
||||||
|
name,
|
||||||
|
threads,
|
||||||
|
budget,
|
||||||
|
sp_str,
|
||||||
|
a.n,
|
||||||
|
lat(a.p50_us),
|
||||||
|
lat(a.p90_us),
|
||||||
|
lat(a.p99_us),
|
||||||
|
lat(a.min_us),
|
||||||
|
lat(a.max_us),
|
||||||
|
a.cores_busy,
|
||||||
|
if a.ops_s > 0.0 { format!("{:.0}", a.ops_s) } else { "-".to_string() },
|
||||||
|
);
|
||||||
|
println!(
|
||||||
|
"SPINCSV,{},{},{},{},{},{},{},{:.3},{:.3},{:.3},{:.3},{:.3},{:.4},{:.0}",
|
||||||
|
variant(), name, threads, budget, sp_str, gap.as_micros(),
|
||||||
|
a.n, a.p50_us, a.p90_us, a.p99_us, a.min_us, a.max_us, a.cores_busy, a.ops_s
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -243,6 +243,90 @@ def check_regressions(current: dict, baseline: dict) -> bool:
|
|||||||
# Pretty print
|
# Pretty print
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
def _threads(label: str) -> int | None:
|
||||||
|
"""Worker-thread count implied by a runtime label.
|
||||||
|
|
||||||
|
tokio's `current_thread` is a single-threaded executor (1); an explicit
|
||||||
|
`multi N-thread` is N; a bare `multi-thread` has no count (None) — tokio's
|
||||||
|
default work-stealing pool, paired against smarm's widest config.
|
||||||
|
"""
|
||||||
|
if "current_thread" in label:
|
||||||
|
return 1
|
||||||
|
m = re.search(r"(\d+)-thread", label)
|
||||||
|
return int(m.group(1)) if m else None
|
||||||
|
|
||||||
|
|
||||||
|
def vs_tokio(results: dict) -> list[tuple]:
|
||||||
|
"""Per bench, like-for-like smarm-vs-tokio rows matched by thread count.
|
||||||
|
|
||||||
|
Exact thread-count matches are paired directly (smarm 1-thread vs tokio
|
||||||
|
current_thread, smarm 4-thread vs tokio multi 4-thread, …). tokio's bare
|
||||||
|
`multi-thread` (no explicit count) is paired against the widest unmatched
|
||||||
|
smarm multi-thread config. Lower median µs = faster; ratio = tokio_med /
|
||||||
|
smarm_med, so ratio > 1 means smarm is that many times faster.
|
||||||
|
|
||||||
|
Returns rows of (bench, smarm_label, smarm_med, tokio_label, tokio_med,
|
||||||
|
ratio, winner). Benches without a comparable pair are skipped.
|
||||||
|
"""
|
||||||
|
rows: list[tuple] = []
|
||||||
|
for bench, runtimes in sorted(results.items()):
|
||||||
|
smarm: dict[int, tuple[str, int]] = {}
|
||||||
|
tokio: dict[int, tuple[str, int]] = {}
|
||||||
|
tokio_default: tuple[str, int] | None = None # bare 'multi-thread'
|
||||||
|
for label, data in runtimes.items():
|
||||||
|
n = _threads(label)
|
||||||
|
if label.startswith("smarm"):
|
||||||
|
if n is not None:
|
||||||
|
smarm[n] = (label, data["median"])
|
||||||
|
elif label.startswith("tokio"):
|
||||||
|
if n is None:
|
||||||
|
tokio_default = (label, data["median"])
|
||||||
|
else:
|
||||||
|
tokio[n] = (label, data["median"])
|
||||||
|
|
||||||
|
def row(s: tuple[str, int], t: tuple[str, int]):
|
||||||
|
s_label, s_med = s
|
||||||
|
t_label, t_med = t
|
||||||
|
if s_med == 0:
|
||||||
|
return None
|
||||||
|
ratio = t_med / s_med
|
||||||
|
return (bench, s_label, s_med, t_label, t_med, ratio,
|
||||||
|
"smarm" if ratio >= 1.0 else "tokio")
|
||||||
|
|
||||||
|
matched: set[int] = set()
|
||||||
|
for n in sorted(set(smarm) & set(tokio)):
|
||||||
|
r = row(smarm[n], tokio[n])
|
||||||
|
if r:
|
||||||
|
rows.append(r)
|
||||||
|
matched.add(n)
|
||||||
|
# tokio's default multi pool vs the widest smarm config not already paired.
|
||||||
|
if tokio_default is not None:
|
||||||
|
rem = [n for n in smarm if n not in matched and n > 1]
|
||||||
|
if rem:
|
||||||
|
r = row(smarm[max(rem)], tokio_default)
|
||||||
|
if r:
|
||||||
|
rows.append(r)
|
||||||
|
return rows
|
||||||
|
|
||||||
|
|
||||||
|
def print_vs_tokio(results: dict) -> None:
|
||||||
|
"""Human summary + greppable VSTOKIO lines (best smarm vs best tokio)."""
|
||||||
|
rows = vs_tokio(results)
|
||||||
|
if not rows:
|
||||||
|
return
|
||||||
|
print("\n vs tokio (like-for-like by thread count; ratio>1 = smarm faster, lower µs better)")
|
||||||
|
print(f" {'-'*78}")
|
||||||
|
for bench, s_label, s_med, t_label, t_med, ratio, winner in rows:
|
||||||
|
print(
|
||||||
|
f" {bench:<22} {s_label} {s_med}µs vs {t_label} {t_med}µs"
|
||||||
|
f" → {ratio:.2f}x ({winner})"
|
||||||
|
)
|
||||||
|
# Machine-readable, one line per bench:
|
||||||
|
# VSTOKIO,<bench>,<smarm_label>,<smarm_us>,<tokio_label>,<tokio_us>,<ratio>,<winner>
|
||||||
|
for bench, s_label, s_med, t_label, t_med, ratio, winner in rows:
|
||||||
|
print(f"VSTOKIO,{bench},{s_label},{s_med},{t_label},{t_med},{ratio:.3f},{winner}")
|
||||||
|
|
||||||
|
|
||||||
def print_results(results: dict, label: str = "") -> None:
|
def print_results(results: dict, label: str = "") -> None:
|
||||||
if label:
|
if label:
|
||||||
print(f"\n{'='*70}")
|
print(f"\n{'='*70}")
|
||||||
@@ -257,6 +341,7 @@ def print_results(results: dict, label: str = "") -> None:
|
|||||||
f" {rt_label:>28} | {data['result']:>10} | "
|
f" {rt_label:>28} | {data['result']:>10} | "
|
||||||
f"{data['median']:>10} | {data['min']:>8} | {data['max']:>8}"
|
f"{data['median']:>10} | {data['min']:>8} | {data['max']:>8}"
|
||||||
)
|
)
|
||||||
|
print_vs_tokio(results)
|
||||||
|
|
||||||
|
|
||||||
def print_sweep_table(sweep_results: list[tuple[int, int, dict]]) -> None:
|
def print_sweep_table(sweep_results: list[tuple[int, int, dict]]) -> None:
|
||||||
|
|||||||
+2
-2
@@ -242,7 +242,7 @@ impl<T> Drop for RawMutexGuard<'_, T> {
|
|||||||
// futex (x86-64 Linux; master is x86-only, see arm-port branch)
|
// futex (x86-64 Linux; master is x86-only, see arm-port branch)
|
||||||
// ---------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------
|
||||||
|
|
||||||
fn futex_wait(state: &AtomicU32, expected: u32) {
|
pub(crate) fn futex_wait(state: &AtomicU32, expected: u32) {
|
||||||
// SAFETY: `state` is a valid, aligned u32 for the duration of the call.
|
// SAFETY: `state` is a valid, aligned u32 for the duration of the call.
|
||||||
// Spurious wakeups and EAGAIN (value already changed) are both handled by
|
// Spurious wakeups and EAGAIN (value already changed) are both handled by
|
||||||
// the caller's retry loop.
|
// the caller's retry loop.
|
||||||
@@ -257,7 +257,7 @@ fn futex_wait(state: &AtomicU32, expected: u32) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
fn futex_wake(state: &AtomicU32, n: i32) {
|
pub(crate) fn futex_wake(state: &AtomicU32, n: i32) {
|
||||||
// SAFETY: as above.
|
// SAFETY: as above.
|
||||||
unsafe {
|
unsafe {
|
||||||
libc::syscall(
|
libc::syscall(
|
||||||
|
|||||||
+214
-3
@@ -118,8 +118,9 @@ use crate::timer::Timers;
|
|||||||
use crate::context::{get_actor_sp, set_actor_sp, switch_to_actor};
|
use crate::context::{get_actor_sp, set_actor_sp, switch_to_actor};
|
||||||
|
|
||||||
use std::sync::atomic::{
|
use std::sync::atomic::{
|
||||||
AtomicBool, AtomicPtr, AtomicU32, AtomicU64, AtomicUsize, Ordering,
|
fence, AtomicBool, AtomicPtr, AtomicU32, AtomicU64, AtomicUsize, Ordering,
|
||||||
};
|
};
|
||||||
|
use crate::raw_mutex::{futex_wait, futex_wake};
|
||||||
use std::sync::{Arc, Mutex};
|
use std::sync::{Arc, Mutex};
|
||||||
use std::thread;
|
use std::thread;
|
||||||
|
|
||||||
@@ -155,6 +156,11 @@ pub struct Config {
|
|||||||
stack_pool_cap: usize,
|
stack_pool_cap: usize,
|
||||||
max_actors: usize,
|
max_actors: usize,
|
||||||
wake_slot: bool,
|
wake_slot: bool,
|
||||||
|
/// RFC 004: TSC cycles a scheduler spins before parking. `0` (default)
|
||||||
|
/// disables spinning entirely — the historical park-immediately behaviour.
|
||||||
|
spin_budget_cycles: u64,
|
||||||
|
/// RFC 004: cap on concurrent spinners. `None` resolves to N/2 at init.
|
||||||
|
max_spinners: Option<u32>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Config {
|
impl Config {
|
||||||
@@ -168,6 +174,8 @@ impl Config {
|
|||||||
stack_pool_cap: n * 4,
|
stack_pool_cap: n * 4,
|
||||||
max_actors: DEFAULT_MAX_ACTORS,
|
max_actors: DEFAULT_MAX_ACTORS,
|
||||||
wake_slot: false,
|
wake_slot: false,
|
||||||
|
spin_budget_cycles: 0,
|
||||||
|
max_spinners: None,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -185,6 +193,8 @@ impl Config {
|
|||||||
stack_pool_cap: max * 4,
|
stack_pool_cap: max * 4,
|
||||||
max_actors: DEFAULT_MAX_ACTORS,
|
max_actors: DEFAULT_MAX_ACTORS,
|
||||||
wake_slot: false,
|
wake_slot: false,
|
||||||
|
spin_budget_cycles: 0,
|
||||||
|
max_spinners: None,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -241,6 +251,25 @@ impl Config {
|
|||||||
self
|
self
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// RFC 004: TSC cycles an idle scheduler spins searching for work before
|
||||||
|
/// parking on the futex. The idle-CPU-vs-wake-latency dial, in the same
|
||||||
|
/// unit as [`timeslice_cycles`](Self::timeslice_cycles) (uses `rdtsc()`).
|
||||||
|
/// `0` (the default) disables spinning and the futex park entirely,
|
||||||
|
/// recovering the historical `thread::sleep` idle behaviour exactly.
|
||||||
|
pub fn spin_budget_cycles(mut self, n: u64) -> Self {
|
||||||
|
self.spin_budget_cycles = n;
|
||||||
|
self
|
||||||
|
}
|
||||||
|
|
||||||
|
/// RFC 004: cap on the number of schedulers that may spin concurrently.
|
||||||
|
/// Default (`None`) resolves to N/2 at init. Capping at half keeps half
|
||||||
|
/// the pool hot and parks the rest — the half-load case is then the easy
|
||||||
|
/// case. Ignored when `spin_budget_cycles == 0`.
|
||||||
|
pub fn max_spinners(mut self, n: u32) -> Self {
|
||||||
|
self.max_spinners = Some(n);
|
||||||
|
self
|
||||||
|
}
|
||||||
|
|
||||||
/// The number of scheduler threads this config resolves to.
|
/// The number of scheduler threads this config resolves to.
|
||||||
pub fn resolved_thread_count(&self) -> usize {
|
pub fn resolved_thread_count(&self) -> usize {
|
||||||
if let Some(e) = self.exact {
|
if let Some(e) = self.exact {
|
||||||
@@ -265,6 +294,8 @@ impl Default for Config {
|
|||||||
stack_pool_cap: avail * 4,
|
stack_pool_cap: avail * 4,
|
||||||
max_actors: DEFAULT_MAX_ACTORS,
|
max_actors: DEFAULT_MAX_ACTORS,
|
||||||
wake_slot: false,
|
wake_slot: false,
|
||||||
|
spin_budget_cycles: 0,
|
||||||
|
max_spinners: None,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -484,6 +515,33 @@ pub(crate) struct RuntimeInner {
|
|||||||
/// Incremented in `spawn` before the enqueue; decremented at the very end
|
/// Incremented in `spawn` before the enqueue; decremented at the very end
|
||||||
/// of `finalize_actor`, after every wakeup finalize produces.
|
/// of `finalize_actor`, after every wakeup finalize produces.
|
||||||
pub(crate) live_actors: AtomicU32,
|
pub(crate) live_actors: AtomicU32,
|
||||||
|
/// RFC 004: lock-free shared-run-queue depth. `+1` in `enqueue` (the sole
|
||||||
|
/// `run_queue.push` site — covers spawn-publish, unpark, and slot
|
||||||
|
/// displacement), `-1` on a successful pop. This is the poll source the
|
||||||
|
/// idle spinners read; it never takes the queue mutex. Slot-parked work is
|
||||||
|
/// deliberately NOT counted: it is thread-local and only its owning thread
|
||||||
|
/// can resume it, so it is not work a spinning sibling could grab.
|
||||||
|
pub(crate) queue_len: AtomicU64,
|
||||||
|
/// RFC 004: schedulers currently *searching* for work (spinning) — the
|
||||||
|
/// middle state between running and parked. The submit rule loads this to
|
||||||
|
/// decide whether a wake is needed (a live spinner will catch the work
|
||||||
|
/// with no syscall). Capped at `max_spinners`.
|
||||||
|
pub(crate) n_spinning: AtomicU32,
|
||||||
|
/// RFC 004: the scheduler park futex word. A monotonic sequence: a waker
|
||||||
|
/// does `fetch_add(1) + futex_wake`, a parker captures the value and
|
||||||
|
/// `futex_wait`s on it, so a bump landing in the arm→wait gap returns the
|
||||||
|
/// wait immediately (EAGAIN) instead of sleeping — the lost-wakeup guard.
|
||||||
|
/// One shared word; `futex_wake(1)` wakes exactly one parked scheduler.
|
||||||
|
pub(crate) park_seq: AtomicU32,
|
||||||
|
/// RFC 004: `spin_budget_cycles > 0`. When false the feature is fully off —
|
||||||
|
/// the idle path is the historical `thread::sleep` park, `n_spinning` and
|
||||||
|
/// `park_seq` are never touched, and the submit rule is a no-op. This is
|
||||||
|
/// what makes `spin_budget_cycles = 0` recover today's behaviour exactly.
|
||||||
|
pub(crate) spinning: bool,
|
||||||
|
/// RFC 004: TSC cycles a scheduler searches before parking (`Config`).
|
||||||
|
pub(crate) spin_budget_cycles: u64,
|
||||||
|
/// RFC 004: cap on concurrent spinners (`Config`, default N/2).
|
||||||
|
pub(crate) max_spinners: u32,
|
||||||
/// Timer heap. Independent lock: never nested with any other.
|
/// Timer heap. Independent lock: never nested with any other.
|
||||||
pub(crate) timers: Mutex<Timers>,
|
pub(crate) timers: Mutex<Timers>,
|
||||||
/// IO subsystem. `None` between runs. Lock order: io before everything.
|
/// IO subsystem. `None` between runs. Lock order: io before everything.
|
||||||
@@ -521,6 +579,8 @@ impl RuntimeInner {
|
|||||||
stack_pool_cap: usize,
|
stack_pool_cap: usize,
|
||||||
max_actors: usize,
|
max_actors: usize,
|
||||||
wake_slot: bool,
|
wake_slot: bool,
|
||||||
|
spin_budget_cycles: u64,
|
||||||
|
max_spinners: u32,
|
||||||
) -> Arc<Self> {
|
) -> Arc<Self> {
|
||||||
let stats = (0..thread_count).map(|_| SchedulerStats::new()).collect();
|
let stats = (0..thread_count).map(|_| SchedulerStats::new()).collect();
|
||||||
let slots: Box<[Slot]> = (0..max_actors).map(|_| Slot::vacant()).collect();
|
let slots: Box<[Slot]> = (0..max_actors).map(|_| Slot::vacant()).collect();
|
||||||
@@ -531,6 +591,12 @@ impl RuntimeInner {
|
|||||||
slots,
|
slots,
|
||||||
free: RawMutex::new(free),
|
free: RawMutex::new(free),
|
||||||
live_actors: AtomicU32::new(0),
|
live_actors: AtomicU32::new(0),
|
||||||
|
queue_len: AtomicU64::new(0),
|
||||||
|
n_spinning: AtomicU32::new(0),
|
||||||
|
park_seq: AtomicU32::new(0),
|
||||||
|
spinning: spin_budget_cycles > 0,
|
||||||
|
spin_budget_cycles,
|
||||||
|
max_spinners,
|
||||||
timers: Mutex::new(Timers::new()),
|
timers: Mutex::new(Timers::new()),
|
||||||
io: Mutex::new(None),
|
io: Mutex::new(None),
|
||||||
next_monitor_id: AtomicU64::new(0),
|
next_monitor_id: AtomicU64::new(0),
|
||||||
@@ -572,9 +638,127 @@ impl RuntimeInner {
|
|||||||
"enqueue of a pid not in (gen, Queued)"
|
"enqueue of a pid not in (gen, Queued)"
|
||||||
);
|
);
|
||||||
self.run_queue.push(pid);
|
self.run_queue.push(pid);
|
||||||
|
// RFC 004: maintain the lock-free depth the idle spinners poll. Release
|
||||||
|
// so a spinner's Acquire load observes the pushed entry.
|
||||||
|
self.queue_len.fetch_add(1, Ordering::Release);
|
||||||
|
// RFC 004 submit rule: ensure a searcher will see this work. A live
|
||||||
|
// spinner (n_spinning > 0) claims it with no syscall — the common,
|
||||||
|
// cheap path; only when nobody is searching do we pay a wake. The
|
||||||
|
// SeqCst fence forbids reordering the n_spinning load ahead of the
|
||||||
|
// queue_len store above. Paired with the parker's symmetric
|
||||||
|
// (decrement n_spinning; SeqCst fence; re-check queue_len), it
|
||||||
|
// guarantees at least one of {we observe the spinner, the parker
|
||||||
|
// observes this work} — the Dekker property that keeps work from
|
||||||
|
// stranding with every scheduler asleep. Loom does not model the
|
||||||
|
// futex path (see sync_shim.rs), so this is correct by construction.
|
||||||
|
if self.spinning {
|
||||||
|
fence(Ordering::SeqCst);
|
||||||
|
if self.n_spinning.load(Ordering::Acquire) == 0 {
|
||||||
|
self.unpark_one_scheduler();
|
||||||
|
}
|
||||||
|
}
|
||||||
crate::te!(crate::trace::Event::Enqueue(pid));
|
crate::te!(crate::trace::Event::Enqueue(pid));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// RFC 004: wake exactly one parked scheduler. Bumping the futex sequence
|
||||||
|
/// before the wake closes the arm→wait race: a parker that already
|
||||||
|
/// captured the old sequence fails its `futex_wait` value-compare and
|
||||||
|
/// returns (EAGAIN) instead of sleeping through the wake. Only called when
|
||||||
|
/// no scheduler is searching, so it is off the common path.
|
||||||
|
pub(crate) fn unpark_one_scheduler(&self) {
|
||||||
|
self.park_seq.fetch_add(1, Ordering::Release);
|
||||||
|
futex_wake(&self.park_seq, 1);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// RFC 004: release ALL futex-parked schedulers (termination). Required in
|
||||||
|
/// a no-io runtime, where there is no wake pipe to fall back on: a sibling
|
||||||
|
/// parked indefinitely on the futex would otherwise sleep past program
|
||||||
|
/// end. Each woken thread re-runs the idle verdict, reaches AllDone, and
|
||||||
|
/// returns — idempotent, mirroring the io wake-pipe terminal broadcast.
|
||||||
|
pub(crate) fn unpark_all_schedulers(&self) {
|
||||||
|
self.park_seq.fetch_add(1, Ordering::Release);
|
||||||
|
futex_wake(&self.park_seq, i32::MAX);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// RFC 004: the no-timer / no-io idle policy. Spin on `queue_len` for the
|
||||||
|
/// configured budget (if under the spinner cap), then park on the futex.
|
||||||
|
/// Returns when the caller should loop back and re-attempt a pop. Only
|
||||||
|
/// invoked when `self.spinning` (budget > 0); with budget 0 the caller
|
||||||
|
/// keeps the historical `thread::sleep` park.
|
||||||
|
fn idle_spin_then_park(&self) {
|
||||||
|
// 1. Try to enlist as a spinner, strictly honouring the cap (CAS, so
|
||||||
|
// `n_spinning` never exceeds `max_spinners` even transiently).
|
||||||
|
let mut spinning = false;
|
||||||
|
loop {
|
||||||
|
let c = self.n_spinning.load(Ordering::Acquire);
|
||||||
|
if c >= self.max_spinners {
|
||||||
|
break; // cap met (or 0): skip the spin, park directly
|
||||||
|
}
|
||||||
|
if self
|
||||||
|
.n_spinning
|
||||||
|
.compare_exchange_weak(c, c + 1, Ordering::AcqRel, Ordering::Acquire)
|
||||||
|
.is_ok()
|
||||||
|
{
|
||||||
|
spinning = true;
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if spinning {
|
||||||
|
let start = crate::preempt::rdtsc();
|
||||||
|
loop {
|
||||||
|
if self.queue_len.load(Ordering::Acquire) > 0 {
|
||||||
|
// Found work. Leave the spinner set; if we were the LAST
|
||||||
|
// spinner (1 → 0), wake one parked sibling before going
|
||||||
|
// heads-down (rule 4) so a burst still arriving is still
|
||||||
|
// observed by a searcher.
|
||||||
|
let prev = self.n_spinning.fetch_sub(1, Ordering::AcqRel);
|
||||||
|
if prev == 1 {
|
||||||
|
fence(Ordering::SeqCst);
|
||||||
|
if self.queue_len.load(Ordering::Acquire) > 0 {
|
||||||
|
self.unpark_one_scheduler();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return; // back to the pop
|
||||||
|
}
|
||||||
|
if crate::preempt::rdtsc().wrapping_sub(start) >= self.spin_budget_cycles {
|
||||||
|
break; // budget exhausted: park
|
||||||
|
}
|
||||||
|
core::hint::spin_loop();
|
||||||
|
}
|
||||||
|
// Budget exhausted: leave the spinner set, then park (below). The
|
||||||
|
// decrement must precede the park's queue re-check — that is the
|
||||||
|
// parker half of the submit-rule Dekker pairing.
|
||||||
|
self.n_spinning.fetch_sub(1, Ordering::AcqRel);
|
||||||
|
}
|
||||||
|
|
||||||
|
// 2. Park on the futex. Three guards close every lost-wakeup / hang:
|
||||||
|
// - SeqCst fence + queue_len re-check: the Dekker pairing with the
|
||||||
|
// submit rule (which loads n_spinning AFTER publishing the work),
|
||||||
|
// so a submit that declined to wake us because it saw us spinning
|
||||||
|
// is observed here instead.
|
||||||
|
// - capture park_seq BEFORE the final re-check: a submit-wake
|
||||||
|
// landing in the gap bumps the sequence, so futex_wait's
|
||||||
|
// value-compare returns immediately rather than sleeping.
|
||||||
|
// - live_actors re-check: the shutdown guard. In this arm io_out is
|
||||||
|
// 0 by construction (the io>0 case is handled by a sibling arm),
|
||||||
|
// so live_actors == 0 IS the AllDone verdict; if termination is
|
||||||
|
// racing our park we loop back and reach AllDone instead of
|
||||||
|
// sleeping past program end. (unpark_all_schedulers covers the
|
||||||
|
// already-parked case from the other side.)
|
||||||
|
fence(Ordering::SeqCst);
|
||||||
|
let observed = self.park_seq.load(Ordering::Acquire);
|
||||||
|
if self.queue_len.load(Ordering::Acquire) > 0 {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
if self.live_actors.load(Ordering::Acquire) == 0 {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
futex_wait(&self.park_seq, observed);
|
||||||
|
// Woken by a wake, a stale-sequence EAGAIN, or spuriously: loop back
|
||||||
|
// and re-pop. The verdict re-runs from scratch, so any wake is benign.
|
||||||
|
}
|
||||||
|
|
||||||
/// Make `pid` runnable if it is parked; coalesce or defer otherwise.
|
/// Make `pid` runnable if it is parked; coalesce or defer otherwise.
|
||||||
/// The runtime-internal core of `scheduler::unpark`. WILDCARD wake:
|
/// The runtime-internal core of `scheduler::unpark`. WILDCARD wake:
|
||||||
/// consumes the epoch but does not check it — reserved for terminal
|
/// consumes the epoch but does not check it — reserved for terminal
|
||||||
@@ -704,6 +888,10 @@ pub struct Runtime {
|
|||||||
/// Initialise the runtime with the given config. Returns a reusable handle.
|
/// Initialise the runtime with the given config. Returns a reusable handle.
|
||||||
pub fn init(config: Config) -> Runtime {
|
pub fn init(config: Config) -> Runtime {
|
||||||
let n = config.resolved_thread_count();
|
let n = config.resolved_thread_count();
|
||||||
|
// RFC 004: default spinner cap is N/2 (capping at half keeps half the pool
|
||||||
|
// hot, parks the rest). With N=1 this is 0 — a lone thread has no
|
||||||
|
// cross-thread handoff to hide, so it never spins.
|
||||||
|
let max_spinners = config.max_spinners.unwrap_or((n / 2) as u32);
|
||||||
Runtime {
|
Runtime {
|
||||||
inner: RuntimeInner::new(
|
inner: RuntimeInner::new(
|
||||||
n,
|
n,
|
||||||
@@ -712,6 +900,8 @@ pub fn init(config: Config) -> Runtime {
|
|||||||
config.stack_pool_cap,
|
config.stack_pool_cap,
|
||||||
config.max_actors,
|
config.max_actors,
|
||||||
config.wake_slot,
|
config.wake_slot,
|
||||||
|
config.spin_budget_cycles,
|
||||||
|
max_spinners,
|
||||||
),
|
),
|
||||||
thread_count: n,
|
thread_count: n,
|
||||||
}
|
}
|
||||||
@@ -1207,7 +1397,12 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot_idx: usize) {
|
|||||||
|
|
||||||
stats.run_queue_len.store(inner.run_queue.len(), Ordering::Relaxed);
|
stats.run_queue_len.store(inner.run_queue.len(), Ordering::Relaxed);
|
||||||
let pop = match inner.run_queue.pop() {
|
let pop = match inner.run_queue.pop() {
|
||||||
Some(pid) => Pop::Got(pid),
|
Some(pid) => {
|
||||||
|
// RFC 004: mirror the enqueue increment. Release pairs with
|
||||||
|
// the spinners' Acquire load of queue_len.
|
||||||
|
inner.queue_len.fetch_sub(1, Ordering::Release);
|
||||||
|
Pop::Got(pid)
|
||||||
|
}
|
||||||
None => {
|
None => {
|
||||||
// Termination does not lean on pop-None being a fence (with
|
// Termination does not lean on pop-None being a fence (with
|
||||||
// the ring queues it is only a snapshot). The argument is
|
// the ring queues it is only a snapshot). The argument is
|
||||||
@@ -1249,6 +1444,13 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot_idx: usize) {
|
|||||||
if let Some(io) = inner.io.lock().unwrap().as_ref() {
|
if let Some(io) = inner.io.lock().unwrap().as_ref() {
|
||||||
io.wake();
|
io.wake();
|
||||||
}
|
}
|
||||||
|
// RFC 004: a no-io runtime has no wake pipe; futex-parked
|
||||||
|
// siblings must be released or shutdown hangs. Broadcast;
|
||||||
|
// each re-runs the verdict and reaches AllDone itself. The
|
||||||
|
// sequence bump also EAGAINs any sibling mid-park.
|
||||||
|
if inner.spinning {
|
||||||
|
inner.unpark_all_schedulers();
|
||||||
|
}
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
Pop::Idle { io_outstanding, wake_fd } => {
|
Pop::Idle { io_outstanding, wake_fd } => {
|
||||||
@@ -1274,7 +1476,16 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot_idx: usize) {
|
|||||||
crate::io::drain_wake_pipe(fd);
|
crate::io::drain_wake_pipe(fd);
|
||||||
}
|
}
|
||||||
_ => {
|
_ => {
|
||||||
thread::sleep(std::time::Duration::from_micros(100));
|
// RFC 004: the no-timer / no-io idle arm — the
|
||||||
|
// ~100µs thread::sleep worst case this RFC targets.
|
||||||
|
// With spinning enabled, spin-before-park on a
|
||||||
|
// wakeable futex; with budget 0 the historical
|
||||||
|
// sleep is preserved exactly.
|
||||||
|
if inner.spinning {
|
||||||
|
inner.idle_spin_then_park();
|
||||||
|
} else {
|
||||||
|
thread::sleep(std::time::Duration::from_micros(100));
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
continue;
|
continue;
|
||||||
|
|||||||
@@ -0,0 +1,260 @@
|
|||||||
|
//! RFC 004 spinning-worker tests.
|
||||||
|
//!
|
||||||
|
//! The futex scheduler-park path is NOT loom-modelled (sync_shim.rs: the full
|
||||||
|
//! runtime — context switches, futexes, TLS — is never run under cfg(loom)),
|
||||||
|
//! so the Dekker submit/park ordering is guarded here instead, by driving the
|
||||||
|
//! one true deadlock state and catching it as a timeout rather than a hung
|
||||||
|
//! binary: every scheduler parked on the futex while runnable work exists.
|
||||||
|
//!
|
||||||
|
//! All tests use a NO-IO runtime (pure message passing) so the idle path is
|
||||||
|
//! the bare `_` arm RFC 004 replaces — the futex park — not the timer/io arms,
|
||||||
|
//! which keep their own wake sources. A lost wakeup therefore manifests as a
|
||||||
|
//! permanent stall: a `recv` that never returns, an `rt.run` that never
|
||||||
|
//! reaches AllDone. The watchdog turns that into a failed assertion.
|
||||||
|
|
||||||
|
use smarm::channel::channel;
|
||||||
|
use smarm::runtime::{init, Config};
|
||||||
|
use smarm::spawn;
|
||||||
|
use std::sync::atomic::{AtomicU64, Ordering};
|
||||||
|
use std::sync::Arc;
|
||||||
|
use std::time::Duration;
|
||||||
|
|
||||||
|
/// Run `body` on a side thread; fail (rather than hang the test binary) if it
|
||||||
|
/// does not finish within `secs`. A hang here means a scheduler parked on the
|
||||||
|
/// futex with work outstanding and nothing woke it — exactly the RFC 004
|
||||||
|
/// failure mode. The leaked thread is acceptable: the process exits non-zero.
|
||||||
|
fn run_with_timeout(secs: u64, body: impl FnOnce() + Send + 'static) {
|
||||||
|
let (tx, rx) = std::sync::mpsc::channel();
|
||||||
|
std::thread::spawn(move || {
|
||||||
|
body();
|
||||||
|
let _ = tx.send(());
|
||||||
|
});
|
||||||
|
if rx.recv_timeout(Duration::from_secs(secs)).is_err() {
|
||||||
|
panic!(
|
||||||
|
"RFC 004 deadlock: workload did not terminate within {secs}s \
|
||||||
|
— a scheduler wakeup was lost (work stranded with all schedulers \
|
||||||
|
parked on the futex)"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A single token ping-ponging between two actors for `rounds` rounds, on a
|
||||||
|
/// `threads`-wide no-io runtime with the given spin budget. With more threads
|
||||||
|
/// than active actors, the surplus schedulers go idle and park; every send to
|
||||||
|
/// the (parked) peer is a cross-thread wakeup. A lost wake stalls the bounce.
|
||||||
|
/// The post-run count confirms no message was silently dropped.
|
||||||
|
fn bounce(threads: usize, rounds: u64, budget: u64) {
|
||||||
|
run_with_timeout(30, move || {
|
||||||
|
let rt = init(Config::exact(threads).spin_budget_cycles(budget));
|
||||||
|
let echoed = Arc::new(AtomicU64::new(0));
|
||||||
|
let echoed_actor = echoed.clone();
|
||||||
|
rt.run(move || {
|
||||||
|
let (tx_ab, rx_ab) = channel::<u64>();
|
||||||
|
let (tx_ba, rx_ba) = channel::<u64>();
|
||||||
|
let echo = spawn(move || {
|
||||||
|
for _ in 0..rounds {
|
||||||
|
let v = rx_ab.recv().expect("echo recv");
|
||||||
|
echoed_actor.fetch_add(1, Ordering::Relaxed);
|
||||||
|
tx_ba.send(v + 1).expect("echo send");
|
||||||
|
}
|
||||||
|
});
|
||||||
|
for i in 0..rounds {
|
||||||
|
tx_ab.send(i).expect("ping send");
|
||||||
|
let r = rx_ba.recv().expect("ping recv");
|
||||||
|
assert_eq!(r, i + 1, "token corrupted in flight");
|
||||||
|
}
|
||||||
|
echo.join().expect("echo join");
|
||||||
|
});
|
||||||
|
assert_eq!(
|
||||||
|
echoed.load(Ordering::Relaxed),
|
||||||
|
rounds,
|
||||||
|
"messages were lost, not just delayed"
|
||||||
|
);
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Tiny budget: schedulers park almost immediately, so essentially every
|
||||||
|
/// cross-thread handoff exercises the futex wake. Sharpest stress on the
|
||||||
|
/// submit-rule / lost-wakeup guards.
|
||||||
|
#[test]
|
||||||
|
fn bounce_forces_park_and_wake() {
|
||||||
|
bounce(4, 20_000, 2_000);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Large budget: idle schedulers stay hot spinners, so the submit rule mostly
|
||||||
|
/// takes the `n_spinning > 0` skip and the last-spinner handoff (rule 4) does
|
||||||
|
/// the waking. Confirms that path also makes progress and stays correct.
|
||||||
|
#[test]
|
||||||
|
fn bounce_with_hot_spinners() {
|
||||||
|
bounce(4, 20_000, 2_000_000);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Oversubscribed: more active actors than schedulers, so the run queue is
|
||||||
|
/// rarely empty and the park/spin/submit machinery is hammered from every
|
||||||
|
/// thread at once.
|
||||||
|
#[test]
|
||||||
|
fn bounce_oversubscribed() {
|
||||||
|
bounce(2, 30_000, 5_000);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Bursty fork: a parent makes many children runnable at once (the motivating
|
||||||
|
/// fan-out), each sends one reply, the parent collects all of them. Exercises
|
||||||
|
/// the submit rule under a burst — the first job eats one wake, the woken
|
||||||
|
/// worker becomes a spinner and absorbs the rest — and confirms none are lost.
|
||||||
|
fn fanout(threads: usize, children: u64, budget: u64) {
|
||||||
|
run_with_timeout(30, move || {
|
||||||
|
let rt = init(Config::exact(threads).spin_budget_cycles(budget));
|
||||||
|
rt.run(move || {
|
||||||
|
let (tx, rx) = channel::<u64>();
|
||||||
|
for c in 0..children {
|
||||||
|
let tx = tx.clone();
|
||||||
|
spawn(move || {
|
||||||
|
tx.send(c).expect("child send");
|
||||||
|
});
|
||||||
|
}
|
||||||
|
drop(tx);
|
||||||
|
let mut sum = 0u64;
|
||||||
|
for _ in 0..children {
|
||||||
|
sum += rx.recv().expect("collector recv");
|
||||||
|
}
|
||||||
|
assert_eq!(sum, (0..children).sum(), "a child reply was lost");
|
||||||
|
});
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn fanout_burst_parks() {
|
||||||
|
fanout(4, 256, 2_000);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn fanout_burst_hot() {
|
||||||
|
fanout(4, 256, 2_000_000);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Shutdown guard: a mostly-idle spinning runtime must still terminate. Most
|
||||||
|
/// schedulers park on the futex with no work ever coming; only the AllDone
|
||||||
|
/// broadcast (`unpark_all_schedulers`) can release them. If that broadcast is
|
||||||
|
/// missing, `rt.run` never returns and the watchdog fires.
|
||||||
|
#[test]
|
||||||
|
fn idle_runtime_shuts_down() {
|
||||||
|
run_with_timeout(15, || {
|
||||||
|
let rt = init(Config::exact(8).spin_budget_cycles(2_000));
|
||||||
|
rt.run(|| {
|
||||||
|
// One trivial cross-thread handoff, then everyone goes idle and the
|
||||||
|
// runtime must drain to AllDone through parked schedulers.
|
||||||
|
let (tx, rx) = channel::<u64>();
|
||||||
|
let h = spawn(move || {
|
||||||
|
tx.send(42).expect("send");
|
||||||
|
});
|
||||||
|
assert_eq!(rx.recv().expect("recv"), 42);
|
||||||
|
h.join().expect("join");
|
||||||
|
});
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Repeated lifecycles: init→run→AllDone several times in one process, to
|
||||||
|
/// catch any park/broadcast state that fails to reset between runtimes.
|
||||||
|
#[test]
|
||||||
|
fn repeated_spinning_lifecycles() {
|
||||||
|
run_with_timeout(30, || {
|
||||||
|
for _ in 0..20 {
|
||||||
|
let rt = init(Config::exact(4).spin_budget_cycles(2_000));
|
||||||
|
let echoed = Arc::new(AtomicU64::new(0));
|
||||||
|
let e = echoed.clone();
|
||||||
|
rt.run(move || {
|
||||||
|
let (tx, rx) = channel::<u64>();
|
||||||
|
let worker = spawn(move || {
|
||||||
|
for _ in 0..500 {
|
||||||
|
let v = rx.recv().expect("recv");
|
||||||
|
e.fetch_add(v, Ordering::Relaxed);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
for i in 0..500u64 {
|
||||||
|
tx.send(i).expect("send");
|
||||||
|
}
|
||||||
|
worker.join().expect("worker join");
|
||||||
|
});
|
||||||
|
assert_eq!(echoed.load(Ordering::Relaxed), (0..500u64).sum());
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Shutdown with siblings *pinned* in the futex park — the test that actually
|
||||||
|
/// exercises the AllDone broadcast. A single long-running actor holds `live`
|
||||||
|
/// at 1 (CPU-bound, never sending or spawning) long enough for every other
|
||||||
|
/// scheduler to idle, spin out its budget, and commit to `futex_wait`. Because
|
||||||
|
/// nothing is ever enqueued during that window, the submit-rule wakes never
|
||||||
|
/// fire and those siblings stay parked; the final finalize enqueues no one, so
|
||||||
|
/// the live-check self-termination cannot save them either. Only
|
||||||
|
/// `unpark_all_schedulers` at AllDone can. Gut that broadcast and this hangs.
|
||||||
|
#[test]
|
||||||
|
fn shutdown_releases_pinned_parked_siblings() {
|
||||||
|
run_with_timeout(15, || {
|
||||||
|
let rt = init(Config::exact(8).spin_budget_cycles(2_000));
|
||||||
|
rt.run(|| {
|
||||||
|
// ~50ms of pure CPU: far longer than the µs it takes the other
|
||||||
|
// seven schedulers to park, so they are provably in futex_wait
|
||||||
|
// before this returns and live drops to 0.
|
||||||
|
let start = std::time::Instant::now();
|
||||||
|
let mut x = 0u64;
|
||||||
|
while start.elapsed() < Duration::from_millis(50) {
|
||||||
|
x = x.wrapping_mul(6364136223846793005).wrapping_add(1);
|
||||||
|
std::hint::black_box(x);
|
||||||
|
}
|
||||||
|
std::hint::black_box(x);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
/// cross-thread handoff — but the futex park path must still be inert and
|
||||||
|
/// correct (the lone thread is never parked while it has work). Pins the
|
||||||
|
/// degenerate cap=0 case.
|
||||||
|
#[test]
|
||||||
|
fn single_thread_spinning_is_inert() {
|
||||||
|
run_with_timeout(15, || {
|
||||||
|
let rt = init(Config::exact(1).spin_budget_cycles(2_000));
|
||||||
|
let echoed = Arc::new(AtomicU64::new(0));
|
||||||
|
let e = echoed.clone();
|
||||||
|
rt.run(move || {
|
||||||
|
let (tx, rx) = channel::<u64>();
|
||||||
|
let worker = spawn(move || {
|
||||||
|
for _ in 0..1_000 {
|
||||||
|
let v = rx.recv().expect("recv");
|
||||||
|
e.fetch_add(v, Ordering::Relaxed);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
for i in 0..1_000u64 {
|
||||||
|
tx.send(i).expect("send");
|
||||||
|
}
|
||||||
|
worker.join().expect("join");
|
||||||
|
});
|
||||||
|
assert_eq!(echoed.load(Ordering::Relaxed), (0..1_000u64).sum());
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
/// `spin_budget_cycles == 0` is the opt-out: the feature is fully off and the
|
||||||
|
/// runtime uses the historical `thread::sleep` idle path. Must still run a
|
||||||
|
/// cross-thread workload to completion (it always did) — this just guards that
|
||||||
|
/// the gating left the off-path intact.
|
||||||
|
#[test]
|
||||||
|
fn budget_zero_keeps_old_behaviour() {
|
||||||
|
run_with_timeout(20, || {
|
||||||
|
let rt = init(Config::exact(4).spin_budget_cycles(0));
|
||||||
|
let sum = Arc::new(AtomicU64::new(0));
|
||||||
|
let s = sum.clone();
|
||||||
|
rt.run(move || {
|
||||||
|
let (tx, rx) = channel::<u64>();
|
||||||
|
let worker = spawn(move || {
|
||||||
|
for _ in 0..1_000 {
|
||||||
|
s.fetch_add(rx.recv().expect("recv"), Ordering::Relaxed);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
for i in 0..1_000u64 {
|
||||||
|
tx.send(i).expect("send");
|
||||||
|
}
|
||||||
|
worker.join().expect("join");
|
||||||
|
});
|
||||||
|
assert_eq!(sum.load(Ordering::Relaxed), (0..1_000u64).sum());
|
||||||
|
});
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user