This is an automated email from the ASF dual-hosted git repository. CritasWang pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/iotdb-client-rust.git
commit d6d3ed3e989af4c3004764f03bc02a49e615d283 Author: CritasWang <[email protected]> AuthorDate: Mon Jul 13 19:32:40 2026 +0800 Benchmark: mirror iot-benchmark measurement semantics (prep inside timed span, failures excluded from latency, Result/Latency Matrix output with P10-P999 + SLOWEST_THREAD) --- README.md | 8 ++- README_ZH.md | 8 ++- examples/benchmark.rs | 159 +++++++++++++++++++++++++++++++++++++++++--------- 3 files changed, 141 insertions(+), 34 deletions(-) diff --git a/README.md b/README.md index fbb379f..a403b7d 100644 --- a/README.md +++ b/README.md @@ -253,7 +253,9 @@ cargo test # now includes the live-server tests ## Benchmark -`examples/benchmark.rs` is a write-performance benchmark modeled on the Node.js client's `benchmark/` suite (which follows [thulab/iot-benchmark](https://github.com/thulab/iot-benchmark)); metric definitions match, so results are comparable across the SDKs. Tablets are pre-generated outside the timed section; N worker threads each own a pooled session and insert `insert_tablet` batches round-robin over their devices. Timestamps are sequential per device from a fixed base, so runs are det [...] +`examples/benchmark.rs` is a write-performance benchmark modeled on the Node.js client's `benchmark/` suite and on [thulab/iot-benchmark](https://github.com/thulab/iot-benchmark). Tablets are pre-generated outside the timed section; N worker threads each own a pooled session and insert `insert_tablet` batches round-robin over their devices. Timestamps are sequential per device from a fixed base, so runs are deterministic. + +> **Statistics semantics now mirror iot-benchmark:** the per-operation timed span includes batch preparation (not just the insert RPC), failed operations are excluded from latency samples (counted as `failOperation`/`failPoint`), and the output includes iot-benchmark-style Result Matrix and Latency (ms) Matrix sections (AVG…P999/MAX/SLOWEST_THREAD; percentiles are exact, whereas iot-benchmark uses a t-digest approximation). Numbers produced by earlier versions of this benchmark (RPC-only [...] ```sh # tree model, defaults: 100 devices × 10 sensors × 20 batches × 1000 rows = 20M points, 8 clients @@ -264,9 +266,9 @@ cargo run --release --example benchmark -- --mode table \ --devices 20 --sensors 10 --batches 100 --batch-size 100 --clients 8 --cleanup ``` -Knobs: `--mode tree|table`, `--devices`, `--sensors`, `--batches` (per device), `--batch-size` (rows per tablet), `--clients` (worker threads = pool size), `--host/--port/--user/--password` (also via `IOTDB_HOST/PORT/USER/PASSWORD`), `--base-ts`, `--point-step`, `--reuse-tablets` (pre-generate only N tablets per worker and re-send them with rebased timestamps — bounds memory for very large runs; the per-batch timestamp rewrite happens inside the timed loop, like a real streaming producer [...] +Knobs: `--mode tree|table`, `--devices`, `--sensors`, `--batches` (per device), `--batch-size` (rows per tablet), `--clients` (worker threads = pool size), `--host/--port/--user/--password` (also via `IOTDB_HOST/PORT/USER/PASSWORD`), `--base-ts`, `--point-step`, `--reuse-tablets` (pre-generate only N tablets per worker and re-send them with rebased timestamps — bounds memory for very large runs; the per-batch timestamp rewrite happens inside the timed loop, like a real streaming producer [...] -Measured on an Apple M2 Pro (10 cores), IoTDB 2.0.6 standalone in Docker on the same machine (Docker VM: all 10 CPUs / 8 GB; JVM heap 1 GB), release build: +Measured on an Apple M2 Pro (10 cores), IoTDB 2.0.6 standalone in Docker on the same machine (Docker VM: all 10 CPUs / 8 GB; JVM heap 1 GB), release build, **with the old RPC-only timing** (see note above — expect somewhat lower throughput/higher latency with the current semantics): | Mode | Devices × Sensors × Batches × Rows | Clients | Points | Throughput | p50 / p99 latency | | --- | --- | --- | --- | --- | --- | diff --git a/README_ZH.md b/README_ZH.md index dc5807f..6f62b4f 100644 --- a/README_ZH.md +++ b/README_ZH.md @@ -253,7 +253,9 @@ cargo test # 此时包含在线测试 ## 性能基准测试 -`examples/benchmark.rs` 是写入性能基准，参考 Node.js 客户端的 `benchmark/` 套件（后者遵循 [thulab/iot-benchmark](https://github.com/thulab/iot-benchmark)）；指标定义一致，因此各语言 SDK 的结果可直接对比。Tablet 在计时区间之外预先生成；N 个工作线程各持有一个池化会话，按批次轮询各自的设备执行 `insert_tablet`。时间戳从固定基准按设备顺序递增，运行结果可复现。 +`examples/benchmark.rs` 是写入性能基准，参考 Node.js 客户端的 `benchmark/` 套件与 [thulab/iot-benchmark](https://github.com/thulab/iot-benchmark)。Tablet 在计时区间之外预先生成；N 个工作线程各持有一个池化会话，按批次轮询各自的设备执行 `insert_tablet`。时间戳从固定基准按设备顺序递增，运行结果可复现。 + +> **统计口径现已对齐 iot-benchmark：**单次操作的计时区间包含批次准备（而非仅 insert RPC）；失败操作不计入延迟样本（单独统计 `failOperation`/`failPoint`）；输出包含 iot-benchmark 风格的 Result Matrix 与 Latency (ms) Matrix（AVG…P999/MAX/SLOWEST_THREAD；百分位为精确值，iot-benchmark 使用 t-digest 近似）。旧版基准（仅计 RPC 时间，含下表数据）的数字与新输出**不可直接对比**。 ```sh # 树模型，默认规模：100 设备 × 10 传感器 × 20 批 × 1000 行 = 2000 万点，8 客户端 @@ -264,9 +266,9 @@ cargo run --release --example benchmark -- --mode table \ --devices 20 --sensors 10 --batches 100 --batch-size 100 --clients 8 --cleanup ``` -参数：`--mode tree|table`、`--devices`、`--sensors`、`--batches`（每设备批数）、`--batch-size`（每 tablet 行数）、`--clients`（工作线程数 = 池大小）、`--host/--port/--user/--password`（亦支持 `IOTDB_HOST/PORT/USER/PASSWORD` 环境变量）、`--base-ts`、`--point-step`、`--reuse-tablets`（每工作线程仅预生成 N 个 tablet 并循环重发，每批重写时间戳基准——为超大规模运行限定内存占用；时间戳重写在计时循环内进行，等同真实流式生产者的按批打时间戳）、`--tablets-per-rpc`（树模型：每次 RPC 通过 `insert_tablets` 批量发送 N 个 tablet）、`--cleanup`。传感器类型分布沿用 Node.js 默认比例（30% FLOAT、20% DOUBLE、20% INT32、10% INT64、10% TEXT、10% BOOLEAN）。报告 [...] +参数：`--mode tree|table`、`--devices`、`--sensors`、`--batches`（每设备批数）、`--batch-size`（每 tablet 行数）、`--clients`（工作线程数 = 池大小）、`--host/--port/--user/--password`（亦支持 `IOTDB_HOST/PORT/USER/PASSWORD` 环境变量）、`--base-ts`、`--point-step`、`--reuse-tablets`（每工作线程仅预生成 N 个 tablet 并循环重发，每批重写时间戳基准——为超大规模运行限定内存占用；时间戳重写在计时循环内进行，等同真实流式生产者的按批打时间戳）、`--tablets-per-rpc`（树模型：每次 RPC 通过 `insert_tablets` 批量发送 N 个 tablet）、`--cleanup`。传感器类型分布沿用 Node.js 默认比例（30% FLOAT、20% DOUBLE、20% INT32、10% INT64、10% TEXT、10% BOOLEAN）。报告 [...] -实测环境：Apple M2 Pro（10 核），IoTDB 2.0.6 standalone（Docker，与客户端同机；Docker VM 全部 10 核 / 8 GB，JVM 堆 1 GB），release 构建： +实测环境：Apple M2 Pro（10 核），IoTDB 2.0.6 standalone（Docker，与客户端同机；Docker VM 全部 10 核 / 8 GB，JVM 堆 1 GB），release 构建，**采用旧的仅计 RPC 口径**（见上方说明——新口径下吞吐会略低、延迟会略高）： | 模式 | 设备 × 传感器 × 批数 × 行数 | 客户端 | 总点数 | 吞吐量 | p50 / p99 延迟 | | --- | --- | --- | --- | --- | --- | diff --git a/examples/benchmark.rs b/examples/benchmark.rs index 3f44afb..b949cae 100644 --- a/examples/benchmark.rs +++ b/examples/benchmark.rs @@ -16,14 +16,28 @@ // under the License. //! Write-performance benchmark for the tree and table models, modeled on the -//! Node.js client's `benchmark/` suite (which in turn follows -//! [thulab/iot-benchmark](https://github.com/thulab/iot-benchmark)). Metric -//! definitions match the Node.js `MetricsCollector` so results are directly -//! comparable: per-`insert_tablet` latency percentiles (p = ceil(p% × n) − 1 -//! over the sorted samples), points/sec over the wall time of the timed -//! insert phase only. Schema setup and data generation happen **outside** -//! the timed section ("Pre-generated Test Data"); timestamps are sequential -//! per device from a fixed base, so runs are deterministic. +//! Node.js client's `benchmark/` suite and on +//! [thulab/iot-benchmark](https://github.com/thulab/iot-benchmark). The +//! **measurement semantics mirror iot-benchmark** (`DBWrapper` / +//! `Measurement`): +//! +//! - the per-operation latency span starts **before** batch preparation for +//! that batch (Java times `genTablet` + submit + `Future.get`, +//! `SessionStrategy.insertOneBatchByTablet`), not just the insert RPC; +//! - failed operations record **no latency sample** — they only count toward +//! `failOperation` / `failPoint` (`DBWrapper.measureOneBatch`); +//! - throughput = okPoint / elapsed, where elapsed is the wall time of the +//! whole data phase (schema creation excluded), matching +//! `BaseMode.measure`; +//! - output includes an iot-benchmark-style "Result Matrix" and +//! "Latency (ms) Matrix" (AVG MIN P10 P25 MEDIAN P75 P90 P95 P99 P999 MAX +//! SLOWEST_THREAD) so runs diff side-by-side with Java logs. Percentiles +//! here are exact over the sorted samples (p = ceil(p% × n) − 1); +//! iot-benchmark uses a t-digest approximation. SLOWEST_THREAD is the max +//! over threads of that thread's latency sum (`Measurement.mergeMeasurement`). +//! +//! Schema setup happens **outside** the timed section; timestamps are +//! sequential per device from a fixed base, so runs are deterministic. //! //! Sensor data types follow the Node.js default distribution: //! 30% FLOAT, 20% DOUBLE, 20% INT32, 10% INT64, 10% TEXT, 10% BOOLEAN. @@ -462,17 +476,23 @@ fn verify_row_count(pool: &Pool, config: &BenchConfig) -> Result<()> { } // --------------------------------------------------------------------------- -// Metrics (definitions match Node.js benchmark-core.js MetricsCollector) +// Metrics (semantics mirror iot-benchmark's Measurement / DBWrapper) // --------------------------------------------------------------------------- #[derive(Default)] struct WorkerStats { - /// One entry per insert_tablet call (ms), success or failure. + /// One entry per **successful** operation (ms). The timed span covers + /// batch preparation + the insert RPC (iot-benchmark's DBWrapper times + /// genTablet + submit + Future.get; failures record no latency). latencies_ms: Vec<f64>, + /// All operations attempted (ok + failed). ops: u64, - failures: u64, - /// Points from successful operations only. + /// Failed operations (iot-benchmark failOperation). + fail_operations: u64, + /// Points from successful operations only (iot-benchmark okPoint). points: u64, + /// Points belonging to failed operations (iot-benchmark failPoint). + fail_points: u64, error_samples: Vec<String>, } @@ -492,14 +512,23 @@ fn print_summary(config: &BenchConfig, wall: Duration, workers: &[WorkerStats]) .collect(); latencies.sort_by(|a, b| a.partial_cmp(b).expect("latency is never NaN")); let ops: u64 = workers.iter().map(|w| w.ops).sum(); - let failures: u64 = workers.iter().map(|w| w.failures).sum(); - let points: u64 = workers.iter().map(|w| w.points).sum(); + let fail_operations: u64 = workers.iter().map(|w| w.fail_operations).sum(); + let ok_operations = ops - fail_operations; + let ok_points: u64 = workers.iter().map(|w| w.points).sum(); + let fail_points: u64 = workers.iter().map(|w| w.fail_points).sum(); let secs = wall.as_secs_f64(); + // AVG = sum(ok latencies) / okOperations (Measurement.calculateMetrics). let avg = if latencies.is_empty() { 0.0 } else { latencies.iter().sum::<f64>() / latencies.len() as f64 }; + // SLOWEST_THREAD = max over threads of that thread's latency sum + // (Measurement.mergeMeasurement → MAX_THREAD_LATENCY_SUM). + let slowest_thread: f64 = workers + .iter() + .map(|w| w.latencies_ms.iter().sum::<f64>()) + .fold(0.0, f64::max); let sep = "=".repeat(80); println!("\n{sep}\nBENCHMARK RESULTS\n{sep}"); @@ -511,22 +540,25 @@ fn print_summary(config: &BenchConfig, wall: Duration, workers: &[WorkerStats]) ); println!("\n[Operations]"); println!(" Total Operations: {ops}"); - println!(" Successful: {}", ops - failures); - println!(" Failed: {failures}"); + println!(" Successful: {ok_operations}"); + println!(" Failed: {fail_operations}"); println!( " Success Rate: {:.2}%", if ops == 0 { 0.0 } else { - (ops - failures) as f64 / ops as f64 * 100.0 + ok_operations as f64 / ops as f64 * 100.0 } ); println!("\n[Data Points]"); - println!(" Total Points Written: {points}"); + println!(" Total Points Written: {ok_points}"); + if fail_points > 0 { + println!(" Failed Points: {fail_points}"); + } println!("\n[Throughput]"); println!(" Operations/sec: {:.2}", ops as f64 / secs); - println!(" Points/sec: {:.0}", points as f64 / secs); - println!("\n[Latency (ms)]"); + println!(" Points/sec: {:.0}", ok_points as f64 / secs); + println!("\n[Latency (ms)] (successful operations only; span includes batch preparation)"); println!( " Min: {:.2}ms", latencies.first().copied().unwrap_or(0.0) @@ -575,6 +607,67 @@ fn print_summary(config: &BenchConfig, wall: Duration, workers: &[WorkerStats]) config.clients, config.total_points(), ); + + // --- iot-benchmark-style matrices (columns/order match Measurement) ---- + // %-25s per Result-Matrix cell, %-12s per Latency-Matrix metric cell. + println!("\nTest elapsed time (not include schema creation): {secs:.2} second"); + println!( + "----------------------------------------------------------Result Matrix----------------------------------------------------------" + ); + println!( + "{:<25}{:<25}{:<25}{:<25}{:<25}{:<25}", + "Operation", "okOperation", "okPoint", "failOperation", "failPoint", "throughput(point/s)" + ); + println!( + "{:<25}{:<25}{:<25}{:<25}{:<25}{:<25}", + "INGESTION", + ok_operations, + ok_points, + fail_operations, + fail_points, + format!( + "{:.2}", + if secs > 0.0 { + ok_points as f64 / secs + } else { + 0.0 + } + ) + ); + println!( + "---------------------------------------------------------------------------------------------------------------------------------" + ); + println!( + "--------------------------------------------------------------------------Latency (ms) Matrix--------------------------------------------------------------------------" + ); + print!("{:<25}", "Operation"); + let metrics: [(&str, f64); 12] = [ + ("AVG", avg), + ("MIN", latencies.first().copied().unwrap_or(0.0)), + ("P10", percentile(&latencies, 10.0)), + ("P25", percentile(&latencies, 25.0)), + ("MEDIAN", percentile(&latencies, 50.0)), + ("P75", percentile(&latencies, 75.0)), + ("P90", percentile(&latencies, 90.0)), + ("P95", percentile(&latencies, 95.0)), + ("P99", percentile(&latencies, 99.0)), + ("P999", percentile(&latencies, 99.9)), + ("MAX", latencies.last().copied().unwrap_or(0.0)), + ("SLOWEST_THREAD", slowest_thread), + ]; + for (name, _) in &metrics { + print!("{name:<12}"); + } + println!(); + print!("{:<25}", "INGESTION"); + for (_, value) in &metrics { + print!("{:<12}", format!("{value:.2}")); + } + println!(); + println!( + "-----------------------------------------------------------------------------------------------------------------------------------------------------------------------" + ); + println!("Note: exact percentiles (iot-benchmark uses t-digest approximation)"); } // --------------------------------------------------------------------------- @@ -724,6 +817,17 @@ fn main() -> Result<()> { let materialized = tablets.len(); let mut i = 0; while i < schedule_len { + // Per-op timing mirrors iot-benchmark: the timer + // starts BEFORE batch preparation, as Java's + // DBWrapper.insertOneBatch times genTablet + submit + // + Future.get. Mapping per mode: + // * reuse mode — timestamp rebase (the per-batch + // work a streaming producer would do) + the RPC + // are inside the timed span; + // * non-reuse — tablets are pre-built (Java's + // pre-built workload), so the span covers + // borrowing the tablet slice + the RPC. + let start = Instant::now(); // A chunk never wraps the materialized ring, so it // always maps to one contiguous slice. let idx = i % materialized; @@ -734,9 +838,7 @@ fn main() -> Result<()> { if config.reuse_tablets > 0 { // Rebase each reused tablet onto its iteration's // disjoint time window so every insert writes - // fresh timestamps. This runs inside the timed - // loop on purpose: it's the same per-batch - // timestamping a real streaming producer would do. + // fresh timestamps. for (k, tablet) in tablets[idx..idx + chunk].iter_mut().enumerate() { let window = config.base_ts + ((i + k) * config.batch_size) as i64 * config.point_step; @@ -750,24 +852,25 @@ fn main() -> Result<()> { .iter() .map(|t| (t.row_count() * config.sensors) as u64) .sum(); - let start = Instant::now(); let outcome = if chunk == 1 { session.insert_tablet(&batch[0]) } else { session.insert_tablets(batch, false) }; - stats - .latencies_ms - .push(start.elapsed().as_secs_f64() * 1000.0); + let elapsed_ms = start.elapsed().as_secs_f64() * 1000.0; stats.ops += 1; ops_done.fetch_add(1, Ordering::Relaxed); match outcome { Ok(()) => { + // Only successful operations record a latency + // sample (DBWrapper.measureOneBatch). + stats.latencies_ms.push(elapsed_ms); stats.points += points; points_done.fetch_add(points, Ordering::Relaxed); } Err(e) => { - stats.failures += 1; + stats.fail_operations += 1; + stats.fail_points += points; if stats.error_samples.len() < 5 { stats.error_samples.push(e.to_string()); }
