Add CPU ClickHouse host scaling benchmarks

docs/benchmarks.md

@@ -30,11 +30,14 @@ make -C native host -j
 
 ```bash
 python -m benchmarks.bench_clickhouse_host \
-  --rows 10000 \
+  --rows 100000 \
   --payload-bytes 64KiB \
+  --parallelism-sweep 1,2,4,8 \
+  --trials 5 \
   --min-batch-bytes 16MiB \
   --max-batch-bytes 64MiB \
   --compression lz4 \
+  --socket-timeout-seconds 30 \
   --json-output host-clickhouse.json
 ```
 
@@ -43,22 +46,53 @@ connecting to ClickHouse. The benchmark creates a uniquely named table and
 drops it after collection; `--keep-table` preserves it for inspection. Set the
 password with `DMI_CLICKHOUSE_PASSWORD` to keep it out of shell history.
 
-The JSON separates enqueue time from total drain time. It also reports logical
-payload throughput, sampled submit latency, DMI process CPU and peak RSS, active
-MergeTree parts, compression size, primary-key size, and insert batching from
-`system.query_log`. Query-log metrics degrade to a warning when the account
-lacks access. The process measurements include the Python-to-C++ call used by
-this synthetic driver; they do not include a ClickHouse server running in a
-different process.
-
-Compare one setting at a time with the same row count, payload pattern, seed,
-and pool size. Useful sweeps are `--parallelism`, batch byte limits,
-`--compression`, and `--async-insert`. Synchronous batching remains the default.
+The scaling report retains every raw trial and summarizes median throughput,
+variance, speedup over the reported baseline, and gain over the preceding
+worker count. `speedup_vs_one` is populated only when the sweep includes one
+worker. Trial order is deterministically shuffled. Startup ends only after
+every native worker has connected and initialized, so steady-state throughput
+does not mix in connection setup.
+
+Each trial reports enqueue and total drain time, sampled submit latency, DMI
+process CPU, process-lifetime peak RSS, per-worker batches/rows/bytes/insert
+time, and peak simultaneous native inserts. The RSS value is not an isolated
+per-trial peak, so do not use later values in an in-process sweep for memory
+scaling. A 50 ms sampler records server-side active inserts,
+query/merge/connection gauges, normalized CPU and I/O wait, resident memory,
+part pressure, and block/network counter deltas over the same steady-state
+interval as throughput. Set
+`--server-sample-interval-ms 0` when the benchmark account cannot read those
+tables. Query-log and server metrics degrade to warnings when unavailable.
+
+Treat saturation as evidence from several signals, not the first flat number:
+throughput gain should fall below the configured plateau threshold across
+repeated trials, realized insert concurrency should reach the requested worker
+count, and client CPU/backpressure plus ClickHouse query/merge metrics should
+identify which side is limiting progress. The process measurements include the
+Python-to-C++ synthetic producer but not ClickHouse server CPU when the server
+runs in another process.
+
+Queue admission uses a finite timeout. The drain deadline is not restarted
+during abort cleanup, and native connect/send/receive calls use the configured
+socket timeout. An in-flight native call cannot be cancelled, so failure
+cleanup can extend past the drain deadline by up to the socket timeout.
+
+Use `--parallelism-sweep` for worker scaling, then compare one additional
+setting at a time with the same row count, payload pattern, seed, and pool size.
+Useful follow-up sweeps are batch byte limits, `--compression`, and
+`--async-insert`. The benchmark explicitly sets `async_insert=0` for
+synchronous trials instead of inheriting the server or user profile; the JSON
+records the effective client settings. This matters on releases such as
+[ClickHouse 26.3](https://clickhouse.com/blog/clickhouse-release-26-03), which
+changed the server default.
 ClickHouse recommends batching synchronous inserts, commonly at least 1,000
 rows and ideally 10,000–100,000 where row size permits; tensor workloads may
 reach practical byte limits earlier. See ClickHouse's
 [insert strategy](https://clickhouse.com/docs/concepts/best-practices/selecting-an-insert-strategy),
 [`system.query_log`](https://clickhouse.com/docs/reference/system-tables/query_log),
+[`system.processes`](https://clickhouse.com/docs/reference/system-tables/processes),
+[`system.metrics`](https://clickhouse.com/docs/reference/system-tables/metrics),
+[`system.asynchronous_metrics`](https://clickhouse.com/docs/reference/system-tables/asynchronous_metrics),
 and [`system.parts`](https://clickhouse.com/docs/reference/system-tables/parts)
 documentation when interpreting results.
 

docs/clickhouse-offload-pipeline.html

@@ -0,0 +1,110 @@
+<!doctype html>
+<html lang="en">
+<head>
+  <meta charset="utf-8">
+  <meta name="viewport" content="width=device-width, initial-scale=1">
+  <title>DMI host-to-ClickHouse benchmark</title>
+  <style>
+    :root {
+      color-scheme: light dark;
+      --bg: #09121d;
+      --panel: #122235;
+      --text: #eef7ff;
+      --muted: #a7bdd2;
+      --line: #35526c;
+      --host: #55d6be;
+      --db: #ffc857;
+    }
+    * { box-sizing: border-box; }
+    body {
+      margin: 0;
+      background: radial-gradient(circle at top, #173653, var(--bg) 55%);
+      color: var(--text);
+      font: 16px/1.5 system-ui, sans-serif;
+    }
+    main { width: min(1120px, 94vw); margin: auto; padding: 48px 0 64px; }
+    h1 { margin: 0 0 8px; font-size: clamp(2rem, 5vw, 3.4rem); }
+    h2 { margin: 0 0 8px; font-size: 1.05rem; }
+    p { color: var(--muted); }
+    .flow, .measure { display: grid; gap: 14px; margin-top: 28px; }
+    .flow { grid-template-columns: repeat(4, 1fr); }
+    .measure { grid-template-columns: repeat(3, 1fr); }
+    .card {
+      min-height: 180px;
+      padding: 20px;
+      border: 1px solid var(--line);
+      border-radius: 16px;
+      background: color-mix(in srgb, var(--panel) 94%, transparent);
+      box-shadow: 0 16px 40px #0004;
+    }
+    .host { border-color: var(--host); }
+    .db { border-color: var(--db); }
+    .tag {
+      display: inline-block;
+      margin: 4px 4px 0 0;
+      padding: 3px 8px;
+      border: 1px solid var(--line);
+      border-radius: 999px;
+      color: var(--muted);
+      font-size: .8rem;
+    }
+    @media (max-width: 900px) {
+      .flow, .measure { grid-template-columns: 1fr; }
+      .card { min-height: 0; }
+    }
+  </style>
+</head>
+<body>
+  <main>
+    <h1>CPU host to ClickHouse</h1>
+    <p>The benchmark isolates ClickHouse ingestion from model execution and
+      accelerator transport. Deterministic CPU tensors exercise the same native
+      host queue and ClickHouse clients used by DMI.</p>
+
+    <section class="flow" aria-label="Host benchmark stages">
+      <article class="card host">
+        <h2>1. Synthetic producer</h2>
+        <p>Reuses a seeded pool of contiguous CPU tensors and submits a fixed
+          logical payload without model or device variance.</p>
+        <span class="tag">CPU tensors</span><span class="tag">repeatable</span>
+      </article>
+      <article class="card host">
+        <h2>2. Bounded queue</h2>
+        <p>Applies byte and row watermarks, finite admission waits, and
+          configurable batch size and linger thresholds.</p>
+        <span class="tag">backpressure</span><span class="tag">16–64 MiB batches</span>
+      </article>
+      <article class="card host">
+        <h2>3. Native clients</h2>
+        <p>Each worker owns one ClickHouse connection. Readiness, per-worker
+          rows and bytes, insert time, and realized concurrency are recorded.</p>
+        <span class="tag">1 / 2 / 4 / 8 workers</span><span class="tag">socket timeouts</span>
+      </article>
+      <article class="card db">
+        <h2>4. ClickHouse</h2>
+        <p>Stores tensor metadata and binary payloads in a MergeTree table,
+          then verifies row and byte counts after the queue drains.</p>
+        <span class="tag">native protocol</span><span class="tag">verified output</span>
+      </article>
+    </section>
+
+    <section class="measure" aria-label="Scaling evidence">
+      <article class="card host">
+        <h2>Steady-state boundary</h2>
+        <p>Throughput starts only after every worker connects, selects the
+          database, and applies session settings.</p>
+      </article>
+      <article class="card host">
+        <h2>Repeated scaling sweep</h2>
+        <p>Shuffled trials retain raw results and summarize median throughput,
+          variance, speedup, and marginal gain.</p>
+      </article>
+      <article class="card db">
+        <h2>Server evidence</h2>
+        <p>Active inserts, queries, merges, connections, CPU, I/O wait, memory,
+          and network counters are sampled over the measured interval.</p>
+      </article>
+    </section>
+  </main>
+</body>
+</html>

docs/integration-api-v1.md

@@ -844,12 +844,15 @@ Construct `ClickHouseClientConfig()` and set all fields before passing it to
 | `drop_existing_database` | `False` | Drop the entire configured database before setup. Destructive; isolated tests only. |
 | `client_side_compress` | `none` | `none`, `lz4`, `zstd`, `true`, or `false`. |
 | `index_granularity` | `8192` | MergeTree index granularity for a created table. |
+| `connect_timeout_ms` | `5000` | Native socket connect timeout. |
+| `receive_timeout_ms` | `0` | Native socket receive timeout; `0` is unbounded. |
+| `send_timeout_ms` | `0` | Native socket send timeout; `0` is unbounded. |
 
 Database connection and schema initialization occur asynchronously in stage
 worker threads after host start, not in the config/factory constructor.
-Existing incompatible tables are not migrated. Current v1 also protects schema
-DDL with one process-global one-time guard: use one writer destination per
-process or pre-create any later destination yourself.
+Existing incompatible tables are not migrated. Schema DDL runs once per
+ClickHouse stage, so separate stages may target separate destinations in one
+process.
 
 ### `StageConfig`
 
@@ -874,6 +877,11 @@ The object also exposes mutable `name`, `parallelism`, and
 worker labels. Configure all fields before constructing `DMXHostEngine`; the
 engine copies the stage.
 
+The ClickHouse factory targets 16 MiB batches with a 50 ms linger, caps batches
+at 10,000 rows, and applies backpressure at 20,000 rows or 512 MiB. It leaves
+`max_batch_size` unset so a larger activation can be inserted as a singleton.
+Callers may override these values before constructing `DMXHostEngine`.
+
 ### `QueueConfig`
 
 ```python
@@ -925,6 +933,8 @@ database. Public lifecycle and diagnostics are:
 
 ```python
 start() -> None
+wait_until_ready(timeout_s: float) -> bool
+clickhouse_metrics() -> ClickHouseMetricsSnapshot
 close_input() -> None
 stop(graceful: bool = True, timeout_s: float | None = None) -> bool
 request_abort() -> None
@@ -934,9 +944,24 @@ raise_if_failed() -> None
 ```
 
 `start()` is asynchronous: it can return before a worker fails to connect or
-initialize. `stop()` returning true means threads joined, not that inserts
-succeeded. After shutdown, call `raise_if_failed()`; `failures()` returns
-records with `stage`, `thread_name`, `where`, `exc_type`, and `exc_what`.
+initialize. `wait_until_ready()` returns when every configured ClickHouse
+worker has initialized, a worker fails, or the timeout expires. On false, call
+`raise_if_failed()` to distinguish failure from timeout. `stop()` returning true
+means threads joined, not that inserts succeeded. After shutdown, call
+`raise_if_failed()`; `failures()` returns records with `stage`, `thread_name`,
+`where`, `exc_type`, and `exc_what`.
+
+`clickhouse_metrics()` returns a read-only snapshot with `expected_workers`,
+`ready_workers`, `active_inserts`, `peak_active_inserts`, `batches`, `rows`,
+`logical_bytes`, `insert_seconds`, and a `workers` list. Each worker record has
+`worker_index`, `batches`, `rows`, `logical_bytes`, and `insert_seconds`.
+Insert time covers the synchronous native ClickHouse call; it excludes queue
+wait, column construction, and worker initialization.
+
+Host-engine construction creates fresh schema-initialization and metrics state
+using the stage's final `parallelism`. Mutating a factory-created stage before
+construction therefore keeps readiness and per-worker metrics aligned; engines
+constructed from the same stage do not share accumulated metrics.
 
 ### `ThreadFailure`
 

native/csrc/bindings.cpp

@@ -7,6 +7,7 @@
 #endif
 #include <pybind11/pybind11.h>
 #include <pybind11/stl.h>
+#include <cmath>
 namespace py = pybind11;
 
 #include "clickhouse_client.h"
@@ -54,6 +55,12 @@ PYBIND11_MODULE(TORCH_EXTENSION_NAME, m) {
                      &dmx_host::ClickHouseClientConfig::client_side_compress)
       .def_readwrite("index_granularity",
                      &dmx_host::ClickHouseClientConfig::index_granularity)
+      .def_readwrite("connect_timeout_ms",
+                     &dmx_host::ClickHouseClientConfig::connect_timeout_ms)
+      .def_readwrite("receive_timeout_ms",
+                     &dmx_host::ClickHouseClientConfig::receive_timeout_ms)
+      .def_readwrite("send_timeout_ms",
+                     &dmx_host::ClickHouseClientConfig::send_timeout_ms)
 
       // Expose client_settings as a dict, store internally as unordered_map<string, variant<...>>.
       // This avoids requiring <pybind11/stl_variant.h>.
@@ -107,6 +114,24 @@ PYBIND11_MODULE(TORCH_EXTENSION_NAME, m) {
   using EnqueuePolicy = DMXHostEngine::EnqueuePolicy;
   using Duration = DMXHostEngine::Duration;
 
+  py::class_<dmx_host::ClickHouseWorkerMetrics>(m, "ClickHouseWorkerMetrics")
+      .def_readonly("worker_index", &dmx_host::ClickHouseWorkerMetrics::worker_index)
+      .def_readonly("batches", &dmx_host::ClickHouseWorkerMetrics::batches)
+      .def_readonly("rows", &dmx_host::ClickHouseWorkerMetrics::rows)
+      .def_readonly("logical_bytes", &dmx_host::ClickHouseWorkerMetrics::logical_bytes)
+      .def_readonly("insert_seconds", &dmx_host::ClickHouseWorkerMetrics::insert_seconds);
+
+  py::class_<dmx_host::ClickHouseMetricsSnapshot>(m, "ClickHouseMetricsSnapshot")
+      .def_readonly("expected_workers", &dmx_host::ClickHouseMetricsSnapshot::expected_workers)
+      .def_readonly("ready_workers", &dmx_host::ClickHouseMetricsSnapshot::ready_workers)
+      .def_readonly("active_inserts", &dmx_host::ClickHouseMetricsSnapshot::active_inserts)
+      .def_readonly("peak_active_inserts", &dmx_host::ClickHouseMetricsSnapshot::peak_active_inserts)
+      .def_readonly("batches", &dmx_host::ClickHouseMetricsSnapshot::batches)
+      .def_readonly("rows", &dmx_host::ClickHouseMetricsSnapshot::rows)
+      .def_readonly("logical_bytes", &dmx_host::ClickHouseMetricsSnapshot::logical_bytes)
+      .def_readonly("insert_seconds", &dmx_host::ClickHouseMetricsSnapshot::insert_seconds)
+      .def_readonly("workers", &dmx_host::ClickHouseMetricsSnapshot::workers);
+
   py::enum_<dmx_host::OnFullPolicy>(m, "OnFullPolicy")
       .value("RAISE", dmx_host::OnFullPolicy::RAISE)
       .value("DROP", dmx_host::OnFullPolicy::DROP)
@@ -184,11 +209,17 @@ PYBIND11_MODULE(TORCH_EXTENSION_NAME, m) {
             StageConfig cfg;
             cfg.name = std::move(name);
             cfg.parallelism = parallelism;
+            cfg.input_queue.min_batch_items.reset();
+            cfg.input_queue.min_batch_size = 16ULL * 1024 * 1024;
+            cfg.input_queue.max_linger = Duration(0.05);
+            cfg.input_queue.max_batch_items = 10000;
+            cfg.input_queue.high_watermark_items = 20000;
+            cfg.input_queue.high_watermark_size = 512ULL * 1024 * 1024;
             cfg.process_fn = [](std::vector<dmx_host::dmx_host_queue_item> batch, QueueT* next_q) {
               return dmx_host::ClickHouseInsertStage::ProcessFn<QueueT>(std::move(batch), next_q);
             };
-            // Stored by value in std::any; ClickHouseInsertStage::ThreadInitAny will any_cast it.
-            cfg.thread_init_config = ch_cfg;
+            auto thread_cfg = ch_cfg;
+            cfg.thread_init_config = std::move(thread_cfg);
             cfg.thread_init = &dmx_host::ClickHouseInsertStage::ThreadInitAny;
             cfg.thread_cleanup = &dmx_host::ClickHouseInsertStage::ThreadCleanupAny;
             return cfg;
@@ -200,9 +231,22 @@ PYBIND11_MODULE(TORCH_EXTENSION_NAME, m) {
   py::class_<DMXHostEngine, std::shared_ptr<DMXHostEngine>>(m, "DMXHostEngine")
       .def(py::init<StageConfig>(), py::arg("insert_stage"))
       .def("start", &DMXHostEngine::start)
+      .def("wait_until_ready",
+           [](DMXHostEngine& self, double timeout_s) {
+             if (!std::isfinite(timeout_s)) {
+               throw std::invalid_argument("timeout_s must be finite and non-negative");
+             }
+             return self.wait_until_ready(DMXHostEngine::Duration(timeout_s));
+           },
+           py::arg("timeout_s"),
+           py::call_guard<py::gil_scoped_release>())
+      .def("clickhouse_metrics", &DMXHostEngine::clickhouse_metrics)
       .def("stop",
            [](DMXHostEngine& self, bool graceful, std::optional<double> timeout_s) {
              if (timeout_s) {
+               if (!std::isfinite(*timeout_s)) {
+                 throw std::invalid_argument("timeout_s must be finite and non-negative");
+               }
                return self.stop(graceful, DMXHostEngine::Duration(*timeout_s));
              }
              return self.stop(graceful, std::nullopt);
@@ -214,7 +258,12 @@ PYBIND11_MODULE(TORCH_EXTENSION_NAME, m) {
       .def("request_abort", &DMXHostEngine::request_abort)
       .def("join",
            [](DMXHostEngine& self, std::optional<double> timeout_s) {
-             if (timeout_s) return self.join(DMXHostEngine::Duration(*timeout_s));
+             if (timeout_s) {
+               if (!std::isfinite(*timeout_s)) {
+                 throw std::invalid_argument("timeout_s must be finite and non-negative");
+               }
+               return self.join(DMXHostEngine::Duration(*timeout_s));
+             }
              return self.join(std::nullopt);
            },
            py::arg("timeout_s") = std::optional<double>(),