[Core] Expose ring preparation, graph toggle, gather upload, and staging wait telemetry

docs/ring-telemetry.md

@@ -0,0 +1,30 @@
+# Legacy ring telemetry
+
+`RingEngine.get_stats()` returns a `RingFlushStats` snapshot of host-only,
+cumulative counters for waits and copies the legacy ring already performs.
+Reading it does no CUDA work and adds no synchronization; each field is
+monotonic on its own.
+
+| fields | what is counted |
+|---|---|
+| `cpu_payload_head`, `cpu_payload_tail_committed`, `cpu_task_head`, `cpu_task_tail_committed` | reservation heads and drain tails |
+| `prepare_ok_calls`, `prepare_flushed_calls`, `prepare_oversized_calls` | `prepare_step` outcomes |
+| `prepare_sync_{calls,ns,errors}` | stream synchronizations inside `prepare_step` (flushed and oversized steps) |
+| `prepare_drain_{calls,ns,errors}` | forced flushes inside `prepare_step` |
+| `toggle_wait_{calls,ns,errors}` | node-toggle pre-apply waits (`set_toggle_pre_apply_sync(True)` only) |
+| `staging_wait_{calls,ns,errors}` | drain waits for pinned staging space (an uncontended check reads no clock) |
+| `ready_commit_{tasks,us,invalid}` | per task, summed: first seen published to committed after its D2H copy |
+| `d2h_copy_errors`, `d2h_sync_errors` | failed D2H copies and synchronizations |
+| `pageable_copy_{calls,ns,bytes}` | staging-to-pageable copies in the P2P thread |
+
+`dmi.sync_stats` reads them as versioned dicts:
+`synchronization_stats(engine)` (prepare outcomes and waits, toggle waits),
+`pipeline_stats(engine)` (drain and P2P stages) and
+`gather_sync_stats(source)` (an integration's per-HookPoint row-table
+uploads, reuses and pre-write waits, from `source.row_table_stats()`). A reader returns `{"available": False, ...}` rather than
+zeros when a counter is missing, and raises on a negative one.
+
+Building a legacy ring logs its device payload ring, pinned staging and task
+entries at INFO (`dmi.engine`). The payload ring is allocated when the ring
+is built, so a serving framework that sizes its KV cache from the free memory
+it profiles afterwards loses KV capacity byte for byte.

native/Makefile

@@ -31,7 +31,8 @@ HEADERS := csrc/dmx_host_utils.h \
            csrc/ring/toggle_device.h csrc/ring/toggle_registry.h \
            csrc/ring/capture_lease.h csrc/ring/drop_notice.h \
            csrc/ring/slice_queue.h csrc/ring/batch_queue.h \
-           csrc/ring/buffer_pool.h csrc/ring/pooled_block.h
+           csrc/ring/buffer_pool.h csrc/ring/pooled_block.h \
+           csrc/ring/host_wait_counter.h csrc/ring/pipeline_telemetry.h
 BUILD_DIR := build
 HOST_BUILD_DIR := $(BUILD_DIR)/host
 OBJS := $(patsubst csrc/%.cpp,$(BUILD_DIR)/%.o,$(SRCS))

native/csrc/bindings.cpp

@@ -949,6 +949,36 @@ PYBIND11_MODULE(TORCH_EXTENSION_NAME, m) {
            },
            py::arg("max_items") = 64, py::arg("timeout_ms") = 50);
 
