The host thread on the last core, the pool off the second GPU's interrupt core, copies in batches

Windows sends each GPU's interrupts to one logical processor (here the 5070 Ti's to the first, the 3090's to
the second, one MSI each), and the host thread ran on the first: generating IQ3_XXS thinking text the 5070 Ti
raised ~12,000 interrupts a second, each ~13 us in dxgkrnl's ISR and ~3 us in DPCs, a fifth of that processor.
A copy raises one when it lands, so the prefetch's copies, one per expert, raised most of them.  The host saw
a layer's ring up to 30-110 us late (p99), its call submitting the 5070 Ti's share took 18-24 us (14-17 on
another core), and in the pool's gate/up phase it finished last in 42% of the layers with a share.

- The host thread on the last physical core, the pool's workers on the others; with a second GPU the first
  core stays with the OS and its interrupts (6 workers here, from 7).  core_plan() decides; the startup log
  says where the threads run.
- The 5070 Ti's prefetch copies and each adaptive-tier call's copies in one cudaMemcpyBatchAsync (copy_blobs):
  the first processor's interrupt time 10% -> 4% on IQ3_XXS thinking text.

Window hashes identical to 7105fcc's (--adapt-every 0, both models); pool_test --stress passes.  Fixed text,
4 runs each in both orders, t/s: both GPUs IQ3_XXS code +1.2%, thinking +0.5%, after 16K +1.3%, edit -0.7%;
UD-Q4_K_XL +5.9 / +0.7 / +1.7 / -0.1%; the 3090 alone IQ3_XXS code +4.9%, thinking +0.9%, UD-Q4_K_XL +3.2 /
+2.1%.  16K prompt times unchanged (IQ3_XXS 8.11 -> 7.93 s, UD-Q4_K_XL 8.53 -> 8.48 s).

Claude-Session: https://claude.ai/code/session_019XxRxQQNyk7Ycdr4ComSs1

include/strata/core/adaptive_tier.hpp

@@ -106,9 +106,14 @@ private:
     cudaEvent_t evs_[kBatches] = {};
     size_t batch_sent_[kBatches] = {};
     int first_ = 0, flying_ = 0;
+    // a call's copies, sent as one batch (copy_blobs)
+    std::vector<void*> cp_dst_;
+    std::vector<const void*> cp_src_;
+    std::vector<size_t> cp_bytes_;
     uint64_t bytes_of(const Move& m) const;
     void evict(const Move& m);
     bool copy(const Move& m, uint64_t off, uint64_t n, std::string& err);
+    bool send(std::string& err);
     bool time_begin(bool worth);
     bool end_batch(bool timed, uint64_t bytes, std::string& err);
 };

include/strata/core/expert_cache.hpp

@@ -152,4 +152,9 @@ private:
     int64_t admitted_ = 0;
 };
 
+/// Copies `n` pinned host blobs into VRAM on `stream` as one submission (cudaMemcpyBatchAsync: the copies in any
+/// order, the batch in the stream's).  Under WDDM a copy submitted alone raises an interrupt when it completes, on the
+/// CPU its GPU's interrupts go to (~10 us there, in the ISR and a DPC); a batch raises one.
+bool copy_blobs(void* const* dst, const void* const* src, const size_t* bytes, size_t n, void* stream);
+
 }  // namespace strata::core

include/strata/kernels/cpu/pool.hpp

