vrr/BigMoeOnEdge
2
0
Fork 0
mirror of https://github.com/Helldez/BigMoeOnEdge.git synced 2026-10-03 03:25:42 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

feat(metrics): --compute-trace and --io-trace decompose the decode

Browse source 
The per-token CSV reports compute as a residual (wall - io - mgmt), so everything the
engine does not itself clock is pooled into it: page faults, scheduler stalls, and the
matmuls. A residual cannot say which. That is the whole reason gpt-oss-120b reads as
"compute-bound" at 1.7 s/token while faulting 8.4k pages per token — the flash wait is
billed to compute because it happens under llama_decode.

--compute-trace measures it instead. Asking the eval callback to isolate a node makes
ggml compute exactly up to it and synchronize, so the wall delta between consecutive
boundaries is that node's real compute time; sampling major faults across the same
boundaries attributes the >RAM stall to the node that paid it. Still no llama.cpp patch:
this rides the public cb_eval ABI, whose ask/no-ask contract already specifies the
isolation. It costs a barrier per node and forbids operator coalescing, so it is a
diagnostic — a traced run is not a benchmark run, and only the shares are meaningful.
Unlike the other traces it does not need --moe-stream: it times the graph, so a dense
mmap baseline can be traced and compared against a streamed run.

--io-trace records one row per pread: latency, requested vs aligned bytes, lane, and the
(layer, expert, projection) it serves — values already computed at every enqueue site and
until now discarded. This is where the flash floor is: the aggregate 760 MiB/s sits far
below the drive's sequential ceiling because routed slices are scattered, and the trace
says whether that is per-read latency, request size, or lanes idling. It also measures
the adjacency the roadmap's read-coalescing and expert-contiguous-layout items assume.

Node classification stays out of the engine: which node is attention vs dense FFN vs
expert matmul is naming policy that varies by architecture, so the rows carry the raw op
and name and scripts/decode-analyze.py classifies. Verified on both gate models that the
generated text, cache hit rate and bytes read are identical with the traces on.
This commit is contained in:
Helldez 2026-07-15 11:11:22 +02:00
parent 5db3399897
commit f9e408f542
14 changed files with 764 additions and 39 deletions
Download patch file Download diff file Expand all files Collapse all files

33
cli/main.cpp
View file

