diff --git a/benchmarks/bench_clickhouse_host.py b/benchmarks/bench_clickhouse_host.py index 01aae389e..d091a8d52 100644 --- a/benchmarks/bench_clickhouse_host.py +++ b/benchmarks/bench_clickhouse_host.py @@ -6,12 +6,15 @@ import json import math import os +import random import re import resource +import statistics import sys +import threading import time import uuid -from dataclasses import asdict, dataclass, field +from dataclasses import asdict, dataclass, field, replace from datetime import datetime from typing import Any, Callable, Sequence @@ -74,6 +77,8 @@ class BenchmarkConfig: compression: str = "lz4" async_insert: bool = False drain_timeout_seconds: float = 300.0 + socket_timeout_seconds: float = 30.0 + server_sample_interval_ms: int = 50 host: str = "localhost" port: int = 9000 user: str = "default" @@ -109,6 +114,10 @@ def __post_init__(self) -> None: raise ValueError("max_linger_ms must be a finite non-negative number") if not math.isfinite(self.drain_timeout_seconds) or self.drain_timeout_seconds <= 0: raise ValueError("drain_timeout_seconds must be a finite positive number") + if not math.isfinite(self.socket_timeout_seconds) or self.socket_timeout_seconds <= 0: + raise ValueError("socket_timeout_seconds must be a finite positive number") + if self.server_sample_interval_ms < 0: + raise ValueError("server_sample_interval_ms must be non-negative") if not 1 <= self.port <= 65535: raise ValueError("port must be between 1 and 65535") if self.dtype not in _DTYPE_BYTES: @@ -155,8 +164,9 @@ class TrialMeasurement: total_seconds: float enqueue_latencies_ns: tuple[int, ...] process_cpu_seconds: float = 0.0 - peak_rss_before_bytes: int = 0 - peak_rss_after_bytes: int = 0 + process_lifetime_peak_rss_bytes: int = 0 + startup_seconds: float = 0.0 + client_metrics: dict[str, Any] | None = None def as_dict(self) -> dict[str, Any]: logical_bytes = self.rows * self.payload_bytes @@ -177,26 +187,181 @@ def rates(seconds: float) -> dict[str, float]: return { "rows": self.rows, "logical_payload_bytes": logical_bytes, + "startup_seconds": self.startup_seconds, "enqueue": enqueue, "drain_seconds": max(0.0, self.total_seconds - self.enqueue_seconds), "total": rates(self.total_seconds), "host_process": { "cpu_seconds": self.process_cpu_seconds, "effective_cpu_cores": self.process_cpu_seconds / self.total_seconds, - "peak_rss_bytes": self.peak_rss_after_bytes, - "peak_rss_growth_bytes": max( - 0, self.peak_rss_after_bytes - self.peak_rss_before_bytes - ), + "process_lifetime_peak_rss_bytes": self.process_lifetime_peak_rss_bytes, }, + "client": self.client_metrics, } def build_client_settings(async_insert: bool) -> dict[str, int]: if not async_insert: - return {} + return {"async_insert": 0} return {"async_insert": 1, "wait_for_async_insert": 1} +def parse_parallelism_sweep(value: str) -> tuple[int, ...]: + try: + parsed = [int(part.strip()) for part in value.split(",") if part.strip()] + except ValueError as exc: + raise ValueError("parallelism values must be integers") from exc + if not parsed or any(value <= 0 for value in parsed): + raise ValueError("parallelism values must be positive") + return tuple(dict.fromkeys(parsed)) + + +def _client_metrics_as_dict(metrics: Any) -> dict[str, Any]: + workers = [ + { + "worker_index": int(worker.worker_index), + "batches": int(worker.batches), + "rows": int(worker.rows), + "logical_bytes": int(worker.logical_bytes), + "insert_seconds": float(worker.insert_seconds), + } + for worker in metrics.workers + ] + return { + "expected_workers": int(metrics.expected_workers), + "ready_workers": int(metrics.ready_workers), + "active_inserts": int(metrics.active_inserts), + "peak_active_inserts": int(metrics.peak_active_inserts), + "batches": int(metrics.batches), + "rows": int(metrics.rows), + "logical_bytes": int(metrics.logical_bytes), + "insert_seconds": float(metrics.insert_seconds), + "workers": workers, + } + + +class ServerTelemetrySampler: + _METRICS_QUERY = """ + SELECT metric, value + FROM system.metrics + WHERE metric IN ('Query', 'Merge', 'BackgroundMergesAndMutationsPoolTask', + 'TCPConnection') + """ + _ASYNC_METRICS_QUERY = """ + SELECT metric, + if(isFinite(value), value, arraySum(mapValues(key_values))) AS scalar_value + FROM system.asynchronous_metrics + WHERE metric IN ('OSUserTimeNormalized', 'OSSystemTimeNormalized', + 'OSIOWaitTimeNormalized', 'MemoryResident', 'LoadAverage1', + 'MaxPartCountForPartition', 'LongestRunningMerge', + 'BlockReadBytes', 'BlockWriteBytes', + 'NetworkReceiveBytes', 'NetworkSendBytes') + """ + + def __init__( + self, + client_factory: Callable[[], Any], + interval_ms: int = 50, + database: str | None = None, + table: str | None = None, + ): + if interval_ms <= 0: + raise ValueError("interval_ms must be positive") + self._client_factory = client_factory + self._interval_seconds = interval_ms / 1000.0 + if database is not None and table is not None: + qualified = f"{database}.{table}" + self._process_query = ( + "SELECT count() FROM system.processes" + " WHERE query_kind = 'Insert'" + " AND has(tables, %(table)s)" + ) + self._process_query_params: dict[str, Any] = {"table": qualified} + else: + self._process_query = ( + "SELECT count() FROM system.processes WHERE query_kind = 'Insert'" + ) + self._process_query_params = {} + self._stop = threading.Event() + self._thread: threading.Thread | None = None + self._client: Any = None + self._lock = threading.Lock() + self._samples = 0 + self._peak_active_inserts = 0 + self._metric_values: dict[str, list[float]] = {} + self._errors: list[str] = [] + + def start(self) -> None: + if self._thread is not None: + return + self.sample_once() + if self._stop.is_set(): + return + self._thread = threading.Thread( + target=self._run, + name="dmi-clickhouse-telemetry", + daemon=True, + ) + self._thread.start() + + def stop(self) -> None: + self._stop.set() + if self._thread is not None: + self._thread.join() + if not self._errors: + self.sample_once() + if self._client is not None: + try: + self._client.disconnect() + except Exception: + pass + + def _run(self) -> None: + while not self._stop.is_set(): + self.sample_once() + self._stop.wait(self._interval_seconds) + + def sample_once(self) -> None: + try: + if self._client is None: + self._client = self._client_factory() + active = int(self._client.execute(self._process_query, self._process_query_params)[0][0]) + metrics = self._client.execute(self._METRICS_QUERY) + async_metrics = self._client.execute(self._ASYNC_METRICS_QUERY) + with self._lock: + self._samples += 1 + self._peak_active_inserts = max(self._peak_active_inserts, active) + for name, value in metrics: + self._metric_values.setdefault(str(name), []).append(float(value)) + for name, value in async_metrics: + key = f"async.{name}" + self._metric_values.setdefault(key, []).append(float(value)) + except Exception as exc: + with self._lock: + if not self._errors: + self._errors.append(str(exc)) + self._stop.set() + + def snapshot(self) -> dict[str, Any]: + with self._lock: + return { + "samples": self._samples, + "peak_active_inserts": self._peak_active_inserts, + "metric_peaks": { + name: max(values) for name, values in self._metric_values.items() + }, + "metric_means": { + name: statistics.fmean(values) + for name, values in self._metric_values.items() + }, + "metric_deltas": { + name: values[-1] - values[0] + for name, values in self._metric_values.items() + }, + "errors": list(self._errors), + } + + def generate_payload_pool(config: BenchmarkConfig) -> list[Any]: import torch @@ -229,6 +394,10 @@ def configure_stage(stage: Any, config: BenchmarkConfig) -> None: queue.high_watermark_items = config.queue_capacity_items queue.high_watermark_size = config.queue_capacity_bytes stage.ingress_policy.block = True + stage.ingress_policy.timeout_s = min( + config.socket_timeout_seconds, + config.drain_timeout_seconds, + ) def _peak_rss_bytes() -> int: @@ -256,6 +425,10 @@ def _build_engine(config: BenchmarkConfig) -> Any: clickhouse.client_side_compress = config.compression clickhouse.client_settings = build_client_settings(config.async_insert) clickhouse.index_granularity = config.index_granularity + timeout_ms = round(config.socket_timeout_seconds * 1000) + clickhouse.connect_timeout_ms = min(5000, timeout_ms) + clickhouse.send_timeout_ms = timeout_ms + clickhouse.receive_timeout_ms = timeout_ms stage = StageConfig.clickhouse_insert( clickhouse, @@ -274,15 +447,29 @@ def _submit_and_drain( timeout_seconds: float, max_latency_samples: int = 10_000, clock: Callable[[], int] = time.perf_counter_ns, + deadline_clock: Callable[[], float] = time.monotonic, + sampler: ServerTelemetrySampler | None = None, ) -> TrialMeasurement: - engine.start() - cpu_start = time.process_time() - peak_rss_before = _peak_rss_bytes() - total_start = clock() - latencies = [] - sample_stride = max(1, math.ceil(rows / max_latency_samples)) + startup_start = clock() + sampler_started = False + deadline = deadline_clock() + timeout_seconds try: + engine.start() + if not engine.wait_until_ready(timeout_seconds): + engine.raise_if_failed() + raise TimeoutError(f"ClickHouse workers were not ready after {timeout_seconds:.1f}s") + startup_end = clock() + deadline = deadline_clock() + timeout_seconds + cpu_start = time.process_time() + if sampler is not None: + sampler.start() + sampler_started = True + total_start = clock() + latencies = [] + sample_stride = max(1, math.ceil(rows / max_latency_samples)) for row in range(rows): + if deadline_clock() >= deadline: + raise TimeoutError(f"ClickHouse benchmark exceeded {timeout_seconds:.1f}s") started = clock() if row % sample_stride == 0 else None engine.submit_direct( model_id, @@ -298,16 +485,28 @@ def _submit_and_drain( latencies.append(clock() - started) enqueue_end = clock() engine.close_input() - if not engine.join(timeout_seconds): + remaining = max(0.0, deadline - deadline_clock()) + if not engine.join(remaining): raise TimeoutError(f"ClickHouse drain exceeded {timeout_seconds:.1f}s") engine.raise_if_failed() total_end = clock() + if sampler is not None: + sampler.stop() + sampler_started = False cpu_end = time.process_time() - peak_rss_after = _peak_rss_bytes() + process_lifetime_peak_rss = _peak_rss_bytes() + client_metrics = _client_metrics_as_dict(engine.clickhouse_metrics()) except BaseException: engine.request_abort() - engine.join(timeout_seconds) + remaining = max(0.0, deadline - deadline_clock()) + try: + engine.join(remaining) + except BaseException: + pass raise + finally: + if sampler_started: + sampler.stop() return TrialMeasurement( rows=rows, @@ -316,23 +515,27 @@ def _submit_and_drain( total_seconds=(total_end - total_start) / 1_000_000_000, enqueue_latencies_ns=tuple(latencies), process_cpu_seconds=cpu_end - cpu_start, - peak_rss_before_bytes=peak_rss_before, - peak_rss_after_bytes=peak_rss_after, + process_lifetime_peak_rss_bytes=process_lifetime_peak_rss, + startup_seconds=(startup_end - startup_start) / 1_000_000_000, + client_metrics=client_metrics, ) -def _connect(config: BenchmarkConfig) -> Any: +def _connect(config: BenchmarkConfig, timeout_seconds: float | None = None) -> Any: try: from clickhouse_driver import Client except ImportError as exc: raise RuntimeError("clickhouse-driver is required for benchmark verification") from exc + socket_timeout = timeout_seconds or config.socket_timeout_seconds connection = { "host": config.host, "port": config.port, "user": config.user, "password": config.password, "secure": config.secure, + "connect_timeout": min(5.0, socket_timeout), + "send_receive_timeout": socket_timeout, } if config.create_database: bootstrap = Client(database="default", **connection) @@ -441,13 +644,7 @@ def _safe_config(config: BenchmarkConfig) -> dict[str, Any]: def _ensure_table(client: Any, config: BenchmarkConfig) -> None: - """Explicitly create the capture table if it does not yet exist. - - The native backend guards its own DDL with a process-wide once_flag, so a - second ``run()`` call in the same process will not recreate a table that - was dropped by a previous run. Pre-creating the table here means every - ``run()`` starts with a live table regardless of the once_flag state. - """ + """Create the trial table before native workers enter the timed path.""" table = f"{quote_identifier(config.database)}.{quote_identifier(config.table)}" cols = ", ".join([ f"{quote_identifier('model_id')} String", @@ -497,6 +694,14 @@ def run(config: BenchmarkConfig) -> dict[str, Any]: client.execute(f"TRUNCATE TABLE {table}") started_at = _server_time(client) + sampler = None + if config.server_sample_interval_ms: + sampler = ServerTelemetrySampler( + lambda: _connect(config, min(2.0, config.socket_timeout_seconds)), + interval_ms=config.server_sample_interval_ms, + database=config.database, + table=config.table, + ) measurement = _submit_and_drain( _build_engine(config), payloads, @@ -504,7 +709,11 @@ def run(config: BenchmarkConfig) -> dict[str, Any]: measured_model_id, config.drain_timeout_seconds, config.latency_samples, + sampler=sampler, ) + server_telemetry = sampler.snapshot() if sampler is not None else None + if server_telemetry and server_telemetry["errors"]: + warnings.append(f"server telemetry unavailable: {server_telemetry['errors'][0]}") verification = _verification(client, config, measured_model_id) parts = _parts_metrics(client, config) query_log, warning = _query_log_metrics(client, config, started_at) @@ -514,10 +723,12 @@ def run(config: BenchmarkConfig) -> dict[str, Any]: "benchmark": "dmi_clickhouse_host", "server_version": server_version, "config": _safe_config(config), + "effective_client_settings": build_client_settings(config.async_insert), "measurement": measurement.as_dict(), "verification": verification, "parts": parts, "query_log": query_log, + "server_telemetry": server_telemetry, "warnings": warnings, } finally: @@ -529,6 +740,88 @@ def run(config: BenchmarkConfig) -> dict[str, Any]: client.disconnect() +def summarize_scaling_trials( + trials: Sequence[dict[str, Any]], + plateau_threshold_percent: float = 5.0, +) -> list[dict[str, Any]]: + grouped: dict[int, list[float]] = {} + for trial in trials: + grouped.setdefault(int(trial["parallelism"]), []).append(float(trial["throughput"])) + if not grouped: + return [] + + parallelisms = sorted(grouped) + medians = {value: statistics.median(grouped[value]) for value in parallelisms} + baseline_parallelism = 1 if 1 in medians else parallelisms[0] + baseline = medians[baseline_parallelism] + summary = [] + previous = None + for parallelism in parallelisms: + values = grouped[parallelism] + mean = statistics.fmean(values) + stdev = statistics.stdev(values) if len(values) > 1 else 0.0 + median = medians[parallelism] + gain = None if previous is None else (median / previous - 1.0) * 100.0 + summary.append({ + "parallelism": parallelism, + "trials": len(values), + "median_gib_per_second": median, + "mean_gib_per_second": mean, + "stdev_gib_per_second": stdev, + "coefficient_of_variation_percent": stdev / mean * 100.0 if mean else None, + "min_gib_per_second": min(values), + "max_gib_per_second": max(values), + "baseline_parallelism": baseline_parallelism, + "speedup_vs_baseline": median / baseline if baseline else None, + "speedup_vs_one": median / baseline if baseline and baseline_parallelism == 1 else None, + "gain_vs_previous_percent": gain, + "plateau": None if gain is None else gain < plateau_threshold_percent, + }) + previous = median + return summary + + +def run_sweep( + config: BenchmarkConfig, + parallelisms: Sequence[int], + trials: int, + *, + plateau_threshold_percent: float = 5.0, + runner: Callable[[BenchmarkConfig], dict[str, Any]] = run, +) -> dict[str, Any]: + if trials <= 0: + raise ValueError("trials must be positive") + values = tuple(dict.fromkeys(int(value) for value in parallelisms)) + if not values or any(value <= 0 for value in values): + raise ValueError("parallelism values must be positive") + + schedule = [(parallelism, trial) for trial in range(trials) for parallelism in values] + random.Random(config.seed).shuffle(schedule) + raw = [] + for run_index, (parallelism, trial_index) in enumerate(schedule): + table = f"{config.table}_p{parallelism}_t{trial_index}_{uuid.uuid4().hex[:6]}" + trial_config = replace(config, parallelism=parallelism, table=table) + report = runner(trial_config) + throughput = float(report["measurement"]["total"]["logical_gib_per_second"]) + raw.append({ + "run_index": run_index, + "trial_index": trial_index, + "parallelism": parallelism, + "throughput": throughput, + "report": report, + }) + + return { + "benchmark": "dmi_clickhouse_host_scaling", + "base_config": _safe_config(config), + "parallelisms": list(values), + "trials_per_parallelism": trials, + "plateau_threshold_percent": plateau_threshold_percent, + "trials": raw, + "summary": summarize_scaling_trials(raw, plateau_threshold_percent), + } + + def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--rows", type=int, default=10_000) @@ -540,6 +833,9 @@ def build_parser() -> argparse.ArgumentParser: parser.add_argument("--latency-samples", type=int, default=10_000) parser.add_argument("--warmup-rows", type=int, default=512) parser.add_argument("--parallelism", type=int, default=4) + parser.add_argument("--parallelism-sweep") + parser.add_argument("--trials", type=int, default=1) + parser.add_argument("--plateau-threshold-percent", type=float, default=5.0) parser.add_argument("--min-batch-bytes", type=parse_byte_size, default=16 * 1024**2) parser.add_argument("--max-batch-bytes", type=parse_byte_size, default=64 * 1024**2) parser.add_argument("--max-batch-items", type=int, default=10_000) @@ -550,6 +846,8 @@ def build_parser() -> argparse.ArgumentParser: parser.add_argument("--compression", choices=("none", "lz4", "zstd"), default="lz4") parser.add_argument("--async-insert", action="store_true") parser.add_argument("--drain-timeout-seconds", type=float, default=300.0) + parser.add_argument("--socket-timeout-seconds", type=float, default=30.0) + parser.add_argument("--server-sample-interval-ms", type=int, default=50) parser.add_argument("--host", default="localhost") parser.add_argument("--port", type=int, default=9000) parser.add_argument("--user", default="default") @@ -587,6 +885,8 @@ def _config_from_args(args: argparse.Namespace) -> BenchmarkConfig: compression=args.compression, async_insert=args.async_insert, drain_timeout_seconds=args.drain_timeout_seconds, + socket_timeout_seconds=args.socket_timeout_seconds, + server_sample_interval_ms=args.server_sample_interval_ms, host=args.host, port=args.port, user=args.user, @@ -602,14 +902,32 @@ def _config_from_args(args: argparse.Namespace) -> BenchmarkConfig: def main(argv: Sequence[str] | None = None) -> int: args = build_parser().parse_args(argv) config = _config_from_args(args) + parallelisms = ( + parse_parallelism_sweep(args.parallelism_sweep) + if args.parallelism_sweep + else (config.parallelism,) + ) + if len(parallelisms) == 1: + config = replace(config, parallelism=parallelisms[0]) + if args.trials <= 0: + raise ValueError("trials must be positive") if args.dry_run: report: dict[str, Any] = { "benchmark": "dmi_clickhouse_host", "dry_run": True, "config": _safe_config(config), + "parallelisms": list(parallelisms), + "trials_per_parallelism": args.trials, "logical_payload_bytes": config.rows * config.payload_bytes, "payload_pool_bytes": config.pool_size * config.payload_bytes, } + elif len(parallelisms) > 1 or args.trials > 1: + report = run_sweep( + config, + parallelisms, + args.trials, + plateau_threshold_percent=args.plateau_threshold_percent, + ) else: report = run(config) output = json.dumps(report, indent=2, sort_keys=True, default=str) diff --git a/docs/benchmarks.md b/docs/benchmarks.md index 951c19c74..801b3fd4d 100644 --- a/docs/benchmarks.md +++ b/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. diff --git a/docs/clickhouse-offload-pipeline.html b/docs/clickhouse-offload-pipeline.html new file mode 100644 index 000000000..2152136e2 --- /dev/null +++ b/docs/clickhouse-offload-pipeline.html @@ -0,0 +1,110 @@ + + + + + + DMI host-to-ClickHouse benchmark + + + +
+

CPU host to ClickHouse

+

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.

+ +
+
+

1. Synthetic producer

+

Reuses a seeded pool of contiguous CPU tensors and submits a fixed + logical payload without model or device variance.

+ CPU tensorsrepeatable +
+
+

2. Bounded queue

+

Applies byte and row watermarks, finite admission waits, and + configurable batch size and linger thresholds.

+ backpressure16–64 MiB batches +
+
+

3. Native clients

+

Each worker owns one ClickHouse connection. Readiness, per-worker + rows and bytes, insert time, and realized concurrency are recorded.

+ 1 / 2 / 4 / 8 workerssocket timeouts +
+
+

4. ClickHouse

+

Stores tensor metadata and binary payloads in a MergeTree table, + then verifies row and byte counts after the queue drains.

+ native protocolverified output +
+
+ +
+
+

Steady-state boundary

+

Throughput starts only after every worker connects, selects the + database, and applies session settings.

+
+
+

Repeated scaling sweep

+

Shuffled trials retain raw results and summarize median throughput, + variance, speedup, and marginal gain.

+
+
+

Server evidence

+

Active inserts, queries, merges, connections, CPU, I/O wait, memory, + and network counters are sampled over the measured interval.

+
+
+
+ + diff --git a/docs/integration-api-v1.md b/docs/integration-api-v1.md index 36cc7c985..deb25fd33 100644 --- a/docs/integration-api-v1.md +++ b/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` diff --git a/native/csrc/bindings.cpp b/native/csrc/bindings.cpp index e4ac9970e..cf2b0524d 100644 --- a/native/csrc/bindings.cpp +++ b/native/csrc/bindings.cpp @@ -7,6 +7,7 @@ #endif #include #include +#include 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>. // This avoids requiring . @@ -107,6 +114,24 @@ PYBIND11_MODULE(TORCH_EXTENSION_NAME, m) { using EnqueuePolicy = DMXHostEngine::EnqueuePolicy; using Duration = DMXHostEngine::Duration; + py::class_(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_(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_(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 batch, QueueT* next_q) { return dmx_host::ClickHouseInsertStage::ProcessFn(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_>(m, "DMXHostEngine") .def(py::init(), 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()) + .def("clickhouse_metrics", &DMXHostEngine::clickhouse_metrics) .def("stop", [](DMXHostEngine& self, bool graceful, std::optional 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 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(), diff --git a/native/csrc/clickhouse_client.cpp b/native/csrc/clickhouse_client.cpp index 454aa2901..56f4815c1 100644 --- a/native/csrc/clickhouse_client.cpp +++ b/native/csrc/clickhouse_client.cpp @@ -7,6 +7,7 @@ #include #include #include +#include #include #include #include @@ -19,6 +20,75 @@ #include namespace dmx_host { + +ClickHouseRuntimeMetrics::ClickHouseRuntimeMetrics(int expected_workers) + : expected_workers_(expected_workers), + ready_(static_cast(std::max(0, expected_workers)), false), + workers_(static_cast(std::max(0, expected_workers))) { + if (expected_workers < 0) { + throw std::invalid_argument("expected_workers must be non-negative"); + } + for (int i = 0; i < expected_workers; ++i) workers_[i].worker_index = i; +} + +void ClickHouseRuntimeMetrics::WorkerReady(int worker_index) { + std::lock_guard lock(mu_); + if (worker_index < 0 || worker_index >= expected_workers_) { + throw std::out_of_range("ClickHouse worker index is out of range"); + } + if (!ready_[worker_index]) { + ready_[worker_index] = true; + ++ready_workers_; + ready_cv_.notify_all(); + } +} + +bool ClickHouseRuntimeMetrics::WaitUntilReady(std::chrono::milliseconds timeout) { + std::unique_lock lock(mu_); + return ready_cv_.wait_for(lock, timeout, [&] { + return ready_workers_ == expected_workers_; + }); +} + +void ClickHouseRuntimeMetrics::BeginInsert() { + std::lock_guard lock(mu_); + ++active_inserts_; + peak_active_inserts_ = std::max(peak_active_inserts_, active_inserts_); +} + +void ClickHouseRuntimeMetrics::EndInsert(int worker_index, std::uint64_t rows, + std::uint64_t logical_bytes, + double seconds) { + std::lock_guard lock(mu_); + if (active_inserts_ > 0) --active_inserts_; + ++batches_; + rows_ += rows; + logical_bytes_ += logical_bytes; + insert_seconds_ += seconds; + if (worker_index >= 0 && worker_index < expected_workers_) { + auto& worker = workers_[worker_index]; + ++worker.batches; + worker.rows += rows; + worker.logical_bytes += logical_bytes; + worker.insert_seconds += seconds; + } +} + +ClickHouseMetricsSnapshot ClickHouseRuntimeMetrics::Snapshot() const { + std::lock_guard lock(mu_); + ClickHouseMetricsSnapshot snapshot; + snapshot.expected_workers = expected_workers_; + snapshot.ready_workers = ready_workers_; + snapshot.active_inserts = active_inserts_; + snapshot.peak_active_inserts = peak_active_inserts_; + snapshot.batches = batches_; + snapshot.rows = rows_; + snapshot.logical_bytes = logical_bytes_; + snapshot.insert_seconds = insert_seconds_; + snapshot.workers = workers_; + return snapshot; +} + namespace { thread_local std::unique_ptr tl_client; @@ -26,8 +96,8 @@ thread_local bool tl_inited = false; thread_local bool tl_cleaned = false; thread_local std::string tl_db; thread_local std::string tl_table; - -std::once_flag g_schema_once; +thread_local int tl_worker_index = -1; +thread_local std::shared_ptr tl_runtime_metrics; // --------------------- SQL helpers --------------------- @@ -377,7 +447,7 @@ StagedRow StageOneRow(ClickHouseRow&& row) { // ===================== Stage API ===================== -void ClickHouseInsertStage::ThreadInit(int /*thread_idx*/, const ClickHouseClientConfig& cfg) { +void ClickHouseInsertStage::ThreadInit(int thread_idx, const ClickHouseClientConfig& cfg) { if (tl_inited) { throw std::runtime_error("ClickHouseInsertStage::ThreadInit() called more than once in the same thread"); } @@ -393,6 +463,13 @@ void ClickHouseInsertStage::ThreadInit(int /*thread_idx*/, const ClickHouseClien opts.SetPort(static_cast(cfg.port)); opts.SetUser(cfg.username); opts.SetPassword(cfg.password); + if (cfg.connect_timeout_ms < 0 || cfg.receive_timeout_ms < 0 || + cfg.send_timeout_ms < 0) { + throw std::invalid_argument("ClickHouse socket timeouts must be non-negative"); + } + opts.SetConnectionConnectTimeout(std::chrono::milliseconds(cfg.connect_timeout_ms)); + opts.SetConnectionRecvTimeout(std::chrono::milliseconds(cfg.receive_timeout_ms)); + opts.SetConnectionSendTimeout(std::chrono::milliseconds(cfg.send_timeout_ms)); // stable default database for handshake opts.SetDefaultDatabase("default"); @@ -416,8 +493,8 @@ void ClickHouseInsertStage::ThreadInit(int /*thread_idx*/, const ClickHouseClien } try { - // DDL init once globally (copy cfg/opts into the call_once closure) - std::call_once(g_schema_once, [cfg, opts]() { RunSchemaInitOnce(cfg, opts); }); + std::call_once(*cfg.schema_once, + [cfg, opts]() { RunSchemaInitOnce(cfg, opts); }); // Per-thread client tl_client = std::make_unique(opts); @@ -435,19 +512,28 @@ void ClickHouseInsertStage::ThreadInit(int /*thread_idx*/, const ClickHouseClien throw; } + tl_worker_index = thread_idx; + tl_runtime_metrics = cfg.runtime_metrics; tl_inited = true; + tl_runtime_metrics->WorkerReady(thread_idx); } void ClickHouseInsertStage::ThreadCleanup() noexcept { if (!tl_inited || tl_cleaned) return; tl_cleaned = true; try { tl_client.reset(); } catch (...) {} + tl_runtime_metrics.reset(); + tl_worker_index = -1; } void ClickHouseInsertStage::InsertBatch(std::vector&& batch) { clickhouse::Client& client = ClientOrThrow(); if (batch.empty()) return; + std::uint64_t logical_bytes = 0; + for (const auto& item : batch) logical_bytes += item.item_size; + const std::uint64_t row_count = batch.size(); + // Stage rows (keeps tensor memory alive for AppendNoManagedLifetime) std::vector rows; rows.reserve(batch.size()); @@ -511,7 +597,19 @@ void ClickHouseInsertStage::InsertBatch(std::vector&& batch block.AppendColumn("shape", col_shape); block.AppendColumn("bytes", col_bytes); - client.Insert(fq_table, block); + const auto started = std::chrono::steady_clock::now(); + tl_runtime_metrics->BeginInsert(); + try { + client.Insert(fq_table, block); + } catch (...) { + const double seconds = std::chrono::duration( + std::chrono::steady_clock::now() - started).count(); + tl_runtime_metrics->EndInsert(tl_worker_index, row_count, logical_bytes, seconds); + throw; + } + const double seconds = std::chrono::duration( + std::chrono::steady_clock::now() - started).count(); + tl_runtime_metrics->EndInsert(tl_worker_index, row_count, logical_bytes, seconds); } // Force emission of the specialization referenced from bindings.cpp: diff --git a/native/csrc/clickhouse_client.h b/native/csrc/clickhouse_client.h index bf0e2ba0d..6093d4ef7 100644 --- a/native/csrc/clickhouse_client.h +++ b/native/csrc/clickhouse_client.h @@ -2,7 +2,11 @@ #define DMX_HOST_CLICKHOUSE_CLIENT_H_ #include +#include +#include #include +#include +#include #include #include #include @@ -19,6 +23,52 @@ namespace dmx_host { // Settings value types for session SET ... (bound from Python too). using ClickHouseSettingValue = std::variant; +struct ClickHouseWorkerMetrics { + int worker_index = 0; + std::uint64_t batches = 0; + std::uint64_t rows = 0; + std::uint64_t logical_bytes = 0; + double insert_seconds = 0.0; +}; + +struct ClickHouseMetricsSnapshot { + int expected_workers = 0; + int ready_workers = 0; + std::uint64_t active_inserts = 0; + std::uint64_t peak_active_inserts = 0; + std::uint64_t batches = 0; + std::uint64_t rows = 0; + std::uint64_t logical_bytes = 0; + double insert_seconds = 0.0; + std::vector workers; +}; + +class ClickHouseRuntimeMetrics { + public: + explicit ClickHouseRuntimeMetrics(int expected_workers = 0); + + void WorkerReady(int worker_index); + bool WaitUntilReady(std::chrono::milliseconds timeout); + void BeginInsert(); + void EndInsert(int worker_index, std::uint64_t rows, + std::uint64_t logical_bytes, double seconds); + ClickHouseMetricsSnapshot Snapshot() const; + + private: + mutable std::mutex mu_; + std::condition_variable ready_cv_; + int expected_workers_ = 0; + int ready_workers_ = 0; + std::uint64_t active_inserts_ = 0; + std::uint64_t peak_active_inserts_ = 0; + std::uint64_t batches_ = 0; + std::uint64_t rows_ = 0; + std::uint64_t logical_bytes_ = 0; + double insert_seconds_ = 0.0; + std::vector ready_; + std::vector workers_; +}; + /** * ClickHouse connection + schema init configuration. */ @@ -43,6 +93,14 @@ struct ClickHouseClientConfig { std::string client_side_compress = "none"; int index_granularity = 8192; + int connect_timeout_ms = 5000; + int receive_timeout_ms = 0; + int send_timeout_ms = 0; + + std::shared_ptr schema_once = + std::make_shared(); + std::shared_ptr runtime_metrics = + std::make_shared(); }; /** diff --git a/native/csrc/dmx_host_engine.h b/native/csrc/dmx_host_engine.h index f424b9114..11f63da44 100644 --- a/native/csrc/dmx_host_engine.h +++ b/native/csrc/dmx_host_engine.h @@ -1,18 +1,49 @@ #ifndef DMX_HOST_ENGINE__ #define DMX_HOST_ENGINE__ +#include "clickhouse_client.h" #include "dmx_host_utils.h" #include "pipelined_engine.hpp" +#include +#include + namespace dmx_host{ -// DMXHostEngine is a single-stage ClickHouse insert pipeline. -// Pre-assembled ClickHouseRows are submitted via submit_direct(). -class DMXHostEngine : public PipelinedEngine, false, -NoOutputHandler >{ +using DMXHostEngineBase = PipelinedEngine< + dmx_host_queue_item, uint64_t, 1, QueueOptions, false, + NoOutputHandler>; + +class DMXHostEngine : public DMXHostEngineBase { public: - explicit DMXHostEngine(StageConfig insert_stage): - PipelinedEngine(std::array{std::move(insert_stage)}, EngineConfig{}){} + using Base = DMXHostEngineBase; + using StageConfig = Base::StageConfig; + using EngineConfig = Base::EngineConfig; + using Duration = Base::Duration; + + explicit DMXHostEngine(StageConfig insert_stage) + : DMXHostEngine(Prepare(std::move(insert_stage))) {} + + bool wait_until_ready(Duration timeout) { + if (timeout.count() < 0.0) { + throw std::invalid_argument("timeout must be non-negative"); + } + const auto deadline = std::chrono::steady_clock::now() + + std::chrono::duration_cast(timeout); + while (std::chrono::steady_clock::now() < deadline) { + const auto remaining = std::chrono::duration_cast( + deadline - std::chrono::steady_clock::now()); + const auto slice = std::min(remaining, std::chrono::milliseconds(50)); + if (metrics_->WaitUntilReady(slice)) return true; + if (!failures().empty()) return false; + } + const auto snapshot = metrics_->Snapshot(); + return snapshot.ready_workers == snapshot.expected_workers; + } + + ClickHouseMetricsSnapshot clickhouse_metrics() const { + return metrics_->Snapshot(); + } // Submit a pre-assembled ClickHouseRow directly to the insert stage. // Fields must match the order expected by ClickHouseInsertStage: @@ -29,6 +60,32 @@ NoOutputHandler >{ items.emplace_back(std::move(row), nbytes); submit_items(std::move(items)); } + +private: + struct PreparedStage { + StageConfig stage; + std::shared_ptr metrics; + }; + + static PreparedStage Prepare(StageConfig stage) { + if (stage.parallelism <= 0) { + throw std::invalid_argument("parallelism must be positive"); + } + auto config = + std::any_cast(stage.thread_init_config); + config.schema_once = std::make_shared(); + auto metrics = + std::make_shared(stage.parallelism); + config.runtime_metrics = metrics; + stage.thread_init_config = std::move(config); + return PreparedStage{std::move(stage), std::move(metrics)}; + } + + explicit DMXHostEngine(PreparedStage prepared) + : Base(std::array{std::move(prepared.stage)}, EngineConfig{}), + metrics_(std::move(prepared.metrics)) {} + + std::shared_ptr metrics_; }; } diff --git a/tests/test_clickhouse_host_benchmark.py b/tests/test_clickhouse_host_benchmark.py index 7e3ed77be..06407c437 100644 --- a/tests/test_clickhouse_host_benchmark.py +++ b/tests/test_clickhouse_host_benchmark.py @@ -9,6 +9,7 @@ import benchmarks.bench_clickhouse_host as benchmark from benchmarks.bench_clickhouse_host import ( BenchmarkConfig, + ServerTelemetrySampler, TrialMeasurement, _ensure_table, _parts_metrics, @@ -18,8 +19,11 @@ build_client_settings, configure_stage, generate_payload_pool, + parse_parallelism_sweep, parse_byte_size, quote_identifier, + run_sweep, + summarize_scaling_trials, ) pytestmark = pytest.mark.cpu @@ -78,6 +82,13 @@ def test_config_rejects_non_positive_index_granularity(): BenchmarkConfig(index_granularity=0) +def test_config_rejects_invalid_socket_and_sampling_timeouts(): + with pytest.raises(ValueError, match="socket_timeout_seconds"): + BenchmarkConfig(socket_timeout_seconds=0) + with pytest.raises(ValueError, match="server_sample_interval_ms"): + BenchmarkConfig(server_sample_interval_ms=-1) + + def test_config_uses_a_unique_benchmark_table_by_default(): first = BenchmarkConfig().table second = BenchmarkConfig().table @@ -132,16 +143,31 @@ def test_stage_configuration_sets_batching_and_backpressure(): assert queue.high_watermark_items == 100 assert queue.high_watermark_size == 256 assert policy.block is True + assert policy.timeout_s == cfg.socket_timeout_seconds def test_async_insert_settings_wait_for_durability(): - assert build_client_settings(False) == {} + assert build_client_settings(False) == {"async_insert": 0} assert build_client_settings(True) == { "async_insert": 1, "wait_for_async_insert": 1, } +def test_single_value_sweep_uses_requested_parallelism(monkeypatch, capsys): + seen = [] + + def runner(config): + seen.append(config.parallelism) + return {"parallelism": config.parallelism} + + monkeypatch.setattr(benchmark, "run", runner) + + assert benchmark.main(["--parallelism-sweep", "8"]) == 0 + assert seen == [8] + assert '"parallelism": 8' in capsys.readouterr().out + + def test_server_time_uses_clickhouse_clock(): class Client: def execute(self, query): @@ -279,6 +305,8 @@ def test_trial_measurement_reports_enqueue_and_drain_separately(): enqueue_seconds=2.0, total_seconds=5.0, enqueue_latencies_ns=(1_000_000, 2_000_000, 3_000_000, 4_000_000), + startup_seconds=0.25, + client_metrics={"peak_active_inserts": 3}, ) report = trial.as_dict() @@ -289,6 +317,8 @@ def test_trial_measurement_reports_enqueue_and_drain_separately(): assert report["enqueue"]["latency_ms"]["p50"] == 2.5 assert report["enqueue"]["latency_ms"]["p95"] == pytest.approx(3.85) assert report["enqueue"]["latency_samples"] == 4 + assert report["startup_seconds"] == 0.25 + assert report["client"]["peak_active_inserts"] == 3 def test_submit_and_drain_uses_graceful_lifecycle(): @@ -299,6 +329,10 @@ def __init__(self): def start(self): self.calls.append("start") + def wait_until_ready(self, timeout): + self.calls.append(("wait_until_ready", timeout)) + return True + def submit_direct(self, *args): self.calls.append(("submit", args)) @@ -312,10 +346,30 @@ def join(self, timeout): def raise_if_failed(self): self.calls.append("raise_if_failed") + def clickhouse_metrics(self): + return SimpleNamespace( + expected_workers=2, + ready_workers=2, + active_inserts=0, + peak_active_inserts=2, + batches=2, + rows=2, + logical_bytes=32, + insert_seconds=0.5, + workers=[], + ) + def request_abort(self): self.calls.append("request_abort") - ticks = iter((0, 10, 15, 20, 30, 40, 70)) + class Sampler: + def start(self): + engine.calls.append("sampler_start") + + def stop(self): + engine.calls.append("sampler_stop") + + ticks = iter((0, 10, 20, 30, 40, 50, 60, 70, 90)) engine = Engine() payload = SimpleNamespace(nbytes=16) @@ -326,9 +380,12 @@ def request_abort(self): model_id="run", timeout_seconds=5.0, clock=lambda: next(ticks), + deadline_clock=lambda: 0.0, + sampler=Sampler(), ) assert [call[0] for call in engine.calls if isinstance(call, tuple)] == [ + "wait_until_ready", "submit", "submit", "join", @@ -336,8 +393,144 @@ def request_abort(self): assert "close_input" in engine.calls assert "raise_if_failed" in engine.calls assert "request_abort" not in engine.calls - assert result.enqueue_seconds == 40 / 1_000_000_000 + assert engine.calls.index("sampler_start") > engine.calls.index(("wait_until_ready", 5.0)) + assert engine.calls.index("sampler_stop") > engine.calls.index(("join", 5.0)) + assert result.startup_seconds == 10 / 1_000_000_000 + assert result.enqueue_seconds == 50 / 1_000_000_000 assert result.total_seconds == 70 / 1_000_000_000 + assert result.client_metrics["peak_active_inserts"] == 2 + + +def test_submit_and_drain_does_not_restart_timeout_during_abort(): + class Engine: + def __init__(self): + self.join_timeouts = [] + + def start(self): + pass + + def wait_until_ready(self, timeout): + return True + + def submit_direct(self, *args): + pass + + def close_input(self): + pass + + def join(self, timeout): + self.join_timeouts.append(timeout) + return False + + def request_abort(self): + pass + + deadlines = iter((0.0, 0.0, 0.0, 0.0, 5.0)) + engine = Engine() + + with pytest.raises(TimeoutError, match="drain exceeded"): + _submit_and_drain( + engine, + [SimpleNamespace(nbytes=16)], + rows=1, + model_id="run", + timeout_seconds=5.0, + clock=iter(range(20)).__next__, + deadline_clock=lambda: next(deadlines), + ) + + assert engine.join_timeouts == [5.0, 0.0] + + +def test_parallelism_sweep_is_unique_and_positive(): + assert parse_parallelism_sweep("1,2,4,2,8") == (1, 2, 4, 8) + with pytest.raises(ValueError, match="positive"): + parse_parallelism_sweep("1,0,4") + + +def test_scaling_summary_reports_speedup_and_variance(): + trials = [ + {"parallelism": 1, "throughput": 1.0}, + {"parallelism": 1, "throughput": 1.2}, + {"parallelism": 2, "throughput": 1.8}, + {"parallelism": 2, "throughput": 2.0}, + ] + + summary = summarize_scaling_trials(trials, plateau_threshold_percent=5.0) + + assert summary[0]["parallelism"] == 1 + assert summary[0]["median_gib_per_second"] == pytest.approx(1.1) + assert summary[1]["speedup_vs_one"] == pytest.approx(1.9 / 1.1) + assert summary[1]["gain_vs_previous_percent"] == pytest.approx((1.9 / 1.1 - 1) * 100) + assert summary[1]["plateau"] is False + + +def test_scaling_summary_does_not_invent_one_worker_baseline(): + summary = summarize_scaling_trials( + [ + {"parallelism": 2, "throughput": 2.0}, + {"parallelism": 4, "throughput": 3.0}, + ] + ) + + assert summary[0]["baseline_parallelism"] == 2 + assert summary[0]["speedup_vs_one"] is None + assert summary[1]["speedup_vs_baseline"] == pytest.approx(1.5) + + +def test_run_sweep_keeps_raw_trials_and_uses_unique_tables(): + calls = [] + + def runner(config): + calls.append(config) + return { + "measurement": { + "total": {"logical_gib_per_second": float(config.parallelism)} + } + } + + report = run_sweep( + BenchmarkConfig(rows=1, payload_bytes=16, min_batch_bytes=16, + max_batch_bytes=16, queue_capacity_bytes=32, + warmup_rows=0), + parallelisms=(1, 2), + trials=2, + runner=runner, + ) + + assert len(report["trials"]) == 4 + assert {trial["parallelism"] for trial in report["trials"]} == {1, 2} + assert len({config.table for config in calls}) == 4 + assert report["summary"][1]["speedup_vs_one"] == 2.0 + + +def test_server_sampler_tracks_insert_concurrency_and_metric_peaks(): + class Client: + def __init__(self): + self.calls = 0 + + def execute(self, query): + self.calls += 1 + if "system.processes" in query: + return [(2,)] + if "system.asynchronous_metrics" in query: + return [("OSUserTimeNormalized", 0.5)] + return [("Merge", 3), ("Query", 5)] + + def disconnect(self): + pass + + sampler = ServerTelemetrySampler(lambda: Client(), interval_ms=1) + sampler.sample_once() + snapshot = sampler.snapshot() + + assert snapshot["samples"] == 1 + assert snapshot["peak_active_inserts"] == 2 + assert snapshot["metric_peaks"] == { + "Merge": 3.0, + "Query": 5.0, + "async.OSUserTimeNormalized": 0.5, + } def test_identifier_quoting_rejects_sql_fragments(): diff --git a/tests/test_cpu_native_build.py b/tests/test_cpu_native_build.py index d98363285..89bd1a0e1 100644 --- a/tests/test_cpu_native_build.py +++ b/tests/test_cpu_native_build.py @@ -9,9 +9,7 @@ import pytest -pytestmark = pytest.mark.cpu - - +@pytest.mark.cpu def test_host_build_plan_has_no_cuda_toolchain_or_libraries(): root = Path(__file__).resolve().parents[1] result = subprocess.run( @@ -31,6 +29,7 @@ def test_host_build_plan_has_no_cuda_toolchain_or_libraries(): assert forbidden not in output +@pytest.mark.cpu def test_host_export_falls_back_to_cpu_backend(monkeypatch): from dmi.transport import native @@ -50,6 +49,53 @@ def load_named(name): assert calls == ["_native_backend", "_host_backend"] +@pytest.mark.native_backend +def test_clickhouse_stage_has_bounded_batching_defaults(): + from dmi.transport.native import ClickHouseClientConfig, StageConfig + + stage = StageConfig.clickhouse_insert(ClickHouseClientConfig()) + queue = stage.input_queue + + assert queue.min_batch_items is None + assert queue.min_batch_size == 16 * 1024**2 + assert queue.max_linger_s == pytest.approx(0.05) + assert queue.max_batch_items == 10_000 + assert queue.max_batch_size is None + assert queue.high_watermark_items == 20_000 + assert queue.high_watermark_size == 512 * 1024**2 + + +@pytest.mark.native_backend +def test_clickhouse_client_exposes_socket_timeouts_and_worker_metrics(): + from dmi.transport.native import ClickHouseClientConfig, DMXHostEngine, StageConfig + + config = ClickHouseClientConfig() + assert config.connect_timeout_ms == 5000 + assert config.receive_timeout_ms == 0 + assert config.send_timeout_ms == 0 + + engine = DMXHostEngine(StageConfig.clickhouse_insert(config, parallelism=3)) + metrics = engine.clickhouse_metrics() + assert metrics.expected_workers == 3 + assert metrics.ready_workers == 0 + assert metrics.peak_active_inserts == 0 + assert [worker.worker_index for worker in metrics.workers] == [0, 1, 2] + + +@pytest.mark.native_backend +def test_engine_metrics_follow_mutated_stage_parallelism(): + from dmi.transport.native import ClickHouseClientConfig, DMXHostEngine, StageConfig + + stage = StageConfig.clickhouse_insert(ClickHouseClientConfig(), parallelism=1) + stage.parallelism = 3 + + metrics = DMXHostEngine(stage).clickhouse_metrics() + + assert metrics.expected_workers == 3 + assert [worker.worker_index for worker in metrics.workers] == [0, 1, 2] + + +@pytest.mark.cpu def test_ring_export_requires_full_backend(monkeypatch): from dmi.transport import native @@ -67,6 +113,7 @@ def load_named(name): assert calls == ["_native_backend"] +@pytest.mark.cpu def test_v1_host_export_does_not_load_ring_backend(monkeypatch): import dmi.api.v1 as api from dmi.transport import native @@ -85,6 +132,7 @@ def test_v1_host_export_does_not_load_ring_backend(monkeypatch): api.__dict__["DMXHostEngine"] = cached +@pytest.mark.cpu def test_v1_model_shape_contract_does_not_load_ring_backend(): import dmi.api.v1 as api