diff --git a/cmake/options.cmake b/cmake/options.cmake index 4c50c17b1d1f..86ff61486206 100644 --- a/cmake/options.cmake +++ b/cmake/options.cmake @@ -137,6 +137,17 @@ option(BUILD_WITH_RAND_ERR "If build with random error injection" OFF) option(BUILD_TSZ_ENABLED "If build with TSZ compression" ON) option(BUILD_USE_PUBLIC_DEPS "Use public (internet) URLs for all external dependencies instead of internal mirrors" OFF) +# Runtime-pluggable accelerated L2 compression backends. +# When ON (Linux only), taosd can dlopen a user-built drop-in zlib/zstd/lz4 +# (e.g. Intel QAT/IAA, ISA-L) at startup and route the L2 dispatch table +# through it. If the env var is unset or the dlopen/dlsym fails, the +# statically linked stock implementations are used unchanged. +# Activated by: TAOS_COMPRESS_ACCEL= (or per-codec TAOS_COMPRESS_ACCEL_{ZLIB,ZSTD,LZ4}). +# See docs/zh/26-tdinternal/11-compress.md for the symbol contract and ABI requirements. +if(TD_LINUX) + option(BUILD_WITH_ACCEL_COMPRESS "Build runtime-pluggable accelerated L2 compression backends (Linux only)" OFF) +endif() + # When BUILD_RELEASE is ON, force CMAKE_BUILD_TYPE to Release so that # CMake built-in Release flags and ExternalProject configuration align. if(BUILD_RELEASE) diff --git a/docs/zh/26-tdinternal/11-compress.md b/docs/zh/26-tdinternal/11-compress.md index d7b0901313c5..c96f53361780 100644 --- a/docs/zh/26-tdinternal/11-compress.md +++ b/docs/zh/26-tdinternal/11-compress.md @@ -35,6 +35,72 @@ TDengine TSDB 在存储架构上采用了列式存储技术，这意味着在存 TDengine TSDB 支持多种压缩算法，包括 LZ4、ZLIB、ZSTD、XZ 等，用户可以根据具体的应用场景和需求，在压缩率和写入速度之间进行灵活权衡，选择最适合的压缩方案。 +#### 使用硬件加速的二级压缩库（可选） + +二级压缩默认链接到打包在 TDengine 中的静态 zlib/zstd/lz4。如果部署环境提供了硬件加速的 ABI 兼容替代品（例如 Intel QAT/IAA 加速版 zlib、ISA-L 的 libz、ARM 上经过 SVE 优化的 zstd 等），可以让 taosd 在启动时把这些替代品挂入二级压缩 dispatch table，从而在不修改 SQL 的前提下获得加速；如果替代品不可用，自动回退到打包的静态实现。 + +该能力仅在 Linux 平台、并且编译时显式开启 `BUILD_WITH_ACCEL_COMPRESS` 时生效： + +```bash +# 编译 +mkdir build && cd build +cmake -DBUILD_WITH_ACCEL_COMPRESS=ON .. +make -j$(nproc) +``` + +启动时通过环境变量告诉 taosd 从哪里加载替代库： + +| 环境变量 | 取值 | 说明 | +|---|---|---| +| `TAOS_COMPRESS_ACCEL` | 目录路径 / 不设置 | 设为目录则按惯例从 `/libz.so`、`/libzstd.so`、`/liblz4.so` 加载；不设置（或设为空）时使用内置实现 | +| `TAOS_COMPRESS_ACCEL_ZLIB` | `.so` 完整路径 | 单独覆盖 zlib 路径，优先于 `TAOS_COMPRESS_ACCEL` | +| `TAOS_COMPRESS_ACCEL_ZSTD` | `.so` 完整路径 | 同上，覆盖 zstd | +| `TAOS_COMPRESS_ACCEL_LZ4` | `.so` 完整路径 | 同上，覆盖 lz4 | + +**符号约定**：替代库必须导出与上游一致的公共符号，TDengine 启动时会 `dlsym` 它们： + +- libz：`compress2`、`uncompress` +- libzstd：`ZSTD_compress`、`ZSTD_decompress` +- liblz4：`LZ4_compress_default`、`LZ4_decompress_safe` + +ABI 必须与上游相同（参数顺序、返回值语义）。绝大多数硬件加速版本都是 drop-in 替换，无需关心。 + +**失败回退**：任一步失败（环境变量未设、文件不存在、`dlopen` 失败、缺符号），都不会中断 taosd 启动；该 codec 继续使用静态打包的实现，并在日志中输出 `UTL WARN accel : ...`。 + +**确认加载成功**：taosd 启动日志会显示一行类似： + +```text +UTL INFO accel zlib: loaded from /opt/qat-zlib/libz.so, L2_ZLIB dispatch patched +UTL INFO accel zstd: loaded from /opt/qat-zlib/libzstd.so, L2_ZSTD dispatch patched +``` + +如果完全没设环境变量，则会看到： + +```text +UTL INFO accel compression: TAOS_COMPRESS_ACCEL{,_ZLIB,_ZSTD,_LZ4} unset; using stock L2 implementations +``` + +**评估加速效果**：与 `BUILD_TOOLS=ON` 一起编译会得到 `compressBench`，可以直接对二级压缩 dispatch table 做微基准（绕开 SQL、网络、WAL，结果只反映压缩本身）： + +```bash +# Stock 基线（不设 TAOS_COMPRESS_ACCEL） +./build/bin/compressBench --codec all --size 1 --iters 30 --warmup 5 \ + --shape mixed --label stock --csv result.csv + +# 切换到加速库再跑一次，对比同 codec 的 throughput +export TAOS_COMPRESS_ACCEL=/opt/qat-zlib +./build/bin/compressBench --codec all --size 1 --iters 30 --warmup 5 \ + --shape mixed --label accel --csv result.csv +``` + +输出每个 codec 的 mean / p50 / p95 / stdev 以及 MB/s 吞吐和压缩率，并把 `backend=stock|accel` 标在行尾以便对照。`--shape` 提供 `random / repeating / sequential / mixed` 四种数据形态，可分别评估高熵、低熵、单调时间戳和混合负载下的表现；`--size` 支持 0.004（4 KiB）、0.0625（64 KiB）、1（1 MiB）等，覆盖真实列块的常见尺寸。建议至少跑 30 轮 measure + 5 轮 warmup，并在不同空闲时段重复两次以排除瞬时干扰。 + +**注意事项**： + +- 替代库会被 `dlopen` 一次并常驻整个 taosd 生命周期，因此磁盘上别在运行期间替换或删除该文件。 +- 如果替代库本身又依赖其他动态库（例如 QAT 用户态驱动），需要确保它们也能被 `dlopen` 找到——常规手段（`/etc/ld.so.conf.d/`、`LD_LIBRARY_PATH`）都适用。 +- TSZ（浮点有损压缩）和 XZ 不在替换范围内：前者是 TDengine 内部实现，后者使用的是 fast-lzma2 而非主流 xz，没有通用的 drop-in 加速版。 + ### 有损压缩 TDengine TSDB 引擎为浮点数类型数据提供了无损压缩和有损压缩两种模式。浮点数的精度通常由其小数点后的位数决定。在某些情况下，设备采集的浮点数精度较高，但实际应用中关注的精度却较低，此时采用有损压缩可以有效地节约存储空间。TDengine TSDB 的有损压缩算法基于预测模型，其核心思想是利用前序数据点的趋势来预测后续数据点的走势。这种算法能够显著提高压缩率，相比之下，其压缩效果远超无损压缩。有损压缩算法的名称为 TSZ。 diff --git a/source/util/CMakeLists.txt b/source/util/CMakeLists.txt index 85c181eb3cab..e3353f4478a9 100644 --- a/source/util/CMakeLists.txt +++ b/source/util/CMakeLists.txt @@ -23,6 +23,11 @@ DEP_ext_lz4(util) if(TD_LINUX) DEP_ext_xxhash(util) DEP_ext_lzma2(util) + if(BUILD_WITH_ACCEL_COMPRESS) + # tcompression_accel.c needs dlopen/dlsym/dlerror at runtime + target_compile_definitions(util PRIVATE TD_BUILD_WITH_ACCEL_COMPRESS) + target_link_libraries(util PUBLIC dl) + endif() endif() if (NOT ${TD_WINDOWS}) DEP_ext_ssl(util) diff --git a/source/util/src/tcompression.c b/source/util/src/tcompression.c index 79870d79ce03..ecceb7eff594 100644 --- a/source/util/src/tcompression.c +++ b/source/util/src/tcompression.c @@ -334,6 +334,10 @@ static const int32_t TEST_NUMBER = 1; bool lossyFloat = false; bool lossyDouble = false; +// Defined in tcompression_accel.c. Compiled to a no-op stub when +// BUILD_WITH_ACCEL_COMPRESS=OFF, so we can call it unconditionally. +extern void tcompressionAccelInit(void); + // init call void tsCompressInit(char *lossyColumns, float fPrecision, double dPrecision, uint32_t maxIntervals, uint32_t intervals, int32_t ifAdtFse, const char *compressor) { @@ -344,7 +348,11 @@ void tsCompressInit(char *lossyColumns, float fPrecision, double dPrecision, uin tdszInit(fPrecision, dPrecision, maxIntervals, intervals, ifAdtFse, compressor); if (lossyFloat) uTrace("lossy compression float is opened. "); if (lossyDouble) uTrace("lossy compression double is opened. "); - return; + + // Patch compressL2Dict[] with accelerated backends if TAOS_COMPRESS_ACCEL + // points at a drop-in shared object. Failure is non-fatal (logged) and the + // stock implementations remain in the dispatch table. + tcompressionAccelInit(); } // exit call void tsCompressExit() { tdszExit(); } diff --git a/source/util/src/tcompression_accel.c b/source/util/src/tcompression_accel.c new file mode 100644 index 000000000000..3356dc722891 --- /dev/null +++ b/source/util/src/tcompression_accel.c @@ -0,0 +1,328 @@ +/* + * Runtime-pluggable accelerated L2 compression backends. + * + * Stock zlib / zstd / lz4 are statically linked into util at build time and + * remain the default. When BUILD_WITH_ACCEL_COMPRESS=ON and the operator + * points TAOS_COMPRESS_ACCEL{,_ZLIB,_ZSTD,_LZ4} at a drop-in ABI-compatible + * shared object (e.g. an Intel QAT/IAA-accelerated libz, ISA-L's libz, an + * arm64-optimized libzstd), startup dlopens it, resolves the public symbols, + * and rebinds the corresponding entries in compressL2Dict[]. Any failure + * (env unset, file missing, missing symbols) leaves the stock implementation + * in place and is logged. + * + * The whole translation unit compiles to an empty object when the build + * option is off, so callers don't need any #ifdef around tcompressionAccelInit(). + * + * See docs/zh/26-tdinternal/11-compress.md for the operator-facing contract. + */ + +#ifdef TD_BUILD_WITH_ACCEL_COMPRESS + +#ifndef _GNU_SOURCE +#define _GNU_SOURCE +#endif + +#include +#include +#include +#include +#include + +#include "tcompression.h" +#include "tdef.h" +#include "tlog.h" + +// Stock dispatch table lives in tcompression.c +extern TCmprL2FnSet compressL2Dict[]; + +// ---- Resolved function pointers per backend --------------------------------- + +typedef struct { + void *handle; + int (*compress2_fn)(unsigned char *dest, unsigned long *destLen, + const unsigned char *source, unsigned long sourceLen, int level); + int (*uncompress_fn)(unsigned char *dest, unsigned long *destLen, + const unsigned char *source, unsigned long sourceLen); +} SZlibAccel; + +typedef struct { + void *handle; + size_t (*compress_fn)(void *dst, size_t dstCapacity, + const void *src, size_t srcSize, int compressionLevel); + size_t (*decompress_fn)(void *dst, size_t dstCapacity, + const void *src, size_t compressedSize); +} SZstdAccel; + +typedef struct { + void *handle; + int (*compress_default_fn)(const char *src, char *dst, int srcSize, int dstCapacity); + int (*decompress_safe_fn)(const char *src, char *dst, int compressedSize, int dstCapacity); +} SLz4Accel; + +static SZlibAccel g_zlibAccel; +static SZstdAccel g_zstdAccel; +static SLz4Accel g_lz4Accel; + +// ---- Wrappers (signatures match l2*Impl_* in tcompression.c) --------------- +// +// Defensive bounds checks mirror the L2 dispatch contract (callers always pass +// outputSize > inputSize + 1 and compressedSize > 1). Without them, an +// out-of-spec caller or an accel library returning a wrong size that we then +// re-feed here would underflow `outputSize - 1` / `compressedSize - 1` into +// ULONG_MAX / SIZE_MAX and turn the subsequent codec call or fallback memcpy +// into a buffer overrun. The stock implementations in tcompression.c rely on +// the same invariants but skip the checks; this dispatch lives behind dlopen +// and is the more exposed surface, so we harden it here (gemini-code-assist +// security review on PR #35367). + +static int32_t accelCompress_zlib(const char *const input, const int32_t inputSize, char *const output, + int32_t outputSize, const char type, int8_t lvl) { + if (inputSize < 0 || outputSize <= 1) { + return TSDB_CODE_INVALID_PARA; + } + unsigned long dstLen = (unsigned long)(outputSize - 1); + int rc = g_zlibAccel.compress2_fn((unsigned char *)(output + 1), &dstLen, + (const unsigned char *)input, (unsigned long)inputSize, lvl); + if (rc == 0 /* Z_OK */) { + output[0] = 1; + return (int32_t)dstLen + 1; + } + // Fallback path stores the input verbatim, prefixed with a 0 marker byte; + // refuse if the output buffer can't fit it rather than overflow it. + if (outputSize - 1 < inputSize) { + return TSDB_CODE_INVALID_PARA; + } + output[0] = 0; + memcpy(output + 1, input, inputSize); + return inputSize + 1; +} + +static int32_t accelDecompress_zlib(const char *const input, const int32_t compressedSize, char *const output, + int32_t outputSize, const char type) { + if (compressedSize <= 0 || outputSize < 0) { + return TSDB_CODE_INVALID_PARA; + } + if (input[0] == 1) { + unsigned long len = (unsigned long)outputSize; + int rc = g_zlibAccel.uncompress_fn((unsigned char *)output, &len, + (const unsigned char *)input + 1, + (unsigned long)(compressedSize - 1)); + if (rc == 0) return (int32_t)len; + return TSDB_CODE_THIRDPARTY_ERROR; + } else if (input[0] == 0) { + if (outputSize < compressedSize - 1) { + return TSDB_CODE_INVALID_PARA; + } + memcpy(output, input + 1, compressedSize - 1); + return compressedSize - 1; + } + return TSDB_CODE_THIRDPARTY_ERROR; +} + +static int32_t accelCompress_zstd(const char *const input, const int32_t inputSize, char *const output, + int32_t outputSize, const char type, int8_t lvl) { + if (inputSize < 0 || outputSize <= 1) { + return TSDB_CODE_INVALID_PARA; + } + size_t len = g_zstdAccel.compress_fn(output + 1, (size_t)(outputSize - 1), input, (size_t)inputSize, lvl); + // ZSTD_isError() returns a value > srcSize for error codes, so the same + // sentinel as the stock path catches both error and ratio-not-worth-it. + if (len > (size_t)inputSize) { + if (outputSize - 1 < inputSize) { + return TSDB_CODE_INVALID_PARA; + } + output[0] = 0; + memcpy(output + 1, input, inputSize); + return inputSize + 1; + } + output[0] = 1; + return (int32_t)len + 1; +} + +static int32_t accelDecompress_zstd(const char *const input, const int32_t compressedSize, char *const output, + int32_t outputSize, const char type) { + if (compressedSize <= 0 || outputSize < 0) { + return TSDB_CODE_INVALID_PARA; + } + if (input[0] == 1) { + return (int32_t)g_zstdAccel.decompress_fn(output, (size_t)outputSize, input + 1, (size_t)(compressedSize - 1)); + } else if (input[0] == 0) { + if (outputSize < compressedSize - 1) { + return TSDB_CODE_INVALID_PARA; + } + memcpy(output, input + 1, compressedSize - 1); + return compressedSize - 1; + } + return TSDB_CODE_THIRDPARTY_ERROR; +} + +static int32_t accelCompress_lz4(const char *const input, const int32_t inputSize, char *const output, + int32_t outputSize, const char type, int8_t lvl) { + if (inputSize < 0 || outputSize <= 1) { + return TSDB_CODE_INVALID_PARA; + } + const int32_t n = g_lz4Accel.compress_default_fn(input, output + 1, inputSize, outputSize - 1); + if (n <= 0 || n > inputSize) { + if (outputSize - 1 < inputSize) { + return TSDB_CODE_INVALID_PARA; + } + output[0] = 0; + memcpy(output + 1, input, inputSize); + return inputSize + 1; + } + output[0] = 1; + return n + 1; +} + +static int32_t accelDecompress_lz4(const char *const input, const int32_t compressedSize, char *const output, + int32_t outputSize, const char type) { + if (compressedSize <= 0 || outputSize < 0) { + return TSDB_CODE_INVALID_PARA; + } + if (input[0] == 1) { + const int32_t n = g_lz4Accel.decompress_safe_fn(input + 1, output, compressedSize - 1, outputSize); + if (n < 0) { + uError("accel lz4: LZ4_decompress_safe returned %d", n); + return TSDB_CODE_THIRDPARTY_ERROR; + } + return n; + } else if (input[0] == 0) { + if (outputSize < compressedSize - 1) { + return TSDB_CODE_INVALID_PARA; + } + memcpy(output, input + 1, compressedSize - 1); + return compressedSize - 1; + } + return TSDB_CODE_THIRDPARTY_ERROR; +} + +// ---- dlopen + dlsym + dispatch-table patch --------------------------------- + +static void *openLibOrLog(const char *codec, const char *path) { + void *h = dlopen(path, RTLD_NOW | RTLD_LOCAL); + if (h == NULL) { + uWarn("accel %s: dlopen(\"%s\") failed: %s", codec, path, dlerror()); + } + return h; +} + +static void *symOrLog(void *h, const char *codec, const char *sym) { + dlerror(); // clear stale error + void *p = dlsym(h, sym); + const char *err = dlerror(); + if (err != NULL) { + uWarn("accel %s: dlsym(\"%s\") failed: %s", codec, sym, err); + return NULL; + } + return p; +} + +static void tryLoadZlibAccel(const char *path) { + void *h = openLibOrLog("zlib", path); + if (h == NULL) return; + + void *fn_c = symOrLog(h, "zlib", "compress2"); + void *fn_u = symOrLog(h, "zlib", "uncompress"); + if (fn_c == NULL || fn_u == NULL) { + dlclose(h); + return; + } + g_zlibAccel.handle = h; + g_zlibAccel.compress2_fn = (int (*)(unsigned char *, unsigned long *, const unsigned char *, unsigned long, int))fn_c; + g_zlibAccel.uncompress_fn = (int (*)(unsigned char *, unsigned long *, const unsigned char *, unsigned long))fn_u; + compressL2Dict[L2_ZLIB].comprFn = accelCompress_zlib; + compressL2Dict[L2_ZLIB].decomprFn = accelDecompress_zlib; + uInfo("accel zlib: loaded from %s, L2_ZLIB dispatch patched", path); +} + +static void tryLoadZstdAccel(const char *path) { + void *h = openLibOrLog("zstd", path); + if (h == NULL) return; + + void *fn_c = symOrLog(h, "zstd", "ZSTD_compress"); + void *fn_u = symOrLog(h, "zstd", "ZSTD_decompress"); + if (fn_c == NULL || fn_u == NULL) { + dlclose(h); + return; + } + g_zstdAccel.handle = h; + g_zstdAccel.compress_fn = (size_t (*)(void *, size_t, const void *, size_t, int))fn_c; + g_zstdAccel.decompress_fn = (size_t (*)(void *, size_t, const void *, size_t))fn_u; + compressL2Dict[L2_ZSTD].comprFn = accelCompress_zstd; + compressL2Dict[L2_ZSTD].decomprFn = accelDecompress_zstd; + uInfo("accel zstd: loaded from %s, L2_ZSTD dispatch patched", path); +} + +static void tryLoadLz4Accel(const char *path) { + void *h = openLibOrLog("lz4", path); + if (h == NULL) return; + + void *fn_c = symOrLog(h, "lz4", "LZ4_compress_default"); + void *fn_u = symOrLog(h, "lz4", "LZ4_decompress_safe"); + if (fn_c == NULL || fn_u == NULL) { + dlclose(h); + return; + } + g_lz4Accel.handle = h; + g_lz4Accel.compress_default_fn = (int (*)(const char *, char *, int, int))fn_c; + g_lz4Accel.decompress_safe_fn = (int (*)(const char *, char *, int, int))fn_u; + compressL2Dict[L2_LZ4].comprFn = accelCompress_lz4; + compressL2Dict[L2_LZ4].decomprFn = accelDecompress_lz4; + uInfo("accel lz4: loaded from %s, L2_LZ4 dispatch patched", path); +} + +// Resolve "/" without overflowing the destination buffer. +// Returns 0 on success, -1 on overflow (with the buffer left empty). +static int joinDirAndName(char *dst, size_t dstSize, const char *dir, const char *base) { + int n = snprintf(dst, dstSize, "%s/%s", dir, base); + if (n < 0 || (size_t)n >= dstSize) { + dst[0] = '\0'; + return -1; + } + return 0; +} + +// An empty value counts as unset, so operators can disable acceleration with +// FOO= in unit files / container images where deleting a variable is awkward. +static const char *getenvNonEmpty(const char *name) { + const char *v = getenv(name); + return (v != NULL && v[0] != '\0') ? v : NULL; +} + +void tcompressionAccelInit(void) { + const char *dir = getenvNonEmpty("TAOS_COMPRESS_ACCEL"); + const char *zlibLib = getenvNonEmpty("TAOS_COMPRESS_ACCEL_ZLIB"); + const char *zstdLib = getenvNonEmpty("TAOS_COMPRESS_ACCEL_ZSTD"); + const char *lz4Lib = getenvNonEmpty("TAOS_COMPRESS_ACCEL_LZ4"); + + if (dir == NULL && zlibLib == NULL && zstdLib == NULL && lz4Lib == NULL) { + uInfo("accel compression: TAOS_COMPRESS_ACCEL{,_ZLIB,_ZSTD,_LZ4} unset; using stock L2 implementations"); + return; + } + + char buf[1024]; + + if (zlibLib != NULL) { + tryLoadZlibAccel(zlibLib); + } else if (dir != NULL && joinDirAndName(buf, sizeof(buf), dir, "libz.so") == 0) { + tryLoadZlibAccel(buf); + } + + if (zstdLib != NULL) { + tryLoadZstdAccel(zstdLib); + } else if (dir != NULL && joinDirAndName(buf, sizeof(buf), dir, "libzstd.so") == 0) { + tryLoadZstdAccel(buf); + } + + if (lz4Lib != NULL) { + tryLoadLz4Accel(lz4Lib); + } else if (dir != NULL && joinDirAndName(buf, sizeof(buf), dir, "liblz4.so") == 0) { + tryLoadLz4Accel(buf); + } +} + +#else // !TD_BUILD_WITH_ACCEL_COMPRESS — no-op stub so callers stay simple + +void tcompressionAccelInit(void) {} + +#endif // TD_BUILD_WITH_ACCEL_COMPRESS diff --git a/tools/CMakeLists.txt b/tools/CMakeLists.txt index 5041f50044b6..4f44cfbdea3b 100644 --- a/tools/CMakeLists.txt +++ b/tools/CMakeLists.txt @@ -7,3 +7,7 @@ add_subdirectory(shell) IF(TD_LINUX) add_subdirectory(rocks-reader) ENDIF() + +IF(BUILD_TOOLS) + add_subdirectory(compressBench) +ENDIF() diff --git a/tools/compressBench/CMakeLists.txt b/tools/compressBench/CMakeLists.txt new file mode 100644 index 000000000000..38833cef06cf --- /dev/null +++ b/tools/compressBench/CMakeLists.txt @@ -0,0 +1,13 @@ +IF(BUILD_TOOLS) + add_executable(compressBench compressBench.c) + + target_link_libraries(compressBench PRIVATE util m) + + target_include_directories( + compressBench + PRIVATE + ${TD_SOURCE_DIR}/include/util + ${TD_SOURCE_DIR}/include/common + ${TD_SOURCE_DIR}/include/os + ) +ENDIF() diff --git a/tools/compressBench/compressBench.c b/tools/compressBench/compressBench.c new file mode 100644 index 000000000000..67305b1e0b99 --- /dev/null +++ b/tools/compressBench/compressBench.c @@ -0,0 +1,557 @@ +/* + * Copyright (c) 2019 TAOS Data, Inc. + * + * L2 compression layer micro-benchmark. + * + * Measures per-block compress and decompress latency + throughput for each + * codec in the TDengine L2 dispatch table. Supports multiple iterations, + * warmup rounds, and outputs p50/p95/stdev so stock vs. accelerated backends + * can be compared side-by-side. + * + * Build: linked via tools/compressBench/CMakeLists.txt (BUILD_TOOLS=ON). + * Run: compressBench --help + */ + +// Standalone bench: it does not participate in the TDengine memory/file/clock +// accounting layer, so it calls libc malloc/free/fopen/clock_gettime directly. +// The escape hatch must be defined BEFORE any include that pulls in os/. +#define ALLOW_FORBID_FUNC + +#include +#include +#include +#include +#include +#include + +/* Pull in the L2 types and tsCompressInit/tsCompressExit */ +#include "tcompression.h" + +/* ------------------------------------------------------------------------- + * External symbols from tcompression.c + * ---------------------------------------------------------------------- */ +extern TCmprL2FnSet compressL2Dict[]; + +/* Stock compress pointers – used for backend=stock|accel detection. + * On Windows/Darwin tcompression.c does NOT define the zlib/zstd/xz variants; + * compressL2Dict[] there aliases all of them to l2CompressImpl_lz4 instead. + * Only extern what actually exists on the target platform or the link breaks + * with "Undefined symbols _l2CompressImpl_{zlib,zstd,xz}" on macOS/Windows. */ +extern int32_t l2CompressImpl_lz4(const char *const, const int32_t, char *const, int32_t, const char, int8_t); +#if !defined(WINDOWS) && !defined(_TD_DARWIN_64) +extern int32_t l2CompressImpl_zlib(const char *const, const int32_t, char *const, int32_t, const char, int8_t); +extern int32_t l2CompressImpl_zstd(const char *const, const int32_t, char *const, int32_t, const char, int8_t); +extern int32_t l2CompressImpl_xz(const char *const, const int32_t, char *const, int32_t, const char, int8_t); +#endif + +/* tsGetCompressL2Level lives in tcompression.c but has no header prototype */ +extern int8_t tsGetCompressL2Level(uint8_t alg, uint8_t lvl); + +/* ------------------------------------------------------------------------- + * Small helpers + * ---------------------------------------------------------------------- */ + +static double ns_to_ms(double ns) { return ns * 1e-6; } +static double ns_to_mbs(double ns, size_t bytes) { + /* MB/s = bytes / (ns * 1e-9) / 1e6 */ + return (ns > 0.0) ? ((double)bytes / ns * 1e3) : 0.0; +} + +/* xorshift64 – fast PRNG with good entropy, no deps */ +static uint64_t xstate = 42ULL; +static uint64_t xorshift64(void) { + xstate ^= xstate << 13; + xstate ^= xstate >> 7; + xstate ^= xstate << 17; + return xstate; +} + +static int cmp_double(const void *a, const void *b) { + double x = *(const double *)a; + double y = *(const double *)b; + return (x > y) - (x < y); +} + +typedef struct { + double mean; + double p50; + double p95; + double stdev; +} Stats; + +static Stats compute_stats(double *samples, int n) { + Stats s = {0}; + if (n <= 0) return s; + qsort(samples, n, sizeof(double), cmp_double); + double sum = 0.0; + for (int i = 0; i < n; i++) sum += samples[i]; + s.mean = sum / n; + s.p50 = samples[(int)(n * 0.50)]; + s.p95 = samples[(int)(n * 0.95 < n - 1 ? n * 0.95 : n - 1)]; + double var = 0.0; + for (int i = 0; i < n; i++) { + double d = samples[i] - s.mean; + var += d * d; + } + s.stdev = (n > 1) ? sqrt(var / (n - 1)) : 0.0; + return s; +} + +/* ------------------------------------------------------------------------- + * Data generators + * ---------------------------------------------------------------------- */ + +typedef enum { SHAPE_RANDOM = 0, SHAPE_REPEATING, SHAPE_SEQUENTIAL, SHAPE_MIXED } Shape; + +static void gen_random(uint8_t *buf, size_t sz) { + size_t i = 0; + for (; i + 8 <= sz; i += 8) { + uint64_t v = xorshift64(); + memcpy(buf + i, &v, 8); + } + if (i < sz) { + uint64_t v = xorshift64(); + memcpy(buf + i, &v, sz - i); + } +} + +static void gen_repeating(uint8_t *buf, size_t sz) { + /* 64-byte pattern repeated */ + uint8_t pat[64]; + for (int i = 0; i < 64; i++) pat[i] = (uint8_t)(i * 3 + 7); + for (size_t i = 0; i < sz; i++) buf[i] = pat[i % 64]; +} + +static void gen_sequential(uint8_t *buf, size_t sz) { + /* Monotonic int64 timestamps: base + i*1000000 (1ms apart) */ + int64_t ts = 1700000000000000000LL; /* 2023-ish nanoseconds */ + size_t i = 0; + for (; i + 8 <= sz; i += 8, ts += 1000000LL) { + memcpy(buf + i, &ts, 8); + } + if (i < sz) { + memcpy(buf + i, &ts, sz - i); + } +} + +static void gen_mixed(uint8_t *buf, size_t sz) { + size_t third = sz / 3; + size_t rem = sz - 2 * third; + gen_random(buf, third); + gen_repeating(buf + third, third); + gen_sequential(buf + 2 * third, rem); +} + +static void fill_input(uint8_t *buf, size_t sz, Shape shape, uint64_t seed) { + xstate = seed ? seed : 42ULL; + switch (shape) { + case SHAPE_RANDOM: gen_random(buf, sz); break; + case SHAPE_REPEATING: gen_repeating(buf, sz); break; + case SHAPE_SEQUENTIAL:gen_sequential(buf, sz); break; + default: gen_mixed(buf, sz); break; + } +} + +/* ------------------------------------------------------------------------- + * Timing + * ---------------------------------------------------------------------- */ + +static inline double now_ns(void) { + struct timespec ts; + clock_gettime(CLOCK_MONOTONIC, &ts); + return (double)ts.tv_sec * 1e9 + (double)ts.tv_nsec; +} + +/* ------------------------------------------------------------------------- + * CSV helpers + * ---------------------------------------------------------------------- */ + +static void csv_ensure_header(FILE *f) { + /* Write header when file is newly created / empty */ + if (ftell(f) == 0) { + fprintf(f, + "timestamp,label,codec,size_bytes,lvl,shape,iters," + "c_mean_ms,c_p50_ms,c_p95_ms,c_stdev_ms,c_throughput_MBs," + "d_mean_ms,d_p50_ms,d_p95_ms,d_stdev_ms,d_throughput_MBs," + "ratio,backend\n"); + } +} + +/* ------------------------------------------------------------------------- + * Backend detection + * ---------------------------------------------------------------------- */ + +static const char *detect_backend(int codec_idx) { + __data_compress_l2_fn_t fn = compressL2Dict[codec_idx].comprFn; + switch (codec_idx) { + case L2_LZ4: return (fn == l2CompressImpl_lz4) ? "stock" : "accel"; +#if !defined(WINDOWS) && !defined(_TD_DARWIN_64) + case L2_ZLIB: return (fn == l2CompressImpl_zlib) ? "stock" : "accel"; + case L2_ZSTD: return (fn == l2CompressImpl_zstd) ? "stock" : "accel"; + case L2_XZ: return (fn == l2CompressImpl_xz) ? "stock" : "accel"; +#else + // Windows / Darwin alias these three to lz4 in tcompression.c, so the + // stock fn pointer is the lz4 one. accel substitution isn't compiled + // on those platforms anyway (BUILD_WITH_ACCEL_COMPRESS is Linux-only). + case L2_ZLIB: + case L2_ZSTD: + case L2_XZ: return (fn == l2CompressImpl_lz4) ? "stock" : "accel"; +#endif + default: return "auto"; + } +} + +/* ------------------------------------------------------------------------- + * Per-codec bench + * ---------------------------------------------------------------------- */ + +typedef struct { + int codec_idx; + const char *codec_name; + size_t input_size; + int warmup; + int iters; + int8_t level; + const char *shape_name; + const char *label; + const char *csv_path; +} BenchParams; + +static int run_bench(const BenchParams *p, const uint8_t *input, uint8_t *compressed, uint8_t *decompressed) { + size_t out_buf_sz = p->input_size * 2 + 1024; + int total = p->warmup + p->iters; + + double *c_ns = (double *)malloc(sizeof(double) * (size_t)total); + double *d_ns = (double *)malloc(sizeof(double) * (size_t)total); + if (!c_ns || !d_ns) { + fprintf(stderr, "OOM allocating sample arrays\n"); + free(c_ns); + free(d_ns); + return 1; + } + + tsCompressInit("", 0.0f, 0.0, 0, 0, 0, ""); + + int exit_code = 0; + for (int it = 0; it < total && exit_code == 0; it++) { + /* Compress */ + double t0 = now_ns(); + int32_t compressed_sz = compressL2Dict[p->codec_idx].comprFn( + (const char *)input, (int32_t)p->input_size, + (char *)compressed, (int32_t)out_buf_sz, + /* type */ 0, p->level); + double t1 = now_ns(); + + if (compressed_sz <= 0) { + fprintf(stderr, "codec %s compress returned %d at iter %d\n", + p->codec_name, compressed_sz, it); + exit_code = 2; + break; + } + + /* Decompress */ + double t2 = now_ns(); + int32_t decompressed_sz = compressL2Dict[p->codec_idx].decomprFn( + (const char *)compressed, compressed_sz, + (char *)decompressed, (int32_t)p->input_size, + /* type */ 0); + double t3 = now_ns(); + + if (decompressed_sz < 0) { + fprintf(stderr, "codec %s decompress returned %d at iter %d\n", + p->codec_name, decompressed_sz, it); + exit_code = 3; + break; + } + + /* Integrity check */ + if ((size_t)decompressed_sz != p->input_size || + memcmp(input, decompressed, p->input_size) != 0) { + fprintf(stderr, + "INTEGRITY FAIL: codec=%s iter=%d decompressed_sz=%d expected=%zu\n", + p->codec_name, it, decompressed_sz, p->input_size); + exit_code = 4; + break; + } + + c_ns[it] = t1 - t0; + d_ns[it] = t3 - t2; + } + + tsCompressExit(); + + if (exit_code != 0) { + free(c_ns); + free(d_ns); + return exit_code; + } + + /* Skip warmup samples */ + double *c_meas = c_ns + p->warmup; + double *d_meas = d_ns + p->warmup; + int n = p->iters; + + /* Convert to ms for display */ + double *c_ms_arr = (double *)malloc(sizeof(double) * (size_t)n); + double *d_ms_arr = (double *)malloc(sizeof(double) * (size_t)n); + if (!c_ms_arr || !d_ms_arr) { + free(c_ns); free(d_ns); free(c_ms_arr); free(d_ms_arr); + return 1; + } + for (int i = 0; i < n; i++) { + c_ms_arr[i] = ns_to_ms(c_meas[i]); + d_ms_arr[i] = ns_to_ms(d_meas[i]); + } + + Stats cs = compute_stats(c_ms_arr, n); + Stats ds = compute_stats(d_ms_arr, n); + + /* Use mean ns for throughput */ + double c_mean_ns = 0.0, d_mean_ns = 0.0; + for (int i = 0; i < n; i++) { c_mean_ns += c_meas[i]; d_mean_ns += d_meas[i]; } + c_mean_ns /= n; d_mean_ns /= n; + + double c_mbps = ns_to_mbs(c_mean_ns, p->input_size); + double d_mbps = ns_to_mbs(d_mean_ns, p->input_size); + + /* Compute compression ratio from last iteration */ + /* Re-run once outside timing to get stable compressed_sz */ + tsCompressInit("", 0.0f, 0.0, 0, 0, 0, ""); + int32_t last_csz = compressL2Dict[p->codec_idx].comprFn( + (const char *)input, (int32_t)p->input_size, + (char *)compressed, (int32_t)out_buf_sz, + 0, p->level); + tsCompressExit(); + double ratio = (last_csz > 0) ? ((double)p->input_size / (double)last_csz) : 0.0; + + const char *backend = detect_backend(p->codec_idx); + + double size_mib = (double)p->input_size / (1024.0 * 1024.0); + char size_str[32]; + if (size_mib >= 1.0) + snprintf(size_str, sizeof(size_str), "%.0fMiB", size_mib); + else + snprintf(size_str, sizeof(size_str), "%.0fKiB", size_mib * 1024.0); + + /* stdout summary */ + printf("BENCH %-6s size=%-7s lvl=%-6s shape=%-10s iters=%d\n" + " compress mean=%6.2fms p50=%6.2f p95=%6.2f stdev=%5.2f throughput=%7.1fMB/s\n" + " decompress mean=%6.2fms p50=%6.2f p95=%6.2f stdev=%5.2f throughput=%7.1fMB/s\n" + " ratio=%.2fx backend=%s\n\n", + p->codec_name, size_str, "medium", p->shape_name, n, + cs.mean, cs.p50, cs.p95, cs.stdev, c_mbps, + ds.mean, ds.p50, ds.p95, ds.stdev, d_mbps, + ratio, backend); + + /* CSV append */ + if (p->csv_path) { + FILE *cf = fopen(p->csv_path, "a"); + if (cf) { + csv_ensure_header(cf); + time_t now = time(NULL); + fprintf(cf, + "%lld,%s,%s,%zu,%d,%s,%d," + "%.4f,%.4f,%.4f,%.4f,%.2f," + "%.4f,%.4f,%.4f,%.4f,%.2f," + "%.4f,%s\n", + (long long)now, p->label, p->codec_name, p->input_size, + (int)p->level, p->shape_name, n, + cs.mean, cs.p50, cs.p95, cs.stdev, c_mbps, + ds.mean, ds.p50, ds.p95, ds.stdev, d_mbps, + ratio, backend); + fclose(cf); + } else { + fprintf(stderr, "warning: cannot open csv file: %s\n", p->csv_path); + } + } + + free(c_ns); free(d_ns); free(c_ms_arr); free(d_ms_arr); + return 0; +} + +/* ------------------------------------------------------------------------- + * CLI parsing + * ---------------------------------------------------------------------- */ + +static void usage(const char *prog) { + fprintf(stderr, + "Usage: %s [options]\n" + "\n" + " --codec {lz4|zlib|zstd|xz|all} default: all\n" + " --size default: 1 (try 0.004 for 4KiB)\n" + " --iters default: 30\n" + " --warmup default: 5\n" + " --lvl {low|medium|high} default: medium\n" + " --shape {random|repeating|sequential|mixed} default: mixed\n" + " --seed default: 42\n" + " --csv append CSV row(s)\n" + " --label opaque label (default: bench)\n" + "\n", prog); +} + +int main(int argc, char **argv) { + /* Defaults */ + int codec_all = 1; + int codec_idx = L2_LZ4; /* used when codec_all==0 */ + const char *codec_name = "lz4"; + double size_mib = 1.0; + int iters = 30; + int warmup = 5; + uint8_t lvl_enum = L2_LVL_MEDIUM; + Shape shape = SHAPE_MIXED; + const char *shape_name = "mixed"; + uint64_t seed = 42ULL; + const char *csv_path = NULL; + const char *label = "bench"; + + /* Hand-rolled option parser */ + for (int i = 1; i < argc; i++) { + if (strcmp(argv[i], "--help") == 0 || strcmp(argv[i], "-h") == 0) { + usage(argv[0]); + return 0; + } +#define NEED_ARG(flag) \ + do { \ + if (i + 1 >= argc) { \ + fprintf(stderr, "error: %s requires an argument\n", flag); \ + return 1; \ + } \ + i++; \ + } while (0) + + else if (strcmp(argv[i], "--codec") == 0) { + NEED_ARG("--codec"); + const char *v = argv[i]; + if (strcmp(v, "all") == 0) { + codec_all = 1; + } else if (strcmp(v, "lz4") == 0) { + codec_all = 0; codec_idx = L2_LZ4; codec_name = "lz4"; + } else if (strcmp(v, "zlib") == 0) { + codec_all = 0; codec_idx = L2_ZLIB; codec_name = "zlib"; + } else if (strcmp(v, "zstd") == 0) { + codec_all = 0; codec_idx = L2_ZSTD; codec_name = "zstd"; + } else if (strcmp(v, "xz") == 0) { + codec_all = 0; codec_idx = L2_XZ; codec_name = "xz"; + } else { + fprintf(stderr, "error: unknown codec '%s'\n", v); + return 1; + } + } else if (strcmp(argv[i], "--size") == 0) { + NEED_ARG("--size"); + size_mib = atof(argv[i]); + if (size_mib <= 0.0) { fprintf(stderr, "error: --size must be > 0\n"); return 1; } + } else if (strcmp(argv[i], "--iters") == 0) { + NEED_ARG("--iters"); + iters = atoi(argv[i]); + if (iters < 1) { fprintf(stderr, "error: --iters must be >= 1\n"); return 1; } + } else if (strcmp(argv[i], "--warmup") == 0) { + NEED_ARG("--warmup"); + warmup = atoi(argv[i]); + if (warmup < 0) { fprintf(stderr, "error: --warmup must be >= 0\n"); return 1; } + } else if (strcmp(argv[i], "--lvl") == 0) { + NEED_ARG("--lvl"); + const char *v = argv[i]; + if (strcmp(v, "low") == 0) lvl_enum = L2_LVL_LOW; + else if (strcmp(v, "medium") == 0) lvl_enum = L2_LVL_MEDIUM; + else if (strcmp(v, "high") == 0) lvl_enum = L2_LVL_HIGH; + else { fprintf(stderr, "error: unknown level '%s'\n", v); return 1; } + } else if (strcmp(argv[i], "--shape") == 0) { + NEED_ARG("--shape"); + const char *v = argv[i]; + if (strcmp(v, "random") == 0) { shape = SHAPE_RANDOM; shape_name = "random"; } + else if (strcmp(v, "repeating") == 0) { shape = SHAPE_REPEATING; shape_name = "repeating"; } + else if (strcmp(v, "sequential") == 0) { shape = SHAPE_SEQUENTIAL; shape_name = "sequential"; } + else if (strcmp(v, "mixed") == 0) { shape = SHAPE_MIXED; shape_name = "mixed"; } + else { fprintf(stderr, "error: unknown shape '%s'\n", v); return 1; } + } else if (strcmp(argv[i], "--seed") == 0) { + NEED_ARG("--seed"); + seed = (uint64_t)strtoull(argv[i], NULL, 10); + } else if (strcmp(argv[i], "--csv") == 0) { + NEED_ARG("--csv"); + csv_path = argv[i]; + } else if (strcmp(argv[i], "--label") == 0) { + NEED_ARG("--label"); + label = argv[i]; + } else { + fprintf(stderr, "error: unknown option '%s'\n", argv[i]); + usage(argv[0]); + return 1; + } +#undef NEED_ARG + } + + /* Allocate buffers */ + size_t input_size = (size_t)(size_mib * 1024.0 * 1024.0); + if (input_size < 16) input_size = 16; + size_t out_buf_sz = input_size * 2 + 1024; + + uint8_t *input = (uint8_t *)malloc(input_size); + uint8_t *compressed = (uint8_t *)malloc(out_buf_sz); + uint8_t *decompressed= (uint8_t *)malloc(input_size); + if (!input || !compressed || !decompressed) { + fprintf(stderr, "OOM: cannot allocate %.2f MiB benchmark buffers\n", size_mib); + free(input); free(compressed); free(decompressed); + return 1; + } + + fill_input(input, input_size, shape, seed); + + /* Print header */ + double size_mib_act = (double)input_size / (1024.0 * 1024.0); + printf("compressBench: size=%.4f MiB iters=%d warmup=%d seed=%llu label=%s\n\n", + size_mib_act, iters, warmup, (unsigned long long)seed, label); + + /* Codec list for --codec=all */ + typedef struct { int idx; const char *name; } CodecEntry; + static const CodecEntry ALL_CODECS[] = { + {L2_LZ4, "lz4"}, + {L2_ZLIB, "zlib"}, + {L2_ZSTD, "zstd"}, + {L2_XZ, "xz"}, + }; + static const int N_CODECS = (int)(sizeof(ALL_CODECS) / sizeof(ALL_CODECS[0])); + + int ret = 0; + if (codec_all) { + for (int c = 0; c < N_CODECS && ret == 0; c++) { + int8_t level = tsGetCompressL2Level((uint8_t)ALL_CODECS[c].idx, lvl_enum); + BenchParams bp = { + .codec_idx = ALL_CODECS[c].idx, + .codec_name = ALL_CODECS[c].name, + .input_size = input_size, + .warmup = warmup, + .iters = iters, + .level = level, + .shape_name = shape_name, + .label = label, + .csv_path = csv_path, + }; + ret = run_bench(&bp, input, compressed, decompressed); + if (ret != 0) { + fprintf(stderr, "FAIL: codec=%s rc=%d\n", ALL_CODECS[c].name, ret); + } + } + } else { + int8_t level = tsGetCompressL2Level((uint8_t)codec_idx, lvl_enum); + BenchParams bp = { + .codec_idx = codec_idx, + .codec_name = codec_name, + .input_size = input_size, + .warmup = warmup, + .iters = iters, + .level = level, + .shape_name = shape_name, + .label = label, + .csv_path = csv_path, + }; + ret = run_bench(&bp, input, compressed, decompressed); + if (ret != 0) { + fprintf(stderr, "FAIL: codec=%s rc=%d\n", codec_name, ret); + } + } + + free(input); + free(compressed); + free(decompressed); + return ret; +}