Peer tier with mutual help: concurrency 2 at 155 t/s aggregate on dual RTX 3090 (elastic peer pair, opt-in)

CMakeLists.txt

@@ -446,7 +446,7 @@ if(STRATA_ENABLE_CUDA OR STRATA_ENABLE_HIP)
   add_library(strata_engine STATIC src/core/layer.cpp src/core/session.cpp src/core/expert_source.cpp src/core/remote_experts.cpp
     src/core/remote_expert_opt.cu
     src/core/expert_cache.cpp src/core/native_head.cpp src/core/native_dense.cpp src/core/verify.cpp src/core/mtp.cpp
-    src/core/conversation_snapshot.cpp src/core/conversation_state.cpp src/core/conversation_memory.cpp src/core/peer_experts.cpp
+    src/core/conversation_snapshot.cpp src/core/conversation_state.cpp src/core/conversation_memory.cpp src/core/peer_experts.cpp src/core/peer_link.cpp
     src/core/arch_defaults.cpp src/core/vmm.cpp src/core/conversation_file.cpp)
   target_include_directories(strata_engine PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/include)
   target_link_libraries(strata_engine PUBLIC strata_core strata_kernels strata_kernels_cpu ${_strata_gpu_runtime_target})

docs/MULTI_GPU.md

@@ -246,3 +246,61 @@ pipeline).
 
 With a layer split, `--batch N --batch-groups G --trim-stage-weights` decodes several conversations together and
 pipelines them through the cards: see [BATCHING.md](BATCHING.md).
