From 7a8abcdda76cd8066fcac219ee31e725ef75c0d8 Mon Sep 17 00:00:00 2001 From: gjjkbssg <74439300+gjjkbssg@users.noreply.github.com> Date: Thu, 27 Aug 2026 23:45:51 +0800 Subject: [PATCH] fix(telemetry): attribute critical-path stall explicitly (#169) Overlap stall is the union of stalled intervals, not the per-thread mean; the app panel draws compute / flash wait / cache mgmt / unattributed from measured terms. Fixes #98. --- CHANGELOG.md | 17 +++ core/include/bmoe/expert_source.h | 5 +- core/src/engine/session.cpp | 8 +- core/src/moe/expert_stream_source.cpp | 14 +- core/src/moe/expert_stream_source.h | 6 +- core/src/moe/stall_union.h | 100 ++++++++++++++ docs/telemetry.md | 42 ++++-- .../io/bigmoeonedge/example/MainActivity.kt | 10 +- .../io/bigmoeonedge/example/MetricFields.kt | 2 +- .../io/bigmoeonedge/example/RunService.kt | 7 +- .../java/io/bigmoeonedge/example/Telemetry.kt | 91 ++++++++----- tests/CMakeLists.txt | 7 + tests/stall_union_test.cpp | 128 ++++++++++++++++++ 13 files changed, 378 insertions(+), 59 deletions(-) create mode 100644 core/src/moe/stall_union.h create mode 100644 tests/stall_union_test.cpp diff --git a/CHANGELOG.md b/CHANGELOG.md index df5df24..2e4b329 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,23 @@ All notable changes to this project are documented here. The format follows [Keep a Changelog](https://keepachangelog.com/), and the project aims to follow Semantic Versioning. +## [0.23.0] - 2026-08-27 + +### Changed +- **Telemetry attribution stops calling unmeasured runtime "compute".** Overlap `stall` is now the + **union of stalled intervals** — the cumulative wall time during which at least one compute thread + was blocked on a streamed expert — instead of the summed per-thread block time divided by the + thread count, a mean that understated the stall whenever a minority of threads did the waiting and + quietly dumped the difference into the `compute_ms` residual. The residual `compute_ms` / + `compute_s_tok` fields keep their existing meaning for protocol compatibility. The app panel now + draws four bars — compute (process CPU time over the compute threads, an attribution proxy), + flash wait (measured `io`/`stall`, in the end-of-run summary too: `io_s_tok`/`stall_s_tok` instead + of inverting the clamped residual), cache mgmt, and **unattributed** (the off-CPU wall time — zram, + preemption, faults — the residual used to absorb). +- **Compatibility note:** under `--overlap`, `stall_ms` and `stall_s_tok` changed meaning — they are + the critical-path union now — so these columns are **not comparable with CSVs produced by older + releases**; compare them only within one engine version. Fixes #98. + ## [0.21.0] - 2026-08-25 ### Added diff --git a/core/include/bmoe/expert_source.h b/core/include/bmoe/expert_source.h index 8167f3d..afa1e3d 100644 --- a/core/include/bmoe/expert_source.h +++ b/core/include/bmoe/expert_source.h @@ -72,7 +72,10 @@ public: long long cache_hits = 0; // expert lookups served from the cache long long cache_lookups = 0; // total expert lookups (hits + misses) uint64_t cache_resident_bytes = 0; // currently resident cached slice bytes - double stall_seconds = 0.0; // overlap: summed across compute threads (0 when serial) + double stall_seconds = 0.0; // overlap: cumulative wall time during which at least one compute + // thread was stalled on a streamed expert — the UNION of stalled + // intervals, not a per-thread sum (0 when serial). Includes an + // interval still open at the snapshot. uint64_t spec_read_bytes = 0; // bytes read speculatively by prefetch (subset of read_bytes) long long spec_experts = 0; // experts fully prefetched long long spec_useful = 0; // prefetched experts that a later lookup actually hit diff --git a/core/src/engine/session.cpp b/core/src/engine/session.cpp index cd9d55c..c7d54c6 100644 --- a/core/src/engine/session.cpp +++ b/core/src/engine/session.cpp @@ -98,7 +98,6 @@ llama_token argmax(const float * logits, int n_vocab) { struct GenTally { // Fixed for the run; kept here so record() needs only the token's own measurements. bool overlap = false; - int n_threads = 1; long long prev_bytes = 0; double prev_io_s = 0.0; @@ -163,7 +162,11 @@ struct GenTally { m.io_ms = (st->read_seconds - prev_io_s) * 1000.0; m.mgmt_ms = (st->mgmt_seconds - prev_mgmt_s) * 1000.0; if (overlap) { - m.stall_ms = (st->stall_seconds - prev_stall_s) * 1000.0 / n_threads; + // stall is already wall-additive (the union of stalled intervals), so no thread-count + // normalization: dividing summed thread time by n_threads was a mean that understated + // the stall whenever a minority of threads did the waiting, and the difference quietly + // became "compute". + m.stall_ms = (st->stall_seconds - prev_stall_s) * 1000.0; m.compute_ms = m.wall_ms - m.stall_ms - m.mgmt_ms; } else { m.compute_ms = m.wall_ms - m.io_ms - m.mgmt_ms; @@ -1143,7 +1146,6 @@ RunResult Session::generate(const GenerateRequest & req, // counters carry the prior prompts' totals; the deltas make each prompt self-relative. GenTally tally; tally.overlap = moe.overlap; - tally.n_threads = im.cfg.n_threads; if (moe.enabled) { const IExpertSource::Stats st0 = im.source.stats(); tally.prev_bytes = (long long) st0.read_bytes; diff --git a/core/src/moe/expert_stream_source.cpp b/core/src/moe/expert_stream_source.cpp index 005fda3..c5ef819 100644 --- a/core/src/moe/expert_stream_source.cpp +++ b/core/src/moe/expert_stream_source.cpp @@ -214,7 +214,7 @@ bool ExpertStreamSource::init(const std::vector & shard_paths, async_gen_.store(0); cur_il_.store(-1); fatal_.store(false); - stall_ns_.store(0); + stall_union_.reset(); batch_flag_gen_ = 0; staged_.reserve(n_expert_); texp_.clear(); @@ -1305,15 +1305,17 @@ void ExpertStreamSource::on_expert_ready(const ggml_tensor * src0, int expert) { const uint32_t want = async_gen_.load(std::memory_order_relaxed); if (ready_[idx].gen.load(std::memory_order_acquire) == want) return; // already resident - const auto t0 = clock_t_::now(); + // The stall interval opens the moment the need is unmet — before the spin, since the spin is + // already waiting — and closes on whichever exit this thread takes. Union accounting, not a + // per-thread sum: see StallUnion. + stall_union_.enter(); // Short spin first: a slice usually lands within microseconds, cheaper than a syscall. The // beat is a pause instruction, not yield() — 2048 yields burnt up to a millisecond of // sched_yield churn per genuinely slow slice, stealing CPU from the I/O lanes and the // sibling compute threads that would have finished the slice sooner. for (int s = 0; s < 256; ++s) { if (ready_[idx].gen.load(std::memory_order_acquire) == want || fatal_.load(std::memory_order_acquire)) { - stall_ns_.fetch_add( - (long long) std::chrono::duration_cast(clock_t_::now() - t0).count()); + stall_union_.exit(); return; } cpu_relax(); @@ -1331,7 +1333,7 @@ void ExpertStreamSource::on_expert_ready(const ggml_tensor * src0, int expert) { }); } ready_waiters_.fetch_sub(1, std::memory_order_seq_cst); - stall_ns_.fetch_add((long long) std::chrono::duration_cast(clock_t_::now() - t0).count()); + stall_union_.exit(); } void ExpertStreamSource::enable_overlap_hook() { @@ -1360,7 +1362,7 @@ IExpertSource::Stats ExpertStreamSource::stats() const { s.cache_hits = chits_; s.cache_lookups = clookups_; s.cache_resident_bytes = (uint64_t) cresident_; - s.stall_seconds = stall_ns_.load() / 1e9; + s.stall_seconds = stall_union_.total_ns() / 1e9; s.cache_budget_bytes = (uint64_t) cache_max_; s.cache_resizes = cache_resizes_; s.evictions = evictions_; diff --git a/core/src/moe/expert_stream_source.h b/core/src/moe/expert_stream_source.h index 7c8c9a2..99eb6a4 100644 --- a/core/src/moe/expert_stream_source.h +++ b/core/src/moe/expert_stream_source.h @@ -17,6 +17,7 @@ #include "bmoe/expert_source.h" #include "bmoe/config.h" #include "bmoe/decode_trace.h" +#include "stall_union.h" #include "bmoe/recipe.h" #include "../io/platform_io.h" #include "../io/file_reader.h" @@ -342,7 +343,10 @@ private: // re-check a predicate that was almost never its own. Registration and publication are both // seq_cst so the two cannot miss each other — see on_expert_ready. std::atomic ready_waiters_{0}; - std::atomic stall_ns_{0}; // summed across all stalling compute threads + // Overlap stall as the UNION of stalled-thread wall intervals (see stall_union.h), + // not the sum of per-thread waits: one blocked thread already means the graph is not + // progressing, and sum/n_threads understates whenever a minority of threads waits. + StallUnion stall_union_; // expert tensor* -> (il<<8)|p. Sorted by pointer and static after init, and probed by every // compute thread for every routed expert — a flat binary search beats hashing the pointer. std::vector> texp_; diff --git a/core/src/moe/stall_union.h b/core/src/moe/stall_union.h new file mode 100644 index 0000000..9600df6 --- /dev/null +++ b/core/src/moe/stall_union.h @@ -0,0 +1,100 @@ +#pragma once + +#include +#include + +namespace bmoe { + +// The interval-union state machine: the cumulative wall time during which at least one thread was +// stalled, rather than the sum of per-thread stall times. enter/exit bracket each thread's stall; +// the clock starts on the 0→1 transition and the interval accumulates on the 1→0 transition, so +// overlapping waits count once and their union is what a wall-additive "the graph was blocked" +// term needs (summed thread stall divided by thread count is a mean, and understates whenever a +// minority of threads does the waiting — see docs/telemetry.md). +// +// Pure arithmetic over injected timestamps — no locking, no clock — which is what makes the union +// rules deterministically testable: overlap, nesting, separation, the 0→1/1→0 boundaries and the +// open-interval snapshot (tests/stall_union_test.cpp). Concurrency lives in the wrapper below. +class StallUnionState { +public: + // A thread begins a stall at `now_ns`. The 0→1 transition opens the interval. + void enter_at(long long now_ns) { + if (stalled_++ == 0) open_ns_ = now_ns; + } + + // A thread ends its stall. The 1→0 transition closes the interval and accumulates it. + void exit_at(long long now_ns) { + // An exit with nothing stalled cannot arise from a correct caller, but refusing it costs + // nothing and keeps a stray exit from driving stalled_ to -1 — which would make the next + // 0→1 look like a close and accumulate garbage. + if (stalled_ == 0) return; + if (--stalled_ == 0) { + total_ns_ += now_ns - open_ns_; + open_ns_ = 0; + } + } + + // Cumulative union through `now_ns`, including an interval still open at the snapshot: a + // stats() read taken mid-stall must not silently lose the stall in progress, or a token-level + // delta spanning it undercounts by the part already elapsed. + long long total_at(long long now_ns) const { return total_ns_ + (stalled_ > 0 ? now_ns - open_ns_ : 0); } + + void reset() { + stalled_ = 0; + open_ns_ = 0; + total_ns_ = 0; + } + +private: + int stalled_ = 0; // threads currently inside a stall (0 = no interval open) + long long open_ns_ = 0; // when the open interval started; meaningful only while stalled_ > 0 + long long total_ns_ = 0; +}; + +// The production wrapper: mutex + monotonic clock around [StallUnionState]. Deliberately a mutex, +// not lock-free atomics — the obvious lock-free shape (atomic waiter count + open-interval +// timestamp) needs the 0→1 opener's timestamp store to be visible to whichever thread performs the +// 1→0 close; that store is sequenced AFTER the opener's counter RMW, so no acquire/release edge on +// the counter covers it, and a stale read silently drops the interval. A mutex makes the invariant +// local and auditable. +// +// The timestamp is taken INSIDE the critical section, on purpose: mutex acquisition order is the +// transition order, and a timestamp captured before acquiring can be older than a transition that +// acquired first — the last thread to leave could close the interval with a time earlier than a +// thread that already exited, undercounting the union (the same applies to a snapshot racing an +// open). Stall tracking uses this short critical section containing only the union-state +// transition and the timestamp capture; it does not hold the readiness mutex or perform I/O. +class StallUnion { +public: + void enter() { + const std::lock_guard lk(mtx_); + state_.enter_at(steady_ns()); + } + + void exit() { + const std::lock_guard lk(mtx_); + state_.exit_at(steady_ns()); + } + + long long total_ns() const { + const std::lock_guard lk(mtx_); + return state_.total_at(steady_ns()); + } + + void reset() { + const std::lock_guard lk(mtx_); + state_.reset(); + } + +private: + static long long steady_ns() { + return (long long) std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()) + .count(); + } + + mutable std::mutex mtx_; + StallUnionState state_; +}; + +} // namespace bmoe diff --git a/docs/telemetry.md b/docs/telemetry.md index ef46114..2fee20f 100644 --- a/docs/telemetry.md +++ b/docs/telemetry.md @@ -26,17 +26,21 @@ BMOE_PROGRESS {"step":,"steps":,"wall_ms":,"io_ms":, mgmt_ms` in serial, `wall_ms − stall_ms − mgmt_ms` under overlap. When that residual is the number in question, `--compute-trace` measures it directly instead (see [Decode traces](#decode-traces)) — at a cost that makes it a diagnostic, not telemetry. - `compute_ms` is **clamped at 0**: the subtraction can go slightly negative under overlap (where - `stall_ms` is a per-thread mean, not a critical path), and a negative compute would be nonsense. - That clamp means the wall-additive identity is not exact in the pathological case — a consumer - that recovers the flash-wait term as `wall_ms − compute_ms − mgmt_ms` gets `wall_ms − mgmt_ms` - when the clamp fires, over-attributing to flash. Read the wall-additive flash term straight from - `io_ms` (serial) / `stall_ms` (overlap) instead of inverting the residual. - Precisely: `stall_ms` is the summed per-thread block time divided by `n_threads`, which equals the - wall stall only if every compute thread blocks together. When one thread waits on an expert while - the others keep working it **under**-states the stall, and `compute_ms` — being the residual — - absorbs the difference. Read an attribution between compute and flash as approximate, and reach - for `--compute-trace` when the split itself is the question. + `compute_ms` is **clamped at 0** — a negative compute would be nonsense. That clamp means the + wall-additive identity is not exact in the pathological case — a consumer that recovers the + flash-wait term as `wall_ms − compute_ms − mgmt_ms` gets `wall_ms − mgmt_ms` when the clamp + fires, over-attributing to flash. Read the wall-additive flash term straight from `io_ms` + (serial) / `stall_ms` (overlap) instead of inverting the residual. + `stall_ms` is the **union of stalled intervals**: the cumulative wall time during which at least + one compute thread was blocked on a streamed expert. Overlapping waits count once, so it is the + critical-path quantity — one blocked thread already means the graph is not progressing. (It was + previously the summed per-thread block time divided by `n_threads`, a mean that equaled the wall + stall only if every compute thread blocked together and understated it whenever a minority of + threads did the waiting — with the difference silently landing in `compute_ms`.) The interval + opens the moment a thread finds its expert unready — including the short pre-block spin — and a + stats snapshot taken mid-stall includes the open interval up to now. Attribution between compute + and flash is still approximate (see `--compute-trace` when the split itself is the question), but + the flash term no longer depends on how the waiting was distributed across threads. In serial mode `io_ms` is the wall time blocked on reads (a subset of `wall_ms`). Under `--overlap` its meaning changes: it is the **sum of per-lane busy time**, so it can exceed `wall_ms` because lanes read in parallel with compute. Use `stall_ms` for the wall time @@ -62,6 +66,22 @@ BMOE_PROGRESS {"step":,"steps":,"wall_ms":,"io_ms":, `experts_dropped` next to it. - `majflt` / `cpu_ms` **decompose the `compute_ms` residual** — the whole point being that "compute" above is a catch-all that silently absorbs page faults and scheduler stalls, not just matmul. +- **The app panel never reads the residual as compute.** Its four bars are attribution components, + not a partition of wall time: **compute** = `cpu_ms ÷ compute threads` — an attribution *proxy* + (process CPU time, which includes the I/O lanes' work, divided down to a per-compute-thread + figure; it is not a direct measurement of matrix-kernel execution), and because process CPU also + covers the I/O lanes and other process threads, it is an **upper bound** on compute-thread + CPU-equivalent time rather than a disjoint share of the token. Under heavy streaming the lanes' + CPU is large enough to matter: compute + flash wait + mgmt can exceed `wall_ms`, and the panel + clamps the unattributed remainder at 0 in that case instead of displaying the overlap. + **Flash wait** = `io_ms` (serial) / `stall_ms` (overlap) read as + measured in both the live and the end-of-run summary views (`io_s_tok` / `stall_s_tok` in + `BMOE_DONE`), **cache mgmt** = `mgmt_ms`, and **unattributed** = the non-negative wall-time + remainder — the off-CPU time (zram swap-in, preemption, frequency caps) the residual used to + paint as compute. The bars are deliberately not rescaled to total 100 %: measurement noise is + preferred to a fabricated normalization. The CPU-busy diagnostic keeps its own denominator — + `cpu ÷ (wall × busy threads)`, busy threads including the I/O lanes under overlap — which is a + different quantity from the compute bar's and must not be "simplified" into it. They are measured directly around `llama_decode` (no submodule patch needed): `majflt` is the major page faults served this token — a non-zero count means a mmap-resident (dense) weight was re-faulted from flash *inside* the decode, i.e. a >RAM residency stall masquerading as compute. diff --git a/examples/android/app/src/main/java/io/bigmoeonedge/example/MainActivity.kt b/examples/android/app/src/main/java/io/bigmoeonedge/example/MainActivity.kt index 1c4a863..62176d2 100644 --- a/examples/android/app/src/main/java/io/bigmoeonedge/example/MainActivity.kt +++ b/examples/android/app/src/main/java/io/bigmoeonedge/example/MainActivity.kt @@ -555,8 +555,10 @@ private fun TelemetryCard(ui: UiState, threads: Int, overlap: Boolean, ioThreads // The compute-vs-flash split and cache hit rate only mean anything with the streamer // running. Under mmap the model faults in through the OS page cache, invisible here. // The split itself — live last token vs run average, and which term is measured - // rather than residual — is derived in breakdown(); this only draws it. - val b = breakdown(t, overlap, busyThreads = threads + if (overlap) ioThreads else 0) + // rather than residual — is derived in breakdown(); this only draws it. The busy + // denominator counts the I/O lanes; the compute one does not (see breakdown()). + val b = breakdown(t, overlap, + busyThreads = threads + if (overlap) ioThreads else 0, computeThreads = threads) val suffix = if (b.isAverage) " avg" else "" // Headline: token time and its inverse, so no mental arithmetic to get tok/s. @@ -567,6 +569,10 @@ private fun TelemetryCard(ui: UiState, threads: Int, overlap: Boolean, ioThreads MeterRow("compute$suffix", b.computeMs, b.totalMs, MaterialTheme.colorScheme.primary) MeterRow("flash wait$suffix", b.flashWaitMs, b.totalMs, MaterialTheme.colorScheme.tertiary) MeterRow("cache mgmt$suffix", b.mgmtMs, b.totalMs, MaterialTheme.colorScheme.secondary) + // The fourth bar is the wall time nobody claims: off-CPU cost (zram, preemption, + // frequency caps) the residual `compute_ms` used to absorb silently. Muted color — + // it is a question mark, not a workload. + MeterRow("unattributed$suffix", b.unattributedMs, b.totalMs, MaterialTheme.colorScheme.outline) // Diagnostic line: WHY compute is what it is, plus cache hit. Near 100% busy is // genuinely compute-bound, well below means a throttled/preempted core (a frequency diff --git a/examples/android/app/src/main/java/io/bigmoeonedge/example/MetricFields.kt b/examples/android/app/src/main/java/io/bigmoeonedge/example/MetricFields.kt index 26a9da5..80b2c40 100644 --- a/examples/android/app/src/main/java/io/bigmoeonedge/example/MetricFields.kt +++ b/examples/android/app/src/main/java/io/bigmoeonedge/example/MetricFields.kt @@ -24,7 +24,7 @@ object MetricFields { MetricField("wall_ms", "token time", "the whole time this token took — the one number measured directly; tok/s = 1000/wall_ms", Better.LOWER), MetricField("compute_ms", "compute (a residual!)", "with streaming on, what is left after io/stall and mgmt are subtracted — NOT measured, so it also absorbs faults and scheduler stalls, and is clamped at 0. With streaming off it is just wall_ms", Better.LOWER), MetricField("io_ms", "flash read time", "time reading experts from flash. In serial it is part of wall_ms; under overlap it is per-lane busy time summed, so it can exceed wall_ms", Better.LOWER), - MetricField("stall_ms", "wait on flash", "overlap only: the wall time compute actually sat idle waiting for a read (already divided per thread)", Better.LOWER), + MetricField("stall_ms", "wait on flash", "overlap only: the wall time during which at least one compute thread was blocked on a streamed read — the UNION of stalled intervals, not a per-thread mean, so overlapping waits count once", Better.LOWER), MetricField("mgmt_ms", "cache bookkeeping", "time committing, evicting and reordering the LRU cache this token — plus the periodic dense-residency probe, which the engine folds in here", Better.LOWER), MetricField("read_bytes", "flash read", "bytes of experts pulled from flash THIS token (not cumulative)", Better.LOWER), diff --git a/examples/android/app/src/main/java/io/bigmoeonedge/example/RunService.kt b/examples/android/app/src/main/java/io/bigmoeonedge/example/RunService.kt index 0eb5147..b71aa62 100644 --- a/examples/android/app/src/main/java/io/bigmoeonedge/example/RunService.kt +++ b/examples/android/app/src/main/java/io/bigmoeonedge/example/RunService.kt @@ -249,6 +249,11 @@ class RunService : Service() { val nPast = o.optInt("n_past", -1) val avgComputeMs = o.optDouble("compute_s_tok", -0.001) * 1000.0 val avgMgmtMs = o.optDouble("mgmt_s_tok", -0.001) * 1000.0 + // The measured flash terms for the summary panel — serial reads io, overlap reads + // stall — so end-of-run attribution uses the same measured numbers the live panel + // does instead of inverting the clamped compute residual (issue #98). + val avgIoMs = o.optDouble("io_s_tok", -0.001) * 1000.0 + val avgStallMs = o.optDouble("stall_s_tok", -0.001) * 1000.0 val prefillTps = o.optDouble("prefill_tps", -1.0) val loadS = o.optDouble("load_s", -1.0) val readMib = o.optDouble("read_mib", -1.0) @@ -319,7 +324,7 @@ class RunService : Service() { } val tel = telemetry.current.copy( avgTokensPerSecond = tokS, avgComputeMs = avgComputeMs, - avgMgmtMs = avgMgmtMs, + avgMgmtMs = avgMgmtMs, avgIoMs = avgIoMs, avgStallMs = avgStallMs, prefillTps = prefillTps, ttftS = ttft, readMib = readMib, cacheResidentMib = cacheResidentMib, cacheBudgetMib = cacheBudgetMib, avgMajfltPerTok = majfltPerTok, avgCpuSPerTok = cpuSPerTok, diff --git a/examples/android/app/src/main/java/io/bigmoeonedge/example/Telemetry.kt b/examples/android/app/src/main/java/io/bigmoeonedge/example/Telemetry.kt index 8b47acb..c949e80 100644 --- a/examples/android/app/src/main/java/io/bigmoeonedge/example/Telemetry.kt +++ b/examples/android/app/src/main/java/io/bigmoeonedge/example/Telemetry.kt @@ -14,9 +14,9 @@ data class Telemetry( // the token when compute_ms was clamped to 0. var ioMs: Double = 0.0, var stallMs: Double = 0.0, - // Cache-management time this token — the third wall-additive term. wall = compute + flash-wait + - // mgmt exactly (the engine defines compute as that residual), so the panel can show a breakdown - // that sums to the token time and makes tok/s = 1000/wall self-evident. + // Cache-management time this token — measured, unlike the legacy compute_ms residual. The + // four-way breakdown no longer sums to the wall exactly (unattributed is the honest remainder), + // which is the point: a tidy total was coming from pretending unknown time was compute. var mgmtMs: Double = 0.0, var cacheHitPct: Double = -1.0, // Compute-decomposition of the `computeMs` residual (see docs/telemetry.md). Live per-token @@ -33,10 +33,19 @@ data class Telemetry( // generation finishes. The per-token [tokensPerSecond] is instantaneous (last token only), // so the UI shows this average once it is available. var avgTokensPerSecond: Double = -1.0, - // Per-token AVERAGES over the whole run, from the final summary — shown at the end instead of - // the last token's instantaneous [computeMs]. -1 until generation finishes. + // Per-token AVERAGES over the whole run, from the final summary. Kept for protocol/benchmark + // compatibility — the panel's compute bar deliberately does NOT read this residual anymore + // (see breakdown()); compute comes from [avgCpuSPerTok] and what it cannot explain lands in + // the unattributed term. -1 until generation finishes. var avgComputeMs: Double = -1.0, var avgMgmtMs: Double = -1.0, + // The measured flash terms of the same run averages, from BMOE_DONE's io_s_tok / stall_s_tok: + // serial reads io, overlap reads stall. The summary used to reconstruct flash wait from the + // clamped compute residual instead — against the documented contract — which over-attributed + // to flash exactly when compute_ms had been clamped to 0 (issue #98). -1 until generation + // finishes or on an engine older than the fields. + var avgIoMs: Double = -1.0, + var avgStallMs: Double = -1.0, // End-of-run figures from the final summary (BMOE_DONE); -1 / 0 until generation finishes. var prefillTps: Double = -1.0, // prompt prefill rate (tok/s) var ttftS: Double = -1.0, // time-to-first-token = model load + prompt prefill (s) @@ -83,15 +92,20 @@ data class Telemetry( } /** - * One token's time, split into the three wall-additive terms the panel draws, plus the diagnostics - * that explain the compute term. Derived by [breakdown]; see MetricFields for the same contract as - * the CSV states it. + * One token's time, split into the four terms the panel draws — compute, flash wait, cache mgmt + * and the unattributed remainder — plus the diagnostics that explain them. Derived by [breakdown]; + * see MetricFields for the same contract as the CSV states it. */ data class Breakdown( val wallMs: Double, val computeMs: Double, val flashWaitMs: Double, val mgmtMs: Double, + // Wall time none of the three measured terms explains (compute, flash wait, cache mgmt): + // zram swap-in, preemption, frequency caps — the off-CPU time the old three-bar panel painted + // as compute and read as "the model is thinking". Clamped at 0: measurement noise is allowed + // to make the terms overlap the wall, and the bars are not rescaled to force a tidy total. + val unattributedMs: Double, /** These are run averages, not the last token — the panel labels them "avg". */ val isAverage: Boolean, /** CPU-time ÷ (wall × busy threads), or -1 when the platform couldn't measure it. */ @@ -99,40 +113,50 @@ data class Breakdown( /** Major faults per token, or -1 when unmeasured. */ val faultsPerToken: Double, ) { - /** Denominator for the meter bars: the wall time, or the terms themselves before it is known. */ - val totalMs: Double get() = if (wallMs > 0.0) wallMs else computeMs + flashWaitMs + mgmtMs + /** Denominator for the meter bars: the wall time, or the four terms before it is known. */ + val totalMs: Double get() = + if (wallMs > 0.0) wallMs else computeMs + flashWaitMs + mgmtMs + unattributedMs } /** - * Split a token's wall time into compute / flash-wait / cache-mgmt. + * Split a token's wall time into compute / flash-wait / cache-mgmt / unattributed. * - * While generating this reads the live last token; once the run has a summary it switches to the - * run averages. The two derive the split differently on purpose. Live, flash wait is the MEASURED - * wall-additive read term — stall_ms under overlap (the wall time compute sat idle), io_ms in - * serial (the blocking read) — and compute is the leftover, which keeps the clamp on compute so a - * near-0 compute stays honest instead of being dumped into "flash wait". The end-of-run average - * has no per-mode io/stall to read, so it keeps the residual form. + * Both branches — live (last token) and end-of-run (averages) — now derive the SAME way (issue + * #98; they used to disagree, and the end-of-run one inverted the clamped compute residual, which + * the telemetry contract warns against). Flash wait is always the MEASURED wall-additive read + * term: stall under overlap (the wall time at least one compute thread sat idle on a read), io in + * serial (the blocking read). Compute is process CPU time over the compute threads — a measured + * attribution proxy for matmul work, not the legacy `compute_ms` residual, which by definition + * absorbs everything unmeasured (zram swap-in, preemption, faults); what that residual used to + * hide lands in [Breakdown.unattributedMs] instead. A missing measurement contributes 0 to its + * bar and its time stays unattributed — it is never reconstructed from the wall. * - * [busyThreads] must include the I/O lanes under overlap: the CPU numerator is whole-process, so a - * denominator that counts only compute threads reads occupancy above 100%. + * [busyThreads] (CPU-busy diagnostic) must include the I/O lanes under overlap — the CPU numerator + * is whole-process. [computeThreads] (displayed compute) divides that same numerator down to a + * per-compute-thread figure; the two denominators are different on purpose and must not be + * "simplified" into one. */ -fun breakdown(t: Telemetry, overlap: Boolean, busyThreads: Int): Breakdown { - val useAvg = t.avgTokensPerSecond > 0 && t.avgComputeMs >= 0 - val mgmt = if (useAvg) t.avgMgmtMs.coerceAtLeast(0.0) else t.mgmtMs - val wall = if (useAvg) { - if (t.avgTokensPerSecond > 0) 1000.0 / t.avgTokensPerSecond else 0.0 +fun breakdown(t: Telemetry, overlap: Boolean, busyThreads: Int, computeThreads: Int): Breakdown { + // Summary mode is gated by the summary itself, not by the legacy residual: the bars no longer + // read avgComputeMs at all, so it must not control which branch they take either. + val useAvg = t.avgTokensPerSecond > 0 + val mgmt = (if (useAvg) t.avgMgmtMs else t.mgmtMs).coerceAtLeast(0.0) + val wall = if (useAvg) 1000.0 / t.avgTokensPerSecond else t.wallMs + // Every component is measured or zero. A missing measurement is NOT reconstructed from the + // wall — that is exactly the attribution error #98 exists to fix (the residual silently + // absorbed zram, preemption, faults as "compute"). What is unmeasured stays unattributed. + val flashWait = (if (useAvg) { + if (overlap) t.avgStallMs else t.avgIoMs } else { - t.wallMs - } - val compute: Double - val flashWait: Double - if (useAvg) { - compute = t.avgComputeMs - flashWait = (wall - compute - mgmt).coerceAtLeast(0.0) + if (overlap) t.stallMs else t.ioMs + }).coerceAtLeast(0.0) + val cpuMs = if (useAvg) { + if (t.avgCpuSPerTok >= 0) t.avgCpuSPerTok * 1000.0 else 0.0 } else { - flashWait = if (overlap) t.stallMs else t.ioMs - compute = (wall - flashWait - mgmt).coerceAtLeast(0.0) + t.cpuMs.coerceAtLeast(0.0) } + val compute = if (computeThreads > 0) cpuMs / computeThreads else 0.0 + val unattributed = (wall - compute - flashWait - mgmt).coerceAtLeast(0.0) val useAvgCpu = useAvg && t.avgCpuSPerTok >= 0 val cpuSPerTok = if (useAvgCpu) t.avgCpuSPerTok else t.cpuMs / 1000.0 @@ -146,6 +170,7 @@ fun breakdown(t: Telemetry, overlap: Boolean, busyThreads: Int): Breakdown { computeMs = compute, flashWaitMs = flashWait, mgmtMs = mgmt, + unattributedMs = unattributed, isAverage = useAvg, cpuBusyPct = cpuBusy, faultsPerToken = if (useAvgCpu) t.avgMajfltPerTok else t.majflt, diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 94ec200..d299538 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -42,6 +42,13 @@ add_executable(bmoe_ngram_test ngram_test.cpp) target_link_libraries(bmoe_ngram_test PRIVATE bmoe_core) add_test(NAME ngram_draft COMMAND bmoe_ngram_test) +# The interval-union stall clock. Injected timestamps, no wall clock, so overlap/nesting/ +# separation and the open-interval snapshot rule are verified exactly (issue #98). +add_executable(bmoe_stall_union_test stall_union_test.cpp) +target_link_libraries(bmoe_stall_union_test PRIVATE bmoe_core) +target_include_directories(bmoe_stall_union_test PRIVATE ${CMAKE_SOURCE_DIR}/core/src/moe) +add_test(NAME stall_union COMMAND bmoe_stall_union_test) + find_program(PYTHON3 NAMES python3 python) if(PYTHON3) diff --git a/tests/stall_union_test.cpp b/tests/stall_union_test.cpp new file mode 100644 index 0000000..0bba114 --- /dev/null +++ b/tests/stall_union_test.cpp @@ -0,0 +1,128 @@ +// Unit tests for StallUnionState (core/src/moe/stall_union.h) — the interval-union state machine. +// +// The tests drive the pure state machine with injected timestamps: the production StallUnion is +// this class plus a mutex and an in-critical-section clock read, so what needs exhaustive +// testing is the arithmetic — overlap, nesting, separation, the 0→1/1→0 boundaries and the +// open-interval snapshot rule — all deterministic, with no wall clock and no scheduling luck. +// +// Checks are explicit (not ): the Release build defines NDEBUG, which compiles assert out. + +#include "stall_union.h" + +#include + +using namespace bmoe; + +static int failures = 0; + +static void check(const char * what, long long got, long long want) { + const bool ok = got == want; + if (!ok) ++failures; + std::printf("%-44s got %lld want %lld %s\n", what, got, want, ok ? "ok" : "FAIL"); +} + +int main() { + // Case A — one wait: the interval is its length, whole and nothing else. + { + StallUnionState u; + u.enter_at(10); + check("A: one wait [10,30)", u.total_at(29), 19); // open interval, snapshotted mid-stall + u.exit_at(30); + check("A: one wait [10,30)", u.total_at(100), 20); + } + + // Case B — a wait nested inside another counts once: union 20, not 40. + { + StallUnionState u; + u.enter_at(10); + u.enter_at(15); + u.exit_at(25); + check("B: nested snapshot [10,25)", u.total_at(25), 15); + u.exit_at(30); + check("B: nested waits", u.total_at(100), 20); + } + + // Case C — partial overlap abuts into one interval: [10,30)∪[20,40) = 30. + { + StallUnionState u; + u.enter_at(10); + u.enter_at(20); + u.exit_at(30); + u.exit_at(40); + check("C: partial overlap", u.total_at(100), 30); + } + + // Case D — separated intervals both count: [10,20) + [30,50) = 30. + { + StallUnionState u; + u.enter_at(10); + u.exit_at(20); + u.enter_at(30); + u.exit_at(50); + check("D: separated intervals", u.total_at(100), 30); + } + + // Case E — three waits, two nested in the first: [10,50)∪[20,30)∪[25,45) = 40. + { + StallUnionState u; + u.enter_at(10); + u.enter_at(20); + u.enter_at(25); + u.exit_at(30); + u.exit_at(45); + check("E: snapshot, one still open", u.total_at(45), 35); + u.exit_at(50); + check("E: triple, nested", u.total_at(100), 40); + } + + // Case F — the boundaries: a second enter while the interval is open must not restart it, and + // only the 1->0 exit may close it. Interleave a third thread to exercise 2->3->2. + { + StallUnionState u; + u.enter_at(100); // 0 -> 1: opens + u.enter_at(110); // 1 -> 2: no restart + u.enter_at(120); // 2 -> 3 + u.exit_at(130); // 3 -> 2 + check("F: mid-flight snapshot", u.total_at(130), 30); + u.exit_at(999); // 2 -> 1: must NOT close — a bogus close would stop at 899 here + check("F: still open after 2->1", u.total_at(999), 899); + u.exit_at(200); // 1 -> 0: closes at 200 + check("F: closed at 1->0", u.total_at(1000), 100); + } + + // The stats() rule: a snapshot taken while stalled includes the open interval up to now, so a + // per-token delta spanning it cannot lose the part already elapsed. + { + StallUnionState u; + u.enter_at(1000); + const long long first = u.total_at(1500); // 500 elapsed + u.exit_at(2000); + const long long second = u.total_at(2000) - first; + check("snapshot deltas sum to the union", first + second, 1000); + } + + // reset() zeroes everything, interval included. + { + StallUnionState u; + u.enter_at(10); + u.exit_at(30); + u.enter_at(40); + u.reset(); + check("reset clears total", u.total_at(100), 0); + u.exit_at(50); // stray exit on an empty state: refused, count not driven negative + check("exit on empty is inert", u.total_at(100), 0); + // ...and the refusal is what keeps the NEXT interval honest: without the guard the count + // sits at -1, the following enter reads as a "close" of a garbage interval, and every + // number after it is wrong. + u.enter_at(1000); + u.exit_at(1050); + check("interval after stray exit is correct", u.total_at(2000), 50); + } + + if (failures) { + std::printf("stall_union_test: %d FAILURE(S)\n", failures); + return 1; + } + std::printf("stall_union_test: all ok\n"); + return 0; +}