+  // ---- RingFlushStats: host-only telemetry snapshot ----------------------
+  py::class_<ring_py::RingFlushStats>(m, "RingFlushStats")
+      .def_readonly("cpu_payload_head", &ring_py::RingFlushStats::cpu_payload_head)
+      .def_readonly("cpu_payload_tail_committed", &ring_py::RingFlushStats::cpu_payload_tail_committed)
+      .def_readonly("cpu_task_head", &ring_py::RingFlushStats::cpu_task_head)
+      .def_readonly("cpu_task_tail_committed", &ring_py::RingFlushStats::cpu_task_tail_committed)
+      .def_readonly("prepare_ok_calls", &ring_py::RingFlushStats::prepare_ok_calls)
+      .def_readonly("prepare_flushed_calls", &ring_py::RingFlushStats::prepare_flushed_calls)
+      .def_readonly("prepare_oversized_calls", &ring_py::RingFlushStats::prepare_oversized_calls)
+      .def_readonly("prepare_sync_calls", &ring_py::RingFlushStats::prepare_sync_calls)
+      .def_readonly("prepare_sync_ns", &ring_py::RingFlushStats::prepare_sync_ns)
+      .def_readonly("prepare_sync_errors", &ring_py::RingFlushStats::prepare_sync_errors)
+      .def_readonly("prepare_drain_calls", &ring_py::RingFlushStats::prepare_drain_calls)
+      .def_readonly("prepare_drain_ns", &ring_py::RingFlushStats::prepare_drain_ns)
+      .def_readonly("prepare_drain_errors", &ring_py::RingFlushStats::prepare_drain_errors)
+      .def_readonly("toggle_wait_calls", &ring_py::RingFlushStats::toggle_wait_calls)
+      .def_readonly("toggle_wait_ns", &ring_py::RingFlushStats::toggle_wait_ns)
+      .def_readonly("toggle_wait_errors", &ring_py::RingFlushStats::toggle_wait_errors)
+      .def_readonly("staging_wait_calls", &ring_py::RingFlushStats::staging_wait_calls)
+      .def_readonly("staging_wait_ns", &ring_py::RingFlushStats::staging_wait_ns)
+      .def_readonly("staging_wait_errors", &ring_py::RingFlushStats::staging_wait_errors)
+      .def_readonly("ready_commit_tasks", &ring_py::RingFlushStats::ready_commit_tasks)
+      .def_readonly("ready_commit_us", &ring_py::RingFlushStats::ready_commit_us)
+      .def_readonly("ready_commit_invalid", &ring_py::RingFlushStats::ready_commit_invalid)
+      .def_readonly("d2h_copy_errors", &ring_py::RingFlushStats::d2h_copy_errors)
+      .def_readonly("d2h_sync_errors", &ring_py::RingFlushStats::d2h_sync_errors)
+      .def_readonly("pageable_copy_calls", &ring_py::RingFlushStats::pageable_copy_calls)
+      .def_readonly("pageable_copy_ns", &ring_py::RingFlushStats::pageable_copy_ns)
+      .def_readonly("pageable_copy_bytes", &ring_py::RingFlushStats::pageable_copy_bytes);
+
   py::class_<ring_py::RingConfig>(m, "RingConfig")
       .def(py::init<>())
       .def_readwrite("task_ring_entries",         &ring_py::RingConfig::task_ring_entries)
