From 1c12f52fa7f0c1df2d082ebf4d383a88c51a168f Mon Sep 17 00:00:00 2001 From: Chris Lalancette Date: Thu, 26 Mar 2026 13:59:20 -0400 Subject: [PATCH] Add cross-language MCAP performance benchmarks. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit I've recently been looking at/thinking about MCAP performance. However, it turns out that we don't currently have a good way to talk about performance since we can't measure it. Fix that by adding a benchmarking/ directory with read and write benchmarks for C++, Rust, Go, Python, and TypeScript. Each language has its own *_bench/ subdirectory. TypeScript supports unchunked/chunked/zstd (LZ4 compression is unavailable in wasm-lz4). Benchmark modes: - Fixed-payload: 1M messages x 100 bytes, single channel - Mixed-payload: 10-second simulated robot recording with 5 channels (/imu 96B@200Hz, /odom 296B@50Hz, /tf 80-1600B@100Hz, /lidar 230KB@10Hz, /camera 512KB@15Hz) — 3750 messages, ~102 MB - Filtered reads: topic filter (/imu), time range (seconds 3-5), and combined topic+time (/lidar seconds 4-6) using mixed files All modes run across 4 compression types (unchunked, chunked, zstd, lz4). Message payloads are sliced from a shared 16 MiB blob generated once by gen_blob.py (deterministic, fixed seed), so every language feeds byte-identical data to its writer and the comparison stays fair by construction. The blob is shaped like sensor data (a triangle wave with noise on ~1 in 4 samples, ~0.43 zstd ratio) so compression work is realistic. Each write bench emits a CRC-32 of its payload stream, and run_bench.sh aborts if the CRCs differ across languages. `make bench` runs the full matrix: all compression modes, fixed + mixed payloads, and filtered reads. Includes peak memory tracking and summary tables via run_bench.sh. Co-Authored-By: Claude Opus 4.6 (1M context) Co-Authored-By: Claude Fable 5 --- .prettierignore | 1 + benchmarking/.gitignore | 5 + benchmarking/Makefile | 50 ++ benchmarking/README.md | 265 ++++++++++ benchmarking/cpp_bench/bench_read.cpp | 101 ++++ benchmarking/cpp_bench/bench_write.cpp | 324 ++++++++++++ benchmarking/cspell.json | 25 + benchmarking/gen_blob.py | 64 +++ benchmarking/go_bench/cmd/bench_read/main.go | 133 +++++ benchmarking/go_bench/cmd/bench_write/main.go | 349 +++++++++++++ benchmarking/go_bench/go.mod | 18 + benchmarking/go_bench/go.sum | 22 + benchmarking/python_bench/bench_read.py | 79 +++ benchmarking/python_bench/bench_write.py | 207 ++++++++ benchmarking/run_bench.sh | 473 ++++++++++++++++++ benchmarking/rust_bench/Cargo.lock | 397 +++++++++++++++ benchmarking/rust_bench/Cargo.toml | 14 + benchmarking/rust_bench/src/bin/bench_read.rs | 137 +++++ .../rust_bench/src/bin/bench_write.rs | 270 ++++++++++ benchmarking/typescript_bench/bench_read.ts | 116 +++++ benchmarking/typescript_bench/bench_write.ts | 273 ++++++++++ 21 files changed, 3323 insertions(+) create mode 100644 benchmarking/.gitignore create mode 100644 benchmarking/Makefile create mode 100644 benchmarking/README.md create mode 100644 benchmarking/cpp_bench/bench_read.cpp create mode 100644 benchmarking/cpp_bench/bench_write.cpp create mode 100644 benchmarking/cspell.json create mode 100644 benchmarking/gen_blob.py create mode 100644 benchmarking/go_bench/cmd/bench_read/main.go create mode 100644 benchmarking/go_bench/cmd/bench_write/main.go create mode 100644 benchmarking/go_bench/go.mod create mode 100644 benchmarking/go_bench/go.sum create mode 100644 benchmarking/python_bench/bench_read.py create mode 100644 benchmarking/python_bench/bench_write.py create mode 100755 benchmarking/run_bench.sh create mode 100644 benchmarking/rust_bench/Cargo.lock create mode 100644 benchmarking/rust_bench/Cargo.toml create mode 100644 benchmarking/rust_bench/src/bin/bench_read.rs create mode 100644 benchmarking/rust_bench/src/bin/bench_write.rs create mode 100644 benchmarking/typescript_bench/bench_read.ts create mode 100644 benchmarking/typescript_bench/bench_write.ts diff --git a/.prettierignore b/.prettierignore index 762810ae2..d7a06d48a 100644 --- a/.prettierignore +++ b/.prettierignore @@ -5,3 +5,4 @@ tests/conformance/data/**/*.json typescript/examples/flatbuffer/output/**/*.ts website/.docusaurus /target +benchmarking/rust_bench/target diff --git a/benchmarking/.gitignore b/benchmarking/.gitignore new file mode 100644 index 000000000..188662988 --- /dev/null +++ b/benchmarking/.gitignore @@ -0,0 +1,5 @@ +cpp_bench/bench_read +cpp_bench/bench_write +go_bench/bench_read +go_bench/bench_write +rust_bench/target/ diff --git a/benchmarking/Makefile b/benchmarking/Makefile new file mode 100644 index 000000000..6c6ccad48 --- /dev/null +++ b/benchmarking/Makefile @@ -0,0 +1,50 @@ +CXX ?= g++ +CXXFLAGS ?= -O2 -Wall -Wextra -Wpedantic -Wconversion \ + -Wundef -Wshadow -Walloca -Wcast-qual \ + -Wdisabled-optimization -Wdouble-promotion -Wfloat-equal \ + -Wformat-signedness -Winit-self -Wmissing-include-dirs -Wmultichar \ + -Wpacked -Wpointer-arith -Wredundant-decls \ + -Wswitch-default -Wwrite-strings \ + -fvisibility=hidden \ + -std=c++17 -I../cpp/mcap/include + +RUST_WRITE = rust_bench/target/release/bench_write +RUST_READ = rust_bench/target/release/bench_read + +GO_WRITE = go_bench/bench_write +GO_READ = go_bench/bench_read + +.PHONY: all bench cpp_bench rust_bench go_bench python_bench typescript_bench clean + +all: cpp_bench rust_bench go_bench python_bench typescript_bench + +cpp_bench: cpp_bench/bench_write cpp_bench/bench_read + +python_bench: + @test -f python_bench/bench_write.py || (echo "ERROR: python_bench/bench_write.py not found" >&2; exit 1) + @test -f python_bench/bench_read.py || (echo "ERROR: python_bench/bench_read.py not found" >&2; exit 1) + @test -f gen_blob.py || (echo "ERROR: gen_blob.py not found" >&2; exit 1) + +cpp_bench/bench_write: cpp_bench/bench_write.cpp + $(CXX) $(CXXFLAGS) cpp_bench/bench_write.cpp -o cpp_bench/bench_write -llz4 -lzstd + +cpp_bench/bench_read: cpp_bench/bench_read.cpp + $(CXX) $(CXXFLAGS) cpp_bench/bench_read.cpp -o cpp_bench/bench_read -llz4 -lzstd + +rust_bench: rust_bench/src/bin/bench_write.rs rust_bench/src/bin/bench_read.rs rust_bench/Cargo.toml + cd rust_bench && cargo build --release + +go_bench: go_bench/cmd/bench_write/main.go go_bench/cmd/bench_read/main.go go_bench/go.mod + cd go_bench && go build -o bench_write ./cmd/bench_write && go build -o bench_read ./cmd/bench_read + +typescript_bench: + @test -f typescript_bench/bench_write.ts || (echo "ERROR: typescript_bench/bench_write.ts not found" >&2; exit 1) + @test -f typescript_bench/bench_read.ts || (echo "ERROR: typescript_bench/bench_read.ts not found" >&2; exit 1) + +bench: all + ./run_bench.sh + +clean: + rm -f cpp_bench/bench_write cpp_bench/bench_read + cd rust_bench && cargo clean + rm -f go_bench/bench_write go_bench/bench_read diff --git a/benchmarking/README.md b/benchmarking/README.md new file mode 100644 index 000000000..cd307398d --- /dev/null +++ b/benchmarking/README.md @@ -0,0 +1,265 @@ +# MCAP Cross-Language Benchmarks + +Read and write benchmarks for the MCAP libraries across five languages: +C++, Rust, Go, Python, and TypeScript. Each language has its own +`*_bench/` subdirectory. + +Three benchmark scenarios are included: + +- **Fixed-payload** — 1M messages with a fixed 100-byte payload on a + single channel, across all compression modes +- **Mixed-payload** — simulated 10-second robot recording with 5 + channels at realistic rates and sizes (3750 messages, ~102 MB) +- **Filtered reads** — topic filter, time range filter, and combined + topic+time filter using the mixed-payload files + +## Directory structure + +``` +benchmarking/ + cpp_bench/ C++ benchmarks (header-only mcap library) + rust_bench/ Rust benchmarks (mcap crate) + go_bench/ Go benchmarks (mcap module) + python_bench/ Python benchmarks (mcap package) + typescript_bench/ TypeScript benchmarks (@mcap/core) + gen_blob.py Generator for the shared payload blob + Makefile Build targets for all languages + run_bench.sh Unified benchmark runner with result tables +``` + +## Dependencies + +### C++ + +- **g++** (or another C++17 compiler) +- **liblz4-dev** — LZ4 compression library +- **libzstd-dev** — Zstandard compression library + +On Debian/Ubuntu: + +``` +sudo apt install g++ liblz4-dev libzstd-dev +``` + +### Rust + +- **cargo** and a Rust toolchain (stable) + +Install via [rustup](https://rustup.rs/): + +``` +curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh +``` + +### Go + +- **go** 1.23 or later + +Install from https://go.dev/dl/ or via your package manager. + +### Python + +- **python3** +- The **mcap** package from this repo (added to `PYTHONPATH` automatically by `run_bench.sh`) + +No additional install is needed; the benchmark script imports from `../python/mcap`. + +### TypeScript + +- **Node.js** (v20.15 or later, for `crc32` in `node:zlib`) +- **npx** (included with Node.js) +- **tsx** (invoked via `npx tsx`; no global install required) +- Node modules must be installed at the repo root (`npm install` from the repo root) + +## Building + +From the `benchmarking/` directory: + +``` +make all +``` + +This will: + +- Compile the C++ benchmarks (`cpp_bench/bench_write`, `cpp_bench/bench_read`) +- Build the Rust benchmarks in release mode (`rust_bench/target/release/`) +- Build the Go benchmarks (`go_bench/bench_write`, `go_bench/bench_read`) +- Verify the Python and TypeScript scripts exist (no compilation needed) + +To build a single language: + +``` +make cpp_bench # C++ only +make rust_bench # Rust only +make go_bench # Go only +``` + +## Running + +### Full benchmark suite + +``` +make bench +``` + +This runs all languages across all compression modes (unchunked, chunked, +zstd, lz4), all three benchmark scenarios (fixed-payload, mixed-payload, +filtered reads), with 5 iterations each. Expect ~10-15 minutes on a +modern machine. + +### Configuration + +The benchmark runner accepts environment variables: + +| Variable | Default | Description | +| -------------------- | --------------------------------- | ----------------------------------------------------------------------- | +| `NUM_MESSAGES` | `1000000` | Number of messages for fixed-payload benchmarks | +| `PAYLOAD_SIZE` | `100` | Message payload size in bytes for fixed-payload benchmarks (max 524288) | +| `BENCH_ITERS` | `5` | Number of iterations per (language, mode) pair | +| `BENCH_DIR` | `/tmp` | Directory for temporary MCAP files and results | +| `BLOB_FILE` | `$BENCH_DIR/bench_fill.bin` | Path of the shared payload blob | +| `MODES` | `unchunked chunked zstd lz4` | Compression modes for fixed-payload benchmarks | +| `MIXED_MODES` | `unchunked chunked zstd lz4` | Compression modes for mixed-payload benchmarks | +| `FILTER_COMPRESSION` | `chunked zstd` | Compression modes for filtered read benchmarks | +| `FILTER_MODES` | `topic timerange topic_timerange` | Filter types to benchmark | + +Example: run a quick benchmark with fewer messages and iterations: + +``` +NUM_MESSAGES=10000 BENCH_ITERS=2 ./run_bench.sh +``` + +### Output + +Results are written to TSV files in `$BENCH_DIR` and summarized in +tables printed to stdout: + +**Fixed-payload benchmarks** (`bench_results.tsv`): + +- File size comparison with compression ratios +- Peak memory usage (write and read) +- Write performance — median/min/max time, messages/sec, MB/sec +- Read performance — median/min/max time, messages/sec, MB/sec + +**Mixed-payload benchmarks** (`bench_mixed_results.tsv`): + +- Write performance — median/min/max time +- Read performance — median/min/max time + +**Filtered read benchmarks** (`bench_filter_results.tsv`): + +- Filtered read performance — median/min/max time per filter type + +Each TSV row has the columns `op lang mode num_msgs payload_size +file_size elapsed_ns wall_sec peak_rss_kb`, plus a tenth column: +`payload_crc32` on write rows (see below) and `msg_count` on read rows. +`run_bench.sh` verifies the message counts: fixed and mixed reads must +equal the number of messages written, and a filtered read returning +zero messages aborts the run. Filtered counts are also compared across +languages; a disagreement is reported as a warning rather than an +error, since time-range boundary semantics may legitimately differ +between library APIs. + +### Timing convention + +The timed region for write benchmarks is: message loop + library +finish/close + flush of user-space buffers, i.e. it ends once all bytes have +been handed to the OS. The file-descriptor close falls outside the timed +region, except in C++ where the library owns the file and closes it inside +`writer.close()`; the extra close syscall is noise at benchmark timescales. + +### Memory measurement convention + +Each bench reports peak RSS in kilobytes, and any platform normalization +happens inside the bench, not in `run_bench.sh`. The native benches (C++, +Rust, Go, Python) read `ru_maxrss`, which is KB on Linux but bytes on macOS, +so they divide by 1024 on macOS. TypeScript's +`process.resourceUsage().maxRSS` is already normalized to KB on all +platforms by libuv. New benches must follow the same convention: emit KB, +normalize at the source. Every write bench holds the 16 MB payload blob +resident, so write RSS numbers include that constant equally across +languages. Unfiltered read benches stream the file rather than buffering +it wholesale, so read RSS reflects the library, not the harness; the one +exception is the Rust filtered-read path, which loads the whole file +because its indexed reader operates on byte slices — filtered results do +not feed the memory table. + +## Payload data + +All write benchmarks draw their message payloads from a single shared +16 MiB blob, `$BENCH_DIR/bench_fill.bin`, generated once by +`gen_blob.py` (deterministic, fixed seed). This guarantees every +language feeds byte-identical data to its writer — the comparison +between implementations stays fair by construction, with no +per-language payload-generation code to keep in sync. + +Message `i`'s payload is a window into the blob: + +``` +offset(i) = (i * 7919) % (16 MiB - 512 KiB) +``` + +where `i` is the global message index in write order (the loop index in +fixed-payload mode, the schedule index in mixed mode). The 7919-byte +stride means consecutive small messages get disjoint windows, while +large payloads (e.g. the 512 KiB camera messages) overlap between +messages — similar to the redundancy between consecutive frames in real +recordings. Since MCAP compresses per chunk, only overlap within a +chunk is visible to the compressor. + +The blob itself is shaped to compress like real sensor data rather than +sitting at either extreme: it is a stream of little-endian 16-bit +"samples" — a slowly-varying triangle wave with noise added to roughly +1 in 4 samples — so exact repeats are common but interrupted, which zstd +compresses at about a 0.43 ratio. Tune `GATE_MASK` / `NOISE_BITS` / +`TRI_PERIOD` in `gen_blob.py` to adjust the ratio, and delete the blob +file to regenerate it. + +To catch any divergence, each write bench computes a CRC-32 of the +exact payload byte stream it hands to the writer (outside the timed +region) and emits it as the tenth TSV column. `run_bench.sh` verifies +the CRC matches across all languages and iterations for each mode, and +aborts on mismatch. + +## Mixed-payload scenario + +The mixed-payload benchmark simulates a 10-second robot recording: + +| Channel | Topic | Payload | Rate | Messages | +| -------- | -------------------- | ----------------------- | ------ | -------- | +| IMU | `/imu` | 96 bytes | 200 Hz | 2000 | +| Odometry | `/odom` | 296 bytes | 50 Hz | 500 | +| TF | `/tf` | 80-1600 bytes (cycling) | 100 Hz | 1000 | +| LiDAR | `/lidar` | 230,400 bytes | 10 Hz | 100 | +| Camera | `/camera/compressed` | 524,288 bytes | 15 Hz | 150 | + +Total: 3750 messages, ~102 MB. Messages are interleaved by timestamp. + +## Filtered read benchmarks + +Filtered reads use the mixed-payload files and test three filter types: + +- **topic** — read only `/imu` messages (2000 of 3750) +- **timerange** — read messages from seconds 3-5 (20% of the recording) +- **topic_timerange** — read `/lidar` messages from seconds 4-6 (~20 messages) + +These benchmarks reveal whether each language's reader uses the MCAP +index to skip irrelevant chunks, or falls back to a linear scan. + +## Notes + +- TypeScript benchmarks skip LZ4 writes because `@foxglove/wasm-lz4` + only provides decompression. TypeScript can still read LZ4-compressed + files. +- The C++ benchmarks link against system lz4/zstd libraries. The Rust + and Go benchmarks use their own compression implementations. +- Python and TypeScript benchmarks are interpreted/JIT and will be + significantly slower than the compiled language benchmarks. + +## Cleaning up + +``` +make clean +``` + +This removes compiled C++ binaries, Rust build artifacts, and Go binaries. diff --git a/benchmarking/cpp_bench/bench_read.cpp b/benchmarking/cpp_bench/bench_read.cpp new file mode 100644 index 000000000..0ca97d3e7 --- /dev/null +++ b/benchmarking/cpp_bench/bench_read.cpp @@ -0,0 +1,101 @@ +#define MCAP_IMPLEMENTATION +#include "mcap/reader.hpp" + +#include +#include +#include +#include +#include +#include +#include + +/* ru_maxrss is KB on Linux but bytes on macOS; normalize to KB. */ +static long peak_rss_kb(void) +{ + struct rusage ru; + getrusage(RUSAGE_SELF, &ru); +#ifdef __APPLE__ + return ru.ru_maxrss / 1024; +#else + return ru.ru_maxrss; +#endif +} + +int main(int argc, char* argv[]) +{ + if (argc < 2 || argc > 6) { + fprintf(stderr, "Usage: %s [mode] [num_messages] [payload_size] [filter]\n", argv[0]); + return 1; + } + + const char* filename = argv[1]; + const char* mode = (argc >= 3) ? argv[2] : "unknown"; + const char* num_messages_str = (argc >= 4) ? argv[3] : "0"; + const char* payload_size_str = (argc >= 5) ? argv[4] : "0"; + const char* filter = (argc >= 6) ? argv[5] : ""; + + struct timespec t_start, t_end; + clock_gettime(CLOCK_MONOTONIC, &t_start); + + mcap::McapReader reader; + auto res = reader.open(filename); + if (!res.ok()) { + fprintf(stderr, "Failed to open %s: %s\n", filename, res.message.c_str()); + return 1; + } + + auto sres = reader.readSummary(mcap::ReadSummaryMethod::AllowFallbackScan); + if (!sres.ok()) { + fprintf(stderr, "Failed to read summary: %s\n", sres.message.c_str()); + reader.close(); + return 1; + } + + mcap::ReadMessageOptions opts; + if (std::strcmp(filter, "topic") == 0) { + opts.topicFilter = [](std::string_view topic) { return topic == "/imu"; }; + } else if (std::strcmp(filter, "timerange") == 0) { + opts.startTime = 3000000000; + opts.endTime = 5000000000; + } else if (std::strcmp(filter, "topic_timerange") == 0) { + opts.topicFilter = [](std::string_view topic) { return topic == "/lidar"; }; + opts.startTime = 4000000000; + opts.endTime = 6000000000; + } + + long msg_count = 0; + auto onProblem = [](const mcap::Status& status) { + fprintf(stderr, "Reader problem: %s\n", status.message.c_str()); + }; + auto messageView = reader.readMessages(onProblem, opts); + for (auto it = messageView.begin(); it != messageView.end(); ++it) { + msg_count++; + /* Touch the data to prevent dead-code elimination */ + if (it->message.dataSize == 0) { + fprintf(stderr, "Empty message\n"); + } + } + + reader.close(); + + clock_gettime(CLOCK_MONOTONIC, &t_end); + + struct stat st; + if (stat(filename, &st) != 0) { + fprintf(stderr, "Failed to stat file\n"); + return 1; + } + long file_size = static_cast(st.st_size); + + long long elapsed_ns = (long long)(t_end.tv_sec - t_start.tv_sec) * 1000000000LL + + (long long)(t_end.tv_nsec - t_start.tv_nsec); + double wall_sec = static_cast(elapsed_ns) / 1e9; + + /* TSV output: op lang mode num_msgs payload_size file_size elapsed_ns wall_sec peak_rss_kb + * msg_count */ + printf("read\tcpp\t%s\t%s\t%s\t%ld\t%lld\t%.6f\t%ld\t%ld\n", + mode, num_messages_str, payload_size_str, file_size, elapsed_ns, wall_sec, peak_rss_kb(), + msg_count); + + return 0; +} diff --git a/benchmarking/cpp_bench/bench_write.cpp b/benchmarking/cpp_bench/bench_write.cpp new file mode 100644 index 000000000..e96ba4c79 --- /dev/null +++ b/benchmarking/cpp_bench/bench_write.cpp @@ -0,0 +1,324 @@ +#define MCAP_IMPLEMENTATION +#include "mcap/writer.hpp" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +/* Shared payload blob parameters; must match gen_blob.py and the other + * language benches. Message i's payload is the window of the blob starting + * at (i * kStride) % kWindowSpan, so all implementations feed identical + * bytes to their writers. */ +static const size_t kBlobSize = 16777216; +static const size_t kMaxPayload = 524288; +static const size_t kWindowSpan = kBlobSize - kMaxPayload; +static const uint64_t kStride = 7919; + +static size_t payload_offset(uint64_t msg_index) +{ + return static_cast((msg_index * kStride) % kWindowSpan); +} + +static bool load_blob(const char* path, std::vector& blob) +{ + FILE* f = fopen(path, "rb"); + if (f == nullptr) { + fprintf(stderr, "Failed to open blob file: %s\n", path); + return false; + } + blob.resize(kBlobSize); + size_t nread = fread(blob.data(), 1, kBlobSize, f); + bool at_eof = (fgetc(f) == EOF); + fclose(f); + if (nread != kBlobSize || !at_eof) { + fprintf(stderr, "Blob file %s is not exactly %zu bytes\n", path, kBlobSize); + return false; + } + return true; +} + +/* CRC-32 (IEEE, zlib-compatible) over the payload stream, used by + * run_bench.sh to verify all languages fed identical bytes. */ +static uint32_t crc32_update(uint32_t crc, const std::byte* data, size_t len) +{ + static uint32_t table[256]; + static bool table_init = false; + if (!table_init) { + for (uint32_t i = 0; i < 256; i++) { + uint32_t c = i; + for (int k = 0; k < 8; k++) { + c = (c & 1) ? 0xEDB88320U ^ (c >> 1) : c >> 1; + } + table[i] = c; + } + table_init = true; + } + crc ^= 0xFFFFFFFFU; + for (size_t i = 0; i < len; i++) { + crc = table[(crc ^ static_cast(data[i])) & 0xFFU] ^ (crc >> 8); + } + return crc ^ 0xFFFFFFFFU; +} + +/* ru_maxrss is KB on Linux but bytes on macOS; normalize to KB. */ +static long peak_rss_kb(void) +{ + struct rusage ru; + getrusage(RUSAGE_SELF, &ru); +#ifdef __APPLE__ + return ru.ru_maxrss / 1024; +#else + return ru.ru_maxrss; +#endif +} + +int main(int argc, char* argv[]) +{ + if (argc != 6) { + fprintf(stderr, "Usage: %s \n", argv[0]); + fprintf(stderr, " mode: unchunked | chunked | zstd | lz4\n"); + return 1; + } + + const char* filename = argv[1]; + const char* mode = argv[2]; + bool mixed_mode = (strcmp(argv[4], "mixed") == 0); + long num_messages = mixed_mode ? 0 : strtol(argv[3], nullptr, 10); + long payload_size = mixed_mode ? 0 : strtol(argv[4], nullptr, 10); + + if (!mixed_mode && (num_messages <= 0 || payload_size <= 0)) { + fprintf(stderr, "num_messages and payload_size must be positive\n"); + return 1; + } + if (!mixed_mode && static_cast(payload_size) > kMaxPayload) { + fprintf(stderr, "payload_size must be <= %zu\n", kMaxPayload); + return 1; + } + + std::vector blob; + if (!load_blob(argv[5], blob)) { + return 1; + } + + mcap::McapWriterOptions opts("bench"); + opts.library = "cpp-bench"; + + if (strcmp(mode, "unchunked") == 0) { + opts.noChunking = true; + opts.compression = mcap::Compression::None; + } else if (strcmp(mode, "chunked") == 0) { + opts.chunkSize = 786432; + opts.compression = mcap::Compression::None; + } else if (strcmp(mode, "zstd") == 0) { + opts.chunkSize = 786432; + opts.compression = mcap::Compression::Zstd; + } else if (strcmp(mode, "lz4") == 0) { + opts.chunkSize = 786432; + opts.compression = mcap::Compression::Lz4; + } else { + fprintf(stderr, "Unknown mode: %s\n", mode); + return 1; + } + + mcap::McapWriter writer; + auto res = writer.open(filename, opts); + if (!res.ok()) { + fprintf(stderr, "Failed to open writer: %s\n", res.message.c_str()); + return 1; + } + + if (mixed_mode) { + /* Mixed payload mode: simulate a 10-second robot recording */ + + /* Channel definitions: topic, schema_name, payload_size(s), period_ns, count */ + struct ChannelDef { + const char* topic; + const char* schema_name; + std::vector payload_sizes; + uint64_t period_ns; + long count; + }; + + ChannelDef channel_defs[] = { + {"/imu", "IMU", {96}, 5000000ULL, 2000}, + {"/odom", "Odometry", {296}, 20000000ULL, 500}, + {"/tf", "TFMessage", {80, 160, 320, 800, 1600}, 10000000ULL, 1000}, + {"/lidar", "PointCloud2", {230400}, 100000000ULL, 100}, + {"/camera/compressed", "CompressedImage", {524288}, 66666667ULL, 150}, + }; + const int num_channels = 5; + + /* Register schemas and channels (not timed) */ + mcap::Schema schemas[5]; + mcap::Channel channels[5]; + for (int c = 0; c < num_channels; c++) { + schemas[c] = mcap::Schema(channel_defs[c].schema_name, "jsonschema", "{\"type\":\"object\"}"); + writer.addSchema(schemas[c]); + channels[c] = mcap::Channel(channel_defs[c].topic, "json", schemas[c].id); + writer.addChannel(channels[c]); + } + + /* Pre-generate sorted message schedule: (timestamp, channel_index) */ + struct ScheduleEntry { + uint64_t timestamp; + int channel_index; + }; + + std::vector schedule; + schedule.reserve(3750); + for (int c = 0; c < num_channels; c++) { + for (long i = 0; i < channel_defs[c].count; i++) { + ScheduleEntry e; + e.timestamp = static_cast(i) * channel_defs[c].period_ns; + e.channel_index = c; + schedule.push_back(e); + } + } + std::sort(schedule.begin(), schedule.end(), [](const ScheduleEntry& a, const ScheduleEntry& b) { + if (a.timestamp != b.timestamp) return a.timestamp < b.timestamp; + return a.channel_index < b.channel_index; + }); + + num_messages = 3750; + + /* Not timed: CRC of the payload stream for cross-language verification */ + uint32_t payload_crc = 0; + { + long crc_seq[5] = {0, 0, 0, 0, 0}; + for (size_t i = 0; i < schedule.size(); i++) { + int c = schedule[i].channel_index; + const auto& cdef = channel_defs[c]; + size_t psize = cdef.payload_sizes[static_cast(crc_seq[c]) % cdef.payload_sizes.size()]; + crc_seq[c]++; + payload_crc = crc32_update(payload_crc, blob.data() + payload_offset(i), psize); + } + } + + /* Time the message-writing loop + close */ + struct timespec t_start, t_end; + clock_gettime(CLOCK_MONOTONIC, &t_start); + + /* Track per-channel sequence numbers for tf cycling */ + long chan_seq[5] = {0, 0, 0, 0, 0}; + + for (size_t i = 0; i < schedule.size(); i++) { + const auto& entry = schedule[i]; + int c = entry.channel_index; + const auto& cdef = channel_defs[c]; + + /* Determine payload size (cycles for /tf) and blob window */ + size_t psize; + if (cdef.payload_sizes.size() == 1) { + psize = cdef.payload_sizes[0]; + } else { + psize = cdef.payload_sizes[static_cast(chan_seq[c]) % cdef.payload_sizes.size()]; + } + + mcap::Message msg; + msg.channelId = channels[c].id; + msg.sequence = static_cast(chan_seq[c]); + msg.logTime = entry.timestamp; + msg.publishTime = entry.timestamp; + msg.data = blob.data() + payload_offset(i); + msg.dataSize = psize; + auto wres = writer.write(msg); + if (!wres.ok()) { + fprintf(stderr, "Failed to write message %zu: %s\n", i, + wres.message.c_str()); + writer.close(); + return 1; + } + + chan_seq[c]++; + } + + writer.close(); + + clock_gettime(CLOCK_MONOTONIC, &t_end); + + struct stat st; + if (stat(filename, &st) != 0) { + fprintf(stderr, "Failed to stat file\n"); + return 1; + } + long file_size = static_cast(st.st_size); + + long long elapsed_ns = (long long)(t_end.tv_sec - t_start.tv_sec) * 1000000000LL + + (long long)(t_end.tv_nsec - t_start.tv_nsec); + double wall_sec = static_cast(elapsed_ns) / 1e9; + + /* TSV output: op lang mode num_msgs payload_size file_size elapsed_ns wall_sec peak_rss_kb payload_crc32 */ + printf("write\tcpp\t%s\t%ld\t%s\t%ld\t%lld\t%.6f\t%ld\t%u\n", + mode, num_messages, "mixed", file_size, elapsed_ns, wall_sec, peak_rss_kb(), + payload_crc); + + } else { + /* Fixed payload mode (original code path) */ + + /* Schema (not timed) */ + mcap::Schema schema("BenchMsg", "jsonschema", "{\"type\":\"object\"}"); + writer.addSchema(schema); + + /* Channel (not timed) */ + mcap::Channel channel("/bench", "json", schema.id); + writer.addChannel(channel); + + /* Not timed: CRC of the payload stream for cross-language verification */ + uint32_t payload_crc = 0; + for (long i = 0; i < num_messages; i++) { + payload_crc = crc32_update(payload_crc, blob.data() + payload_offset(static_cast(i)), + static_cast(payload_size)); + } + + /* Time the message-writing loop + close */ + struct timespec t_start, t_end; + clock_gettime(CLOCK_MONOTONIC, &t_start); + + for (long i = 0; i < num_messages; i++) { + mcap::Message msg; + msg.channelId = channel.id; + msg.sequence = static_cast(i); + msg.logTime = static_cast(i) * 1000; + msg.publishTime = msg.logTime; + msg.data = blob.data() + payload_offset(static_cast(i)); + msg.dataSize = static_cast(payload_size); + auto wres = writer.write(msg); + if (!wres.ok()) { + fprintf(stderr, "Failed to write message %ld: %s\n", i, + wres.message.c_str()); + writer.close(); + return 1; + } + } + + writer.close(); + + clock_gettime(CLOCK_MONOTONIC, &t_end); + + struct stat st; + if (stat(filename, &st) != 0) { + fprintf(stderr, "Failed to stat file\n"); + return 1; + } + long file_size = static_cast(st.st_size); + + long long elapsed_ns = (long long)(t_end.tv_sec - t_start.tv_sec) * 1000000000LL + + (long long)(t_end.tv_nsec - t_start.tv_nsec); + double wall_sec = static_cast(elapsed_ns) / 1e9; + + /* TSV output: op lang mode num_msgs payload_size file_size elapsed_ns wall_sec peak_rss_kb payload_crc32 */ + printf("write\tcpp\t%s\t%ld\t%ld\t%ld\t%lld\t%.6f\t%ld\t%u\n", + mode, num_messages, payload_size, file_size, elapsed_ns, wall_sec, peak_rss_kb(), + payload_crc); + } + + return 0; +} diff --git a/benchmarking/cspell.json b/benchmarking/cspell.json new file mode 100644 index 000000000..84d9f8ba3 --- /dev/null +++ b/benchmarking/cspell.json @@ -0,0 +1,25 @@ +{ + "words": [ + "cdef", + "cmds", + "fsize", + "getrusage", + "libc", + "liblz", + "libuv", + "libzstd", + "maxrss", + "Maxrss", + "Mersenne", + "randbytes", + "Odometry", + "outfile", + "PYPATH", + "PYTHONPATH", + "RUSAGE", + "rustup", + "sres", + "syscall", + "timerange" + ] +} diff --git a/benchmarking/gen_blob.py b/benchmarking/gen_blob.py new file mode 100644 index 000000000..cc89310bd --- /dev/null +++ b/benchmarking/gen_blob.py @@ -0,0 +1,64 @@ +#!/usr/bin/env python3 +"""Generate the shared payload blob used by all benchmark write programs. + +Every language's write benchmark slices its message payloads out of this +one file, so the input data is byte-identical across implementations by +construction. The blob is deterministic (fixed seed), so results are +reproducible across runs and machines. + +The data is shaped to compress like real robot sensor data rather than +sitting at either extreme (a repeating pattern compresses to nearly +nothing; pure random data doesn't compress at all). It is a stream of +little-endian 16-bit "sensor samples": a slowly-varying triangle wave +with random noise added to roughly 1 in 4 samples, which lands around a +2.5x zstd compression ratio. Adjust GATE_MASK (noise on 1 in +(GATE_MASK + 1) samples), NOISE_BITS, or TRI_PERIOD to tune the ratio. +""" + +import hashlib +import random +import sys + +BLOB_SIZE = 16 * 1024 * 1024 +NUM_SAMPLES = BLOB_SIZE // 2 +TRI_PERIOD = 4096 # samples per triangle-wave cycle +NOISE_BITS = 8 # noise amplitude: 0..(2^NOISE_BITS - 1) +GATE_MASK = 0x03 # noise hits samples where gate byte & GATE_MASK == 0 +SEED = 1337 + + +def main(): + if len(sys.argv) != 2: + print(f"Usage: {sys.argv[0]} ", file=sys.stderr) + return 1 + + # Triangle wave lookup table: rises 0..2048, falls back to 0, scaled + # so samples span 0..8192 (comfortably within 16 bits after noise). + half = TRI_PERIOD // 2 + tri = [(k if k < half else TRI_PERIOD - k) * 4 for k in range(TRI_PERIOD)] + + # random.Random is a seeded Mersenne Twister; CPython guarantees the + # same seed yields the same sequence across versions and platforms. + rng = random.Random(SEED) + noise = rng.randbytes(NUM_SAMPLES) + gate = rng.randbytes(NUM_SAMPLES) + noise_mask = (1 << NOISE_BITS) - 1 + + out = bytearray(BLOB_SIZE) + for k in range(NUM_SAMPLES): + s = tri[k % TRI_PERIOD] + if gate[k] & GATE_MASK == 0: + s += noise[k] & noise_mask + out[2 * k] = s & 0xFF + out[2 * k + 1] = (s >> 8) & 0xFF + + with open(sys.argv[1], "wb") as f: + f.write(out) + + digest = hashlib.sha256(out).hexdigest() + print(f"wrote {sys.argv[1]}: {BLOB_SIZE} bytes, sha256 {digest}") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/benchmarking/go_bench/cmd/bench_read/main.go b/benchmarking/go_bench/cmd/bench_read/main.go new file mode 100644 index 000000000..7effd9bbc --- /dev/null +++ b/benchmarking/go_bench/cmd/bench_read/main.go @@ -0,0 +1,133 @@ +package main + +import ( + "fmt" + "io" + "os" + "runtime" + "syscall" + "time" + + "github.com/foxglove/mcap/go/mcap" +) + +func run() error { + if len(os.Args) < 2 || len(os.Args) > 6 { + return fmt.Errorf("Usage: %s [mode] [num_messages] [payload_size] [filter]", os.Args[0]) + } + + filename := os.Args[1] + mode := "unknown" + numMessagesStr := "0" + payloadSizeStr := "0" + filter := "" + if len(os.Args) >= 3 { + mode = os.Args[2] + } + if len(os.Args) >= 4 { + numMessagesStr = os.Args[3] + } + if len(os.Args) >= 5 { + payloadSizeStr = os.Args[4] + } + if len(os.Args) >= 6 { + filter = os.Args[5] + } + + // Timed: file open + message iteration + start := time.Now() + + f, err := os.Open(filename) + if err != nil { + return fmt.Errorf("failed to open file: %w", err) + } + defer f.Close() + + reader, err := mcap.NewReader(f) + if err != nil { + return fmt.Errorf("failed to create reader: %w", err) + } + defer reader.Close() + + var opts []mcap.ReadOpt + switch filter { + case "": + // no filter — read all messages + case "topic": + opts = append(opts, + mcap.WithTopics([]string{"/imu"}), + mcap.UsingIndex(true), + ) + case "timerange": + opts = append(opts, + mcap.AfterNanos(3000000000), + mcap.BeforeNanos(5000000000), + mcap.UsingIndex(true), + ) + case "topic_timerange": + opts = append(opts, + mcap.WithTopics([]string{"/lidar"}), + mcap.AfterNanos(4000000000), + mcap.BeforeNanos(6000000000), + mcap.UsingIndex(true), + ) + default: + return fmt.Errorf("unknown filter mode: %s (expected topic, timerange, or topic_timerange)", filter) + } + + it, err := reader.Messages(opts...) + if err != nil { + return fmt.Errorf("failed to create message iterator: %w", err) + } + + msgCount := int64(0) + msg := &mcap.Message{} + for { + _, _, _, err = it.NextInto(msg) + if err != nil { + if err == io.EOF { + break + } + return fmt.Errorf("failed to read message: %w", err) + } + // Touch data to prevent dead-code elimination + if len(msg.Data) == 0 { + fmt.Fprintf(os.Stderr, "Empty message\n") + } + msgCount++ + } + + elapsed := time.Since(start) + + fi, err := f.Stat() + if err != nil { + return fmt.Errorf("failed to stat file: %w", err) + } + fileSize := fi.Size() + + // TSV output: op lang mode num_msgs payload_size file_size elapsed_ns wall_sec peak_rss_kb msg_count + fmt.Printf("read\tgo\t%s\t%s\t%s\t%d\t%d\t%.6f\t%d\t%d\n", + mode, numMessagesStr, payloadSizeStr, fileSize, elapsed.Nanoseconds(), elapsed.Seconds(), peakRssKb(), + msgCount) + + return nil +} + +// peakRssKb returns peak RSS in KB. Rusage.Maxrss is KB on Linux but +// bytes on macOS. +func peakRssKb() int64 { + var rusage syscall.Rusage + syscall.Getrusage(syscall.RUSAGE_SELF, &rusage) + rss := int64(rusage.Maxrss) + if runtime.GOOS == "darwin" { + rss /= 1024 + } + return rss +} + +func main() { + if err := run(); err != nil { + fmt.Fprintf(os.Stderr, "%v\n", err) + os.Exit(1) + } +} diff --git a/benchmarking/go_bench/cmd/bench_write/main.go b/benchmarking/go_bench/cmd/bench_write/main.go new file mode 100644 index 000000000..866cfbe73 --- /dev/null +++ b/benchmarking/go_bench/cmd/bench_write/main.go @@ -0,0 +1,349 @@ +package main + +import ( + "fmt" + "hash/crc32" + "os" + "runtime" + "sort" + "strconv" + "syscall" + "time" + + "github.com/foxglove/mcap/go/mcap" +) + +// Shared payload blob parameters; must match gen_blob.py and the other +// language benches. Message i's payload is the window of the blob starting +// at (i * blobStride) % blobWindowSpan, so all implementations feed +// identical bytes to their writers. +const ( + blobSize = 16777216 + blobMaxPayload = 524288 + blobWindowSpan = blobSize - blobMaxPayload + blobStride = 7919 +) + +func payloadOffset(msgIndex int64) int64 { + return (msgIndex * blobStride) % blobWindowSpan +} + +func loadBlob(path string) ([]byte, error) { + blob, err := os.ReadFile(path) + if err != nil { + return nil, fmt.Errorf("failed to read blob file: %w", err) + } + if len(blob) != blobSize { + return nil, fmt.Errorf("blob file %s is not exactly %d bytes", path, blobSize) + } + return blob, nil +} + +type scheduledMsg struct { + timestamp uint64 + channelID uint16 +} + +func run() error { + if len(os.Args) != 6 { + return fmt.Errorf("Usage: %s \n mode: unchunked | chunked | zstd | lz4", os.Args[0]) + } + + blob, err := loadBlob(os.Args[5]) + if err != nil { + return err + } + + filename := os.Args[1] + mode := os.Args[2] + mixed := os.Args[4] == "mixed" + + var numMessages int64 + var payloadSize int64 + if !mixed { + var err error + numMessages, err = strconv.ParseInt(os.Args[3], 10, 64) + if err != nil { + return fmt.Errorf("invalid num_messages: %w", err) + } + payloadSize, err = strconv.ParseInt(os.Args[4], 10, 64) + if err != nil { + return fmt.Errorf("invalid payload_size: %w", err) + } + if payloadSize > blobMaxPayload { + return fmt.Errorf("payload_size must be <= %d", blobMaxPayload) + } + } + + var opts mcap.WriterOptions + opts.IncludeCRC = true + opts.OverrideLibrary = true + + switch mode { + case "unchunked": + opts.Chunked = false + case "chunked": + opts.Chunked = true + opts.ChunkSize = 786432 + opts.Compression = mcap.CompressionNone + case "zstd": + opts.Chunked = true + opts.ChunkSize = 786432 + opts.Compression = mcap.CompressionZSTD + case "lz4": + opts.Chunked = true + opts.ChunkSize = 786432 + opts.Compression = mcap.CompressionLZ4 + default: + return fmt.Errorf("Unknown mode: %s", mode) + } + + f, err := os.Create(filename) + if err != nil { + return fmt.Errorf("failed to create file: %w", err) + } + defer f.Close() + + w, err := mcap.NewWriter(f, &opts) + if err != nil { + return fmt.Errorf("failed to create writer: %w", err) + } + + // Header (not timed) + if err := w.WriteHeader(&mcap.Header{ + Profile: "bench", + Library: "go-bench", + }); err != nil { + return fmt.Errorf("failed to write header: %w", err) + } + + if mixed { + // --- Mixed payload mode: simulate a 10-second robot recording --- + + // Schemas + type schemaInfo struct { + id uint16 + name string + } + schemas := []schemaInfo{ + {1, "IMU"}, + {2, "Odometry"}, + {3, "TFMessage"}, + {4, "PointCloud2"}, + {5, "CompressedImage"}, + } + for _, s := range schemas { + if err := w.WriteSchema(&mcap.Schema{ + ID: s.id, + Name: s.name, + Encoding: "jsonschema", + Data: []byte(`{"type":"object"}`), + }); err != nil { + return fmt.Errorf("failed to write schema %s: %w", s.name, err) + } + } + + // Channels + type channelInfo struct { + id uint16 + schemaID uint16 + topic string + } + channels := []channelInfo{ + {1, 1, "/imu"}, + {2, 2, "/odom"}, + {3, 3, "/tf"}, + {4, 4, "/lidar"}, + {5, 5, "/camera/compressed"}, + } + for _, c := range channels { + if err := w.WriteChannel(&mcap.Channel{ + ID: c.id, + SchemaID: c.schemaID, + Topic: c.topic, + MessageEncoding: "json", + }); err != nil { + return fmt.Errorf("failed to write channel %s: %w", c.topic, err) + } + } + + // Pre-generate message schedule + type chanSpec struct { + channelID uint16 + periodNs uint64 + count int + } + chanSpecs := []chanSpec{ + {1, 5000000, 2000}, + {2, 20000000, 500}, + {3, 10000000, 1000}, + {4, 100000000, 100}, + {5, 66666667, 150}, + } + + schedule := make([]scheduledMsg, 0, 3750) + for _, cs := range chanSpecs { + for i := 0; i < cs.count; i++ { + schedule = append(schedule, scheduledMsg{ + timestamp: uint64(i) * cs.periodNs, + channelID: cs.channelID, + }) + } + } + sort.Slice(schedule, func(i, j int) bool { + if schedule[i].timestamp != schedule[j].timestamp { + return schedule[i].timestamp < schedule[j].timestamp + } + return schedule[i].channelID < schedule[j].channelID + }) + + // TF payload sizes cycle + tfSizes := []int{80, 160, 320, 800, 1600} + + // Channel ID -> fixed payload size (0 means variable/TF) + fixedPayload := map[uint16]int{ + 1: 96, + 2: 296, + 4: 230400, + 5: 524288, + } + + // Not timed: CRC of the payload stream for cross-language verification + var payloadCrc uint32 + { + crcSeq := make([]uint32, 6) // index by channelID (1-based) + for i, msg := range schedule { + seq := crcSeq[msg.channelID] + crcSeq[msg.channelID] = seq + 1 + size := fixedPayload[msg.channelID] + if msg.channelID == 3 { + size = tfSizes[seq%uint32(len(tfSizes))] + } + off := payloadOffset(int64(i)) + payloadCrc = crc32.Update(payloadCrc, crc32.IEEETable, blob[off:off+int64(size)]) + } + } + + // Per-channel sequence counters + chanSeq := make([]uint32, 6) // index by channelID (1-based) + + // Timed: message loop + close + start := time.Now() + + for i, msg := range schedule { + seq := chanSeq[msg.channelID] + chanSeq[msg.channelID] = seq + 1 + size := fixedPayload[msg.channelID] + if msg.channelID == 3 { + size = tfSizes[seq%uint32(len(tfSizes))] + } + off := payloadOffset(int64(i)) + data := blob[off : off+int64(size)] + if err := w.WriteMessage(&mcap.Message{ + ChannelID: msg.channelID, + Sequence: seq, + LogTime: msg.timestamp, + PublishTime: msg.timestamp, + Data: data, + }); err != nil { + return fmt.Errorf("failed to write message %d: %w", seq, err) + } + } + + if err := w.Close(); err != nil { + return fmt.Errorf("failed to close writer: %w", err) + } + + elapsed := time.Since(start) + + fi, err := f.Stat() + if err != nil { + return fmt.Errorf("failed to stat file: %w", err) + } + fileSize := fi.Size() + + fmt.Printf("write\tgo\t%s\t%d\t%v\t%d\t%d\t%.6f\t%d\t%d\n", + mode, 3750, "mixed", fileSize, elapsed.Nanoseconds(), elapsed.Seconds(), peakRssKb(), + payloadCrc) + } else { + // --- Fixed payload mode --- + if err := w.WriteSchema(&mcap.Schema{ + ID: 1, + Name: "BenchMsg", + Encoding: "jsonschema", + Data: []byte(`{"type":"object"}`), + }); err != nil { + return fmt.Errorf("failed to write schema: %w", err) + } + + if err := w.WriteChannel(&mcap.Channel{ + ID: 1, + SchemaID: 1, + Topic: "/bench", + MessageEncoding: "json", + }); err != nil { + return fmt.Errorf("failed to write channel: %w", err) + } + + // Not timed: CRC of the payload stream for cross-language verification + var payloadCrc uint32 + for i := int64(0); i < numMessages; i++ { + off := payloadOffset(i) + payloadCrc = crc32.Update(payloadCrc, crc32.IEEETable, blob[off:off+payloadSize]) + } + + // Timed: message loop + close + start := time.Now() + + for i := int64(0); i < numMessages; i++ { + logTime := uint64(i) * 1000 + off := payloadOffset(i) + if err := w.WriteMessage(&mcap.Message{ + ChannelID: 1, + Sequence: uint32(i), + LogTime: logTime, + PublishTime: logTime, + Data: blob[off : off+payloadSize], + }); err != nil { + return fmt.Errorf("failed to write message %d: %w", i, err) + } + } + + if err := w.Close(); err != nil { + return fmt.Errorf("failed to close writer: %w", err) + } + + elapsed := time.Since(start) + + fi, err := f.Stat() + if err != nil { + return fmt.Errorf("failed to stat file: %w", err) + } + fileSize := fi.Size() + + fmt.Printf("write\tgo\t%s\t%d\t%d\t%d\t%d\t%.6f\t%d\t%d\n", + mode, numMessages, payloadSize, fileSize, elapsed.Nanoseconds(), elapsed.Seconds(), peakRssKb(), + payloadCrc) + } + + return nil +} + +// peakRssKb returns peak RSS in KB. Rusage.Maxrss is KB on Linux but +// bytes on macOS. +func peakRssKb() int64 { + var rusage syscall.Rusage + syscall.Getrusage(syscall.RUSAGE_SELF, &rusage) + rss := int64(rusage.Maxrss) + if runtime.GOOS == "darwin" { + rss /= 1024 + } + return rss +} + +func main() { + if err := run(); err != nil { + fmt.Fprintf(os.Stderr, "%v\n", err) + os.Exit(1) + } +} diff --git a/benchmarking/go_bench/go.mod b/benchmarking/go_bench/go.mod new file mode 100644 index 000000000..446bd0211 --- /dev/null +++ b/benchmarking/go_bench/go.mod @@ -0,0 +1,18 @@ +module go_bench + +go 1.23 + +require github.com/foxglove/mcap/go/mcap v0.0.0 + +require ( + github.com/davecgh/go-spew v1.1.1 // indirect + github.com/klauspost/compress v1.16.7 // indirect + github.com/kr/text v0.2.0 // indirect + github.com/pierrec/lz4/v4 v4.1.22 // indirect + github.com/pmezard/go-difflib v1.0.0 // indirect + github.com/rogpeppe/go-internal v1.14.1 // indirect + github.com/stretchr/testify v1.9.0 // indirect + gopkg.in/yaml.v3 v3.0.1 // indirect +) + +replace github.com/foxglove/mcap/go/mcap => ../../go/mcap diff --git a/benchmarking/go_bench/go.sum b/benchmarking/go_bench/go.sum new file mode 100644 index 000000000..8f5bfaf7c --- /dev/null +++ b/benchmarking/go_bench/go.sum @@ -0,0 +1,22 @@ +github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/klauspost/compress v1.16.7 h1:2mk3MPGNzKyxErAw8YaohYh69+pa4sIQSC0fPGCFR9I= +github.com/klauspost/compress v1.16.7/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE= +github.com/kr/pretty v0.3.0 h1:WgNl7dwNpEZ6jJ9k1snq4pZsg7DOEN8hP9Xw0Tsjwk0= +github.com/kr/pretty v0.3.0/go.mod h1:640gp4NfQd8pI5XOwp5fnNeVWj67G7CFk/SaSQn7NBk= +github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= +github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/pierrec/lz4/v4 v4.1.22 h1:cKFw6uJDK+/gfw5BcDL0JL5aBsAFdsIT18eRtLj7VIU= +github.com/pierrec/lz4/v4 v4.1.22/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= +github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= +github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg= +github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY= +gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/benchmarking/python_bench/bench_read.py b/benchmarking/python_bench/bench_read.py new file mode 100644 index 000000000..1d1752215 --- /dev/null +++ b/benchmarking/python_bench/bench_read.py @@ -0,0 +1,79 @@ +#!/usr/bin/env python3 +"""MCAP read benchmark for Python.""" + +import os +import resource +import sys +import time + + +def peak_rss_kb() -> int: + """Peak RSS in KB. ru_maxrss is KB on Linux but bytes on macOS.""" + rss = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss + if sys.platform == "darwin": + rss //= 1024 + return rss + + +def main(): + if len(sys.argv) < 2: + print( + f"Usage: {sys.argv[0]} [mode] [num_messages] [payload_size] [filter]", + file=sys.stderr, + ) + print( + " filter: topic | timerange | topic_timerange (default: no filter)", + file=sys.stderr, + ) + return 1 + + filename = sys.argv[1] + mode = sys.argv[2] if len(sys.argv) >= 3 else "unknown" + num_messages_str = sys.argv[3] if len(sys.argv) >= 4 else "0" + payload_size_str = sys.argv[4] if len(sys.argv) >= 5 else "0" + filter_mode = sys.argv[5] if len(sys.argv) >= 6 else "" + + filter_kwargs = {} + if filter_mode == "topic": + filter_kwargs["topics"] = ["/imu"] + elif filter_mode == "timerange": + filter_kwargs["start_time"] = 3000000000 + filter_kwargs["end_time"] = 5000000000 + elif filter_mode == "topic_timerange": + filter_kwargs["topics"] = ["/lidar"] + filter_kwargs["start_time"] = 4000000000 + filter_kwargs["end_time"] = 6000000000 + + from mcap.reader import make_reader + + # Time file open + reader creation + message iteration + t_start = time.perf_counter_ns() + + msg_count = 0 + with open(filename, "rb") as f: + reader = make_reader(f) + for _schema, _channel, message in reader.iter_messages(**filter_kwargs): + # Touch the data to prevent dead-code elimination + if len(message.data) == 0: + print("Empty message", file=sys.stderr) + msg_count += 1 + + t_end = time.perf_counter_ns() + + elapsed_ns = t_end - t_start + wall_sec = elapsed_ns / 1e9 + file_size = os.path.getsize(filename) + + rss_kb = peak_rss_kb() + + # TSV output: op lang mode num_msgs payload_size file_size elapsed_ns wall_sec peak_rss_kb + # msg_count + print( + f"read\tpython\t{mode}\t{num_messages_str}\t{payload_size_str}\t{file_size}\t{elapsed_ns}\t{wall_sec:.6f}\t{rss_kb}\t{msg_count}" + ) + + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/benchmarking/python_bench/bench_write.py b/benchmarking/python_bench/bench_write.py new file mode 100644 index 000000000..a9e955303 --- /dev/null +++ b/benchmarking/python_bench/bench_write.py @@ -0,0 +1,207 @@ +#!/usr/bin/env python3 +"""MCAP write benchmark for Python.""" + +import os +import resource +import sys +import time +import zlib + +# Shared payload blob parameters; must match gen_blob.py and the other +# language benches. Message i's payload is the window of the blob starting +# at (i * BLOB_STRIDE) % BLOB_WINDOW_SPAN, so all implementations feed +# identical bytes to their writers. +BLOB_SIZE = 16777216 +BLOB_MAX_PAYLOAD = 524288 +BLOB_WINDOW_SPAN = BLOB_SIZE - BLOB_MAX_PAYLOAD +BLOB_STRIDE = 7919 + + +def payload_offset(msg_index): + return (msg_index * BLOB_STRIDE) % BLOB_WINDOW_SPAN + + +def load_blob(path): + with open(path, "rb") as f: + blob = f.read() + if len(blob) != BLOB_SIZE: + print(f"Blob file {path} is not exactly {BLOB_SIZE} bytes", file=sys.stderr) + sys.exit(1) + return memoryview(blob) + + +def peak_rss_kb() -> int: + """Peak RSS in KB. ru_maxrss is KB on Linux but bytes on macOS.""" + rss = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss + if sys.platform == "darwin": + rss //= 1024 + return rss + + +def main(): + if len(sys.argv) != 6: + print( + f"Usage: {sys.argv[0]} ", + file=sys.stderr, + ) + return 1 + + blob = load_blob(sys.argv[5]) + + filename = sys.argv[1] + mode = sys.argv[2] + mixed = sys.argv[4] == "mixed" + + if mixed: + num_messages = 3750 + payload_size_str = "mixed" + else: + num_messages = int(sys.argv[3]) + payload_size_str = sys.argv[4] + if int(sys.argv[4]) > BLOB_MAX_PAYLOAD: + print(f"payload_size must be <= {BLOB_MAX_PAYLOAD}", file=sys.stderr) + return 1 + + from mcap.writer import Writer, CompressionType + + with open(filename, "wb") as f: + if mode == "unchunked": + writer = Writer(f, use_chunking=False) + elif mode == "chunked": + writer = Writer(f, compression=CompressionType.NONE, chunk_size=786432) + elif mode == "zstd": + writer = Writer(f, compression=CompressionType.ZSTD, chunk_size=786432) + elif mode == "lz4": + writer = Writer(f, compression=CompressionType.LZ4, chunk_size=786432) + else: + print(f"Unknown mode: {mode}", file=sys.stderr) + return 1 + + writer.start(profile="bench", library="py-bench") + + if mixed: + # Channel definitions: (topic, schema_name, base_payload_size, period_ns, count) + channel_defs = [ + ("/imu", "IMU", 96, 5_000_000, 2000), + ("/odom", "Odometry", 296, 20_000_000, 500), + ("/tf", "TFMessage", None, 10_000_000, 1000), + ("/lidar", "PointCloud2", 230_400, 100_000_000, 100), + ("/camera/compressed", "CompressedImage", 524_288, 66_666_667, 150), + ] + + tf_payload_cycle = [80, 160, 320, 800, 1600] + + schema_ids = [] + channel_ids = [] + for topic, schema_name, _, _, _ in channel_defs: + sid = writer.register_schema( + name=schema_name, + encoding="jsonschema", + data=b'{"type":"object"}', + ) + cid = writer.register_channel( + topic=topic, + message_encoding="json", + schema_id=sid, + ) + schema_ids.append(sid) + channel_ids.append(cid) + + # Pre-generate the message schedule sorted by (timestamp, channel_index) + schedule = [] + for ch_idx, (_, _, _, period_ns, count) in enumerate(channel_defs): + for msg_i in range(count): + ts = msg_i * period_ns + schedule.append((ts, ch_idx, msg_i)) + schedule.sort(key=lambda x: (x[0], x[1])) + + # Not timed: CRC of the payload stream for cross-language verification + payload_crc = 0 + for i, (ts, ch_idx, msg_i) in enumerate(schedule): + if ch_idx == 2: # /tf + size = tf_payload_cycle[msg_i % len(tf_payload_cycle)] + else: + size = channel_defs[ch_idx][2] + off = payload_offset(i) + payload_crc = zlib.crc32(blob[off : off + size], payload_crc) + + # Per-channel sequence counters + seq = [0] * len(channel_defs) + + # Time the message-writing loop + finish + flush to the OS + t_start = time.perf_counter_ns() + + for i, (ts, ch_idx, msg_i) in enumerate(schedule): + if ch_idx == 2: # /tf + size = tf_payload_cycle[msg_i % len(tf_payload_cycle)] + else: + size = channel_defs[ch_idx][2] + off = payload_offset(i) + payload = blob[off : off + size] + writer.add_message( + channel_id=channel_ids[ch_idx], + sequence=seq[ch_idx], + log_time=ts, + publish_time=ts, + data=payload, + ) + seq[ch_idx] += 1 + + writer.finish() + f.flush() + t_end = time.perf_counter_ns() + else: + payload_size = int(sys.argv[4]) + + schema_id = writer.register_schema( + name="BenchMsg", + encoding="jsonschema", + data=b'{"type":"object"}', + ) + + channel_id = writer.register_channel( + topic="/bench", + message_encoding="json", + schema_id=schema_id, + ) + + # Not timed: CRC of the payload stream for cross-language verification + payload_crc = 0 + for i in range(num_messages): + off = payload_offset(i) + payload_crc = zlib.crc32(blob[off : off + payload_size], payload_crc) + + # Time the message-writing loop + finish + flush to the OS + t_start = time.perf_counter_ns() + + for i in range(num_messages): + log_time = i * 1000 + off = payload_offset(i) + writer.add_message( + channel_id=channel_id, + sequence=i, + log_time=log_time, + publish_time=log_time, + data=blob[off : off + payload_size], + ) + + writer.finish() + f.flush() + t_end = time.perf_counter_ns() + + elapsed_ns = t_end - t_start + wall_sec = elapsed_ns / 1e9 + file_size = os.path.getsize(filename) + + rss_kb = peak_rss_kb() + + # TSV output: op lang mode num_msgs payload_size file_size elapsed_ns wall_sec peak_rss_kb payload_crc32 + print( + f"write\tpython\t{mode}\t{num_messages}\t{payload_size_str}\t{file_size}\t{elapsed_ns}\t{wall_sec:.6f}\t{rss_kb}\t{payload_crc}" + ) + + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/benchmarking/run_bench.sh b/benchmarking/run_bench.sh new file mode 100755 index 000000000..3e17185e3 --- /dev/null +++ b/benchmarking/run_bench.sh @@ -0,0 +1,473 @@ +#!/usr/bin/env bash +set -euo pipefail + +# Configuration via environment variables +NUM_MESSAGES="${NUM_MESSAGES:-1000000}" +PAYLOAD_SIZE="${PAYLOAD_SIZE:-100}" +BENCH_ITERS="${BENCH_ITERS:-5}" +BENCH_DIR="${BENCH_DIR:-/tmp}" +MODES="${MODES:-unchunked chunked zstd lz4}" + +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" +RESULTS_FILE="${BENCH_DIR}/bench_results.tsv" +BLOB_FILE="${BLOB_FILE:-${BENCH_DIR}/bench_fill.bin}" + +# Paths to binaries +RUST_WRITE="${SCRIPT_DIR}/rust_bench/target/release/bench_write" +RUST_READ="${SCRIPT_DIR}/rust_bench/target/release/bench_read" +GO_WRITE="${SCRIPT_DIR}/go_bench/bench_write" +GO_READ="${SCRIPT_DIR}/go_bench/bench_read" +CPP_WRITE="${SCRIPT_DIR}/cpp_bench/bench_write" +CPP_READ="${SCRIPT_DIR}/cpp_bench/bench_read" + +# Python config +PYTHON="${PYTHON:-python3}" +INTEROP_PYPATH="${INTEROP_PYPATH:-${SCRIPT_DIR}/../python/mcap}" +PY_WRITE="${SCRIPT_DIR}/python_bench/bench_write.py" +PY_READ="${SCRIPT_DIR}/python_bench/bench_read.py" + +# TypeScript config +TSX="${TSX:-npx tsx}" +TS_WRITE="${SCRIPT_DIR}/typescript_bench/bench_write.ts" +TS_READ="${SCRIPT_DIR}/typescript_bench/bench_read.ts" + +# Verify binaries exist +for bin in "$RUST_WRITE" "$RUST_READ" "$GO_WRITE" "$GO_READ" "$CPP_WRITE" "$CPP_READ"; do + if [ ! -x "$bin" ]; then + echo "ERROR: Binary not found or not executable: $bin" >&2 + echo "Run 'make all' first." >&2 + exit 1 + fi +done + +# Verify Python scripts exist +for script in "$PY_WRITE" "$PY_READ" "${SCRIPT_DIR}/gen_blob.py"; do + if [ ! -f "$script" ]; then + echo "ERROR: Python script not found: $script" >&2 + echo "Run 'make all' first." >&2 + exit 1 + fi +done + +# Verify TypeScript scripts exist +for script in "$TS_WRITE" "$TS_READ"; do + if [ ! -f "$script" ]; then + echo "ERROR: TypeScript script not found: $script" >&2 + echo "Run 'make all' first." >&2 + exit 1 + fi +done + +echo "=== MCAP Cross-Language Benchmark ===" +echo "Messages: ${NUM_MESSAGES}, Payload: ${PAYLOAD_SIZE} bytes, Iterations: ${BENCH_ITERS}" +echo "Output dir: ${BENCH_DIR}" +echo "Modes: ${MODES}" +echo "" + +# Generate the shared payload blob (deterministic; reused if present) +if [ ! -f "$BLOB_FILE" ]; then + echo "Generating payload blob ${BLOB_FILE}..." + "$PYTHON" "${SCRIPT_DIR}/gen_blob.py" "$BLOB_FILE" + echo "" +fi + +# Verify that every language fed identical payload bytes to its writer: +# column 10 of each write row is a CRC-32 of the payload stream, which +# must match across languages (and iterations) for each mode. +verify_checksums() { + local results_file="$1" modes="$2" bad=0 mode n + for mode in $modes; do + n=$(awk -F'\t' -v m="$mode" '$1=="write" && $3==m {print $10}' "$results_file" | sort -u | wc -l) + if [ "$n" -gt 1 ]; then + echo "ERROR: payload CRC mismatch across languages for mode ${mode}:" >&2 + awk -F'\t' -v m="$mode" '$1=="write" && $3==m {print " "$2"\t"$10}' "$results_file" | sort -u >&2 + bad=1 + fi + done + if [ "$bad" -ne 0 ]; then + echo "Aborting: implementations did not write identical payloads." >&2 + exit 1 + fi +} + +# Verify that every read bench consumed the expected number of messages: +# column 10 of each read row is the message count reported by the bench. A +# reader that silently under-reads would otherwise post a fast time with no +# error. +verify_read_counts() { + local results_file="$1" expected="$2" rows + rows=$(awk -F'\t' -v e="$expected" \ + '$1=="read" && $10 != e {print " "$2"/"$3": read "$10" messages, expected "e}' "$results_file") + if [ -n "$rows" ]; then + echo "ERROR: read benches did not consume the expected message count:" >&2 + echo "$rows" >&2 + echo "Aborting: incomplete reads invalidate the read timings." >&2 + exit 1 + fi +} + +# Filtered reads: a zero count means a broken filter, so fail hard. Exact +# counts may legitimately differ between languages if their time-range +# boundary semantics differ, so cross-language disagreement is reported +# loudly but does not abort. +check_filter_counts() { + local results_file="$1" zero mode n + zero=$(awk -F'\t' '$1=="read" && $10+0 == 0 {print " "$2"/"$3}' "$results_file") + if [ -n "$zero" ]; then + echo "ERROR: filtered read returned zero messages:" >&2 + echo "$zero" >&2 + echo "Aborting: an empty filtered read invalidates the filtered timings." >&2 + exit 1 + fi + for mode in $(awk -F'\t' '$1=="read" {print $3}' "$results_file" | sort -u); do + n=$(awk -F'\t' -v m="$mode" '$1=="read" && $3==m {print $10}' "$results_file" | sort -u | wc -l) + if [ "$n" -gt 1 ]; then + echo "WARNING: message count disagreement across languages for ${mode}:" + awk -F'\t' -v m="$mode" '$1=="read" && $3==m {print " "$2"\t"$10}' "$results_file" | sort -u + fi + done +} + +# Clear results file +> "$RESULTS_FILE" + +LANGS="rust go python cpp typescript" + +# Set write_cmd and read_cmd for a given language. +set_lang_cmds() { + local lang="$1" + case "$lang" in + rust) write_cmd="$RUST_WRITE"; read_cmd="$RUST_READ" ;; + go) write_cmd="$GO_WRITE"; read_cmd="$GO_READ" ;; + python) write_cmd="PYTHONPATH=$INTEROP_PYPATH $PYTHON $PY_WRITE"; read_cmd="PYTHONPATH=$INTEROP_PYPATH $PYTHON $PY_READ" ;; + cpp) write_cmd="$CPP_WRITE"; read_cmd="$CPP_READ" ;; + typescript) write_cmd="$TSX $TS_WRITE"; read_cmd="$TSX $TS_READ" ;; + *) echo "unknown lang: $lang" >&2; exit 1 ;; + esac +} + +for mode in $MODES; do + for lang in $LANGS; do + set_lang_cmds "$lang" + + # TypeScript doesn't support LZ4 compression for writes + if [ "$lang" = "typescript" ] && [ "$mode" = "lz4" ]; then + continue + fi + + outfile="${BENCH_DIR}/bench_${lang}_${mode}.mcap" + + for iter in $(seq 1 "$BENCH_ITERS"); do + echo -n " ${lang}/${mode} write iter ${iter}/${BENCH_ITERS}..." + result=$(eval "$write_cmd" "$outfile" "$mode" "$NUM_MESSAGES" "$PAYLOAD_SIZE" "$BLOB_FILE") + echo "$result" >> "$RESULTS_FILE" + echo " done" + + echo -n " ${lang}/${mode} read iter ${iter}/${BENCH_ITERS}..." + result=$(eval "$read_cmd" "$outfile" "$mode" "$NUM_MESSAGES" "$PAYLOAD_SIZE") + echo "$result" >> "$RESULTS_FILE" + echo " done" + done + done +done + +verify_checksums "$RESULTS_FILE" "$MODES" +verify_read_counts "$RESULTS_FILE" "$NUM_MESSAGES" + +echo "" +echo "=== Raw results ===" +cat "$RESULTS_FILE" +echo "" + +# --- Compute and display tables --- + +# median helper: takes a list of numbers (one per line), outputs the median +median() { + sort -n | awk '{a[NR]=$1} END {if (NR%2==1) print a[(NR+1)/2]; else print (a[NR/2]+a[NR/2+1])/2}' +} + +# Collect unique file sizes per (lang, mode) for the write operations +echo "=== FILE SIZE COMPARISON ===" +echo "" +printf "%-6s %-12s %12s %10s\n" "Lang" "Mode" "FileSize" "Ratio" +printf "%-6s %-12s %12s %10s\n" "----" "--------" "--------" "-----" + +# Get baseline (unchunked) sizes for ratio calculation +declare -A file_sizes +for mode in $MODES; do + for lang in $LANGS; do + # Get file size from first write result for this (lang, mode) + fsize=$(awk -F'\t' -v l="$lang" -v m="$mode" '$1=="write" && $2==l && $3==m {print $6; exit}' "$RESULTS_FILE") + file_sizes["${lang}_${mode}"]="$fsize" + done +done + +for lang in $LANGS; do + base_size="${file_sizes[${lang}_unchunked]:-0}" + for mode in $MODES; do + fsize="${file_sizes[${lang}_${mode}]:-0}" + if [ "$base_size" -gt 0 ] 2>/dev/null; then + ratio=$(awk "BEGIN {printf \"%.2fx\", $fsize/$base_size}") + else + ratio="N/A" + fi + # Human-readable file size + if [ "$fsize" -gt 1048576 ] 2>/dev/null; then + hr_size=$(awk "BEGIN {printf \"%.1f MB\", $fsize/1048576}") + elif [ "$fsize" -gt 1024 ] 2>/dev/null; then + hr_size=$(awk "BEGIN {printf \"%.1f KB\", $fsize/1024}") + else + hr_size="${fsize} B" + fi + printf "%-6s %-12s %12s %10s\n" "$lang" "$mode" "$hr_size" "$ratio" + done +done + +echo "" + +# Display memory usage table +echo "=== MEMORY USAGE (median of ${BENCH_ITERS} iterations) ===" +echo "" +printf "%-6s %-12s %12s %12s\n" "Lang" "Mode" "Write(MB)" "Read(MB)" +printf "%-6s %-12s %12s %12s\n" "----" "--------" "---------" "--------" + +for mode in $MODES; do + for lang in $LANGS; do + write_rss_values=$(awk -F'\t' -v l="$lang" -v m="$mode" \ + '$1=="write" && $2==l && $3==m {print $9}' "$RESULTS_FILE") + read_rss_values=$(awk -F'\t' -v l="$lang" -v m="$mode" \ + '$1=="read" && $2==l && $3==m {print $9}' "$RESULTS_FILE") + + if [ -z "$write_rss_values" ] && [ -z "$read_rss_values" ]; then + continue + fi + + write_mb="N/A" + read_mb="N/A" + if [ -n "$write_rss_values" ]; then + write_median_kb=$(echo "$write_rss_values" | median) + write_mb=$(awk "BEGIN {printf \"%.1f\", $write_median_kb / 1024}") + fi + if [ -n "$read_rss_values" ]; then + read_median_kb=$(echo "$read_rss_values" | median) + read_mb=$(awk "BEGIN {printf \"%.1f\", $read_median_kb / 1024}") + fi + + printf "%-6s %-12s %12s %12s\n" "$lang" "$mode" "$write_mb" "$read_mb" + done +done + +echo "" + +# Display write and read results +for op in write read; do + echo "=== $(echo "$op" | tr '[:lower:]' '[:upper:]') RESULTS (median of ${BENCH_ITERS} iterations) ===" + echo "" + printf "%-6s %-12s %12s %12s %12s %12s %10s\n" \ + "Lang" "Mode" "Median(ms)" "Min(ms)" "Max(ms)" "Msg/sec" "MB/sec" + printf "%-6s %-12s %12s %12s %12s %12s %10s\n" \ + "----" "--------" "----------" "--------" "--------" "--------" "------" + + for mode in $MODES; do + for lang in $LANGS; do + # Extract elapsed_ns values for this (op, lang, mode) + ns_values=$(awk -F'\t' -v o="$op" -v l="$lang" -v m="$mode" \ + '$1==o && $2==l && $3==m {print $7}' "$RESULTS_FILE") + + if [ -z "$ns_values" ]; then + continue + fi + + median_ns=$(echo "$ns_values" | median) + min_ns=$(echo "$ns_values" | sort -n | head -1) + max_ns=$(echo "$ns_values" | sort -n | tail -1) + + # Compute derived metrics from median + awk -v med="$median_ns" -v mn="$min_ns" -v mx="$max_ns" \ + -v nm="$NUM_MESSAGES" -v ps="$PAYLOAD_SIZE" \ + -v lang="$lang" -v mode="$mode" \ + 'BEGIN { + med_ms = med / 1e6 + min_ms = mn / 1e6 + max_ms = mx / 1e6 + med_sec = med / 1e9 + if (med_sec > 0) { + msg_per_sec = nm / med_sec + mb_per_sec = (nm * ps) / med_sec / 1048576 + } else { + msg_per_sec = 0 + mb_per_sec = 0 + } + printf "%-6s %-12s %12.1f %12.1f %12.1f %12.0f %10.1f\n", \ + lang, mode, med_ms, min_ms, max_ms, msg_per_sec, mb_per_sec + }' + done + done + echo "" +done + +# --- Mixed-payload benchmarks --- + +MIXED_MODES="${MIXED_MODES:-unchunked chunked zstd lz4}" +MIXED_RESULTS_FILE="${BENCH_DIR}/bench_mixed_results.tsv" +> "$MIXED_RESULTS_FILE" + +echo "=== MCAP Mixed-Payload Benchmark (10s robot recording) ===" +echo "Channels: /imu(96B@200Hz) /odom(296B@50Hz) /tf(80-1600B@100Hz) /lidar(230KB@10Hz) /camera(512KB@15Hz)" +echo "Total: 3750 messages, ~102 MB" +echo "Iterations: ${BENCH_ITERS}" +echo "Modes: ${MIXED_MODES}" +echo "" + +for mode in $MIXED_MODES; do + for lang in $LANGS; do + set_lang_cmds "$lang" + + # TypeScript doesn't support LZ4 compression for writes + if [ "$lang" = "typescript" ] && [ "$mode" = "lz4" ]; then + continue + fi + + outfile="${BENCH_DIR}/bench_${lang}_mixed_${mode}.mcap" + + for iter in $(seq 1 "$BENCH_ITERS"); do + echo -n " ${lang}/mixed-${mode} write iter ${iter}/${BENCH_ITERS}..." + result=$(eval "$write_cmd" "$outfile" "$mode" 0 mixed "$BLOB_FILE") + echo "$result" >> "$MIXED_RESULTS_FILE" + echo " done" + + echo -n " ${lang}/mixed-${mode} read iter ${iter}/${BENCH_ITERS}..." + result=$(eval "$read_cmd" "$outfile" "$mode" 0 mixed) + echo "$result" >> "$MIXED_RESULTS_FILE" + echo " done" + done + done +done + +verify_checksums "$MIXED_RESULTS_FILE" "$MIXED_MODES" +verify_read_counts "$MIXED_RESULTS_FILE" 3750 + +echo "" +echo "=== Mixed-payload raw results ===" +cat "$MIXED_RESULTS_FILE" +echo "" + +# Mixed-payload results table +for op in write read; do + echo "=== MIXED $(echo "$op" | tr '[:lower:]' '[:upper:]') RESULTS (median of ${BENCH_ITERS} iterations) ===" + echo "" + printf "%-12s %-12s %12s %12s %12s\n" \ + "Lang" "Mode" "Median(ms)" "Min(ms)" "Max(ms)" + printf "%-12s %-12s %12s %12s %12s\n" \ + "----" "--------" "----------" "--------" "--------" + + for mode in $MIXED_MODES; do + for lang in $LANGS; do + ns_values=$(awk -F'\t' -v o="$op" -v l="$lang" -v m="$mode" \ + '$1==o && $2==l && $3==m {print $7}' "$MIXED_RESULTS_FILE") + + if [ -z "$ns_values" ]; then + continue + fi + + median_ns=$(echo "$ns_values" | median) + min_ns=$(echo "$ns_values" | sort -n | head -1) + max_ns=$(echo "$ns_values" | sort -n | tail -1) + + awk -v med="$median_ns" -v mn="$min_ns" -v mx="$max_ns" \ + -v lang="$lang" -v mode="$mode" \ + 'BEGIN { + med_ms = med / 1e6 + min_ms = mn / 1e6 + max_ms = mx / 1e6 + printf "%-12s %-12s %12.1f %12.1f %12.1f\n", \ + lang, mode, med_ms, min_ms, max_ms + }' + done + done + echo "" +done + +# --- Filtered read benchmarks (using mixed-payload files) --- + +FILTER_MODES="${FILTER_MODES:-topic timerange topic_timerange}" +FILTER_COMPRESSION="${FILTER_COMPRESSION:-chunked zstd}" +FILTER_RESULTS_FILE="${BENCH_DIR}/bench_filter_results.tsv" +> "$FILTER_RESULTS_FILE" + +echo "=== MCAP Filtered Read Benchmark ===" +echo "Filters: topic(/imu) timerange(3-5s) topic_timerange(/lidar 4-6s)" +echo "Compression modes: ${FILTER_COMPRESSION}" +echo "Iterations: ${BENCH_ITERS}" +echo "" + +for compression in $FILTER_COMPRESSION; do + for lang in $LANGS; do + set_lang_cmds "$lang" + + if [ "$lang" = "typescript" ] && [ "$compression" = "lz4" ]; then + continue + fi + + # Use the mixed-payload file written earlier + outfile="${BENCH_DIR}/bench_${lang}_mixed_${compression}.mcap" + if [ ! -f "$outfile" ]; then + echo " SKIP ${lang}/${compression}: mixed file not found (run mixed benchmarks first)" + continue + fi + + for filter in $FILTER_MODES; do + for iter in $(seq 1 "$BENCH_ITERS"); do + echo -n " ${lang}/${compression}/${filter} read iter ${iter}/${BENCH_ITERS}..." + result=$(eval "$read_cmd" "$outfile" "${compression}-${filter}" 0 mixed "$filter") + echo "$result" >> "$FILTER_RESULTS_FILE" + echo " done" + done + done + done +done + +check_filter_counts "$FILTER_RESULTS_FILE" + +echo "" +echo "=== Filtered read raw results ===" +cat "$FILTER_RESULTS_FILE" +echo "" + +echo "=== FILTERED READ RESULTS (median of ${BENCH_ITERS} iterations) ===" +echo "" +printf "%-12s %-20s %12s %12s %12s\n" \ + "Lang" "Compression/Filter" "Median(ms)" "Min(ms)" "Max(ms)" +printf "%-12s %-20s %12s %12s %12s\n" \ + "----" "------------------" "----------" "--------" "--------" + +for compression in $FILTER_COMPRESSION; do + for filter in $FILTER_MODES; do + combined="${compression}-${filter}" + for lang in $LANGS; do + ns_values=$(awk -F'\t' -v l="$lang" -v m="$combined" \ + '$1=="read" && $2==l && $3==m {print $7}' "$FILTER_RESULTS_FILE") + + if [ -z "$ns_values" ]; then + continue + fi + + median_ns=$(echo "$ns_values" | median) + min_ns=$(echo "$ns_values" | sort -n | head -1) + max_ns=$(echo "$ns_values" | sort -n | tail -1) + + awk -v med="$median_ns" -v mn="$min_ns" -v mx="$max_ns" \ + -v lang="$lang" -v combined="$combined" \ + 'BEGIN { + med_ms = med / 1e6 + min_ms = mn / 1e6 + max_ms = mx / 1e6 + printf "%-12s %-20s %12.1f %12.1f %12.1f\n", \ + lang, combined, med_ms, min_ms, max_ms + }' + done + done +done +echo "" + +echo "=== Benchmark complete ===" diff --git a/benchmarking/rust_bench/Cargo.lock b/benchmarking/rust_bench/Cargo.lock new file mode 100644 index 000000000..264e9f852 --- /dev/null +++ b/benchmarking/rust_bench/Cargo.lock @@ -0,0 +1,397 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "array-init" +version = "2.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3d62b7694a562cdf5a74227903507c56ab2cc8bdd1f781ed5cb4cf9c9f810bfc" + +[[package]] +name = "bimap" +version = "0.6.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "230c5f1ca6a325a32553f8640d31ac9b49f2411e901e427570154868b46da4f7" + +[[package]] +name = "binrw" +version = "0.15.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ad120d555272286c1017d25165ab8bd74806f13fc85b258484ec7e4ce75458f" +dependencies = [ + "array-init", + "binrw_derive", + "bytemuck", +] + +[[package]] +name = "binrw_derive" +version = "0.15.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6df92e0e9baae4dc82c7bad7715ca40c0a5c71539057bf2ea04a5c29c980410b" +dependencies = [ + "either", + "owo-colors", + "proc-macro2", + "quote", + "syn 2.0.117", +] + +[[package]] +name = "bytemuck" +version = "1.25.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c8efb64bd706a16a1bdde310ae86b351e4d21550d98d056f22f8a7f7a2183fec" + +[[package]] +name = "byteorder" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b" + +[[package]] +name = "cc" +version = "1.2.57" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7a0dd1ca384932ff3641c8718a02769f1698e7563dc6974ffd03346116310423" +dependencies = [ + "find-msvc-tools", + "jobserver", + "libc", + "shlex", +] + +[[package]] +name = "cfg-if" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" + +[[package]] +name = "crc32fast" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9481c1c90cbf2ac953f07c8d4a58aa3945c425b7185c9154d67a65e4230da511" +dependencies = [ + "cfg-if", +] + +[[package]] +name = "darling" +version = "0.21.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9cdf337090841a411e2a7f3deb9187445851f91b309c0c0a29e05f74a00a48c0" +dependencies = [ + "darling_core", + "darling_macro", +] + +[[package]] +name = "darling_core" +version = "0.21.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1247195ecd7e3c85f83c8d2a366e4210d588e802133e1e355180a9870b517ea4" +dependencies = [ + "fnv", + "ident_case", + "proc-macro2", + "quote", + "syn 2.0.117", +] + +[[package]] +name = "darling_macro" +version = "0.21.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d38308df82d1080de0afee5d069fa14b0326a88c14f15c5ccda35b4a6c414c81" +dependencies = [ + "darling_core", + "quote", + "syn 2.0.117", +] + +[[package]] +name = "either" +version = "1.15.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "48c757948c5ede0e46177b7add2e67155f70e33c07fea8284df6576da70b3719" + +[[package]] +name = "enumset" +version = "1.1.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "25b07a8dfbbbfc0064c0a6bdf9edcf966de6b1c33ce344bdeca3b41615452634" +dependencies = [ + "enumset_derive", +] + +[[package]] +name = "enumset_derive" +version = "0.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f43e744e4ea338060faee68ed933e46e722fb7f3617e722a5772d7e856d8b3ce" +dependencies = [ + "darling", + "proc-macro2", + "quote", + "syn 2.0.117", +] + +[[package]] +name = "find-msvc-tools" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" + +[[package]] +name = "fnv" +version = "1.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1" + +[[package]] +name = "getrandom" +version = "0.3.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd" +dependencies = [ + "cfg-if", + "libc", + "r-efi", + "wasip2", +] + +[[package]] +name = "hermit-abi" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc0fef456e4baa96da950455cd02c081ca953b141298e41db3fc7e36b1da849c" + +[[package]] +name = "ident_case" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9e0384b61958566e926dc50660321d12159025e767c18e043daf26b70104c39" + +[[package]] +name = "jobserver" +version = "0.1.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9afb3de4395d6b3e67a780b6de64b51c978ecf11cb9a462c66be7d4ca9039d33" +dependencies = [ + "getrandom", + "libc", +] + +[[package]] +name = "libc" +version = "0.2.183" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b5b646652bf6661599e1da8901b3b9522896f01e736bad5f723fe7a3a27f899d" + +[[package]] +name = "log" +version = "0.4.29" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897" + +[[package]] +name = "lz4" +version = "1.28.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a20b523e860d03443e98350ceaac5e71c6ba89aea7d960769ec3ce37f4de5af4" +dependencies = [ + "lz4-sys", +] + +[[package]] +name = "lz4-sys" +version = "1.11.1+lz4-1.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6bd8c0d6c6ed0cd30b3652886bb8711dc4bb01d637a68105a3d5158039b418e6" +dependencies = [ + "cc", + "libc", +] + +[[package]] +name = "mcap" +version = "0.25.0" +dependencies = [ + "bimap", + "binrw", + "byteorder", + "crc32fast", + "enumset", + "log", + "lz4", + "num_cpus", + "paste", + "static_assertions", + "thiserror", + "zstd", +] + +[[package]] +name = "num_cpus" +version = "1.17.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "91df4bbde75afed763b708b7eee1e8e7651e02d97f6d5dd763e89367e957b23b" +dependencies = [ + "hermit-abi", + "libc", +] + +[[package]] +name = "owo-colors" +version = "4.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d211803b9b6b570f68772237e415a029d5a50c65d382910b879fb19d3271f94d" + +[[package]] +name = "paste" +version = "1.0.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "57c0d7b74b563b49d38dae00a0c37d4d6de9b432382b2892f0574ddcae73fd0a" + +[[package]] +name = "pkg-config" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7edddbd0b52d732b21ad9a5fab5c704c14cd949e5e9a1ec5929a24fded1b904c" + +[[package]] +name = "proc-macro2" +version = "1.0.106" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8fd00f0bb2e90d81d1044c2b32617f68fcb9fa3bb7640c23e9c748e53fb30934" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quote" +version = "1.0.45" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "41f2619966050689382d2b44f664f4bc593e129785a36d6ee376ddf37259b924" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "r-efi" +version = "5.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f" + +[[package]] +name = "rust_bench" +version = "0.1.0" +dependencies = [ + "libc", + "mcap", +] + +[[package]] +name = "shlex" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64" + +[[package]] +name = "static_assertions" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f" + +[[package]] +name = "syn" +version = "2.0.117" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e665b8803e7b1d2a727f4023456bbbbe74da67099c585258af0ad9c5013b9b99" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "syn" +version = "3.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53e9bae58849f64dfa4f5d5ae372c8341f7305f82a3868709269343628b659a3" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "thiserror" +version = "2.0.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09a43598840e33d5b0331f38c5e30d13bb11c11210a4b58f0d9b18a5a5eefcd9" +dependencies = [ + "thiserror-impl", +] + +[[package]] +name = "thiserror-impl" +version = "2.0.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "43cbfe0cf76104d42a574802844187e84a305e531ed54455f11fbde0f10541cd" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + +[[package]] +name = "unicode-ident" +version = "1.0.24" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" + +[[package]] +name = "wasip2" +version = "1.0.2+wasi-0.2.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9517f9239f02c069db75e65f174b3da828fe5f5b945c4dd26bd25d89c03ebcf5" +dependencies = [ + "wit-bindgen", +] + +[[package]] +name = "wit-bindgen" +version = "0.51.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d7249219f66ced02969388cf2bb044a09756a083d0fab1e566056b04d9fbcaa5" + +[[package]] +name = "zstd" +version = "0.13.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e91ee311a569c327171651566e07972200e76fcfe2242a4fa446149a3881c08a" +dependencies = [ + "zstd-safe", +] + +[[package]] +name = "zstd-safe" +version = "7.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f49c4d5f0abb602a93fb8736af2a4f4dd9512e36f7f570d66e65ff867ed3b9d" +dependencies = [ + "zstd-sys", +] + +[[package]] +name = "zstd-sys" +version = "2.0.16+zstd.1.5.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "91e19ebc2adc8f83e43039e79776e3fda8ca919132d68a1fed6a5faca2683748" +dependencies = [ + "cc", + "pkg-config", +] diff --git a/benchmarking/rust_bench/Cargo.toml b/benchmarking/rust_bench/Cargo.toml new file mode 100644 index 000000000..ecf4fc0ae --- /dev/null +++ b/benchmarking/rust_bench/Cargo.toml @@ -0,0 +1,14 @@ +[package] +name = "rust_bench" +version = "0.1.0" +edition = "2021" + +[workspace] + +[dependencies] +mcap = { path = "../../rust/mcap" } +libc = "0.2" + +[profile.release] +opt-level = 3 +lto = true diff --git a/benchmarking/rust_bench/src/bin/bench_read.rs b/benchmarking/rust_bench/src/bin/bench_read.rs new file mode 100644 index 000000000..52c345af3 --- /dev/null +++ b/benchmarking/rust_bench/src/bin/bench_read.rs @@ -0,0 +1,137 @@ +use std::io::Read; +use std::time::Instant; + +use mcap::sans_io::indexed_reader::{IndexedReadEvent, IndexedReader, IndexedReaderOptions}; +use mcap::sans_io::linear_reader::{LinearReadEvent, LinearReader}; + +/// ru_maxrss is KB on Linux but bytes on macOS; normalize to KB. +fn peak_rss_kb() -> libc::c_long { + let mut rusage: libc::rusage = unsafe { std::mem::zeroed() }; + unsafe { libc::getrusage(libc::RUSAGE_SELF, &mut rusage) }; + if cfg!(target_os = "macos") { + rusage.ru_maxrss / 1024 + } else { + rusage.ru_maxrss + } +} + +fn main() { + let args: Vec = std::env::args().collect(); + if args.len() < 2 || args.len() > 6 { + eprintln!( + "Usage: {} [mode] [num_messages] [payload_size] [filter]", + args[0] + ); + eprintln!(" filter: topic | timerange | topic_timerange"); + std::process::exit(1); + } + + let filename = &args[1]; + let mode = if args.len() >= 3 { &args[2] } else { "unknown" }; + let num_messages_str = if args.len() >= 4 { &args[3] } else { "0" }; + let payload_size_str = if args.len() >= 5 { &args[4] } else { "0" }; + let filter = if args.len() >= 6 { + Some(args[5].as_str()) + } else { + None + }; + + // Timed: file read + message iteration + let start = Instant::now(); + + let mut msg_count: u64 = 0; + + match filter { + None => { + // Stream the file through the sans-io LinearReader to keep + // memory bounded rather than buffering the whole file. The + // reader requests a few bytes at a time on unchunked files, so + // buffer the underlying reads. + let file = std::fs::File::open(filename).expect("Failed to open file"); + let mut file = std::io::BufReader::with_capacity(1 << 20, file); + let mut reader = LinearReader::new(); + while let Some(event) = reader.next_event() { + match event.expect("Failed to read event") { + LinearReadEvent::ReadRequest(need) => { + let written = file.read(reader.insert(need)).expect("Failed to read file"); + reader.notify_read(written); + } + LinearReadEvent::Record { opcode, data } => { + if opcode == mcap::records::op::MESSAGE { + let record = mcap::parse_record(opcode, data) + .expect("Failed to parse message record"); + if let mcap::records::Record::Message { data, .. } = record { + // Touch data to prevent dead-code elimination + if data.is_empty() { + eprintln!("Empty message"); + } + msg_count += 1; + } + } + } + } + } + } + Some(filter_mode) => { + // The indexed reader operates on byte slices, so the filtered + // path buffers the whole file. Filtered results do not feed + // the memory table. + let buf = std::fs::read(filename).expect("Failed to read file"); + let summary = mcap::Summary::read(&buf) + .expect("Failed to read summary") + .expect("No summary found in file"); + + let options = match filter_mode { + "topic" => IndexedReaderOptions::new().include_topics(vec!["/imu"]), + "timerange" => IndexedReaderOptions::new() + .log_time_on_or_after(3_000_000_000) + .log_time_before(5_000_000_000), + "topic_timerange" => IndexedReaderOptions::new() + .include_topics(vec!["/lidar"]) + .log_time_on_or_after(4_000_000_000) + .log_time_before(6_000_000_000), + other => { + eprintln!("Unknown filter mode: {}", other); + std::process::exit(1); + } + }; + + let mut reader = IndexedReader::new_with_options(&summary, options) + .expect("Failed to create indexed reader"); + while let Some(event) = reader.next_event() { + match event.expect("Failed to read event") { + IndexedReadEvent::ReadChunkRequest { offset, length } => { + let chunk_data = &buf[offset as usize..][..length]; + reader + .insert_chunk_record_data(offset, chunk_data) + .expect("Failed to insert chunk data"); + } + IndexedReadEvent::Message { header: _, data } => { + msg_count += 1; + if data.is_empty() { + eprintln!("Empty message"); + } + } + } + } + } + } + + let elapsed = start.elapsed(); + let elapsed_ns = elapsed.as_nanos(); + let wall_sec = elapsed.as_secs_f64(); + + let file_size = std::fs::metadata(filename) + .expect("Failed to stat file") + .len(); + + let peak_rss_kb = peak_rss_kb(); + + // TSV output: op lang mode num_msgs payload_size file_size elapsed_ns wall_sec peak_rss_kb + // msg_count + println!( + "read\trust\t{}\t{}\t{}\t{}\t{}\t{:.6}\t{}\t{}", + mode, num_messages_str, payload_size_str, file_size, elapsed_ns, wall_sec, peak_rss_kb, + msg_count + ); +} diff --git a/benchmarking/rust_bench/src/bin/bench_write.rs b/benchmarking/rust_bench/src/bin/bench_write.rs new file mode 100644 index 000000000..f14141ed0 --- /dev/null +++ b/benchmarking/rust_bench/src/bin/bench_write.rs @@ -0,0 +1,270 @@ +use mcap::write::WriteOptions; +use mcap::Compression; +use std::collections::BTreeMap; +use std::io::BufWriter; +use std::time::Instant; + +/// Shared payload blob parameters; must match gen_blob.py and the other +/// language benches. Message i's payload is the window of the blob starting +/// at (i * STRIDE) % WINDOW_SPAN, so all implementations feed identical +/// bytes to their writers. +const BLOB_SIZE: usize = 16_777_216; +const MAX_PAYLOAD: usize = 524_288; +const WINDOW_SPAN: u64 = (BLOB_SIZE - MAX_PAYLOAD) as u64; +const STRIDE: u64 = 7919; + +fn payload_offset(msg_index: u64) -> usize { + ((msg_index * STRIDE) % WINDOW_SPAN) as usize +} + +fn load_blob(path: &str) -> Vec { + let blob = std::fs::read(path).expect("Failed to read blob file"); + assert!( + blob.len() == BLOB_SIZE, + "Blob file {} is not exactly {} bytes", + path, + BLOB_SIZE + ); + blob +} + +/// CRC-32 (IEEE, zlib-compatible) over the payload stream, used by +/// run_bench.sh to verify all languages fed identical bytes. +fn crc32_table() -> [u32; 256] { + let mut table = [0u32; 256]; + for (i, entry) in table.iter_mut().enumerate() { + let mut c = i as u32; + for _ in 0..8 { + c = if c & 1 != 0 { 0xEDB88320 ^ (c >> 1) } else { c >> 1 }; + } + *entry = c; + } + table +} + +fn crc32_update(table: &[u32; 256], crc: u32, data: &[u8]) -> u32 { + let mut crc = crc ^ 0xFFFFFFFF; + for &byte in data { + crc = table[((crc ^ byte as u32) & 0xFF) as usize] ^ (crc >> 8); + } + crc ^ 0xFFFFFFFF +} + +/// ru_maxrss is KB on Linux but bytes on macOS; normalize to KB. +fn peak_rss_kb() -> libc::c_long { + let mut rusage: libc::rusage = unsafe { std::mem::zeroed() }; + unsafe { libc::getrusage(libc::RUSAGE_SELF, &mut rusage) }; + if cfg!(target_os = "macos") { + rusage.ru_maxrss / 1024 + } else { + rusage.ru_maxrss + } +} + +fn main() { + let args: Vec = std::env::args().collect(); + if args.len() != 6 { + eprintln!("Usage: {} ", args[0]); + eprintln!(" mode: unchunked | chunked | zstd | lz4"); + std::process::exit(1); + } + + let blob = load_blob(&args[5]); + let crc_table = crc32_table(); + + let filename = &args[1]; + let mode = &args[2]; + let is_mixed = args[4] == "mixed"; + + let opts = match mode.as_str() { + "unchunked" => WriteOptions::new() + .use_chunks(false) + .profile("bench") + .library("rust-bench"), + "chunked" => WriteOptions::new() + .compression(None) + .chunk_size(Some(786432)) + .profile("bench") + .library("rust-bench"), + "zstd" => WriteOptions::new() + .compression(Some(Compression::Zstd)) + .chunk_size(Some(786432)) + .profile("bench") + .library("rust-bench"), + "lz4" => WriteOptions::new() + .compression(Some(Compression::Lz4)) + .chunk_size(Some(786432)) + .profile("bench") + .library("rust-bench"), + _ => { + eprintln!("Unknown mode: {}", mode); + std::process::exit(1); + } + }; + + let file = std::fs::File::create(filename).expect("Failed to create file"); + let buf_writer = BufWriter::new(file); + let mut writer = opts.create(buf_writer).expect("Failed to create writer"); + + let schema_data = b"{\"type\":\"object\"}"; + let metadata = BTreeMap::new(); + + if is_mixed { + // ── Mixed-payload mode: simulate a 10-second robot recording ── + + // Channel definitions: (topic, schema_name, base_payload_sizes, period_ns, count) + struct ChanDef { + topic: &'static str, + schema_name: &'static str, + payload_sizes: &'static [usize], + period_ns: u64, + count: u64, + } + let chan_defs: [ChanDef; 5] = [ + ChanDef { topic: "/imu", schema_name: "IMU", payload_sizes: &[96], period_ns: 5_000_000, count: 2000 }, + ChanDef { topic: "/odom", schema_name: "Odometry", payload_sizes: &[296], period_ns: 20_000_000, count: 500 }, + ChanDef { topic: "/tf", schema_name: "TFMessage", payload_sizes: &[80, 160, 320, 800, 1600], period_ns: 10_000_000, count: 1000 }, + ChanDef { topic: "/lidar", schema_name: "PointCloud2", payload_sizes: &[230_400], period_ns: 100_000_000, count: 100 }, + ChanDef { topic: "/camera/compressed", schema_name: "CompressedImage", payload_sizes: &[524_288], period_ns: 66_666_667, count: 150 }, + ]; + + // Register schemas and channels (not timed) + let mut channel_ids: Vec = Vec::new(); + for def in &chan_defs { + let sid = writer + .add_schema(def.schema_name, "jsonschema", schema_data) + .expect("Failed to add schema"); + let cid = writer + .add_channel(sid, def.topic, "json", &metadata) + .expect("Failed to add channel"); + channel_ids.push(cid); + } + + // Pre-generate sorted schedule: (timestamp, channel_index) + let mut schedule: Vec<(u64, usize)> = Vec::new(); + for (ch_idx, def) in chan_defs.iter().enumerate() { + for msg_i in 0..def.count { + let timestamp = msg_i * def.period_ns; + schedule.push((timestamp, ch_idx)); + } + } + // Sort by timestamp; ties broken by channel index (stable sort preserves push order) + schedule.sort_by(|a, b| a.0.cmp(&b.0).then(a.1.cmp(&b.1))); + + // Not timed: CRC of the payload stream for cross-language verification + let mut payload_crc: u32 = 0; + { + let mut crc_seq: [u32; 5] = [0; 5]; + for (i, &(_, ch_idx)) in schedule.iter().enumerate() { + let def = &chan_defs[ch_idx]; + let seq = crc_seq[ch_idx]; + crc_seq[ch_idx] += 1; + let payload_size = def.payload_sizes[seq as usize % def.payload_sizes.len()]; + let off = payload_offset(i as u64); + payload_crc = crc32_update(&crc_table, payload_crc, &blob[off..off + payload_size]); + } + } + + // Per-channel sequence counters and message index (for cycling /tf sizes) + let mut seq_counters: [u32; 5] = [0; 5]; + + // ── Timed: message loop + finish ── + let start = Instant::now(); + + for (i, &(timestamp, ch_idx)) in schedule.iter().enumerate() { + let def = &chan_defs[ch_idx]; + let seq = seq_counters[ch_idx]; + seq_counters[ch_idx] += 1; + + let payload_size = def.payload_sizes[seq as usize % def.payload_sizes.len()]; + let off = payload_offset(i as u64); + let payload = &blob[off..off + payload_size]; + + writer + .write_to_known_channel( + &mcap::records::MessageHeader { + channel_id: channel_ids[ch_idx], + sequence: seq, + log_time: timestamp, + publish_time: timestamp, + }, + payload, + ) + .expect("Failed to write message"); + } + + writer.finish().expect("Failed to finish"); + + let elapsed = start.elapsed(); + let elapsed_ns = elapsed.as_nanos(); + let wall_sec = elapsed.as_secs_f64(); + + let file_size = std::fs::metadata(filename) + .expect("Failed to stat file") + .len(); + + let peak_rss_kb = peak_rss_kb(); + + println!( + "write\trust\t{}\t{}\t{}\t{}\t{}\t{:.6}\t{}\t{}", + mode, 3750, "mixed", file_size, elapsed_ns, wall_sec, peak_rss_kb, payload_crc + ); + } else { + // ── Fixed-payload mode (original behavior) ── + let num_messages: u64 = args[3].parse().expect("invalid num_messages"); + let payload_size: usize = args[4].parse().expect("invalid payload_size"); + assert!(payload_size <= MAX_PAYLOAD, "payload_size must be <= {}", MAX_PAYLOAD); + + let schema_id = writer + .add_schema("BenchMsg", "jsonschema", schema_data) + .expect("Failed to add schema"); + + let channel_id = writer + .add_channel(schema_id, "/bench", "json", &metadata) + .expect("Failed to add channel"); + + // Not timed: CRC of the payload stream for cross-language verification + let mut payload_crc: u32 = 0; + for i in 0..num_messages { + let off = payload_offset(i); + payload_crc = crc32_update(&crc_table, payload_crc, &blob[off..off + payload_size]); + } + + // Timed: message loop + finish + let start = Instant::now(); + + for i in 0..num_messages { + let log_time = i * 1000; + let off = payload_offset(i); + writer + .write_to_known_channel( + &mcap::records::MessageHeader { + channel_id, + sequence: i as u32, + log_time, + publish_time: log_time, + }, + &blob[off..off + payload_size], + ) + .expect("Failed to write message"); + } + + writer.finish().expect("Failed to finish"); + + let elapsed = start.elapsed(); + let elapsed_ns = elapsed.as_nanos(); + let wall_sec = elapsed.as_secs_f64(); + + let file_size = std::fs::metadata(filename) + .expect("Failed to stat file") + .len(); + + let peak_rss_kb = peak_rss_kb(); + + println!( + "write\trust\t{}\t{}\t{}\t{}\t{}\t{:.6}\t{}\t{}", + mode, num_messages, payload_size, file_size, elapsed_ns, wall_sec, peak_rss_kb, + payload_crc + ); + } +} diff --git a/benchmarking/typescript_bench/bench_read.ts b/benchmarking/typescript_bench/bench_read.ts new file mode 100644 index 000000000..3e5797b56 --- /dev/null +++ b/benchmarking/typescript_bench/bench_read.ts @@ -0,0 +1,116 @@ +#!/usr/bin/env -S npx tsx +/** + * MCAP read benchmark for TypeScript. + */ + +import { open, stat } from "fs/promises"; + +import { + McapIndexedReader, + McapStreamReader, +} from "../../typescript/core/src/index.ts"; +import { FileHandleReadable } from "../../typescript/nodejs/src/index.ts"; +import { loadDecompressHandlers } from "../../typescript/support/src/decompressHandlers.ts"; + +async function main(): Promise { + if (process.argv.length < 3 || process.argv.length > 7) { + process.stderr.write( + `Usage: ${process.argv[1]} [mode] [num_messages] [payload_size] [filter]\n` + + ` filter: topic | timerange | topic_timerange\n`, + ); + return 1; + } + + const filename = process.argv[2]!; + const mode = process.argv[3] ?? "unknown"; + const numMessagesStr = process.argv[4] ?? "0"; + const payloadSizeStr = process.argv[5] ?? "0"; + const filter = process.argv[6]; + + const decompressHandlers = await loadDecompressHandlers(); + + let msgCount = 0; + + // Time file open + reader creation + message iteration + const tStart = process.hrtime.bigint(); + + if (filter != null && filter !== "") { + // Filtered reads: use McapIndexedReader to leverage the chunk index + const fileHandle = await open(filename, "r"); + const reader = await McapIndexedReader.Initialize({ + readable: new FileHandleReadable(fileHandle), + decompressHandlers, + }); + + const readArgs: { + topics?: string[]; + startTime?: bigint; + endTime?: bigint; + } = {}; + if (filter === "topic") { + readArgs.topics = ["/imu"]; + } else if (filter === "timerange") { + readArgs.startTime = 3000000000n; + readArgs.endTime = 5000000000n; + } else if (filter === "topic_timerange") { + readArgs.topics = ["/lidar"]; + readArgs.startTime = 4000000000n; + readArgs.endTime = 6000000000n; + } + + for await (const message of reader.readMessages(readArgs)) { + if (message.data.length === 0) { + process.stderr.write("Empty message\n"); + } + msgCount++; + } + + await fileHandle.close(); + } else { + // Unfiltered reads: use McapStreamReader to handle all file types. + // Stream the file in chunks to keep memory bounded; append() copies + // into the reader's internal buffer, so the read buffer is reusable. + const fileHandle = await open(filename, "r"); + const reader = new McapStreamReader({ decompressHandlers }); + const chunk = new Uint8Array(1024 * 1024); + for (;;) { + const { bytesRead } = await fileHandle.read(chunk, 0, chunk.length); + if (bytesRead === 0) { + break; + } + reader.append(chunk.subarray(0, bytesRead)); + for (;;) { + const record = reader.nextRecord(); + if (record == null) { + break; + } + if (record.type === "Message") { + if (record.data.length === 0) { + process.stderr.write("Empty message\n"); + } + msgCount++; + } + } + } + await fileHandle.close(); + } + + const tEnd = process.hrtime.bigint(); + + const elapsedNs = tEnd - tStart; + const wallSec = Number(elapsedNs) / 1e9; + const fileSize = (await stat(filename)).size; + + const peakRssKb = process.resourceUsage().maxRSS; + + // TSV output: op lang mode num_msgs payload_size file_size elapsed_ns wall_sec peak_rss_kb msg_count + process.stdout.write( + `read\ttypescript\t${mode}\t${numMessagesStr}\t${payloadSizeStr}\t${fileSize}\t${elapsedNs}\t${wallSec.toFixed( + 6, + )}\t${peakRssKb}\t${msgCount}\n`, + ); + + return 0; +} + +main().then((code) => process.exit(code)); diff --git a/benchmarking/typescript_bench/bench_write.ts b/benchmarking/typescript_bench/bench_write.ts new file mode 100644 index 000000000..9fe03ccc8 --- /dev/null +++ b/benchmarking/typescript_bench/bench_write.ts @@ -0,0 +1,273 @@ +#!/usr/bin/env -S npx tsx +/** + * MCAP write benchmark for TypeScript. + */ + +import { open, readFile, stat } from "fs/promises"; +import { crc32 } from "zlib"; + +import { McapWriter } from "../../typescript/core/src/index.ts"; +import { FileHandleWritable } from "../../typescript/nodejs/src/index.ts"; + +// Shared payload blob parameters; must match gen_blob.py and the other +// language benches. Message i's payload is the window of the blob starting +// at (i * BLOB_STRIDE) % BLOB_WINDOW_SPAN, so all implementations feed +// identical bytes to their writers. +const BLOB_SIZE = 16777216; +const BLOB_MAX_PAYLOAD = 524288; +const BLOB_WINDOW_SPAN = BLOB_SIZE - BLOB_MAX_PAYLOAD; +const BLOB_STRIDE = 7919; + +function payloadOffset(msgIndex: number): number { + return (msgIndex * BLOB_STRIDE) % BLOB_WINDOW_SPAN; +} + +async function main(): Promise { + if (process.argv.length !== 7) { + process.stderr.write( + `Usage: ${process.argv[1]} \n`, + ); + return 1; + } + + const blob: Uint8Array = await readFile(process.argv[6]!); + if (blob.byteLength !== BLOB_SIZE) { + process.stderr.write( + `Blob file ${process.argv[6]!} is not exactly ${BLOB_SIZE} bytes\n`, + ); + return 1; + } + + const filename = process.argv[2]!; + const mode = process.argv[3]!; + const isMixed = process.argv[5] === "mixed"; + const numMessages = isMixed ? 3750 : parseInt(process.argv[4]!, 10); + const payloadSize = isMixed ? 0 : parseInt(process.argv[5]!, 10); + + if (!isMixed && (numMessages <= 0 || payloadSize <= 0)) { + process.stderr.write("num_messages and payload_size must be positive\n"); + return 1; + } + if (!isMixed && payloadSize > BLOB_MAX_PAYLOAD) { + process.stderr.write(`payload_size must be <= ${BLOB_MAX_PAYLOAD}\n`); + return 1; + } + + let useChunks = true; + let compressChunk: + | ((data: Uint8Array) => { + compression: string; + compressedData: Uint8Array; + }) + | undefined; + + if (mode === "unchunked") { + useChunks = false; + } else if (mode === "chunked") { + // chunked, no compression + } else if (mode === "zstd") { + const zstdMod = await import("@foxglove/wasm-zstd"); + const zstd = zstdMod.default; + await zstd.isLoaded; + compressChunk = (data: Uint8Array) => ({ + compression: "zstd", + compressedData: new Uint8Array(zstd.compress(data)), + }); + } else if (mode === "lz4") { + process.stderr.write( + "LZ4 compression is not available in the TypeScript MCAP library\n", + ); + return 1; + } else { + process.stderr.write(`Unknown mode: ${mode}\n`); + return 1; + } + + const fileHandle = await open(filename, "w"); + const writer = new McapWriter({ + writable: new FileHandleWritable(fileHandle), + useChunks, + chunkSize: 786432, + compressChunk, + }); + + await writer.start({ profile: "bench", library: "ts-bench" }); + + let tStart: bigint; + let tEnd: bigint; + let payloadCrc = 0; + + if (isMixed) { + // Mixed-payload mode: simulate a 10-second robot recording. + const channelDefs = [ + { + topic: "/imu", + schema: "IMU", + sizes: [96], + periodNs: 5_000_000n, + count: 2000, + }, + { + topic: "/odom", + schema: "Odometry", + sizes: [296], + periodNs: 20_000_000n, + count: 500, + }, + { + topic: "/tf", + schema: "TFMessage", + sizes: [80, 160, 320, 800, 1600], + periodNs: 10_000_000n, + count: 1000, + }, + { + topic: "/lidar", + schema: "PointCloud2", + sizes: [230400], + periodNs: 100_000_000n, + count: 100, + }, + { + topic: "/camera/compressed", + schema: "CompressedImage", + sizes: [524288], + periodNs: 66_666_667n, + count: 150, + }, + ]; + + const schemaData = new TextEncoder().encode('{"type":"object"}'); + const channelIds: number[] = []; + for (const def of channelDefs) { + const sid = await writer.registerSchema({ + name: def.schema, + encoding: "jsonschema", + data: schemaData, + }); + const cid = await writer.registerChannel({ + topic: def.topic, + schemaId: sid, + messageEncoding: "json", + metadata: new Map(), + }); + channelIds.push(cid); + } + + // Pre-generate sorted message schedule. + const schedule: { timestamp: bigint; channelIndex: number }[] = []; + for (let ci = 0; ci < channelDefs.length; ci++) { + const def = channelDefs[ci]!; + for (let m = 0; m < def.count; m++) { + schedule.push({ + timestamp: BigInt(m) * def.periodNs, + channelIndex: ci, + }); + } + } + schedule.sort((a, b) => { + if (a.timestamp < b.timestamp) return -1; + if (a.timestamp > b.timestamp) return 1; + return a.channelIndex - b.channelIndex; + }); + + // Not timed: CRC of the payload stream for cross-language verification. + { + const crcSeq = new Array(channelDefs.length).fill(0) as number[]; + for (let i = 0; i < schedule.length; i++) { + const ci = schedule[i]!.channelIndex; + const def = channelDefs[ci]!; + const seq = crcSeq[ci]!; + crcSeq[ci] = seq + 1; + const sz = def.sizes[seq % def.sizes.length]!; + const off = payloadOffset(i); + payloadCrc = crc32(blob.subarray(off, off + sz), payloadCrc); + } + } + + // Per-channel sequence counters for payload size cycling. + const chanSeq = new Array(channelDefs.length).fill(0) as number[]; + + // Time the message-writing loop + end. + tStart = process.hrtime.bigint(); + + for (let i = 0; i < schedule.length; i++) { + const msg = schedule[i]!; + const ci = msg.channelIndex; + const def = channelDefs[ci]!; + const seq = chanSeq[ci]!; + chanSeq[ci] = seq + 1; + const sz = def.sizes[seq % def.sizes.length]!; + const off = payloadOffset(i); + const data = blob.subarray(off, off + sz); + await writer.addMessage({ + channelId: channelIds[ci]!, + sequence: seq, + logTime: msg.timestamp, + publishTime: msg.timestamp, + data, + }); + } + + await writer.end(); + tEnd = process.hrtime.bigint(); + await fileHandle.close(); + } else { + // Fixed-payload mode. + const schemaId = await writer.registerSchema({ + name: "BenchMsg", + encoding: "jsonschema", + data: new TextEncoder().encode('{"type":"object"}'), + }); + + const channelId = await writer.registerChannel({ + topic: "/bench", + schemaId, + messageEncoding: "json", + metadata: new Map(), + }); + + // Not timed: CRC of the payload stream for cross-language verification. + for (let i = 0; i < numMessages; i++) { + const off = payloadOffset(i); + payloadCrc = crc32(blob.subarray(off, off + payloadSize), payloadCrc); + } + + // Time the message-writing loop + end + tStart = process.hrtime.bigint(); + + for (let i = 0; i < numMessages; i++) { + const logTime = BigInt(i) * 1000n; + const off = payloadOffset(i); + await writer.addMessage({ + channelId, + sequence: i, + logTime, + publishTime: logTime, + data: blob.subarray(off, off + payloadSize), + }); + } + + await writer.end(); + tEnd = process.hrtime.bigint(); + await fileHandle.close(); + } + + const elapsedNs = tEnd - tStart; + const wallSec = Number(elapsedNs) / 1e9; + const fileSize = (await stat(filename)).size; + + const peakRssKb = process.resourceUsage().maxRSS; + + // TSV output: op lang mode num_msgs payload_size file_size elapsed_ns wall_sec peak_rss_kb payload_crc32 + const payloadSizeStr = isMixed ? "mixed" : String(payloadSize); + process.stdout.write( + `write\ttypescript\t${mode}\t${numMessages}\t${payloadSizeStr}\t${fileSize}\t${elapsedNs}\t${wallSec.toFixed( + 6, + )}\t${peakRssKb}\t${payloadCrc}\n`, + ); + + return 0; +} + +main().then((code) => process.exit(code));