@@ -62,15 +62,24 @@ struct ExpertJobMulti {
 /// worker.  On the 6-core/12-thread machine this project measures on, `hardware_concurrency()/2` workers on
 /// logical processors 0..5 would put every worker on a sibling pair and halve the useful bandwidth - which is
 /// exactly the kind of error that shows up as "the CPU path is slower than the model says" with no clue why.
-///
-/// `skip_first` drops the first core, which P2.S3 reserves for the host loop.
-std::vector<int> physical_cores(bool skip_first);
+std::vector<int> physical_cores();
+
+/// Where the host loop and the pool's workers run, one logical processor each: the host on the last physical core,
+/// the workers on the others.  With `spare_first` (a second GPU) the first core stays with the OS: Windows sends the
+/// GPUs' interrupts there (here the 5070 Ti's to its first logical processor, the 3090's to its second), and the
+/// second GPU's copies and launches raise thousands a second during generation, each ~16 us there in the ISR and a
+/// DPC.  With four physical cores or fewer the first core takes a worker anyway.
+struct CorePlan {
+    int host = -1;
+    std::vector<int> workers;
+};
+CorePlan core_plan(bool spare_first);
 
 /// **THE RESERVATION IS A FICTION UNLESS THE HOST IS ACTUALLY PUT THERE.**
 ///
-/// `physical_cores(true)` keeps the workers off the first physical core so that the host loop can spin on
-/// `cudaEventQuery` without stealing a worker's cycles.  Nothing in the pool can enforce the other half of
-/// that, so this is it: the host loop calls this on entry and restores on exit.
+/// `core_plan()` keeps the workers off the host's core so that the host loop can spin on `cudaEventQuery`
+/// without stealing a worker's cycles.  Nothing in the pool can enforce the other half of that, so this is it:
+/// the host loop calls this on entry and restores on exit.
 ///
 /// MEASURED, and this is why it exists: the pool runs at **36.32 GB/s on 5 workers with nothing else running**
 /// - exactly 5/6 of L9's 44.14 on 6 - and at **26.9 GB/s inside the host loop**, where the unpinned spinning
@@ -86,13 +95,13 @@ void restore_thread_affinity(long long previous);
 #endif
 class ExpertPool {
 public:
-    /// `n_workers <= 0` means "every physical core except the first".  Workers are pinned to physical cores
-    /// (minus core 0 by default) and each owns one `ExpertScratch`, so nothing in the token path allocates.
+    /// `n_workers <= 0` means a worker on each of `core_plan(spare_first).workers`.  Workers are pinned to those
+    /// cores and each owns one `ExpertScratch`, so nothing in the token path allocates.
     ///
     /// **`host_works` PUTS THE HOST THREAD INTO THE DRAIN (R2.2's FIRST HALF).**
     ///
-    /// The pool reserves core 0 for the host loop so the doorbell spin cannot steal a worker's cycles - but
-    /// during `run()` the host does not spin, it waits, so core 0 is idle for the whole drain. Measured on the
+    /// The pool keeps a core for the host loop so the doorbell spin cannot steal a worker's cycles - but
+    /// during `run()` the host does not spin, it waits, so that core is idle for the whole drain. Measured on the
     /// 6-core machine this project targets: the engine's pool drains at **33.7 GB/s** (663.6 MB of expert
     /// blobs in 19.71 ms/token) where the same kernel on 5 workers should reach 5/6 x 44.14 = 36.8 and the
     /// machine measures 44.14 GB/s on all six. So the sixth core is being paid for and not used.
@@ -100,7 +109,7 @@ public:
     /// With `host_works`, `run()` claims jobs itself instead of spinning on `done_`, and the pool is six
     /// threads on six cores. `false` is the A/B arm and exists so the change is measurable rather than
     /// asserted - the counter it moves is `pool phases ... drain`, which is host-side and needs no profiler.
-    explicit ExpertPool(int n_workers = 0, bool pin = true, bool host_works = true);
+    explicit ExpertPool(int n_workers = 0, bool pin = true, bool host_works = true, bool spare_first = false);
     ~ExpertPool();
     ExpertPool(const ExpertPool&) = delete;
     ExpertPool& operator=(const ExpertPool&) = delete;

src/core/adaptive_tier.cpp

@@ -63,16 +63,26 @@ void AdaptiveTier::evict(const Move& m) {
 
 bool AdaptiveTier::copy(const Move& m, uint64_t off, uint64_t n, std::string& err) {
     const uint8_t* b = src_->blob(m.layer, m.in);
-    if (b == nullptr ||
-        cudaMemcpyAsync(cache_->device_slot(m.slot) + off, b + off, (size_t) n, cudaMemcpyHostToDevice, stream_) !=
-            cudaSuccess) {
+    if (b == nullptr) {
         err = "adaptive tier: a refill copy failed";
         return false;
     }
+    cp_dst_.push_back(cache_->device_slot(m.slot) + off);
+    cp_src_.push_back(b + off);
+    cp_bytes_.push_back((size_t) n);
     sent_bytes += n;
     return true;
 }
 
+bool AdaptiveTier::send(std::string& err) {
+    const bool ok = copy_blobs(cp_dst_.data(), cp_src_.data(), cp_bytes_.data(), cp_dst_.size(), stream_);
+    cp_dst_.clear();
+    cp_src_.clear();
+    cp_bytes_.clear();
+    if (!ok) err = "adaptive tier: a refill copy failed";
+    return ok;
+}
+
 // Takes the last timed copies' rate once they have landed; then starts timing these when `worth` (long enough for
 // their start and end not to dominate) and none are being timed.
 bool AdaptiveTier::time_begin(bool worth) {
@@ -129,7 +139,7 @@ bool AdaptiveTier::pump_bytes(uint64_t bytes, std::string& err) {
             ++sent_;
         }
     }
-    return done == 0 || end_batch(timed, done, err);
+    return done == 0 || (send(err) && end_batch(timed, done, err));
 }
 
 bool AdaptiveTier::pump(uint64_t budget, uint64_t& sent, std::string& err) {
@@ -153,7 +163,7 @@ bool AdaptiveTier::pump(uint64_t budget, uint64_t& sent, std::string& err) {
         if (!copy(queued_[sent_], 0, bytes_of(queued_[sent_]), err)) return false;
     }
     staged_ = sent_;
-    return end_batch(timed, sent, err);
+    return send(err) && end_batch(timed, sent, err);
 }
 
 bool AdaptiveTier::adapt(std::vector<float>& usage, std::string& err, bool decay) {

src/core/expert_cache.cpp

@@ -245,6 +245,15 @@ bool ExpertCache::fill_slot(int32_t slot, const uint8_t* host_blob, void* stream
     return true;
 }
 
+bool copy_blobs(void* const* dst, const void* const* src, const size_t* bytes, size_t n, void* stream) {
+    if (n == 0) return true;
+    cudaMemcpyAttributes attr{};
+    attr.srcAccessOrder = cudaMemcpySrcAccessOrderStream;
+    attr.flags = cudaMemcpyFlagPreferOverlapWithCompute;   // the copy engines, beside the kernels
+    size_t first = 0;
+    return cudaMemcpyBatchAsync(dst, src, bytes, n, &attr, &first, 1, (cudaStream_t) stream) == cudaSuccess;
+}
+
 bool ExpertCache::fill_slot_blocking(int32_t slot, const uint8_t* host_blob, std::string& err, int64_t bytes) {
     const size_t n = (size_t) (bytes > 0 && bytes <= blob_ ? bytes : blob_);
     uint8_t* dst = device_slot(slot);

src/core/second_gpu.cpp

@@ -117,11 +117,15 @@ bool SecondGpu::prefetch(int64_t layer, const int32_t* ids, const uint8_t* const
     if (n <= 0) return true;
     if (n > pre_max_ || bytes > pre_cap_) { err = "second GPU: a prefetch does not fit its slots"; return false; }
     DeviceScope scope(dev_, main_);
-    bool ok = cudaStreamWaitEvent(pre_s_, ev_, 0) == cudaSuccess;   // the last submitted layer may read the slots
-    for (int i = 0; ok && i < n; ++i)
-        ok = cudaMemcpyAsync(d_pre_ + (size_t) i * pre_cap_, src[i], (size_t) bytes, cudaMemcpyHostToDevice, pre_s_) ==
-             cudaSuccess;
-    if (!ok || cudaEventRecord(pre_ev_, pre_s_) != cudaSuccess) {
+    void* dst[kPrefetchMax];
+    size_t size[kPrefetchMax];
+    for (int i = 0; i < n; ++i) {
+        dst[i] = d_pre_ + (size_t) i * pre_cap_;
+        size[i] = (size_t) bytes;
+    }
+    if (cudaStreamWaitEvent(pre_s_, ev_, 0) != cudaSuccess ||   // the last submitted layer may read the slots
+        !copy_blobs(dst, (const void* const*) src, size, (size_t) n, pre_s_) ||
+        cudaEventRecord(pre_ev_, pre_s_) != cudaSuccess) {
         err = std::string("second GPU: prefetch: ") + cudaGetErrorString(cudaGetLastError());
         return false;
     }

src/core/session.cpp

@@ -569,15 +569,15 @@ bool SessionLoopScratch::init(size_t parts_bytes_in, std::string& err) {
         free();
         return false;
     }
-    // **PIN THE HOST ONCE, NOT ONCE PER TOKEN.**  `ExpertPool` builds its workers from `physical_cores(true)`,
-    // which drops the first physical core so the host loop can spin without taking a worker's cycles - and
+    // **PIN THE HOST ONCE, NOT ONCE PER TOKEN.**  `ExpertPool` builds its workers from `core_plan()`, which
+    // keeps them off the host's core so the host loop can spin without taking a worker's cycles - and
     // nothing in the pool can pin the host, so if this does not happen the spin is free to land on a worker's
     // core or its SMT sibling.  The symptom is not an error: it is a CPU path at 26.9 GB/s where the same pool
     // runs at 36.32.  It was being done and undone on EVERY token, which is a syscall pair on the critical path
     // for a property that wants to hold for the whole session.
-    const std::vector<int> cores = strata::kernels::cpu::physical_cores(false);
-    if (!cores.empty()) {
-        pinned_core = strata::kernels::cpu::pin_current_thread(cores[0]);
+    const int host = strata::kernels::cpu::core_plan(false).host;   // the last core, a second GPU or not
+    if (host >= 0) {
+        pinned_core = strata::kernels::cpu::pin_current_thread(host);
         pinned = true;
     }
     return true;

src/kernels/cpu/pool.cpp

@@ -19,7 +19,7 @@
 
 namespace strata::kernels::cpu {
 
-std::vector<int> physical_cores(bool skip_first) {
+std::vector<int> physical_cores() {
     std::vector<int> cores;
 #if defined(_WIN32)
     // Ask the OS rather than assuming a layout.  `hardware_concurrency()` returns LOGICAL processors, and on
@@ -55,10 +55,18 @@ std::vector<int> physical_cores(bool skip_first) {
     else
         for (unsigned i = 0; i < std::thread::hardware_concurrency(); ++i) cores.push_back((int) i);
 #endif
-    if (skip_first && !cores.empty()) cores.erase(cores.begin());
     return cores;
 }
 
+CorePlan core_plan(bool spare_first) {
+    const std::vector<int> cores = physical_cores();
+    CorePlan plan;
+    if (cores.empty()) return plan;
+    plan.host = cores.back();
+    plan.workers.assign(cores.begin() + (spare_first && cores.size() > 4 ? 1 : 0), cores.end() - 1);
+    return plan;
+}
+
 namespace {
 
 // `state_`: the epoch in the high half, then the closed bit, then the parked workers
@@ -112,8 +120,8 @@ void restore_thread_affinity(long long previous) {
 #endif
 }
 
-ExpertPool::ExpertPool(int n_workers, bool pin, bool host_works) : host_works_(host_works) {
-    const std::vector<int> cores = physical_cores(true);
+ExpertPool::ExpertPool(int n_workers, bool pin, bool host_works, bool spare_first) : host_works_(host_works) {
+    const std::vector<int> cores = core_plan(spare_first).workers;
     n_ = n_workers > 0 ? n_workers : (int) cores.size();
     if (n_ < 1) n_ = 1;
     state_.store(kClosed | (uint64_t) n_, std::memory_order_relaxed);   // epoch 0, closed, everyone parked
@@ -390,8 +398,8 @@ void ExpertPool::run(ExpertJob* jobs, int n) {
 
     // ---- **THE HOST DRAINS TOO (R2.2), INSTEAD OF SPINNING ON `done_`.**
     //
-    // The loop below used to be `while (done_ != n) _mm_pause();`.  The host is pinned to core 0 - the core
-    // `physical_cores(true)` deliberately keeps the five workers off - so for the whole drain that core was
+    // The loop below used to be `while (done_ != n) _mm_pause();`.  The host is pinned to its own core - the
+    // one `core_plan()` deliberately keeps the workers off - so for the whole drain that core was
     // idle while five cores did six cores' worth of work.  Measured before the change: 33.7 GB/s against
     // 5/6 x 44.14 = 36.8 for five workers and 44.14 for six.
     //

src/kernels/cpu/pool_test.cpp

@@ -314,9 +314,9 @@ int main(int argc, char** argv) {
     }
 
     // ---- the pool
-    const std::vector<int> cores = cpu::physical_cores(true);
+    const std::vector<int> cores = cpu::core_plan(false).workers;
     const int hw = (int) std::thread::hardware_concurrency();
-    std::printf("  %-44s %d logical, %d physical (skipping the first)\n", "cores the pool will use",
+    std::printf("  %-44s %d logical, %d physical (not the last, the host's)\n", "cores the pool will use",
                 hw, (int) cores.size());
 
     cpu::ExpertPool pool;

src/program/generate.cpp

@@ -1393,7 +1393,9 @@ int main(int argc, char** argv) {
     // --ple-io ram: the table goes into RAM once the SSD has delivered the experts and the draft layer
     if (ple_table.is_open() && !ple_table.start_ram_load(err))
         std::fprintf(stderr, "strata generate: %s; its rows stay on the SSD\n", err.c_str());
-    strata::kernels::cpu::ExpertPool pool(o.pool_workers, /*pin=*/true, /*host_works=*/!o.no_host_worker);
+    // with a second GPU the first core stays with the OS: that GPU's interrupts go there (core_plan)
+    const bool spare_first = o.second_gpu >= 0;
+    strata::kernels::cpu::ExpertPool pool(o.pool_workers, /*pin=*/true, /*host_works=*/!o.no_host_worker, spare_first);
     if (o.no_ple_prefetch) strata::kernels::ple_prefetch_enable(false);
     std::fprintf(stderr, "strata generate: session is up; locating the head\n");
     const strata::core::WeightRef* wo = wt.find("output.weight");
@@ -1688,8 +1690,12 @@ int main(int argc, char** argv) {
     strata::core::HitFn hit_fn =
         (o.no_pool || o.expert_cache <= 0) ? nullptr : &strata::core::expert_hit_run;
     void* pool_user = o.no_pool ? nullptr : (void*) &drive;
-    std::fprintf(stderr, "strata generate: %d expert-pool workers%s%s\n", pool.workers(),
-                 pool.host_works() ? " + the host thread" : "",
+    const strata::kernels::cpu::CorePlan cores = strata::kernels::cpu::core_plan(spare_first);
+    std::string on;
+    for (int i = 0; i < pool.workers() && i < (int) cores.workers.size(); ++i)
+        on += (i > 0 ? "," : "") + std::to_string(cores.workers[(size_t) i]);
+    std::fprintf(stderr, "strata generate: %d pool workers on logical processors %s, the host thread on %d%s%s\n",
+                 pool.workers(), on.c_str(), cores.host, pool.host_works() ? " (draining too)" : "",
                  o.no_pool ? " (UNUSED: --no-pool)" : "");
 
     // **THE MISALIGNMENT WARNING THAT STOOD HERE IS GONE, BECAUSE THE MISALIGNMENT IS FIXED.**