You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by cr...@apache.org on 2026/07/20 03:08:25 UTC
(iotdb-client-rust) 16/20: Benchmark: mirror iot-benchmark measurement semantics (prep inside timed span, failures excluded from latency, Result/Latency Matrix output with P10-P999 + SLOWEST_THREAD)
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 <cr...@outlook.com>
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());
}