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 6d64d2a87b4e612423227b25daebba367b9cbca8 Author: CritasWang <[email protected]> AuthorDate: Mon Jul 13 15:26:14 2026 +0800 Benchmark scale-up: --reuse-tablets (bounded memory) + --tablets-per-rpc (insert_tablets batching); 22.5M pts/s sustained over 250M points, server-side bottleneck attributed (1G heap flush stalls) --- README.md | 18 +++++---- README_ZH.md | 18 +++++---- examples/benchmark.rs | 103 +++++++++++++++++++++++++++++++++++++++++++------- src/data/tablet.rs | 8 ++++ 4 files changed, 118 insertions(+), 29 deletions(-) diff --git a/README.md b/README.md index 2b5dfd5..fbb379f 100644 --- a/README.md +++ b/README.md @@ -264,17 +264,19 @@ 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`, `--cleanup`. Sensor types follow the Node.js default distribution (30% FLOAT, 20% DOUBLE, 20% INT32, 10% INT64, 10% TEXT, 10% BOOLEAN). The report includes total points, wall time, points/sec, per-batch latency p50/p90/p95/p99/ma [...] +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, release build, 8 clients: +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: -| Mode | Devices × Sensors × Batches × Rows | Points | Throughput | p50 / p99 latency | -| --- | --- | --- | --- | --- | -| tree | 20 × 10 × 100 × 100 | 2M | ~1.98M pts/s | 2.46 ms / 8.38 ms | -| table | 20 × 10 × 100 × 100 | 2M | ~1.97M pts/s | 2.13 ms / 9.97 ms | -| tree | 100 × 10 × 10 × 1000 | 10M | ~9.73M pts/s | 4.36 ms / 72.18 ms | +| Mode | Devices × Sensors × Batches × Rows | Clients | Points | Throughput | p50 / p99 latency | +| --- | --- | --- | --- | --- | --- | +| tree | 20 × 10 × 100 × 100 | 8 | 2M | ~1.98M pts/s | 2.46 ms / 8.38 ms | +| table | 20 × 10 × 100 × 100 | 8 | 2M | ~1.97M pts/s | 2.13 ms / 9.97 ms | +| tree | 100 × 10 × 20 × 1000 | 8 | 20M | ~12.4M pts/s | 4.45 ms / 27.03 ms | +| tree | 100 × 100 × 4 × 1000 | 10 | 40M | ~15–20M pts/s | 31 ms / 156 ms | +| tree | 100 × 100 × 25 × 1000, `--tablets-per-rpc 4` | 10 | 250M | **~21–22.5M pts/s** | 105 ms / 907 ms | -Throughput scales with tablet size (rows × sensors per RPC); 1000-row tablets give ~5× the throughput of 100-row tablets at the same client count. Numbers are client+server on one machine — treat them as an upper bound on client overhead, not a server capacity measurement. +Throughput scales with points per RPC: wider tablets (100 sensors = 100k points per 1000-row tablet) and multi-tablet `insert_tablets` batching lift the same hardware from ~12M to ~22M pts/s sustained (250M points, `--reuse-tablets`). Beyond ~400k points per RPC — or more clients than cores — throughput plateaus and tail latency grows; during peak runs the server JVM bursts to ~3 cores then stalls on memtable flushes while the Rust client sits at ~30% of one core, so the ceiling here is [...] ## Project layout diff --git a/README_ZH.md b/README_ZH.md index 25e384b..dc5807f 100644 --- a/README_ZH.md +++ b/README_ZH.md @@ -264,17 +264,19 @@ 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`、`--cleanup`。传感器类型分布沿用 Node.js 默认比例（30% FLOAT、20% DOUBLE、20% INT32、10% INT64、10% TEXT、10% BOOLEAN）。报告包含总点数、耗时、points/sec、单批延迟 p50/p90/p95/p99/max、错误数，以及读回行数校验。 +参数：`--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，与客户端同机），release 构建，8 客户端： +实测环境：Apple M2 Pro（10 核），IoTDB 2.0.6 standalone（Docker，与客户端同机；Docker VM 全部 10 核 / 8 GB，JVM 堆 1 GB），release 构建： -| 模式 | 设备 × 传感器 × 批数 × 行数 | 总点数 | 吞吐量 | p50 / p99 延迟 | -| --- | --- | --- | --- | --- | -| tree | 20 × 10 × 100 × 100 | 200 万 | ~198 万 pts/s | 2.46 ms / 8.38 ms | -| table | 20 × 10 × 100 × 100 | 200 万 | ~197 万 pts/s | 2.13 ms / 9.97 ms | -| tree | 100 × 10 × 10 × 1000 | 1000 万 | ~973 万 pts/s | 4.36 ms / 72.18 ms | +| 模式 | 设备 × 传感器 × 批数 × 行数 | 客户端 | 总点数 | 吞吐量 | p50 / p99 延迟 | +| --- | --- | --- | --- | --- | --- | +| tree | 20 × 10 × 100 × 100 | 8 | 200 万 | ~198 万 pts/s | 2.46 ms / 8.38 ms | +| table | 20 × 10 × 100 × 100 | 8 | 200 万 | ~197 万 pts/s | 2.13 ms / 9.97 ms | +| tree | 100 × 10 × 20 × 1000 | 8 | 2000 万 | ~1240 万 pts/s | 4.45 ms / 27.03 ms | +| tree | 100 × 100 × 4 × 1000 | 10 | 4000 万 | ~1500–2000 万 pts/s | 31 ms / 156 ms | +| tree | 100 × 100 × 25 × 1000，`--tablets-per-rpc 4` | 10 | 2.5 亿 | **~2100–2250 万 pts/s** | 105 ms / 907 ms | -吞吐量随 tablet 大小（每 RPC 的行 × 传感器数）增长；同等客户端数下 1000 行 tablet 的吞吐约为 100 行的 5 倍。数据为客户端与服务器同机测得——应视为客户端开销的上界，而非服务器容量。 +吞吐量随每 RPC 点数增长：更宽的 tablet（100 传感器 × 1000 行 = 每 tablet 10 万点）加上 `insert_tablets` 多 tablet 批量，可将同一硬件从 ~1200 万提升到 ~2200 万 pts/s 持续吞吐（2.5 亿点，`--reuse-tablets`）。每 RPC 超过约 40 万点、或客户端数超过核数后，吞吐进入平台期且尾延迟上升；峰值运行期间服务器 JVM 写入突发时占 ~3 核、随后阻塞在 memtable 刷盘上，而 Rust 客户端仅占单核 ~30%——瓶颈在同机 Docker 内的服务器（1 GB 堆），不在客户端。数据为客户端与服务器同机测得——应视为客户端开销的上界，而非服务器容量。 ## 项目结构 diff --git a/examples/benchmark.rs b/examples/benchmark.rs index 9c1fe3b..3f44afb 100644 --- a/examples/benchmark.rs +++ b/examples/benchmark.rs @@ -81,6 +81,16 @@ struct BenchConfig { /// Time interval between consecutive points per device (ms), Node.js /// `POINT_STEP`. point_step: i64, + /// When > 0, pre-generate only this many tablets per worker and re-send + /// them cyclically, rebasing timestamps per batch so every insert still + /// lands on a fresh, disjoint time range (values repeat; timestamps + /// don't). Bounds pre-generation memory for very large point counts. + /// 0 = pre-generate everything (default, original behavior). + reuse_tablets: usize, + /// Tree model only: batch this many tablets (different devices) into one + /// `insert_tablets` RPC. 1 = one `insert_tablet` per RPC (default, + /// original behavior). Latency percentiles are then per multi-tablet RPC. + tablets_per_rpc: usize, cleanup: bool, } @@ -108,6 +118,11 @@ OPTIONS: --password <PASS> Password (default: $IOTDB_PASSWORD or root) --base-ts <MS> Base epoch-ms timestamp (default: 1720000000000) --point-step <MS> Interval between points per device (default: 1000) + --reuse-tablets <N> Pre-generate only N tablets per worker and re-send + them with rebased timestamps (bounds memory for + very large runs; 0 = pre-generate all, default) + --tablets-per-rpc <N> Tree model only: send N tablets per RPC via + insert_tablets (default: 1 = insert_tablet) --cleanup Drop the benchmark database after the run --help Print this help @@ -132,6 +147,8 @@ fn parse_args() -> BenchConfig { password: env_or("IOTDB_PASSWORD", "root"), base_ts: 1_720_000_000_000, point_step: 1000, + reuse_tablets: 0, + tablets_per_rpc: 1, cleanup: false, }; @@ -168,6 +185,8 @@ fn parse_args() -> BenchConfig { "--password" => config.password = value(&mut i, flag), "--base-ts" => config.base_ts = parse_num(&value(&mut i, flag), flag) as i64, "--point-step" => config.point_step = parse_num(&value(&mut i, flag), flag) as i64, + "--reuse-tablets" => config.reuse_tablets = parse_num(&value(&mut i, flag), flag), + "--tablets-per-rpc" => config.tablets_per_rpc = parse_num(&value(&mut i, flag), flag), "--cleanup" => config.cleanup = true, "--help" | "-h" => { println!("{USAGE}"); @@ -187,12 +206,17 @@ fn parse_args() -> BenchConfig { ("--batches", config.batches), ("--batch-size", config.batch_size), ("--clients", config.clients), + ("--tablets-per-rpc", config.tablets_per_rpc), ] { if v == 0 { eprintln!("{name} must be positive"); exit(2); } } + if config.tablets_per_rpc > 1 && config.mode == Mode::Table { + eprintln!("--tablets-per-rpc requires --mode tree (insert_tablets is tree-model only)"); + exit(2); + } config } @@ -609,14 +633,26 @@ fn main() -> Result<()> { // batch-major, i.e. round-robin over its devices. println!("[Setup] pre-generating test data..."); let t0 = Instant::now(); - let mut worker_tablets: Vec<Vec<Tablet>> = Vec::with_capacity(config.clients); + // Each worker's full schedule is `batches × its devices` inserts. With + // --reuse-tablets N only the first N tablets are materialized; the worker + // cycles over them and rebases timestamps per iteration (see worker loop). + let mut worker_tablets: Vec<(Vec<Tablet>, usize)> = Vec::with_capacity(config.clients); for w in 0..config.clients { let devices: Vec<usize> = (0..config.devices) .filter(|d| d % config.clients == w) .collect(); - let mut tablets = Vec::with_capacity(devices.len() * config.batches); - for batch in 0..config.batches { + let schedule_len = devices.len() * config.batches; + let materialized = if config.reuse_tablets > 0 { + schedule_len.min(config.reuse_tablets) + } else { + schedule_len + }; + let mut tablets = Vec::with_capacity(materialized); + 'gen: for batch in 0..config.batches { for &device in &devices { + if tablets.len() == materialized { + break 'gen; + } tablets.push(build_tablet( &config, device, @@ -626,12 +662,20 @@ fn main() -> Result<()> { )?); } } - worker_tablets.push(tablets); + worker_tablets.push((tablets, schedule_len)); } println!( - "[Setup] {} tablets generated in {:.2}s", - worker_tablets.iter().map(Vec::len).sum::<usize>(), - t0.elapsed().as_secs_f64() + "[Setup] {} tablets generated in {:.2}s{}", + worker_tablets.iter().map(|(t, _)| t.len()).sum::<usize>(), + t0.elapsed().as_secs_f64(), + if config.reuse_tablets > 0 { + format!( + " (reused cyclically over {} inserts)", + worker_tablets.iter().map(|(_, n)| n).sum::<usize>() + ) + } else { + String::new() + } ); // --- Timed insert phase ------------------------------------------------ @@ -665,21 +709,53 @@ fn main() -> Result<()> { }); let handles: Vec<_> = worker_tablets - .iter() - .map(|tablets| { + .iter_mut() + .map(|(tablets, schedule_len)| { + let schedule_len = *schedule_len; let pool = &pool; let ops_done = &ops_done; let points_done = &points_done; scope.spawn(move || -> Result<WorkerStats> { let mut session = pool.acquire()?; let mut stats = WorkerStats { - latencies_ms: Vec::with_capacity(tablets.len()), + latencies_ms: Vec::with_capacity(schedule_len), ..WorkerStats::default() }; - for tablet in tablets { - let points = (tablet.row_count() * config.sensors) as u64; + let materialized = tablets.len(); + let mut i = 0; + while i < schedule_len { + // A chunk never wraps the materialized ring, so it + // always maps to one contiguous slice. + let idx = i % materialized; + let chunk = config + .tablets_per_rpc + .min(schedule_len - i) + .min(materialized - idx); + 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. + 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; + for (r, ts) in tablet.timestamps_mut().iter_mut().enumerate() { + *ts = window + r as i64 * config.point_step; + } + } + } + let batch = &tablets[idx..idx + chunk]; + let points: u64 = batch + .iter() + .map(|t| (t.row_count() * config.sensors) as u64) + .sum(); let start = Instant::now(); - let outcome = session.insert_tablet(tablet); + 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); @@ -697,6 +773,7 @@ fn main() -> Result<()> { } } } + i += chunk; } Ok(stats) }) diff --git a/src/data/tablet.rs b/src/data/tablet.rs index 0bc3127..177ece4 100644 --- a/src/data/tablet.rs +++ b/src/data/tablet.rs @@ -128,6 +128,14 @@ impl Tablet { &self.timestamps } + /// Mutable access to the timestamp column, e.g. to rebase a reused + /// tablet onto a fresh time range without regenerating its values. + /// Rows are re-sorted on serialization, so callers need not preserve + /// ordering. + pub fn timestamps_mut(&mut self) -> &mut [i64] { + &mut self.timestamps + } + /// `Some` iff this is a table-model tablet. pub fn column_categories(&self) -> Option<&[ColumnCategory]> { self.column_categories.as_deref()
