Merge branch 'feat/router-multigpu' into feat/wp-phase2

Author: kmbandy

docs/dev/weight-paging-batch-eval-continuation.md

@@ -511,3 +511,26 @@ still ahead.
 hits (routing locality low at this cache size). Levers to go higher: (1) fix the depth>4 hang → QD>4;
 (2) Phase 2 (dense on 6900XT eGPU frees full 32GB R9700 for expert cache); (3) Phase 3 frequency-biased
 retention + prefetch coverage. Slot-count alone is a weak lever. Phase 1 gate: PASSED (coherent, >>0.4).
+
+---
+
+## 2026-07-07 — QD>4 HANG FIXED (Codex, validated) + parallel Codex work
+
+**QD>4 hang FIXED** (commit 945f9074a, merged): io_uring backends counted `pending_` when SQEs
+were PREPARED but ignored `io_uring_submit()`'s return. At depth 8/16 under resident-dense churn the
+CQ ring fills → `io_uring_submit` returns -EBUSY (SQEs NOT submitted) → a blocking wait_cqe waits for
+a req_id never kernel-owned → hang. Fix: track prepared-but-unsubmitted req_ids (pending_submit_
+deque), drain the CQ into the demux ready_ buffer on -EBUSY and retry; synthesize ErrorNoSubmit if
+truly stuck so waiters never hang. Applied to host + P2P io_uring. Regression test
+test_file_io_submit_batch_depth_one_targeted_waits. GPU-VALIDATED: depth-8 P2P resident-dense (was
+hanging) now health@40s, decode 0.9872 t/s, coherent ("...Rome in 753 BC."), sync_fallbacks=0.
+
+**KEY FINDING: QD is NOT the bottleneck.** depth-8 (0.99) ≈ depth-4 (1.04). page_ins 11348,
+evictions 9359, prefetch_hit_rate 0% — the expert cache thrashes (2000 slots ≪ working set, low
+cross-token routing locality). The path to 1-3 t/s is Phase 2 (dense on 6900XT eGPU → full 32GB
+R9700 for expert cache) + Phase 3 (frequency-biased retention + prefetch coverage), NOT queue depth.
+
+**Phase 2 committed on feat/wp-phase2 (Codex, d8196031a) — NOT yet GPU-validated.**
+--weight-paging-resident-device flag, dense→resident/expert→paging tensor_buft_overrides, per-device
+pager pools + relaxed >1-device guard, two-pool unit test. 38/38 unit tests. Needs multi-device
+forward validation on the R9700+6900XT(TB3) rig (cross-device activation copies are code-trace-only).

src/weight-pager/wp-file-io-p2p.cpp

@@ -5,6 +5,7 @@
 #include <cerrno>
 #include <cstdlib>
 #include <cstring>
+#include <deque>
 #include <memory>
 #include <vector>
 #include <unistd.h>
