You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tsfile.apache.org by co...@apache.org on 2026/09/10 10:59:44 UTC
(tsfile) 10/17: docs(cpp): document and benchmark LabVIEW bulk I/O
This is an automated email from the ASF dual-hosted git repository.
ColinLeeo pushed a commit to branch feat/labview-bulk-io
in repository https://gitbox.apache.org/repos/asf/tsfile.git
commit 2824d3b7559e2b363df61b65b6aae4c65f15271b
Author: ColinLee <sh...@163.com>
AuthorDate: Thu Sep 10 17:17:53 2026 +0800
docs(cpp): document and benchmark LabVIEW bulk I/O
---
cpp/src/labview_wrapper/CMakeLists.txt | 10 +
...223\215\344\275\234\346\211\213\345\206\214.md" | 174 ++++++
cpp/src/labview_wrapper/README.md | 42 +-
cpp/src/labview_wrapper/benchmark_block.py | 178 +++---
cpp/src/labview_wrapper/benchmark_codec_matrix.c | 603 +++++++++++++++++++++
cpp/src/labview_wrapper/benchmark_read_block.py | 256 +++++++++
6 files changed, 1180 insertions(+), 83 deletions(-)
diff --git a/cpp/src/labview_wrapper/CMakeLists.txt b/cpp/src/labview_wrapper/CMakeLists.txt
index 2b79e6744..d984fae18 100644
--- a/cpp/src/labview_wrapper/CMakeLists.txt
+++ b/cpp/src/labview_wrapper/CMakeLists.txt
@@ -63,4 +63,14 @@ if (BUILD_LABVIEW_WRAPPER_TESTS)
C_STANDARD 99
C_STANDARD_REQUIRED ON
RUNTIME_OUTPUT_DIRECTORY ${LIBRARY_OUTPUT_PATH})
+
+ add_executable(tsfile_labview_codec_benchmark benchmark_codec_matrix.c)
+ target_include_directories(tsfile_labview_codec_benchmark
+ PRIVATE ${PROJECT_SRC_DIR})
+ target_link_libraries(tsfile_labview_codec_benchmark
+ PRIVATE tsfile_labview tsfile)
+ set_target_properties(tsfile_labview_codec_benchmark PROPERTIES
+ C_STANDARD 99
+ C_STANDARD_REQUIRED ON
+ RUNTIME_OUTPUT_DIRECTORY ${LIBRARY_OUTPUT_PATH})
endif ()
diff --git "a/cpp/src/labview_wrapper/LabVIEW\350\257\273\345\206\231TsFile\346\223\215\344\275\234\346\211\213\345\206\214.md" "b/cpp/src/labview_wrapper/LabVIEW\350\257\273\345\206\231TsFile\346\223\215\344\275\234\346\211\213\345\206\214.md"
new file mode 100644
index 000000000..5a5762f82
--- /dev/null
+++ "b/cpp/src/labview_wrapper/LabVIEW\350\257\273\345\206\231TsFile\346\223\215\344\275\234\346\211\213\345\206\214.md"
@@ -0,0 +1,174 @@
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+-->
+
+# LabVIEW 读写 TsFile 操作手册
+
+## 1. 适用范围
+
+本 Wrapper 面向 LabVIEW Call Library Function Node（CLFN），对外只使用
+固定宽度整数、C 字符串、Array Data Pointer 和 U64 不透明句柄。旧版逐行
+接口继续兼容；新增的块写入和块读取适合 INT32、FLOAT、DOUBLE 同类型通道。
+混合类型、TAG、字符串或布尔列仍使用 Tablet Builder 与逐行 ResultSet API。
+
+LabVIEW 进程与动态库必须同位数：32 位 LabVIEW 使用 x86 DLL，64 位
+LabVIEW 使用 x64 DLL。`LV_Handle` 即使在 32 位 LabVIEW 中也配置为 U64。
+
+## 2. CLFN 通用映射
+
+| C 类型 | LabVIEW CLFN 配置 |
+|---|---|
+| `LV_Status` / `int32_t` | Signed 32-bit Integer，by value |
+| `LV_Handle` / `uint64_t` | Unsigned 64-bit Integer；输出时 Pointer to Value |
+| `const char*` | String，C String Pointer |
+| `const int64_t*` / `int64_t*` | 1D I64 Array，Array Data Pointer |
+| `const int32_t*` / `int32_t*` | 1D I32 Array，Array Data Pointer |
+| `const float*` / `float*` | 1D SGL Array，Array Data Pointer |
+| `const double*` / `double*` | 1D DBL Array，Array Data Pointer |
+| `uint8_t*` | 1D U8 Array，Array Data Pointer；可选参数可传空指针 |
+| `int32_t* out_rows` | Signed 32-bit Integer，Pointer to Value |
+
+所有可能失败的调用都检查返回状态：`0` 表示成功，其他值来自 TsFile
+`ERRNO`。输入数组只在一次调用期间被借用，DLL 不保存 LabVIEW 数组地址。
+
+## 3. 连续块写入
+
+推荐初始化顺序：
+
+1. `lv_tsfile_set_global_compression` 和
+ `lv_tsfile_set_datatype_encoding`（如果需要覆盖默认 Codec）。
+2. `lv_tsfile_schema_builder_new` 创建 Schema Builder。
+3. 每列调用 `lv_tsfile_schema_builder_add_column`。
+4. `lv_tsfile_writer_open` 获取 Writer U64 Handle。
+5. 释放 Schema Builder。
+6. 每个采集块调用匹配类型的 `lv_tsfile_write_block_*`。
+7. 可在受控时机调用 `lv_tsfile_writer_flush`；它同步执行且 Writer 仍可继续写。
+8. 文件轮转或停止时调用一次 `lv_tsfile_writer_close`，随后把本地 Handle 置零。
+
+块写入参数：
+
+```c
+lv_tsfile_write_block_f64(writer, ts, data, nrows, ncols);
+```
+
+`ts` 有 `nrows` 个 I64 时间戳。`data` 有 `nrows * ncols` 个元素，采用
+row-major 布局：
+
+```text
+data[row * ncols + column]
+```
+
+例如 3 行、2 通道的扁平数组为
+`[r0c0, r0c1, r1c0, r1c1, r2c0, r2c1]`。Wrapper 内部把它拆分为列数据，
+缓存 Tablet 和暂存区；相同或更小的后续批次不会重新创建 Tablet，只有更大
+批次才扩容。建议从每批 1,000～10,000 行开始，用真实采样数据调优。
+
+## 4. 批量读取
+
+首先创建 batch-mode ResultSet：
+
+```c
+lv_tsfile_query_table_batch(reader, table, "x\ny\nz",
+ start_time, end_time, batch_rows, &result_set);
+```
+
+列字符串用实际 LF 字节分隔，不包含时间戳列。然后预分配：
+
+- `out_ts`：至少 `capacity_rows` 个 I64；
+- `out_data`：至少 `capacity_rows * ncols` 个对应数值；
+- `out_is_null`：可选，至少 `capacity_rows * ncols` 个 U8；
+- `out_rows`：一个 I32 输出值。
+
+读取 DOUBLE 示例：
+
+```c
+lv_tsfile_rs_read_block_f64(result_set, out_ts, out_data, out_is_null,
+ capacity_rows, ncols, &out_rows);
+```
+
+`out_data` 同样是 row-major。`out_is_null[index] == 1` 表示 null，此时
+`out_data[index]` 确定性写为零；不关心 null 时可传空指针。返回状态为 0 且
+`out_rows == 0` 表示 EOF，不是错误。
+
+`capacity_rows` 必须大于等于创建查询时的 `batch_rows`。容量不足会在获取
+下一 TsBlock 之前返回错误，因此修正容量后仍能读取原来的下一批。不要在
+batch ResultSet 上调用 `lv_tsfile_rs_next`/标量 Getter，也不要在 row-mode
+ResultSet 上调用块读取函数。批量路径仅支持所选值列全部为同一种 i32、f32
+或 f64；混合类型查询使用旧逐行接口。
+
+读取结束后依次调用 `lv_tsfile_rs_free` 和 `lv_tsfile_reader_close`。
+
+## 5. 为什么通道不多，close 仍可能阻塞
+
+通道数只影响工作量的一部分。`writer_close` 还可能执行：
+
+- 把尚未达到内存阈值的 Page/Chunk 编码并写出；
+- 汇总 Chunk/ChunkGroup 元数据；
+- 构建或序列化文件索引、统计信息、Bloom Filter 和 footer；
+- 触发运行库及文件系统写入，耗时会受磁盘、杀毒软件和系统缓存影响。
+
+因此，即使设备和通道较少，只要单文件累计时间长、页/Chunk 多，或大量数据
+仍留在内存，close 都可能出现数百毫秒尾延迟。显式 flush 可把数据页落盘与
+最终 close 分开观察，但 flush 本身也是同步调用，而且 close 仍需写最终索引。
+
+采集程序使用单一 Writer Owner：
+
+```text
+DAQ 采集循环 ──> 有界 Queue/FIFO ──> TsFile 写入循环
+ ├─ block write
+ ├─ 可控 flush
+ └─ 文件轮转与 close
+```
+
+DAQ 循环只负责采集和入队，不能直接 close。Queue 必须有明确容量、溢出策略
+和水位监控；Writer Handle 只在写入循环使用，避免并发 write/flush/close。
+
+## 6. 性能测量
+
+先执行小规模自检：
+
+```bash
+python3 cpp/src/labview_wrapper/benchmark_block.py --smoke
+python3 cpp/src/labview_wrapper/benchmark_read_block.py --smoke
+```
+
+正式对比：
+
+```bash
+python3 cpp/src/labview_wrapper/benchmark_block.py \
+ --rows 10000 --cols 9 --batches 30 --repeats 3
+python3 cpp/src/labview_wrapper/benchmark_read_block.py \
+ --rows 150000 --cols 9 --batch-rows 10000 --repeats 3
+```
+
+写基准分别报告宿主数组准备、block write、flush 和 close；读基准分别报告
+旧逐行耗时、批量 DLL 解码/复制调用耗时和宿主遍历耗时。基准会校验实际生效
+的 Codec，并验证逐行与批量结果一致。Codec 选择必须用真实设备数据比较：
+合成正弦数据、噪声数据与现场振动波形的压缩率和 CPU 开销可能完全不同。
+
+## 7. 文件与错误处理
+
+- 输出目录必须存在；Writer 使用新文件路径，轮转时生成唯一文件名。
+- Handle 为 0 表示无效。成功 close/free 后立即把 LabVIEW Shift Register
+ 中的 Handle 清零。
+- 输出 Handle 和 `out_rows` 在失败路径保持为零，不能继续使用旧值。
+- 同一 Writer、Reader 或 ResultSet Handle 的调用由调用方串行化。
+- 出错时先停止继续写入，记录状态码；仍持有有效 Writer 时在写入线程中执行
+ 清理，不能让 C++ 异常或资源所有权越过 CLFN 边界。
diff --git a/cpp/src/labview_wrapper/README.md b/cpp/src/labview_wrapper/README.md
index 54089e2fc..b15ebbc5d 100644
--- a/cpp/src/labview_wrapper/README.md
+++ b/cpp/src/labview_wrapper/README.md
@@ -31,6 +31,9 @@ Library Function Node (CLFN).
- Continuous acquisition: open one writer, call
`lv_tsfile_write_block_f64` once per acquisition block, and close the writer
when the file rolls over.
+- Homogeneous numeric reads: create a batch result set with
+ `lv_tsfile_query_table_batch`, then call the matching
+ `lv_tsfile_rs_read_block_i32/f32/f64` function until it returns zero rows.
- Existing mixed-type applications: use the schema, writer, and tablet builder
functions in `tsfile_labview.h`.
@@ -85,10 +88,39 @@ The streaming sequence is:
3. Call `lv_tsfile_write_block_f64` for each block. Configure `writer` as U64
by value, `ts` as 1D I64 Array Data Pointer, and `data` as 1D DBL Array Data
Pointer.
-4. Call `lv_tsfile_writer_close` exactly once.
+4. Optionally call `lv_tsfile_writer_flush` at a controlled point. It is
+ synchronous and leaves the writer open.
+5. Call `lv_tsfile_writer_close` exactly once.
-The block call creates and releases its temporary tablet. The caller retains
-ownership of both input arrays.
+The writer caches and reuses its internal tablet and typed deinterleave buffer;
+only a larger batch grows them. The caller retains ownership of both input
+arrays, and the wrapper never stores their addresses after the call returns.
+
+## Batch reads
+
+Pass only homogeneous `INT32`, `FLOAT`, or `DOUBLE` value columns to
+`lv_tsfile_query_table_batch`. Allocate `batch_rows` I64 timestamps and
+`batch_rows * ncols` typed values, then call the matching block-read function.
+Values use the same row-major indexing as writes. The optional byte-per-cell
+null array uses `0` for valid and `1` for null; null numeric values are written
+as zero. Status `0` with `out_rows == 0` means EOF.
+
+`capacity_rows` must be at least the `batch_rows` used to create the query.
+This is checked before the next TsBlock is consumed. Row-mode scalar getters
+and batch-mode block functions must not be mixed on one result-set handle.
+
+For exact LabVIEW CLFN mappings and acquisition-loop guidance, see
+[LabVIEW读写TsFile操作手册.md](LabVIEW读写TsFile操作手册.md).
+
+## Acquisition and file rotation
+
+`flush` and `close` are synchronous. File size is not the only factor in close
+latency: the writer may still need to encode buffered pages and serialize
+chunk metadata, indexes, Bloom filters, footer data, and filesystem writes.
+Keep acquisition timing isolated with a Producer/Consumer design: the DAQ loop
+enqueues blocks, while one writer loop owns write, flush, rotation, and close.
+Typical starting batch sizes are 1,000 to 10,000 rows; benchmark real device
+data before selecting codecs or rotation thresholds.
## Build and package
@@ -141,4 +173,8 @@ dist\TsFile-LabVIEW-x86\examples\run_block_test.cmd
python cpp/src/labview_wrapper/benchmark_block.py \
--dll /path/to/libtsfile_labview.so \
--rows 10000 --cols 9 --batches 3 --repeats 5
+
+python cpp/src/labview_wrapper/benchmark_read_block.py \
+ --dll /path/to/libtsfile_labview.so \
+ --rows 150000 --cols 9 --batch-rows 10000
```
diff --git a/cpp/src/labview_wrapper/benchmark_block.py b/cpp/src/labview_wrapper/benchmark_block.py
index 3d830749f..a06e9c61e 100644
--- a/cpp/src/labview_wrapper/benchmark_block.py
+++ b/cpp/src/labview_wrapper/benchmark_block.py
@@ -15,7 +15,7 @@
# specific language governing permissions and limitations
# under the License.
-"""Compare per-cell FFI calls with one-call LabVIEW block writes."""
+"""Benchmark legacy per-cell writes against reusable numeric block writes."""
from __future__ import annotations
@@ -24,6 +24,7 @@ import ctypes
import math
import os
import statistics
+import sys
import time
from pathlib import Path
@@ -33,24 +34,22 @@ LV_ENCODING_GORILLA = 8
LV_COMPRESSION_LZ4 = 7
+def default_library() -> Path:
+ name = {
+ "darwin": "libtsfile_labview.dylib",
+ "win32": "tsfile_labview.dll",
+ }.get(sys.platform, "libtsfile_labview.so")
+ return Path(__file__).resolve().parents[2] / "target" / "build" / "lib" / name
+
+
def bind(dll_path: Path) -> ctypes.CDLL:
- dll_directory_handles = []
+ directory_handles = []
if os.name == "nt":
- search_dirs = [dll_path.parent.resolve()]
- search_dirs.extend(
- Path(entry)
- for entry in os.environ.get("PATH", "").split(os.pathsep)
- if entry
- )
- for directory in search_dirs:
- if directory.is_dir():
- dll_directory_handles.append(
- os.add_dll_directory(str(directory.resolve()))
- )
+ directory_handles.append(os.add_dll_directory(str(dll_path.parent.resolve())))
try:
lib = ctypes.CDLL(str(dll_path.resolve()))
finally:
- for handle in dll_directory_handles:
+ for handle in directory_handles:
handle.close()
handle = ctypes.c_uint64
@@ -59,6 +58,9 @@ def bind(dll_path: Path) -> ctypes.CDLL:
lib.lv_tsfile_set_global_compression.restype = status
lib.lv_tsfile_set_datatype_encoding.argtypes = [ctypes.c_uint8, ctypes.c_uint8]
lib.lv_tsfile_set_datatype_encoding.restype = status
+ lib.get_global_compression.restype = ctypes.c_uint8
+ lib.get_datatype_encoding.argtypes = [ctypes.c_uint8]
+ lib.get_datatype_encoding.restype = ctypes.c_uint8
lib.lv_tsfile_schema_builder_new.argtypes = [ctypes.c_char_p]
lib.lv_tsfile_schema_builder_new.restype = handle
lib.lv_tsfile_schema_builder_add_column.argtypes = [
@@ -69,7 +71,6 @@ def bind(dll_path: Path) -> ctypes.CDLL:
]
lib.lv_tsfile_schema_builder_add_column.restype = status
lib.lv_tsfile_schema_builder_free.argtypes = [handle]
-
lib.lv_tsfile_writer_open.argtypes = [
ctypes.c_char_p,
handle,
@@ -87,16 +88,13 @@ def bind(dll_path: Path) -> ctypes.CDLL:
ctypes.c_int32,
]
lib.lv_tsfile_write_block_f64.restype = status
+ lib.lv_tsfile_writer_flush.argtypes = [handle]
+ lib.lv_tsfile_writer_flush.restype = status
lib.lv_tsfile_writer_close.argtypes = [handle]
lib.lv_tsfile_writer_close.restype = status
-
lib.lv_tsfile_tablet_new.argtypes = [ctypes.c_uint32]
lib.lv_tsfile_tablet_new.restype = handle
- lib.lv_tsfile_tablet_add_column.argtypes = [
- handle,
- ctypes.c_char_p,
- ctypes.c_uint8,
- ]
+ lib.lv_tsfile_tablet_add_column.argtypes = [handle, ctypes.c_char_p, ctypes.c_uint8]
lib.lv_tsfile_tablet_add_column.restype = status
lib.lv_tsfile_tablet_finalize_columns.argtypes = [handle]
lib.lv_tsfile_tablet_finalize_columns.restype = status
@@ -122,7 +120,8 @@ def check(status: int, operation: str) -> None:
raise RuntimeError(f"{operation} failed with status {status}")
-def open_writer(lib: ctypes.CDLL, path: Path, columns: list[bytes]) -> ctypes.c_uint64:
+def open_writer(lib: ctypes.CDLL, path: Path, columns: list[bytes]) -> int:
+ path.unlink(missing_ok=True)
builder = lib.lv_tsfile_schema_builder_new(b"vib")
if not builder:
raise RuntimeError("schema builder creation failed")
@@ -134,20 +133,20 @@ def open_writer(lib: ctypes.CDLL, path: Path, columns: list[bytes]) -> ctypes.c_
),
"schema_builder_add_column",
)
- path.unlink(missing_ok=True)
writer = ctypes.c_uint64()
check(
lib.lv_tsfile_writer_open(
- os.fsencode(path), builder, 128 * 1024 * 1024, ctypes.byref(writer)
+ os.fsencode(path), builder, 0, ctypes.byref(writer)
),
"writer_open",
)
- return writer
+ return writer.value
finally:
lib.lv_tsfile_schema_builder_free(builder)
def make_batches(rows: int, cols: int, count: int):
+ started = time.perf_counter()
timestamp_type = ctypes.c_int64 * rows
data_type = ctypes.c_double * (rows * cols)
batches = []
@@ -162,17 +161,10 @@ def make_batches(rows: int, cols: int, count: int):
)
)
batches.append((timestamps, data))
- return batches
+ return batches, time.perf_counter() - started
-def run_cell(
- lib: ctypes.CDLL,
- path: Path,
- columns: list[bytes],
- batches,
- rows: int,
- cols: int,
-) -> float:
+def run_cell(lib, path, columns, batches, rows, cols) -> float:
started = time.perf_counter()
writer = open_writer(lib, path, columns)
for timestamps, data in batches:
@@ -185,76 +177,99 @@ def run_cell(
lib.lv_tsfile_tablet_add_column(tablet, name, LV_TYPE_DOUBLE),
"tablet_add_column",
)
- check(
- lib.lv_tsfile_tablet_finalize_columns(tablet),
- "tablet_finalize_columns",
- )
+ check(lib.lv_tsfile_tablet_finalize_columns(tablet), "tablet_finalize")
for row in range(rows):
check(
lib.lv_tsfile_tablet_set_timestamp(tablet, row, timestamps[row]),
"tablet_set_timestamp",
)
- row_offset = row * cols
for col in range(cols):
check(
lib.lv_tsfile_tablet_set_f64(
- tablet, row, col, data[row_offset + col]
+ tablet, row, col, data[row * cols + col]
),
"tablet_set_f64",
)
check(lib.lv_tsfile_writer_write(writer, tablet), "writer_write")
finally:
lib.lv_tsfile_tablet_free(tablet)
+ check(lib.lv_tsfile_writer_flush(writer), "writer_flush")
check(lib.lv_tsfile_writer_close(writer), "writer_close")
return time.perf_counter() - started
-def run_block(
- lib: ctypes.CDLL,
- path: Path,
- columns: list[bytes],
- batches,
- rows: int,
- cols: int,
-) -> float:
- started = time.perf_counter()
+def run_block(lib, path, columns, batches, rows, cols) -> dict[str, float]:
+ opened = time.perf_counter()
writer = open_writer(lib, path, columns)
+ open_seconds = time.perf_counter() - opened
+
+ started = time.perf_counter()
for timestamps, data in batches:
check(
lib.lv_tsfile_write_block_f64(writer, timestamps, data, rows, cols),
"write_block_f64",
)
+ writer_seconds = time.perf_counter() - started
+
+ started = time.perf_counter()
+ check(lib.lv_tsfile_writer_flush(writer), "writer_flush")
+ flush_seconds = time.perf_counter() - started
+
+ started = time.perf_counter()
check(lib.lv_tsfile_writer_close(writer), "writer_close")
- return time.perf_counter() - started
+ close_seconds = time.perf_counter() - started
+ return {
+ "open_seconds": open_seconds,
+ "writer_seconds": writer_seconds,
+ "flush_seconds": flush_seconds,
+ "close_seconds": close_seconds,
+ }
+
+
+def median(samples: list[dict[str, float]], key: str) -> float:
+ return statistics.median(sample[key] for sample in samples)
def main() -> None:
parser = argparse.ArgumentParser()
- parser.add_argument("--dll", type=Path, required=True)
- parser.add_argument("--rows", type=int, default=2_000)
+ parser.add_argument("--dll", type=Path, default=default_library())
+ parser.add_argument("--rows", type=int, default=10_000)
parser.add_argument("--cols", type=int, default=9)
- parser.add_argument("--batches", type=int, default=3)
+ parser.add_argument("--batches", type=int, default=30)
parser.add_argument("--repeats", type=int, default=3)
parser.add_argument("--output-dir", type=Path, default=Path.cwd())
+ parser.add_argument("--smoke", action="store_true")
args = parser.parse_args()
- if args.rows <= 0 or args.cols <= 0 or args.batches <= 0 or args.repeats <= 0:
+ if args.smoke:
+ args.rows, args.cols, args.batches, args.repeats = 100, 3, 2, 2
+ if min(args.rows, args.cols, args.batches, args.repeats) <= 0:
parser.error("rows, cols, batches, and repeats must be positive")
args.output_dir.mkdir(parents=True, exist_ok=True)
lib = bind(args.dll)
- check(
- lib.lv_tsfile_set_global_compression(LV_COMPRESSION_LZ4),
- "set_global_compression(LZ4)",
- )
+ check(lib.lv_tsfile_set_global_compression(LV_COMPRESSION_LZ4), "set LZ4")
check(
lib.lv_tsfile_set_datatype_encoding(LV_TYPE_DOUBLE, LV_ENCODING_GORILLA),
- "set_datatype_encoding(DOUBLE, GORILLA)",
+ "set GORILLA",
)
- columns = [f"ch_{col}".encode() for col in range(args.cols)]
- batches = make_batches(args.rows, args.cols, args.batches)
+ effective_compression = int(lib.get_global_compression())
+ effective_encoding = int(lib.get_datatype_encoding(LV_TYPE_DOUBLE))
+ if (
+ effective_compression != LV_COMPRESSION_LZ4
+ or effective_encoding != LV_ENCODING_GORILLA
+ ):
+ raise RuntimeError(
+ "effective codec configuration differs from requested values"
+ )
+ columns = [f"ch_{col}".encode() for col in range(args.cols)]
+ batches, materialize_seconds = make_batches(args.rows, args.cols, args.batches)
cell_path = args.output_dir / "lv_cell_benchmark.tsfile"
block_path = args.output_dir / "lv_block_benchmark.tsfile"
+ warmup_path = args.output_dir / "lv_block_benchmark_warmup.tsfile"
+ run_block(lib, warmup_path, columns, batches[:1], args.rows, args.cols)
+ warmup_path.unlink(missing_ok=True)
+
cell_samples = []
block_samples = []
for repeat in range(args.repeats):
@@ -273,32 +288,35 @@ def main() -> None:
run_cell(lib, cell_path, columns, batches, args.rows, args.cols)
)
- cell_seconds = statistics.median(cell_samples)
- block_seconds = statistics.median(block_samples)
+ phases = {
+ "materialize_seconds": materialize_seconds,
+ "writer_seconds": median(block_samples, "writer_seconds"),
+ "flush_seconds": median(block_samples, "flush_seconds"),
+ "close_seconds": median(block_samples, "close_seconds"),
+ }
+ if args.smoke and any(value <= 0 for value in phases.values()):
+ raise RuntimeError(f"non-positive smoke phase: {phases}")
+ cell_seconds = statistics.median(cell_samples)
+ block_seconds = sum(
+ median(block_samples, key)
+ for key in ("open_seconds", "writer_seconds", "flush_seconds", "close_seconds")
+ )
points = args.rows * args.cols * args.batches
- value_bytes = points * ctypes.sizeof(ctypes.c_double)
- cell_calls = args.batches * (args.rows * (args.cols + 1) + args.cols + 4)
- block_calls = args.batches
print(
f"rows={args.rows} cols={args.cols} batches={args.batches} "
- f"repeats={args.repeats} points={points} value_bytes={value_bytes}"
- )
- print("cell samples : " + ", ".join(f"{sample:.6f}s" for sample in cell_samples))
- print("block samples: " + ", ".join(f"{sample:.6f}s" for sample in block_samples))
- print(
- f"cell : {cell_seconds:.6f}s, {points / cell_seconds:,.0f} points/s, "
- f"{value_bytes / cell_seconds / (1024 * 1024):,.2f} MiB/s values, "
- f"~{cell_calls:,} FFI calls in batch loop, "
- f"{cell_path.stat().st_size:,} file bytes"
+ f"repeats={args.repeats} points={points}"
)
print(
- f"block: {block_seconds:.6f}s, {points / block_seconds:,.0f} points/s, "
- f"{value_bytes / block_seconds / (1024 * 1024):,.2f} MiB/s values, "
- f"{block_calls:,} FFI calls in batch loop, "
- f"{block_path.stat().st_size:,} file bytes"
+ f"effective_value_encoding={effective_encoding} "
+ f"effective_compression={effective_compression}"
)
- print(f"speedup: {cell_seconds / block_seconds:.2f}x")
+ for key, value in phases.items():
+ print(f"{key}={value:.9f}")
+ print(f"legacy_cell_seconds={cell_seconds:.9f}")
+ print(f"block_total_seconds={block_seconds:.9f}")
+ print(f"block_speedup={cell_seconds / block_seconds:.2f}x")
+ print(f"output_bytes={block_path.stat().st_size}")
if __name__ == "__main__":
diff --git a/cpp/src/labview_wrapper/benchmark_codec_matrix.c b/cpp/src/labview_wrapper/benchmark_codec_matrix.c
new file mode 100644
index 000000000..818237526
--- /dev/null
+++ b/cpp/src/labview_wrapper/benchmark_codec_matrix.c
@@ -0,0 +1,603 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/*
+ * Reproducible Windows-oriented benchmark for separating TsFile value
+ * encoding, value compression, timestamp encoding/compression, and the
+ * LabVIEW wrapper's block materialization cost.
+ */
+
+#include <inttypes.h>
+#include <stdint.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/stat.h>
+
+#ifdef _WIN32
+/* The TsFile C API names its opaque file handle WriteFile. Rename the Win32
+ * function declaration while including windows.h to avoid a C identifier
+ * collision; this benchmark does not call the Win32 WriteFile function. */
+#define WriteFile Win32WriteFile
+#include <windows.h>
+#undef WriteFile
+#else
+#include <time.h>
+#endif
+
+#include "cwrapper/tsfile_cwrapper.h"
+#include "tsfile_labview.h"
+
+typedef enum BenchmarkMode {
+ MODE_WRAPPER,
+ MODE_CORE,
+} BenchmarkMode;
+
+typedef struct PhaseTimes {
+ double open_seconds;
+ double prepare_seconds;
+ double tablet_new_seconds;
+ double tablet_fill_seconds;
+ double writer_write_seconds;
+ double tablet_free_seconds;
+ double wrapper_block_seconds;
+ double buffer_free_seconds;
+ double flush_seconds;
+ double close_seconds;
+ double write_seconds;
+ double full_seconds;
+ double process_cpu_seconds;
+ uint64_t process_write_bytes;
+} PhaseTimes;
+
+static double monotonic_seconds(void) {
+#ifdef _WIN32
+ LARGE_INTEGER counter;
+ LARGE_INTEGER frequency;
+ QueryPerformanceFrequency(&frequency);
+ QueryPerformanceCounter(&counter);
+ return (double)counter.QuadPart / (double)frequency.QuadPart;
+#else
+ struct timespec value;
+ clock_gettime(CLOCK_MONOTONIC, &value);
+ return (double)value.tv_sec + (double)value.tv_nsec / 1000000000.0;
+#endif
+}
+
+static double process_cpu_seconds(void) {
+#ifdef _WIN32
+ FILETIME creation;
+ FILETIME exit;
+ FILETIME kernel;
+ FILETIME user;
+ ULARGE_INTEGER kernel_value;
+ ULARGE_INTEGER user_value;
+ if (!GetProcessTimes(GetCurrentProcess(), &creation, &exit, &kernel,
+ &user)) {
+ return 0.0;
+ }
+ kernel_value.LowPart = kernel.dwLowDateTime;
+ kernel_value.HighPart = kernel.dwHighDateTime;
+ user_value.LowPart = user.dwLowDateTime;
+ user_value.HighPart = user.dwHighDateTime;
+ return (double)(kernel_value.QuadPart + user_value.QuadPart) / 10000000.0;
+#else
+ return (double)clock() / (double)CLOCKS_PER_SEC;
+#endif
+}
+
+static uint64_t process_write_bytes(void) {
+#ifdef _WIN32
+ IO_COUNTERS counters;
+ if (GetProcessIoCounters(GetCurrentProcess(), &counters)) {
+ return (uint64_t)counters.WriteTransferCount;
+ }
+#endif
+ return 0;
+}
+
+static int check_status(int status, const char* operation) {
+ if (status != 0) {
+ fprintf(stderr, "%s failed with status %d\n", operation, status);
+ return 0;
+ }
+ return 1;
+}
+
+static int parse_mode(const char* text, BenchmarkMode* mode) {
+ if (strcmp(text, "wrapper") == 0) {
+ *mode = MODE_WRAPPER;
+ return 1;
+ }
+ if (strcmp(text, "core") == 0) {
+ *mode = MODE_CORE;
+ return 1;
+ }
+ return 0;
+}
+
+static int parse_encoding(const char* text, uint8_t* encoding) {
+ if (strcmp(text, "plain") == 0) {
+ *encoding = TS_ENCODING_PLAIN;
+ return 1;
+ }
+ if (strcmp(text, "ts2diff") == 0) {
+ *encoding = TS_ENCODING_TS_2DIFF;
+ return 1;
+ }
+ if (strcmp(text, "gorilla") == 0) {
+ *encoding = TS_ENCODING_GORILLA;
+ return 1;
+ }
+ return 0;
+}
+
+static int parse_compression(const char* text, uint8_t* compression) {
+ if (strcmp(text, "uncompressed") == 0) {
+ *compression = TS_COMPRESSION_UNCOMPRESSED;
+ return 1;
+ }
+ if (strcmp(text, "lz4") == 0) {
+ *compression = TS_COMPRESSION_LZ4;
+ return 1;
+ }
+ return 0;
+}
+
+static void free_column_names(char** names, int32_t count) {
+ int32_t i;
+ if (names == NULL) {
+ return;
+ }
+ for (i = 0; i < count; ++i) {
+ free(names[i]);
+ }
+ free(names);
+}
+
+static char** make_column_names(int32_t count) {
+ char** names = (char**)calloc((size_t)count, sizeof(char*));
+ int32_t i;
+ if (names == NULL) {
+ return NULL;
+ }
+ for (i = 0; i < count; ++i) {
+ names[i] = (char*)malloc(32);
+ if (names[i] == NULL) {
+ free_column_names(names, count);
+ return NULL;
+ }
+ if (count == 1) {
+ snprintf(names[i], 32, "channel");
+ } else {
+ snprintf(names[i], 32, "ch_%d", (int)i);
+ }
+ }
+ return names;
+}
+
+static int configure_codecs(uint8_t value_encoding, uint8_t value_compression,
+ uint8_t time_encoding, uint8_t time_compression) {
+ return check_status(
+ set_datatype_encoding(TS_DATATYPE_DOUBLE, value_encoding),
+ "set_datatype_encoding") &&
+ check_status(set_global_compression(value_compression),
+ "set_global_compression") &&
+ check_status(set_global_time_encoding(time_encoding),
+ "set_global_time_encoding") &&
+ check_status(set_global_time_compression(time_compression),
+ "set_global_time_compression");
+}
+
+static LV_Handle wrapper_open(const char* output_path, char** column_names,
+ int32_t ncols) {
+ LV_Handle builder = lv_tsfile_schema_builder_new("vib");
+ LV_Handle writer = 0;
+ int32_t col;
+ if (builder == 0) {
+ return 0;
+ }
+ for (col = 0; col < ncols; ++col) {
+ if (!check_status(
+ lv_tsfile_schema_builder_add_column(
+ builder, column_names[col], LV_TYPE_DOUBLE, LV_CAT_FIELD),
+ "schema_builder_add_column")) {
+ lv_tsfile_schema_builder_free(builder);
+ return 0;
+ }
+ }
+ if (!check_status(lv_tsfile_writer_open(output_path, builder, 0, &writer),
+ "writer_open")) {
+ lv_tsfile_schema_builder_free(builder);
+ return 0;
+ }
+ lv_tsfile_schema_builder_free(builder);
+ return writer;
+}
+
+static TsFileWriter core_open(const char* output_path, char** column_names,
+ int32_t ncols, WriteFile* out_file) {
+ ColumnSchema* columns = NULL;
+ TableSchema schema;
+ TsFileWriter writer = NULL;
+ ERRNO status = 0;
+ int32_t col;
+
+ *out_file = write_file_new(output_path, &status);
+ if (*out_file == NULL || status != 0) {
+ fprintf(stderr, "write_file_new failed with status %d\n", status);
+ return NULL;
+ }
+ columns = (ColumnSchema*)calloc((size_t)ncols, sizeof(ColumnSchema));
+ if (columns == NULL) {
+ free_write_file(out_file);
+ return NULL;
+ }
+ for (col = 0; col < ncols; ++col) {
+ columns[col].column_name = column_names[col];
+ columns[col].data_type = TS_DATATYPE_DOUBLE;
+ columns[col].column_category = FIELD;
+ }
+ schema.table_name = "vib";
+ schema.column_schemas = columns;
+ schema.column_num = ncols;
+ writer = tsfile_writer_new(*out_file, &schema, &status);
+ free(columns);
+ if (writer == NULL || status != 0) {
+ fprintf(stderr, "tsfile_writer_new failed with status %d\n", status);
+ free_write_file(out_file);
+ return NULL;
+ }
+ return writer;
+}
+
+int main(int argc, char** argv) {
+ BenchmarkMode mode;
+ const char* output_path;
+ uint64_t total_value_bytes;
+ uint64_t block_value_bytes;
+ int32_t ncols;
+ const char* raw_block_path;
+ uint8_t value_encoding;
+ uint8_t value_compression;
+ uint8_t time_encoding;
+ uint8_t time_compression;
+ uint64_t values_per_block;
+ int32_t rows_per_block;
+ uint64_t block_count;
+ uint64_t total_rows;
+ double* source_values = NULL;
+ FILE* source = NULL;
+ char** column_names = NULL;
+ TSDataType* column_types = NULL;
+ PhaseTimes phases;
+ LV_Handle wrapper_writer = 0;
+ WriteFile core_file = NULL;
+ TsFileWriter core_writer = NULL;
+ double full_started;
+ double write_started;
+ double cpu_started;
+ uint64_t io_started;
+ uint64_t block;
+ struct stat file_info;
+ int exit_code = 1;
+
+ memset(&phases, 0, sizeof(phases));
+ if (argc != 11) {
+ fprintf(stderr,
+ "usage: %s <wrapper|core> <output.tsfile> "
+ "<total-value-bytes> <block-value-bytes> <ncols> "
+ "<raw-block.bin> <value-encoding> <value-compression> "
+ "<time-encoding> <time-compression>\n",
+ argv[0]);
+ return 2;
+ }
+ if (!parse_mode(argv[1], &mode) ||
+ !parse_encoding(argv[7], &value_encoding) ||
+ !parse_compression(argv[8], &value_compression) ||
+ !parse_encoding(argv[9], &time_encoding) ||
+ !parse_compression(argv[10], &time_compression)) {
+ fprintf(stderr, "invalid mode, encoding, or compression name\n");
+ return 2;
+ }
+ output_path = argv[2];
+ total_value_bytes = strtoull(argv[3], NULL, 10);
+ block_value_bytes = strtoull(argv[4], NULL, 10);
+ ncols = (int32_t)strtol(argv[5], NULL, 10);
+ raw_block_path = argv[6];
+ if (total_value_bytes == 0 || block_value_bytes == 0 || ncols <= 0 ||
+ total_value_bytes % block_value_bytes != 0 ||
+ block_value_bytes % sizeof(double) != 0) {
+ fprintf(stderr, "invalid byte counts or column count\n");
+ return 2;
+ }
+ values_per_block = block_value_bytes / sizeof(double);
+ if (values_per_block % (uint64_t)ncols != 0 ||
+ values_per_block / (uint64_t)ncols > INT32_MAX) {
+ fprintf(stderr, "block size is not divisible into rows and columns\n");
+ return 2;
+ }
+ rows_per_block = (int32_t)(values_per_block / (uint64_t)ncols);
+ block_count = total_value_bytes / block_value_bytes;
+ total_rows = (uint64_t)rows_per_block * block_count;
+ if (total_rows > INT64_MAX) {
+ fprintf(stderr, "timestamp range is too large\n");
+ return 2;
+ }
+
+ source = fopen(raw_block_path, "rb");
+ source_values = (double*)malloc((size_t)block_value_bytes);
+ if (source == NULL || source_values == NULL ||
+ fread(source_values, 1, (size_t)block_value_bytes, source) !=
+ (size_t)block_value_bytes ||
+ fgetc(source) != EOF) {
+ fprintf(stderr, "could not read an exact raw source block\n");
+ if (source != NULL) {
+ fclose(source);
+ }
+ free(source_values);
+ return 2;
+ }
+ fclose(source);
+ source = NULL;
+
+ column_names = make_column_names(ncols);
+ column_types = (TSDataType*)malloc((size_t)ncols * sizeof(TSDataType));
+ if (column_names == NULL || column_types == NULL) {
+ fprintf(stderr, "column allocation failed\n");
+ goto cleanup;
+ }
+ for (int32_t col = 0; col < ncols; ++col) {
+ column_types[col] = TS_DATATYPE_DOUBLE;
+ }
+ if (!configure_codecs(value_encoding, value_compression, time_encoding,
+ time_compression)) {
+ goto cleanup;
+ }
+ if (get_datatype_encoding(TS_DATATYPE_DOUBLE) != value_encoding ||
+ get_global_compression() != value_compression ||
+ get_global_time_encoding() != time_encoding ||
+ get_global_time_compression() != time_compression) {
+ fprintf(stderr, "effective codec configuration differs from request\n");
+ goto cleanup;
+ }
+
+ remove(output_path);
+ full_started = monotonic_seconds();
+ {
+ const double started = monotonic_seconds();
+ if (mode == MODE_WRAPPER) {
+ wrapper_writer = wrapper_open(output_path, column_names, ncols);
+ if (wrapper_writer == 0) {
+ goto cleanup;
+ }
+ } else {
+ core_writer =
+ core_open(output_path, column_names, ncols, &core_file);
+ if (core_writer == NULL) {
+ goto cleanup;
+ }
+ }
+ phases.open_seconds = monotonic_seconds() - started;
+ }
+
+ write_started = monotonic_seconds();
+ cpu_started = process_cpu_seconds();
+ io_started = process_write_bytes();
+ for (block = 0; block < block_count; ++block) {
+ int64_t* timestamps = NULL;
+ double* values = NULL;
+ const int64_t base = (int64_t)(block * (uint64_t)rows_per_block);
+ double started = monotonic_seconds();
+ int32_t row;
+ int32_t col;
+
+ timestamps = (int64_t*)malloc((size_t)rows_per_block * sizeof(int64_t));
+ values = (double*)malloc((size_t)block_value_bytes);
+ if (timestamps == NULL || values == NULL) {
+ free(timestamps);
+ free(values);
+ fprintf(stderr, "block buffer allocation failed\n");
+ goto cleanup;
+ }
+ memcpy(values, source_values, (size_t)block_value_bytes);
+ for (row = 0; row < rows_per_block; ++row) {
+ timestamps[row] = base + row;
+ }
+ phases.prepare_seconds += monotonic_seconds() - started;
+
+ if (mode == MODE_WRAPPER) {
+ started = monotonic_seconds();
+ if (!check_status(
+ lv_tsfile_write_block_f64(wrapper_writer, timestamps,
+ values, rows_per_block, ncols),
+ "lv_tsfile_write_block_f64")) {
+ free(timestamps);
+ free(values);
+ goto cleanup;
+ }
+ phases.wrapper_block_seconds += monotonic_seconds() - started;
+ } else {
+ Tablet tablet;
+ started = monotonic_seconds();
+ tablet = tablet_new(column_names, column_types, (uint32_t)ncols,
+ (uint32_t)rows_per_block);
+ phases.tablet_new_seconds += monotonic_seconds() - started;
+ if (tablet == NULL) {
+ free(timestamps);
+ free(values);
+ fprintf(stderr, "tablet_new failed\n");
+ goto cleanup;
+ }
+
+ started = monotonic_seconds();
+ for (row = 0; row < rows_per_block; ++row) {
+ const size_t offset = (size_t)row * (size_t)ncols;
+ if (!check_status(tablet_add_timestamp(tablet, (uint32_t)row,
+ timestamps[row]),
+ "tablet_add_timestamp")) {
+ free_tablet(&tablet);
+ free(timestamps);
+ free(values);
+ goto cleanup;
+ }
+ for (col = 0; col < ncols; ++col) {
+ if (!check_status(tablet_add_value_by_index_double(
+ tablet, (uint32_t)row, (uint32_t)col,
+ values[offset + (size_t)col]),
+ "tablet_add_value_by_index_double")) {
+ free_tablet(&tablet);
+ free(timestamps);
+ free(values);
+ goto cleanup;
+ }
+ }
+ }
+ phases.tablet_fill_seconds += monotonic_seconds() - started;
+
+ started = monotonic_seconds();
+ if (!check_status(tsfile_writer_write(core_writer, tablet),
+ "tsfile_writer_write")) {
+ free_tablet(&tablet);
+ free(timestamps);
+ free(values);
+ goto cleanup;
+ }
+ phases.writer_write_seconds += monotonic_seconds() - started;
+
+ started = monotonic_seconds();
+ free_tablet(&tablet);
+ phases.tablet_free_seconds += monotonic_seconds() - started;
+ }
+
+ started = monotonic_seconds();
+ free(timestamps);
+ free(values);
+ phases.buffer_free_seconds += monotonic_seconds() - started;
+ }
+ phases.process_cpu_seconds = process_cpu_seconds() - cpu_started;
+ phases.process_write_bytes = process_write_bytes() - io_started;
+ phases.write_seconds = monotonic_seconds() - write_started;
+
+ {
+ const double started = monotonic_seconds();
+ if (mode == MODE_WRAPPER) {
+ if (!check_status(lv_tsfile_writer_flush(wrapper_writer),
+ "lv_tsfile_writer_flush")) {
+ goto cleanup;
+ }
+ } else if (!check_status(tsfile_writer_flush(core_writer),
+ "tsfile_writer_flush")) {
+ goto cleanup;
+ }
+ phases.flush_seconds = monotonic_seconds() - started;
+ }
+
+ {
+ const double started = monotonic_seconds();
+ if (mode == MODE_WRAPPER) {
+ if (!check_status(lv_tsfile_writer_close(wrapper_writer),
+ "lv_tsfile_writer_close")) {
+ wrapper_writer = 0;
+ goto cleanup;
+ }
+ wrapper_writer = 0;
+ } else {
+ if (!check_status(tsfile_writer_close(core_writer),
+ "tsfile_writer_close")) {
+ core_writer = NULL;
+ goto cleanup;
+ }
+ core_writer = NULL;
+ free_write_file(&core_file);
+ }
+ phases.close_seconds = monotonic_seconds() - started;
+ }
+ phases.full_seconds = monotonic_seconds() - full_started;
+
+ if (stat(output_path, &file_info) != 0) {
+ perror("stat");
+ goto cleanup;
+ }
+
+ printf("MODE=%s\n", mode == MODE_WRAPPER ? "wrapper" : "core");
+ printf("VALUE_ENCODING=%s\n", argv[7]);
+ printf("VALUE_COMPRESSION=%s\n", argv[8]);
+ printf("TIME_ENCODING=%s\n", argv[9]);
+ printf("TIME_COMPRESSION=%s\n", argv[10]);
+ printf("EFFECTIVE_VALUE_ENCODING=%u\n",
+ (unsigned)get_datatype_encoding(TS_DATATYPE_DOUBLE));
+ printf("EFFECTIVE_VALUE_COMPRESSION=%u\n",
+ (unsigned)get_global_compression());
+ printf("EFFECTIVE_TIME_ENCODING=%u\n",
+ (unsigned)get_global_time_encoding());
+ printf("EFFECTIVE_TIME_COMPRESSION=%u\n",
+ (unsigned)get_global_time_compression());
+ printf("TOTAL_VALUE_BYTES=%" PRIu64 "\n", total_value_bytes);
+ printf("TOTAL_TIMESTAMP_BYTES=%" PRIu64 "\n", total_rows * sizeof(int64_t));
+ printf("BLOCK_VALUE_BYTES=%" PRIu64 "\n", block_value_bytes);
+ printf("ROWS_PER_BLOCK=%d\n", (int)rows_per_block);
+ printf("COLS=%d\n", (int)ncols);
+ printf("BLOCKS=%" PRIu64 "\n", block_count);
+ printf("TOTAL_ROWS=%" PRIu64 "\n", total_rows);
+ printf("TOTAL_POINTS=%" PRIu64 "\n", total_rows * (uint64_t)ncols);
+ printf("OPEN_SECONDS=%.9f\n", phases.open_seconds);
+ printf("PREPARE_SECONDS=%.9f\n", phases.prepare_seconds);
+ printf("TABLET_NEW_SECONDS=%.9f\n", phases.tablet_new_seconds);
+ printf("TABLET_FILL_SECONDS=%.9f\n", phases.tablet_fill_seconds);
+ printf("WRITER_WRITE_SECONDS=%.9f\n", phases.writer_write_seconds);
+ printf("TABLET_FREE_SECONDS=%.9f\n", phases.tablet_free_seconds);
+ printf("WRAPPER_BLOCK_SECONDS=%.9f\n", phases.wrapper_block_seconds);
+ printf("BUFFER_FREE_SECONDS=%.9f\n", phases.buffer_free_seconds);
+ printf("WRITE_SECONDS=%.9f\n", phases.write_seconds);
+ printf("FLUSH_SECONDS=%.9f\n", phases.flush_seconds);
+ printf("CLOSE_SECONDS=%.9f\n", phases.close_seconds);
+ printf("FULL_SECONDS=%.9f\n", phases.full_seconds);
+ printf("PROCESS_CPU_SECONDS=%.9f\n", phases.process_cpu_seconds);
+ printf("CPU_TO_WALL_RATIO=%.6f\n",
+ phases.write_seconds > 0.0
+ ? phases.process_cpu_seconds / phases.write_seconds
+ : 0.0);
+ printf("PROCESS_WRITE_BYTES=%" PRIu64 "\n", phases.process_write_bytes);
+ printf("VALUE_BYTES_PER_SECOND=%.3f\n",
+ (double)total_value_bytes / phases.write_seconds);
+ printf("POINTS_PER_SECOND=%.3f\n",
+ (double)(total_rows * (uint64_t)ncols) / phases.write_seconds);
+ printf("OUTPUT_BYTES=%lld\n", (long long)file_info.st_size);
+ printf("OUTPUT_TO_VALUE_RATIO=%.9f\n",
+ (double)file_info.st_size / (double)total_value_bytes);
+ printf("OUTPUT_TO_LOGICAL_INPUT_RATIO=%.9f\n",
+ (double)file_info.st_size /
+ (double)(total_value_bytes + total_rows * sizeof(int64_t)));
+ exit_code = 0;
+
+cleanup:
+ if (wrapper_writer != 0) {
+ lv_tsfile_writer_close(wrapper_writer);
+ }
+ if (core_writer != NULL) {
+ tsfile_writer_close(core_writer);
+ }
+ if (core_file != NULL) {
+ free_write_file(&core_file);
+ }
+ free(column_types);
+ free_column_names(column_names, ncols > 0 ? ncols : 0);
+ free(source_values);
+ return exit_code;
+}
diff --git a/cpp/src/labview_wrapper/benchmark_read_block.py b/cpp/src/labview_wrapper/benchmark_read_block.py
new file mode 100644
index 000000000..8fb756e1f
--- /dev/null
+++ b/cpp/src/labview_wrapper/benchmark_read_block.py
@@ -0,0 +1,256 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Compare LabVIEW scalar-row reads with numeric TsBlock reads."""
+
+from __future__ import annotations
+
+import argparse
+import ctypes
+import math
+import os
+import statistics
+import time
+from pathlib import Path
+
+from benchmark_block import bind, check, default_library, open_writer
+
+
+def bind_reader(dll_path: Path) -> ctypes.CDLL:
+ lib = bind(dll_path)
+ handle = ctypes.c_uint64
+ status = ctypes.c_int32
+ lib.lv_tsfile_reader_open.argtypes = [ctypes.c_char_p, ctypes.POINTER(handle)]
+ lib.lv_tsfile_reader_open.restype = status
+ lib.lv_tsfile_reader_close.argtypes = [handle]
+ lib.lv_tsfile_reader_close.restype = status
+ lib.lv_tsfile_query_table.argtypes = [
+ handle,
+ ctypes.c_char_p,
+ ctypes.c_char_p,
+ ctypes.c_int64,
+ ctypes.c_int64,
+ ctypes.POINTER(handle),
+ ]
+ lib.lv_tsfile_query_table.restype = status
+ lib.lv_tsfile_query_table_batch.argtypes = [
+ handle,
+ ctypes.c_char_p,
+ ctypes.c_char_p,
+ ctypes.c_int64,
+ ctypes.c_int64,
+ ctypes.c_int32,
+ ctypes.POINTER(handle),
+ ]
+ lib.lv_tsfile_query_table_batch.restype = status
+ lib.lv_tsfile_rs_next.argtypes = [handle, ctypes.POINTER(status)]
+ lib.lv_tsfile_rs_next.restype = ctypes.c_int32
+ lib.lv_tsfile_rs_get_i64.argtypes = [handle, ctypes.c_uint32]
+ lib.lv_tsfile_rs_get_i64.restype = ctypes.c_int64
+ lib.lv_tsfile_rs_get_f64.argtypes = [handle, ctypes.c_uint32]
+ lib.lv_tsfile_rs_get_f64.restype = ctypes.c_double
+ lib.lv_tsfile_rs_read_block_f64.argtypes = [
+ handle,
+ ctypes.POINTER(ctypes.c_int64),
+ ctypes.POINTER(ctypes.c_double),
+ ctypes.POINTER(ctypes.c_uint8),
+ ctypes.c_int32,
+ ctypes.c_int32,
+ ctypes.POINTER(ctypes.c_int32),
+ ]
+ lib.lv_tsfile_rs_read_block_f64.restype = status
+ lib.lv_tsfile_rs_free.argtypes = [handle]
+ return lib
+
+
+def create_input(lib, path: Path, rows: int, cols: int, columns: list[bytes]) -> None:
+ timestamp_type = ctypes.c_int64 * rows
+ data_type = ctypes.c_double * (rows * cols)
+ timestamps = timestamp_type(*(1000 + row for row in range(rows)))
+ values = data_type(
+ *(
+ math.sin(row * 0.01 + col * 0.1)
+ for row in range(rows)
+ for col in range(cols)
+ )
+ )
+ writer = open_writer(lib, path, columns)
+ check(
+ lib.lv_tsfile_write_block_f64(writer, timestamps, values, rows, cols), "write"
+ )
+ check(lib.lv_tsfile_writer_flush(writer), "flush")
+ check(lib.lv_tsfile_writer_close(writer), "close")
+
+
+def open_reader(lib, path: Path) -> int:
+ reader = ctypes.c_uint64()
+ check(
+ lib.lv_tsfile_reader_open(os.fsencode(path), ctypes.byref(reader)),
+ "reader_open",
+ )
+ return reader.value
+
+
+def read_rows(lib, path: Path, table: bytes, columns: bytes, cols: int):
+ reader = open_reader(lib, path)
+ result_set = ctypes.c_uint64()
+ check(
+ lib.lv_tsfile_query_table(
+ reader, table, columns, 0, (1 << 63) - 1, ctypes.byref(result_set)
+ ),
+ "query_table",
+ )
+ started = time.perf_counter()
+ row_count = 0
+ checksum = 0.0
+ error = ctypes.c_int32()
+ while lib.lv_tsfile_rs_next(result_set, ctypes.byref(error)) == 1:
+ checksum += int(lib.lv_tsfile_rs_get_i64(result_set, 0)) * 1e-12
+ for col in range(cols):
+ checksum += float(lib.lv_tsfile_rs_get_f64(result_set, col + 1))
+ row_count += 1
+ elapsed = time.perf_counter() - started
+ check(error.value, "rs_next")
+ lib.lv_tsfile_rs_free(result_set)
+ check(lib.lv_tsfile_reader_close(reader), "reader_close")
+ return row_count, checksum, elapsed
+
+
+def read_blocks(
+ lib, path: Path, table: bytes, columns: bytes, cols: int, batch_rows: int
+):
+ reader = open_reader(lib, path)
+ result_set = ctypes.c_uint64()
+ check(
+ lib.lv_tsfile_query_table_batch(
+ reader,
+ table,
+ columns,
+ 0,
+ (1 << 63) - 1,
+ batch_rows,
+ ctypes.byref(result_set),
+ ),
+ "query_table_batch",
+ )
+ timestamps = (ctypes.c_int64 * batch_rows)()
+ values = (ctypes.c_double * (batch_rows * cols))()
+ copied = ctypes.c_int32()
+ row_count = 0
+ checksum = 0.0
+ decode_seconds = 0.0
+ host_copy_seconds = 0.0
+ while True:
+ started = time.perf_counter()
+ check(
+ lib.lv_tsfile_rs_read_block_f64(
+ result_set,
+ timestamps,
+ values,
+ None,
+ batch_rows,
+ cols,
+ ctypes.byref(copied),
+ ),
+ "read_block_f64",
+ )
+ decode_seconds += time.perf_counter() - started
+ if copied.value == 0:
+ break
+ started = time.perf_counter()
+ for row in range(copied.value):
+ checksum += int(timestamps[row]) * 1e-12
+ offset = row * cols
+ for col in range(cols):
+ checksum += float(values[offset + col])
+ host_copy_seconds += time.perf_counter() - started
+ row_count += copied.value
+ lib.lv_tsfile_rs_free(result_set)
+ check(lib.lv_tsfile_reader_close(reader), "reader_close")
+ return row_count, checksum, decode_seconds, host_copy_seconds
+
+
+def main() -> None:
+ parser = argparse.ArgumentParser()
+ parser.add_argument("--dll", type=Path, default=default_library())
+ parser.add_argument("--rows", type=int, default=150_000)
+ parser.add_argument("--cols", type=int, default=9)
+ parser.add_argument("--batch-rows", type=int, default=10_000)
+ parser.add_argument("--repeats", type=int, default=3)
+ parser.add_argument("--output-dir", type=Path, default=Path.cwd())
+ parser.add_argument("--smoke", action="store_true")
+ args = parser.parse_args()
+ if args.smoke:
+ args.rows, args.cols, args.batch_rows, args.repeats = 100, 3, 64, 2
+ if min(args.rows, args.cols, args.batch_rows, args.repeats) <= 0:
+ parser.error("rows, cols, batch-rows, and repeats must be positive")
+
+ args.output_dir.mkdir(parents=True, exist_ok=True)
+ path = args.output_dir / "lv_read_benchmark.tsfile"
+ table = b"vib"
+ column_names = [f"ch_{col}".encode() for col in range(args.cols)]
+ columns = b"\n".join(column_names)
+ lib = bind_reader(args.dll)
+ create_input(lib, path, args.rows, args.cols, column_names)
+
+ warm_rows = read_rows(lib, path, table, columns, args.cols)
+ warm_blocks = read_blocks(lib, path, table, columns, args.cols, args.batch_rows)
+ if warm_rows[0] != warm_blocks[0] or not math.isclose(
+ warm_rows[1], warm_blocks[1], rel_tol=1e-12, abs_tol=1e-9
+ ):
+ raise RuntimeError("warm-up row and block results differ")
+
+ legacy_samples = []
+ decode_samples = []
+ copy_samples = []
+ for _ in range(args.repeats):
+ row_result = read_rows(lib, path, table, columns, args.cols)
+ block_result = read_blocks(
+ lib, path, table, columns, args.cols, args.batch_rows
+ )
+ if row_result[0] != args.rows or block_result[0] != args.rows:
+ raise RuntimeError("unexpected result row count")
+ if not math.isclose(
+ row_result[1], block_result[1], rel_tol=1e-12, abs_tol=1e-9
+ ):
+ raise RuntimeError("row and block checksums differ")
+ legacy_samples.append(row_result[2])
+ decode_samples.append(block_result[2])
+ copy_samples.append(block_result[3])
+
+ result = {
+ "legacy_row_seconds": statistics.median(legacy_samples),
+ "batch_decode_seconds": statistics.median(decode_samples),
+ "host_copy_seconds": statistics.median(copy_samples),
+ }
+ if args.smoke and any(value <= 0 for value in result.values()):
+ raise RuntimeError(f"non-positive smoke phase: {result}")
+ batch_total = result["batch_decode_seconds"] + result["host_copy_seconds"]
+ print(
+ f"rows={args.rows} cols={args.cols} batch_rows={args.batch_rows} "
+ f"repeats={args.repeats}"
+ )
+ for key, value in result.items():
+ print(f"{key}={value:.9f}")
+ print(f"batch_total_seconds={batch_total:.9f}")
+ print(f"batch_speedup={result['legacy_row_seconds'] / batch_total:.2f}x")
+ print(f"total_rows={args.rows}")
+ print(f"total_values={args.rows * args.cols}")
+
+
+if __name__ == "__main__":
+ main()