Waits like the OpenMP runtimes: bounded poll spin, yields, then sleep; wait benchmarks (faer#107)

Cargo.toml

@@ -83,3 +83,8 @@ unexpected_cfgs = { level = "warn", check-cfg = ['cfg(loom)'] }
 [[bench]]
 name = "cost_lookup"
 harness = false
+
+[[bench]]
+name = "wait_scenarios"
+harness = false
+required-features = ["parallel"]

README.md

@@ -58,7 +58,8 @@ the token to kernels by value. Kernels never store a token and never mint one.
 | `Exec::team(f)` / `team_with(over, f)` | reserve `budget - 1` awake pool members for a scope and run barrier-bearing sections through `TeamExec` (`run`, `run_with`, `split`, `fork`, `reborrow`), or `Err(Refused::Unavailable)`; see the design record |
 | `Pool::into_team(width)` -> `Team` (owns the pool; `IntoTeamError` gives it back), `Team::exec(budget)` / `try_exec`, `Team::pool`, `Team::into_pool` | persistent reservation for an engine that owns its pool (`Pool` is not `Clone`); minted tokens are ordinary `Exec`s whose `fork`, `for_ranges`, `broadcast`, `team` dispatch on the reserved members; one leader at a time (`exec` blocks, `try_exec` -> `Refused::Busy`); idle members park while reserved |
 | `PoolBuilder::qos(Qos)`, `raise_current_thread_qos(Qos)` | worker QoS on macOS (default `UserInitiated`); raise the leading thread without ever lowering it |
-| `PoolBuilder::idle_spin`, `idle_runs_rayon`, `acquire_timeout`; `Pool::release_idle` | waiting policy of idle team members; a reserved member spins with one PAUSE per poll (never `sched_yield`), parks after the delay, and parks after a short spin in a root scope that has more participants than the caller's CPUs |
+| `PoolBuilder::idle_spin`, `idle_runs_rayon`, `acquire_timeout`; `Pool::release_idle` | waiting policy of idle team members (an idle member returns to Rayon after `idle_spin` or the spin budget, whichever comes first) |
+| `PoolBuilder::spin_polls`, `Pool::spin_polls` | spin budget of every team wait, in polls (the shape of libgomp's `GOMP_SPINCOUNT`): a reserved member waiting for work, the leader in a join and a participant in a barrier spin that many polls, yield a few times, then sleep until woken; throttled to 100 polls in an oversubscribed scope or after a waiter was descheduled ([`docs/design/waiting.md`](docs/design/waiting.md)) |
 | `PoolBuilder::pin_participants(bool)`, `Pool::pin_participants()` | opt-in placement of the **members** of a root team scope (Linux; off by default, which is consistent with the rule it protects): member `k` (participant `k >= 1`) pins itself onto the `k`-th CPU of the calling thread's affinity set, `S[0]` is left to the leader, and a member restores its own mask before it returns to Rayon or serves foreign work. **The leader is never pinned**, because a vendor library creates its thread team from the calling thread and a one-CPU caller would confine the whole provider team; a member that calls such a library directly inside a share does confine it, so use `Exec::external` (which restores the member's mask first) or leave pinning off. For tensor4all/faer#107; give each team its own CPU set (`taskset -c`), there is no CPU allocator |
 | `DispatchCosts`, `WidthCosts`, `PoolBuilder::dispatch_costs`, `Exec::dispatch_cost(width)`, `TeamExec::dispatch_cost(width)` | cost of one parallel section per kind of token (team, in pool, outside), steady state; one row per `HostClass` (`HostClass::detect()`, keyed by kernel dispatch: `x86_64-avx512`, `x86_64-avx2`, `x86_64-scalar`, `aarch64-apple`, `aarch64`, `generic`), `DispatchCosts::for_class(class)`; measured rows for `x86_64-avx512` (team anchors at widths 2, 4, 6 and 8) and `aarch64-apple`, a measured Zen2 team row for `x86_64-avx2` (anchors at ten widths from 2 to 64; its plain kinds stay generic), every other class takes the conservative `generic` row (3 / 15 / 40 us at the anchors 2, 4, 6); a row holds measured anchors (`WidthCosts::new`, `WidthCosts::flat`, or `WidthCosts::from_anchors` for measured anchors up to 64 and beyond), interpolates between them and grows linearly above the largest one, so a width above the anchors is never charged the cost of the largest anchor; `DispatchCosts::host_default()` is the detected class's row; `Duration::MAX` when serial; override per pool |
 | `pays_at(est, sections, cost)`, `Exec::pays(est, sections, width)`, `Exec::width_for(est, sections)`, `width_for_by`, `PAYS` | the one rule: parallel iff `est >= PAYS * sections * D(width)`; `width_for` minimises `sections * D(k) + est / k` |
@@ -107,6 +108,50 @@ Consumers assert `ledger::max_live(root) <= budget` in their own tests, and keep
 a test binary that runs serial and pooled work and then checks that
 `rayon::ThreadPoolBuilder::new().build_global()` still succeeds.
 
+## Wait benchmarks
+
+`benches/wait_scenarios.rs` measures how team threads wait, the way callers use them; the design
+is in [`docs/design/waiting.md`](docs/design/waiting.md). A change to the waits is judged by running
+it on the base and on the head of the change, on a quiet host and on a many-core shared host.
+
+| scenario | a call is | reports |
+| --- | --- | --- |
+| `sections` | 200 sections of 5, 20 or 50 µs, team held | per call, and the overhead over `sections * work` |
+| `gaps` | 10 sections of 5 µs, the caller sleeps 0 / 100 µs / 1 ms / 10 ms between calls | the first section (wake latency) and the call |
+| `acquire` | 50 sections of 20 µs: the team held (a token per call), the pool held (members acquired per call), a new pool per call | per call; for a new pool also the total from build to drop |
+| `nested` | 100 splits of the team into pairs, each pair one 20 µs section | per call |
+
+Each scenario runs at width = all CPUs of the affinity set and at three quarters of them, on
+threadteam and on a Rayon `scope`; `benches/omp/wait_omp.c` runs `sections` and `gaps` on libgomp
+(one `omp parallel` region per section, runtime defaults). `--load K:P` adds `K` child processes,
+each busy `P` % of every 10 ms, for the whole run: the stand-in for another user's processes.
+
+```sh
+# one checkout, both passes (quiet, then 8 processes at 10 %), rows appended to
+# docs/design/waiting-data/<host>.csv
+scripts/bench-waits.sh <label>                 # e.g. main-4c73c2b, then the PR head
+scripts/bench-waits.sh <label> 64,48 8:10      # explicit widths and load
+cargo bench --bench wait_scenarios -- --scenario gaps --widths 8   # one scenario; options in the file header
+```
+
+* **evo** (quiet): `scripts/bench-waits.sh <label> "" none` on the base and on the head. The
+  acceptance check is that `sections` and `gaps` do not regress.
+* **broccoli** (64 CPUs, shared): `taskset -c 0-63 scripts/bench-waits.sh <label> 64,48 8:10` on the
+  base and on the head, plus the faer#107 arms of `acquire_probe_107` (tensor4all/faer branch
+  `probe/107-acquire-per-call`). The acceptance check is that width 64 under load is no longer
+  slow and width 48 does not regress.
+
+To compare two commits on a shared host, `scripts/bench-ab.sh <base-rev> <head-rev> <rounds>
+[bench args]` builds both and alternates them, so a drift of other users' load falls on both sides;
+`HEAD_VARIANTS='--spin-polls 300000;--own-hold-ms 100'` adds head runs with tuning flags (this
+version only), and `OMP=1` runs the libgomp driver in every round with the same scenario, widths,
+calls and load (`acquire` is libgomp's held team). The base of the waits change is its bench commit `a3a9320` (the bench on the 0.1.3
+waits). Under every threadteam row, a `waits per call` line counts join and barrier sleeps, member
+parks and the preemptions waiters saw.
+
+Results are committed as CSV under [`docs/design/waiting-data/`](docs/design/waiting-data/), one
+file per host, with the label naming the commit; the summary tables live in `waiting.md`.
+
 ## License
 
 Licensed under either of MIT or Apache-2.0 at your option.