@@ -1256,6 +1286,8 @@ PYBIND11_MODULE(TORCH_EXTENSION_NAME, m) {
       .def("request_flush",
            &ring_py::RingEnginePy::request_flush,
            py::call_guard<py::gil_scoped_release>())
+      .def("get_stats", &ring_py::RingEnginePy::get_stats,
+           py::call_guard<py::gil_scoped_release>())
       .def_property_readonly("submit_exceptions",
            &ring_py::RingEnginePy::submit_exceptions)
       .def_property_readonly("last_sink_error",

native/csrc/ring/drain_thread.cpp

@@ -416,7 +416,8 @@ void DrainThread::do_full_flush() {
         }
         {
             std::unique_lock<std::mutex> lk(staging_mu_);
-            staging_cv_.wait(lk, [&] { return staging_.free_bytes() >= flush_bytes; });
+            telemetry_.wait_for_space(staging_cv_, lk,
+                [&] { return staging_.free_bytes() >= flush_bytes; });
         }
         enqueue_d2h(flush_bytes);
         sync_stream();
@@ -465,7 +466,8 @@ void DrainThread::flush_allowed() {
         }
         {
             std::unique_lock<std::mutex> lk(staging_mu_);
-            staging_cv_.wait(lk, [&] { return staging_.free_bytes() >= flush_bytes; });
+            telemetry_.wait_for_space(staging_cv_, lk,
+                [&] { return staging_.free_bytes() >= flush_bytes; });
         }
         enqueue_d2h(flush_bytes);
         sync_stream();
@@ -597,7 +599,7 @@ void DrainThread::loop() {
             try {
                 {
                     std::unique_lock<std::mutex> lk(staging_mu_);
-                    staging_cv_.wait(lk, [&] {
+                    telemetry_.wait_for_space(staging_cv_, lk, [&] {
                         return staging_.free_bytes() >= flush_bytes;
                     });
                 }
@@ -749,8 +751,10 @@ void DrainThread::flush_state_update(uint64_t flush_count, uint64_t flush_bytes)
 }
 
 void DrainThread::sync_stream() {
-    throw_cuda_failure("cudaStreamSynchronize",
-                       cudaStreamSynchronize(stream_));
+    const cudaError_t synced = cudaStreamSynchronize(stream_);
+    if (synced != cudaSuccess)
+        telemetry_.d2h_sync_errors.fetch_add(1, std::memory_order_relaxed);
+    throw_cuda_failure("cudaStreamSynchronize", synced);
 }
 
 // ---------------------------------------------------------------------------
@@ -776,6 +780,8 @@ void DrainThread::enqueue_d2h(uint64_t flush_bytes) {
         cudaError_t err = cudaMemcpyAsync(staging_.base() + stg_cursor,
                         ring_.payload_buf + gpu_cursor,
                         chunk, cudaMemcpyDeviceToHost, stream_);
+        if (err != cudaSuccess)
+            telemetry_.d2h_copy_errors.fetch_add(1, std::memory_order_relaxed);
         throw_cuda_failure("cudaMemcpyAsync", err);
         RING_DBG("[enqueue_d2h] chunk=%d enqueued OK\n", chunk_idx);
 
@@ -793,6 +799,9 @@ void DrainThread::enqueue_d2h(uint64_t flush_bytes) {
 void DrainThread::submit_to_p2p(uint64_t flush_count, uint64_t flush_bytes) {
     uint64_t cumulative = 0;
     const uint64_t staging_batch_start = staging_.head();
+    // Called after the batch's D2H copy completed: the commit instant of
+    // every task in it.
+    const int64_t committed_us = steady_us(std::chrono::steady_clock::now());
 
     for (uint64_t i = 0; i < flush_count; ++i) {
         const uint64_t actual_bytes = scanned_[i];
@@ -803,6 +812,7 @@ void DrainThread::submit_to_p2p(uint64_t flush_count, uint64_t flush_bytes) {
         task.tensor_total_bytes = actual_bytes;
         task.alloc_bytes        = alloc;
         task.capture_ts_us      = scanned_ready_us_[i];
+        telemetry_.record_commit(task.capture_ts_us, committed_us);
 
         if (data_len > 0) {
             uint64_t staging_logical = staging_batch_start + cumulative;

native/csrc/ring/drain_thread.h

@@ -16,6 +16,7 @@
 #include "ring_config.h"
 #include "pinned_staging.h"
 #include "drain_task.h"
+#include "pipeline_telemetry.h"
 
 #include <cuda_runtime.h>
 #include <atomic>
@@ -85,6 +86,8 @@ class DrainThread {
 
     bool is_running() const { return running_.load(std::memory_order_relaxed); }
 
+    const DrainTelemetry& telemetry() const { return telemetry_; }
+
     // Cumulative count of DrainTasks pushed to the P2P queue.  Paired with
     // P2PThread::tasks_processed() for a barrier through the consumer.
     uint64_t tasks_enqueued() const {
@@ -110,6 +113,7 @@ class DrainThread {
     void rethrow_drain_failure();
 
 private:
+    DrainTelemetry  telemetry_;
     RingState&      ring_;
     PinnedStaging&  staging_;
     RingConfig      cfg_;

native/csrc/ring/host_wait_counter.h

@@ -0,0 +1,57 @@
+// ring/host_wait_counter.h -- host elapsed-time counters for waits the ring
+// already performs.  No CUDA work, no device memory, no allocation, no
+// polling: a counter only times a call that would run anyway.  Snapshot
+// fields are individually monotonic, not one atomic multi-field read while a
+// measured call is completing.
+#pragma once
+#include <atomic>
+#include <chrono>
+#include <cstdint>
+#include <type_traits>
+
+namespace ring_py {
+
+struct HostWaitSnapshot {
+    uint64_t calls{0};
+    uint64_t ns{0};
+    uint64_t errors{0};
+};
+
+class HostWaitCounter {
+public:
+    // Time f().  A non-void result is a CUDA status: nonzero counts as an
+    // error.  An exception counts as an error and propagates.
+    template <class F> auto measure(F&& f) -> decltype(f()) {
+        const auto start = std::chrono::steady_clock::now();
+        try {
+            if constexpr (std::is_void_v<decltype(f())>) {
+                f();
+                record(start, false);
+            } else {
+                auto result = f();
+                record(start, result != 0);
+                return result;
+            }
+        } catch (...) {
+            record(start, true);
+            throw;
+        }
+    }
+    HostWaitSnapshot snapshot() const {
+        return {calls_.load(std::memory_order_relaxed),
+                ns_.load(std::memory_order_relaxed),
+                errors_.load(std::memory_order_relaxed)};
+    }
+
+private:
+    void record(std::chrono::steady_clock::time_point start, bool error) {
+        const auto elapsed = std::chrono::duration_cast<std::chrono::nanoseconds>(
+            std::chrono::steady_clock::now() - start).count();
+        ns_.fetch_add(static_cast<uint64_t>(elapsed), std::memory_order_relaxed);
+        if (error) errors_.fetch_add(1, std::memory_order_relaxed);
+        calls_.fetch_add(1, std::memory_order_relaxed);
+    }
+    std::atomic<uint64_t> calls_{0}, ns_{0}, errors_{0};
+};
+
+}  // namespace ring_py

native/csrc/ring/p2p_thread.cpp

@@ -244,12 +244,8 @@ void P2PThread::process(std::vector<DrainTask>& tasks) {
         }
         uint8_t* dst = tensor.data_ptr<uint8_t>();
 
-        if (task.data_len1 > 0) {
-            std::memcpy(dst, task.data_ptr1, task.data_len1);
-        }
-        if (task.data_len2 > 0) {
-            std::memcpy(dst + task.data_len1, task.data_ptr2, task.data_len2);
-        }
+        copy_telemetry_.copy(dst, task.data_ptr1, task.data_len1,
+                             task.data_ptr2, task.data_len2);
 
         // Release staging space
         if (task.alloc_bytes > 0) {

native/csrc/ring/p2p_thread.h

@@ -9,6 +9,7 @@
 #include "drain_thread.h"
 #include "drain_task.h"
 #include "ring_config.h"
+#include "pipeline_telemetry.h"
 #include "record_consumer.h"
 #include "tensor_meta.h"
 
@@ -119,7 +120,10 @@ class P2PThread {
     bool wait_until_processed_until(uint64_t target,
                                     std::chrono::steady_clock::time_point deadline);
 
+    const PageableCopyTelemetry& copy_telemetry() const { return copy_telemetry_; }
+
 private:
+    PageableCopyTelemetry    copy_telemetry_;
     DrainThread&             drain_;
     ring_py::TensorMetaFifo& fifo_;
     RingConfig               cfg_;

native/csrc/ring/pipeline_telemetry.h

@@ -0,0 +1,58 @@
+// ring/pipeline_telemetry.h -- host-only accounting of the drain and P2P
+// stages.  No device reads and no added synchronization.
+#pragma once
+#include "host_wait_counter.h"
+
+#include <atomic>
+#include <condition_variable>
+#include <cstdint>
+#include <cstring>
+#include <mutex>
+
+namespace ring {
+
+class DrainTelemetry {
+public:
+    // Wait for staging space under the caller's lock.  An uncontended check
+    // reads no clock.
+    template <class Predicate>
+    void wait_for_space(std::condition_variable& cv,
+                        std::unique_lock<std::mutex>& lock, Predicate ready) {
+        if (!ready()) staging_wait.measure([&] { cv.wait(lock, ready); });
+    }
+    // One task's latency from first seen published to committed to the P2P
+    // queue (after its D2H copy completed).
+    void record_commit(int64_t ready_us, int64_t committed_us) {
+        if (ready_us <= 0 || committed_us < ready_us) {
+            ready_commit_invalid.fetch_add(1, std::memory_order_relaxed);
+            return;
+        }
+        ready_commit_us.fetch_add(static_cast<uint64_t>(committed_us - ready_us),
+                                  std::memory_order_relaxed);
+        ready_commit_tasks.fetch_add(1, std::memory_order_relaxed);
+    }
+    ring_py::HostWaitCounter staging_wait;
+    std::atomic<uint64_t> ready_commit_tasks{0}, ready_commit_us{0},
+                          ready_commit_invalid{0};
+    std::atomic<uint64_t> d2h_copy_errors{0}, d2h_sync_errors{0};
+};
+
+class PageableCopyTelemetry {
+public:
+    // Staging -> pageable copy of one task (one or two spans when the staging
+    // ring wraps; counted as one operation).  Destination allocation and the
+    // staging release are outside the measurement.
+    void copy(uint8_t* dst, const uint8_t* a, uint64_t n,
+              const uint8_t* b, uint64_t m) {
+        if (n == 0 && m == 0) return;
+        elapsed.measure([&] {
+            if (n) std::memcpy(dst, a, n);
+            if (m) std::memcpy(dst + n, b, m);
+        });
+        bytes.fetch_add(n + m, std::memory_order_relaxed);
+    }
+    ring_py::HostWaitCounter elapsed;
+    std::atomic<uint64_t> bytes{0};
+};
+
+}  // namespace ring