WIP: PD-DFlash — native offloaded-expert stat accessors (unblock §8 B0–B2)

benchmarks/dflash/_serving_measure.py

@@ -130,11 +130,14 @@ def measure_configuration(
         slo_ms=args.slo_ms,
         warnings=warnings,
     )
+    effective_h2d_gbps = args.measured_h2d_gbps
+    if effective_h2d_gbps is None and getattr(args, "probe_h2d", False):
+        effective_h2d_gbps = _probe_h2d_gbps()
     cost_terms = _collect_cost_terms(
         baseline=baseline,
         speculator=speculator,
         engine=engine,
-        measured_h2d_gbps=args.measured_h2d_gbps,
+        measured_h2d_gbps=effective_h2d_gbps,
         warnings=warnings,
     )
     return make_observation_row(
@@ -268,6 +271,42 @@ def _route_ahead_snapshot(speculator: Any) -> Tuple[float, Optional[int]]:
     return coverage, (int(wasted) if wasted is not None else None)
 
 
+def _probe_h2d_gbps(
+    nbytes: int = 256 * 1024 * 1024, iters: int = 20
+) -> Optional[float]:
+    import torch
+
+    if not torch.cuda.is_available():
+        return None
+    try:
+        elements = max(1, nbytes // 2)
+        host = torch.empty(elements, dtype=torch.float16, pin_memory=True)
+        device = torch.empty(elements, dtype=torch.float16, device="cuda:0")
+        torch.cuda.synchronize()
+        device.copy_(host, non_blocking=True)
+        torch.cuda.synchronize()
+        started = time.perf_counter()
+        for _ in range(max(1, iters)):
+            device.copy_(host, non_blocking=True)
+        torch.cuda.synchronize()
+        elapsed = max(time.perf_counter() - started, 1e-9)
+    except Exception:
+        return None
+    moved = float(elements * 2 * max(1, iters))
+    return moved / elapsed / 1_000_000_000.0
+
+
+def _native_wasted_prefetch_bytes(engine: Any) -> Optional[float]:
+    prefetcher = getattr(engine, "expert_prefetcher", None)
+    getter = getattr(prefetcher, "wasted_prefetch_bytes", None)
+    if callable(getter):
+        try:
+            return float(getter())
+        except Exception:
+            return None
+    return None
+
+
 def _extract_float(source: Any, names: Tuple[str, ...]) -> Optional[float]:
     for name in names:
         value = getattr(source, name, None)
@@ -300,11 +339,16 @@ def _collect_metrics(
     per_round_latency = elapsed / rounds
     coverage, wasted_bytes = _route_ahead_snapshot(speculator)
     if wasted_bytes is None:
-        warnings.append(
-            "wasted_prefetch_bytes unavailable from RouteAheadStats; a route-"
-            "ahead configuration on offloaded experts must report real bytes"
-        )
-        wasted_bytes = 0
+        native_wasted = _native_wasted_prefetch_bytes(engine)
+        if native_wasted is not None:
+            wasted_bytes = int(native_wasted)
+        else:
+            warnings.append(
+                "wasted_prefetch_bytes unavailable from RouteAheadStats; a "
+                "route-ahead configuration on offloaded experts must report "
+                "real bytes"
+            )
+            wasted_bytes = 0
 
     hit_rate = _extract_float(
         getattr(engine, "expert_prefetcher", engine),
@@ -358,6 +402,10 @@ def _collect_cost_terms(
     warnings: List[str],
 ) -> Dict[str, Any]:
     coverage, wasted_bytes = _route_ahead_snapshot(speculator)
+    if wasted_bytes is None:
+        native_wasted = _native_wasted_prefetch_bytes(engine)
+        if native_wasted is not None:
+            wasted_bytes = int(native_wasted)
     terms: Dict[str, Any] = {
         "route_ahead_coverage": coverage,
         "wasted_prefetch_bytes": wasted_bytes,

benchmarks/dflash/pd_dflash_serving.py

@@ -15,8 +15,10 @@
 Design contract (frozen here, cross-checked by the aggregator):
 
 * baselines are exactly B0-B3 with the design-doc §8 semantics;
-* the required generalization targets are ``Qwen/Qwen3-Coder-30B-A3B`` and
-  ``openai/gpt-oss-20b`` with their ``z-lab`` DFlash drafts;
+* the required generalization targets are
+  ``Qwen/Qwen3-Coder-30B-A3B-Instruct`` and ``openai/gpt-oss-20b`` with their
+  ``z-lab`` DFlash drafts (the bare ``Qwen/Qwen3-Coder-30B-A3B`` repo returns
+  HTTP 404 and never existed; ``-Instruct`` is the real ``qwen3_moe`` repo);
 * block sizes are 8 and 16, concurrency sweeps 1..32; and
 * every emitted observation carries the full ``REQUIRED_METRICS`` schema, plus
   the byte-accurate route-ahead ``wasted_prefetch_bytes`` from the instrumented
@@ -50,12 +52,12 @@
 }
 
 REQUIRED_MODELS: Tuple[str, ...] = (
-    "Qwen/Qwen3-Coder-30B-A3B",
+    "Qwen/Qwen3-Coder-30B-A3B-Instruct",
     "openai/gpt-oss-20b",
 )
 
 REQUIRED_DRAFTS: Dict[str, str] = {
-    "Qwen/Qwen3-Coder-30B-A3B": "z-lab/Qwen3-Coder-30B-A3B-DFlash",
+    "Qwen/Qwen3-Coder-30B-A3B-Instruct": "z-lab/Qwen3-Coder-30B-A3B-DFlash",
     "openai/gpt-oss-20b": "z-lab/gpt-oss-20b-DFlash",
 }
 
@@ -251,6 +253,7 @@ class RunnerArgs:
     requests: int
     warmup_rounds: int
     measured_h2d_gbps: Optional[float]
+    probe_h2d: bool
     slo_ms: Optional[float]
     seed: int
     device_memory_ratio: float
@@ -294,6 +297,15 @@ def build_arg_parser() -> argparse.ArgumentParser:
     parser.add_argument("--requests", type=int, default=64)
     parser.add_argument("--warmup-rounds", type=int, default=5)
     parser.add_argument("--measured-h2d-gbps", type=float, default=None)
+    parser.add_argument(
+        "--probe-h2d",
+        action="store_true",
+        help=(
+            "when --measured-h2d-gbps is unset, measure host->device "
+            "bandwidth on the visible GPU and use it for the §7 hide "
+            "inequality (never a theoretical PCIe figure)"
+        ),
+    )
     parser.add_argument("--slo-ms", type=float, default=None)
     parser.add_argument("--seed", type=int, default=1408)
     parser.add_argument(
@@ -323,6 +335,7 @@ def parse_args(argv: Optional[Sequence[str]] = None) -> RunnerArgs:
         requests=parsed.requests,
         warmup_rounds=parsed.warmup_rounds,
         measured_h2d_gbps=parsed.measured_h2d_gbps,
+        probe_h2d=parsed.probe_h2d,
         slo_ms=parsed.slo_ms,
         seed=parsed.seed,
         device_memory_ratio=parsed.device_memory_ratio,

core/model/model_topology.cpp

@@ -372,6 +372,24 @@ ArcherTopologyHandle::GetNodeVisitCounts() {
   return node_visit_counts;
 }
 
+std::tuple<std::int64_t, std::int64_t>
+ArcherTopologyHandle::GetResidentAndWastedBytes() {
+  std::lock_guard<std::mutex> lock(mutex_);
+  std::int64_t resident = 0;
+  std::int64_t wasted = 0;
+  for (auto& stage : pipeline_.stages) {
+    for (auto& node_body : stage->nodes) {
+      auto& node = node_body->node;
+      if (node == nullptr) continue;
+      if (node->device.is_cuda()) {
+        resident += node->byte_size;
+      }
+      wasted += node->byte_size * static_cast<std::int64_t>(node->unused_count);
+    }
+  }
+  return std::make_tuple(resident, wasted);
+}
+
 std::vector<std::size_t> ArcherTopologyHandle::GetChildVisitCounts() {
   std::lock_guard<std::mutex> lock(mutex_);
   int num_layers = 0;

core/model/model_topology.h

@@ -185,6 +185,7 @@ class ArcherTopologyHandle : public base::noncopyable {
   void DisableTrace() noexcept { trace_enabled_ = false; }
 
   std::vector<std::vector<std::size_t>> GetNodeVisitCounts();
+  std::tuple<std::int64_t, std::int64_t> GetResidentAndWastedBytes();
   std::vector<std::size_t> GetChildVisitCounts();
   void SetNodeVisitCounts(const std::vector<std::size_t>& visit_counts);
   void SetChildVisitCounts(const std::vector<std::size_t>& visit_counts);

core/parallel/expert_dispatcher.cpp

@@ -233,6 +233,8 @@ void ExpertDispatcher::Enqueue(CallArgs& args) {
     exec_args.out_dtype = c10::typeMetaToScalarType(hidden_states_.dtype());
     exec_args.evict = false;
     exec_args.hit = true;
+    cache_hit_count_.fetch_add(1, std::memory_order_relaxed);
+    cache_access_count_.fetch_add(1, std::memory_order_relaxed);
     // transfer_event = nullptr: expert is already on GPU, no H2D wait needed
 
     // module_->SetTensorsFromIds(expert_node->node->tensor_ids);
@@ -296,6 +298,39 @@ void ExpertDispatcher::ClearExpertCacheCounts() {
       expert_node->node->incache_visit_count = 0;
     }
   }
+  cache_hit_count_.store(0, std::memory_order_relaxed);
+  cache_access_count_.store(0, std::memory_order_relaxed);
+}
+
+std::int64_t ExpertDispatcher::GetCacheOccupancyBytes() {
+  std::int64_t total = 0;
+  for (auto& gpu_cache : cached_experts_) {
+    for (auto key : gpu_cache) {
+      int layer_idx = static_cast<int>(key >> 32);
+      int expert_idx = static_cast<int>(key & 0xFFFFFFFF);
+      if (expert_idx < 0 || expert_idx >= static_cast<int>(experts_.size())) {
+        continue;
+      }
+      if (layer_idx < 0 ||
+          layer_idx >= static_cast<int>(experts_[expert_idx].size())) {
+        continue;
+      }
+      auto& expert_node = experts_[expert_idx][layer_idx];
+      if (expert_node && expert_node->node) {
+        total += expert_node->node->byte_size;
+      }
+    }
+  }
+  return total;
+}
+
+double ExpertDispatcher::GetCacheHitRate() const {
+  std::uint64_t access = cache_access_count_.load(std::memory_order_relaxed);
+  if (access == 0) {
+    return 0.0;
+  }
+  std::uint64_t hits = cache_hit_count_.load(std::memory_order_relaxed);
+  return static_cast<double>(hits) / static_cast<double>(access);
 }
 
 // void ExpertDispatcher::GPUThreadFunc(int gpu_id) {
@@ -388,6 +423,8 @@ void ExpertDispatcher::GPUFetchFunc(int gpu_id) {
         exec_args.out_dtype = c10::typeMetaToScalarType(hidden_states_.dtype());
         exec_args.evict = false;
         exec_args.hit = true;
+        cache_hit_count_.fetch_add(1, std::memory_order_relaxed);
+        cache_access_count_.fetch_add(1, std::memory_order_relaxed);
         exec_args.transfer_event = nullptr;
         exec_queue_[gpu_id].Push(exec_args);
         continue;
@@ -537,6 +574,8 @@ void ExpertDispatcher::GPUFetchFunc(int gpu_id) {
       exec_args.out_dtype = c10::typeMetaToScalarType(hidden_states_.dtype());
       exec_args.evict = gpu_overload_[gpu_id].load(std::memory_order_acquire);
       exec_args.hit = cache_hit;
+      cache_access_count_.fetch_add(1, std::memory_order_relaxed);
+      if (cache_hit) cache_hit_count_.fetch_add(1, std::memory_order_relaxed);
       exec_args.transfer_event = transfer_done;
       // std::lock_guard<std::mutex> lock(exec_mutex_[gpu_id]);
       // exec_queue_[gpu_id].emplace_back(std::move(exec_args));

core/parallel/expert_dispatcher.h

@@ -108,6 +108,9 @@ class ExpertDispatcher : public base::noncopyable {
                       const std::vector<std::uint32_t>& tensor_ids,
                       std::string jit_path);
   void ClearExpertCacheCounts();
+  // Read-only observability accessors; neither alters routing/dispatch.
+  std::int64_t GetCacheOccupancyBytes();
+  double GetCacheHitRate() const;
   void SetExpectedQueue(int expected_pending = 0) {
     pending_.store(expected_pending);
   }
@@ -150,6 +153,10 @@ class ExpertDispatcher : public base::noncopyable {
 
   std::atomic<size_t> pending_;
 
+  // Passive counters for GetCacheHitRate(); reset by ClearExpertCacheCounts().
+  std::atomic<std::uint64_t> cache_hit_count_{0};
+  std::atomic<std::uint64_t> cache_access_count_{0};
+
   std::mutex pending_mutex_;
   std::condition_variable pending_cv_;
 

core/prefetch/archer_prefetch_handle.cpp

@@ -341,6 +341,14 @@ torch::Tensor ArcherPrefetchHandle::GetHitRate() {
   return trace;
 }
 
+std::int64_t ArcherPrefetchHandle::GetExpertOccupancyBytes() {
+  return std::get<0>(kTopologyHandle->GetResidentAndWastedBytes());
+}
+
+std::int64_t ArcherPrefetchHandle::GetWastedPrefetchBytes() {
+  return std::get<1>(kTopologyHandle->GetResidentAndWastedBytes());
+}
+
 void ArcherPrefetchHandle::SetTrace(const torch::Tensor& trace) {
   if (trace.dim() != 3 || !trace.is_contiguous() || !trace.is_cpu()) {
     DLOG_ERROR("Trace should be a contiguous 3D tensor on CPU");

core/prefetch/archer_prefetch_handle.h

@@ -43,6 +43,8 @@ class ArcherPrefetchHandle {
 
   torch::Tensor GetTrace();
   torch::Tensor GetHitRate();
+  std::int64_t GetExpertOccupancyBytes();
+  std::int64_t GetWastedPrefetchBytes();
   void SetTrace(const torch::Tensor& trace);
   void TraceRequest(const std::uint64_t request_id, const TensorID tensor_id);
   void SetTopology(const std::vector<

core/python/py_archer_prefetch.cpp

@@ -53,6 +53,10 @@ PYBIND11_MODULE(TORCH_EXTENSION_NAME, m) {
       //    ArcherPrefetchHandle::GetTrace)
       .def("get_hit_rate", (torch::Tensor(ArcherPrefetchHandle::*)()) &
                                ArcherPrefetchHandle::GetHitRate)
+      .def("get_expert_occupancy_bytes",
+           &ArcherPrefetchHandle::GetExpertOccupancyBytes)
+      .def("get_wasted_prefetch_bytes",
+           &ArcherPrefetchHandle::GetWastedPrefetchBytes)
       .def("set_trace", (void(ArcherPrefetchHandle::*)(const torch::Tensor&)) &
                             ArcherPrefetchHandle::SetTrace)
       //    .def("trace_request",
@@ -111,6 +115,9 @@ PYBIND11_MODULE(TORCH_EXTENSION_NAME, m) {
       .def("notify_fetch_start", &ExpertDispatcher::NotifyFetchStart)
       .def("clear_expert_cache_counts",
            &ExpertDispatcher::ClearExpertCacheCounts)
+      .def("get_cache_occupancy_bytes",
+           &ExpertDispatcher::GetCacheOccupancyBytes)
+      .def("get_cache_hit_rate", &ExpertDispatcher::GetCacheHitRate)
       .def("set_scales", &ExpertDispatcher::SetScales,
            "Store fp8 block scales for dequant-on-copy (fp8-in-store path)");
 }