@@ -103,7 +104,7 @@ class IoUringP2PFileIOLayer : public FileIOLayer {
 
         struct io_uring_sqe * sqe = io_uring_get_sqe(&ring_);
         if (sqe == nullptr) {
-            io_uring_submit(&ring_);
+            flush_submissions_();
             sqe = io_uring_get_sqe(&ring_);
         }
         if (sqe == nullptr) {
@@ -114,18 +115,19 @@ class IoUringP2PFileIOLayer : public FileIOLayer {
         io_uring_prep_read(sqe, fd_idx, mapped_dst, (unsigned) size, (off_t) offset);
         sqe->flags    |= IOSQE_FIXED_FILE;
         sqe->user_data = req_id;
+        pending_submit_.push_back(req_id);
         ++pending_;
         return true;
     }
 
     void flush() override {
+        if (ring_ok_ && !pending_submit_.empty()) {
+            flush_submissions_();
+        }
         if (!p2p_enabled_ && pending_ == 0) {
             host_->flush();
             return;
         }
-        if (ring_ok_ && pending_ > 0) {
-            io_uring_submit(&ring_);
-        }
     }
 
     // Reap one raw completion for the FileIOLayer base demux. Drains the P2P
@@ -136,6 +138,13 @@ class IoUringP2PFileIOLayer : public FileIOLayer {
     // buffering happen in the base.
     bool reap_raw_(int timeout_ms, IoResult & out) override {
         if (ring_ok_ && pending_ > 0) {
+            if (!pending_submit_.empty()) {
+                flush_submissions_();
+            }
+            if (pending_ == 0) {
+                return false;
+            }
+
             struct io_uring_cqe * cqe = nullptr;
             int ret = 0;
             if (timeout_ms < 0) {
@@ -241,6 +250,90 @@ class IoUringP2PFileIOLayer : public FileIOLayer {
     IoUringP2PFileIOLayer(std::vector<int> fds, std::unique_ptr<FileIOLayer> host)
         : fds_(std::move(fds)), host_(std::move(host)) {}
 
+    void synthesize_submit_failures_(int err) {
+        while (!pending_submit_.empty()) {
+            const uint64_t req_id = pending_submit_.front();
+            pending_submit_.pop_front();
+
+            IoResult r;
+            r.req_id     = req_id;
+            r.status     = IoStatus::ErrorNoSubmit;
+            r.bytes_read = err;
+            ready_[req_id] = r;
+            --pending_;
+        }
+    }
+
+    bool flush_submissions_() {
+        if (!ring_ok_ || pending_submit_.empty()) {
+            return true;
+        }
+
+        while (!pending_submit_.empty()) {
+            int ret = 0;
+            do {
+                ret = io_uring_submit(&ring_);
+            } while (ret == -EINTR);
+
+            if (ret == -EBUSY || ret == -EAGAIN || ret == 0) {
+                bool drained = false;
+                while (reap_ready_cqe_()) {
+                    drained = true;
+                }
+                if (drained) {
+                    continue;
+                }
+            }
+            if (ret < 0) {
+                switch_to_host_errno_("io_uring submit failed", -ret);
+                synthesize_submit_failures_(ret);
+                return false;
+            }
+            if (ret == 0) {
+                switch_to_host_("io_uring submit made no progress", 0);
+                synthesize_submit_failures_(-EAGAIN);
+                return false;
+            }
+
+            for (int i = 0; i < ret && !pending_submit_.empty(); ++i) {
+                pending_submit_.pop_front();
+            }
+        }
+        return true;
+    }
+
+    bool reap_ready_cqe_() {
+        if (pending_ == 0) {
+            return false;
+        }
+
+        struct io_uring_cqe * cqe = nullptr;
+        int ret = io_uring_peek_cqe(&ring_, &cqe);
+        if (ret == -EAGAIN || ret == -EINTR || cqe == nullptr) {
+            return false;
+        }
+        if (ret < 0) {
+            return false;
+        }
+
+        IoResult r;
+        r.req_id = cqe->user_data;
+        const int res = cqe->res;
+        io_uring_cqe_seen(&ring_, cqe);
+        --pending_;
+
+        if (res < 0) {
+            r.status     = IoStatus::ErrorIo;
+            r.bytes_read = res;
+            switch_to_host_errno_("read failed", -res);
+        } else {
+            r.status     = IoStatus::Ok;
+            r.bytes_read = res;
+        }
+        ready_[r.req_id] = r;
+        return true;
+    }
+
     bool init_(int queue_depth, const FileIOP2PConfig & cfg) {
         int rt_version = 0;
         if (hipRuntimeGetVersion(&rt_version) == hipSuccess) {
@@ -386,6 +479,7 @@ class IoUringP2PFileIOLayer : public FileIOLayer {
     bool ring_ok_ = false;
     bool files_registered_ = false;
     int pending_ = 0;
+    std::deque<uint64_t> pending_submit_;
     struct io_uring ring_ {};
     void * libhsa_ = nullptr;
     HsaExportDmaBufFn hsa_export_ = nullptr;

src/weight-pager/wp-file-io.cpp

@@ -233,21 +233,21 @@ class IoUringAsyncFileIO : public FileIOLayer {
         struct io_uring_sqe * sqe = io_uring_get_sqe(&ring_);
         if (sqe == nullptr) {
             // Ring full: flush and retry once.
-            io_uring_submit(&ring_);
+            flush_submissions_();
             sqe = io_uring_get_sqe(&ring_);
             if (sqe == nullptr) return false;
         }
         // Use registered-file index (faster than passing raw fds).
         io_uring_prep_read(sqe, fd_idx, dst, (unsigned) size, (off_t) offset);
         sqe->flags     |= IOSQE_FIXED_FILE;
         sqe->user_data  = req_id;
+        pending_submit_.push_back(req_id);
         ++pending_;
         return true;
     }
 
     void flush() override {
-        if (!ring_ok_ || pending_ == 0) return;
-        io_uring_submit(&ring_);
+        flush_submissions_();
     }
 
     // Reap one completion from the ring. The demux/routing (buffering foreign
@@ -265,6 +265,12 @@ class IoUringAsyncFileIO : public FileIOLayer {
         if (pending_ == 0) {
             return false;  // nothing in flight
         }
+        if (!pending_submit_.empty()) {
+            flush_submissions_();
+        }
+        if (pending_ == 0) {
+            return false;
+        }
 
         struct io_uring_cqe * cqe = nullptr;
         int                   ret = 0;
@@ -347,28 +353,113 @@ class IoUringAsyncFileIO : public FileIOLayer {
             struct io_uring_sqe * sqe = io_uring_get_sqe(&ring_);
             if (sqe == nullptr) {
                 // SQ ring full mid-batch — flush what we've prepped, then retry once.
-                io_uring_submit(&ring_);
+                flush_submissions_();
                 sqe = io_uring_get_sqe(&ring_);
                 if (sqe == nullptr) break;
             }
             io_uring_prep_read(sqe, r.fd_idx, r.dst, (unsigned) r.size, (off_t) r.offset);
             sqe->flags     |= IOSQE_FIXED_FILE;
             sqe->user_data  = r.req_id;
+            pending_submit_.push_back(r.req_id);
             ++pending_;
             ++n_queued;
         }
         // Pass 2: single io_uring_submit for the batch. This is the MAD-235
         // win — one syscall covers N expert-prefetches for the MoE layer
         // instead of N separate submits.
         if (n_queued > 0) {
-            io_uring_submit(&ring_);
+            flush_submissions_();
         }
         return n_queued;
     }
 
 private:
     explicit IoUringAsyncFileIO(std::vector<int> fds) : fds_(std::move(fds)) {}
 
+    void synthesize_submit_failures_(int err) {
+        while (!pending_submit_.empty()) {
+            const uint64_t req_id = pending_submit_.front();
+            pending_submit_.pop_front();
+
+            IoResult r;
+            r.req_id     = req_id;
+            r.status     = IoStatus::ErrorNoSubmit;
+            r.bytes_read = err;
+            ready_[req_id] = r;
+            --pending_;
+        }
+    }
+
+    bool flush_submissions_() {
+        if (!ring_ok_ || pending_submit_.empty()) {
+            return true;
+        }
+
+        while (!pending_submit_.empty()) {
+            int ret = 0;
+            do {
+                ret = io_uring_submit(&ring_);
+            } while (ret == -EINTR);
+
+            if (ret == -EBUSY || ret == -EAGAIN || ret == 0) {
+                bool drained = false;
+                while (reap_ready_cqe_()) {
+                    drained = true;
+                }
+                if (drained) {
+                    continue;
+                }
+            }
+            if (ret < 0) {
+                LLAMA_LOG_WARN("wp::IoUringAsyncFileIO: io_uring_submit failed: %s\n",
+                               strerror(-ret));
+                synthesize_submit_failures_(ret);
+                return false;
+            }
+            if (ret == 0) {
+                LLAMA_LOG_WARN("wp::IoUringAsyncFileIO: io_uring_submit made no progress with %zu SQEs pending\n",
+                               pending_submit_.size());
+                synthesize_submit_failures_(-EAGAIN);
+                return false;
+            }
+
+            for (int i = 0; i < ret && !pending_submit_.empty(); ++i) {
+                pending_submit_.pop_front();
+            }
+        }
+        return true;
+    }
+
+    bool reap_ready_cqe_() {
+        if (pending_ == 0) {
+            return false;
+        }
+
+        struct io_uring_cqe * cqe = nullptr;
+        int ret = io_uring_peek_cqe(&ring_, &cqe);
+        if (ret == -EAGAIN || ret == -EINTR || cqe == nullptr) {
+            return false;
+        }
+        if (ret < 0) {
+            return false;
+        }
+
+        IoResult r;
+        r.req_id = cqe->user_data;
+        const int res = cqe->res;
+        if (res < 0) {
+            r.status     = IoStatus::ErrorIo;
+            r.bytes_read = res;
+        } else {
+            r.status     = IoStatus::Ok;
+            r.bytes_read = res;
+        }
+        io_uring_cqe_seen(&ring_, cqe);
+        --pending_;
+        ready_[r.req_id] = r;
+        return true;
+    }
+
     bool init_(int queue_depth) {
         int ret = io_uring_queue_init(queue_depth, &ring_, 0);
         if (ret < 0) {
@@ -397,6 +488,7 @@ class IoUringAsyncFileIO : public FileIOLayer {
     }
 
     std::vector<int>   fds_;
+    std::deque<uint64_t> pending_submit_;
     bool               ring_ok_          = false;
     bool               files_registered_ = false;
     int                pending_          = 0;