+
+## A second request on a peer pair: the elastic pair with mutual help (opt-in)
+
+With `--peer-device` one engine uses both cards for one conversation, and a second request waits. The **elastic pair**
+lets the second card turn into its own engine when a second request arrives, and back into the first engine's peer
+tier when it has been idle for a while. While both run, each engine computes the other's rows for the experts its card
+holds (**mutual help**, `--peer-link`), so neither falls back to the CPU for the experts the other card has. One
+sequence per engine process, as always: two processes, one card each, one shared expert arena in RAM.
+
+What happens:
+
+| | lead engine (card 1) | helper engine (card 2) |
+|---|---|---|
+| one request | `--peer-device`: card 2 is its tier (exactly as without the pair) | asleep: its expert cache given back (`--vram-elastic`), optionally frozen with NVIDIA's `cuda-checkpoint` |
+| a 2nd request arrives | `PEER_DETACH` between two verify windows; its last N CPU workers sit out (`POOL_HOLD`) | thawed, its cache comes back from the shared arena (`VRAM`), then it takes the request |
+| both run | computes its own experts and the helper's rows for its card's experts | the same, the other way round |
+| idle for `hold_s` | `PEER_ATTACH` once its request has ended: card 2 is its tier again | cache given back, frozen |
+
+Config (the lead's run config; nothing changes without the `"elastic"` block):
+
+```json
+"args": [ ..., "--peer-device", "1", "--peer-link", "/dev/shm/strata-link", "--peer-link-role", "0",
+         "--shared-expert-arena", "/dev/shm/strata.arena" ],
+"elastic": { "helper": "strata-<model>-helper.json", "hold_s": 30, "helper_cores": "8-15,24-31",
+             "hold_workers": 8, "checkpoint": "/usr/bin/cuda-checkpoint" }
+```
+
+The helper config is an ordinary one-card run config (`"gpu": 0` for the second card) whose args carry
+`--peer-link <same file> --peer-link-role 1 --vram-elastic --shared-expert-arena <same file> --pool-workers 7`.
+`helper_cores` pins the helper process (and `hold_workers` frees the same number of the lead's workers while it is
+awake). Without `"checkpoint"` the helper stays resident with its cache given back (its dense weights and K/V stay on
+card 2, so the lead's peer tier is that much smaller). The shared arena must fit the folder it is in (`/dev/shm` is
+half of RAM by default: an arena of 72 GB, unsloth Q4_K_XL, needs a larger tmpfs).
+
+`--peer-link FILE --peer-link-role 0|1` alone (two one-card engines started by hand, each with its own server) is the
+mutual help without the elastic part.
+
+Engine control lines (stdin, any time; answered with `CTL ...`): `PEER_DETACH`, `PEER_ATTACH` (between requests),
+`POOL_HOLD n`, `LINK_SERVE 0|1`. Without `--peer-link` and without the `"elastic"` block the engine's output is
+byte-identical to the release.
+
+Measured (2x RTX 3090 + NVLink, Ryzen 9 3950X, 121 GB DDR4, IQ3_S, 262K, `--kv int8`; one request = a cached 54K
+coding prompt, 400-1,000 tokens, sampled):
+
+| | one request (decode / prefill 50K) | two requests at once |
+|---|---|---|
+| `--peer-device 1` (release) | 103-111 / 2,846 t/s (n=6) | one after the other: 99 t/s together |
+| elastic pair | 103-119 / 2,844 t/s (n=6): unchanged | **151-160 t/s together** (75-80 each) |
+| two one-card engines, no mutual help | | 136-139 t/s |
+
+What the first request pays when the second starts (3 runs): its decode goes from 102-112 t/s to 72-77 t/s for the
+4.7 s the helper needs to wake (the second card's experts come from the CPU meanwhile: detach 0.6 s, thaw 3.1 s,
+cache 1.1 s), then 82-87 t/s while both run, and back to 94-104 t/s once the second one has ended; its stream never
+stops (longest gap 0.07 s). The second request's first token comes after those 4.7 s plus its prompt.
+
+Where it does not help: unsloth Q4_K_XL on the same cards. A `--layer-split` already spends both cards' memory
+bandwidth on one conversation (95 t/s); the pair gives 87-88 t/s for two together (and 75-79 for one on the peer pair).
+Its dense weights and head (~2.9 GB) are read per verify window per engine, so two engines read them twice.

include/strata/core/peer_experts.hpp

@@ -17,6 +17,7 @@
 
 #include "strata/core/expert_cache.hpp"
 #include "strata/core/expert_source.hpp"
+#include "strata/core/peer_link.hpp"
 
 #include <cuda_runtime.h>
 
@@ -45,11 +46,19 @@ class PeerExperts {
               ExpertSource& src, int64_t n_layers, int64_t n_expert, int reserve_mib, int64_t max_slots,
               std::string& err);
     void close();
-    bool valid() const { return device_ >= 0; }
+    /// --peer-link: the tier is the OTHER engine's card (another process, see peer_link.hpp) instead of a card of this
+    /// one: `has` reads what that card holds, `launch`/`finish` post the rows to that engine and wait for them.  No
+    /// device of its own (no prompt-path share, no adaptive tier here: the other engine adapts its own cache).
+    void open_link(PeerLink* link) { close(); link_ = link; }
+    bool linked() const { return link_ != nullptr; }
+    bool valid() const { return device_ >= 0 || link_ != nullptr; }
     int device() const { return device_; }
 
     /// Whether the peer holds (layer, expert) right now.
-    bool has(int64_t layer, int64_t expert) const { return res_[(size_t) (layer * n_expert_ + expert)] >= 0; }
+    bool has(int64_t layer, int64_t expert) const {
+        if (link_ != nullptr) return link_->partner_has(layer, expert);
+        return res_[(size_t) (layer * n_expert_ + expert)] >= 0;
+    }
 
     /// The device address of (layer, expert)'s blob on the peer, or null when it is not resident.
     const uint8_t* slot_ptr(int64_t layer, int64_t expert) {
@@ -83,6 +92,7 @@ class PeerExperts {
     double ms_wait = 0;                               ///< host time spent in finish() waiting for the peer
 
 private:
+    PeerLink* link_ = nullptr;
     int device_ = -1;
     bool p2p_ = false;
     int64_t n_layers_ = 0, n_expert_ = 0;

include/strata/core/peer_link.hpp

@@ -0,0 +1,132 @@
+// include/strata/core/peer_link.hpp - multi-GPU, opt-in (--peer-link): two engine PROCESSES, one card each, that are
+// each other's peer expert tier ("mutual help").
+//
+// `--peer-device` (#531) gives ONE engine a second card's VRAM as an extra expert tier.  When a second request
+// arrives, the second card can instead run its own engine for it (one sequence per process, as always) - but then each
+// engine sees only its own card's expert cache, and the experts the other card holds become CPU misses.  The link
+// keeps both caches usable for both engines: each engine publishes which experts its card holds, and computes the
+// other engine's rows for those experts on its own card, with the same kernels as the `--peer-device` tier (q8_1
+// activations + native_expert_grouped), from the same activation values - so an expert gives the same rows on either
+// card.  The two caches are filled from different ranks of the profile (deduplicated), so both engines see roughly the
+// coverage of both cards.
+//
+// The two processes talk through one small shared-memory file (no CUDA IPC, no P2P, no MPS): the requesting engine's
+// pool thread writes the layer's activations and the (expert -> rows) plan, the owning engine's service thread copies
+// them to its card, computes, and writes the rows back; the requester waits for them where the `--peer-device` tier
+// waits for its card (PeerExperts::finish).  Only the OWNER ever touches its card and its cache, so all ordering
+// against the owner's own cache writes (adaptive swaps, the prompt path's loan, VRAM resizes) is local to the owner:
+// before such a write it holds the link (`hold`: the service finishes what it launched and starts nothing new), and
+// it publishes the new residency when the copies have landed.  An expert the owner no longer holds when a request
+// arrives (the requester planned with a residency a moment old) is copied in from the owner's RAM copy for that one
+// request - slower, the same rows.
+#pragma once
+
+#include <atomic>
+#include <cstdint>
+#include <mutex>
+#include <string>
+#include <thread>
+#include <vector>
+
+namespace strata::core {
+
+class ExpertCache;
+class ExpertSource;
+struct PeerLinkShm;
+
+class PeerLink {
+public:
+    PeerLink() = default;
+    ~PeerLink();
+    PeerLink(const PeerLink&) = delete;
+    PeerLink& operator=(const PeerLink&) = delete;
+
+    /// Maps the link file at `path`: the first engine creates it, the other joins (waiting up to `wait_s` for its
+    /// size); both check the model geometry.  Remove a stale file before starting either engine.
+    bool open(const std::string& path, int role, int64_t n_layers, int64_t n_expert, double wait_s, std::string& err);
+    void close();
+    bool valid() const { return shm_ != nullptr; }
+    int role() const { return role_; }
+
+    // ---- requester side (this engine's pool thread) ----
+    /// Whether the other engine serves now and its card holds (layer, expert).
+    bool partner_has(int64_t layer, int64_t expert) const;
+    /// Posts the entries with kind[i] == 2 of one layer (x: n_tok rows of the model's width).
+    bool request(int64_t layer, const float* x, const int32_t* ids, int64_t n_tok, int64_t k, const int32_t* kind,
+                 std::string& err);
+    /// Waits for the posted rows and writes them into `out` (row i = entry i).
+    bool wait(float* out, std::string& err);
+    int64_t entries() const { return entries_; }
+    int64_t experts() const { return experts_; }
+    double ms_wait = 0;
+
+    // ---- owner side (this engine serves the other one) ----
+    /// Starts the service thread on the calling thread's CUDA device, computing from `cache` (slots) with the
+    /// residency `host_res` (published now) and `src` (the RAM copy, for an expert no longer resident).
+    bool serve(ExpertCache& cache, ExpertSource& src, const std::vector<int32_t>& host_res, std::string& err);
+    /// Before this engine writes into or unmaps its cache's slots: the service finishes what it launched and
+    /// starts nothing new until `unhold`.  Nests on the same thread.
+    /// Whether the other engine may plan rows for this card (it sees this before it plans each layer).  Off while
+    /// this engine sleeps; a request already posted is still answered.
+    void set_serving(bool on);
+    void hold();
+    void unhold();
+    /// The residency the service (and the other engine's plans) use from now on.  Call when the copies behind it
+    /// have landed (removals: under `hold`, before the slots are written).
+    void publish(const std::vector<int32_t>& host_res);
+    /// Routed entries this engine computed for the other one per (layer, expert), for the adaptive tier's victims
+    /// (an expert the other engine uses is not the least used one).  Written by the service thread.
+    const float* served_usage() const { return svc_usage_.empty() ? nullptr : svc_usage_.data(); }
+    void decay_served(float f);
+    int64_t served_entries() const { return served_entries_.load(std::memory_order_relaxed); }
+    int64_t served_fallbacks() const { return served_fallback_.load(std::memory_order_relaxed); }
+    void stop();
+    /// timing of the service (owner side), cumulative: requests, ms from post to pickup, ms enqueueing (incl. lock),
+    /// ms in handle (enqueue + GPU)
+    int64_t n_handled() const { return n_handled_; }
+    double t_recv_ms() const { return t_recv_ms_; }
+    double t_enqueue_ms() const { return t_enqueue_ms_; }
+    double t_handle_ms() const { return t_handle_ms_; }
+
+private:
+    int64_t n_handled_ = 0;
+    double t_recv_ms_ = 0, t_enqueue_ms_ = 0, t_handle_ms_ = 0;
+    void loop();
+    bool handle(std::string& err);
+
+    PeerLinkShm* shm_ = nullptr;
+    size_t map_bytes_ = 0;
+    int role_ = -1;
+    int64_t n_layers_ = 0, n_expert_ = 0;
+    int32_t* res_mine_ = nullptr;          // in the file: what this card holds (the other engine plans with it)
+    const int32_t* res_partner_ = nullptr; // in the file: what the other card holds
+    // requester
+    uint64_t seq_ = 0;
+    std::vector<int32_t> row_of_;
+    int64_t rows_posted_ = 0, entries_ = 0, experts_ = 0;
+    // owner
+    ExpertCache* cache_ = nullptr;
+    ExpertSource* src_ = nullptr;
+    int device_ = -1;
+    std::vector<int32_t> svc_res_;
+    std::vector<float> svc_usage_;
+    std::atomic<int64_t> served_entries_{0}, served_fallback_{0};
+    std::recursive_mutex mu_;
+    int hold_depth_ = 0;
+    std::thread thread_;
+    std::atomic<bool> stop_{false};
+    bool registered_ = false;
+    void* stream_ = nullptr;               // cudaStream_t, high priority
+    void* d_shm_ = nullptr;                // the link file's device address (mapped)
+    float* d_x_ = nullptr;                 // (unused: zero-copy)
+    float* d_out_ = nullptr;
+    void* d_meta_ = nullptr;               // device address of h_meta_ (mapped)
+    void* h_meta_ = nullptr;
+    uint8_t* d_q8_ = nullptr;
+    void* d_scratch_ = nullptr;
+    uint8_t* d_stage_ = nullptr;           // blobs of experts no longer resident, kStage at a time
+    uint64_t stage_blob_ = 0;
+    uint64_t last_seq_ = 0;
+};
+
+}  // namespace strata::core

include/strata/hip_compat/cuda_runtime.h

@@ -112,6 +112,8 @@ hipError_t mem_get_info(size_t* free_bytes, size_t* total_bytes);
 #define cudaStreamCaptureStatusNone hipStreamCaptureStatusNone
 #define cudaStreamCreate hipStreamCreate
 #define cudaStreamCreateWithFlags hipStreamCreateWithFlags
+#define cudaStreamCreateWithPriority hipStreamCreateWithPriority
+#define cudaDeviceGetStreamPriorityRange hipDeviceGetStreamPriorityRange
 #define cudaStreamDestroy hipStreamDestroy
 #define cudaStreamEndCapture hipStreamEndCapture
 #define cudaStreamIsCapturing hipStreamIsCapturing

include/strata/kernels/cpu/pool.hpp

@@ -158,6 +158,11 @@ class ExpertPool {
     ExpertPool& operator=(const ExpertPool&) = delete;
 
     int workers() const { return n_; }
+    /// Elastic multi-engine (two engine processes on one CPU): hold the LAST `n` workers out of every batch - they
+    /// sleep (still counted as parked) until released, so another engine's workers can have their cores.  0 releases
+    /// all.  Call between batches (the serve loop does it between verify windows / requests).  Returns the number held.
+    int set_hold(int n);
+    int held() const { return held_; }
     /// Whether the host thread also drains.  Reported at startup, because "the engine adapts to the machine it
     /// is on" is only true if the engine says which adaptation it took.
     bool host_works() const { return host_works_; }
@@ -274,6 +279,10 @@ class ExpertPool {
     alignas(64) std::atomic<uint32_t> sleepers_{0};
     std::mutex sleep_mu_;
     std::condition_variable sleep_cv_;
+    std::unique_ptr<std::atomic<bool>[]> hold_;      ///< set_hold: worker i sits the batches out
+    int held_ = 0;
+    std::mutex hold_mu_;
+    std::condition_variable hold_cv_;
     std::chrono::microseconds spin_before_sleep_{kSpinBeforeSleep};
     std::vector<std::thread> threads_;
     std::vector<ExpertScratch> scratch_;   // one per worker: no allocation, no false sharing of the hot data

include/strata/platform/hip_compat/strata_hip.h

@@ -127,6 +127,8 @@ template <typename T> inline hipError_t strata_host_alloc(T** p, size_t bytes, u
 #define cudaHostGetDevicePointer hipHostGetDevicePointer
 #define cudaStreamCreate hipStreamCreate
 #define cudaStreamCreateWithFlags hipStreamCreateWithFlags
+#define cudaStreamCreateWithPriority hipStreamCreateWithPriority
+#define cudaDeviceGetStreamPriorityRange hipDeviceGetStreamPriorityRange
 #define cudaStreamDestroy hipStreamDestroy
 #define cudaStreamSynchronize hipStreamSynchronize
 #define cudaStreamQuery hipStreamQuery