diff --git a/docs/MULTI_GPU.md b/docs/MULTI_GPU.md index 1270125b3a6..d0065f76831 100644 --- a/docs/MULTI_GPU.md +++ b/docs/MULTI_GPU.md @@ -168,6 +168,45 @@ The Coder on an RTX 5080 + RTX 3090 (Ryzen 9 9950X3D), 32K context; details in 3090 pair slower (68 / 90 tok/s decode): every extra card costs its own round per window. - More cards pay off when the model's routed experts do not fit the faster ones. +## One conversation with both cards busy (`--pipeline-windows`, opt-in) + +With a split the cards take turns on a verify window: the first card runs its layers and hands off, then waits while +the last card runs the rest, the head and the draft. `--pipeline-windows 2` lets the first card start the **next** +window while the last card still verifies this one. The next window is a guess: that this window is accepted whole +and that its bonus token is the one the draft layer predicts (the draft layer is run on through the drafts of the +window in flight). When the guess holds, half of the next window is already done; when it does not, the first card +puts its state back (a copy of its recurrent state taken while the window ran) and the next window is built from the +real tokens. A guessed window is only started when the draft layer's estimate says it is likely to be kept. Windows +copied from earlier context (`--suffix-draft`) are guessed past as well, which is where edits that copy text gain +most. `--pipeline-windows 1` overlaps only the short prompt reads that go through the verify windows +(`--short-read`). + +Two cards, exactly two stages, `--serve`. In the config: + +``` +"args": [ ..., "--pipeline-windows", "2" ], +"layer_split": "20" +``` + +- **Cost**: a second verify window on each card, 160 MiB more kept out of each card's expert cache, plus two copies + of the first card's recurrent state (about 3 MiB per GDN layer it runs) on the first card with `2`. +- **Same text**: the last card only ever runs windows that are verified, and every window row computes what it + would in any other window, so the tokens are the serial loop's. With `STRATA_IQ_MT_MIN=1 --pcie-frac 0 + --adapt-every 0` the greedy output is identical bit for bit to the serial loop's with the same expert caches. The pipeline keeps + its VRAM out of the caches, so against a run without the flag a few experts move from a card to the CPU, which + rounds them differently, and a near-tie can flip (a serial run given the same caches through `--vram-reserve-mib` + matches it exactly). +- **Off, with one line in the log saying why**, with `--batch` slots, `--peer-device`, the helper caches + (`--expert-cache-device1..3`, `--remote-expert-opt`), a split into three or more stages or onto one GPU + (`--split-device 0`), or no draft layer. A request with repetition penalties (`penalty_last_n`) or coupled + draft sampling decodes serially. +- **Measured** (Swift 1.5 IQ3_XXS, 160K context, q4_0 KV, the stock draft layer, RTX 4060 Ti (layers 0-19) + + RTX 5080 (20-47), i9-14900KF, 32 GB of RAM with the resident RAM mode on the split (#848); greedy, 500 tokens, two + interleaved pairs of three rounds, decode tok/s): Python code 90.4 -> 103.1, C code 76.6 -> 81.6, English prose + 63.0 -> 71.1, Italian prose 41.6 -> 46.3, a copy-heavy edit (a 5 KB file back with a rename) 90.5 -> 117.9; mean + 72.4 -> 84.0 (+16%). It pays when the windows are GPU-bound: with the experts read through the OS file cache + (`--mmap-experts` on that 32 GB PC) the file reads dominate and it measured no faster. + ## Several conversations at once With a layer split, `--batch N --batch-groups G --trim-stage-weights` decodes several conversations together and diff --git a/include/strata/core/expert_source.hpp b/include/strata/core/expert_source.hpp index 8fbc4a58cd2..2ce7a44e3ba 100644 --- a/include/strata/core/expert_source.hpp +++ b/include/strata/core/expert_source.hpp @@ -483,6 +483,9 @@ class FileExpertSource : public ExpertSource { uint8_t* exchange_buffer(int64_t q) const; /// Requires `has_resident(layer, in)`, `!has_resident(layer, out)` and `exchange_buffer(q)` holding out's blob. bool stage_exchange(int64_t layer, int64_t in, int64_t out, int64_t q); + /// --pipeline-windows: size the exchange table now, so a `stage_exchange` on the adaptive tier's thread never + /// reallocates it under a concurrent `blob` (the pool reads it while windows are in flight). + void prepare_overrides() { if (override_.empty()) override_.assign((size_t) blobs_, nullptr); } /// After the GPU copies of every staged swap have landed. Returns how many exchanges were applied. int64_t commit_exchanges(); int64_t exchanges() const { return exchanges_; } diff --git a/include/strata/core/mtp.hpp b/include/strata/core/mtp.hpp index bfb6ee0f57c..c802f87f4a6 100644 --- a/include/strata/core/mtp.hpp +++ b/include/strata/core/mtp.hpp @@ -48,6 +48,7 @@ class MtpDrafter { void set_prompt_len(int64_t n) { prompt_len_ = n; } /// At most this many drafts per round (below max_t - 1): a window longer than the MTP's comes from elsewhere. void set_max_drafts(int k) { max_drafts_ = k; } + int max_t() const { return max_t_; } uint64_t vram_bytes() const { return vram_; } /// The draft layer's K/V state (read-only: --serve's STRATA_STATE_HASH check hashes it) const QsaState& kv_state() const { return st_; } @@ -61,8 +62,14 @@ class MtpDrafter { bool bind(const WeightTable& wt, const NativeHead* head, const float* window_R, std::string& err); /// Prompt cells [cell0, cell0 + n): residual rows `R_rows` (device, hc*n_embd each) and `next_tokens` (host, - /// the token at position cell+1). Runs in batches of up to max_t rows. - bool prefill(const float* R_rows, const int32_t* next_tokens, int64_t n, int64_t cell0, std::string& err); + /// the token at position cell+1). Runs in batches of up to max_t rows. `sync` false: returns without waiting + /// for the drafter's stream (the caller orders on it: `stream()`). + bool prefill(const float* R_rows, const int32_t* next_tokens, int64_t n, int64_t cell0, std::string& err, + bool sync = true); + /// --pipeline-windows: capture the prefill graphs and size its input records for windows of up to max_t rows now, + /// so a later prefill(sync = false) of a window neither captures nor allocates (both wait for the device). + bool prepare_prefill(std::string& err); + cudaStream_t stream() const { return cs_; } /// One round: catch-up over T cells from `p` (rows = the window's final residuals, `tokens` = the window's /// argmaxes: row t pairs R_{p+t} with the token at p+t+1), then the draft chain from row `a` (the last @@ -85,6 +92,31 @@ class MtpDrafter { void set_ple_session(SessionState* s) { ple_ss_ = s; } bool coupled() const { return coupled_active_; } + // ---- --pipeline-windows 2: the chain as one asynchronous launch (the round and its steps back to back, no host + // wait), its first `n_force` steps fed the given tokens instead of their own picks (teacher forcing: the drafts of + // a window already in flight), each output read once the event after its step has completed. + /// Before `load`: the round/step graphs get the forcing kernel (a no-op while its token is -1), and the drafter's + /// stream the highest priority (STRATA_MTP_PRIORITY=0: the default priority). + void set_force_capture(bool on) { force_on_ = on; } + /// The rows the next round reads (null: the bound `window_R`), copied into the bound buffer before the round. + void set_source_R(const float* rows) { src_R_ = rows; } + /// Capture the round graphs for T = 1..max_t and every step graph now (a capture syncs the drafter's stream). + bool prepare_chain(std::string& err); + /// The catch-up over T cells from `p` (as draft()), then `n_out` outputs from row `a`: output j is the chain's + /// own pick, and step j (1 <= j <= n_force) takes force[j - 1] as its input token instead of output j - 1. + /// `n_early`: an event after each of the first `n_early` outputs (chain_outputs_ready). + bool chain_launch(int T, const int32_t* tokens, int64_t p, int a, const int32_t* force, int n_force, int n_out, + int n_early, std::string& err); + /// 1: every output is in chain_tok()/chain_prob(); 0: still running; -1: an error. + int chain_poll(std::string& err); + /// How many of the chain's first outputs (up to `n_early`; all of them once the chain is done) have landed in + /// chain_tok()/chain_prob(); -1: an error. A window's size is often decided by its first outputs (a draft below + /// spec_min_p ends it), so the next launch need not wait for all of them. + int chain_outputs_ready(std::string& err); + bool chain_live() const { return chain_live_; } + const int32_t* chain_tok() const { return chain_tok_; } + const float* chain_prob() const { return chain_prob_; } + double ms_draft = 0, ms_prefill = 0; int64_t rounds = 0; @@ -132,6 +164,16 @@ class MtpDrafter { int max_t_ = 0; int device_ = -1; ///< the device `load` ran on: the public calls switch to it (layer split) int max_drafts_ = 1 << 30; + // --pipeline-windows 2 (chain_launch) + bool stage_source_R(int T, std::string& err); ///< set_source_R's rows into the bound buffer + bool force_on_ = false, chain_live_ = false; + const float* src_R_ = nullptr; + int32_t *h_force_ = nullptr, *m_force_ = nullptr; ///< the forced tokens (mapped), -1 = the step's own pick + cudaEvent_t ev_chain_ = nullptr; + cudaEvent_t ev_step_[8] = {}; ///< after each of the first n_early outputs + int steps_seen_ = 0, chain_n_ = 0, chain_early_ = 0; + int32_t chain_tok_[8] = {}; + float chain_prob_[8] = {}; int64_t n_vocab_ = 0; uint64_t vram_ = 0; cudaStream_t cs_ = nullptr; diff --git a/include/strata/core/progress.hpp b/include/strata/core/progress.hpp index 46c93783cb7..0d982c1d8c2 100644 --- a/include/strata/core/progress.hpp +++ b/include/strata/core/progress.hpp @@ -39,6 +39,8 @@ inline std::atomic& release_gpu_fn() { static std::atomic f{null inline void release_gpu_waits(std::FILE* f) { if (auto fn = release_gpu_fn().load()) fn(f); } +/// --pipeline-windows: the pipelined loop's windows and the drafter's chain (set while that loop runs) +inline std::atomic& diag_pipeline_fn() { static std::atomic f{nullptr}; return f; } inline Progress& progress() { static Progress p; diff --git a/include/strata/core/verify.hpp b/include/strata/core/verify.hpp index e68f7b3a4b6..7c638ed307f 100644 --- a/include/strata/core/verify.hpp +++ b/include/strata/core/verify.hpp @@ -180,6 +180,45 @@ class Verifier { /// false with `err` when it failed. Free when nothing is pending. bool wait_commit(std::string& err); + // ---- PIPELINED WINDOWS (--pipeline-windows, a layer split on two GPUs). One conversation's windows with the + // stages overlapped: stage 0 runs window K+1 while stage 1 still runs window K. The same window as `run`, driven + // without blocking the host, so one host thread keeps a window in flight on each stage (the batch pipeline's + // pattern, batch_launch / batch_poll, for one sequence with drafts). Two verifiers per stage (one per window + // parity) share the stage's stream and each has its own hand-off. `pl_launch` stages and launches (it never + // captures: `capture_all` first, with nothing in flight), `service` serves the layers whose doorbells have rung + // and returns at once, `done` polls the window's completion, `pl_finish` reads the picks, `pl_commit_async` + // queues the commit. Nothing chains to `next_`: the caller drives every stage. + /// The compute stream to use instead of a private one (the two verifiers of one stage share it). Before `init`. + void set_stream(cudaStream_t s) { ext_stream_ = s; } + /// Every layer copies the token rows to the host (doorbell_publish), not only the layers with a routed expert + /// outside the VRAM tier by the device's residency table: with windows in flight the adaptive tier marks an + /// expert evicted on the host (the pool then computes it on the CPU) before the device table follows. Also turns + /// the device-planned layers (E-6) off. Before `init`. + void set_always_publish(bool on) { always_publish_ = on; } + cudaStream_t stream() const { return cs_; } + int device() const { return device_; } + /// Capture every window size and the commit graph now (a capture syncs the stream: never with a window in flight). + bool capture_all(std::string& err); + bool pl_launch(int T, const int32_t* tokens, int64_t pos0, std::string& err); + /// Stage a window ahead of its launch (positions, and the PLE rows from `ple_prev` = the two tokens before the + /// window as they WILL be, their pages prefetched). A later `pl_launch` of the same window (T, pos0, tokens, and + /// `ss.ple_prev` equal to `ple_prev` by then) skips the staging; anything else stages again. + bool prestage(int T, const int32_t* tokens, int64_t pos0, const int32_t ple_prev[2], std::string& err); + /// 1: every layer served; 0: the GPU has not reached the next layer yet; -1: an error (`err`). + int service(PoolMultiFn pool, void* user, std::string& err); + bool in_flight() const { return fl_active_; } + /// The window's graph (and its profile copy) completed; false while it runs. An error sets `err`. + bool done(std::string& err); + /// After `done`: the profile, the last stage's host sampling and picks (`out` may be null on an earlier stage). + bool pl_finish(int32_t* out, std::string& err); + /// The commit without a host sync; `ss.ple_prev` advances now (host side). A second call for the same window + /// (after its state was restored) replays it with another count. + bool pl_commit_async(int n_keep, std::string& err); + /// Fold another verifier's counters and GPU profile into this one's (the two verifiers of one stage report once). + void absorb_stats(Verifier& o); + /// The watchdog's line for a pipelined verifier: in flight, layers served, the GPU's ring and flags, its events. + void diag_pipelined(std::FILE* f, const char* name) const; + /// Measurement hook (STRATA_LOGPOS): after run(), write one line per row t of the last window's head - /// "pos target logprob top top_logprob hit extra_logprob target_logprob_without_extra" - where row t is the /// distribution at pos0 + t, targets[t] is the token at pos0 + t + 1, extra_logprob is the log-probability of @@ -242,6 +281,19 @@ class Verifier { int64_t last_pos_b_[8] = {}; bool capture_batch(const int* rows, int S, int hbase, std::string& err); void collect_profile(); ///< STRATA_VERIFY_PROFILE: add the last window's stamps to prof_sum_ + void accumulate_profile(const unsigned long long* stamps); ///< one window's stamps (host copy) into prof_sum_ + // pipelined windows (pl_launch ...) + cudaStream_t ext_stream_ = nullptr; ///< set_stream: the stage's shared stream (not destroyed here) + bool always_publish_ = false; + void pl_stage(int T, const int32_t* tokens, int64_t pos0, const int32_t ple_prev[2]); + cudaEvent_t ev_done_ = nullptr, ev_commit_ = nullptr; + unsigned long long* prof_pin_ = nullptr; ///< pinned host copy of the stamps (pipelined windows) + bool fl_active_ = false, fl_prof_ = false, fl_ple_ = false, commit_live_ = false, pl_prestaged_ = false; + int fl_T_ = 0; + int64_t fl_k_ = 0, fl_total_ = 0; + double fl_since_ms_ = 0, fl_flush_ms_ = 0, fl_launch_ms_ = 0; + int32_t pl_prev_[2] = {-1, -1}; + std::vector pl_ple_rows_; ///< the window's PLE rows (T x PLE_N_HEADS), gathered when layer 0 is served bool capture_commit_batch(const int* rows, int S, int hbase, std::string& err); bool stage_batch(const int* rows, int S, int hbase, const int32_t* tokens, const int64_t* pos, std::string& err); strata::kernels::SamplerParams sampling_ = [] { diff --git a/include/strata/kernels/verify_kernels.hpp b/include/strata/kernels/verify_kernels.hpp index 50d951a6966..c42d119096d 100644 --- a/include/strata/kernels/verify_kernels.hpp +++ b/include/strata/kernels/verify_kernels.hpp @@ -83,6 +83,9 @@ void ident_hits(const int32_t* ids, int n, int32_t* slot, int32_t* dst, int32_t* void mtp_select(const float* R_src, int64_t R_stride, const int32_t* ids, const int32_t* row_dev, float* R_dst, int32_t* tok_dst, int32_t* out, int j, void* stream, const float* probs = nullptr, float* out_p = nullptr); +/// --pipeline-windows, the drafter's chain teacher forced: `*tok = force[j]` when force[j] >= 0 (`force` is mapped +/// host memory, read when the kernel runs), else `*tok` is left as it is. +void force_token(int32_t* tok, const int32_t* force, int j, void* stream); /// dst row i = src row ids[i] (row_bytes each, multiple of 16), for n rows. void gather_rows(const uint8_t* src, int64_t row_bytes, const int32_t* ids, int64_t n, uint8_t* dst, void* stream); /// ids[t] = table[ids[t]] for n entries (a subset index back to a token id). diff --git a/src/core/mtp.cpp b/src/core/mtp.cpp index e9953e30025..9ab1349ae2a 100644 --- a/src/core/mtp.cpp +++ b/src/core/mtp.cpp @@ -130,7 +130,9 @@ MtpDrafter::~MtpDrafter() { if (head_logits_) cudaFree(head_logits_); if (dhead_) cudaFree(dhead_); if (dvocab_) cudaFree(dvocab_); - void* hosts[] = {h_tok_, h_step_, h_pos_, h_row_, h_out_, h_prob_}; + if (ev_chain_) cudaEventDestroy(ev_chain_); + for (cudaEvent_t e : ev_step_) if (e) cudaEventDestroy(e); + void* hosts[] = {h_tok_, h_step_, h_pos_, h_row_, h_out_, h_prob_, h_force_}; for (void* h : hosts) if (h) cudaFreeHost(h); } @@ -256,8 +258,10 @@ bool MtpDrafter::load(const std::string& rt_dir, const ModelGeometry& g, Session mapped(R2 * NH * 4 + 64, (void**) &h_pos_, (void**) &m_pos_) && mapped(64, (void**) &h_row_, (void**) &m_row_) && mapped(T * 4 + 64, (void**) &h_out_, (void**) &m_out_) && - mapped(T * 4 + 64, (void**) &h_prob_, (void**) &m_prob_); + mapped(T * 4 + 64, (void**) &h_prob_, (void**) &m_prob_) && + (!force_on_ || mapped(64, (void**) &h_force_, (void**) &m_force_)); if (!ok) { err = "mtp: mapped staging failed"; return false; } + if (h_force_ != nullptr) for (int j = 0; j < 16; ++j) h_force_[j] = -1; auto carve = [&](Bump& b) { tok_ = b.take(T); step_ = b.take(R2 * 4); pos_ = b.take(R2 * NH); row_ = b.take(4); ident_ = b.take(T * (uint64_t) cap_); @@ -300,7 +304,17 @@ bool MtpDrafter::load(const std::string& rt_dir, const ModelGeometry& g, Session for (int64_t i = 0; i < cap_; ++i) id[(size_t) (t * (uint64_t) cap_ + (uint64_t) i)] = (int32_t) i; cudaMemcpy(ident_, id.data(), id.size() * 4, cudaMemcpyHostToDevice); } - if (cudaStreamCreateWithFlags(&cs_, cudaStreamNonBlocking) != cudaSuccess) { err = "mtp: stream"; return false; } + // --pipeline-windows 2 (the forcing graphs): the chain shares its card with stage 1's windows and the next launch + // on stage 0 waits for it, so its stream gets the highest priority there (STRATA_MTP_PRIORITY=0: the default) + bool prio = false; +#if !defined(STRATA_USE_HIP) // (HIP: the default priority) + static const bool hi_prio = [] { const char* v = std::getenv("STRATA_MTP_PRIORITY"); return v == nullptr || std::atoi(v) != 0; }(); + int prio_lo = 0, prio_hi = 0; + if (force_on_ && hi_prio && cudaDeviceGetStreamPriorityRange(&prio_lo, &prio_hi) == cudaSuccess) + prio = cudaStreamCreateWithPriority(&cs_, cudaStreamNonBlocking, prio_hi) == cudaSuccess; + cudaGetLastError(); +#endif + if (!prio && cudaStreamCreateWithFlags(&cs_, cudaStreamNonBlocking) != cudaSuccess) { err = "mtp: stream"; return false; } const double files_s = std::chrono::duration(std::chrono::steady_clock::now() - t_files).count(); std::fprintf(stderr, "strata mtp: draft layer loaded, %.0f MiB of VRAM (experts %.0f, dense %.0f), files read in %.2f s (%.0f MiB/s)\n", (double) vram_ / 1048576.0, (double) g.n_expert * strata::kernels::cpu::BLOB / 1048576.0, @@ -717,6 +731,7 @@ bool MtpDrafter::capture_round(int T, bool coupled, std::string& err) { coupled_rec_ = false; } if (ok) mtp_select(R_, HCN, out_ids_, row_ + 1, Rin_, tok_, m_out_, 0, cs_, probs_, m_prob_); + if (ok && force_on_ && !coupled) force_token(tok_, m_force_, 0, cs_); // chain_launch: step 1's input return finish_capture(cs_, ok, exec, coupled ? "round (coupled)" : "round", err); } @@ -736,6 +751,7 @@ bool MtpDrafter::capture_step(int j, bool coupled, std::string& err) { bool ok = record_forward(1, row, cs_, err); coupled_rec_ = false; if (ok) mtp_select(R_, HCN, out_ids_, row_ + 1, Rin_, tok_, m_out_, j, cs_, probs_, m_prob_); + if (ok && force_on_ && !coupled) force_token(tok_, m_force_, j, cs_); // chain_launch: step j+1's input return finish_capture(cs_, ok, exec, coupled ? "step (coupled)" : "step", err); } @@ -749,7 +765,26 @@ void MtpDrafter::kv_restore(int64_t upto) { cudaStreamSynchronize(cs_); } -bool MtpDrafter::prefill(const float* R_rows, const int32_t* next_tokens, int64_t n, int64_t cell0, std::string& err) { +bool MtpDrafter::prepare_prefill(std::string& err) { + const OnDevice on_device(device_); + const int64_t per_row = 1 + 4 + g_->n_head; + if (pf_cap_ < (int64_t) max_t_ * per_row) { + if (pf_dev_) cudaFree(pf_dev_); + pf_dev_ = nullptr; + pf_cap_ = 0; + if (cudaMalloc((void**) &pf_dev_, (size_t) (max_t_ * per_row) * sizeof(int32_t)) != cudaSuccess) { + err = "mtp prefill: the input records do not fit"; + return false; + } + pf_cap_ = max_t_ * per_row; + } + for (int T = 1; T <= max_t_; ++T) + if (!capture_prefill_dev(T, err)) return false; // (STRATA_MTP_PREFILL_SYNC's loop syncs anyway) + return true; +} + +bool MtpDrafter::prefill(const float* R_rows, const int32_t* next_tokens, int64_t n, int64_t cell0, std::string& err, + bool sync) { const OnDevice on_device(device_); const Clock::time_point t0 = Clock::now(); const int64_t HCN = g_->hc * g_->n_embd; @@ -819,6 +854,11 @@ bool MtpDrafter::prefill(const float* R_rows, const int32_t* next_tokens, int64_ return false; } } + if (!sync) { // the caller orders on stream(); the records are staged (a pageable copy returns once taken) + (void) cudaStreamQuery(cs_); + ms_prefill += ms_since(t0); + return true; + } if (cudaStreamSynchronize(cs_) != cudaSuccess) { err = std::string("mtp prefill: ") + cudaGetErrorString(cudaGetLastError()); return false; @@ -855,6 +895,7 @@ bool MtpDrafter::draft(int T, const int32_t* tokens, int64_t p, int a, int32_t* float* probs, float min_p, int* n_drafts) { const OnDevice on_device(device_); if (T < 1 || T > max_t_ || a < 0 || a >= T) { err = "mtp: draft arguments out of range"; return false; } + if (chain_live_) { err = "mtp: a pipelined chain is still in flight"; return false; } const bool cp = coupled_active_; // coupled draft sampling for this request: its own graphs if (!capture_round(T, cp, err)) return false; const int max_steps = std::min(max_t_ - 1, max_drafts_); @@ -879,7 +920,9 @@ bool MtpDrafter::draft(int T, const int32_t* tokens, int64_t p, int a, int32_t* for (int j = 0; j < max_steps; ++j) ((volatile int32_t*) h_out_)[j] = -1; h_row_[0] = a; h_row_[1] = 0; + if (h_force_ != nullptr) for (int j = 0; j < 16; ++j) h_force_[j] = -1; // the forcing kernels: no-ops here std::atomic_thread_fence(std::memory_order_seq_cst); + if (!stage_source_R(T, err)) return false; SessionState* const pss = ple_ss_ ? ple_ss_ : ss_; const bool do_ple = pss != nullptr && pss->ple.ready() && pss->ple.table != nullptr; @@ -975,6 +1018,117 @@ bool MtpDrafter::draft(int T, const int32_t* tokens, int64_t p, int a, int32_t* return true; } +bool MtpDrafter::stage_source_R(int T, std::string& err) { + if (src_R_ == nullptr || src_R_ == window_R_) return true; + if (cudaMemcpyAsync((void*) window_R_, src_R_, (size_t) T * (size_t) (g_->hc * g_->n_embd) * sizeof(float), + cudaMemcpyDeviceToDevice, cs_) != cudaSuccess) { + err = "mtp: staging the window's residual rows failed"; + return false; + } + return true; +} + +bool MtpDrafter::prepare_chain(std::string& err) { + const OnDevice on_device(device_); + if (!force_on_ || h_force_ == nullptr) { err = "mtp: chains need set_force_capture before load"; return false; } + for (int T = 1; T <= max_t_; ++T) + if (!capture_round(T, false, err)) return false; + for (int j = 1; j <= max_t_ - 2; ++j) + if (!capture_step(j, false, err)) return false; + if (ev_chain_ == nullptr && cudaEventCreateWithFlags(&ev_chain_, cudaEventDisableTiming) != cudaSuccess) { + err = "mtp: event creation failed"; + return false; + } + for (cudaEvent_t& e : ev_step_) + if (e == nullptr && cudaEventCreateWithFlags(&e, cudaEventDisableTiming) != cudaSuccess) { + err = "mtp: event creation failed"; + return false; + } + return true; +} + +bool MtpDrafter::chain_launch(int T, const int32_t* tokens, int64_t p, int a, const int32_t* force, int n_force, + int n_out, int n_early, std::string& err) { + const OnDevice on_device(device_); + if (chain_live_) { err = "mtp: a chain is already in flight"; return false; } + if (!force_on_ || ev_chain_ == nullptr) { err = "mtp: chains need prepare_chain"; return false; } + if (T < 1 || T > max_t_ || a < 0 || a >= T || n_out < 1 || n_out > max_t_ - 1 || n_force < 0 || n_force > n_out || + n_force > 16) { + err = "mtp: chain arguments out of range"; + return false; + } + if (round_exec_[T] == nullptr) { err = "mtp: chain graphs not prepared"; return false; } + for (int j = 1; j < n_out; ++j) + if (step_exec_[j] == nullptr) { err = "mtp: chain graphs not prepared"; return false; } + const Clock::time_point t0 = Clock::now(); + const int64_t NH = g_->n_head; + auto put = [&](int row, int64_t cell) { + h_step_[row * 4 + 0] = (int32_t) cell; + h_step_[row * 4 + 1] = (int32_t) (cell + 1); + h_step_[row * 4 + 2] = (int32_t) ((cell + 1) / 4); + h_step_[row * 4 + 3] = (int32_t) (cell + 1); + for (int64_t h = 0; h < NH; ++h) h_pos_[row * NH + h] = (int32_t) cell; + }; + for (int t = 0; t < T; ++t) { + h_tok_[t] = tokens[t]; + put(t, p + t); + } + put(2 * max_t_ - 1, p + a); // output 0's cell + for (int j = 1; j < n_out; ++j) put(max_t_ + j - 1, p + a + j); // every step's cell, staged at once + for (int j = 0; j < 16; ++j) h_force_[j] = j < n_force ? force[j] : -1; + h_row_[0] = a; + h_row_[1] = 0; + std::atomic_thread_fence(std::memory_order_seq_cst); + if (!stage_source_R(T, err)) return false; + chain_early_ = std::max(0, std::min(n_early, n_out)); + bool ok = cudaGraphLaunch(round_exec_[T], cs_) == cudaSuccess; + if (ok && chain_early_ >= 1) ok = cudaEventRecord(ev_step_[0], cs_) == cudaSuccess; + for (int j = 1; ok && j < n_out; ++j) { + ok = cudaGraphLaunch(step_exec_[j], cs_) == cudaSuccess; + if (ok && j < chain_early_) ok = cudaEventRecord(ev_step_[j], cs_) == cudaSuccess; + } + if (!ok || cudaEventRecord(ev_chain_, cs_) != cudaSuccess) { + err = std::string("mtp chain: ") + cudaGetErrorString(cudaGetLastError()); + return false; + } + (void) cudaStreamQuery(cs_); // WDDM: submit now + steps_seen_ = 0; + chain_live_ = true; + chain_n_ = n_out; + ms_draft += ms_since(t0); + ++rounds; + return true; +} + +int MtpDrafter::chain_outputs_ready(std::string& err) { + if (!chain_live_) return chain_n_; + const OnDevice on_device(device_); + while (steps_seen_ < chain_early_) { + const cudaError_t q = cudaEventQuery(ev_step_[steps_seen_]); + if (q == cudaErrorNotReady) break; + if (q != cudaSuccess) { err = std::string("mtp chain: ") + cudaGetErrorString(q); return -1; } + chain_tok_[steps_seen_] = ((volatile int32_t*) h_out_)[steps_seen_]; + chain_prob_[steps_seen_] = ((volatile float*) h_prob_)[steps_seen_]; + ++steps_seen_; + } + return steps_seen_; +} + +int MtpDrafter::chain_poll(std::string& err) { + if (!chain_live_) return 1; + const OnDevice on_device(device_); + const cudaError_t q = cudaEventQuery(ev_chain_); + if (q == cudaErrorNotReady) return 0; + chain_live_ = false; + if (q != cudaSuccess) { err = std::string("mtp chain: ") + cudaGetErrorString(q); return -1; } + for (int j = 0; j < chain_n_; ++j) { + chain_tok_[j] = ((volatile int32_t*) h_out_)[j]; + chain_prob_[j] = ((volatile float*) h_prob_)[j]; + } + for (int j = chain_n_; j < 8; ++j) { chain_tok_[j] = 0; chain_prob_[j] = 0.0f; } + return 1; +} + bool MtpDrafter::draft_first(int T, const float* R_row, int32_t token, int64_t cell, int32_t* drafts, std::string& err, float* probs, float min_p, int* n_drafts) { const OnDevice on_device(device_); diff --git a/src/core/verify.cpp b/src/core/verify.cpp index dd5362bacd9..bcc35ff2521 100644 --- a/src/core/verify.cpp +++ b/src/core/verify.cpp @@ -285,10 +285,13 @@ Verifier::~Verifier() { if (kv.second) cudaGraphExecDestroy(kv.second); if (arena_b_) cudaFree(arena_b_); if (h_commitb_) cudaFreeHost(h_commitb_); - if (cs_) cudaStreamDestroy(cs_); + if (cs_ && cs_ != ext_stream_) cudaStreamDestroy(cs_); // set_stream: the stage's stream, shared, not ours if (sh_cs_) cudaStreamDestroy(sh_cs_); if (copy_) { cudaStreamSynchronize(copy_); cudaStreamDestroy(copy_); } if (commit_done_) cudaEventDestroy(commit_done_); + if (ev_done_) cudaEventDestroy(ev_done_); + if (ev_commit_) cudaEventDestroy(ev_commit_); + if (prof_pin_) cudaFreeHost(prof_pin_); if (ev_fork_) cudaEventDestroy(ev_fork_); if (ev_join_) cudaEventDestroy(ev_join_); if (arena_) cudaFree(arena_); @@ -477,8 +480,9 @@ bool Verifier::init(const WeightTable& wt, const ModelGeometry& g, SessionState& err = "verify: copy stream create failed"; return false; } - if (cudaStreamCreateWithFlags(&cs_, cudaStreamNonBlocking) != cudaSuccess || - cudaStreamCreateWithFlags(&sh_cs_, cudaStreamNonBlocking) != cudaSuccess) { + if (ext_stream_ != nullptr) cs_ = ext_stream_; // set_stream (pipelined windows): the stage's shared stream + else if (cudaStreamCreateWithFlags(&cs_, cudaStreamNonBlocking) != cudaSuccess) cs_ = nullptr; + if (cs_ == nullptr || cudaStreamCreateWithFlags(&sh_cs_, cudaStreamNonBlocking) != cudaSuccess) { err = "verify: stream create failed"; return false; } @@ -511,7 +515,8 @@ bool Verifier::init(const WeightTable& wt, const ModelGeometry& g, SessionState& // exact, but neutral on RIBPC 1-2 GPUs: off by default) { const char* v = std::getenv("STRATA_VERIFY_DEVICE_PLAN"); - device_plan_ = !all_resident_ && (v != nullptr && std::atoi(v) != 0); + // (not with set_always_publish: the device table may lag the host's while windows are in flight) + device_plan_ = !all_resident_ && !always_publish_ && (v != nullptr && std::atoi(v) != 0); } if (all_resident_ || device_plan_) { bool ok2 = true; @@ -950,7 +955,9 @@ bool Verifier::record_window(int T, cudaStream_t cs, std::string& err) { m_ids_ + tb * K, m_w_ + tb * K, m_seq_, (uint32_t) ((l - lb_) * G + grp + 1), cs); else #endif - if (remote_opt_) // #578: the helper GPUs reduce with the routing weights - publish them + // #578: the helper GPUs reduce with the routing weights - publish them; set_always_publish: the rows + // whatever the device's residency table says (pipelined windows) + if (remote_opt_ || always_publish_) doorbell_publish(xm, ids_ + tb * K, w_ + tb * K, (int64_t) n * N, (int64_t) n * K, m_x_ + tb * N, m_ids_ + tb * K, m_w_ + tb * K, m_seq_, cs); else { @@ -1354,10 +1361,14 @@ void Verifier::stage_inputs(int T, const int32_t* tokens, int64_t pos0) { } void Verifier::collect_profile() { - const ModelGeometry& g = *g_; cudaMemcpy(prof_h_.data(), prof_, prof_h_.size() * 8, cudaMemcpyDeviceToHost); + accumulate_profile(prof_h_.data()); +} + +void Verifier::accumulate_profile(const unsigned long long* stamps) { + const ModelGeometry& g = *g_; const int64_t L = g.n_layers; - auto at = [&](int64_t l, int i) { return prof_h_[(size_t) (l * kProfPer + i)]; }; + auto at = [&](int64_t l, int i) { return stamps[(size_t) (l * kProfPer + i)]; }; // D8: the derived columns below read stamps the hc-read kernels write themselves or the next // layer's first. A slot no kernel stamped is 0 and its unsigned difference wrapped to ~1e19 ns - // which is why (gap)/head/hc0 printed ~1e14 ms per window. A missing or out-of-order stamp now @@ -1365,7 +1376,8 @@ void Verifier::collect_profile() { const auto gap = [](unsigned long long to, unsigned long long from) { return (from != 0 && to != 0 && to >= from) ? (double) (to - from) : 0.0; }; - for (int64_t l = 0; l < L; ++l) { + // only this stage's layers [lb_, le_) are ever stamped (a layer split); the head only on the last stage + for (int64_t l = lb_; l < le_; ++l) { const int kind = is_qsa_layer(g, l) ? 1 : 0; unsigned long long prev = at(l, 0); for (int i = 1; i <= 24; ++i) { @@ -1374,7 +1386,7 @@ void Verifier::collect_profile() { prof_sum_[kind][i] += (double) (x - prev); prev = x; } - if (l + 1 < L) prof_sum_[kind][25] += gap(at(l + 1, 0), at(l, 24)); + if (l + 1 < le_) prof_sum_[kind][25] += gap(at(l + 1, 0), at(l, 24)); const double dn = gap(at(l, 27), at(l, 0)), dd = gap(at(l, 28), at(l, 27)), du = gap(at(l, 1), at(l, 28)); if (dn > 0 && dd > 0 && du > 0) { // the split exists: show it split, not twice prof_sum_[kind][27] += dn; // hc-read0: norm @@ -1383,7 +1395,7 @@ void Verifier::collect_profile() { prof_sum_[kind][1] -= gap(at(l, 1), at(l, 0)); // (hc-read0 shown split) } } - prof_sum_[0][26] += gap(at(L, 1), at(L, 0)); + if (le_ == L) prof_sum_[0][26] += gap(at(L, 1), at(L, 0)); ++prof_windows_; } @@ -2155,4 +2167,289 @@ bool Verifier::copy_logits(int t, float* host) const { cudaMemcpyDeviceToHost) == cudaSuccess; } + +// ================================ PIPELINED WINDOWS (see verify.hpp) ================================ +// +// The window is run()'s, step for step: the same graph, the same staging, the same per-layer service (the PLE rows +// gathered while layer 0 is served, the zero-doorbell graph's one flag), the same commit graph. Only the waits are +// split up: the host polls instead of spinning, so it can serve the other stage's window in between. + +namespace { +double now_ms() { return std::chrono::duration(Clock::now().time_since_epoch()).count(); } +} // namespace + +void Verifier::diag_pipelined(std::FILE* f, const char* name) const { + auto rd = [](const uint32_t* p) { return p ? *(const volatile uint32_t*) p : 0u; }; + auto ev = [](cudaEvent_t e) { + if (e == nullptr) return "none"; + const cudaError_t q = cudaEventQuery(e); + return q == cudaSuccess ? "done" : q == cudaErrorNotReady ? "PENDING" : "error"; + }; + std::fprintf(f, " %s: %s T=%d pos %lld served %lld/%lld; GPU rang %u, flags served %u A %u B %u; window event %s, " + "commit event %s (commit launched %d)\n", name, fl_active_ ? "IN FLIGHT" : "idle", last_t_, + (long long) last_pos0_, (long long) fl_k_, (long long) fl_total_, rd(h_seq_), rd(h_flag_), rd(h_flagA_), + rd(h_flagB_), ev(ev_done_), ev(ev_commit_), (int) commit_live_); +} + +bool Verifier::capture_all(std::string& err) { + const OnDevice on_device(device_); + if (g_ == nullptr) { err = "verify: capture_all before init"; return false; } + if (remote_opt_ != nullptr) { err = "verify: pipelined windows do not serve --remote-expert-opt"; return false; } + if (released_.load()) { err = "verify: an earlier window never finished on the GPU (#267); restart the engine"; return false; } + for (int T = 1; T <= max_t_; ++T) + if (!capture(T, err)) return false; + if (!capture_commit(err)) return false; + if ((ev_done_ == nullptr && cudaEventCreateWithFlags(&ev_done_, cudaEventDisableTiming) != cudaSuccess) || + (ev_commit_ == nullptr && cudaEventCreateWithFlags(&ev_commit_, cudaEventDisableTiming) != cudaSuccess)) { + err = "verify: event create failed"; + return false; + } + if (prof_on_ && prof_pin_ == nullptr && + cudaHostAlloc((void**) &prof_pin_, prof_h_.size() * 8, cudaHostAllocDefault) != cudaSuccess) { + prof_pin_ = nullptr; // the pipelined windows go unprofiled + cudaGetLastError(); + } + pl_ple_rows_.assign((size_t) strata::kernels::kVerifyMaxT * strata::kernels::PLE_N_HEADS, 0u); + return true; +} + +// The host staging of a pipelined window: run()'s (stage_inputs), and its PLE rows from `ple_prev` with their pages +// prefetched (they are gathered into the mapped rows when layer 0 is served, as run() does). +void Verifier::pl_stage(int T, const int32_t* tokens, int64_t pos0, const int32_t ple_prev[2]) { + using namespace strata::kernels; + stage_inputs(T, tokens, pos0); + staged_ = false; // a later run() stages its own window + SessionState& ss = *ss_; + fl_ple_ = ss.ple.ready() && ple_stage(); + if (fl_ple_) { + int32_t prev[2] = {ple_prev[0], ple_prev[1]}; + for (int t = 0; t < T; ++t) { + ngram_rows(&tokens[t], prev, 1, ss.ple.consts, pl_ple_rows_.data() + t * PLE_N_HEADS); + prev[0] = prev[1]; + prev[1] = tokens[t]; + ss.ple.table->prefetch_rows(pl_ple_rows_.data() + t * PLE_N_HEADS); + } + } + pl_prev_[0] = ple_prev[0]; + pl_prev_[1] = ple_prev[1]; +} + +bool Verifier::prestage(int T, const int32_t* tokens, int64_t pos0, const int32_t ple_prev[2], std::string& err) { + if (fl_active_) { err = "verify: a window is in flight on this verifier"; return false; } + if (T < 1 || T > max_t_) { err = "verify: window size out of range"; return false; } + if (pl_ple_rows_.empty()) { err = "verify: pipelined window not prepared (capture_all)"; return false; } + const Clock::time_point t0 = Clock::now(); + pl_stage(T, tokens, pos0, ple_prev); + pl_prestaged_ = true; + ms_host += ms_since(t0); + return true; +} + +bool Verifier::pl_launch(int T, const int32_t* tokens, int64_t pos0, std::string& err) { + const OnDevice on_device(device_); + if (fl_active_) { err = "verify: a window is already in flight on this verifier"; return false; } + if (released_.load()) { err = "verify: an earlier window never finished on the GPU (#267); restart the engine"; return false; } + if (T < 1 || T > max_t_) { err = "verify: window size out of range"; return false; } + SessionState& ss = *ss_; + if (pos0 + T > ss.qsa_states[ss.qsa_primary()].max_cells) { err = "verify: the window runs past the context"; return false; } + if (exec_[T] == nullptr || commit_exec_ == nullptr || ev_done_ == nullptr || pl_ple_rows_.empty()) { + err = "verify: pipelined window not prepared (capture_all)"; + return false; + } + const Clock::time_point t0 = Clock::now(); + last_batch_ = false; + bool staged = pl_prestaged_ && last_t_ == T && last_pos0_ == pos0; + for (int t = 0; staged && t < T; ++t) staged = last_tokens_[t] == tokens[t]; + if (staged && ss.ple.ready() && ple_stage()) + staged = pl_prev_[0] == ss.ple_prev[0] && pl_prev_[1] == ss.ple_prev[1]; + pl_prestaged_ = false; + if (!staged) pl_stage(T, tokens, pos0, ss.ple_prev); + const int G = groups_[T] > 0 ? groups_[T] : 1; + fl_T_ = T; + fl_k_ = 0; + fl_total_ = (le_ - lb_) * G; + fl_prof_ = prof_on_ && G == 1 && prof_pin_ != nullptr; + if (trace_h_ != nullptr) std::memset(trace_h_, 0, trace_n_ * 8); // #649: this window's breadcrumbs only + std::atomic_thread_fence(std::memory_order_seq_cst); + trace_ev("WINDOW (pipelined)", -1, -1, pos0 * 16 + T); + ms_host += ms_since(t0); + const cudaError_t le = cudaGraphLaunch(exec_[T], cs_); + trace_ev("LAUNCHED", -1, -1, (int64_t) le); + if (le != cudaSuccess) { err = std::string("verify: launch: ") + cudaGetErrorString(le); return false; } + if (fl_prof_) cudaMemcpyAsync(prof_pin_, prof_, prof_h_.size() * 8, cudaMemcpyDeviceToHost, cs_); + if (cudaEventRecord(ev_done_, cs_) != cudaSuccess) { err = "verify: event record failed"; return false; } + (void) cudaStreamQuery(cs_); // WDDM: submit now + fl_active_ = true; + fl_since_ms_ = fl_flush_ms_ = fl_launch_ms_ = now_ms(); + return true; +} + +int Verifier::service(PoolMultiFn pool, void* user, std::string& err) { + if (!fl_active_) return 1; + if (fl_k_ >= fl_total_) return 1; + const OnDevice on_device(device_); + const ModelGeometry& g = *g_; + SessionState& ss = *ss_; + const int T = fl_T_; + // the PLE rows into the mapped staging (layer 1's pre() copies them once flag 1 is up), as run() does at k == 0 + auto gather_ple = [&]() -> bool { + if (!fl_ple_) return true; + const Clock::time_point tp = Clock::now(); + if (!ss.ple.table->gather_batch(pl_ple_rows_.data(), (size_t) T, h_ple_, err)) return false; + std::atomic_thread_fence(std::memory_order_seq_cst); + _mm_sfence(); + ms_host += ms_since(tp); + return true; + }; + if (all_resident_) { // the zero-doorbell graph never rings: it waits only for flag 1 (stage 0's PLE rows) + if (!gather_ple()) return -1; + *(volatile uint32_t*) h_flag_ = 1; + fl_k_ = fl_total_; + progress_tick(); + return 1; + } + const int G = groups_[T] > 0 ? groups_[T] : 1; + const int gtb[2] = {0, (T + 1) / 2}, gte[2] = {G == 2 ? (T + 1) / 2 : T, T}; + while (fl_k_ < fl_total_) { + const int64_t l = lb_ + fl_k_ / G; + const uint32_t want = (uint32_t) (fl_k_ + 1); + if (*(volatile uint32_t*) h_seq_ < want) { + const double now = now_ms(); + if (now - fl_flush_ms_ > 2.0) { // flush WDDM and notice a dead graph, as run() does + fl_flush_ms_ = now; + const cudaError_t q = cudaEventQuery(ev_done_); + if (q != cudaErrorNotReady && *(volatile uint32_t*) h_seq_ < want) { + trace_ev("NEVER-RANG", fl_k_, l, (int64_t) q); + err = "verify: layer " + std::to_string(l) + " never rang (" + + (q == cudaSuccess ? std::string("graph finished") : std::string(cudaGetErrorString(q))) + ")"; + return -1; + } + (void) cudaStreamQuery(cs_); + } + if (now - fl_since_ms_ > 20000.0) { // #267: no spin kernel may outlive the engine + trace_ev("TIMEOUT", fl_k_, l, 0); + err = "verify: timed out at layer " + std::to_string(l) + released_note(release_gpu_waits(5000)); + return -1; + } + return 0; + } + const Clock::time_point b = Clock::now(); + ms_wait += now_ms() - fl_since_ms_; + const int grp = (int) (fl_k_ % G); + cur_layer_ = want - 1; + set_plan_slot(grp); + const int tb = gtb[grp], n = gte[grp] - gtb[grp]; + progress_at("verify window (pipelined): the CPU experts of layer", l); + if (pool != nullptr) + pool(user, h_x_ + (size_t) tb * g.n_embd, h_ids_ + (size_t) tb * ss.k, n, ss.k, + h_ymiss_ + (size_t) tb * ss.k * g.n_embd, l); + if (g_trace) trace_ev(*(volatile uint32_t*) h_flagA_ == want ? "SERVED" : "SERVED-NO-PLAN-YET", fl_k_, l, + (int64_t) ms_since(b)); + progress_tick(); + std::atomic_thread_fence(std::memory_order_seq_cst); + _mm_sfence(); + if (*(volatile uint32_t*) h_flagA_ != want) { // the pool did not publish a plan: an empty one + sink_.counts[0] = 0; + sink_.counts[1] = 0; + sink_.counts[2] = 0; + sink_.start[0] = 0; + sink_.start2[0] = 0; + std::atomic_thread_fence(std::memory_order_seq_cst); + *(volatile uint32_t*) h_flagA_ = want; + raise_flag(h_flagB_, want); + } + if (fl_k_ == 0 && !gather_ple()) return -1; + *(volatile uint32_t*) h_flag_ = want; + ++fl_k_; + ms_pool += ms_since(b); + fl_since_ms_ = fl_flush_ms_ = now_ms(); + } + return 1; +} + +bool Verifier::done(std::string& err) { + if (!fl_active_ || fl_k_ < fl_total_) return false; + const OnDevice on_device(device_); + const cudaError_t q = cudaEventQuery(ev_done_); + if (q == cudaErrorNotReady) { + const double now = now_ms(); + if (now - fl_flush_ms_ > 2.0) { fl_flush_ms_ = now; (void) cudaStreamQuery(cs_); } + if (now - fl_since_ms_ > 20000.0) { // every layer served, and the graph still runs: #267 as above + err = "verify: the window never finished" + released_note(release_gpu_waits(5000)); + } + return false; + } + if (q != cudaSuccess) { err = std::string("verify: ") + cudaGetErrorString(q); return false; } + if (copy_used_) { // no host function of this window may raise flag B in the next one + if (cudaStreamQuery(copy_) == cudaErrorNotReady) return false; + copy_used_ = false; + } + return true; +} + +bool Verifier::pl_finish(int32_t* out, std::string& err) { + using namespace strata::kernels; + const OnDevice on_device(device_); + const ModelGeometry& g = *g_; + fl_active_ = false; + if (fl_prof_) accumulate_profile(prof_pin_); + ++windows; + trace_ev("DONE (pipelined)", -1, -1, (int64_t) (now_ms() - fl_launch_ms_)); + if (le_ < g.n_layers) return true; // an earlier stage: the hand-off is written + const int T = fl_T_; + const bool sampled = !sampling_.greedy && sampling_.temperature > 0.0f; + if (head_sampling_ && (sampled || hist_d_ != nullptr)) { // run()'s host-side sampling, Philox(seed, pos0 + t) + SamplerParams sp = sampling_; + sp.counter = (uint64_t) last_pos0_; + sample_tokens(head_logits_, T, (int) n_vocab_, hist_d_, hist_len_, sp, m_out_, cs_); + if (cudaStreamSynchronize(cs_) != cudaSuccess) { + err = "verify: the head sampling failed"; + return false; + } + } + if (out != nullptr) + for (int t = 0; t < T; ++t) out[t] = ((volatile int32_t*) h_out_)[t]; + progress_beat(); + return true; +} + +bool Verifier::pl_commit_async(int n_keep, std::string& err) { + const OnDevice on_device(device_); + if (n_keep < 1 || n_keep > last_t_) { err = "verify: commit count out of range"; return false; } + if (commit_exec_ == nullptr || ev_commit_ == nullptr) { err = "verify: pipelined commit not prepared"; return false; } + const Clock::time_point t0 = Clock::now(); + if (commit_live_) { // its graph reads the words below when it starts: the previous one has (a window ago) + cudaError_t q; + while ((q = cudaEventQuery(ev_commit_)) == cudaErrorNotReady) _mm_pause(); + if (q != cudaSuccess) { err = std::string("verify: commit: ") + cudaGetErrorString(q); return false; } + } + h_commit_[0] = n_keep; + h_commit_[1] = n_keep - 1; + for (int t = 0; t < max_t_; ++t) h_commit_[2 + t] = t < n_keep ? (int32_t) (last_pos0_ + t) : -1; + std::atomic_thread_fence(std::memory_order_seq_cst); + const cudaError_t le = cudaGraphLaunch(commit_exec_, cs_); + if (le != cudaSuccess) { err = std::string("verify: commit launch: ") + cudaGetErrorString(le); return false; } + if (cudaEventRecord(ev_commit_, cs_) != cudaSuccess) { err = "verify: event record failed"; return false; } + (void) cudaStreamQuery(cs_); + commit_live_ = true; + if (ple_stage()) + for (int t = 0; t < n_keep; ++t) { + ss_->ple_prev[0] = ss_->ple_prev[1]; + ss_->ple_prev[1] = last_tokens_[t]; + } + ms_commit += ms_since(t0); + return true; +} + +void Verifier::absorb_stats(Verifier& o) { + ms_wait += o.ms_wait; ms_pool += o.ms_pool; ms_host += o.ms_host; ms_commit += o.ms_commit; + windows += o.windows; + o.ms_wait = o.ms_pool = o.ms_host = o.ms_commit = 0; + o.windows = 0; + for (int k = 0; k < 2; ++k) + for (int i = 0; i < kProfPer; ++i) { prof_sum_[k][i] += o.prof_sum_[k][i]; o.prof_sum_[k][i] = 0; } + prof_windows_ += o.prof_windows_; + o.prof_windows_ = 0; +} + } // namespace strata::core diff --git a/src/kernels/cuda/verify_kernels.cu b/src/kernels/cuda/verify_kernels.cu index f7c48438937..5a2dbdb2e1f 100644 --- a/src/kernels/cuda/verify_kernels.cu +++ b/src/kernels/cuda/verify_kernels.cu @@ -358,6 +358,12 @@ __global__ void dense_steps_kernel(const int32_t* __restrict__ cells, int n, int steps[i * 4 + 3] = c + 1; } +// --pipeline-windows: the drafter's chain, teacher forced (see force_token) +__global__ void force_token_kernel(int32_t* tok, const int32_t* force, int j) { + const int32_t f = ((const volatile int32_t*) force)[j]; + if (f >= 0) *tok = f; +} + } // namespace void fetch_blobs(const unsigned long long* src, const int32_t* n, uint8_t* dst, int64_t blob_bytes, int cap, void* stream) { @@ -391,6 +397,11 @@ void mtp_select(const float* R_src, int64_t R_stride, const int32_t* ids, const check("mtp_select"); } +void force_token(int32_t* tok, const int32_t* force, int j, void* stream) { + force_token_kernel<<<1, 1, 0, (cudaStream_t) stream>>>(tok, force, j); + check("force_token"); +} + void gather_rows(const uint8_t* src, int64_t row_bytes, const int32_t* ids, int64_t n, uint8_t* dst, void* stream) { cudaStream_t s = (cudaStream_t) stream; if (row_bytes % 16 == 0) diff --git a/src/program/generate.cpp b/src/program/generate.cpp index 085c3d8ced6..fdb2897b1ee 100644 --- a/src/program/generate.cpp +++ b/src/program/generate.cpp @@ -81,6 +81,7 @@ #include #include #include +#include #include #include #include @@ -519,6 +520,11 @@ struct Options { /// result) goes through the verify windows, S tokens at a time, instead of the batched prompt path (0 = always /// the batched path) int64_t short_read = 64; + /// --pipeline-windows N (--serve, a layer split on exactly two GPUs; opt-in): 1 = a prompt's short reads through + /// the verify windows run with the stages overlapped (stage 0 reads window K+1 while stage 1 reads window K); + /// 2 = the decode windows too: stage 0 runs window K+1 speculatively while stage 1 verifies window K (rolled back + /// when K is not accepted as guessed). 0 = off (the stages take turns). + int pipeline_windows = 0; /// The suffix drafter (prompt lookup): when the text being written repeats an earlier stretch of the context (code /// edits, quoted input, tool-call JSON) by at least this many tokens, the window may be filled with what followed /// it there instead of the MTP's drafts, where the MTP's own first guess agrees and the draft policy expects it to @@ -623,6 +629,9 @@ void usage() { " --batch N / --slots N --serve (opt-in): up to N requests decode together in batch slots (2..8),\n" " each slot with its own session (VRAM like the main one); docs/BATCHING.md\n" " --batch-groups G --batch with a layer split: the slots in G groups pipelined through the GPUs\n" + " --pipeline-windows N --serve with a layer split on two GPUs (opt-in): one conversation's verify windows\n" + " with the GPUs overlapped - 1 = the prompt's short reads, 2 = decode as well\n" + " (stage 0 runs the next window while stage 1 verifies this one); docs/MULTI_GPU.md\n" " --trim-stage-weights an explicit --layer-split: each GPU loads only its own layers' dense weights\n" " --prompt-cache-every N --serve: also checkpoint every N fresh prompt tokens (default 16384, 0 = off)\n" " --turn-token ID --serve: the token that opens a chat turn (default 248045, <|im_start|>)\n" @@ -844,6 +853,7 @@ struct GpuStage { std::vector> profile; ///< its layers' share of the profile, hottest first int32_t* d_res = nullptr; ///< the residency table on its device strata::core::Verifier ver; + strata::core::Verifier ver_b; ///< --pipeline-windows: the odd windows' verifier strata::prefill::Prefill sp; cudaStream_t adapt_stream = nullptr; cudaEvent_t adapt_ev = nullptr; @@ -952,6 +962,31 @@ void disk_wait_threads(std::FILE* f) { } #endif +// --pipeline-windows: what the watchdog prints about the pipelined decode loop (set while it runs) +struct PlDiag { + strata::core::Verifier* v[2][2] = {}; + std::atomic a_seq{0}, a_bits{0}, b_bits{0}, d_bits{0}, chain_kind{0}, doomed{0}, ending{0}, chain_live{0}; + std::atomic iters{0}; +}; +PlDiag g_pl_diag; +void pl_diag_print(std::FILE* f) { + PlDiag& d = g_pl_diag; + auto bits = [](int b, char* o) { + std::snprintf(o, 16, "%c%c%c%c%c%c", b & 1 ? 'R' : '-', b & 2 ? 'L' : '-', b & 4 ? 'F' : '-', b & 8 ? 'C' : '-', + b & 16 ? 'S' : '-', b & 32 ? 'V' : '-'); + return (const char*) o; + }; + char ba[16], bb[16], bd[16]; + std::fprintf(f, " pipelined loop: %lld iterations; A seq %d [%s] B [%s] D [%s] (Ready Launched Finished Committed " + "S1-launched S1-done); chain kind %d live %d; doomed %d ending %d\n", d.iters.load(), d.a_seq.load(), + bits(d.a_bits.load(), ba), bits(d.b_bits.load(), bb), bits(d.d_bits.load(), bd), d.chain_kind.load(), + d.chain_live.load(), d.doomed.load(), d.ending.load()); + const char* names[2][2] = {{"stage 0 even", "stage 0 odd"}, {"stage 1 even", "stage 1 odd"}}; + for (int st = 0; st < 2; ++st) + for (int par = 0; par < 2; ++par) + if (d.v[st][par] != nullptr) d.v[st][par]->diag_pipelined(f, names[st][par]); +} + void stall_report(std::FILE* f, uint64_t layers_during) { strata::core::Progress& p = strata::core::progress(); std::fprintf(f, "strata serve: stall report (engine %s): stage \"%s\" for %lld s; %llu layers served since the " @@ -965,6 +1000,7 @@ void stall_report(std::FILE* f, uint64_t layers_during) { } if (auto fn = strata::core::diag_pool_fn().load()) fn(f); if (auto fn = strata::core::diag_verify_fn().load()) fn(f); + if (auto fn = strata::core::diag_pipeline_fn().load()) fn(f); #if !defined(_WIN32) disk_wait_threads(f); #endif @@ -1433,6 +1469,7 @@ int main(int argc, char** argv) { else if (a == "--turn-token") o.turn_token = std::atoll(next("--turn-token")); else if (a == "--tail-role-token") o.tail_role_token = std::atoll(next("--tail-role-token")); else if (a == "--short-read") o.short_read = std::max(0LL, std::atoll(next("--short-read"))); + else if (a == "--pipeline-windows") o.pipeline_windows = std::max(0, std::min(2, std::atoi(next("--pipeline-windows")))); else if (a == "--suffix-draft") o.suffix_draft = std::max(0, std::atoi(next("--suffix-draft"))); else if (a == "--mtp-max-t") o.mtp_max_t = std::max(0, std::atoi(next("--mtp-max-t"))); else if (a == "--control-vector") o.cvec_files.push_back({next("--control-vector"), 1.0f}); @@ -2552,6 +2589,25 @@ int main(int argc, char** argv) { } } strata::core::Verifier::set_commit_async(!multi_gpu); // see Verifier::set_commit_async + // --pipeline-windows (opt-in): one conversation's windows with the two stages of a layer split overlapped. Decided + // here, before any stage sizes its expert cache (the second verifier per stage and the snapshots are allocated + // after the caches, so their room is kept out of them). What it does not support turns it off, said once. + if (o.pipeline_windows > 0) { + const bool helpers = o.expert_cache_remote[0] > 0 || o.expert_cache_remote[1] > 0 || o.expert_cache_remote[2] > 0; + const char* off = !o.serve ? "it needs --serve" + : split_same ? "it needs each stage of the layer split on its own GPU" + : !multi_gpu ? "it needs a layer split on two GPUs" + : split_devs.size() != 1 ? "it needs a layer split into exactly two stages" + : o.mtp.empty() || o.spec < 2 ? "it needs the MTP drafter (--mtp, --spec)" + : o.batch != 0 ? "not with --batch slots" + : o.peer_device >= 1 ? "not with --peer-device" + : helpers ? "not with the helper caches (--expert-cache-device1..3, --remote-expert-opt)" + : nullptr; + if (off != nullptr) { + std::fprintf(stderr, "strata generate: --pipeline-windows %d is off: %s\n", o.pipeline_windows, off); + o.pipeline_windows = 0; + } + } // ---- layer split across GPUs: each later stage's own copy of the dense weights, its session and (the last) the // head, made on its device before the host arena is mapped (as the drafter below, for the same WDDM reason) std::vector> stages; @@ -2727,6 +2783,16 @@ int main(int argc, char** argv) { // the windows and - only on the stage that carries them - the drafter and the head. const int64_t kWindowMib = 96; // the verify windows; 75 MiB measured, rounded up const int64_t kDrafterMib = 1000; // the MTP drafter (839 MiB) + the head, on the last stage only + // --pipeline-windows: a second verifier on every stage (its arena alone is 70-75 MiB and its graphs come on top: + // 64 MiB left two RTX 5070 Ti under Linux short of it) and, with the decode pipelined, two copies of the first + // stage's GDN state (sized from its session where the first card's cache is sized; this is the split search's + // estimate). Allocated after the caches, so kept out of them. + const int64_t kPipeWindowMib = o.pipeline_windows > 0 ? 160 : 0; + const int64_t kPipeSnapMib = o.pipeline_windows >= 2 ? 96 : 0; // ~3 MiB per GDN layer, twice, ~15 layers + auto gdn_snapshot_bytes = [&](const strata::core::SessionState& s0) -> size_t { + return (size_t) s0.gdn_alloc * ((size_t) g.ssm_state_size * g.ssm_v_heads * g.ssm_state_size + + (size_t) g.ssm_conv_channels * (g.ssm_d_conv - 1)) * sizeof(float); + }; // #340: with the own prompt buffers chosen by the split's rule (not asked for with --no-prefill-borrow) the // boundary is searched as the borrowing configuration would (no reserve): the reserve then only makes the caches // smaller, which measured cost no decode (K=28 on 9070 XT + R9700: 58.4 tok/s own vs 58.5 borrowing), while a @@ -2743,8 +2809,8 @@ int main(int argc, char** argv) { std::fprintf(stderr, "strata generate: layer split: CUDA%d free memory: %s\n", dev < 0 ? 0 : dev, cudaGetErrorString(e)); const int64_t pf = search && split_own_auto && !place_with_reserve ? 0 : split_pf_mib; - const int64_t reserve = ((int64_t) o.vram_reserve_mib + pf + (later ? kWindowMib : 0) + - (drafter ? kDrafterMib : 0)) << 20; + const int64_t reserve = ((int64_t) o.vram_reserve_mib + pf + (later ? kWindowMib : kPipeSnapMib) + + kPipeWindowMib + (drafter ? kDrafterMib : 0)) << 20; return std::max((int64_t) fb - reserve, 0); }; if (multi_gpu && split_auto) { @@ -3013,7 +3079,11 @@ int main(int argc, char** argv) { static const strata::core::ModelGeometry draft_geometry{}; // with a layer split across GPUs the drafter reads the last stage's residual: it lives on that device const strata::core::OnDevice on_mtp(last_st ? last_st->dev : -1); - if (!o.mtp.empty() && !mtp.load(o.mtp, draft_geometry, last_st ? last_st->ss : ss, o.spec, err, o.mtp_window)) { std::fprintf(stderr, "strata generate: %s%s\n", err.c_str(), vram_free_note().c_str()); return 1; } + // --pipeline-windows 2: 8 rows (the chain runs on past the window in flight: its drafts, its bonus token and + // the next window's drafts), and the round/step graphs carry the forcing kernel + const int mtp_t = o.pipeline_windows >= 2 ? strata::kernels::kVerifyMaxT : o.spec; + if (o.pipeline_windows >= 2) mtp.set_force_capture(true); + if (!o.mtp.empty() && !mtp.load(o.mtp, draft_geometry, last_st ? last_st->ss : ss, mtp_t, err, o.mtp_window)) { std::fprintf(stderr, "strata generate: %s%s\n", err.c_str(), vram_free_note().c_str()); return 1; } mtp.set_ple_session(&ss); } // THE HEAD BEFORE THE CACHE, AND BEFORE THE ARENA. The expert cache takes what is free minus the reserve, so @@ -3203,6 +3273,13 @@ int main(int argc, char** argv) { // THE HEAD BEFORE THE CACHE: the native head and the logits are allocated above, before the expert arena (#620) const bool auto_cache = o.expert_cache < 0; bool reserve_adapted = false; // #496: the auto sizing lowered the reserve so a small card's cache fits + // --pipeline-windows: what it allocates on CUDA0 after the cache (see kPipeWindowMib) + const int64_t pipe_first = o.pipeline_windows <= 0 ? 0 + : (kPipeWindowMib << 20) + (o.pipeline_windows >= 2 ? 2 * (int64_t) gdn_snapshot_bytes(ss) : 0); + if (pipe_first > 0) + std::fprintf(stderr, "strata generate: --pipeline-windows %d: %lld MiB of CUDA0 kept out of the expert cache " + "(the second verifier%s)\n", o.pipeline_windows, (long long) (pipe_first >> 20), + o.pipeline_windows >= 2 ? ", the GDN snapshots" : ""); if (o.expert_cache < 0) { size_t free_b = 0, total_b = 0; cudaMemGetInfo(&free_b, &total_b); @@ -3215,7 +3292,7 @@ int main(int argc, char** argv) { // ~110-180 MiB larger, and out of the reserve they left 16 GB cards below the stall line (#199) const int64_t mtp_bind = (!o.mtp.empty() && native_head.loaded()) ? (int64_t) mtp.bind_bytes(native_head.row_bytes(), n_vocab) : 0; - const int64_t reserve = (((int64_t) o.vram_reserve_mib + prefill_mib) << 20) + mtp_bind; + const int64_t reserve = (((int64_t) o.vram_reserve_mib + prefill_mib) << 20) + mtp_bind + pipe_first; const int64_t blob = (int64_t) strata::kernels::cpu::expert_layout().max_blob; int64_t slots = ((int64_t) free_b - reserve) / blob; if (!profile.empty()) slots = std::min(slots, (int64_t) profile.size()); @@ -3238,10 +3315,10 @@ int main(int argc, char** argv) { ? ((int64_t) strata::prefill::Prefill::bytes_needed(g, ss, 256) + blob - 1) / blob + 128 : 1; if (o.expert_cache < min_slots && !o.vram_reserve_given && o.vram_reserve_mib > kSmallReserveMib) { // the largest reserve (in MiB) that still leaves min_slots - const int64_t fit_mib = ((int64_t) free_b - mtp_bind - min_slots * blob) / (1 << 20) - prefill_mib; + const int64_t fit_mib = ((int64_t) free_b - mtp_bind - pipe_first - min_slots * blob) / (1 << 20) - prefill_mib; if (fit_mib >= kSmallReserveMib) { const int r = (int) std::min(fit_mib, o.vram_reserve_mib); - int64_t s2 = ((int64_t) free_b - ((((int64_t) r + prefill_mib) << 20) + mtp_bind)) / blob; + int64_t s2 = ((int64_t) free_b - ((((int64_t) r + prefill_mib) << 20) + mtp_bind + pipe_first)) / blob; if (!profile.empty()) s2 = std::min(s2, (int64_t) profile.size()); std::fprintf(stderr, "strata generate: expert cache auto: the %d MiB reserve leaves too few slots on " "this card (a working cache needs %lld): a %d MiB reserve instead -> %lld slots\n", @@ -3255,7 +3332,7 @@ int main(int argc, char** argv) { // what is short, and what makes room: the numbers a small card picks from const int64_t at_reserve = o.vram_reserve_given ? o.vram_reserve_mib : std::min(o.vram_reserve_mib, kSmallReserveMib); - const int64_t need_b = (((int64_t) at_reserve + prefill_mib) << 20) + mtp_bind + min_slots * blob; + const int64_t need_b = (((int64_t) at_reserve + prefill_mib) << 20) + mtp_bind + pipe_first + min_slots * blob; const int64_t short_mib = std::max(1, (need_b - (int64_t) free_b + (1 << 20) - 1) >> 20); const int64_t session_mib = (int64_t) (strata::core::session_bytes(g, o.max_context, K, 0, g.n_layers) >> 20); @@ -3280,7 +3357,7 @@ int main(int argc, char** argv) { size_t free_b = 0, total_b = 0; cudaMemGetInfo(&free_b, &total_b); const int64_t prefill_mib = (o.prefill_chunk > 0 && !pf_borrow) ? 160 + (o.prefill_chunk * 680) / 1024 : 0; - const int64_t reserve = ((int64_t) o.vram_reserve_mib + prefill_mib) << 20; + const int64_t reserve = (((int64_t) o.vram_reserve_mib + prefill_mib) << 20) + pipe_first; const int64_t fit = std::max(((int64_t) free_b - reserve) / (int64_t) strata::kernels::cpu::expert_layout().max_blob, 0); if (o.expert_cache > fit) { // a WARNING that names the knob: the user asked for this size, and gets fewer slots @@ -3302,7 +3379,8 @@ int main(int argc, char** argv) { const auto& lay = strata::kernels::cpu::expert_layout(); const uint64_t budget = (uint64_t) o.expert_cache * lay.max_blob; // what the uniform sizing granted uint64_t used = 0; - size_t free_room = free_b > ((size_t) o.vram_reserve_mib << 20) ? free_b - ((size_t) o.vram_reserve_mib << 20) : 0; + const size_t keep_free = ((size_t) o.vram_reserve_mib << 20) + (size_t) pipe_first; + size_t free_room = free_b > keep_free ? free_b - keep_free : 0; const uint64_t cap = std::min(budget, (uint64_t) free_room); for (const auto& pr : profile) { const uint64_t b = (lay.blob_bytes(pr.first) + 255) / 256 * 256; @@ -3403,7 +3481,7 @@ int main(int argc, char** argv) { cudaDeviceSynchronize(); size_t free_b = 0, total_b = 0; cudaMemGetInfo(&free_b, &total_b); - const int64_t want = (int64_t) o.vram_reserve_mib << 20; + const int64_t want = ((int64_t) o.vram_reserve_mib << 20) + pipe_first; if ((int64_t) free_b >= want - (64ll << 20)) break; // short by (want - free); a figure of 0 only says "at least": the first two such reads give back 1 GiB // each (under WDDM the free figure read before the allocation runs ~0.7 GiB high), later ones a quarter @@ -5075,6 +5153,29 @@ int main(int argc, char** argv) { }; auto pcie_num_of = [](double f) { return std::max(0, std::min(256, (int) (f * 256.0 + 0.5))); }; const int n_stages = split_devs.empty() ? 1 : (int) split_at.size() + 1; + // --pipeline-windows (decided where the split was): two verifiers per stage, one for the even windows and one + // for the odd ones, sharing one stream of their card, and a hand-off per parity - so stage 0 can run window + // K+1 while stage 1 still reads window K's hand-off + const bool pipe = o.pipeline_windows > 0 && n_stages == 2 && !split_same && stages.size() == 1; + strata::core::Verifier ver_b; + SplitDrive split_drive_b; + cudaStream_t pl_stream[2] = {nullptr, nullptr}; + if (pipe) { + bool ok_p = cudaStreamCreateWithFlags(&pl_stream[0], cudaStreamNonBlocking) == cudaSuccess; + { + const strata::core::OnDevice on(stages[0]->dev); + ok_p = ok_p && cudaStreamCreateWithFlags(&pl_stream[1], cudaStreamNonBlocking) == cudaSuccess; + } + if (!ok_p) { std::fprintf(stderr, "strata serve: --pipeline-windows: stream creation failed\n"); return 1; } + for (strata::core::Verifier* v : {&ver, &ver_b}) { + v->set_stream(pl_stream[0]); + v->set_always_publish(true); + } + for (strata::core::Verifier* v : {&stages[0]->ver, &stages[0]->ver_b}) { + v->set_stream(pl_stream[1]); + v->set_always_publish(true); + } + } if (n_stages > 1) { const size_t hb = (size_t) strata::kernels::kVerifyMaxT * (size_t) strata::core::Verifier::handoff_floats(g) * sizeof(float); @@ -5130,12 +5231,58 @@ int main(int argc, char** argv) { " (CUDA" + std::to_string(split_same ? 0 : stages[(size_t) st - 1]->dev) + ")"; std::fprintf(stderr, "strata serve: layer split: layers %s, one hand-off per window\n", plan_s.c_str()); } + // --pipeline-windows: the odd windows' verifiers and their hand-off (initialized before `ver`, which stays the + // watchdog's verifier), and the drafter's own row buffer, which either parity's rows are copied into + float* pl_mtp_R = nullptr; + if (pipe) { + const size_t hb = (size_t) strata::kernels::kVerifyMaxT * + (size_t) strata::core::Verifier::handoff_floats(g) * sizeof(float); + float *hh = nullptr, *hand_b = nullptr; + if (cudaHostAlloc((void**) &hh, hb, cudaHostAllocMapped | cudaHostAllocPortable) != cudaSuccess || + cudaHostGetDevicePointer((void**) &hand_b, hh, 0) != cudaSuccess) { + std::fprintf(stderr, "strata serve: --pipeline-windows: the second hand-off allocation failed\n"); + return 1; + } + std::memset(hh, 0, hb); + GpuStage& gs = *stages[0]; + ver_b.set_stage(0, split_at[0], nullptr, hand_b); + gs.ver_b.set_stage(split_at[0], -1, hand_b, nullptr); + { + const strata::core::OnDevice on(gs.dev); + strata::core::VerifyHits vs; + vs.d_res = gs.d_res; + vs.h_res = host_res.empty() ? nullptr : host_res.data(); + vs.cache_base = gs.cache.device_slot(0); + vs.blob = thits.blob; + vs.slot_off = gs.cache.slot_offsets(); + vs.n_slots = gs.cache.slots(); + if (!gs.ver_b.init(gs.wt, g, gs.ss, vs, gs.head.loaded() ? &gs.head : nullptr, o.spec, err)) { + std::fprintf(stderr, "strata serve: --pipeline-windows: the second verifier on CUDA%d (the later " + "card): %s (raise --vram-reserve-mib by ~200, or --pipeline-windows 0)\n", + gs.dev, err.c_str()); + return 1; + } + if (cudaMalloc((void**) &pl_mtp_R, (size_t) strata::kernels::kVerifyMaxT * (size_t) (g.hc * g.n_embd) * + sizeof(float)) != cudaSuccess) { + std::fprintf(stderr, "strata serve: --pipeline-windows: the drafter's row buffer does not fit\n"); + return 1; + } + } + if (!ver_b.init(wt, g, ss, vh, native_head.loaded() ? &native_head : nullptr, o.spec, err)) { + std::fprintf(stderr, "strata serve: --pipeline-windows: the second verifier on CUDA0 (the first card): " + "%s (raise --vram-reserve-mib by ~200, or --pipeline-windows 0)\n", err.c_str()); + return 1; + } + ver_b.set_next(&gs.ver_b, &split_drive_b); // the setters reach it; the pipeline never chains run/commit + } ver.set_remote_expert_opt(remote_opt.get()); if (!ver.init(wt, g, ss, vh, native_head.loaded() ? &native_head : nullptr, std::max(o.spec, o.batch), err) || - !mtp.bind(last_st ? last_st->wt : wt, last_st ? &last_st->head : &native_head, ver.final_R_all(), err)) { + !mtp.bind(last_st ? last_st->wt : wt, last_st ? &last_st->head : &native_head, + pipe ? pl_mtp_R : ver.final_R_all(), err)) { std::fprintf(stderr, "strata serve: %s\n", err.c_str()); return 1; } + if (pipe) mtp.set_source_R(ver.final_R_all()); // the serial decode's rows (the last stage's even verifier) // each stage's verifier gets its stage's slot sessions if (o.batch > 0) { for (size_t k = 0; k < bslot_ss.size(); ++k) { @@ -5156,6 +5303,15 @@ int main(int argc, char** argv) { stage_ver(st).set_pcie_mode(o.pcie_mode == "dma" ? 0 : o.pcie_mode == "direct" ? 1 : 2); } } + if (pipe) { // the odd windows' pool routing: the same as the even ones', with their plans + split_drive_b = split_drive; + split_drive_b.plan[0] = ver_b.plan_sink(); + split_drive_b.plan[1] = stages[0]->ver_b.plan_sink(); + stages[0]->ver_b.set_split(o.spec_split); + stages[0]->ver_b.set_pcie_mode(o.pcie_mode == "dma" ? 0 : o.pcie_mode == "direct" ? 1 : 2); + ver_b.set_split(o.spec_split); + ver_b.set_pcie_mode(o.pcie_mode == "dma" ? 0 : o.pcie_mode == "direct" ? 1 : 2); + } // the pool the verify windows call: with a layer split, the wrapper that routes each layer to its stage const strata::core::PoolMultiFn win_pool_fn = n_stages > 1 ? &drive_pool_split : &drive_pool_multi; void* const win_pool_user = n_stages > 1 ? (void*) &split_drive : (void*) &drive; @@ -5166,6 +5322,86 @@ int main(int argc, char** argv) { // issue #31's thread dumps show the host stuck in that cudaMemcpyAsync on a driver lock for good. The copy // kernel needs no host CUDA call there, and costs ~1-3% decode on IQ3_S (45.3 -> 44.8 tok/s, 8 requests). ver.set_pcie_mode(o.pcie_mode == "dma" ? 0 : o.pcie_mode == "direct" ? 1 : 2); + // --pipeline-windows 2: stage 0's GDN state before a speculative commit, restored when the guess was wrong. + // Window W reads the state its snapshot must hold and no window writes it (only commits do), so it is copied + // on a side stream while W runs, into the buffer of W's parity, and only W's speculative commit waits for it + // (STRATA_PIPELINE_SNAP_OVERLAP=0: one buffer, copied on the critical path). + const size_t pl_snap_bytes = pipe ? gdn_snapshot_bytes(ss) : 0; + float* pl_snap2[2] = {nullptr, nullptr}; + cudaStream_t pl_snap_stream = nullptr; + cudaEvent_t pl_snap_pre[2] = {nullptr, nullptr}, pl_snap_ev[2] = {nullptr, nullptr}; + bool pl_snap_overlap = false; + if (pipe && o.pipeline_windows >= 2) { + if (cudaMalloc((void**) &pl_snap2[0], pl_snap_bytes) != cudaSuccess) { + std::fprintf(stderr, "strata serve: --pipeline-windows 2: the GDN snapshot (%.1f MiB) does not fit; the " + "decode stays serial\n", (double) pl_snap_bytes / 1048576.0); + cudaGetLastError(); + pl_snap2[0] = nullptr; + } else if ([] { const char* v = std::getenv("STRATA_PIPELINE_SNAP_OVERLAP"); return v == nullptr || std::atoi(v) != 0; }()) { + pl_snap_overlap = cudaMalloc((void**) &pl_snap2[1], pl_snap_bytes) == cudaSuccess && + cudaStreamCreateWithFlags(&pl_snap_stream, cudaStreamNonBlocking) == cudaSuccess; + for (int i = 0; i < 2 && pl_snap_overlap; ++i) + pl_snap_overlap = cudaEventCreateWithFlags(&pl_snap_pre[i], cudaEventDisableTiming) == cudaSuccess && + cudaEventCreateWithFlags(&pl_snap_ev[i], cudaEventDisableTiming) == cudaSuccess; + if (!pl_snap_overlap) cudaGetLastError(); + } + } + if (pipe) + std::fprintf(stderr, "strata serve: --pipeline-windows %d: two verifiers per stage, the stages overlap across " + "windows%s\n", o.pipeline_windows, + pl_snap2[0] == nullptr ? "" : pl_snap_overlap ? " (decode too; the GDN snapshots beside the window)" + : " (decode too; the GDN snapshot on the critical path)"); + // the verifiers by [stage][parity], their pool routing, and the drafter's events (in a pipelined prompt read, a + // stage-1 window waits until the drafter has read the rows of the same parity's previous window) + strata::core::Verifier* PV[2][2] = {{&ver, &ver_b}, {pipe ? &stages[0]->ver : nullptr, pipe ? &stages[0]->ver_b : nullptr}}; + SplitDrive* PSD[2] = {&split_drive, &split_drive_b}; + cudaEvent_t pl_mtp_ev[2] = {nullptr, nullptr}; + bool pl_mtp_live[2] = {false, false}; + bool pl_prepared = false; + // every graph captured and the drafter's prompt path sized while nothing is in flight (a capture syncs its stream) + auto pl_prepare = [&](std::string& e) -> bool { + if (pl_prepared) return true; + for (int st = 0; st < 2; ++st) + for (int par = 0; par < 2; ++par) { + const strata::core::OnDevice on(PV[st][par]->device()); + if (!PV[st][par]->capture_all(e)) return false; + } + if (!mtp.prepare_prefill(e)) return false; + if (o.pipeline_windows >= 2 && pl_snap2[0] != nullptr && !mtp.prepare_chain(e)) return false; + const strata::core::OnDevice on(mtp.device()); + for (cudaEvent_t& ev : pl_mtp_ev) + if (ev == nullptr && cudaEventCreateWithFlags(&ev, cudaEventDisableTiming) != cudaSuccess) { + e = "pipeline: event creation failed"; + return false; + } + pl_prepared = true; + return true; + }; + // INVARIANT: nothing pipelined is in flight outside read_windows_pl and the pipelined decode loop - both end + // here (a checkpoint, a park, the next request and the serial paths all find both stages and the drafter idle). + // Every window in flight is served to its end (its picks discarded), then both stages and the drafter idle. + auto pl_drain = [&]() { + std::string e2; + for (int round = 0; round < 10000000; ++round) { + bool any = false; + for (int st = 0; st < 2; ++st) + for (int par = 0; par < 2; ++par) { + strata::core::Verifier& v = *PV[st][par]; + if (!v.in_flight()) continue; + any = true; + if (v.service(&drive_pool_split, PSD[par], e2) < 0) return; + if (v.done(e2)) v.pl_finish(nullptr, e2); + else if (!e2.empty()) return; + } + if (!any) break; + } + for (int st = 0; st < 2; ++st) { + const strata::core::OnDevice on(PV[st][0]->device()); + cudaStreamSynchronize(PV[st][0]->stream()); + } + const strata::core::OnDevice on(mtp.device()); + cudaStreamSynchronize(mtp.stream()); + }; std::vector cur; // ---- the conversation cache (see ConvCheckpoint). `live` is what the session holds right now: the tokens // it has consumed, so a request that starts with exactly them continues without any copy. `checks` are the @@ -5487,6 +5723,8 @@ int main(int argc, char** argv) { pending.clear(); res_upload(); }; + // --pipeline-windows: set while the tier adapts beside windows in flight (see its use in adapt) + std::function adapt_fence; // the VRAM tier follows the conversation (the same rule as the speculative loop below) auto adapt = [&]() -> bool { if (!pending.empty()) return true; // the previous swaps are still in flight @@ -5524,10 +5762,25 @@ int main(int argc, char** argv) { (uintptr_t) b + strata::kernels::cpu::expert_layout().blob_bytes(s.layer)); pin_blobs(std::move(spans), pin_live); } + // --pipeline-windows: windows are in flight while the tier adapts. The evicted experts stop being planned + // for the GPU first (the pool computes them on the CPU from the rows every layer publishes, see + // Verifier::set_always_publish); then every copy into their slots waits until all the work the stages' + // streams hold now has completed: any window that may still read those slots was launched before this. + std::vector fenced_slots; + if (adapt_fence && !swaps.empty()) { + for (const Swap& s : swaps) { + const size_t out = (size_t) s.layer * g.n_expert + s.out; + fenced_slots.push_back(host_res[out]); + host_res[out] = strata::core::kNotResident; + } + std::atomic_thread_fence(std::memory_order_seq_cst); + adapt_fence(); + } bool main_live = false; - for (const Swap& s : swaps) { + for (size_t si = 0; si < swaps.size(); ++si) { + const Swap& s = swaps[si]; const size_t in = (size_t) s.layer * g.n_expert + s.in, out = (size_t) s.layer * g.n_expert + s.out; - const int32_t slot = host_res[out]; + const int32_t slot = fenced_slots.empty() ? host_res[out] : fenced_slots[si]; const uint8_t* b = srcp->blob(s.layer, s.in); const int stn = multi_gpu ? stage_of(s.layer) : 0; // the swap stays in the layer's own cache GpuStage* gs = stn > 0 ? stages[(size_t) stn - 1].get() : nullptr; @@ -5747,6 +6000,8 @@ int main(int argc, char** argv) { const int S = o.spec; const int S_mtp = o.mtp_max_t > 0 ? std::min(o.mtp_max_t, S) : S; // the MTP's windows; suffixes go up to S if (S_mtp < S) mtp.set_max_drafts(S_mtp - 1); + // --pipeline-windows 2 loads the drafter with 8 rows: the serial loop's chain keeps its own length + else if (mtp.max_t() > S_mtp) mtp.set_max_drafts(S_mtp - 1); strata::spec::SuffixDrafter sfx(std::max(1, o.suffix_draft), 64, (size_t) o.max_context + 4096); strata::spec::DraftPolicy policy(S); // MTP or lookup window, learned over the whole process // The vision path (--vision): GENI carries images. The file is one @@ -6673,15 +6928,132 @@ int main(int argc, char** argv) { // read batched and its header still goes through the windows. Picture rows need the batched path. // STRATA_CKPT_REREAD compares a restored checkpoint with a batched re-read, so it keeps every read batched. static const bool no_short = std::getenv("STRATA_CKPT_REREAD") != nullptr; + int64_t req_short_read = o.short_read; // STRATA_PIPELINE_SWITCH may set it per request (short_read=) auto windows_ok = [&](int64_t a, int64_t b) -> bool { - if (no_short || b - a > o.short_read) return false; + if (no_short || b - a > req_short_read) return false; if (sp.embd_rows != nullptr) for (int64_t i = a; i < b; ++i) if (sp.embd_rows[i] != nullptr) return false; return true; }; + // --pipeline-windows, testing: STRATA_PIPELINE_SWITCH= is read at every request, "pw=<0|1|2> + // theta= force_miss= short_read=" - an A/B of the pipelined and the serial loops (and forced + // rollbacks) on one server, with the same expert placement + int pl_pw = pipe ? o.pipeline_windows : 0; + float pl_theta = [] { // a speculative window is launched only when its estimated chance is at least this + const char* v = std::getenv("STRATA_PIPELINE_THETA"); + return v ? (float) std::atof(v) : 0.2f; + }(); + int pl_force_miss = [] { // exactness test: every k-th speculative window gets a wrong first token + const char* v = std::getenv("STRATA_PIPELINE_FORCE_MISS"); + return v ? std::max(0, std::atoi(v)) : 0; + }(); + if (static const char* sw = std::getenv("STRATA_PIPELINE_SWITCH"); sw != nullptr && pipe) { + if (std::FILE* f = std::fopen(sw, "r")) { + char buf[256] = {}; + const size_t nr = std::fread(buf, 1, sizeof buf - 1, f); + std::fclose(f); + buf[nr] = 0; + if (const char* q = std::strstr(buf, "pw=")) pl_pw = std::max(0, std::min(2, std::atoi(q + 3))); + if (const char* q = std::strstr(buf, "theta=")) pl_theta = (float) std::atof(q + 6); + if (const char* q = std::strstr(buf, "force_miss=")) pl_force_miss = std::max(0, std::atoi(q + 11)); + if (const char* q = std::strstr(buf, "short_read=")) req_short_read = std::max(0LL, std::atoll(q + 11)); + std::fprintf(stderr, "strata pipeline switch: pw=%d theta=%.3f force_miss=%d short_read=%lld\n", pl_pw, + pl_theta, pl_force_miss, (long long) req_short_read); + } + } + // --pipeline-windows: tokens [a, b) through the windows with the stages overlapped. The tokens are known, + // so every window commits whole: stage 0 queues window K, its commit and window K+1 back to back (each + // parity's verifier keeps its own window), and stage 1 runs window K once stage 0 is done with it. Every + // stage runs the same windows in the same order as read_windows below, so the session ends the same. + auto read_windows_pl = [&](int64_t a, int64_t b, std::string& e) -> bool { + strata::core::progress_at("reading the prompt (pipelined verify windows), from token", a); + if (!pl_prepare(e)) return false; + for (int par = 0; par < 2; ++par) PV[0][par]->set_head_sampling(false); // reaches stage 1 (set_next) + struct Restore { + strata::core::Verifier* v[2]; + ~Restore() { v[0]->set_head_sampling(true); v[1]->set_head_sampling(true); } + } restore{{PV[0][0], PV[0][1]}}; + struct Win { int T; int64_t q; }; + std::vector wins; + for (int64_t q = a; q < b;) { + const int T = (int) std::min(S, b - q); + wins.push_back({T, q}); + q += T; + } + const size_t N = wins.size(); + std::vector w0((size_t) S), w1((size_t) S), nx((size_t) S); + size_t l0 = 0, f0 = 0, l1 = 0, f1 = 0; // stage 0 launched / finished, stage 1 launched / finished + drive.d.layers = 0; + drive.d.experts = 0; + drive.d.failed = false; + auto fail = [&](std::string why) -> bool { + pl_drain(); + e = why.empty() ? std::string("a pipelined prompt window failed") : why; + return false; + }; + while (f1 < N) { + if (stop_req.load()) return fail("cancelled"); + // stage 0: up to two windows queued; window K+2 (K's parity) only once stage 1 has finished window + // K, whose hand-off K+2 overwrites + while (l0 < N && l0 < f0 + 2 && l0 < f1 + 2) { + const Win& w = wins[l0]; + for (int t = 0; t < w.T; ++t) w0[(size_t) t] = (int32_t) cur[(size_t) (w.q + t)]; + strata::core::Verifier& v = *PV[0][l0 & 1]; + if (!v.pl_launch(w.T, w0.data(), w.q, e) || !v.pl_commit_async(w.T, e)) return fail(e); + ++l0; + } + if (f0 < l0) { + strata::core::Verifier& v = *PV[0][f0 & 1]; + if (v.service(win_pool_fn, PSD[f0 & 1], e) < 0) return fail(e); + if (v.done(e)) { + if (!v.pl_finish(nullptr, e)) return fail(e); + ++f0; + } else if (!e.empty()) return fail(e); + } + if (l1 < f0 && l1 == f1) { // stage 1: the next window, its stage 0 done + const Win& w = wins[l1]; + for (int t = 0; t < w.T; ++t) w1[(size_t) t] = (int32_t) cur[(size_t) (w.q + t)]; + strata::core::Verifier& v = *PV[1][l1 & 1]; + if (pl_mtp_live[l1 & 1]) { // the drafter has read this parity's previous rows first + const strata::core::OnDevice on(mtp.device()); + cudaStreamWaitEvent(v.stream(), pl_mtp_ev[l1 & 1], 0); + } + if (!v.pl_launch(w.T, w1.data(), w.q, e) || !v.pl_commit_async(w.T, e)) return fail(e); + ++l1; + } + if (f1 < l1) { + strata::core::Verifier& v = *PV[1][f1 & 1]; + if (v.service(win_pool_fn, PSD[f1 & 1], e) < 0) return fail(e); + if (v.done(e)) { + if (!v.pl_finish(nullptr, e)) return fail(e); + const Win& w = wins[f1]; + for (int t = 0; t < w.T; ++t) nx[(size_t) t] = (int32_t) cur[(size_t) (w.q + t + 1)]; + if (!mtp.prefill(v.final_R(0), nx.data(), w.T, w.q, e, false)) return fail(e); + { + const strata::core::OnDevice on(mtp.device()); + cudaEventRecord(pl_mtp_ev[f1 & 1], mtp.stream()); + } + pl_mtp_live[f1 & 1] = true; + pp_reached = w.q + w.T; // #471 + ++f1; + } else if (!e.empty()) return fail(e); + } + if (drive.d.failed) return fail(drive.d.fail ? drive.d.fail : "the expert pool failed"); + } + pl_drain(); // nothing in flight: both stages' commits and the drafter's rows have landed + pl_mtp_live[0] = pl_mtp_live[1] = false; + const double ms = std::chrono::duration(Clock::now() - pp_t0).count(); + std::printf("PP %lld %lld %.0f %.1f\n", (long long) b, (long long) pp_total, ms, + ms > 0.0 ? 1000.0 * (double) (b - pp_from) / ms : 0.0); + strata::core::progress_beat(); + std::fflush(stdout); + return true; + }; // tokens [a, b) through the windows: commit all of them, then give the draft layer their residuals auto read_windows = [&](int64_t a, int64_t b, std::string& e) -> bool { + // (STRATA_LOGPOS reads every window's logits: the serial loop) + if (pipe && pl_pw >= 1 && std::getenv("STRATA_LOGPOS") == nullptr) return read_windows_pl(a, b, e); strata::core::progress_at("reading the prompt (verify windows), from token", a); // #217: not "batched" // every token is committed and the picks are discarded: no head sampling (see set_head_sampling) struct NoHeadSampling { @@ -6951,14 +7323,17 @@ int main(int argc, char** argv) { req_sp.penalty_present = req_penalty_present; req_sp.counter = 0; ver.set_sampling(req_sp); + if (pipe) ver_b.set_sampling(req_sp); // (reaches the later stage's odd verifier) mtp.set_draft_sampling(req_sp); // STRATA_SPEC_COUPLED=1: sampled drafts (a no-op otherwise) drive.d.pcie_num = std::max(0, std::min(256, (int) (req_pcie_frac * 256.0 + 0.5))); // a layer split: CUDA0's share as asked; a later GPU keeps its own (its link) unless the request sets one for (int st = 0; st < split_drive.n; ++st) split_drive.pcie_num[st] = (st == 0 || split_same || req_pcie_frac != o.pcie_frac) ? drive.d.pcie_num : pcie_num_of(stages[(size_t) st - 1]->pcie_frac); + if (pipe) for (int st = 0; st < split_drive.n; ++st) split_drive_b.pcie_num[st] = split_drive.pcie_num[st]; const int hist_n = std::min(req_sp.penalty_last_n, kPenaltyWindowCap); ver.set_history(hist_n > 0 ? d_hist : nullptr, hist_n); + if (pipe) ver_b.set_history(hist_n > 0 ? d_hist : nullptr, hist_n); bool cancelled = false; tr("prompt start", n - 1); // The prompt is read in two parts when it has a turn boundary past `resume`: up to the last <|im_start|> @@ -7047,8 +7422,10 @@ int main(int argc, char** argv) { // the verify windows: the first holds the last prompt token alone int64_t p = n - 1; int32_t x = (int32_t) ids[(size_t) (n - 1)]; - std::vector drafts((size_t) S, 0), window((size_t) S), outv((size_t) S); - std::vector dprob((size_t) S, 0.0f); + // (the drafter writes max_t - 1 drafts: 8 rows with --pipeline-windows 2) + const size_t DS = (size_t) std::max(S, mtp.max_t()); + std::vector drafts(DS, 0), window((size_t) S), outv((size_t) S); + std::vector dprob(DS, 0.0f); std::vector sbuf((size_t) S, 0); if (o.suffix_draft > 0) { sfx.reset(); @@ -7084,7 +7461,595 @@ int main(int argc, char** argv) { const int64_t decode_look0 = drive.d.cache_hits + drive.d.cache_admitted + drive.d.cache_refused; const int64_t offload0 = drive.d.offload_entries; // #588 if (cancelled) finish = "cancel"; - while (!cancelled && produced_n < max_new) { + // ======== --pipeline-windows 2: the decode windows with the two cards overlapped ======== + // Stage 0 (the first card) runs window K+1 while stage 1 verifies window K, on the guess that K is accepted + // whole and that its bonus token is the drafter's. The drafter, teacher-forced through K+1's drafts, + // guesses K+1's bonus and drafts K+2 while stage 1 runs K+1. A wrong guess costs the speculative window: + // stage 0's GDN state comes back from a snapshot and K's commit is replayed with its real count (the + // window's own commit inputs are intact in its parity's verifier). Stage 1 never runs a speculative window + // and a window row computes what it would in any other window, so every emitted token is the serial loop's + // (with STRATA_IQ_MT_MIN=1 bit for bit; sampled requests with the same Philox draw per position). + bool pl_ran = false; + const bool pl_want = pipe && pl_pw >= 2 && pl_snap2[0] != nullptr; + const char* pl_serial = !pl_want ? nullptr + : hist_n > 0 ? "repetition penalties (penalty_last_n)" + : mtp.coupled() ? "coupled draft sampling" : nullptr; + if (pl_serial != nullptr) { // said once per reason + static std::set said; + if (said.insert(pl_serial).second) + std::fprintf(stderr, "strata serve: --pipeline-windows: a request with %s decodes serially\n", + pl_serial); + } + if (pl_want && pl_serial == nullptr && !cancelled && produced_n < max_new) { + pl_ran = true; + if (!pl_prepare(err)) { + std::printf("ERR %s\n", err.c_str()); + return 1; + } + struct PW { + int T = 0, seq = 0; + int64_t p = 0; + int32_t tok[8] = {}; + float prob[8] = {}; // prob[i]: the drafter's probability of tok[i + 1] + float p_on = 0.0f; // the chain's estimate that the window before is accepted whole, tok[0] its bonus + bool ready = false, launched = false, finished = false, committed = false, s1 = false, s1_done = false; + bool spec = false; // launched speculatively (behind the window before, ahead of its verdict) + bool made = false; // made from a chain (its outcome can be scored even when the gate held it) + bool sfx = false; // a lookup window (the suffix drafter's drafts) + int sfx_match = 0; + }; + auto V0 = [&](const PW& w) -> strata::core::Verifier& { return *PV[0][w.seq & 1]; }; + auto V1 = [&](const PW& w) -> strata::core::Verifier& { return *PV[1][w.seq & 1]; }; + auto SDf = [&](const PW& w) { return (void*) PSD[w.seq & 1]; }; + const float theta = pl_theta; + const int force_miss = pl_force_miss; + static const bool pl_log = std::getenv("STRATA_PIPELINE_LOG") != nullptr; + const int dev0 = PV[0][0]->device(); + cudaStream_t s0 = PV[0][0]->stream(); + // the snapshot of the GDN state window `seq` reads (overlapped: before its launch, on the side stream) + auto snap_take = [&](int seq) -> bool { + if (!pl_snap_overlap) return true; + const strata::core::OnDevice on(dev0); + const int q = seq & 1; + return cudaEventRecord(pl_snap_pre[q], s0) == cudaSuccess && + cudaStreamWaitEvent(pl_snap_stream, pl_snap_pre[q], 0) == cudaSuccess && + cudaMemcpyAsync(pl_snap2[q], ss.gdn_state, pl_snap_bytes, cudaMemcpyDeviceToDevice, + pl_snap_stream) == cudaSuccess && + cudaEventRecord(pl_snap_ev[q], pl_snap_stream) == cudaSuccess && + (cudaStreamQuery(pl_snap_stream), true); + }; + PW A, B, D; // A: the oldest window not verified; B: the next; D: a wrong speculative window + bool doomed = false; // D still runs: when it finishes, stage 0 goes back to undo_w's real commit + PW undo_w; + int undo_keep = 0; + int32_t undo_ple[2] = {-1, -1}; + int chain_kind = 0; // 1: after an on-path verdict (A's drafts forced); 2: fresh (A's drafts from it) + bool early_used = false; + bool b_done = false; // B has been made from the chain in flight + int chain_n = 0; + int64_t pl_spec = 0, pl_on = 0, pl_undo = 0, pl_gate = 0, pl_fm = 0; + std::vector outp(8, 0); + bool ending = false; + A.T = 1; + A.p = p; + A.tok[0] = x; + A.ready = true; + drive.d.layers = 0; + drive.d.experts = 0; + drive.d.failed = false; + // the adaptive tier beside the pipeline: the serial loop's rule and adapt() on a thread of its own (its + // copies wait for the windows in flight, which this loop keeps serving meanwhile) + std::thread pl_adapt_thr; + std::atomic pl_adapt_done{true}; + bool pl_adapt_ok = true; + // an error ends the engine, as the serial loop's do: every GPU wait is released first (#267), so no + // window in flight keeps spinning on a flag nobody raises, and the tier's thread can finish + auto die = [&](const std::string& m) -> int { + std::fprintf(stderr, "strata serve: pipelined decode: %s\n", m.c_str()); + strata::core::diag_pipeline_fn().store(nullptr); + strata::core::release_gpu_waits(stderr); + if (pl_adapt_thr.joinable()) pl_adapt_thr.join(); + std::printf("ERR %s\n", m.c_str()); + std::fflush(stdout); + return 1; + }; + // STRATA_PIPELINE_TRACE=: every event of the loop with its time (appended per request) + static const char* pl_trace_path = std::getenv("STRATA_PIPELINE_TRACE"); + std::string pl_trace; + const Clock::time_point pl_tr0 = Clock::now(); + auto tre = [&](const char* ev, int seq, int x0 = 0, int x1 = 0) { + if (pl_trace_path == nullptr) return; + char b[96]; + std::snprintf(b, sizeof b, "%.3f %s %d %d %d\n", + std::chrono::duration(Clock::now() - pl_tr0).count(), ev, seq, x0, x1); + pl_trace += b; + }; + auto pump0 = [&](PW& w) -> bool { + if (!w.launched || w.finished) return true; + strata::core::Verifier& v = V0(w); + // a doomed window (its guess was wrong) gets empty plans: no expert work on the CPU or the GPU for + // its remaining layers, so it frees stage 0 and the pool sooner (its rows are discarded and the + // undo restores everything it wrote). STRATA_PIPELINE_DOOM_SKIP=0: served in full. + static const bool doom_skip = [] { const char* e = std::getenv("STRATA_PIPELINE_DOOM_SKIP"); return e == nullptr || std::atoi(e) != 0; }(); + const bool skip = doom_skip && doomed && &w == &D; + if (v.service(skip ? nullptr : &drive_pool_split, SDf(w), err) < 0) return false; + if (v.done(err)) { + if (!v.pl_finish(nullptr, err)) return false; + w.finished = true; + tre("F0", w.seq, w.T, w.spec); + } + return err.empty(); + }; + // the serial loop's window size over the chain's outputs from index `from` (`avail` of them) + auto t_rule = [&](const float* pr, int from, int avail) { + int T = S_mtp; + if (req_spec_min_p > 0.0) { + T = 1; + while (T < S_mtp && from + T - 1 < avail && pr[from + T - 1] >= (float) req_spec_min_p) ++T; + } + return std::max(1, std::min(T, 1 + avail - from)); + }; + // B from the chain's outputs: its row 0 is the guess at index `base`, its drafts follow + auto make_b = [&](const int32_t* oc, const float* op, int base, int avail, float pa) { + B = PW{}; + if (base >= avail) return; + B.seq = A.seq + 1; + B.p = A.p + A.T; + B.tok[0] = oc[base]; + B.T = t_rule(op, base + 1, avail); + for (int i = 1; i < B.T; ++i) { + B.tok[i] = oc[base + i]; + B.prob[i - 1] = op[base + i]; + } + B.p_on = pa * op[base]; + if (force_miss > 0 && ++pl_fm % force_miss == 0) B.tok[0] = B.tok[0] == 0 ? 1 : 0; + B.ready = true; + B.made = true; + }; + // events after all the work both stages' streams hold now, and whether they have passed + cudaEvent_t fence_ev[2] = {nullptr, nullptr}, exch_ev[2] = {nullptr, nullptr}; + { + const strata::core::OnDevice on0(dev0); + cudaEventCreateWithFlags(&fence_ev[0], cudaEventDisableTiming); + cudaEventCreateWithFlags(&exch_ev[0], cudaEventDisableTiming); + } + { + const strata::core::OnDevice on1(stages[0]->dev); + cudaEventCreateWithFlags(&fence_ev[1], cudaEventDisableTiming); + cudaEventCreateWithFlags(&exch_ev[1], cudaEventDisableTiming); + } + auto stage_record = [&](cudaEvent_t* ev) { + { + const strata::core::OnDevice on(dev0); + cudaEventRecord(ev[0], s0); + (void) cudaStreamQuery(s0); + } + const strata::core::OnDevice on(stages[0]->dev); + cudaEventRecord(ev[1], PV[1][0]->stream()); + (void) cudaStreamQuery(PV[1][0]->stream()); + }; + auto stage_passed = [&](cudaEvent_t* ev) { + return cudaEventQuery(ev[0]) != cudaErrorNotReady && cudaEventQuery(ev[1]) != cudaErrorNotReady; + }; + // adapt()'s fence, on the tier's thread: it waits on the HOST. A device-side wait (the refill streams + // waiting for the stages' events) would queue copies behind windows that need this loop, beside it. + adapt_fence = [&]() { + stage_record(fence_ev); + while (!stage_passed(fence_ev)) std::this_thread::yield(); + }; + src.prepare_overrides(); + // The swaps land in two steps. Once their copies are done, the incoming experts are resident (the plans + // from then on use their slots). What gives their RAM back - the resident RAM mode's exchange, the + // released or unpinned pages - waits until every window that may still pull them over PCIe (a plan + // published before that point) has completed: both stages' streams are fenced there. + bool exch_wait = false; + std::vector pl_landed; + auto pl_release = [&]() { + unpin_blobs(pin_live); + src.commit_exchanges(); + for (const int32_t i : pl_landed) srcp->release((int64_t) i / g.n_expert, (int64_t) i % g.n_expert); + pl_landed.clear(); + exch_wait = false; + }; + auto pl_apply = [&]() { + if (exch_wait) { + if (stage_passed(exch_ev)) pl_release(); + return; + } + if (pending.empty() || cudaEventQuery(adapt_ev) != cudaSuccess) return; + for (auto& st : stages) + if (st->adapt_live && cudaEventQuery(st->adapt_ev) != cudaSuccess) return; + for (auto& st : stages) st->adapt_live = false; + for (const auto& [i, slot] : pending) { + host_res[(size_t) i] = slot; + pl_landed.push_back(i); + } + pending.clear(); + res_upload(); + stage_record(exch_ev); + exch_wait = true; + }; + auto pl_adapt = [&]() -> bool { + if (pl_adapt_thr.joinable() && pl_adapt_done.load()) { + pl_adapt_thr.join(); + if (!pl_adapt_ok) return false; + } + if (pl_adapt_thr.joinable()) return true; // still adapting + pl_apply(); + if (!drive.d.usage.empty() && o.adapt_every > 0 && (rounds % o.adapt_every) == 0 && pending.empty() && + !exch_wait) { + pl_adapt_done = false; + pl_adapt_thr = std::thread([&] { pl_adapt_ok = adapt(); pl_adapt_done = true; }); + } + return true; + }; + // the serial loop's prompt lookup for a fresh window: where the text repeats an earlier stretch and the + // drafter's own first guess agrees, the policy may take the lookup's continuation (up to S rows) + std::vector pl_sbuf((size_t) (2 * S), 0); + // a lookup window's continuation goes on past it: the next window (B) is taken from the same lookup and + // speculated behind it (a copied stretch is mostly accepted whole). STRATA_PIPELINE_LOOKUP_PON: B's + // estimate (default 0.75; 0 = no lookup B) + static const float pl_lookup_pon = [] { + const char* v = std::getenv("STRATA_PIPELINE_LOOKUP_PON"); + return v ? (float) std::atof(v) : 0.75f; + }(); + auto pick_lookup = [&](PW& w, const int32_t* chain0) { + if (o.suffix_draft <= 0 || w.seq == 0) return; + const int k = sfx.propose(2 * S - 1, pl_sbuf.data()); + const int match = sfx.last_match(); + if (k <= 0 || pl_sbuf[0] != chain0[0]) return; + const strata::spec::DraftPolicy::Pick pk = policy.choose(w.T, std::min(k, S - 1), match); + if (!pk.lookup) return; + w.T = pk.t; + for (int i = 1; i < w.T; ++i) { + w.tok[i] = pl_sbuf[(size_t) i - 1]; + w.prob[i - 1] = 1.0f; + } + w.sfx = true; + w.sfx_match = match; + const int rest = k - (w.T - 1); // lookup tokens past w: B's row 0 (w's bonus guess) and its drafts + if (&w == &A && rest >= 1 && pl_lookup_pon > 0.0f && !b_done) { + B = PW{}; + B.seq = A.seq + 1; + B.p = A.p + A.T; + B.T = std::min(S, rest); + for (int i = 0; i < B.T; ++i) B.tok[i] = pl_sbuf[(size_t) (A.T - 1 + i)]; + for (int i = 1; i < B.T; ++i) B.prob[i - 1] = 1.0f; + B.p_on = pl_lookup_pon; + B.ready = true; + B.made = true; + b_done = true; // the chain in flight makes no B for this A + tre("CD", A.seq + 1, 3, (int) (1000.0f * B.p_on)); + } + }; + const Clock::time_point pl_t0 = Clock::now(); + for (int st = 0; st < 2; ++st) + for (int par = 0; par < 2; ++par) g_pl_diag.v[st][par] = PV[st][par]; + strata::core::diag_pipeline_fn().store(&pl_diag_print); + auto pw_bits = [](const PW& w) { + return (w.ready ? 1 : 0) | (w.launched ? 2 : 0) | (w.finished ? 4 : 0) | (w.committed ? 8 : 0) | + (w.s1 ? 16 : 0) | (w.s1_done ? 32 : 0); + }; + // per window class (0: fresh, 1: run speculatively ahead of its predecessor's verdict): the time from + // the previous verdict, the tokens it emitted, how many + double cls_ms[2] = {0, 0}, cls_tok[2] = {0, 0}, cls_n[2] = {0, 0}; + int64_t cal_n[10] = {}, cal_on[10] = {}; // per p_on decile: windows scored, on the path + double last_verdict = 0.0; + int64_t pl_disagree = 0, pl_late = 0; + static const bool pl_prestage = [] { // STRATA_PIPELINE_PRESTAGE=0: B staged at its launch + const char* v = std::getenv("STRATA_PIPELINE_PRESTAGE"); + return v == nullptr || std::atoi(v) != 0; + }(); + static const bool pl_agree = [] { // STRATA_PIPELINE_AGREE=0: only the old chain's probabilities + const char* v = std::getenv("STRATA_PIPELINE_AGREE"); + return v == nullptr || std::atoi(v) != 0; + }(); + auto ms_now = [&]() { return std::chrono::duration(Clock::now() - pl_t0).count(); }; + while (true) { + g_pl_diag.iters.fetch_add(1, std::memory_order_relaxed); + g_pl_diag.a_seq.store(A.seq, std::memory_order_relaxed); + g_pl_diag.a_bits.store(pw_bits(A), std::memory_order_relaxed); + g_pl_diag.b_bits.store(pw_bits(B), std::memory_order_relaxed); + g_pl_diag.d_bits.store(pw_bits(D), std::memory_order_relaxed); + g_pl_diag.chain_kind.store(chain_kind, std::memory_order_relaxed); + g_pl_diag.chain_live.store(mtp.chain_live() ? 1 : 0, std::memory_order_relaxed); + g_pl_diag.doomed.store(doomed ? 1 : 0, std::memory_order_relaxed); + g_pl_diag.ending.store(ending ? 1 : 0, std::memory_order_relaxed); + // ---- every window in flight served + if ((doomed && !pump0(D)) || !pump0(A) || !pump0(B)) return die(err); + if (A.s1 && !A.s1_done) { + strata::core::Verifier& v = V1(A); + if (v.service(&drive_pool_split, SDf(A), err) < 0) return die(err); + if (v.done(err)) { + if (!v.pl_finish(outp.data(), err)) return die(err); + A.s1_done = true; + tre("F1", A.seq, A.T); + } else if (!err.empty()) return die(err); + } + if (drive.d.failed) return die(drive.d.fail ? drive.d.fail : "the expert pool failed"); + // ---- a wrong speculative window has finished: stage 0 back to the verified window's real commit + if (doomed && D.finished) { + { + const strata::core::OnDevice on(dev0); + if (cudaMemcpyAsync(ss.gdn_state, pl_snap2[pl_snap_overlap ? (undo_w.seq & 1) : 0], + pl_snap_bytes, cudaMemcpyDeviceToDevice, s0) != cudaSuccess) + return die("restoring the GDN snapshot failed"); + } + ss.ple_prev[0] = undo_ple[0]; + ss.ple_prev[1] = undo_ple[1]; + if (!V0(undo_w).pl_commit_async(undo_keep, err)) return die(err); + doomed = false; + ++pl_undo; + tre("U", undo_w.seq, undo_keep); + } + if (ending) { + if (!doomed) break; + continue; + } + // ---- the drafter's chain, read as its outputs land (an event after each): a fresh A once its size + // is decided, then B once ITS size is decided - its bonus guess and a draft below spec_min_p (or + // S_mtp - 1 drafts) - so stage 0 need not wait for the whole chain + if (chain_kind != 0) { + const int k_ready = mtp.chain_outputs_ready(err); + if (k_ready < 0) return die(err); + if (chain_kind == 2 && !early_used) { + // A's size is decided by its first outputs: a draft below spec_min_p ends it (the serial + // loop's rule), so A goes to stage 0 as soon as that is known, not after all S_mtp - 1 + const int k = k_ready; + bool decided = k >= S_mtp - 1; + if (!decided && req_spec_min_p > 0.0) + for (int j = 0; j < k; ++j) + if (mtp.chain_prob()[j] < (float) req_spec_min_p) { decided = true; break; } + if (decided) { + A.T = t_rule(mtp.chain_prob(), 0, std::min(k, S_mtp - 1)); + for (int i = 1; i < A.T; ++i) { + A.tok[i] = mtp.chain_tok()[i - 1]; + A.prob[i - 1] = mtp.chain_prob()[i - 1]; + } + pick_lookup(A, mtp.chain_tok()); + A.ready = true; + early_used = true; + tre("CE", A.seq, A.T); + } + } + const int r = mtp.chain_poll(err); + if (r < 0) return die(err); + const int k = r == 1 ? chain_n : k_ready; + const int32_t* oc = mtp.chain_tok(); + const float* op = mtp.chain_prob(); + if (r == 1 && chain_kind == 2 && !early_used) { + A.T = t_rule(op, 0, std::min(chain_n, S_mtp - 1)); + for (int i = 1; i < A.T; ++i) { + A.tok[i] = oc[i - 1]; + A.prob[i - 1] = op[i - 1]; + } + pick_lookup(A, oc); + A.ready = true; + early_used = true; + } + if (!b_done && (chain_kind == 1 || early_used)) { + const int base = A.T - 1; // B's row 0: the guess of A's bonus + bool decided = r == 1 || k >= base + S_mtp; + if (!decided && k > base + 1 && req_spec_min_p > 0.0) + for (int j = base + 1; j < k; ++j) + if (op[j] < (float) req_spec_min_p) { decided = true; break; } + if (decided) { + float pa = 1.0f; + if (chain_kind == 1) { + // forced through A's drafts: its outputs there are the drafter's own picks from + // A's verified predecessor, a better estimate of A's acceptance than the chain that + // drafted A; a disagreement means A is unlikely to be accepted whole + bool agree = true; + for (int i = 0; i + 1 < A.T; ++i) { + if (oc[i] != A.tok[i + 1]) agree = false; + pa *= pl_agree ? op[i] : A.prob[i]; + } + if (pl_agree && !agree) { pa = 0.0f; ++pl_disagree; } + } else { + for (int i = 0; i + 1 < A.T; ++i) pa *= A.prob[i]; + } + if (!A.sfx) make_b(oc, op, base, k, pa); // a lookup window's bonus is not the chain's + // stage B now (its PLE rows from the tokens before it as they will be once A is + // committed whole), so its launch behind A is only the graph launch (never while a + // rollback is pending: B's verifier is the one the undo commit reads) + if (B.ready && B.p_on >= theta && pl_prestage && !V0(B).in_flight() && !doomed) { + int32_t prev[2] = {ss.ple_prev[0], ss.ple_prev[1]}; + for (int i = 0; i < A.T; ++i) { prev[0] = prev[1]; prev[1] = A.tok[i]; } + if (!V0(B).prestage(B.T, B.tok, B.p, prev, err)) return die(err); + } + b_done = true; + tre("CD", A.seq, chain_kind, (int) (1000.0f * B.p_on)); + if (pl_log) + std::fprintf(stderr, "strata pipeline: chain %d for p=%lld T=%d: B T=%d p_on %.3f (%d " + "outputs)\n", chain_kind, (long long) A.p, A.T, B.T, B.p_on, k); + } + } + if (r == 1) chain_kind = 0; + } + // ---- stage 0: A (never while a wrong window still holds stage 0's state) + if (A.ready && !A.launched && !doomed) { + if (A.p + A.T > o.max_context) { ending = true; continue; } + if (!snap_take(A.seq)) return die("the GDN snapshot failed"); + if (!V0(A).pl_launch(A.T, A.tok, A.p, err)) return die(err); + A.launched = true; + tre("L0", A.seq, A.T, 0); + } + // ---- stage 0: B, speculatively, right behind A + if (B.ready && !B.launched && A.finished && !A.committed && !doomed) { + if (B.p_on >= theta && B.p + B.T <= o.max_context) { + { + const strata::core::OnDevice on(dev0); + if (pl_snap_overlap ? cudaStreamWaitEvent(s0, pl_snap_ev[A.seq & 1], 0) != cudaSuccess + : cudaMemcpyAsync(pl_snap2[0], ss.gdn_state, pl_snap_bytes, + cudaMemcpyDeviceToDevice, s0) != cudaSuccess) + return die("the GDN snapshot failed"); + } + undo_ple[0] = ss.ple_prev[0]; + undo_ple[1] = ss.ple_prev[1]; + if (!V0(A).pl_commit_async(A.T, err) || !snap_take(B.seq) || + !V0(B).pl_launch(B.T, B.tok, B.p, err)) + return die(err.empty() ? std::string("the GDN snapshot failed") : err); + A.committed = true; + B.launched = true; + B.spec = true; + tre("L0", B.seq, B.T, 1); + ++pl_spec; + } else { + B.ready = false; // not worth it: A's verdict decides the next window + ++pl_gate; + } + } + // ---- stage 1: A, once its stage 0 is done + if (A.finished && !A.s1) { + if (chain_kind == 1 && !B.ready) ++pl_late; // stage 0 idles until the chain has B + if (!V1(A).pl_launch(A.T, A.tok, A.p, err)) return die(err); + A.s1 = true; + tre("L1", A.seq, A.T); + } + // ---- A's verdict + if (!A.s1_done) continue; + // the drafter must be idle before the next chain. The chain needs nothing from the host, so this + // wait is bounded even with B (made from its first outputs) in flight on stage 0; in practice the + // chain ended long before a verdict that needs a whole stage-1 window. + while (chain_kind != 0 && mtp.chain_live()) + if (mtp.chain_poll(err) < 0) return die(err); + if (chain_kind != 0 && !B.launched && !B.made) B = PW{}; + chain_kind = 0; + int a = 0; + while (a < A.T - 1 && A.tok[a + 1] == outp[(size_t) a]) ++a; + if (!V1(A).pl_commit_async(a + 1, err)) return die(err); + for (int i = 0; i <= a; ++i) consumed.push_back(A.tok[i]); + draft_offered += A.T - 1; + draft_accepted += a; + ++rounds; + ++dec_windows; + dec_T += A.T; + bool eos = false; + for (int i = 0; i <= a && produced_n < max_new && !eos; ++i) { + std::printf("T %d\n", (int) outp[(size_t) i]); + strata::core::progress_beat(); + ++produced_n; + if (o.suffix_draft > 0) sfx.append(outp[(size_t) i]); + eos = std::find(o.eos_ids.begin(), o.eos_ids.end(), (int64_t) outp[(size_t) i]) != o.eos_ids.end(); + } + std::fflush(stdout); + if (eos) finish = "stop"; + else if (stop_req.load()) finish = "cancel"; + const bool last = eos || produced_n >= max_new || stop_req.load(); + const bool on = B.launched && a == A.T - 1 && B.tok[0] == outp[(size_t) A.T - 1]; + if (B.made) { // the gate's calibration: would B have been on the path, by its estimate p_on + const bool would = a == A.T - 1 && B.tok[0] == outp[(size_t) A.T - 1]; + const int bin = std::min(9, std::max(0, (int) (B.p_on * 10.0f))); + cal_n[bin] += 1; + cal_on[bin] += would ? 1 : 0; + } + tre("V", A.seq, a, on ? 1 : (B.launched ? -1 : 0)); + { + const double now = ms_now(); + const int c = A.spec ? 1 : 0; + if (A.sfx) { ++sfx_windows; sfx_drafts += A.T - 1; sfx_ok += a; } + if (A.seq > 0 && !last) policy.observe(A.sfx, A.T, a, A.sfx_match, now - last_verdict); + cls_ms[c] += now - last_verdict; + cls_tok[c] += a + 1; + cls_n[c] += 1; + last_verdict = now; + } + if (pl_log) + std::fprintf(stderr, "strata pipeline: verdict p=%lld T=%d a=%d B %s%s\n", (long long) A.p, A.T, a, + B.launched ? "launched" : B.ready ? "ready" : "none", on ? " ON" : ""); + x = outp[(size_t) a]; + p = A.p + a + 1; + if (!last && !pl_adapt()) return die("an adaptive refill failed"); + if (on && !last) { + ++pl_on; + // the chain over A, forced through B's drafts: B's bonus guess and the next window's drafts + mtp.set_source_R(V1(A).final_R(0)); + chain_n = std::min(strata::kernels::kVerifyMaxT - 1, B.T + S_mtp - 1); + if (!mtp.chain_launch(A.T, outp.data(), A.p, a, B.tok + 1, B.T - 1, chain_n, chain_n, err)) + return die(err); + chain_kind = 1; + tre("CL", A.seq, 1); + early_used = false; + b_done = false; + A = B; + B = PW{}; + } else { + if (B.launched) { // a wrong guess (or the end): undo once it has finished + doomed = true; + D = B; + undo_w = A; + undo_keep = a + 1; + } else if (!A.committed) { + if (!V0(A).pl_commit_async(a + 1, err)) return die(err); + } + // B is D now (or dropped): never pump it twice (whichever copy saw the window complete first + // would finish it, and the other would wait for it forever) + B = PW{}; + if (!last) { + mtp.set_source_R(V1(A).final_R(0)); + chain_n = strata::kernels::kVerifyMaxT - 1; + if (!mtp.chain_launch(A.T, outp.data(), A.p, a, nullptr, 0, chain_n, chain_n, err)) + return die(err); + chain_kind = 2; + tre("CL", A.seq, 2); + early_used = false; + b_done = false; + PW nA; + nA.seq = A.seq + 1; + nA.p = A.p + a + 1; + nA.tok[0] = outp[(size_t) a]; + A = nA; + } + } + if (last) ending = true; + } + pl_drain(); // nothing in flight: both stages' commits and the drafter have landed + strata::core::diag_pipeline_fn().store(nullptr); + if (pl_adapt_thr.joinable()) pl_adapt_thr.join(); // its copies waited only for finished work + adapt_fence = nullptr; + if (exch_wait) pl_release(); // every window has completed (drained) + for (cudaEvent_t ev : fence_ev) if (ev) cudaEventDestroy(ev); + for (cudaEvent_t ev : exch_ev) if (ev) cudaEventDestroy(ev); + if (pl_trace_path != nullptr && !pl_trace.empty()) { + if (std::FILE* f = std::fopen(pl_trace_path, "a")) { + std::fprintf(f, "# request p0=%lld\n", (long long) (n - 1)); + std::fwrite(pl_trace.data(), 1, pl_trace.size(), f); + std::fclose(f); + } + } + // the odd windows' counters and GPU profiles into the verifiers that report them + PV[0][0]->absorb_stats(*PV[0][1]); + PV[1][0]->absorb_stats(*PV[1][1]); + if (!pl_adapt_ok) { + std::printf("ERR an adaptive refill failed\n"); + return 1; + } + while (mtp.chain_live() && mtp.chain_poll(err) == 0) {} + mtp.set_source_R(ver.final_R_all()); // the serial loop's rows again + const double pl_ms = std::chrono::duration(Clock::now() - pl_t0).count(); + dt_run += pl_ms; + if (dec_timing) { + auto avg = [](double s_, double n_) { return n_ > 0 ? s_ / n_ : 0.0; }; + std::fprintf(stderr, "strata pipeline: %lld windows in %.0f ms (%.2f ms/window): %lld speculative, %lld on " + "the path, %lld rolled back, %lld below the gate (theta %.2f)\n", + (long long) dec_windows, pl_ms, avg(pl_ms, (double) dec_windows), (long long) pl_spec, + (long long) pl_on, (long long) pl_undo, (long long) pl_gate, theta); + std::fprintf(stderr, "strata pipeline classes: fresh %.0f windows %.2f ms %.2f tok | speculative %.0f " + "windows %.2f ms %.2f tok | forced-chain disagreements %lld, chain late %lld\n", + cls_n[0], avg(cls_ms[0], cls_n[0]), avg(cls_tok[0], cls_n[0]), cls_n[1], + avg(cls_ms[1], cls_n[1]), avg(cls_tok[1], cls_n[1]), (long long) pl_disagree, + (long long) pl_late); + std::string cal; + char b[48]; + for (int i = 0; i < 10; ++i) + if (cal_n[i] > 0) { + std::snprintf(b, sizeof b, " %.1f:%lld/%lld", i / 10.0, (long long) cal_on[i], (long long) cal_n[i]); + cal += b; + } + std::fprintf(stderr, "strata pipeline calibration (p_on decile: on/scored):%s\n", cal.c_str()); + } + } + while (!pl_ran && !cancelled && produced_n < max_new) { int T = S_mtp; if (req_spec_min_p > 0.0) { T = 1;