@ -13,6 +13,7 @@
#include "bmoe/recipe.h"
#include "bmoe/metrics.h"
#include "bmoe/route_trace.h"
#include "bmoe/decode_trace.h"
#include <atomic>
#include <condition_variable>
@ -350,6 +351,8 @@ int main(int argc, char ** argv) {
RunConfig cfg;
std::string csv_path;
std::string route_trace_path;
std::string compute_trace_path;
std::string io_trace_path;
bool session_mode = false;
for (int i = 1; i < argc; ++i) {
@ -385,6 +388,10 @@ int main(int argc, char ** argv) {
csv_path = next("--csv");
else if (a == "--route-trace")
route_trace_path = next("--route-trace");
else if (a == "--compute-trace")
compute_trace_path = next("--compute-trace");
else if (a == "--io-trace")
io_trace_path = next("--io-trace");
else if (a == "--moe-stream")
cfg.moe.enabled = true;
else if (a == "--cache-mb") {
@ -466,6 +473,25 @@ int main(int argc, char ** argv) {
}
}
// The compute trace works without streaming — timing the graph is what a dense mmap baseline
// needs too — so unlike the other two it carries no --moe-stream requirement.
std::unique_ptr<IComputeTraceSink> compute_trace;
if (!compute_trace_path.empty()) {
compute_trace.reset(make_csv_compute_trace_sink(compute_trace_path));
if (!compute_trace)
std::fprintf(stderr, "warning: could not open compute trace %s\n", compute_trace_path.c_str());
}
std::unique_ptr<IIoTraceSink> io_trace;
if (!io_trace_path.empty()) {
if (!cfg.moe.enabled) {
std::fprintf(stderr, "warning: --io-trace needs --moe-stream; no trace will be written\n");
} else {
io_trace.reset(make_csv_io_trace_sink(io_trace_path));
if (!io_trace) std::fprintf(stderr, "warning: could not open io trace %s\n", io_trace_path.c_str());
}
}
// Interactive session: one persistent process serves many prompts over stdin, keeping the
// model loaded and the expert cache warm between them. Prompts arrive as JSON requests, not
// via -p. This is a superset of --progress output (BMOE_* lines), so it never streams inline.
@ -493,7 +519,7 @@ int main(int argc, char ** argv) {
}
};
RunResult r = run(cfg, on_token, sink.get(), route_trace.get());
RunResult r = run(cfg, on_token, sink.get(), route_trace.get(), compute_trace.get(), io_trace.get());
if (!r) {
std::fprintf(stderr, "\nerror: %s\n", r.error.c_str());
return 1;
@ -511,9 +537,8 @@ int main(int argc, char ** argv) {
// occupancy = CPU-time ÷ (wall × threads): ~1 is compute-bound, well under 1 is a throttled or
// preempted core; major faults/token > 0 means dense weights re-faulted from flash inside decode.
if (s.cpu_s_per_token > 0.0 || s.majflt_per_token > 0.0) {
const double occ = s.s_per_token > 0 && cfg.n_threads > 0
? s.cpu_s_per_token / (s.s_per_token * cfg.n_threads)
: 0.0;
const double occ =
s.s_per_token > 0 && cfg.n_threads > 0 ? s.cpu_s_per_token / (s.s_per_token * cfg.n_threads) : 0.0;
std::printf("compute: %.1f%% CPU occupancy (%.4f cpu-s/token over %d threads), %.2f major faults/token\n",
occ * 100.0, s.cpu_s_per_token, cfg.n_threads, s.majflt_per_token);
}

1
core/CMakeLists.txt
View file

@ -25,6 +25,7 @@ if(BMOE_HAVE_LLAMA)
src/engine/runtime.cpp
src/metrics/csv_metrics_sink.cpp
src/metrics/route_trace_sink.cpp
src/metrics/decode_trace_sink.cpp
)
# `llama` is the stable public API the streaming seam uses. `llama-common` (llama.cpp's
# `common` utils) is linked only for chat-template rendering + reasoning parsing

96
core/include/bmoe/decode_trace.h Normal file
View file

@ -0,0 +1,96 @@
// Decode traces: what a token's time is actually made of.
//
// The per-token metrics answer *how long*; the route trace answers *what the router asked for*.
// These two answer *where the time went*, and they exist because the headline number they
// decompose is not measured at all: `compute_ms` is a residual (wall − io − mgmt), so every cost
// the engine does not itself clock — page faults, scheduler stalls, the matmuls themselves — is
// silently pooled into it. A residual cannot tell you which of those it is.
//
// Compute trace — the eval callback, asked to isolate nodes, yields real per-node wall time:
// ggml computes exactly up to an isolated node, synchronizes, then calls back, so the delta
// between consecutive boundaries is that node's compute. Sampling major faults across the same
// boundaries attributes the >RAM residency stall to the node that paid it, which is the whole
// point: on a >RAM model most of "compute" is faults, and no residual can show that.
//
// I/O trace — one row per flash read: latency, size, alignment waste and the (layer, expert,
// projection) it served. The aggregate read bandwidth is far below the drive's sequential
// ceiling because routed slices are scattered; this says by how much, and whether the cause is
// per-read latency, request size, or lanes idling.
//
// Both are diagnostics, not telemetry, and both perturb what they measure — isolating nodes
// forbids ggml the operator coalescing it would otherwise do, and the I/O rows take a lock on the
// read path. A traced run is NOT a benchmark run: read the proportions, not the absolutes.
#pragma once
#include <cstdint>
#include <string>
#include <vector>
namespace bmoe {
// One isolated graph node's compute. Emitted only while the compute trace is on.
struct ComputeTraceRow {
int turn = 0; // session-mode turn (0 for a one-shot run)
int phase = 0; // 0 = prefill, 1 = decode
int step = 0; // absolute context position of the token being computed
int seq = 0; // node's position in the graph this decode (0-based), i.e. execution order
int layer = -1;
// ggml's op name (ggml_op_name) and the node's own name. Deliberately raw: which node belongs
// to attention vs the dense FFN vs the expert matmul is naming policy that varies by
// architecture, so the engine reports what the graph said and the analysis script classifies.
std::string op;
std::string name;
uint64_t wall_ns = 0; // time to compute THIS node (delta between isolation boundaries)
uint64_t majflt = 0; // major page faults charged to this node — flash re-reads inside compute
};
// One flash read. Emitted only while the I/O trace is on.
struct IoTraceRow {
int turn = 0;
int phase = 0;
int step = 0;
int layer = -1;
int32_t expert = -1;
int8_t proj = -1; // projection slot within the layer (recipe order)
int8_t lane = -1; // which read lane served it
uint8_t spec = 0; // 1 if issued speculatively by prefetch
uint64_t offset = 0; // absolute file offset requested
uint64_t req_bytes = 0; // bytes the caller wanted
uint64_t read_bytes = 0; // bytes actually read (aligned window; ≥ req_bytes with O_DIRECT)
uint64_t latency_ns = 0; // wall time in the pread loop
};
// Run-level facts, emitted once before any row.
struct DecodeTraceStatic {
std::string model;
std::string arch;
int n_layer = 0;
int n_threads = 0;
int io_threads = 0;
bool o_direct = false;
bool overlap = false;
};
// Optional sinks. The engine calls on_static once at open, then on_rows once per decode with that
// decode's rows. Rows are buffered in RAM while the graph runs — the callback and the read path
// must not do I/O — and drained after llama_decode returns.
class IComputeTraceSink {
public:
virtual ~IComputeTraceSink() = default;
virtual void on_static(const DecodeTraceStatic &) = 0;
virtual void on_rows(const ComputeTraceRow * rows, size_t n) = 0;
};
class IIoTraceSink {
public:
virtual ~IIoTraceSink() = default;
virtual void on_static(const DecodeTraceStatic &) = 0;
virtual void on_rows(const IoTraceRow * rows, size_t n) = 0;
};
// Sinks writing `path` as long-format CSV: a `#` preamble carrying the static block, then one row
// per node / per read. Return nullptr if the file cannot be opened.
IComputeTraceSink * make_csv_compute_trace_sink(const std::string & path);
IIoTraceSink * make_csv_io_trace_sink(const std::string & path);
} // namespace bmoe

10
core/include/bmoe/runtime.h
View file

@ -17,6 +17,8 @@
namespace bmoe {
class IRouteTraceSink;
class IComputeTraceSink;
class IIoTraceSink;
struct RunResult {
bool ok = false;
@ -29,11 +31,13 @@ struct RunResult {
// Run one generation. `on_token` (nullable) is invoked once per generated token before
// the next decode; `sink` (nullable) receives the same per-token metrics plus the final
// summary. `route_trace` (nullable) records the per-step, per-layer routing trace — a
// diagnostic, see bmoe/route_trace.h. Blocks until generation completes or errors.
// summary. The trace sinks (all nullable) are diagnostics that perturb what they measure — see
// bmoe/route_trace.h and bmoe/decode_trace.h. Blocks until generation completes or errors.
RunResult run(const RunConfig & cfg,
const std::function<void(const TokenMetrics &)> & on_token = nullptr,
IMetricsSink * sink = nullptr,
IRouteTraceSink * route_trace = nullptr);
IRouteTraceSink * route_trace = nullptr,
IComputeTraceSink * compute_trace = nullptr,
IIoTraceSink * io_trace = nullptr);
} // namespace bmoe

20
core/include/bmoe/session.h
View file

@ -25,6 +25,8 @@
namespace bmoe {
class IRouteTraceSink;
class IComputeTraceSink;
class IIoTraceSink;
// Everything fixed for the model's lifetime — set once at open(). n_ctx and n_batch are
// baked into the llama context at creation and cannot change per prompt, so size them for
@ -58,11 +60,19 @@ public:
// Returns nullptr and sets `error` on failure. The returned session owns all native
// state and must outlive every generate() call.
//
// `route_trace` (nullable) turns on the per-step, per-layer routing trace for every
// generate() on this session and must outlive it — a diagnostic, ignored when streaming is
// off, and never on for a benchmark run. See bmoe/route_trace.h.
static std::unique_ptr<Session>
open(const SessionConfig & cfg, std::string & error, IRouteTraceSink * route_trace = nullptr);
// The trace sinks (all nullable) turn on their diagnostic for every generate() on this session
// and must outlive it. None is ever on for a benchmark run — each perturbs what it measures.
// * route_trace — per-step, per-layer routing; ignored when streaming is off (no routing to
// observe). See bmoe/route_trace.h.
// * compute_trace — per-node compute and fault attribution. Works with or without streaming,
// so a dense mmap baseline can be compared against a streamed run.
// * io_trace — per-read flash latency/size; ignored when streaming is off (no reads).
// See bmoe/decode_trace.h for the latter two.
static std::unique_ptr<Session> open(const SessionConfig & cfg,
std::string & error,
IRouteTraceSink * route_trace = nullptr,
IComputeTraceSink * compute_trace = nullptr,
IIoTraceSink * io_trace = nullptr);
// Generate one response. Serialized: one generation at a time per session. `on_token`
// and `sink` receive the same per-token metrics as run(). Cache state carries over from

6
core/src/engine/runtime.cpp
View file

@ -13,7 +13,9 @@ namespace bmoe {
RunResult run(const RunConfig & cfg,
const std::function<void(const TokenMetrics &)> & on_token,
IMetricsSink * sink,
IRouteTraceSink * route_trace) {
IRouteTraceSink * route_trace,
IComputeTraceSink * compute_trace,
IIoTraceSink * io_trace) {
ValidationResult v = validate(cfg);
if (!v) {
RunResult r;
@ -31,7 +33,7 @@ RunResult run(const RunConfig & cfg,
sc.moe = cfg.moe;
std::string error;
std::unique_ptr<Session> session = Session::open(sc, error, route_trace);
std::unique_ptr<Session> session = Session::open(sc, error, route_trace, compute_trace, io_trace);
if (!session) {
RunResult r;
r.error = error;

79
core/src/engine/session.cpp
View file

@ -1,6 +1,7 @@
#include "bmoe/session.h"
#include "bmoe/recipe.h"
#include "bmoe/route_trace.h"
#include "bmoe/decode_trace.h"
#include "../moe/router_hook.h"
#include "../moe/expert_stream_source.h"
#include "../moe/gguf_offsets.h"
@ -106,6 +107,13 @@ struct Session::Impl {
IRouteTraceSink * route_trace = nullptr;
int trace_turn = 0;
// Decode traces (diagnostics): null unless requested. The compute trace needs no streaming —
// it measures the graph, which a dense mmap run has too; the I/O trace needs the streamer,
// since there are no engine-issued reads without it. See bmoe/decode_trace.h.
IComputeTraceSink * compute_trace = nullptr;
IIoTraceSink * io_trace = nullptr;
std::vector<IoTraceRow> io_rows_scratch;
std::atomic<bool> cancel_requested{false};
~Impl() {
@ -139,7 +147,11 @@ void Session::cancel() {
impl_->cancel_requested.store(true, std::memory_order_relaxed);
}
std::unique_ptr<Session> Session::open(const SessionConfig & cfg, std::string & error, IRouteTraceSink * route_trace) {
std::unique_ptr<Session> Session::open(const SessionConfig & cfg,
std::string & error,
IRouteTraceSink * route_trace,
IComputeTraceSink * compute_trace,
IIoTraceSink * io_trace) {
auto fail = [&](std::string msg) -> std::unique_ptr<Session> {
error = std::move(msg);
return nullptr;
@ -226,7 +238,10 @@ std::unique_ptr<Session> Session::open(const SessionConfig & cfg, std::string &
cparams.n_ctx = cfg.n_ctx;
cparams.n_batch = cfg.n_batch;
cparams.n_ubatch = cfg.n_batch;
if (cfg.moe.enabled) {
// The streamer needs the callback to see routing; the compute trace needs it to time nodes.
// Installing it for the trace alone is what lets a NON-streamed run be measured — the dense
// mmap baseline the streamed numbers are argued against.
if (cfg.moe.enabled || compute_trace) {
cparams.cb_eval = &RouterHook::c_eval;
cparams.cb_eval_user_data = im.hook.get();
}
@ -317,6 +332,11 @@ std::unique_ptr<Session> Session::open(const SessionConfig & cfg, std::string &
route_trace->on_static(st);
}
if (io_trace) {
im.io_trace = io_trace;
im.source.set_io_trace(true);
}
if (cfg.moe.overlap) {
#ifdef BMOE_HAVE_EXPERT_READY_HOOK
im.source.enable_overlap_hook();
@ -328,6 +348,26 @@ std::unique_ptr<Session> Session::open(const SessionConfig & cfg, std::string &
llama_memory_clear(llama_get_memory(ctx), true); // discard warm-up KV
}
// Decode traces. Outside the streaming block on purpose: the compute trace measures the graph,
// which exists with or without the streamer, so a dense mmap baseline can be traced and
// compared. The I/O trace was armed above (it needs the source) and only reports here.
if (compute_trace || im.io_trace) {
DecodeTraceStatic st;
st.model = cfg.model_path;
st.arch = im.arch;
st.n_layer = im.n_layer;
st.n_threads = cfg.n_threads;
st.io_threads = cfg.moe.enabled ? cfg.moe.io_threads : 0;
st.o_direct = cfg.moe.enabled && cfg.moe.o_direct;
st.overlap = cfg.moe.enabled && cfg.moe.overlap;
if (compute_trace) {
im.compute_trace = compute_trace;
im.hook->set_compute_trace(true);
compute_trace->on_static(st);
}
if (im.io_trace) im.io_trace->on_static(st);
}
im.load_seconds = secs(t_load0, clock_t_::now());
return self;
}
@ -479,15 +519,40 @@ RunResult Session::generate(const GenerateRequest & req,
// Route trace: frame one decode's rows, then hand them to the sink once it has returned —
// never from inside the callback, which runs on a compute thread mid-graph. `base_pos` is
// the context position of the batch's first token, so prefill rows carry real step numbers.
// The frame the I/O rows are stamped with at flush; the other traces carry their own.
int trace_phase = 0, trace_step = 0;
auto trace_begin = [&](int base_pos, int n_tokens, int phase) {
if (im.route_trace) im.hook->begin_trace_batch(base_pos, n_tokens, phase, im.trace_turn);
// A node is computed once for the whole batch, not per token, so a prefill chunk's graph is
// attributed to its last position rather than pretending to split across the chunk.
if (im.compute_trace) im.hook->begin_compute_batch(base_pos + n_tokens - 1, phase, im.trace_turn);
trace_phase = phase;
trace_step = base_pos + n_tokens - 1;
};
auto trace_flush = [&]() {
if (!im.route_trace) return;
im.hook->end_trace_batch();
std::vector<RouteTraceRow> & rows = im.hook->trace_rows();
if (!rows.empty()) im.route_trace->on_rows(rows.data(), rows.size());
rows.clear();
if (im.route_trace) {
im.hook->end_trace_batch();
std::vector<RouteTraceRow> & rows = im.hook->trace_rows();
if (!rows.empty()) im.route_trace->on_rows(rows.data(), rows.size());
rows.clear();
}
if (im.compute_trace) {
std::vector<ComputeTraceRow> & rows = im.hook->compute_rows();
if (!rows.empty()) im.compute_trace->on_rows(rows.data(), rows.size());
rows.clear();
}
if (im.io_trace) {
// The reads carry no frame of their own — a lane does not know which token it serves —
// so stamp them with the decode they were drained after.
im.source.take_io_trace_rows(im.io_rows_scratch);
for (IoTraceRow & r : im.io_rows_scratch) {
r.turn = im.trace_turn;
r.phase = trace_phase;
r.step = trace_step;
}
if (!im.io_rows_scratch.empty()) im.io_trace->on_rows(im.io_rows_scratch.data(), im.io_rows_scratch.size());
im.io_rows_scratch.clear();
}
};
// ── prefill (chunked by n_batch; positions auto-continue from the reused prefix) ──

101
core/src/metrics/decode_trace_sink.cpp Normal file
View file

@ -0,0 +1,101 @@
#include "bmoe/decode_trace.h"
#include <cstdio>
namespace bmoe {
namespace {
// The static preamble both traces share: the facts a row cannot carry, so a trace file stays
// analysable without the run that produced it (same spirit as the metrics CSV's `# summary`).
void write_static(std::FILE * f, const char * kind, const DecodeTraceStatic & s) {
std::fprintf(f, "# %s v1\n", kind);
std::fprintf(f, "# model=%s arch=%s n_layer=%d n_threads=%d io_threads=%d o_direct=%d overlap=%d\n",
s.model.c_str(), s.arch.c_str(), s.n_layer, s.n_threads, s.io_threads, (int) s.o_direct,
(int) s.overlap);
}
// A node name can carry anything ggml put there; commas and quotes would break the column count.
void write_csv_field(std::FILE * f, const std::string & v) {
if (v.find_first_of(",\"\n") == std::string::npos) {
std::fputs(v.c_str(), f);
return;
}
std::fputc('"', f);
for (char c : v) {
if (c == '"') std::fputc('"', f); // RFC4180 doubling
std::fputc(c == '\n' ? ' ' : c, f);
}
std::fputc('"', f);
}
class CsvComputeTraceSink final : public IComputeTraceSink {
public:
explicit CsvComputeTraceSink(std::FILE * f) : f_(f) {}
~CsvComputeTraceSink() override {
if (f_) std::fclose(f_);
}
void on_static(const DecodeTraceStatic & s) override {
write_static(f_, "compute_trace", s);
std::fprintf(f_, "turn,phase,step,seq,layer,op,name,wall_ns,majflt\n");
std::fflush(f_);
}
void on_rows(const ComputeTraceRow * rows, size_t n) override {
for (size_t i = 0; i < n; ++i) {
const ComputeTraceRow & r = rows[i];
std::fprintf(f_, "%d,%d,%d,%d,%d,", r.turn, r.phase, r.step, r.seq, r.layer);
write_csv_field(f_, r.op);
std::fputc(',', f_);
write_csv_field(f_, r.name);
std::fprintf(f_, ",%llu,%llu\n", (unsigned long long) r.wall_ns, (unsigned long long) r.majflt);
}
std::fflush(f_); // once per decode, not per row
}
private:
std::FILE * f_ = nullptr;
};
class CsvIoTraceSink final : public IIoTraceSink {
public:
explicit CsvIoTraceSink(std::FILE * f) : f_(f) {}
~CsvIoTraceSink() override {
if (f_) std::fclose(f_);
}
void on_static(const DecodeTraceStatic & s) override {
write_static(f_, "io_trace", s);
std::fprintf(f_, "turn,phase,step,layer,expert,proj,lane,spec,offset,req_bytes,read_bytes,latency_ns\n");
std::fflush(f_);
}
void on_rows(const IoTraceRow * rows, size_t n) override {
for (size_t i = 0; i < n; ++i) {
const IoTraceRow & r = rows[i];
std::fprintf(f_, "%d,%d,%d,%d,%d,%d,%d,%u,%llu,%llu,%llu,%llu\n", r.turn, r.phase, r.step, r.layer,
(int) r.expert, (int) r.proj, (int) r.lane, (unsigned) r.spec, (unsigned long long) r.offset,
(unsigned long long) r.req_bytes, (unsigned long long) r.read_bytes,
(unsigned long long) r.latency_ns);
}
std::fflush(f_);
}
private:
std::FILE * f_ = nullptr;
};
} // namespace
IComputeTraceSink * make_csv_compute_trace_sink(const std::string & path) {
std::FILE * f = std::fopen(path.c_str(), "wb");
return f ? new CsvComputeTraceSink(f) : nullptr;
}
IIoTraceSink * make_csv_io_trace_sink(const std::string & path) {
std::FILE * f = std::fopen(path.c_str(), "wb");
return f ? new CsvIoTraceSink(f) : nullptr;
}
} // namespace bmoe

62
core/src/moe/expert_stream_source.cpp
View file

@ -317,7 +317,9 @@ void ExpertStreamSource::warm_dense_regions(const std::string & gguf_path) {
}
// ── one aligned slice read on a lane ────────────────────────────────────────────────
bool ExpertStreamSource::read_slice(int lane, void * dst, uint64_t off, uint64_t nbytes) {
bool ExpertStreamSource::read_slice(int lane, const IoJob & j) {
void * const dst = j.dst;
const uint64_t off = j.off, nbytes = j.nbytes;
if (nbytes == 0) return true;
const uint64_t a0 = off & ~(uint64_t) (align_ - 1);
const uint64_t a1 = (off + nbytes + align_ - 1) & ~(uint64_t) (align_ - 1);
@ -355,12 +357,43 @@ bool ExpertStreamSource::read_slice(int lane, void * dst, uint64_t off, uint64_t
a += (uint64_t) got;
}
const auto t1 = clock_t_::now();
io_syscall_ns_.fetch_add((long long) std::chrono::duration_cast<std::chrono::nanoseconds>(t1 - t0).count());
const uint64_t lat_ns = (uint64_t) std::chrono::duration_cast<std::chrono::nanoseconds>(t1 - t0).count();
io_syscall_ns_.fetch_add((long long) lat_ns);
std::memcpy(dst, b + (off - a0), (size_t) nbytes);
read_bytes_.fetch_add((long long) (read_end - a0));
// The trace records the read as issued, not as accounted: `read_bytes` is the aligned window
// actually pulled, which is what the drive was asked for and what the effective bandwidth must
// be judged against — `req_bytes` is only what the caller wanted out of it.
if (io_trace_on_) {
IoTraceRow r;
r.layer = j.layer;
r.expert = j.expert;
r.proj = (int8_t) j.proj;
r.lane = (int8_t) lane;
r.spec = j.spec;
r.offset = off;
r.req_bytes = nbytes;
r.read_bytes = read_end - a0;
r.latency_ns = lat_ns;
std::lock_guard<std::mutex> lk(io_trace_mtx_);
io_trace_rows_.push_back(r);
}
return true;
}
void ExpertStreamSource::set_io_trace(bool on) {
std::lock_guard<std::mutex> lk(io_trace_mtx_);
io_trace_on_ = on;
io_trace_rows_.clear();
}
void ExpertStreamSource::take_io_trace_rows(std::vector<IoTraceRow> & out) {
std::lock_guard<std::mutex> lk(io_trace_mtx_);
out.swap(io_trace_rows_);
io_trace_rows_.clear();
}
void ExpertStreamSource::io_drain(int lane, uint64_t my_gen) {
for (;;) {
size_t i;
@ -370,7 +403,7 @@ void ExpertStreamSource::io_drain(int lane, uint64_t my_gen) {
i = next_idx_++;
}
const IoJob & j = jobs_[i];
if (!read_slice(lane, j.dst, j.off, j.nbytes)) {
if (!read_slice(lane, j)) {
io_err_.store(true);
if (overlap_) fatal_.store(true, std::memory_order_release);
}
@ -432,7 +465,8 @@ void ExpertStreamSource::prefetch(int il, const int32_t * ids, int n_ids) {
uintptr_t a0 = (uintptr_t) dst & ~(uintptr_t) (page_ - 1);
uintptr_t a1 = ((uintptr_t) dst + slice + page_ - 1) & ~(uintptr_t) (page_ - 1);
if (!pio::vm_commit((void *) a0, (size_t) (a1 - a0))) return; // low on memory — stop quietly
spec_jobs_.push_back({dst, L.proj[p].file_off + (uint64_t) e * slice, slice, id});
spec_jobs_.push_back(
{dst, L.proj[p].file_off + (uint64_t) e * slice, slice, id, e, (int16_t) il, (int8_t) p, 1});
++njobs;
}
if (njobs == 0) continue;
@ -451,7 +485,7 @@ void ExpertStreamSource::prefetch(int il, const int32_t * ids, int n_ids) {
g = spec_gen_;
j = spec_jobs_[spec_next_++];
++spec_inflight_;
const bool ok = read_slice(0, j.dst, j.off, j.nbytes);
const bool ok = read_slice(0, j);
if (ok && g == spec_gen_) {
spec_read_bytes_.fetch_add((long long) j.nbytes);
if (--spec_remaining_[j.flag] == 0) spec_done_.push_back(j.flag);
@ -478,7 +512,7 @@ void ExpertStreamSource::drain_spec(int lane, uint64_t worker_seen) {
j = spec_jobs_[spec_next_++];
++spec_inflight_;
}
const bool ok = read_slice(lane, j.dst, j.off, j.nbytes);
const bool ok = read_slice(lane, j);
{
std::lock_guard<std::mutex> lk(io_mtx_);
if (ok && g == spec_gen_) { // ignore reads from a cancelled round
@ -681,8 +715,8 @@ bool ExpertStreamSource::load_layer(int il, const int32_t * ids, int n_ids) {
for (int p = 0; p < MoeRecipe::max_exps; ++p) {
const uint64_t slice = L.proj[p].nb2;
if (slice == 0) continue; // absent slot in a fused layout
jobs_.push_back(
{(char *) slot_[p] + (uint64_t) e * slice, L.proj[p].file_off + (uint64_t) e * slice, slice});
jobs_.push_back({(char *) slot_[p] + (uint64_t) e * slice, L.proj[p].file_off + (uint64_t) e * slice,
slice, -1, e, (int16_t) il, (int8_t) p, 0});
}
return true;
}
@ -709,7 +743,8 @@ bool ExpertStreamSource::load_layer(int il, const int32_t * ids, int n_ids) {
std::fprintf(stderr, "bmoe: commit failed\n");
return false;
}
jobs_.push_back({dst, L.proj[p].file_off + (uint64_t) e * slice, slice});
jobs_.push_back(
{dst, L.proj[p].file_off + (uint64_t) e * slice, slice, -1, e, (int16_t) il, (int8_t) p, 0});
}
cvalid_[id] = 1;
cspec_[id] = 0; // a real read, not speculative
@ -752,7 +787,7 @@ bool ExpertStreamSource::load_layer(int il, const int32_t * ids, int n_ids) {
if (io_threads_ <= 1 || njobs <= 1) {
for (size_t i = 0; i < njobs; ++i) {
const IoJob & j = jobs_[i];
if (!read_slice(0, j.dst, j.off, j.nbytes)) return false;
if (!read_slice(0, j)) return false;
}
} else {
uint64_t my_gen;
@ -852,8 +887,8 @@ bool ExpertStreamSource::load_layer_async(int il, const int32_t * ids, int n_ids
if (slice == 0) continue; // absent slot in a fused layout
for (int e : staged_) {
const int32_t flag = (int32_t) ((size_t) p * (size_t) n_expert_ + (size_t) e);
jobs_.push_back(
{(char *) slot_[p] + (uint64_t) e * slice, L.proj[p].file_off + (uint64_t) e * slice, slice, flag});
jobs_.push_back({(char *) slot_[p] + (uint64_t) e * slice, L.proj[p].file_off + (uint64_t) e * slice,
slice, flag, e, (int16_t) il, (int8_t) p, 0});
}
}
} else {
@ -901,7 +936,8 @@ bool ExpertStreamSource::load_layer_async(int il, const int32_t * ids, int n_ids
if (!seen_[e]) continue; // cache hit, already marked ready
const int32_t flag = (int32_t) ((size_t) p * (size_t) n_expert_ + (size_t) e);
jobs_.push_back({(char *) lbuf_[p][il] + (uint64_t) e * slice,
L.proj[p].file_off + (uint64_t) e * slice, slice, flag});
L.proj[p].file_off + (uint64_t) e * slice, slice, flag, e, (int16_t) il, (int8_t) p,
0});
}
}

24
core/src/moe/expert_stream_source.h
View file

@ -16,6 +16,7 @@
#include "bmoe/expert_source.h"
#include "bmoe/config.h"
#include "bmoe/decode_trace.h"
#include "bmoe/recipe.h"
#include "../io/platform_io.h"
@ -85,6 +86,13 @@ public:
// and exercised by the shrink gate. Clamped up: an explicit raise also lifts the grow ceiling.
void set_cache_budget(size_t bytes);
// ── I/O trace (diagnostics; see bmoe/decode_trace.h) ────────────────────────────
// When on, every read_slice records one row. Rows are appended under a dedicated leaf mutex
// (reads happen on N lanes at once), so this costs a lock per read and is off by default.
// take_io_trace_rows moves the buffer out; the caller stamps the frame it belongs to.
void set_io_trace(bool on);
void take_io_trace_rows(std::vector<IoTraceRow> & out);
void shutdown();
private:
@ -93,6 +101,13 @@ private:
uint64_t off = 0;
uint64_t nbytes = 0;
int32_t flag = -1; // overlap: index into ready_ to publish on completion; -1 = serial
// Which (layer, expert, projection) this read serves. Known at every enqueue site and
// otherwise thrown away; carried so the I/O trace can attribute a read without the read
// path having to guess. Inert unless the trace is on.
int32_t expert = -1;
int16_t layer = -1;
int8_t proj = -1;
uint8_t spec = 0; // 1 if enqueued speculatively by prefetch
};
// One readiness cell per (projection, expert). A cell is "ready for the layer in flight"
@ -102,7 +117,8 @@ private:
std::atomic<uint32_t> gen{0};
};
bool read_slice(int lane, void * dst, uint64_t off, uint64_t nbytes);
// `j` carries the read AND (for the trace) what it serves; `lane` is who is doing it.
bool read_slice(int lane, const IoJob & j);
void io_drain(int lane, uint64_t my_gen);
void io_worker(int lane);
@ -135,6 +151,12 @@ private:
size_t entry_bytes(int il) const;
void evict_tail();
// I/O trace buffer. Its own leaf mutex, never held across a read: the lanes append
// concurrently, and the eval thread swaps the buffer out between decodes.
bool io_trace_on_ = false;
std::mutex io_trace_mtx_;
std::vector<IoTraceRow> io_trace_rows_;
bool active_ = false;
bool o_direct_ = false;
bool load_all_ = false;

53
core/src/moe/router_hook.cpp
View file

@ -1,6 +1,7 @@
#include "router_hook.h"
#include "ggml.h"
#include "../io/platform_io.h"
#include <cstdio>
#include <cstring>
@ -153,7 +154,56 @@ bool RouterHook::c_eval(ggml_tensor * t, bool ask, void * user_data) {
return static_cast<RouterHook *>(user_data)->on_eval(t, ask);
}
// Layer id from a node name. llama.cpp suffixes per-layer nodes with "-<il>"; anything else
// (embeddings, the output head, the KQ mask) belongs to no layer and reports -1. Kept generic on
// purpose: the trace must not carry a table of which node names a given architecture emits.
static int node_layer(const char * name) {
const char * dash = std::strrchr(name, '-');
if (!dash || !dash[1]) return -1;
for (const char * p = dash + 1; *p; ++p)
if (*p < '0' || *p > '9') return -1;
return std::atoi(dash + 1);
}
void RouterHook::set_compute_trace(bool on) {
ctrace_on_ = on;
compute_rows_.clear();
}
void RouterHook::begin_compute_batch(int step, int phase, int turn) {
ctrace_step_ = step;
ctrace_phase_ = phase;
ctrace_turn_ = turn;
ctrace_seq_ = 0;
// The first node of the graph is charged from here, so the mark must be taken as close to
// llama_decode as the caller can manage — anything between them lands on node 0.
ctrace_mark_ = std::chrono::steady_clock::now();
ctrace_faults_ = pio::major_faults();
}
bool RouterHook::on_eval(ggml_tensor * t, bool ask) {
// ── compute trace: close the previous node's interval, open the next ──
// Ordering matters: this runs before every other job below, so the timestamp is as close to the
// boundary as possible and the streamer's own work (load_layer, the residency query) lands
// inside the routing node's interval where it belongs — that IS what routing costs here.
if (ctrace_on_ && !ask) {
const auto now = std::chrono::steady_clock::now();
const uint64_t faults = pio::major_faults();
ComputeTraceRow r;
r.turn = ctrace_turn_;
r.phase = ctrace_phase_;
r.step = ctrace_step_;
r.seq = ctrace_seq_++;
r.layer = node_layer(t->name);
r.op = ggml_op_name(t->op);
r.name = t->name;
r.wall_ns = (uint64_t) std::chrono::duration_cast<std::chrono::nanoseconds>(now - ctrace_mark_).count();
r.majflt = faults - ctrace_faults_;
compute_rows_.push_back(std::move(r));
ctrace_mark_ = now;
ctrace_faults_ = faults;
}
// ── capture: harvest expert weight tensors from every node's sources ──
if (capturing_) {
if (ask) {
@ -179,7 +229,8 @@ bool RouterHook::on_eval(ggml_tensor * t, bool ask) {
// Only a traced run asks for the weight nodes: each extra ask is another barrier.
int wl = -1;
const bool is_weights = trace_on_ && match_weights(t->name, wl);
if (ask) return is_topk || is_weights;
// The compute trace wants every node isolated; the streamer only wants the routing ones.
if (ask) return ctrace_on_ || is_topk || is_weights;
// Weights follow their layer's topk, so the pending record is already open; keep the last
// one offered (match_weights explains why) and let the flush read it.

27
core/src/moe/router_hook.h
View file

@ -23,8 +23,10 @@
#include "bmoe/recipe.h"
#include "bmoe/route_trace.h"
#include "bmoe/decode_trace.h"
#include "expert_stream_source.h"
#include <chrono>
#include <cstdint>
#include <unordered_set>
#include <vector>
@ -62,6 +64,22 @@ public:
// has returned. Off by default: asking for the extra nodes costs a barrier per layer.
void set_trace(bool on);
// ── compute trace (diagnostics; see bmoe/decode_trace.h) ────────────────────────
// When on, the hook asks for EVERY node, which makes ggml compute and synchronize each one
// alone — so the wall delta between consecutive callbacks is that node's real compute time,
// and the major-fault delta across it is the flash re-read that time was actually spent on.
// This is the only way to measure compute from outside llama.cpp, and it is expensive by
// construction: a barrier per node, and no operator coalescing. Off by default; a traced run
// is not a benchmark run. Independent of streaming — a dense baseline can be traced too.
void set_compute_trace(bool on);
// Frame the rows of one llama_decode, as begin_trace_batch does for the route trace. Rows are
// stamped with `step`; a prefill chunk attributes its whole graph to the batch's last position,
// since a node is computed once for the batch, not per token.
void begin_compute_batch(int step, int phase, int turn);
std::vector<ComputeTraceRow> & compute_rows() { return compute_rows_; }
// Frame the rows of one llama_decode. `base_pos` is the context position of the batch's
// first token and `n_tokens` its length, so a prefill chunk's rows carry real per-token step
// numbers — and so a layer that saw fewer tokens than the batch can still be placed (see
@ -108,6 +126,15 @@ private:
PendingLayer pending_;
std::vector<RouteTraceRow> trace_rows_;
std::unordered_set<int32_t> charged_; // per-flush scratch: experts already charged for a read
// Compute trace. All of this is inert unless ctrace_on_. `ctrace_mark_` is the previous
// isolation boundary: the node reported by the next callback is charged the wall since it.
bool ctrace_on_ = false;
int ctrace_step_ = 0, ctrace_phase_ = 0, ctrace_turn_ = 0;
int ctrace_seq_ = 0;
std::chrono::steady_clock::time_point ctrace_mark_;
uint64_t ctrace_faults_ = 0;
std::vector<ComputeTraceRow> compute_rows_;
};
} // namespace bmoe

69
docs/telemetry.md
View file

@ -20,9 +20,11 @@ BMOE_PROGRESS {"step":<int>,"steps":<int>,"wall_ms":<float>,"io_ms":<float>,
- `read_mb` is the flash bytes read this token; `stall_ms` is the overlap-only wall time compute
lost to reads (0 in serial mode).
- `wall_ms` = total token time; `io_ms` = flash read time. `compute_ms` is a **residual, not a
measured quantity**: there is no clock around llama.cpp's matmul kernels (adding one would mean
patching the submodule), so compute is whatever wall time is left after the measured terms are
subtracted — `wall_ms − io_ms − mgmt_ms` in serial, `wall_ms − stall_ms − mgmt_ms` under overlap.
measured quantity**: no clock runs around llama.cpp's matmul kernels in a normal run, so compute
is whatever wall time is left after the measured terms are subtracted — `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.
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
@ -234,3 +236,64 @@ total flash streamed this generation, and `stall_s_tok`/`mgmt_s_tok` the per-tok
cache-management cost. `BMOE_ERROR` with `fatal:false` is a rejected
request (e.g. the prompt plus `n_predict` exceeds `n_ctx`) and leaves the session usable;
`fatal:true` means the process is ending.
## Decode traces
`--compute-trace PATH` and `--io-trace PATH` decompose the two terms the per-token CSV can only
report as totals. The route trace answers *what the router asked for*; these answer *where the
time went*, and they exist because the headline number they decompose is not measured at all —
`compute_ms` is a residual, so every cost the engine does not itself clock (page faults, scheduler
stalls, the matmuls) is pooled into it.
Both are **diagnostics, not telemetry**, and both perturb what they measure. **A traced run is not
a benchmark run.** Read the shares, not the absolutes.
### `--compute-trace` — one row per graph node
Returning `true` from the eval callback makes ggml compute exactly up to that node, synchronize,
and call back — so the wall delta between consecutive boundaries is that node's real compute time,
measured, not inferred. The same boundaries sample major faults, which is the point: on a >RAM
model most of "compute" is flash faults, and no residual can show that. The cost is a barrier per
node and no operator coalescing, so the total is inflated well above an untraced run.
Unlike the other traces this one does **not** need `--moe-stream`: it times the graph, which a
plain mmap run has too, so a dense baseline can be traced and compared against a streamed one.
```
# compute_trace v1
# model=... arch=qwen3moe n_layer=48 n_threads=4 io_threads=4 o_direct=1 overlap=0
turn,phase,step,seq,layer,op,name,wall_ns,majflt
0,1,29,0,-1,GET_ROWS,embd,428500,0
0,1,29,1,0,RMS_NORM,norm-0,19500,0
```
`seq` is the node's execution order in the decode; `layer` is parsed from the node name's `-<il>`
suffix (`-1` = belongs to no layer: embeddings, the output head, masks). `op` and `name` are raw —
which node is attention vs dense FFN vs expert matmul is naming policy that varies by
architecture, so the engine reports what the graph said and the analysis script classifies.
### `--io-trace` — one row per flash read
Needs `--moe-stream` (no engine-issued reads without it). Records every `pread` the streamer
issues, tagged with the `(layer, expert, projection)` it serves.
```
# io_trace v1
turn,phase,step,layer,expert,proj,lane,spec,offset,req_bytes,read_bytes,latency_ns
0,1,29,0,87,1,0,0,1526304,65536,69632,416800
```
`req_bytes` is what the caller wanted; `read_bytes` is the aligned window actually pulled — the
gap is O_DIRECT alignment waste, and `read_bytes` is what effective bandwidth must be judged
against. `spec=1` marks a speculative prefetch read. Rows are stamped with the decode they were
drained after, so a read straddling a token boundary is attributed to the decode that flushed it.
### Reading them
`scripts/decode-analyze.py` (stdlib only) reports what each file is for:
```bash
scripts/decode-analyze.py compute ct.csv --layers # share by op, fault attribution, by layer
scripts/decode-analyze.py io io.csv --adjacent # latency percentiles, size/bandwidth, lanes,
# and the coalescing ceiling
```

222
scripts/decode-analyze.py Normal file
View file

@ -0,0 +1,222 @@
#!/usr/bin/env python3
"""Read a --compute-trace / --io-trace pair without a spreadsheet.
The per-token CSV says how long a token took and calls the leftover "compute". These traces say
what that leftover is actually made of, and what the flash floor under it really is. Stdlib only.
decode-analyze.py compute ct.csv # where compute goes, by op / by layer / faults
decode-analyze.py io io.csv # the flash floor: latency, size, waste, lanes
decode-analyze.py io io.csv --adjacent # how much of a token's reads could coalesce
A traced run is not a benchmark run: isolating every node forbids ggml the operator coalescing it
would normally do, and the I/O rows take a lock per read. Read the proportions, not the absolutes.
"""
import argparse
import csv
import sys
from collections import defaultdict
def read_trace(path):
"""Split the `#` preamble from the rows. Mirrors scripts/route-analyze.py's reader."""
meta, rows = {}, []
with open(path, newline="", encoding="utf-8") as f:
lines = f.readlines()
body = []
for ln in lines:
if ln.startswith("#"):
for tok in ln.lstrip("# ").rstrip().split():
if "=" in tok:
k, v = tok.split("=", 1)
meta[k] = v
else:
body.append(ln)
rows = list(csv.DictReader(body))
return meta, rows
def fmt_ms(ns):
return f"{ns / 1e6:9.2f}"
def bar(frac, width=28):
n = int(round(frac * width))
return "#" * n + "." * (width - n)
def decode_only(rows):
"""Steady-state decode: phase 1. Prefill is one batch of many tokens, a different regime."""
d = [r for r in rows if r["phase"] == "1"]
return d if d else rows
def cmd_compute(args):
meta, rows = read_trace(args.path)
rows = decode_only(rows)
if not rows:
sys.exit("no rows")
steps = sorted({int(r["step"]) for r in rows})
n_steps = len(steps)
total = sum(int(r["wall_ns"]) for r in rows)
faults = sum(int(r["majflt"]) for r in rows)
print(f"model={meta.get('model','?')} arch={meta.get('arch','?')} "
f"n_layer={meta.get('n_layer','?')} threads={meta.get('n_threads','?')}")
print(f"decode steps={n_steps} nodes/step={len(rows)//max(1,n_steps)} "
f"measured compute={fmt_ms(total/max(1,n_steps))} ms/token majflt={faults/max(1,n_steps):.0f}/token")
print("\nNOTE: measured, not a residual - each node was isolated and synchronized. The barrier\n"
" that makes it measurable also inflates it; compare shares, not absolutes.\n")
# ── by op ──
by_op = defaultdict(lambda: [0, 0, 0]) # ns, count, majflt
for r in rows:
e = by_op[r["op"]]
e[0] += int(r["wall_ns"])
e[1] += 1
e[2] += int(r["majflt"])
print(f"{'op':<18}{'ms/token':>10}{'share':>8} {'majflt/tok':>10} {'':<28}")
for op, (ns, cnt, mf) in sorted(by_op.items(), key=lambda kv: -kv[1][0])[: args.top]:
share = ns / total if total else 0
print(f"{op:<18}{fmt_ms(ns/max(1,n_steps))}{share*100:7.1f}% {mf/max(1,n_steps):10.1f} {bar(share)}")
# ── faults: the question the residual could never answer ──
if faults:
print("\nfault attribution - where the >RAM stall is billed as compute")
by_fault = defaultdict(lambda: [0, 0])
for r in rows:
mf = int(r["majflt"])
if not mf:
continue
e = by_fault[r["op"]]
e[0] += mf
e[1] += int(r["wall_ns"])
for op, (mf, ns) in sorted(by_fault.items(), key=lambda kv: -kv[1][0])[:8]:
print(f" {op:<16}{mf/max(1,n_steps):9.1f} majflt/tok in {fmt_ms(ns/max(1,n_steps))} ms/tok "
f"({ns/total*100:4.1f}% of compute)")
faulting = sum(v[1] for v in by_fault.values())
print(f" {'TOTAL':<16}{faults/max(1,n_steps):9.1f} majflt/tok in {fmt_ms(faulting/max(1,n_steps))} ms/tok "
f"({faulting/total*100:4.1f}% of compute)")
print(" ^ nodes that faulted. Their time is flash wait, not arithmetic - subtract it before\n"
" calling this model compute-bound.")
# ── by layer ──
if args.layers:
by_layer = defaultdict(int)
for r in rows:
by_layer[int(r["layer"])] += int(r["wall_ns"])
print("\nby layer (-1 = no layer: embeddings, output head, masks)")
mx = max(by_layer.values()) or 1
for il in sorted(by_layer):
ns = by_layer[il]
print(f" layer {il:>3} {fmt_ms(ns/max(1,n_steps))} ms/tok {bar(ns/mx)}")
def cmd_io(args):
meta, rows = read_trace(args.path)
rows = decode_only(rows)
if not rows:
sys.exit("no rows")
steps = sorted({int(r["step"]) for r in rows})
n_steps = len(steps)
lat = [int(r["latency_ns"]) for r in rows]
got = sum(int(r["read_bytes"]) for r in rows)
want = sum(int(r["req_bytes"]) for r in rows)
busy = sum(lat)
print(f"model={meta.get('model','?')} io_threads={meta.get('io_threads','?')} "
f"o_direct={meta.get('o_direct','?')} overlap={meta.get('overlap','?')}")
print(f"decode steps={n_steps} reads={len(rows)} ({len(rows)/max(1,n_steps):.0f}/token) "
f"read={got/2**20/max(1,n_steps):.1f} MiB/token")
# Per-lane time is what the drive was actually asked to do; wall is hidden by overlap.
print(f"lane-busy bandwidth={got/2**20/(busy/1e9):.0f} MiB/s aggregated over lanes "
f"({busy/1e9/max(1,n_steps)*1000:.0f} ms lane-busy/token)")
waste = (got - want) / got * 100 if got else 0
print(f"alignment waste={waste:.1f}% ({(got-want)/2**20/max(1,n_steps):.2f} MiB/token read but not wanted)")
# ── latency distribution: is the floor seek-bound or size-bound? ──
lat.sort()
def pct(p):
return lat[min(len(lat) - 1, int(len(lat) * p))] / 1e3
print("\nper-read latency (us)")
for p in (0.5, 0.9, 0.99):
print(f" p{int(p*100):<3} {pct(p):9.1f}")
print(f" max {lat[-1]/1e3:9.1f}")
# ── size vs bandwidth: the coalescing case, in one table ──
print("\nby request size - the per-read cost of scattering")
buckets = defaultdict(lambda: [0, 0, 0]) # count, bytes, ns
for r in rows:
kb = int(r["read_bytes"]) // 1024
b = 1 << (kb.bit_length() - 1) if kb else 0
e = buckets[b]
e[0] += 1
e[1] += int(r["read_bytes"])
e[2] += int(r["latency_ns"])
print(f" {'size':>8}{'reads':>9}{'MiB':>9}{'MiB/s':>9}{'us/read':>9}")
for b in sorted(buckets):
cnt, by, ns = buckets[b]
print(f" {b:>6} K{cnt:>9}{by/2**20:9.1f}{by/2**20/(ns/1e9):9.0f}{ns/cnt/1e3:9.1f}")
# ── lanes ──
by_lane = defaultdict(lambda: [0, 0])
for r in rows:
e = by_lane[int(r["lane"])]
e[0] += int(r["read_bytes"])
e[1] += int(r["latency_ns"])
print("\nby lane - uneven busy time means a lane is starved, not that the drive is full")
for ln in sorted(by_lane):
by, ns = by_lane[ln]
print(f" lane {ln}: {by/2**20/max(1,n_steps):7.1f} MiB/tok busy {ns/1e9/max(1,n_steps)*1000:7.1f} ms/tok"
f" {by/2**20/(ns/1e9):6.0f} MiB/s")
spec = [r for r in rows if r["spec"] == "1"]
if spec:
sb = sum(int(r["read_bytes"]) for r in spec)
print(f"\nspeculative: {len(spec)} reads, {sb/2**20/max(1,n_steps):.1f} MiB/token "
f"({sb/got*100:.0f}% of bytes read)")
if args.adjacent:
# How much of a step's reads are back-to-back in the file? That is the ceiling on what an
# expert-contiguous layout / runtime coalescing could merge — the roadmap's read-bandwidth item.
print("\nadjacency - the coalescing ceiling")
merged_tot = runs_tot = 0
for st in steps:
ext = sorted((int(r["offset"]), int(r["offset"]) + int(r["read_bytes"]))
for r in rows if int(r["step"]) == st)
if not ext:
continue
runs, cur_end = 1, ext[0][1]
for a, b in ext[1:]:
if a > cur_end:
runs += 1
cur_end = max(cur_end, b)
merged_tot += len(ext)
runs_tot += runs
if merged_tot:
print(f" {merged_tot/max(1,n_steps):.0f} reads/token span {runs_tot/max(1,n_steps):.0f} "
f"contiguous runs/token")
print(f" perfectly coalesced, that is {merged_tot/max(1,runs_tot):.1f}x fewer requests "
f"for the same bytes")
def main():
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
sub = ap.add_subparsers(dest="cmd", required=True)
c = sub.add_parser("compute", help="what the compute residual is actually made of")
c.add_argument("path")
c.add_argument("--top", type=int, default=12, help="how many ops to list")
c.add_argument("--layers", action="store_true", help="also break down by layer")
c.set_defaults(fn=cmd_compute)
i = sub.add_parser("io", help="the flash floor: latency, size, waste, lanes")
i.add_argument("path")
i.add_argument("--adjacent", action="store_true", help="estimate the coalescing ceiling")
i.set_defaults(fn=cmd_io)
args = ap.parse_args()
args.fn(args)
if __name__ == "__main__":
main()