benches/omp/wait_omp.c

@@ -0,0 +1,169 @@
+/* The libgomp reference for benches/wait_scenarios.rs (docs/design/waiting.md).
+ *
+ * The same work as the Rust bench: a section is one `omp parallel` region in which every thread
+ * busy-waits `work` microseconds on the clock; a call is a run of sections. Prints rows in the
+ * Rust bench's format and appends them to a CSV file when one is given.
+ *
+ * Build and run with scripts/bench-waits.sh or scripts/bench-ab.sh (OMP=1), or:
+ *   gcc -O2 -fopenmp benches/omp/wait_omp.c -o target/wait_omp
+ *   target/wait_omp <width> <calls> [csv] [label]
+ * The OpenMP runtime runs with its defaults (no OMP_WAIT_POLICY, no GOMP_SPINCOUNT, no binding)
+ * unless the caller sets them.
+ *
+ * Environment:
+ *   WAIT_OMP_SCENARIO  all (default) | sections | gaps | acquire
+ *   WAIT_OMP_LOAD      K:P, foreign load: K child processes each busy P % of every 10 ms
+ *
+ * `acquire` / `held_team` is libgomp's counterpart of the Rust bench's held team: calls of 50
+ * sections of 20 us on the same (hot) team, back to back.
+ */
+#define _GNU_SOURCE
+#include <omp.h>
+#include <sched.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <time.h>
+#include <signal.h>
+#include <sys/wait.h>
+#include <unistd.h>
+
+static double now_us(void) {
+  struct timespec t;
+  clock_gettime(CLOCK_MONOTONIC, &t);
+  return t.tv_sec * 1e6 + t.tv_nsec / 1e3;
+}
+
+static inline void relax(void) {
+#if defined(__x86_64__) || defined(__i386__)
+  __builtin_ia32_pause();
+#elif defined(__aarch64__)
+  __asm__ __volatile__("yield");
+#endif
+}
+
+static void busy(double us) {
+  double end = now_us() + us;
+  while (now_us() < end) relax();
+}
+
+static int cmp(const void *a, const void *b) {
+  double x = *(const double *)a, y = *(const double *)b;
+  return (x > y) - (x < y);
+}
+
+static FILE *csv;
+static char prefix[512];
+
+static void row(const char *scenario, int width, const char *param, const char *metric, double *v,
+                int n, double ideal) {
+  qsort(v, n, sizeof *v, cmp);
+  double med = v[n / 2], p90 = v[(n * 9 / 10) < n ? n * 9 / 10 : n - 1], max = v[n - 1];
+  printf("%-9s %-10s %5d %-22s %-7s %11.1f %11.1f %11.1f %10.1f\n", scenario, "libgomp", width,
+         param, metric, med, p90, max, med - ideal);
+  if (csv)
+    fprintf(csv, "%s,%s,libgomp,%d,%s,%s,%.2f,%.2f,%.2f,%.2f\n", prefix, scenario, width, param,
+            metric, med, p90, max, med - ideal);
+}
+
+/* One call: the time of the first section and of the whole call, in us. */
+static void call(int width, int sections, double work, double *first, double *whole) {
+  double t0 = now_us();
+  for (int s = 0; s < sections; s++) {
+#pragma omp parallel num_threads(width)
+    busy(work);
+    if (s == 0) *first = now_us() - t0;
+  }
+  *whole = now_us() - t0;
+}
+
+static pid_t load_children[256];
+static int load_n;
+
+/* Foreign load: K children each busy P % of every 10 ms, until killed. */
+static void load_start(const char *spec) {
+  int k = 0, p = 0;
+  if (!spec || sscanf(spec, "%d:%d", &k, &p) != 2) return;
+  for (int i = 0; i < k && i < 256; i++) {
+    pid_t c = fork();
+    if (c == 0) {
+      for (;;) {
+        double t = now_us();
+        while (now_us() - t < 100.0 * p) relax();
+        usleep((useconds_t)(10000 - 100 * p > 0 ? 10000 - 100 * p : 0));
+      }
+    }
+    if (c > 0) load_children[load_n++] = c;
+  }
+  usleep(200000);
+}
+
+static void load_stop(void) {
+  for (int i = 0; i < load_n; i++) {
+    kill(load_children[i], SIGKILL);
+    waitpid(load_children[i], NULL, 0);
+  }
+}
+
+static int want(const char *s) {
+  const char *w = getenv("WAIT_OMP_SCENARIO");
+  return !w || !strcmp(w, "all") || !strcmp(w, s);
+}
+
+int main(int argc, char **argv) {
+  if (argc < 3) {
+    fprintf(stderr, "usage: wait_omp <width> <calls> [csv] [label]\n");
+    return 2;
+  }
+  int width = atoi(argv[1]), calls = atoi(argv[2]);
+  cpu_set_t set;
+  int cpus = sched_getaffinity(0, sizeof set, &set) == 0 ? CPU_COUNT(&set) : 0;
+  char host[256] = "";
+  gethostname(host, sizeof host - 1);
+  const char *label = argc > 4 ? argv[4] : "";
+  const char *load_spec = getenv("WAIT_OMP_LOAD");
+  char load_text[64];
+  int lk = 0, lp = 0;
+  if (load_spec && sscanf(load_spec, "%d:%d", &lk, &lp) == 2)
+    snprintf(load_text, sizeof load_text, "%dx%d%%", lk, lp);
+  else
+    snprintf(load_text, sizeof load_text, "%s",
+             getenv("WAIT_BENCH_LOAD") ? getenv("WAIT_BENCH_LOAD") : "none");
+  const char *load = load_text;
+  snprintf(prefix, sizeof prefix, "%s,%s,%d,%s", host, label, cpus, load);
+  if (argc > 3) {
+    int fresh = access(argv[3], F_OK) != 0;
+    csv = fopen(argv[3], "a");
+    if (csv && fresh)
+      fprintf(csv, "host,label,cpus,load,scenario,impl,width,param,metric,median_us,p90_us,max_us,overhead_us\n");
+  }
+  double *a = malloc(sizeof(double) * calls), *b = malloc(sizeof(double) * calls);
+  char param[64];
+  load_start(load_spec);
+  /* warm the hot team */
+  for (int i = 0; i < 10; i++) call(width, 10, 5, a, b);
+
+  int works[3] = {5, 20, 50};
+  for (int k = 0; k < 3 && want("sections"); k++) {
+    for (int i = 0; i < calls; i++) call(width, 200, works[k], &a[i], &b[i]);
+    snprintf(param, sizeof param, "work=%dus,sections=200", works[k]);
+    row("sections", width, param, "call", b, calls, 200.0 * works[k]);
+  }
+  int gaps[4] = {0, 100, 1000, 10000};
+  for (int k = 0; k < 4 && want("gaps"); k++) {
+    for (int i = 0; i < calls / 2; i++) {
+      if (gaps[k]) usleep(gaps[k]);
+      call(width, 10, 5, &a[i], &b[i]);
+    }
+    snprintf(param, sizeof param, "gap=%dus", gaps[k]);
+    row("gaps", width, param, "first", a, calls / 2, 5);
+    row("gaps", width, param, "call", b, calls / 2, 50);
+  }
+  if (want("acquire")) {
+    for (int i = 0; i < calls / 2; i++) call(width, 50, 20, &a[i], &b[i]);
+    row("acquire", width, "held_team", "call", b, calls / 2, 1000);
+  }
+  load_stop();
+  if (csv) fclose(csv);
+  return 0;
+}