From ad13038ce0ef203d503fcc6af7d6ddbd22d09cfa Mon Sep 17 00:00:00 2001 From: Helldez <74499579+Helldez@users.noreply.github.com> Date: Tue, 28 Jul 2026 15:46:52 +0200 Subject: [PATCH] feat(moe): --io-two-wave, publish a layer's read batch in two waves (#128) A cold layer's batch became visible to the I/O lanes only after every miss took its page commits - up to three vm_commit syscalls per cold expert of bookkeeping sitting in front of the first byte of I/O, which is the latency-to-first-slice the sidecar refutation (PR #90) identified as the binding constraint. (#118) With the flag on, only the first present projection - the one mul_mat_id blocks on first - is committed up front; its jobs publish and wake the lanes immediately, and the remaining projections are committed and appended while the lanes already read. The drain protocol grew the one thing this needs: - io_drain copies each job out under the lock, so jobs_ growing (and possibly reallocating) mid-batch cannot leave a worker holding a dangling reference; - the worker wait predicate admits next_idx_ < batch_njobs_, so a worker that drained wave one and left comes back for a batch that grew in the SAME generation - the gen comparison alone never would; - a wave-two commit failure goes fatal and wakes the ready waiters, because wave one already published flags this batch will never flip. Batch completion cannot fire between the waves: the only thread that waits on done_cnt_ == batch_njobs_ is the eval thread, and it is the one appending wave two. Overlap + LRU cache only (validate() enforces both); recorded in the CSV preamble as io_two_wave. Default off: the win is bounded by the commit cost per cold layer, and the failure mode of a drain-protocol bug is a hang the host cannot reproduce - so a new gate (G4d) holds two-wave output byte-identical to serial streaming, and the flag stays off until the on-device A/B (#120) says the win is real. --- CHANGELOG.md | 12 +++ cli/main.cpp | 5 + core/include/bmoe/config.h | 9 ++ core/include/bmoe/metrics.h | 1 + core/src/config.cpp | 9 ++ core/src/engine/session.cpp | 1 + core/src/metrics/csv_metrics_sink.cpp | 6 +- core/src/moe/expert_stream_source.cpp | 138 ++++++++++++++++++++------ core/src/moe/expert_stream_source.h | 8 +- docs/telemetry.md | 2 +- tests/moe_gates.cpp | 18 +++- 11 files changed, 172 insertions(+), 37 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6848e80..201b202 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,18 @@ Semantic Versioning. ## [Unreleased] +### Added +- **`--io-two-wave` — publish a layer's read batch in two waves (experimental, off by default).** + A cold layer's batch used to become visible to the I/O lanes only after every miss took its page + commits — roughly two dozen syscalls of bookkeeping sitting in front of the first byte of I/O on + an eight-miss layer, which is exactly the latency-to-first-slice the sidecar refutation + identified as the binding constraint (#118). With the flag on, the first projection's jobs (the + ones `mul_mat_id` blocks on first) are committed and published immediately, and the remaining + projections are committed and appended while the lanes already read. The drain protocol grew the + one thing it needed — a worker that finished wave one comes back for a batch that grew in the + same generation — and a new gate (`G4d`) holds the two-wave output byte-identical to serial + streaming. Off by default until the on-device A/B (#120) says the bounded win is real. + ### Changed - **`BMOE_PROGRESS` carries the answer as a delta, not cumulatively.** Every per-token line repeated the whole answer and reasoning so far, so a generation of n tokens wrote, JSON-escaped diff --git a/cli/main.cpp b/cli/main.cpp index a1ea495..689088e 100644 --- a/cli/main.cpp +++ b/cli/main.cpp @@ -402,6 +402,9 @@ static void print_usage(const char * argv0) { " --load-all debug: read ALL experts each token (A/B baseline)\n" " --force-cache allow a cache-mb in the pathological band\n" " --overlap overlap async expert reads with FFN compute (needs the fork)\n" + " --io-two-wave publish a layer's first-projection reads before committing the\n" + " rest, so the lanes start sooner (needs --overlap and the cache;\n" + " experimental, off by default pending the on-device A/B)\n" " --prefetch K temporally prefetch the next K layers' experts (needs the cache)\n" " --drop-cold-experts F skip a routed expert that is a cache MISS and carries less than\n" " F x (1/top-k) of the routing's weight. F in (0, 1]; 1.0 is the\n" @@ -592,6 +595,8 @@ int main(int argc, char ** argv) { cfg.moe.force_cache = true; else if (a == "--overlap") cfg.moe.overlap = true; + else if (a == "--io-two-wave") + cfg.moe.io_two_wave = true; else if (a == "--prefetch") cfg.moe.prefetch_layers = std::atoi(next("--prefetch")); else if (a == "--prefetch-sync") // debug: complete speculative reads synchronously diff --git a/core/include/bmoe/config.h b/core/include/bmoe/config.h index 32dd903..2e13cda 100644 --- a/core/include/bmoe/config.h +++ b/core/include/bmoe/config.h @@ -71,6 +71,15 @@ struct MoeStreamConfig { // Requires the Helldez/llama.cpp fork submodule (the hook); run() fails fast otherwise. bool overlap = false; + // Two-wave batch publish (#118). A layer's read batch normally becomes visible to the I/O + // lanes only after ALL of its staging finished — up to three page-commit syscalls per cold + // expert sitting in front of the first byte of I/O, on the latency-to-first-slice path the + // sidecar refutation identified as the binding constraint. With this on, the jobs of the + // first projection (the one mul_mat_id blocks on first) are committed and published + // immediately, and the remaining projections are committed and appended while the lanes + // already read. Overlap + LRU cache only. Default off pending the on-device A/B. + bool io_two_wave = false; + // Temporal prefetch: while a token computes layer l, speculatively read on the idle I/O // lanes the experts the PREVIOUS token routed at layers l+1..l+prefetch_layers, betting on // the strong temporal locality of routing. A correct guess turns the next layer's read into diff --git a/core/include/bmoe/metrics.h b/core/include/bmoe/metrics.h index 053752f..fab6047 100644 --- a/core/include/bmoe/metrics.h +++ b/core/include/bmoe/metrics.h @@ -189,6 +189,7 @@ struct RunInfo { int io_threads = 0; bool o_direct = false; bool overlap = false; + bool io_two_wave = false; // two-wave batch publish (#118): first projection published early int prefetch_layers = 0; bool predict_prefetch = false; // stale-gate predictive prefetch (see MoeStreamConfig) bool predict_log = false; // the accuracy probe: costs a barrier and two GEMVs per layer diff --git a/core/src/config.cpp b/core/src/config.cpp index 75085a8..f2e0e0a 100644 --- a/core/src/config.cpp +++ b/core/src/config.cpp @@ -110,6 +110,15 @@ ValidationResult validate(const RunConfig & cfg) { "speculative reads land in the per-layer cache buffers, which do not exist " "with the cache off."); } + if (m.io_two_wave && !m.overlap) { + return fail("moe.io_two_wave requires moe.overlap: without overlap the caller drains the " + "batch synchronously and there is no lane idling on the publish to wake early."); + } + if (m.io_two_wave && !cache_on) { + return fail("moe.io_two_wave requires the LRU cache (cache_mb > 0 or cache_auto): the wave " + "split exists to move per-expert page commits off the publish path, and the " + "shared-slot path has none."); + } if (m.predict_spec_max < 0 || m.predict_spec_max > 8) { return fail("moe.predict_spec_max must be in [0, 8] (0 = retention only, no speculation)"); } diff --git a/core/src/engine/session.cpp b/core/src/engine/session.cpp index 2c6d8d8..9176a71 100644 --- a/core/src/engine/session.cpp +++ b/core/src/engine/session.cpp @@ -583,6 +583,7 @@ std::unique_ptr Session::open(const SessionConfig & cfg, ri.io_threads = cfg.moe.enabled ? cfg.moe.io_threads : 0; ri.o_direct = cfg.moe.enabled && cfg.moe.o_direct; ri.overlap = cfg.moe.enabled && cfg.moe.overlap; + ri.io_two_wave = cfg.moe.enabled && cfg.moe.io_two_wave; ri.prefetch_layers = cfg.moe.enabled ? cfg.moe.prefetch_layers : 0; ri.predict_prefetch = cfg.moe.enabled && cfg.moe.predict_prefetch; ri.predict_log = cfg.moe.enabled && cfg.moe.predict_log; diff --git a/core/src/metrics/csv_metrics_sink.cpp b/core/src/metrics/csv_metrics_sink.cpp index f785945..e415962 100644 --- a/core/src/metrics/csv_metrics_sink.cpp +++ b/core/src/metrics/csv_metrics_sink.cpp @@ -36,12 +36,12 @@ public: r.n_ubatch, r.chatml); std::fprintf(f_, "# moe_stream=%d cache_mb=%d cache_auto=%d cache_floor_mb=%d cache_ceil_mb=%d " - "force_cache=%d load_all=%d io_threads=%d o_direct=%d overlap=%d prefetch=%d " + "force_cache=%d load_all=%d io_threads=%d o_direct=%d overlap=%d io_two_wave=%d prefetch=%d " "predict_prefetch=%d predict_log=%d predict_spec_max=%d prefetch_sync=%d " "dense_weights=%s drop_cold_frac=%.4g drop_renorm=%d drop_prefill=%d\n", r.moe_stream, r.cache_mb, r.cache_auto, r.cache_floor_mb, r.cache_ceil_mb, r.force_cache, - r.load_all, r.io_threads, r.o_direct, r.overlap, r.prefetch_layers, r.predict_prefetch, - r.predict_log, r.predict_spec_max, r.prefetch_sync, r.dense_weights.c_str(), + r.load_all, r.io_threads, r.o_direct, r.overlap, r.io_two_wave, r.prefetch_layers, + r.predict_prefetch, r.predict_log, r.predict_spec_max, r.prefetch_sync, r.dense_weights.c_str(), (double) r.drop_cold_frac, r.drop_renorm, r.drop_prefill); std::fprintf(f_, "# temp=%.4g top_k=%d top_p=%.4g seed=%u compute_trace_layers=%d\n", (double) r.temp, r.top_k, (double) r.top_p, r.seed, r.compute_trace_layers); diff --git a/core/src/moe/expert_stream_source.cpp b/core/src/moe/expert_stream_source.cpp index ff8b1c4..bac1e8b 100644 --- a/core/src/moe/expert_stream_source.cpp +++ b/core/src/moe/expert_stream_source.cpp @@ -40,6 +40,7 @@ bool ExpertStreamSource::init(const std::string & gguf_path, n_layer_ = (int) layers_.size(); load_all_ = cfg.load_all; overlap_ = cfg.overlap; + two_wave_ = cfg.io_two_wave; prefetch_sync_ = cfg.prefetch_sync && !cfg.overlap; // serial only: overlap lane 0 is a worker cache_max_ = (size_t) std::max(0, cfg.cache_mb) * 1024ull * 1024ull; io_threads_ = std::max(1, std::min(MoeStreamConfig::io_threads_max, cfg.io_threads)); @@ -265,13 +266,14 @@ void ExpertStreamSource::take_io_trace_rows(std::vector & out) { void ExpertStreamSource::io_drain(int lane, uint64_t my_gen) { for (;;) { - size_t i; + IoJob j; { std::lock_guard lk(io_mtx_); if (batch_gen_ != my_gen || next_idx_ >= batch_njobs_) return; - i = next_idx_++; + // Copy under the lock: a two-wave publish appends to jobs_ mid-batch, and a vector + // that reallocates would leave a reference taken here dangling. + j = jobs_[next_idx_++]; } - const IoJob & j = jobs_[i]; if (!read_slice(lane, j)) { io_err_.store(true); if (overlap_) fatal_.store(true, std::memory_order_release); @@ -302,7 +304,12 @@ void ExpertStreamSource::io_worker(int lane) { for (;;) { { std::unique_lock lk(io_mtx_); - io_cv_.wait(lk, [&] { return io_stop_ || batch_gen_ > seen || spec_next_ < spec_jobs_.size(); }); + // `next_idx_ < batch_njobs_` admits a batch that GREW in the same generation (a + // two-wave publish): a worker that drained wave one and left has seen == batch_gen_, + // so the gen comparison alone would never bring it back for wave two. + io_cv_.wait(lk, [&] { + return io_stop_ || batch_gen_ > seen || next_idx_ < batch_njobs_ || spec_next_ < spec_jobs_.size(); + }); if (io_stop_) return; } uint64_t g; @@ -310,10 +317,10 @@ void ExpertStreamSource::io_worker(int lane) { std::lock_guard lk(io_mtx_); g = batch_gen_; } - if (g > seen) { // real batch first — it is latency-critical - seen = g; - io_drain(lane, g); - } + if (g > seen) seen = g; + // Real batch first — it is latency-critical. With nothing (left) to drain this returns + // on its first lock, so calling it on a spec-only wake costs one mutex round. + io_drain(lane, seen); // Then use spare capacity on queued speculative reads, yielding the moment a real batch // arrives (drain_spec bails when batch_gen_ advances past this worker's `seen`). drain_spec(lane, seen); @@ -634,7 +641,21 @@ uint64_t ExpertStreamSource::expert_bytes(int il) const { // The cache accounting both load paths share. See the header for why this is the part that is shared // and the job emission is not. Eval-thread only, like every other LRU mutation. -bool ExpertStreamSource::touch_entry(int il, int e, bool & hit, bool promote) { +bool ExpertStreamSource::commit_proj_pages(int il, int e, int p) { + const LayerExperts & L = layers_[il]; + const uint64_t slice = L.proj[p].nb2; + if (slice == 0) return true; // absent slot in a fused layout + char * dst = (char *) lbuf_[p][il] + (uint64_t) e * slice; + 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))) { + std::fprintf(stderr, "bmoe: commit failed\n"); + return false; + } + return true; +} + +bool ExpertStreamSource::touch_entry(int il, int e, bool & hit, bool promote, int commit_only_proj) { const int32_t id = il * n_expert_ + e; clookups_++; cstamp_[id] = cgen_; @@ -654,17 +675,12 @@ bool ExpertStreamSource::touch_entry(int il, int e, bool & hit, bool promote) { // Miss: commit the pages the caller's reads will fill, then enter the expert as resident. The // entry is valid from here on because the caller is contracted to schedule those reads before // anything can consume the slices — the barrier that makes this safe is the eval-callback's. - const LayerExperts & L = layers_[il]; + // (With commit_only_proj, the caller also contracts to commit the other projections before it + // emits their jobs; the residency accounting below still covers the whole entry, because the + // entry lives and is evicted as a unit either way.) 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 - char * dst = (char *) lbuf_[p][il] + (uint64_t) e * slice; - 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))) { - std::fprintf(stderr, "bmoe: commit failed\n"); - return false; - } + if (commit_only_proj >= 0 && p != commit_only_proj) continue; + if (!commit_proj_pages(il, e, p)) return false; } cvalid_[id] = 1; cspec_[id] = 0; // a real read, not speculative @@ -849,6 +865,8 @@ bool ExpertStreamSource::load_layer_async(int il, const int32_t * ids, int n_ids account_demand(il, (int) staged_.size()); maybe_sample_dense(); + bool published = false; // a two-wave batch publishes inside the staging branch + if (cache_max_ == 0) { // Cache off: every (projection, expert) is a fresh read into the shared full-size slot // at its canonical offset e*slice. Emit projection-major. @@ -864,12 +882,26 @@ bool ExpertStreamSource::load_layer_async(int il, const int32_t * ids, int n_ids } else { // Cache on: per-expert LRU bookkeeping (hit → mark all projections ready now; miss → // commit the pages and remember it). Then emit the misses' jobs projection-major. + // + // Two-wave publish (#118): normally no lane can start reading until every miss took up + // to three page-commit syscalls — bookkeeping sitting in front of the first byte of I/O, + // on the latency-to-first-slice path. With two_wave_, only the FIRST present projection + // (the one mul_mat_id blocks on first) is committed up front; its jobs publish + // immediately, and the remaining projections are committed and appended while the lanes + // already read. io_drain copies jobs under the lock and the worker wait predicate admits + // a grown batch, so appending mid-generation is safe. + const int p0 = [&] { + for (int p = 0; p < MoeRecipe::max_exps; ++p) + if (L.proj[p].nb2) return p; + return -1; + }(); + const bool two_wave = two_wave_ && !load_all_ && p0 >= 0; seen_.assign(seen_.size(), 0); // reuse as a per-staged miss marker keyed by expert for (int e : staged_) { bool hit = false; // promote=false: the token-major loop below re-orders every touched id anyway, so a // hit-path LRU move here was k wasted pointer operations per layer per token. - if (!touch_entry(il, e, hit, /*promote=*/false)) { + if (!touch_entry(il, e, hit, /*promote=*/false, two_wave ? p0 : -1)) { // Nothing waits on a flag we will never publish: abort the graph instead. fatal_.store(true, std::memory_order_release); return false; @@ -881,9 +913,9 @@ bool ExpertStreamSource::load_layer_async(int il, const int32_t * ids, int n_ids } seen_[e] = 1; // this expert missed → its projections need reads } - for (int p = 0; p < MoeRecipe::max_exps; ++p) { + auto emit_proj = [&](int p) { const uint64_t slice = L.proj[p].nb2; - if (slice == 0) continue; + if (slice == 0) return; for (int e : staged_) { 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); @@ -891,6 +923,47 @@ bool ExpertStreamSource::load_layer_async(int il, const int32_t * ids, int n_ids L.proj[p].file_off + (uint64_t) e * slice, slice, flag, e, (int16_t) il, (int8_t) p, 0}); } + }; + if (!two_wave) { + for (int p = 0; p < MoeRecipe::max_exps; ++p) emit_proj(p); + } else { + // Wave one: the first projection's jobs, published before anything else is committed. + emit_proj(p0); + { + std::lock_guard lk(io_mtx_); + batch_njobs_ = jobs_.size(); + next_idx_ = 0; + done_cnt_ = 0; + io_err_.store(false); + batch_flag_gen_ = gen; + ++batch_gen_; + } + io_cv_.notify_all(); + published = true; + // Wave two: commit the remaining projections' pages, then append their jobs. A commit + // failure here is after wave one published, so waiters may already be blocked on + // flags this batch will now never flip — go fatal and wake them to observe it. + for (int e : staged_) { + if (!seen_[e]) continue; + for (int p = 0; p < MoeRecipe::max_exps; ++p) { + if (p == p0) continue; + if (!commit_proj_pages(il, e, p)) { + fatal_.store(true, std::memory_order_release); + { + std::lock_guard lk(ready_mtx_); + ready_cv_.notify_all(); + } + return false; + } + } + } + { + std::lock_guard lk(io_mtx_); + for (int p = 0; p < MoeRecipe::max_exps; ++p) + if (p != p0) emit_proj(p); + batch_njobs_ = jobs_.size(); + } + io_cv_.notify_all(); } // Promote every touched expert in raw id order (token-major) so the LRU order reflects @@ -912,16 +985,19 @@ bool ExpertStreamSource::load_layer_async(int il, const int32_t * ids, int n_ids // 4. Publish the batch and return immediately — no drain. Workers fill the slices and // flip the flags as they go; the compute threads block on those flags via the hook. - { - std::lock_guard lk(io_mtx_); - batch_njobs_ = jobs_.size(); - next_idx_ = 0; - done_cnt_ = 0; - io_err_.store(false); - batch_flag_gen_ = gen; - ++batch_gen_; + // (A two-wave batch already published both waves inside the staging branch.) + if (!published) { + { + std::lock_guard lk(io_mtx_); + batch_njobs_ = jobs_.size(); + next_idx_ = 0; + done_cnt_ = 0; + io_err_.store(false); + batch_flag_gen_ = gen; + ++batch_gen_; + } + io_cv_.notify_all(); } - io_cv_.notify_all(); // 5. Evict cold entries to budget. Safe to run concurrently with this batch's reads: step 1 // guaranteed no stale in-flight jobs, and current-gen entries (cstamp_ == cgen_) — the diff --git a/core/src/moe/expert_stream_source.h b/core/src/moe/expert_stream_source.h index 5bddcd5..f9d550a 100644 --- a/core/src/moe/expert_stream_source.h +++ b/core/src/moe/expert_stream_source.h @@ -158,7 +158,12 @@ private: // LRU mode only (cache_max_ > 0) — the shared-slot path has no entries to account for. // `promote` = false skips the hit-path LRU move for callers that re-order every touched id // afterwards anyway (the overlap path's token-major promote loop); a miss is always linked. - bool touch_entry(int il, int e, bool & hit, bool promote = true); + // `commit_only_proj` >= 0 commits only that projection's pages on a miss (two-wave publish: + // the caller contracts to commit the rest via commit_proj_pages before emitting their jobs). + bool touch_entry(int il, int e, bool & hit, bool promote = true, int commit_only_proj = -1); + + // Commit the pages of one (layer, expert, projection) cache slice so a read can land in it. + bool commit_proj_pages(int il, int e, int p); // LRU helpers (active only when cache_max_ > 0) void lru_unlink(int32_t id); @@ -176,6 +181,7 @@ private: bool active_ = false; bool load_all_ = false; bool overlap_ = false; + bool two_wave_ = false; // publish the first projection's jobs before committing the rest (#118) bool prefetch_sync_ = false; // test only: drain prefetch reads synchronously (serial mode) int n_layer_ = 0; int n_expert_ = 0; diff --git a/docs/telemetry.md b/docs/telemetry.md index 0455f25..caf7180 100644 --- a/docs/telemetry.md +++ b/docs/telemetry.md @@ -167,7 +167,7 @@ prints just the summary lines. # model= arch= n_layer= n_expert= n_expert_used= threads= n_ctx= n_ubatch= chatml=<0|1> # moe_stream=<0|1> cache_mb= cache_auto=<0|1> cache_floor_mb= cache_ceil_mb= - force_cache=<0|1> load_all=<0|1> io_threads= o_direct=<0|1> overlap=<0|1> prefetch= + force_cache=<0|1> load_all=<0|1> io_threads= o_direct=<0|1> overlap=<0|1> io_two_wave=<0|1> prefetch= predict_prefetch=<0|1> predict_log=<0|1> predict_spec_max= prefetch_sync=<0|1> dense_weights= drop_cold_frac= drop_renorm=<0|1> drop_prefill=<0|1> # temp= top_k= top_p= seed= compute_trace_layers= diff --git a/tests/moe_gates.cpp b/tests/moe_gates.cpp index 97c138e..0938147 100644 --- a/tests/moe_gates.cpp +++ b/tests/moe_gates.cpp @@ -256,7 +256,18 @@ int main(int argc, char ** argv) { ov1.moe.io_threads = 1; ov1.moe.overlap = true; - std::string s_ov0, s_ovc, s_ov1; + // overlap, cache, two-wave publish: the lanes wake on the first projection's jobs and the + // batch grows mid-generation. Exercises the grown-batch worker predicate and the split + // commit; the wait-per-expert hook must still gate every projection to the same bytes. + RunConfig ov2w = base(model); + ov2w.moe.enabled = true; + ov2w.moe.cache_mb = 2; + ov2w.moe.force_cache = true; + ov2w.moe.io_threads = 4; + ov2w.moe.overlap = true; + ov2w.moe.io_two_wave = true; + + std::string s_ov0, s_ovc, s_ov1, s_ov2w; if (!gen(ov0, s_ov0, err)) { std::fprintf(stderr, "overlap(cache off) run failed: %s\n", err.c_str()); return 2; @@ -269,9 +280,14 @@ int main(int argc, char ** argv) { std::fprintf(stderr, "overlap(io_threads=1) run failed: %s\n", err.c_str()); return 2; } + if (!gen(ov2w, s_ov2w, err)) { + std::fprintf(stderr, "overlap(two-wave) run failed: %s\n", err.c_str()); + return 2; + } fails += check("G4a overlap(cache off) == streaming(cache off)", s_s0, s_ov0); fails += check("G4b overlap(LRU cache) == streaming(cache off)", s_s0, s_ovc); fails += check("G4c overlap(io_threads=1) == streaming(cache off)", s_s0, s_ov1); + fails += check("G4d overlap(two-wave publish) == streaming(cache off)", s_s0, s_ov2w); #else std::printf("[SKIP] G4 (expert-ready hook not built)\n"); #endif