diff --git a/.agents/specs/expert-stream-device-slots.md b/.agents/specs/expert-stream-device-slots.md index 87c4696f0..1c4e0000a 100644 --- a/.agents/specs/expert-stream-device-slots.md +++ b/.agents/specs/expert-stream-device-slots.md @@ -14,6 +14,159 @@ platform may read it. correctness gate that would let us publish a number does not pass.** W0e ran on 2026-08-19 inside one `rc hold` on `dgx:gpu0` at source `9c783a8be`. +**W1 HAS LANDED, and it lands UNREACHED, which this section says out loud +because `## Nothing lands dead` allows the shape only when it is declared.** +`DeviceExpertSlotStore` (`include/vllm/model_executor/device_expert_slot_store.h`, +`src/vllm/model_executor/device_expert_slot_store.cpp`) allocates one contiguous +device arena through `vt::Backend::Alloc` plus one pinned host staging slot, and +`ExpertSlotStore` gains the `CommitSlot(int32_t, size_t)` without which that +class could not be filled by any caller. **Nothing selects the device store.** +`Qwen35ExpertStream::store_` is still a `std::unique_ptr` +and `Qwen35ExpertStream::Slice` still reads the concrete +`HostExpertSlotStore::Slot`, so no load can reach the new class; the wiring is +W2 of this row and is tracked by +[#1124](https://github.com/mudler/vllm.cpp/issues/1124). The entry under +`## Owed` records the same thing. The split from W2 was chosen by the dispatch +rather than by this spec, whose own recommendation stays "land W1 and W2 as one +pull request". + +* **G1: PASS, red-first and on a CPU `vt::Backend`.** The RED was taken with + everything present except the streamer's publish call, which is exactly the + defect #1124's third piece names: `test_device_expert_slot_store` returned + `Status: FAILURE!`, exit status 1, compile status 0 and no ENOSPC in the build + log. All four slices failed both halves of the comparison — against the host + store AND against the file — because the bytes sat in staging and never + reached the device slot. **Re-measured at the repaired head, where the suite is + 10 cases / 112 assertions rather than the 9 / 97 the first offering had: 10 + cases with 2 failed and 112 assertions with 18 failed.** With `CommitSlot` + called from `EnsureFile` the same binary is 10 cases / 112 assertions / 0 + failed at exit status 0. +* **The host path is byte-identical, which is W1's stop condition.** + `HostExpertSlotStore::SlotForWrite` still returns the slot itself, its + `CommitSlot` is a bounds-checked no-op, and no staging buffer is allocated, + touched or copied on that path. `test_host_expert_slot_store` is 9 cases / 203 + assertions / 0 failed, unchanged in count from before the contract change. +* **The PUBLISH arm of the fill's undo was UNGATED when W1 was first offered, + and the fresh review of [#1735](https://github.com/mudler/vllm.cpp/pull/1735) + found it (F1).** `store_.CommitSlot(...)` sits inside `EnsureFile`'s `try` so + that a failed publish takes the same `cache_.Invalidate` a failed read takes, + and that placement was argued in four places and attacked in none. Moving the + call to just after the `catch` left every suite GREEN, because no store in the + tree could fail a publish. It is now gated: `RecordingStore` takes a + `throw_on_commit` flag and `test_expert_streamer` carries "a PUBLISH that + throws leaves nothing resident either", which mirrors the existing case for the + `pread` arm. Red-first with the call moved out — 10 cases with 1 failed, 182 + assertions with 6 failed, exit status 1, compile status 0, no ENOSPC — and the + RED is the corruption itself rather than a proxy: `cache.IsResident(key)` stays + TRUE and the retry comes back `hit` with `filled` false, which is exactly the + "next acquisition is an ordinary HIT over a slot nobody published" the comment + predicts. Green with the call restored, re-measured at the repaired head: 12 + cases / 269 assertions / 0 failed, and `expert_streamer.cpp` restored + byte-identical by sha256. The arm becomes REACHABLE in W2, when a store whose + `CommitSlot` really copies to a device is selected; it is gated now because W1 + is where the placement was decided. +* **THE SAME WINDOW WAS OPEN AT TWO MORE ENTRY POINTS, and the second fresh + review of [#1735](https://github.com/mudler/vllm.cpp/pull/1735) found them + (F1).** `EnsureFile` was wrapped and `EnsureSpan` and `Ensure` were not, and + their `store_.WriteSlot(...)` calls are the identical corruption one step + earlier. W1 is what opens it: before this wave every store's `WriteSlot` was a + `memcpy` and could not throw, and `DeviceExpertSlotStore::WriteSlot` calls + `vt::Backend::Copy` and `Synchronize`, which throw `std::runtime_error` out of + the CUDA backend. **`EnsureSpan` is a PRODUCTION call site** — `qwen3_5.cpp`'s + `Qwen35ExpertStream::Slice` reaches it from `Qwen3_5Model::Forward` — so this + half is not the wait-for-W2 shape the publish arm has. Both now take the same + `try` / `catch (...) { cache_.Invalidate(key); throw; }` as `EnsureFile`. Gated + red-first by a `throw_on_write` flag on `RecordingStore` and one case per entry + point, each asserting CONSISTENCY rather than emptiness: the key is not + resident, `SlotOf` is empty, `resident()` is 0, the fill counters are unmoved, + the slot still holds the EVICTED expert's bytes, and the retry is a real MISS. + RED with neither `try` present — 12 cases with 2 failed, 232 assertions with 16 + failed, exit status 1, compile status 0, no ENOSPC — and the red is the + corruption itself: `retry.hit` is true and the slot holds expert 4's bytes under + expert 6's key. GREEN with both: 12 cases / 269 assertions / 0 failed. The two + `try`s are proven independent by mutation, because wrapping one leaves the other + exactly as exposed: M12 (delete `EnsureSpan`'s) reds the streamer suite at 12 + cases with 1 failed and 235 assertions with 8 failed, M13 (delete `Ensure`'s) at + 12 with 1 and 266 with 8. +* **The constructor leaked on the failure that happens and guarded one that + cannot (F2 of the same review).** No backend in this tree returns nullptr from + an allocator — `CpuBackend::Alloc` refuses with `VT_CHECK`, + `CudaBackend::Alloc` and `AllocPinned` through `Check(...)`, and the base + `Backend::AllocPinned` forwards to `Alloc` — so both nullptr branches were + unreachable while a throw from `Alloc` stranded the queue and a throw from + `AllocPinned` stranded the queue AND the whole device arena, 18.55 GiB on the + target checkpoint, at the moment the device has no memory left to lose. Out of + memory is this class's headline failure: + [#1123](https://github.com/mudler/vllm.cpp/issues/1123) is literally + `vt cuda: cudaMalloc: out of memory`. The acquisitions now sit in a `try` whose + `catch` runs the destructor's body and rethrows unchanged, so the caller still + sees the backend's own message. The nullptr branches are KEPT, deliberately: + `vt::Backend` is an interface, a nullptr-returning implementation would + otherwise hand out slot pointers off a null arena, and they now cost one branch + and no cleanup code because the catch owns the release. The header no longer + claims the constructor's own `std::runtime_error` is what an allocation failure + raises. Gated by `throw_on_alloc` and `throw_on_pinned_alloc` on the suite's + `CountingBackend`: RED first at 10 cases with 1 failed and 112 assertions with + 4 failed, GREEN at 112 / 0. Mutation M14 (drop the arena release from the catch) + reds at 112 assertions with 2 failed and M15 (drop the whole catch) at 112 with + 4. +* **`CommitSlot` is PURE on the interface, not a defaulted no-op.** A default + would be correct for exactly one implementation — the host one — and silently + wrong for every store whose slots the host cannot write, which is the entire + population the method was added for. The cost is two overrides: the host store + and `test_expert_streamer`'s `RecordingStore`, which now counts the calls so + a case can assert the streamer publishes exactly the fills it performed and + never a hit, a refused acquire or a failed read. +* **Two claims the first review corrected, neither of them a defect in the code.** + The G1 comparison against the FILE is not the thing that catches an unpublished + slot: the host arm is filled by its own streamer and is non-zero, so the + host-versus-device comparison reds on its own. Measured rather than conceded, + and re-measured at the repaired head — with both file `CHECK`s deleted AND the + H2D copy deleted the suite is still RED at 10 cases with 3 failed, 104 + assertions with 10 failed, exit status 1. The second correction is that + "byte-identical to the host store" is true over the bytes a fill WROTE and says + nothing past them: the host arena is a zero-filled `std::vector` and the device + arena is a raw `vt::Backend::Alloc` that is not initialised at all. The store's + header, the gate case and — since the second review's F3 — the normative G1 + definition under `## Gates` and the `## Tests to port` row all say so, which + matters because those last two are what a W2 or G-DISCRETE implementer reads to + learn what PASS means. Zeroing the device arena would cost a full write of the + whole budget at load — 18.55 GiB on the target checkpoint — to define bytes + the streamer never hands out. +* **The file `CHECK`'s stated reason was wrong, the check itself is right, and + two recorded mutation counts were allocator-dependent (F4 and F5 of the second + review).** The "each slot holds a DIFFERENT slice" assertion compared two device + slots that a publish-suppressing mutation leaves UNWRITTEN, so its outcome under + mutation was decided by `std::aligned_alloc` garbage: the review measured M4 at + 112 assertions with 18 failed and the F2 combination at 104 with 10 where the + record said 17 and 9, and both deltas were that one assertion. That is the very + allocator non-determinism the F2 correction invokes as its justification, + appearing inside the gate's own assertions. The gate now writes every device + slot to a known byte before the fills, so an unmutated fill is the only thing + that can make two slots differ. Proven deterministic rather than asserted: under + M4 the assertion at `test_device_expert_slot_store.cpp:374` fails on 25 + consecutive runs and the suite reads 112 with 18 failed on all 25. The file + `CHECK` is KEPT, on the stronger ground the review named: host-arm-against- + device-arm is a SHARED-HELPER comparison, both arms running the same + `ExpertStreamer` over the same descriptor at the same `file_offset`, so a + streamer that read the wrong offset, read short, or read one slice twice makes + both arms identically wrong and passes it. The bytes on disk are the only input + neither arm computed. **Measured, not argued:** mutation M16 makes + `EnsureFile`'s `pread` ignore `file_offset`, so every key in both arms fills + from offset 0. The host-against-device assertion at + `test_device_expert_slot_store.cpp:359` stays GREEN through it, and the only + assertions that red are the two file `CHECK`s at `:370` and `:371` and the + different-slice check at `:375` — 10 cases with 2 failed, 112 assertions with + 12 failed, exit status 1, compile status 0, tree restored byte-identical by + sha256. +* **The gate file is `tests/vllm/model_executor/test_device_expert_slot_store.cpp`, + not the `test_expert_slot_store.cpp` this spec's `## Tests to port` table named + when it was written.** Stated rather than done quietly: the header it gates is + `device_expert_slot_store.h`, and the suite sits beside a + `test_host_expert_slot_store.cpp` that is its ORACLE, so a name that does not + say which store it is about would be the one thing a reader has to + disambiguate every time. The table below is corrected to match the tree. + * **G0-LIVE: PASS.** 32/32 steps where seven previous attempts produced ZERO; decode-phase `exhausted` delta **0** (6077 at step 1 and at step 32; the total is the structural prefill number this spec predicted); `W0E_DOCKER_RC=0`, no @@ -460,8 +613,13 @@ records parsed against 1702 declared). is the only production `ExpertSlotStore`; the only other subclass is a test double. `include/vllm/model_executor/expert_streamer.h:8-9,30-31` says "the production destination is a contiguous device-side slot array" and "production - writes to device memory". **Both sentences are false today**, and W1 makes them - true rather than adding a second claim beside them. + writes to device memory". **Both sentences are false today.** W1 was written + here as "makes them true"; what it actually did is replace them, and the + difference is worth the line. A second production `ExpertSlotStore` existing + does not make "the production destination is a contiguous device-side slot + array" true, because production still selects the host one — that is W2. The + header now names both implementations and says which one anything reaches, + which is the correction the false sentences needed. 2. **No device-capable read.** `Qwen35ExpertStream` holds `std::unique_ptr store_` (`qwen3_5.cpp:5621`) and reads `store_->Slot(r.slot)` at `:5381` and `:5437` — the CONCRETE class. There is no @@ -579,7 +737,8 @@ mutation: | Test | Proves | Mutation that must red it | |---|---|---| | `tests/vllm/platforms/test_platform.cpp` (extend) | the new predicate defaults false and the CUDA/ROCm assembly threads the probed value | flip the default to true; drop the assignment | -| `tests/vllm/model_executor/test_expert_slot_store.cpp` (new) | a device-flavoured store filled via `EnsureFile` yields byte-identical slot content to the host store | delete `CommitSlot`'s copy; return staging from `SlotForRead` | +| `tests/vllm/model_executor/test_device_expert_slot_store.cpp` (new, W1) | a device store filled via `EnsureFile` yields slot content byte-identical to the host store's **over the bytes a fill wrote** (the device arena is uninitialised where the host arena is zero-filled; see G1); the arena is ONE device allocation and staging is ONE pinned slot; `SlotForWrite` hands out staging and `SlotForRead` hands out the slot; a hit, a refused acquire and a failed read publish nothing; the host path is unchanged | delete `CommitSlot`'s copy; delete its `Synchronize`; return the slot from `SlotForWrite`; return staging from `SlotForRead`; delete the `CommitSlot` call in `EnsureFile`; delete the staged-slot identity check; drop the overflow guard; acquire the queue above the budget refusals | +| `tests/vllm/model_executor/test_expert_streamer.cpp` (extend, W1) | the streamer publishes exactly the fills it performed and never a hit, a refused acquire or a failed read; a publish that THROWS undoes the acquisition, so the key is not left resident over a slot nobody published | publish on the hit path; move `store_.CommitSlot(...)` out of `EnsureFile`'s `try` | | `tests/vllm/model_executor/test_gguf_device_fit.cpp` (extend) | with the lane on, the bound excludes `*_exps` and adds the arena; with it off, the bound is byte-identical to today | make the exclusion unconditional | | `tests/vllm/entrypoints/test_gguf_device_fit_reach.cpp` (extend) | the loader reaches the conditional refusal from the production entry point | delete the production call site | | a `qwen3_5` slot-arm unit gate | the slot branch never calls `ResidentWeight`, and a streamed tower reaching device staging throws by name | remove the `VT_CHECK`; restore the `ResidentWeight` call | @@ -622,8 +781,16 @@ real, publishable result that closes the unified shortcut — recorded in `docs/BENCHMARKS.md` as a measured negative, not as a failure to be tuned away. **G1 (W1).** `DeviceExpertSlotStore` driven through `ExpertStreamer::EnsureFile` -produces byte-identical slot contents to `HostExpertSlotStore` on the same -input, on a CPU `vt::Backend`, red-first and mutation-proven per the table above. +produces slot contents byte-identical to `HostExpertSlotStore`'s **over the +bytes a fill WROTE**, on the same input, on a CPU `vt::Backend`, red-first and +mutation-proven per the table above. The qualification is normative and not a +caveat: the host arena is a zero-filled `std::vector` and the device arena is a +raw `vt::Backend::Alloc` that is not initialised at all, so past a fill's last +byte the two stores are asymmetric and nothing promises otherwise. Every fill +this gate performs writes a whole slot, so here the written prefix IS the slot; +a caller streaming a SHORT slice into a full-sized slot would find them +disagreeing past the slice, and a W2 or G-DISCRETE gate written against the +unqualified sentence would be gating a property the class does not have. **G2 (W2, reachability).** Per `## Nothing lands dead`: delete the production selection of the device store in a scratch copy and rerun the focused gate. A @@ -1045,6 +1212,31 @@ name W2 as the owning wiring. **The cheaper and more honest shape is to land W1 and W2 as one pull request**, and that is the recommendation here; splitting them is a scheduling choice that costs an explicitly-declared unreached slice. +**LANDED, split from W2, and the cost the paragraph above predicted was paid in +full.** `DeviceExpertSlotStore` is unreached, it is declared in the commit body, +the pull request body and `## Owed`, and the class carries the reachability +statement in its own header comment as well. One qualifier belongs here rather +than in a summary, because it is the part a reader would otherwise get wrong in +the generous direction: the CONTRACT half is not in the same position as the +class. `store_.CommitSlot(acq.slot, bytes)` sits inside +`ExpertStreamer::EnsureFile`, which IS a production call site — +`qwen3_5.cpp`'s `Qwen35ExpertStream::Slice` reaches it from +`Qwen3_5Model::Forward` — so the line executes on every real streamed fill +today. What it does there is nothing, because the store production selects is +the host one and its `CommitSlot` is a no-op. So the reachability mutation on +that call site reds the new suite and CANNOT red the model path, and reporting +it as reach would be reading a call count as a capability. Both halves wait on +W2 for a consumer. + +**One design decision was taken inside the wave and is recorded rather than +left in the diff.** `CommitSlot` is PURE on `ExpertSlotStore`. A defaulted +no-op would have cost nothing at the two existing implementations and would have +been silently wrong at the next one: the failure it hides — a device store that +compiles, fills its staging buffer and publishes nothing — is precisely the RED +this wave was gated on, and it presents as zeros in a slot rather than as a +compile error. The two overrides a pure method costs are one no-op and one +counter. + **Stop condition:** if `CommitSlot` cannot be added without changing every existing `ExpertSlotStore` caller's contract in a way that alters host-path behaviour, stop — the host path must stay byte-identical, and a change that @@ -1098,4 +1290,4 @@ re-derived here. | **The family-wide copy of this change: `include/vllm/model_executor/models/dense_attn_block.h`'s `ResidentWeight` still stages unconditionally.** The measurement: on a host-addressable staging platform, load any of the ~50 models that include that header and show peak resident bytes falling by the model's weight size, with tokens unchanged. | W0f deliberately changes only `qwen3_5.cpp`'s PRIVATE copy, which is the one that governs `Qwen3.8-2.4T-A95B UD-Q1_0` (that file kept its own helper; the header's copy is not on the Qwen3.5 path). The header's version is reached from `ModelRegistry::Forward` for every model that includes it, so extending it is not dead code — but nothing on a CPU tier can drive one of those forwards on a staging platform, so the extension would land with its reachability argued rather than gated, across ~50 architectures at once. That is a scope and a review question, not a line of code, and it gets its own row. | | **The missing CPU-platform gate on `p.quant_repack` itself ([#1320](https://github.com/mudler/vllm.cpp/issues/1320)).** The measurement: `elem_kn_repack` is resolved with `CurrentPlatform().device_type() == kCPU` and `quant_repack` is not, so a device load can still perform a CPU-only transform and be caught afterwards instead of never doing it. | W0f fixes the CONSEQUENCE in flow — a named refusal on both arms of `ResidentWeight`, red-first and mutation-proven — because that is the small and clear part. Moving the gate into the loader policy changes what a GGUF load DOES on a device rather than what it refuses, which is `QUANT-GGUF-KEEPQ-LOADER`'s semantics and needs its own red-first evidence. | | **`.agents/specs/expert-streaming.md`'s `## Owed` entry for #1124 still names no owning row ID.** | Not edited here on purpose; PRs #1200 and #1216 both edit that file. One-line follow-up once both land. | -| **W1 may land UNREACHED if it is split from W2.** | The recommendation is one pull request. If a split is chosen, the commit body and the PR body must name what is unreached and name W2 as the owning wiring, per `## Nothing lands dead`. | +| **W1 LANDED UNREACHED, and W2 owns the wiring.** What is not reached: `DeviceExpertSlotStore` — no loader, no model and no registered command constructs one, because `Qwen35ExpertStream::store_` is still a `std::unique_ptr` and `Qwen35ExpertStream::Slice` reads the concrete `HostExpertSlotStore::Slot`. Owning row: `ENG-EXPERT-STREAM-DEVICE`, wave W2. Tracking issue: [#1124](https://github.com/mudler/vllm.cpp/issues/1124), which stays OPEN. | The split from W2 was a dispatch decision, not this spec's recommendation, which is still one pull request. Named in the landing commit body and the pull request body as well as here, because `## Nothing lands dead` requires all three and the spec alone is not the disclosure. The narrower half: the `CommitSlot` CALL is on a production path (`ExpertStreamer::EnsureFile`, reached from `Qwen3_5Model::Forward` via `Qwen35ExpertStream::Slice`) and executes on every streamed fill, but its effect is a no-op until a store that needs it is selected, so the call site's reachability is not the class's. | diff --git a/CMakeLists.txt b/CMakeLists.txt index af1e6bfac..47cce6e7f 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -744,6 +744,7 @@ add_library(vllm STATIC src/vllm/model_executor/expert_slot_cache.cpp src/vllm/model_executor/model_loader/gguf_expert_span.cpp src/vllm/model_executor/expert_streamer.cpp + src/vllm/model_executor/device_expert_slot_store.cpp src/vllm/config/multimodal.cpp src/vllm/config/speculative.cpp src/vllm/outputs.cpp diff --git a/include/vllm/model_executor/device_expert_slot_store.h b/include/vllm/model_executor/device_expert_slot_store.h new file mode 100644 index 000000000..02516e000 --- /dev/null +++ b/include/vllm/model_executor/device_expert_slot_store.h @@ -0,0 +1,158 @@ +// vllm.cpp original (ENG-EXPERT-STREAM-DEVICE W1, issue #1124, spec +// .agents/specs/expert-stream-device-slots.md). A DEVICE-memory +// `ExpertSlotStore`. +// +// There is no upstream for this file. Pinned vLLM `555967922` has no +// inference-time expert paging at all — `vllm/model_executor/offloader/uva.py` +// is a CPU-blanket UVA offloader over whole parameters and +// `vllm/model_executor/offloader/prefetch.py` is cpu-only — so nothing here is +// mirrored and nothing here may claim to be. The reference for correctness is +// `HostExpertSlotStore` on the same input, which is what gate G1 compares +// against. +// +// WHY THIS EXISTS. `HostExpertSlotStore` is the only production destination +// today, so `--device cuda` serves this checkpoint only where the platform's +// kernels can dereference host storage (W0, keyed on the probed +// `host_memory_is_device_addressable()`). That is one part, a GB10. A DISCRETE +// device cannot read the host arena at all, and for it the slice has to live in +// device memory. This is that store. +// +// WHY IT CANNOT SIMPLY BE FILLED. `ExpertStreamer::EnsureFile` hands +// `SlotForWrite()`'s pointer straight to `::pread`, and +// `vt::Backend::DeviceMemoryIsHostAddressable()` is false for CUDA, so a device +// slot pointer is not a legal `pread` destination. That is why W1 carries a +// fill-contract change rather than deferring it: without `CommitSlot` this +// class could not be filled AT ALL, so planning the contract as a later wave +// would plan a wave that deadlocks its predecessor (spec, "Verdict on issue +// #1124's piece 3"). +// +// THE FILL IS A STAGING BOUNCE, BY CHOICE AND NOT BY DEFAULT. +// `SlotForWrite` returns ONE pinned host slot, `pread` fills it exactly as it +// fills a host slot today, and `CommitSlot` performs the single contiguous H2D +// into the device slot. A true zero-copy filler (GPUDirect Storage / `cuFile`, +// or `O_DIRECT` DMA into a device BAR mapping) moves fewer bytes and needs a +// driver capability probe, a mount-level check, an aligned-I/O path and a +// fallback for each of those; the bounce costs one extra host-to-device copy of +// one slice per MISS, on top of a disk read of the same size. The measurement +// that would justify replacing it is a device-arm decode where the H2D leg is a +// measurable fraction of fill time, and that measurement does not exist yet. +// Recorded under `## Owed` in the spec, not decided here. +// +// ONE STAGING SLOT, NOT `slots` OF THEM. The filler is synchronous by design +// (`expert_streamer.h`: no async I/O, no prefetch, no read-ahead; overlap is +// `ENG-EXPERT-STREAM` W6 and is conditional on a measurement), so exactly one +// fill is ever in flight. A staging slot per device slot would double the +// arena's host cost — 18.55 GiB on the target checkpoint — to buffer a +// concurrency that does not exist. The single buffer is therefore also the +// thing that makes the write/commit pairing a CONTRACT rather than a +// suggestion, and `CommitSlot` refuses a slot that is not the one `SlotForWrite` +// last handed out instead of committing another expert's bytes. +#ifndef VLLM_MODEL_EXECUTOR_DEVICE_EXPERT_SLOT_STORE_H_ +#define VLLM_MODEL_EXECUTOR_DEVICE_EXPERT_SLOT_STORE_H_ + +#include +#include + +#include "vllm/model_executor/expert_streamer.h" +#include "vt/backend.h" + +namespace vllm { + +class DeviceExpertSlotStore final : public ExpertSlotStore { + public: + // `slot_bytes` must be the LARGEST expert slice the caller will stream and is + // fixed for the store's life, exactly as it is for the host store: a ragged + // budget makes eviction unpredictable, and a slice that does not fit is + // refused by the streamer rather than silently truncated. + // + // The arena is ONE allocation of `slots * slot_bytes` through + // `vt::Backend::Alloc`, so a slot is a fixed offset into a contiguous device + // block — the shape `expert_streamer.h` always described. Throws + // std::invalid_argument on a degenerate budget or one whose product overflows + // `size_t`. An allocator that fails PROPAGATES ITS OWN EXCEPTION unchanged -- + // that is how every backend here reports failure, `VT_CHECK` on the CPU + // backend and `Check(cudaMalloc)`/`Check(cudaHostAlloc)` on CUDA -- and the + // std::runtime_error this constructor raises itself is reserved for a backend + // that reports failure by returning nullptr instead, which none in this tree + // does. Either way NOTHING IS LEAKED: the queue and the arena are given back + // before the exception leaves, because a constructor that throws runs no + // destructor. + DeviceExpertSlotStore(vt::Backend& backend, int32_t slots, size_t slot_bytes); + ~DeviceExpertSlotStore() override; + + DeviceExpertSlotStore(const DeviceExpertSlotStore&) = delete; + DeviceExpertSlotStore& operator=(const DeviceExpertSlotStore&) = delete; + + size_t slot_bytes() const override { return slot_bytes_; } + int32_t slot_count() const override { return slots_; } + + // Copy `bytes` from a HOST buffer into the device slot. This is the path + // `Ensure`/`EnsureSpan` take, where the caller already holds the bytes, so no + // staging is involved: the backend copy is itself the H2D. + void WriteSlot(int32_t slot, const uint8_t* src, size_t bytes) override; + + // The STAGING buffer, not the slot. `pread` writes here; the bytes reach the + // device only in `CommitSlot`. The returned pointer is the same for every + // slot and is never inside the arena, which is what makes forgetting the + // commit a visible defect rather than a silent one. + uint8_t* SlotForWrite(int32_t slot) override; + + // The single contiguous H2D of the staged bytes into `slot`, then a queue + // synchronize. The synchronize is load-bearing: `vt::Backend::Copy` is + // `cudaMemcpyAsync` on CUDA, and both the reuse of the one staging buffer by + // the next fill and the GEMM that reads the slot immediately after + // `EnsureFile` returns would otherwise race the transfer. + // + // Throws std::logic_error when `slot` is not the slot `SlotForWrite` last + // handed out, because with one staging buffer that mismatch commits another + // expert's bytes under this slot's key — the same silent-and-plausible + // corruption the streamer's `Invalidate` on a failed read exists to prevent. + void CommitSlot(int32_t slot, size_t bytes) override; + + // The DEVICE bytes of `slot`, for a kernel to read in place. Not host + // memory: only the backend may dereference it. Non-virtual here on purpose — + // making the read virtual on `ExpertSlotStore`, so that + // `Qwen35ExpertStream::Slice` stops reading the concrete + // `HostExpertSlotStore`, is W2's change and this wave does not pre-empt it. + uint8_t* SlotForRead(int32_t slot); + + // Device bytes held. The arena is allocated once and never grows, so this is + // the whole device cost of the lane. + // + // It is NOT initialised. `vt::Backend::Alloc` returns raw device memory -- + // `cudaMalloc` on CUDA, `std::aligned_alloc` on the CPU backend -- so a + // slot holds whatever the allocator last left there until something fills it, + // and the host store differs here: its arena is a `std::vector` and + // is zero-filled at construction. A slot's contents therefore match the host + // store's over the bytes a fill WROTE and are unspecified past them. Zeroing + // the arena would cost a full write of the whole budget at load -- 18.55 GiB + // on the target checkpoint -- to define bytes no reader may look at, since the + // streamer never hands out a slot it has not filled. Stated because + // "byte-identical to the host store" is this row's gate and it is true over + // the filled prefix rather than over the slot. + int64_t resident_bytes() const { + return static_cast(static_cast(slots_) * slot_bytes_); + } + + private: + uint8_t* SlotPtr(int32_t slot) const; + + vt::Backend& b_; + // Default-constructed, then replaced in the constructor BODY: every budget + // refusal has to happen before anything is acquired, because a constructor + // that throws runs no destructor. What is acquired after them is released by + // the constructor's own catch, for the same reason. + vt::Queue q_; + int32_t slots_; + size_t slot_bytes_; + uint8_t* arena_ = nullptr; + uint8_t* staging_ = nullptr; + // Which slot `SlotForWrite` last handed the staging buffer to, or -1 when + // nothing is staged. Reset by every commit, so a second commit of the same + // slot is refused too. + int32_t staged_slot_ = -1; +}; + +} // namespace vllm + +#endif // VLLM_MODEL_EXECUTOR_DEVICE_EXPERT_SLOT_STORE_H_ diff --git a/include/vllm/model_executor/expert_streamer.h b/include/vllm/model_executor/expert_streamer.h index f715d8dea..b485e4907 100644 --- a/include/vllm/model_executor/expert_streamer.h +++ b/include/vllm/model_executor/expert_streamer.h @@ -4,12 +4,23 @@ // whether it is already there, the span says which bytes it is, and this fills // the slot on a miss. // -// WHY THE DESTINATION IS AN INTERFACE. The production destination is a -// contiguous device-side slot array, so a concrete device dependency here would -// make the whole streaming policy untestable without a GPU, and this row's -// hardware is the scarcest thing about it. `ExpertSlotStore` is the seam: -// production writes to device memory, tests write to a host buffer, and the -// policy above it is identical in both. +// WHY THE DESTINATION IS AN INTERFACE. A destination may be a contiguous +// device-side slot array, so a concrete device dependency here would make the +// whole streaming policy untestable without a GPU, and this row's hardware is +// the scarcest thing about it. `ExpertSlotStore` is the seam and the policy +// above it is identical for every implementation. +// +// WHICH DESTINATION PRODUCTION ACTUALLY USES, as of ENG-EXPERT-STREAM-DEVICE W1 +// (issue #1124). Two sentences here used to say "the production destination is +// a contiguous device-side slot array" and "production writes to device +// memory". Both were false: `HostExpertSlotStore` was the only production +// implementation, and it still is the only one anything SELECTS. +// `DeviceExpertSlotStore` (device_expert_slot_store.h) now exists and is +// filled through `EnsureFile` below, but `Qwen35ExpertStream` still holds the +// concrete host store and reads it through `HostExpertSlotStore::Slot`, so no +// load can reach the device one yet. Making the read virtual and selecting the +// store from the platform is W2 of the same row, and until it lands this +// comment says what is true rather than what is intended. // // WHAT THIS DELIBERATELY DOES NOT DO. It performs no asynchronous I/O, no // prefetch and no read-ahead. A miss is a synchronous fill. That is the spec's @@ -45,10 +56,36 @@ class ExpertSlotStore { // streamer has already validated the slot index and the size. virtual void WriteSlot(int32_t slot, const uint8_t* src, size_t bytes) = 0; - // The slot's writable bytes, for a filler that produces them in place (a - // pread writes straight here, so the data never passes through a staging - // buffer). Must return at least `slot_bytes()` writable bytes or throw. + // A HOST-WRITABLE destination for `slot`'s bytes, for a filler that produces + // them in place: `pread` writes straight here. Must return at least + // `slot_bytes()` writable bytes or throw. + // + // It is NOT required to be the slot itself, and on a device store it is not: + // a `cudaMalloc` pointer is not a legal `pread` destination, because + // `vt::Backend::DeviceMemoryIsHostAddressable()` is false for CUDA. The + // host store returns the slot and the device store returns a staging buffer; + // `CommitSlot` below is what makes the difference invisible to the filler. virtual uint8_t* SlotForWrite(int32_t slot) = 0; + + // Publish the `bytes` a filler just wrote through `SlotForWrite(slot)`, so + // that a later read of `slot` sees them. + // + // WHY THIS EXISTS AT ALL, since a host store needs nothing here. Without it + // `ExpertSlotStore` cannot describe a destination that is not host memory, + // and a device store could not be filled by ANY caller — the fill contract + // was `pread`-into-the-slot and nothing else (issue #1124, piece 3). It is + // therefore part of W1 rather than a wave after it: deferring it would defer + // the only thing that makes the class fillable. + // + // PURE, not a defaulted no-op. A default would be correct for exactly one + // implementation, the host one, and silently wrong for every store whose + // slots the host cannot write — which is the whole population this method was + // added for. An implementer who does nothing must say so. + // + // Called by `ExpertStreamer::EnsureFile` on the SUCCESS path only: a fill + // that threw leaves nothing to publish, and the cache entry is invalidated + // instead. + virtual void CommitSlot(int32_t slot, size_t bytes) = 0; }; class ExpertStreamer { diff --git a/include/vllm/model_executor/host_expert_slot_store.h b/include/vllm/model_executor/host_expert_slot_store.h index 7ad45e2fb..b3fe22c09 100644 --- a/include/vllm/model_executor/host_expert_slot_store.h +++ b/include/vllm/model_executor/host_expert_slot_store.h @@ -59,6 +59,26 @@ class HostExpertSlotStore final : public ExpertSlotStore { // through it. uint8_t* SlotForWrite(int32_t slot) override { return Slot(slot); } + // Nothing to publish: `SlotForWrite` returned the slot itself, so `pread` + // already wrote the bytes where a reader will look for them. The bounds check + // is kept because it is the only thing this override can still get wrong, and + // because an out-of-range commit means the streamer and the store disagree + // about which slot was filled. + // + // This is what keeps the host path BYTE-IDENTICAL across the W1 contract + // change (spec G1's premise, and the wave's stop condition): the direct + // `pread`-into-slot fill is unchanged and no staging buffer is allocated, + // touched, or copied on this path. + void CommitSlot(int32_t slot, size_t bytes) override { + if (slot < 0 || slot >= slots_) + throw std::out_of_range("HostExpertSlotStore: slot " + + std::to_string(slot) + " out of range"); + if (bytes > slot_bytes_) + throw std::invalid_argument("HostExpertSlotStore: commit of " + + std::to_string(bytes) + + " bytes exceeds the slot"); + } + uint8_t* Slot(int32_t slot) { if (slot < 0 || slot >= slots_) throw std::out_of_range("HostExpertSlotStore: slot " + diff --git a/src/vllm/model_executor/device_expert_slot_store.cpp b/src/vllm/model_executor/device_expert_slot_store.cpp new file mode 100644 index 000000000..cdd116454 --- /dev/null +++ b/src/vllm/model_executor/device_expert_slot_store.cpp @@ -0,0 +1,150 @@ +// ENG-EXPERT-STREAM-DEVICE W1 (issue #1124). See device_expert_slot_store.h for +// why the destination is device memory, why the fill is a staging bounce, and +// why there is exactly one staging buffer. +#include "vllm/model_executor/device_expert_slot_store.h" + +#include +#include +#include + +namespace vllm { + +DeviceExpertSlotStore::DeviceExpertSlotStore(vt::Backend& backend, + int32_t slots, size_t slot_bytes) + : b_(backend), slots_(slots), slot_bytes_(slot_bytes) { + // Every BUDGET refusal below happens before the queue is created, + // deliberately: a constructor that throws has no destructor run, so anything + // acquired above the throw leaks. Nothing is acquired until the budget is + // known good. The acquisitions themselves throw too, and they are wrapped for + // the same reason -- see the note above them. + if (slots <= 0) { + throw std::invalid_argument( + "DeviceExpertSlotStore: slot count must be > 0"); + } + if (slot_bytes == 0) { + throw std::invalid_argument("DeviceExpertSlotStore: slot bytes must be > 0"); + } + // The host store's arena is a `std::vector`, whose own length check catches + // this; a raw `Alloc` has no such backstop, and an arena that wrapped would + // hand out in-range slot pointers past its end — a silent overwrite rather + // than a refusal. + if (slot_bytes > std::numeric_limits::max() / + static_cast(slots)) { + throw std::invalid_argument( + "DeviceExpertSlotStore: " + std::to_string(slots) + " slots of " + + std::to_string(slot_bytes) + " bytes overflows size_t"); + } + + const size_t total = static_cast(slots) * slot_bytes; + // ACQUISITION IS ALL-OR-NOTHING FROM HERE DOWN, and the failure it is written + // for is the one that actually happens. No backend in this tree returns + // nullptr from an allocator: `CpuBackend::Alloc` refuses with `VT_CHECK`, + // `CudaBackend::Alloc` and `AllocPinned` refuse through `Check(...)`, and the + // base `Backend::AllocPinned` forwards to `Alloc`. They THROW, and out of + // memory is this class's headline failure -- issue #1123 is literally + // `vt cuda: cudaMalloc: out of memory`. Unwrapped, a throw from `Alloc` + // strands the queue and a throw from `AllocPinned` strands the queue and the + // whole device arena, 18.55 GiB on the target checkpoint, at the exact moment + // the device has no memory left to lose. The catch below gives back whatever + // this constructor took and rethrows unchanged, so the caller still sees the + // backend's own message. + q_ = b_.CreateQueue(); + try { + arena_ = static_cast(b_.Alloc(total)); + // Kept although unreachable through any backend here, because `vt::Backend` + // is an interface and a nullptr-returning implementation would otherwise + // hand out slot pointers off a null arena instead of being refused. It now + // costs one branch and no cleanup code, since the catch owns the release. + if (arena_ == nullptr) { + throw std::runtime_error("DeviceExpertSlotStore: device allocation of " + + std::to_string(total) + " bytes failed"); + } + // Page-locked, because this buffer is the source of every H2D the lane + // issues: on CUDA a copy from pageable memory stages through a driver + // bounce of its own, which is the one thing this design must not pay twice. + // The base implementation returns ordinary host memory, which is correct on + // a backend where the distinction does not exist. + staging_ = static_cast(b_.AllocPinned(slot_bytes)); + if (staging_ == nullptr) { // unreachable here for the same reason + throw std::runtime_error("DeviceExpertSlotStore: pinned staging " + "allocation of " + std::to_string(slot_bytes) + + " bytes failed"); + } + } catch (...) { + // The destructor's body, because that is exactly what did not run. + if (staging_ != nullptr) b_.FreePinned(staging_); + if (arena_ != nullptr) b_.Free(arena_); + b_.DestroyQueue(q_); + throw; + } +} + +DeviceExpertSlotStore::~DeviceExpertSlotStore() { + if (staging_ != nullptr) b_.FreePinned(staging_); + if (arena_ != nullptr) b_.Free(arena_); + b_.DestroyQueue(q_); +} + +uint8_t* DeviceExpertSlotStore::SlotPtr(int32_t slot) const { + if (slot < 0 || slot >= slots_) { + throw std::out_of_range("DeviceExpertSlotStore: slot " + + std::to_string(slot) + " out of range"); + } + return arena_ + static_cast(slot) * slot_bytes_; +} + +void DeviceExpertSlotStore::WriteSlot(int32_t slot, const uint8_t* src, + size_t bytes) { + uint8_t* dst = SlotPtr(slot); // bounds first, as the host store does + if (bytes > slot_bytes_) { + throw std::invalid_argument("DeviceExpertSlotStore: write of " + + std::to_string(bytes) + + " bytes exceeds the slot"); + } + if (src == nullptr && bytes > 0) { + throw std::invalid_argument("DeviceExpertSlotStore: null source"); + } + if (bytes == 0) return; + b_.Copy(q_, dst, src, bytes); + // Synchronous by contract: the caller binds this slot to a GEMM as soon as + // the streamer returns, and `Copy` is asynchronous on CUDA. + b_.Synchronize(q_); +} + +uint8_t* DeviceExpertSlotStore::SlotForWrite(int32_t slot) { + if (slot < 0 || slot >= slots_) { + throw std::out_of_range("DeviceExpertSlotStore: slot " + + std::to_string(slot) + " out of range"); + } + staged_slot_ = slot; + return staging_; +} + +void DeviceExpertSlotStore::CommitSlot(int32_t slot, size_t bytes) { + uint8_t* dst = SlotPtr(slot); + if (bytes > slot_bytes_) { + throw std::invalid_argument("DeviceExpertSlotStore: commit of " + + std::to_string(bytes) + + " bytes exceeds the slot"); + } + if (staged_slot_ != slot) { + // With one staging buffer this is not a bookkeeping slip: the bytes in + // staging belong to whichever slot asked for it last, so committing them + // here would file one expert's weights under another expert's key. The + // cache would then report a HIT for a key whose slot holds the wrong + // expert, and the GEMM would multiply it without a symptom. + throw std::logic_error( + "DeviceExpertSlotStore: commit of slot " + std::to_string(slot) + + " but SlotForWrite last staged " + std::to_string(staged_slot_)); + } + staged_slot_ = -1; + if (bytes == 0) return; + b_.Copy(q_, dst, staging_, bytes); + b_.Synchronize(q_); +} + +uint8_t* DeviceExpertSlotStore::SlotForRead(int32_t slot) { + return SlotPtr(slot); +} + +} // namespace vllm diff --git a/src/vllm/model_executor/expert_streamer.cpp b/src/vllm/model_executor/expert_streamer.cpp index f83fd95a8..4abf34a83 100644 --- a/src/vllm/model_executor/expert_streamer.cpp +++ b/src/vllm/model_executor/expert_streamer.cpp @@ -59,8 +59,14 @@ ExpertStreamer::Result ExpertStreamer::EnsureFile(const ExpertKey& key, int fd, } // pread in a loop: a short read is legal and must be finished, not accepted. - // The destination is the slot itself, so the bytes never pass through a - // staging buffer or the page tables of the mapping. + // The destination is whatever the store says is host-writable, so on the host + // store the bytes never pass through a staging buffer or the page tables of + // the mapping at all. On a store whose slots are DEVICE memory that pointer + // cannot be the slot -- `vt::Backend::DeviceMemoryIsHostAddressable()` is + // false for CUDA -- so `SlotForWrite` hands back a staging buffer and + // `CommitSlot` below publishes it. Without that pair a device store could not + // be filled by this function at all, which is issue #1124's third piece and + // the reason the contract change rides in W1 rather than after it. // // THE ACQUISITION IS UNDONE IF THE READ THROWS, and that is the whole reason // this loop sits inside a try. Acquire must run first, because the read needs @@ -92,6 +98,13 @@ ExpertStreamer::Result ExpertStreamer::EnsureFile(const ExpertKey& key, int fd, } done += static_cast(n); } + // INSIDE the try, deliberately. Publishing is the last step of the fill and + // it can fail for the same class of reason the read can -- a device copy + // that throws leaves the slot holding whatever it held before, under a cache + // entry that already claims the key is resident. That is the same silent, + // plausible and wrong outcome the loop above is wrapped for, so it takes the + // same undo. + store_.CommitSlot(acq.slot, bytes); } catch (...) { cache_.Invalidate(key); throw; @@ -135,7 +148,22 @@ ExpertStreamer::Result ExpertStreamer::EnsureSpan(const ExpertKey& key, return out; } - store_.WriteSlot(acq.slot, src, bytes); + // THE ACQUISITION IS UNDONE IF THE WRITE THROWS, for the same reason the read + // in `EnsureFile` is wrapped, one call earlier. Acquire must run first because + // the write needs a slot, so by the time `WriteSlot` throws the cache already + // says the key is resident -- over a slot still holding the expert that + // acquisition just evicted. A throw that escapes leaves that entry standing, + // the next request for the key is an ordinary HIT, no bytes move because a hit + // moves none, and the GEMM multiplies the evicted expert. Silent, plausible + // and wrong. Before ENG-EXPERT-STREAM-DEVICE W1 no store could reach this: + // every `WriteSlot` was a memcpy. `DeviceExpertSlotStore::WriteSlot` calls + // `vt::Backend::Copy` and `Synchronize`, and both throw on CUDA. + try { + store_.WriteSlot(acq.slot, src, bytes); + } catch (...) { + cache_.Invalidate(key); + throw; + } bytes_filled_ += static_cast(bytes); ++fills_; out.filled = true; @@ -179,7 +207,14 @@ ExpertStreamer::Result ExpertStreamer::Ensure(const ExpertKey& key, } const GgufExpertSpan span = GgufExpertSpanOf(tensor, layout, key.expert); - store_.WriteSlot(acq.slot, span.data, span.bytes); + // Wrapped for the same reason as `EnsureSpan` above: a `try` on one entry + // point leaves the other exactly as exposed as it was. + try { + store_.WriteSlot(acq.slot, span.data, span.bytes); + } catch (...) { + cache_.Invalidate(key); + throw; + } bytes_filled_ += static_cast(span.bytes); ++fills_; out.filled = true; diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 26947a6d2..2557d5c73 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -1289,6 +1289,14 @@ vllm_cpp_add_test(test_expert_slot_cache vllm/model_executor/test_expert_slot_ca vllm_cpp_add_test(test_gguf_expert_span vllm/model_executor/test_gguf_expert_span.cpp) vllm_cpp_add_test(test_expert_streamer vllm/model_executor/test_expert_streamer.cpp) vllm_cpp_add_test(test_host_expert_slot_store vllm/model_executor/test_host_expert_slot_store.cpp) +# ENG-EXPERT-STREAM-DEVICE W1 (#1124), gate G1: the DEVICE slot store and the +# fill contract that makes it fillable. It needs no separate binary -- it +# registers nothing process-global, reads no environment latch, and drives its +# own `vt::Backend` instances by hand -- but it is its own file because the +# oracle it compares against is `HostExpertSlotStore`, so a suite that IS the +# host store cannot also be the comparison. +vllm_cpp_add_test(test_device_expert_slot_store + vllm/model_executor/test_device_expert_slot_store.cpp) # ENG-EXPERT-STREAM F3 (#912): the REACHABILITY gate, in its own binary because # VT_MOE_EXPERT_STREAM is read ONCE into a function-local static, so one process # cannot run both the streaming and non-streaming arms from the environment. diff --git a/tests/vllm/model_executor/test_device_expert_slot_store.cpp b/tests/vllm/model_executor/test_device_expert_slot_store.cpp new file mode 100644 index 000000000..dde5af703 --- /dev/null +++ b/tests/vllm/model_executor/test_device_expert_slot_store.cpp @@ -0,0 +1,516 @@ +// ENG-EXPERT-STREAM-DEVICE W1 (issue #1124, gate G1): can a slot store whose +// slots are DEVICE allocations be filled at all, and does what lands in it match +// what lands in `HostExpertSlotStore` byte for byte? +// +// THE ORACLE IS THE HOST STORE, and that is a measured statement rather than a +// search that came up empty. Pinned vLLM `555967922` has no inference-time +// expert paging anywhere (`model_executor/offloader/uva.py` is a CPU-blanket UVA +// offloader over whole parameters, `.../prefetch.py` is cpu-only), and the +// secondary-oracle table does not rescue it — llama.cpp's `-ot`/`-ncmoe` moves +// expert COMPUTE to the host, which is a different design and not a slot store. +// So the reference is our own host arm on the same input, driven through the +// same `ExpertStreamer`, which is the shape `ENG-EXPERT-STREAM` already gates on. +// +// WHY IT CAN RUN WITHOUT A GPU. The store's device-ness is entirely +// `vt::Backend`: one `Alloc` for the arena, one `AllocPinned` for staging, and +// `Copy`/`Synchronize` for the transfer. On the CPU backend those are malloc and +// memcpy, so the bytes are checkable, and the two structural claims a CPU tier +// COULD get wrong instead of measuring — that `SlotForWrite` hands out staging +// rather than the slot, and that `SlotForRead` hands out the slot rather than +// staging — are asserted as pointer relationships rather than inferred from +// content that a unified allocator would make identical either way. Nothing here +// dereferences a device pointer from the host; every read of a slot is a +// backend copy, which is the only access a discrete part would allow. +// +// WHAT THIS FILE DOES NOT CLAIM. It does not claim anything selects this store. +// Nothing does: `Qwen35ExpertStream` still holds a concrete +// `HostExpertSlotStore` and reads it through `HostExpertSlotStore::Slot`, and +// making that read virtual and choosing the store from the platform is W2 of the +// same row (#1124). This suite gates a class, deliberately and with that said +// out loud, per `## Nothing lands dead`. +#include +#include + +#include + +#include +#include +#include +#include +#include +#include + +#include "vllm/model_executor/device_expert_slot_store.h" +#include "vllm/model_executor/expert_slot_cache.h" +#include "vllm/model_executor/expert_streamer.h" +#include "vllm/model_executor/host_expert_slot_store.h" +#include "vt/backend.h" +#include "vt/device.h" + +namespace { + +using vllm::DeviceExpertSlotStore; +using vllm::ExpertKey; +using vllm::ExpertSlotCache; +using vllm::ExpertStreamer; +using vllm::HostExpertSlotStore; + +// A backend over ordinary host memory that COUNTS what the store asked it for. +// The same trick `test_expert_stream_device_slot` uses: malloc stands in for a +// device allocator, so an allocation is a real inspectable address, and the +// counters let a case say HOW MANY allocations and copies it observed rather +// than only that the bytes came out right. +// +// `DeviceMemoryIsHostAddressable()` stays FALSE — the default — because that is +// the honest answer for the device this store exists for, and it is the property +// that makes `pread`-into-the-slot illegal in the first place. +class CountingBackend final : public vt::Backend { + public: + void* Alloc(size_t bytes) override { + ++allocs; + last_alloc_bytes = bytes; + // OUT OF MEMORY is what this reproduces, and it is this class's headline + // failure rather than a hypothetical: #1123 is literally + // `vt cuda: cudaMalloc: out of memory`. Every backend in the tree reports + // it by THROWING -- `VT_CHECK` on the CPU backend, `Check(cudaMalloc)` on + // CUDA -- and none returns nullptr, so a throw is the only shape a real + // allocation failure takes here. + if (throw_on_alloc) throw std::runtime_error("counting backend: alloc failed"); + last_alloc = std::malloc(bytes == 0 ? 1 : bytes); + return last_alloc; + } + void Free(void* p) override { + ++frees; + last_freed = p; + std::free(p); + } + void Memset(vt::Queue&, void* p, int v, size_t bytes) override { + std::memset(p, v, bytes); + } + void Copy(vt::Queue&, void* dst, const void* src, size_t bytes) override { + ++copies; + copied_bytes += bytes; + std::memcpy(dst, src, bytes); + } + vt::Queue CreateQueue() override { + ++queues; + return vt::Queue{vt::Device{vt::DeviceType::kXPU, 0}, nullptr}; + } + void DestroyQueue(vt::Queue&) override { ++queues_destroyed; } + void Synchronize(vt::Queue&) override { ++syncs; } + bool UnifiedMemory() const override { return false; } + void* AllocPinned(size_t bytes) override { + ++pinned_allocs; + last_pinned_bytes = bytes; + // The same failure one allocation later, and the expensive one: by here the + // whole device arena is already held. + if (throw_on_pinned_alloc) + throw std::runtime_error("counting backend: pinned alloc failed"); + last_pinned = std::malloc(bytes == 0 ? 1 : bytes); + return last_pinned; + } + void FreePinned(void* p) override { + ++pinned_frees; + std::free(p); + } + + int allocs = 0; + int pinned_allocs = 0; + int frees = 0; + int pinned_frees = 0; + bool throw_on_alloc = false; + bool throw_on_pinned_alloc = false; + int copies = 0; + int syncs = 0; + int queues = 0; + int queues_destroyed = 0; + int64_t copied_bytes = 0; + size_t last_alloc_bytes = 0; + size_t last_pinned_bytes = 0; + void* last_alloc = nullptr; + void* last_pinned = nullptr; + void* last_freed = nullptr; +}; + +// Read a device slot the only way a discrete part would allow: a backend copy +// back to host memory. Never a host dereference of the slot pointer. +std::vector ReadBack(vt::Backend& b, uint8_t* device_slot, + size_t bytes) { + std::vector out(bytes, 0); + vt::Queue q = b.CreateQueue(); + b.Copy(q, out.data(), device_slot, bytes); + b.Synchronize(q); + b.DestroyQueue(q); + return out; +} + +// A file of bytes no two slices of which are equal, so a wrong offset shows up +// in the CONTENT and not only in a length. +struct SliceFile { + char path[64] = "/tmp/vllm_device_slot_XXXXXX"; + int fd = -1; + std::vector bytes; + + SliceFile(size_t slices, size_t slice_bytes) : bytes(slices * slice_bytes) { + for (size_t i = 0; i < bytes.size(); ++i) + bytes[i] = static_cast((i * 31u + (i >> 5) * 7u + 1u) & 0xFFu); + fd = ::mkstemp(path); + REQUIRE(fd >= 0); + REQUIRE(::write(fd, bytes.data(), bytes.size()) == + static_cast(bytes.size())); + } + ~SliceFile() { + if (fd >= 0) ::close(fd); + ::unlink(path); + } + const uint8_t* slice(size_t i, size_t slice_bytes) const { + return bytes.data() + i * slice_bytes; + } +}; + +constexpr size_t kSliceBytes = 96; +constexpr size_t kSlices = 4; + +} // namespace + +TEST_CASE("DeviceExpertSlotStore refuses a budget it cannot honour") { + CountingBackend b; + CHECK_THROWS_AS(DeviceExpertSlotStore(b, 0, 64), std::invalid_argument); + CHECK_THROWS_AS(DeviceExpertSlotStore(b, -1, 64), std::invalid_argument); + CHECK_THROWS_AS(DeviceExpertSlotStore(b, 4, 0), std::invalid_argument); + // The host store's `std::vector` catches this for it; a raw `Alloc` has no + // backstop, and a wrapped product would hand out in-range slot pointers past + // the end of a far smaller arena. + CHECK_THROWS_AS(DeviceExpertSlotStore(b, 4, SIZE_MAX / 2), + std::invalid_argument); + // Nothing was allocated for any of the four refusals. + CHECK(b.allocs == 0); + CHECK(b.pinned_allocs == 0); + // ...and no queue either: a constructor that throws runs no destructor, so + // anything acquired above the refusal would leak. + CHECK(b.queues == 0); +} + +TEST_CASE("an ALLOCATION that throws gives back everything already acquired") { + // The SECOND fresh review of PR #1735 (F2) -- not the first review's F2, + // which was the file `CHECK`'s justification. The constructor's own comment says a + // throwing constructor runs no destructor, and then guarded the failure that + // cannot happen while leaking on the one that does. NO BACKEND IN THIS TREE + // RETURNS NULLPTR: `CpuBackend::Alloc` refuses with `VT_CHECK`, + // `CudaBackend::Alloc` and `AllocPinned` refuse through `Check(...)`, and the + // base `Backend::AllocPinned` forwards to `Alloc`. They all THROW, and out of + // memory is this class's headline failure -- #1123 is + // `vt cuda: cudaMalloc: out of memory`. So the leak was on the live path and + // the guard was on the dead one. + // + // What leaks is not small. A throw from `Alloc` strands the queue; a throw + // from `AllocPinned` strands the queue AND the whole device arena, 18.55 GiB + // on the target checkpoint, at the exact moment the device is out of memory. + SUBCASE("the arena allocation throws") { + CountingBackend b; + b.throw_on_alloc = true; + CHECK_THROWS_AS(DeviceExpertSlotStore(b, 8, 1024), std::runtime_error); + // It was attempted, so this is the failure path and not an early refusal. + CHECK(b.allocs == 1); + // The queue was taken before it, and is given back. + CHECK(b.queues == 1); + CHECK(b.queues_destroyed == 1); + // Nothing else was ever acquired, so nothing else is released. + CHECK(b.pinned_allocs == 0); + CHECK(b.frees == 0); + CHECK(b.pinned_frees == 0); + } + SUBCASE("the pinned staging allocation throws") { + CountingBackend b; + b.throw_on_pinned_alloc = true; + CHECK_THROWS_AS(DeviceExpertSlotStore(b, 8, 1024), std::runtime_error); + CHECK(b.allocs == 1); + CHECK(b.pinned_allocs == 1); + // THE ARENA COMES BACK. This is the 18.55 GiB, and `last_freed` says it was + // the arena rather than merely some pointer. + CHECK(b.frees == 1); + CHECK(b.last_freed == b.last_alloc); + CHECK(b.queues == 1); + CHECK(b.queues_destroyed == 1); + // Nothing pinned was ever handed over, so nothing pinned is released. + CHECK(b.pinned_frees == 0); + } +} + +TEST_CASE("the arena is ONE contiguous device allocation and staging is ONE pinned slot") { + CountingBackend b; + { + DeviceExpertSlotStore s(b, 8, 1024); + CHECK(s.slot_count() == 8); + CHECK(s.slot_bytes() == 1024); + CHECK(s.resident_bytes() == 8 * 1024); + // One arena of the whole budget, decided up front and never grown: the + // point of a slot array is that a model larger than memory cannot page + // itself to death by admitting one more expert. + CHECK(b.allocs == 1); + CHECK(b.last_alloc_bytes == 8u * 1024u); + // ONE staging slot, not eight. The filler is synchronous, so exactly one + // fill is ever in flight; a buffer per slot would double the arena's host + // cost to buffer a concurrency that does not exist. + CHECK(b.pinned_allocs == 1); + CHECK(b.last_pinned_bytes == 1024u); + CHECK(b.queues == 1); + + // Slots are fixed offsets into that one block, in order. + CHECK(s.SlotForRead(0) == static_cast(b.last_alloc)); + CHECK(s.SlotForRead(3) == static_cast(b.last_alloc) + 3 * 1024); + CHECK(s.SlotForRead(7) == static_cast(b.last_alloc) + 7 * 1024); + CHECK_THROWS_AS(s.SlotForRead(8), std::out_of_range); + CHECK_THROWS_AS(s.SlotForRead(-1), std::out_of_range); + } + // The store owns both allocations and the queue, and gives all three back. + CHECK(b.queues_destroyed == 1); +} + +TEST_CASE("SlotForWrite hands out STAGING, never the device slot") { + // This is the claim a unified allocator would hide if it were asserted + // through content: on this backend a slot pointer IS host memory, so a + // `SlotForWrite` that returned the slot would still produce the right bytes. + // It is asserted as a pointer relationship for that reason. + CountingBackend b; + DeviceExpertSlotStore s(b, 4, 64); + uint8_t* const arena = static_cast(b.last_alloc); + + uint8_t* w0 = s.SlotForWrite(0); + CHECK(w0 == static_cast(b.last_pinned)); + CHECK((w0 < arena || w0 >= arena + 4 * 64)); // outside the arena entirely + // One staging buffer means every slot gets the same address back. + CHECK(s.SlotForWrite(1) == w0); + CHECK(s.SlotForWrite(3) == w0); + CHECK_THROWS_AS(s.SlotForWrite(4), std::out_of_range); + CHECK_THROWS_AS(s.SlotForWrite(-1), std::out_of_range); + // Handing out staging moves no bytes by itself. + CHECK(b.copies == 0); +} + +TEST_CASE("G1: a device store filled through EnsureFile is BYTE-IDENTICAL to the host store") { + // The gate. Both arms run the same cache policy, the same streamer, the same + // descriptor and the same offsets; the only difference is where the slot + // lives. On a CPU `vt::Backend`, per the spec — this row has no discrete + // NVIDIA GPU to reach, and that limitation is recorded as G-DISCRETE rather + // than dressed up as this gate. + // + // "BYTE-IDENTICAL" IS OVER THE FILLED PREFIX, and the stores are asymmetric + // beyond it: the host arena is a `std::vector` and is zero-filled at + // construction, while the device arena is a raw `vt::Backend::Alloc` and is + // not initialised at all. Every fill here writes a whole slot, so the prefix + // is the slot and the distinction does not reach this gate -- but a caller + // that streamed a SHORT slice into a full-sized slot would find the two stores + // disagreeing past the slice, and nothing promises otherwise. Zeroing the + // device arena would cost a full write of the whole budget at load, 18.55 GiB + // on the target checkpoint, to hide bytes no reader may look at. (Fresh review + // of PR #1735, F3.) + vt::Backend& cpu = vt::GetBackend(vt::DeviceType::kCPU); + SliceFile f(kSlices, kSliceBytes); + + ExpertSlotCache host_cache(static_cast(kSlices)); + HostExpertSlotStore host(static_cast(kSlices), kSliceBytes); + ExpertStreamer host_st(host_cache, host); + + ExpertSlotCache dev_cache(static_cast(kSlices)); + DeviceExpertSlotStore dev(cpu, static_cast(kSlices), kSliceBytes); + ExpertStreamer dev_st(dev_cache, dev); + + // THE DEVICE ARENA IS PUT INTO A KNOWN STATE FIRST, and a gate assertion + // below depends on it rather than this being a tidiness habit. "Each slot + // holds a DIFFERENT slice" compares two device slots that a + // publish-suppressing mutation leaves UNWRITTEN, and `vt::Backend::Alloc` + // does not initialise them (`std::aligned_alloc` on the CPU backend), so + // whether that mutation red that assertion was decided by whatever the + // allocator last left there -- the second fresh review of #1735 measured this + // suite one assertion down from the recorded count for exactly that reason. + // Writing every slot to the SAME known byte makes an unmutated fill the only + // thing that can make two slots differ, so the assertion measures the store. + // It says nothing about the CLASS, whose arena is uninitialised as its header + // states; it is this test defining its own starting state. + const std::vector known(kSliceBytes, 0x5A); + for (size_t i = 0; i < kSlices; ++i) + dev.WriteSlot(static_cast(i), known.data(), known.size()); + + int32_t host_slot[kSlices]; + int32_t dev_slot[kSlices]; + for (size_t i = 0; i < kSlices; ++i) { + const ExpertKey key{3, static_cast(i)}; + const size_t off = i * kSliceBytes; + const ExpertStreamer::Result h = + host_st.EnsureFile(key, f.fd, off, kSliceBytes); + const ExpertStreamer::Result d = + dev_st.EnsureFile(key, f.fd, off, kSliceBytes); + REQUIRE(h.filled); + REQUIRE(d.filled); + // Same cache policy, so the same key lands in the same slot index. If this + // ever diverged the byte comparison below would be comparing the wrong + // pair, so it is REQUIRED rather than checked. + REQUIRE(h.slot == d.slot); + host_slot[i] = h.slot; + dev_slot[i] = d.slot; + } + + for (size_t i = 0; i < kSlices; ++i) { + const std::vector got = ReadBack(cpu, dev.SlotForRead(dev_slot[i]), + kSliceBytes); + const uint8_t* want = host.Slot(host_slot[i]); + // Byte-identical to the host store... + CHECK(std::memcmp(got.data(), want, kSliceBytes) == 0); + // ...and equal to the FILE, which is the one assertion in this case that can + // see a defect in the SHARED HELPER. The comparison above is host-arm + // against device-arm, and both arms run the same `ExpertStreamer` over the + // same descriptor at the same `file_offset`: a streamer that read the wrong + // offset, or read short, or read one slice twice makes both arms + // identically wrong and passes it. The bytes on disk are the only input + // neither arm computed, so comparing against them is what breaks that tie. + // (An earlier draft justified this check as making a both-arms-empty red + // DETERMINISTIC; the arena prefill above now does that job, and it was never + // the stronger ground. Second fresh review of PR #1735, F5.) + CHECK(std::memcmp(got.data(), f.slice(i, kSliceBytes), kSliceBytes) == 0); + CHECK(got[0] == f.slice(i, kSliceBytes)[0]); + } + // Each slot holds a DIFFERENT slice, which is what a `SlotForRead` that + // returned the staging buffer would break: staging holds only the last fill. + CHECK(std::memcmp(ReadBack(cpu, dev.SlotForRead(dev_slot[0]), kSliceBytes).data(), + ReadBack(cpu, dev.SlotForRead(dev_slot[kSlices - 1]), kSliceBytes).data(), + kSliceBytes) != 0); + + CHECK(dev_st.fills() == host_st.fills()); + CHECK(dev_st.bytes_filled() == host_st.bytes_filled()); +} + +TEST_CASE("G1: the SPAN and TENSOR fills land in the device slot too") { + // `EnsureFile` is the production filler, but `WriteSlot` is still reachable + // through `EnsureSpan`, and a device store that only honoured one of the two + // would be a trap for the next caller. + vt::Backend& cpu = vt::GetBackend(vt::DeviceType::kCPU); + std::vector src(kSliceBytes); + for (size_t i = 0; i < src.size(); ++i) + src[i] = static_cast(0xC0u + i); + + ExpertSlotCache host_cache(2); + HostExpertSlotStore host(2, kSliceBytes); + ExpertStreamer host_st(host_cache, host); + ExpertSlotCache dev_cache(2); + DeviceExpertSlotStore dev(cpu, 2, kSliceBytes); + ExpertStreamer dev_st(dev_cache, dev); + + const ExpertKey key{1, 9}; + const ExpertStreamer::Result h = + host_st.EnsureSpan(key, src.data(), src.size()); + const ExpertStreamer::Result d = + dev_st.EnsureSpan(key, src.data(), src.size()); + REQUIRE(h.filled); + REQUIRE(d.filled); + const std::vector got = + ReadBack(cpu, dev.SlotForRead(d.slot), kSliceBytes); + CHECK(std::memcmp(got.data(), host.Slot(h.slot), kSliceBytes) == 0); + CHECK(std::memcmp(got.data(), src.data(), kSliceBytes) == 0); + + // The same refusals the host store makes, so a caller cannot learn one + // contract from one store and be surprised by the other. + CHECK_THROWS_AS(dev.WriteSlot(2, src.data(), src.size()), std::out_of_range); + CHECK_THROWS_AS(dev.WriteSlot(-1, src.data(), src.size()), std::out_of_range); + std::vector big(kSliceBytes + 1, 0xEE); + CHECK_THROWS_AS(dev.WriteSlot(0, big.data(), big.size()), + std::invalid_argument); +} + +TEST_CASE("the fill is a bounce: staging is written, then ONE copy publishes it") { + CountingBackend b; + SliceFile f(kSlices, kSliceBytes); + ExpertSlotCache cache(static_cast(kSlices)); + DeviceExpertSlotStore s(b, static_cast(kSlices), kSliceBytes); + ExpertStreamer st(cache, s); + + const ExpertKey key{5, 2}; + const ExpertStreamer::Result r = st.EnsureFile(key, f.fd, 0, kSliceBytes); + REQUIRE(r.filled); + // Exactly one H2D of exactly the slice, and a synchronize after it: `Copy` is + // `cudaMemcpyAsync` on CUDA, and both the reuse of the single staging buffer + // by the next fill and the GEMM that reads this slot as soon as the streamer + // returns would otherwise race the transfer. + CHECK(b.copies == 1); + CHECK(b.copied_bytes == static_cast(kSliceBytes)); + CHECK(b.syncs == 1); + + // A hit publishes nothing: no copy, no synchronize, no bytes. + const ExpertStreamer::Result hit = st.EnsureFile(key, f.fd, 0, kSliceBytes); + REQUIRE(hit.hit); + CHECK(b.copies == 1); + CHECK(b.syncs == 1); + CHECK(st.fills() == 1); +} + +TEST_CASE("a fill that THROWS publishes nothing and leaves the key non-resident") { + CountingBackend b; + SliceFile f(1, kSliceBytes); // one slice on disk, so a second read is short + ExpertSlotCache cache(2); + DeviceExpertSlotStore s(b, 2, kSliceBytes); + ExpertStreamer st(cache, s); + + const ExpertKey key{4, 4}; + // 96 bytes from offset 64 of a 96-byte file: 32 land in staging and then + // pread returns 0. + CHECK_THROWS_AS(st.EnsureFile(key, f.fd, 64, kSliceBytes), + std::runtime_error); + // Nothing reached the device, so the slot still holds what it held before — + // and, crucially, the cache does not claim the key is resident over it. + CHECK(b.copies == 0); + CHECK_FALSE(cache.IsResident(key)); + CHECK(st.fills() == 0); + CHECK(st.bytes_filled() == 0); +} + +TEST_CASE("CommitSlot refuses to publish staging under the WRONG slot") { + // With one staging buffer this is not bookkeeping. The bytes in staging + // belong to whichever slot asked for the buffer last, so committing them + // elsewhere files one expert's weights under another expert's key; the cache + // then reports a HIT for a slot holding the wrong expert and the GEMM + // multiplies it without a symptom. + CountingBackend b; + DeviceExpertSlotStore s(b, 4, 64); + + s.SlotForWrite(2); + CHECK_THROWS_AS(s.CommitSlot(1, 64), std::logic_error); + CHECK(b.copies == 0); + // The right slot goes through. + s.CommitSlot(2, 64); + CHECK(b.copies == 1); + // ...and once only: the staging buffer is spent, so a repeat is refused too. + CHECK_THROWS_AS(s.CommitSlot(2, 64), std::logic_error); + CHECK(b.copies == 1); + // A commit with nothing staged at all is the same refusal. + DeviceExpertSlotStore fresh(b, 4, 64); + CHECK_THROWS_AS(fresh.CommitSlot(0, 64), std::logic_error); + + // Bounds and size are checked before the staging identity, as they are on + // every other entry point. + CHECK_THROWS_AS(s.CommitSlot(4, 64), std::out_of_range); + CHECK_THROWS_AS(s.CommitSlot(-1, 64), std::out_of_range); + s.SlotForWrite(0); + CHECK_THROWS_AS(s.CommitSlot(0, 65), std::invalid_argument); +} + +TEST_CASE("the HOST path is byte-identical across the contract change") { + // W1's stop condition: the host path must stay exactly what it was. It writes + // in place, so `SlotForWrite` still returns the slot itself, `CommitSlot` is a + // no-op, and no staging buffer exists to be allocated or copied. + HostExpertSlotStore h(3, 64); + CHECK(h.SlotForWrite(1) == h.Slot(1)); + + std::vector src(64, 0x7E); + h.WriteSlot(1, src.data(), src.size()); + std::vector before(h.Slot(1), h.Slot(1) + 64); + h.CommitSlot(1, 64); + CHECK(std::memcmp(before.data(), h.Slot(1), 64) == 0); + CHECK(h.Slot(1)[0] == 0x7E); + CHECK(h.Slot(0)[0] == 0x00); // untouched neighbour + + // It still refuses a slot it does not have, because an out-of-range commit + // means the streamer and the store disagree about which slot was filled. + CHECK_THROWS_AS(h.CommitSlot(3, 64), std::out_of_range); + CHECK_THROWS_AS(h.CommitSlot(-1, 64), std::out_of_range); + CHECK_THROWS_AS(h.CommitSlot(0, 65), std::invalid_argument); +} diff --git a/tests/vllm/model_executor/test_expert_streamer.cpp b/tests/vllm/model_executor/test_expert_streamer.cpp index f78d404c0..5145d3a25 100644 --- a/tests/vllm/model_executor/test_expert_streamer.cpp +++ b/tests/vllm/model_executor/test_expert_streamer.cpp @@ -8,6 +8,9 @@ // The store is an interface precisely so these run without a GPU; the host // implementation below records what was written, which is what makes "a hit // wrote nothing" checkable at all. +#include +#include + #include #include @@ -42,6 +45,15 @@ class RecordingStore final : public ExpertSlotStore { REQUIRE(slot >= 0); REQUIRE(slot < slots_); REQUIRE(bytes <= bytes_); + // A WRITE can fail for the same class of reason a publish can, and on the + // store W1 added it for it fails for exactly that reason: + // `DeviceExpertSlotStore::WriteSlot` calls `vt::Backend::Copy` and + // `Synchronize`, and a real CUDA backend throws out of both. Before W1 no + // store in the tree could do that, because every `WriteSlot` was a memcpy. + // The throw is raised BEFORE the copy so the slot keeps the bytes of the + // expert that used to live there, which is the state the streamer's undo + // has to survive. + if (throw_on_write) throw std::runtime_error("recording store: write failed"); std::memcpy(mem_.data() + static_cast(slot) * bytes_, src, bytes); ++writes; written_bytes += static_cast(bytes); @@ -58,11 +70,33 @@ class RecordingStore final : public ExpertSlotStore { return mem_.data() + static_cast(slot) * bytes_; } + // ENG-EXPERT-STREAM-DEVICE W1 (#1124): the double writes in place like the + // host store, so there is nothing to publish — but it COUNTS the calls, so a + // case can assert that the streamer commits exactly the fills it performed + // and never a hit. + void CommitSlot(int32_t slot, size_t bytes) override { + REQUIRE(slot >= 0); + REQUIRE(slot < slots_); + REQUIRE(bytes <= bytes_); + ++commits; + last_commit_slot = slot; + // A PUBLISH can fail, and on the store this method was added for it fails + // for the same class of reason a read does: `DeviceExpertSlotStore` calls + // `vt::Backend::Copy` and `Synchronize`, and a real CUDA backend throws out + // of both. The double can be asked to do that, because the placement of the + // call relative to the streamer's `try` is only observable when it throws. + if (throw_on_commit) throw std::runtime_error("recording store: commit failed"); + } + const uint8_t* slot(int32_t s) const { return mem_.data() + static_cast(s) * bytes_; } int writes = 0; int64_t written_bytes = 0; int32_t last_slot = -1; + int commits = 0; + int32_t last_commit_slot = -1; + bool throw_on_commit = false; + bool throw_on_write = false; private: int32_t slots_; @@ -239,3 +273,213 @@ TEST_CASE("experts of different layers do not share a slot") { CHECK(r1.filled); CHECK(s.fills() == 2); } + + +TEST_CASE("EnsureFile PUBLISHES exactly the fills, and never a hit or a failure") { + // ENG-EXPERT-STREAM-DEVICE W1 (#1124). `CommitSlot` is what lets a store whose + // slots the host cannot write be filled at all, and it is only correct if the + // streamer calls it in exactly the places a fill succeeded. A commit on a HIT + // would republish a slot whose staging buffer now holds a different expert; a + // commit after a failed read would publish a partial slice under a key the + // cache is about to invalidate. + char path[] = "/tmp/vllm_expert_commit_XXXXXX"; + const int fd = ::mkstemp(path); + REQUIRE(fd >= 0); + std::vector file(96); + for (size_t i = 0; i < file.size(); ++i) file[i] = static_cast(i + 3); + REQUIRE(::write(fd, file.data(), file.size()) == + static_cast(file.size())); + + ExpertSlotCache cache(2); + RecordingStore store(2, 32); + ExpertStreamer s(cache, store); + + const ExpertKey key{2, 5}; + const ExpertStreamer::Result first = s.EnsureFile(key, fd, 0, 32); + REQUIRE(first.filled); + CHECK(store.commits == 1); + // The slot published is the slot filled, not merely some slot. + CHECK(store.last_commit_slot == first.slot); + + // A hit moves no bytes, so there is nothing to publish. + REQUIRE(s.EnsureFile(key, fd, 0, 32).hit); + CHECK(store.commits == 1); + + // A short read throws; nothing is published and the key is not resident. + CHECK_THROWS_AS(s.EnsureFile(ExpertKey{2, 6}, fd, 80, 32), std::runtime_error); + CHECK(store.commits == 1); + CHECK_FALSE(cache.IsResident(ExpertKey{2, 6})); + + // A slice too large is refused before the cache is touched, so no slot was + // ever handed out to publish. + CHECK_THROWS_AS(s.EnsureFile(ExpertKey{2, 7}, fd, 0, 33), std::invalid_argument); + CHECK(store.commits == 1); + + // An exhausted budget returns an invalid slot with neither flag set. + REQUIRE(s.EnsureFile(ExpertKey{2, 8}, fd, 32, 32).filled); + CHECK(store.commits == 2); + const ExpertStreamer::Result none = s.EnsureFile(ExpertKey{2, 9}, fd, 0, 32); + CHECK(none.slot == -1); + CHECK_FALSE(none.filled); + CHECK(store.commits == 2); + + ::close(fd); + ::unlink(path); +} + + +TEST_CASE("a PUBLISH that throws leaves nothing resident either") { + // ENG-EXPERT-STREAM-DEVICE W1 (#1124), the fresh review of PR #1735 (F1). + // `store_.CommitSlot(...)` sits INSIDE `EnsureFile`'s try, and until this case + // existed nothing measured that: moving the call to just after the `catch` + // left every suite green, because no store in the tree could fail a publish. + // + // The failure it guards is the SAME one the read arm is wrapped for, one step + // later. Acquire must run before the fill, so by the time a publish throws the + // cache already says the key is resident -- over a slot still holding the + // expert that used to live there, because the bytes never left staging. If the + // throw escapes without undoing the acquisition, the next request for that key + // is an ordinary HIT, no read is issued because a hit moves no bytes, and the + // GEMM multiplies the wrong expert. Silent, plausible and wrong. + // + // This arm becomes REACHABLE in W2, when a store whose `CommitSlot` really + // copies to a device is selected; it is gated now because that is when the + // placement decision was made. + char path[] = "/tmp/vllm_expert_publish_XXXXXX"; + const int fd = ::mkstemp(path); + REQUIRE(fd >= 0); + std::vector file(96); + for (size_t i = 0; i < file.size(); ++i) file[i] = static_cast(i + 5); + REQUIRE(::write(fd, file.data(), file.size()) == + static_cast(file.size())); + + ExpertSlotCache cache(2); + RecordingStore store(2, 32); + ExpertStreamer s(cache, store); + + const ExpertKey key{8, 1}; + store.throw_on_commit = true; + CHECK_THROWS_AS(s.EnsureFile(key, fd, 0, 32), std::runtime_error); + + // THE ASSERTIONS THAT THE PLACEMENT BUYS. The read succeeded, so only the + // publish can have undone the acquisition. + CHECK(store.commits == 1); + CHECK_FALSE(cache.IsResident(key)); + CHECK_FALSE(cache.SlotOf(key).has_value()); + CHECK(cache.resident() == 0); + // A fill that did not publish moved no bytes, so the counters must not claim + // it did -- the same rule the read arm follows. + CHECK(s.fills() == 0); + CHECK(s.bytes_filled() == 0); + + // The slot came back to the budget, and the retry is a MISS that really + // refills rather than a hit over a stale slot. + store.throw_on_commit = false; + const ExpertStreamer::Result retry = s.EnsureFile(key, fd, 0, 32); + REQUIRE(retry.slot >= 0); + CHECK(retry.filled); + CHECK_FALSE(retry.hit); + CHECK(s.fills() == 1); + for (int i = 0; i < 32; ++i) + REQUIRE(store.slot(retry.slot)[i] == static_cast(i + 5)); + + ::close(fd); + ::unlink(path); +} + + +TEST_CASE("a WRITE that throws leaves nothing resident either -- EnsureSpan") { + // ENG-EXPERT-STREAM-DEVICE W1 (#1124), the SECOND fresh review of PR #1735 + // (F1) -- not the first review's F1, which was the publish arm one call later. + // `EnsureFile`'s publish is wrapped and `EnsureSpan`'s write was not, and the + // two are the same window one call earlier. W1 is what opens it: before this + // wave every store's `WriteSlot` was a memcpy and could not throw, and + // `DeviceExpertSlotStore::WriteSlot` calls `vt::Backend::Copy` and + // `Synchronize`, which throw `std::runtime_error` out of the CUDA backend. + // + // `EnsureSpan` is a PRODUCTION call site -- `qwen3_5.cpp`'s + // `Qwen35ExpertStream::Slice` reaches it from `Qwen3_5Model::Forward` -- so + // the failure is the deployed one: Acquire has already evicted the previous + // expert and claimed the key, the write throws, and the entry stands over a + // slot still holding the EVICTED expert's bytes. The next request for that key + // is an ordinary HIT, no bytes move because a hit moves none, and the GEMM + // multiplies the wrong expert. Silent, plausible and wrong. + ExpertSlotCache cache(1); + RecordingStore store(1, 32); + ExpertStreamer s(cache, store); + + // A resident expert first, so the failed write has something to have evicted: + // the slot's stale content is the corruption, not merely an empty slot. + std::vector a(32, 0xA1); + const ExpertKey key_a{7, 1}; + REQUIRE(s.EnsureSpan(key_a, a.data(), a.size()).filled); + REQUIRE(store.slot(0)[0] == 0xA1); + s.EndStep(); + + std::vector b(32, 0xB2); + const ExpertKey key_b{7, 2}; + store.throw_on_write = true; + CHECK_THROWS_AS(s.EnsureSpan(key_b, b.data(), b.size()), std::runtime_error); + + // THE ASSERTIONS THE UNDO BUYS. The acquisition happened -- the cache had to + // hand out a destination before the write could be attempted -- so only the + // catch can have taken it back. + CHECK_FALSE(cache.IsResident(key_b)); + CHECK_FALSE(cache.SlotOf(key_b).has_value()); + CHECK(cache.resident() == 0); + // The write moved no bytes, so the counters must not claim it did. + CHECK(s.fills() == 1); + CHECK(s.bytes_filled() == 32); + CHECK(store.writes == 1); + // ...and the slot really does hold the evicted expert, which is the state a + // surviving cache entry would have made a HIT over. + CHECK(store.slot(0)[0] == 0xA1); + CHECK(store.slot(0)[31] == 0xA1); + + // The retry is a real MISS that refills, not a hit over the stale slot. + store.throw_on_write = false; + const ExpertStreamer::Result retry = s.EnsureSpan(key_b, b.data(), b.size()); + REQUIRE(retry.slot >= 0); + CHECK(retry.filled); + CHECK_FALSE(retry.hit); + CHECK(s.fills() == 2); + CHECK(s.bytes_filled() == 64); + for (int i = 0; i < 32; ++i) REQUIRE(store.slot(retry.slot)[i] == 0xB2); +} + + +TEST_CASE("a WRITE that throws leaves nothing resident either -- Ensure") { + // The tensor overload has the identical window, for the identical reason, and + // it is gated separately because the two entry points wrap their own writes: + // a `try` added to one of them leaves the other exactly as exposed as before. + FakeTensor t = MakeTensor(8, 2, 64); + const auto L = GgufExpertLayoutOf(t.info, 8); + ExpertSlotCache cache(1); + RecordingStore store(1, L.expert_bytes); + ExpertStreamer s(cache, store); + + REQUIRE(s.Ensure(K(0, 4), t.info, L).filled); + REQUIRE(store.slot(0)[0] == 4); // MakeTensor stamps each expert with its index + s.EndStep(); + + const ExpertKey key{0, 6}; + store.throw_on_write = true; + CHECK_THROWS_AS(s.Ensure(key, t.info, L), std::runtime_error); + + CHECK_FALSE(cache.IsResident(key)); + CHECK_FALSE(cache.SlotOf(key).has_value()); + CHECK(cache.resident() == 0); + CHECK(s.fills() == 1); + CHECK(s.bytes_filled() == static_cast(L.expert_bytes)); + CHECK(store.writes == 1); + CHECK(store.slot(0)[0] == 4); // still expert 4's bytes, never expert 6's + + store.throw_on_write = false; + const ExpertStreamer::Result retry = s.Ensure(key, t.info, L); + REQUIRE(retry.slot >= 0); + CHECK(retry.filled); + CHECK_FALSE(retry.hit); + CHECK(s.fills() == 2); + CHECK(store.slot(retry.slot)[0] == 6); + CHECK(store.slot(retry.slot)[L.expert_bytes - 1] == 6); +}