diff --git a/TASK_DETAILS.md b/TASK_DETAILS.md index b331f7d8..e1690821 100644 --- a/TASK_DETAILS.md +++ b/TASK_DETAILS.md @@ -204,7 +204,7 @@ We welcome new engineering problem ideas — even without complete verification Polarization-multiplexed holography - ComputerSystems + ComputerSystems MallocLab High-performance C memory allocator (utilization & throughput) @@ -212,6 +212,10 @@ We welcome new engineering problem ideas — even without complete verification DuckDBWorkloadOptimization Index / materialized-view selection and query rewriting on official DuckDB workloads + + AdaptiveCompressedTelemetryExecution + CPU-measured adaptive telemetry compression with lossless decoding, compressed-domain queries, and storage/compute cost trade-offs + EngDesign CY_03, WJ_01, XY_05, AM_02, AM_03, YJ_02, YJ_03 diff --git a/TASK_DETAILS_zh-CN.md b/TASK_DETAILS_zh-CN.md index 44d1b1fa..ff0fc982 100644 --- a/TASK_DETAILS_zh-CN.md +++ b/TASK_DETAILS_zh-CN.md @@ -204,7 +204,7 @@ Frontier-Eng 目前已覆盖以下领域的任务。每个任务均配有可运 偏振复用全息 - ComputerSystems + ComputerSystems MallocLab 高性能 C 动态内存分配器（utilization & throughput） @@ -212,6 +212,10 @@ Frontier-Eng 目前已覆盖以下领域的任务。每个任务均配有可运 DuckDBWorkloadOptimization 基于 DuckDB 官方 workload 的索引 / 物化视图选择与查询改写 + + AdaptiveCompressedTelemetryExecution + 实测 CPU 的自适应遥测数据压缩、无损解码、压缩态查询与存储/计算成本权衡 + EngDesign CY_03, WJ_01, XY_05, AM_02, AM_03, YJ_02, YJ_03 diff --git a/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/README.md b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/README.md new file mode 100644 index 00000000..5230e96d --- /dev/null +++ b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/README.md @@ -0,0 +1,79 @@ +# Adaptive Compressed Telemetry Execution + +This CPU benchmark co-designs a telemetry column format and the operators that +consume it. Candidate C++ code encodes five correlated columns into opaque blocks, +decodes them losslessly, and answers exact filters and conditional sums directly +from those blocks. The evaluator compiles and runs the candidate, measures real +CPU time, checks every decoded byte and query answer, and reports both absolute +performance and a public cost-per-logical-TiB objective. + +The task is motivated by the CPU and representation-efficiency requirements in the +[OpenTelemetry Logs Data Model](https://opentelemetry.io/docs/specs/otel/logs/data-model/) +and by recent work on composable, query-aware formats such as +[FastLanes](https://vldb.org/pvldb/vol18/p4629-afroozeh.pdf). + +## Files + +- `Task.md`: full API, workload, correctness, measurement, and scoring contract. +- `references/problem_config.json`: public execution, pricing, and scenario settings. +- `scripts/init.cpp`: editable raw-column starter implementation. +- `baseline/solution.cpp`: immutable copy of the raw starter. +- `baseline/result_log.txt`: recorded starter measurement. +- `verification/codec_api.h`: immutable candidate ABI. +- `verification/benchmark_driver.cpp`: immutable timed C++ driver. +- `verification/evaluator.py`: dataset generator, compiler, oracle, and scorer. +- `verification/test_evaluator.py`: end-to-end and integrity regression tests. +- `frontier_eval/`: unified-task metadata. + +## Requirements + +- Python 3.10 or newer; only the standard library is used. +- `g++` with C++20 support. +- A Linux-like host for CPU affinity and resource limits. + +No database, service, network access, accelerator, or task-specific Python package +is required. The driver pins each timed process to one available CPU. + +## Direct evaluation + +From this task directory: + +```bash +python verification/evaluator.py scripts/init.cpp +``` + +The uncompressed starter must be correct and scores exactly `1.0`. Measurements +vary by CPU, but the evaluator measures the immutable baseline in the same run for +every changed candidate. A score above `1.0` reduces the modeled monthly cost using +measured storage bytes and CPU time. + +Compilation deliberately uses `-march=native`. Absolute throughput, modeled cost, +and candidate-to-baseline ratios can therefore vary across CPU models and compiler +versions. Compare leaderboard runs only within the same evaluation host and +environment; the co-measured immutable baseline controls within-run comparisons. + +To retain detailed measurements: + +```bash +python verification/evaluator.py scripts/init.cpp \ + --metrics-out metrics.json --artifacts-out artifacts.json +``` + +## Unified evaluation + +From the repository root: + +```bash +python -m frontier_eval \ + task=unified \ + task.benchmark=ComputerSystems/AdaptiveCompressedTelemetryExecution \ + algorithm=openevolve \ + algorithm.iterations=0 +``` + +## Editing contract + +Only edit the code between `EVOLVE-BLOCK-START` and `EVOLVE-BLOCK-END` in +`scripts/init.cpp`. Preserve the function signatures in `verification/codec_api.h`. +The editable block may add standard or compiler-provided headers, helpers, metadata, +adaptive encoding selection, SIMD code, and any self-contained block representation. diff --git a/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/Task.md b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/Task.md new file mode 100644 index 00000000..8a02fcbf --- /dev/null +++ b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/Task.md @@ -0,0 +1,179 @@ +# Task: Adaptive Compressed Telemetry Execution + +## 1. Engineering problem + +Telemetry backends continuously ingest timestamped events and later scan them for +incident response, dashboards, retention, and audit queries. Storing every field as +an uncompressed 64-bit value is fast but expensive. Aggressive compression can +reduce storage traffic while making ingestion, full decoding, or selective queries +more CPU-intensive. The useful engineering problem is therefore not compression in +isolation: it is joint design of a block representation and the operators that use +that representation. + +The [OpenTelemetry Logs Data Model](https://opentelemetry.io/docs/specs/otel/logs/data-model/) +defines common timestamp, severity, resource, and attribute concepts and explicitly +calls out efficient serialization and space use. Modern systems research similarly +co-designs compression and execution: the +[FastLanes file format](https://vldb.org/pvldb/vol18/p4629-afroozeh.pdf) uses +composable data-parallel encodings, multi-column relationships, partial decoding, +and compressed-vector access, while +[MorphStore](https://arxiv.org/abs/2004.09350) studies complete analytical pipelines +over compressed representations. + +This benchmark presents the same open design space in a self-contained CPU task. +Candidate code is actually compiled and timed; there is no analytical prediction of +codec or query performance. + +## 2. Logical schema + +Every row contains five unsigned 64-bit columns: + +| Index | Column | Meaning | +|---:|---|---| +| 0 | `timestamp_ns` | Event time in Unix nanoseconds; nondecreasing within a scenario | +| 1 | `service_id` | Service or tenant identifier | +| 2 | `severity` | OpenTelemetry-style numeric severity | +| 3 | `duration_us` | Request or operation duration in microseconds | +| 4 | `payload_bytes` | Event payload size in bytes | + +Rows are divided into public `block_rows`-sized blocks. The candidate owns the +entire byte representation inside each block and may encode columns independently, +jointly, or adaptively. + +## 3. Candidate API + +The immutable declarations are in `verification/codec_api.h`: + +```cpp +bool encode_block(const BlockView& input, + std::vector& encoded); + +bool decode_block(const EncodedView& encoded, + const MutableBlock& output); + +std::uint64_t query_block(const EncodedView& encoded, + const QuerySpec& query); +``` + +`EncodedView` supplies the opaque bytes and row count to a fresh process. An encoded +block must therefore be self-contained; pointers, process-global state, source-file +references, and external side files are invalid representations. + +`decode_block` must reproduce every original value exactly. The evaluator allocates +the output columns. + +## 4. Exact query semantics + +All ranges are inclusive and arithmetic uses unsigned 64-bit values. Inputs are +chosen so correct aggregates do not overflow. + +- `CountEqual`: count rows where `filter_column == low`. +- `CountRange`: count rows where `low <= filter_column <= high`. +- `SumWhereEqual`: sum `value_column` where `filter_column == low`. +- `SumWhereRange`: sum `value_column` where + `low <= filter_column <= high`. + +A correct baseline may decode inside `query_block`. More advanced solutions can +answer from metadata, dictionaries, bit-packed vectors, learned residuals, or other +compressed representations. + +## 5. Workloads + +The evaluator generates three deterministic, multi-column datasets: + +1. `steady_api_traffic`: regular timestamps, low service cardinality, and mostly + informational events. +2. `bursty_multitenant_observability`: rotating hot tenants, burst boundaries, + higher cardinality, and localized error periods. +3. `incident_distribution_shift`: correlated changes in service popularity, + severity, duration, and payload during an incident. + +Each scenario uses a separate seed and contains eight exact query templates covering +equality, ranges, conditional sums, time windows, hot values, and low-selectivity +conditions. The generator seed can be overridden by the evaluator CLI for robustness +testing. + +## 6. Verification and measurement + +The evaluator performs the following steps: + +1. Enforce the source-size limit and compile the candidate with + `g++ -std=c++20 -O3 -march=native`. +2. Generate each raw dataset outside the timed region. +3. Run encoding in a CPU-pinned child process after warm-up. +4. Delete the raw input, then start a fresh process to decode the encoded file. +5. Compare the complete decoded binary output byte-for-byte with the original. +6. Start another fresh process, execute every query over every block, and compare + all integer results with an independent Python oracle. +7. Report median, minimum, and maximum wall time over the configured measured rounds. + +File loading, compilation, encoded-file writing, decoded-file writing, and result +serialization are excluded from the timed regions. Candidate allocation and the +actual encode/decode/query functions are included. Each child is pinned to one CPU, +so the elapsed time is a practical single-core performance measurement. A memory +limit, process CPU limit, wall timeout, encoded-size limit, and output-file limit are +enforced. + +The evaluator reports: + +- encode and decode GiB/s; +- compressed-query batch GiB/s; +- encoded bytes and compression ratio; +- per-phase timing samples and peak resident memory; and +- cost components and score for every scenario. + +Scenario failures are isolated. The evaluator records baseline, candidate, and +score errors on the affected scenario and continues through the remaining public +scenarios. Successful ratios remain available through `partial_combined_score` for +diagnosis, but any failed scenario keeps the official `valid` and `combined_score` +at zero so a candidate cannot benefit by skipping a difficult workload. + +## 7. Economic objective + +Let: + +- `r = encoded_bytes / logical_bytes`; +- `H_e`, `H_d`, and `H_q` be measured encode, full-decode, and fixed-query-batch + core-hours extrapolated to one logical TiB; +- `P_s` be the public storage price coefficient in dollars per GiB-month; +- `P_c` be the public CPU price coefficient in dollars per core-hour; and +- `D` and `Q` be scenario-specific monthly full-decode and query-batch counts. + +The scenario cost is: + +```text +storage_cost = 1024 * r * P_s +cpu_cost = P_c * (H_e + D * H_d + Q * H_q) +monthly_cost_per_logical_TiB = storage_cost + cpu_cost +``` + +The coefficients are versioned workload parameters, not claims about a particular +cloud vendor. They make the storage/CPU trade-off explicit while every performance +input to the formula comes from executed candidate code. + +For scenario `i`: + +```text +ratio_i = baseline_monthly_cost_i / candidate_monthly_cost_i +``` + +`combined_score` is the geometric mean of the three ratios. The raw starter is the +immutable baseline and scores exactly `1.0`; larger is better. + +## 8. Correctness and resource gates + +Any compilation error, timeout, crash, false return, oversized block, incomplete +decode, byte mismatch, query mismatch, or non-finite metric yields +`valid=0.0` and `combined_score=0.0`. + +The public limits are in `references/problem_config.json`: + +- candidate source: at most 1,000,000 bytes; +- encoded block: at most twice its raw bytes plus a small fixed allowance; +- process address space: 1 GiB; +- at most 64 candidate processes per real user where `RLIMIT_NPROC` is supported; +- one pinned CPU per timed process; and +- no external libraries, assets, services, or network access. + +The task permits portable C++, compiler intrinsics, adaptive block selection, and +self-contained metadata. Only the code inside the EVOLVE markers may change. diff --git a/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/baseline/result_log.txt b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/baseline/result_log.txt new file mode 100644 index 00000000..258fcbc6 --- /dev/null +++ b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/baseline/result_log.txt @@ -0,0 +1,62 @@ +AdaptiveCompressedTelemetryExecution baseline validation +Date: 2026-07-17 + +Host + OS: Linux 5.15.0-153-generic x86_64 + CPU: 2 x AMD EPYC 7313 (32 physical / 64 logical cores total) + Python: 3.10.12 + Compiler: g++ 11.4.0 + Compile target: -O3 -march=native (measurements are host-dependent) + Candidate process cap: RLIMIT_NPROC=64 where supported + GPU/container/external service: not used + +Direct command (from task directory) + python verification/evaluator.py scripts/init.cpp \ + --metrics-out metrics.json --artifacts-out artifacts.json + +Direct metrics + valid: 1.0 + combined_score: 1.0 + partial_combined_score: 1.0 + runtime_s: 6.591881075873971 + scenario_count: 3 + successful_scenario_count: 3 + failed_scenario_count: 0 + candidate_compression_ratio: 1.0000755310058593 + candidate_encode_gib_s: 15.891468137733515 + candidate_decode_gib_s: 36.09563159308368 + candidate_query_batch_gib_s: 3.204557804794705 + candidate_total_monthly_cost_per_tib: 74.20395162021418 + +Per-scenario frozen starter result + steady_api_traffic + compression_ratio=1.0000755310058593 + encode_gib_s=16.26411464925721 + decode_gib_s=35.57928773112305 + query_scanned_gib_s=26.527365035557704 + monthly_cost_per_logical_tib=25.678239670068667 + score_ratio=1.0 + bursty_multitenant_observability + compression_ratio=1.0000755310058593 + encode_gib_s=16.092824855684697 + decode_gib_s=36.23796783506379 + query_scanned_gib_s=27.275291745248598 + monthly_cost_per_logical_tib=24.575535172063333 + score_ratio=1.0 + incident_distribution_shift + compression_ratio=1.0000755310058593 + encode_gib_s=15.347782218457338 + decode_gib_s=36.48177895660945 + query_scanned_gib_s=23.440786398253035 + monthly_cost_per_logical_tib=23.950176778082177 + score_ratio=1.0 + +Development headroom check + A temporary fixed-width delta codec was validated by the same evaluator. It + stored timestamp deltas plus narrowed service, severity, duration, and payload + fields; it was not committed as the starter. The candidate passed byte-exact + decoding and all independent query oracles, scored 2.556401106833809, reduced + compression ratio from 1.0000755310058593 to 0.32512435913085935, and reduced + total modeled monthly cost from 74.1947212474412 to 29.255996084436042. Its + minimum per-scenario score was 2.248306349794879. This confirms substantial + headroom for richer adaptive formats and compressed execution strategies. diff --git a/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/baseline/solution.cpp b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/baseline/solution.cpp new file mode 100644 index 00000000..1e378093 --- /dev/null +++ b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/baseline/solution.cpp @@ -0,0 +1,96 @@ +#include "codec_api.h" + +// EVOLVE-BLOCK-START +#include +#include + +namespace frontier_telemetry { + +namespace { + +constexpr std::size_t column_index(Column column) { + return static_cast(column); +} + +bool raw_size(std::size_t count, std::size_t &bytes) { + constexpr std::size_t width = kColumnCount * sizeof(std::uint64_t); + if (count > std::numeric_limits::max() / width) { + return false; + } + bytes = count * width; + return true; +} + +std::uint64_t load_value(const EncodedView &encoded, Column column, + std::size_t row) { + const std::size_t offset = + (column_index(column) * encoded.count + row) * sizeof(std::uint64_t); + std::uint64_t value = 0; + std::memcpy(&value, encoded.data + offset, sizeof(value)); + return value; +} + +} // namespace + +bool encode_block(const BlockView &input, std::vector &encoded) { + std::size_t bytes = 0; + if (!raw_size(input.count, bytes)) { + return false; + } + encoded.resize(bytes); + for (std::size_t column = 0; column < kColumnCount; ++column) { + if (input.count != 0 && input.columns[column] == nullptr) { + return false; + } + std::memcpy(encoded.data() + column * input.count * sizeof(std::uint64_t), + input.columns[column], input.count * sizeof(std::uint64_t)); + } + return true; +} + +bool decode_block(const EncodedView &encoded, const MutableBlock &output) { + std::size_t expected = 0; + if (encoded.count != output.count || !raw_size(encoded.count, expected) || + encoded.size != expected) { + return false; + } + for (std::size_t column = 0; column < kColumnCount; ++column) { + if (output.count != 0 && output.columns[column] == nullptr) { + return false; + } + std::memcpy(output.columns[column], + encoded.data + column * output.count * sizeof(std::uint64_t), + output.count * sizeof(std::uint64_t)); + } + return true; +} + +std::uint64_t query_block(const EncodedView &encoded, const QuerySpec &query) { + std::uint64_t result = 0; + for (std::size_t row = 0; row < encoded.count; ++row) { + const std::uint64_t filter = + load_value(encoded, query.filter_column, row); + switch (query.kind) { + case QueryKind::CountEqual: + result += static_cast(filter == query.low); + break; + case QueryKind::CountRange: + result += + static_cast(filter >= query.low && filter <= query.high); + break; + case QueryKind::SumWhereEqual: + if (filter == query.low) { + result += load_value(encoded, query.value_column, row); + } + break; + case QueryKind::SumWhereRange: + if (filter >= query.low && filter <= query.high) { + result += load_value(encoded, query.value_column, row); + } + break; + } + } + return result; +} +} // namespace frontier_telemetry +// EVOLVE-BLOCK-END diff --git a/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/frontier_eval/agent_files.txt b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/frontier_eval/agent_files.txt new file mode 100644 index 00000000..1505e8b1 --- /dev/null +++ b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/frontier_eval/agent_files.txt @@ -0,0 +1,10 @@ +README.md +Task.md +scripts/init.cpp +verification/codec_api.h +verification/benchmark_driver.cpp +verification/evaluator.py +verification/test_evaluator.py +baseline/solution.cpp +references/problem_config.json +frontier_eval/constraints.txt diff --git a/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/frontier_eval/artifact_files.txt b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/frontier_eval/artifact_files.txt new file mode 100644 index 00000000..76dc893a --- /dev/null +++ b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/frontier_eval/artifact_files.txt @@ -0,0 +1,2 @@ +metrics.json +artifacts.json diff --git a/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/frontier_eval/candidate_destination.txt b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/frontier_eval/candidate_destination.txt new file mode 100644 index 00000000..d26dd4fb --- /dev/null +++ b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/frontier_eval/candidate_destination.txt @@ -0,0 +1 @@ +scripts/init.cpp diff --git a/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/frontier_eval/constraints.txt b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/frontier_eval/constraints.txt new file mode 100644 index 00000000..2e294554 --- /dev/null +++ b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/frontier_eval/constraints.txt @@ -0,0 +1,16 @@ +AdaptiveCompressedTelemetryExecution constraints: +1) Only modify code between `// EVOLVE-BLOCK-START` and + `// EVOLVE-BLOCK-END` in `scripts/init.cpp`; marker lines must remain intact. +2) Preserve the three function signatures declared in `verification/codec_api.h`: + `encode_block`, `decode_block`, and `query_block`. +3) Encoded blocks must be self-contained. Do not use external files, network access, + subprocesses, process-global source data, or pointers embedded in encoded bytes. +4) Decoding must reproduce all five uint64 columns exactly, and all four query kinds + must implement the inclusive semantics documented in `Task.md`. +5) Candidate source must stay below 1,000,000 bytes. Each encoded block must stay + below the configured size bound, and each timed process has a 1 GiB address-space + limit and a single pinned CPU. +6) Use self-contained C++20 and compiler-provided intrinsics only. No external + libraries, downloaded assets, accelerators, or services are available. +7) Do not modify the API, driver, evaluator, configuration, baseline, documentation, + or `frontier_eval` metadata; these paths are checked read-only. diff --git a/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/frontier_eval/copy_files.txt b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/frontier_eval/copy_files.txt new file mode 100644 index 00000000..9c558e35 --- /dev/null +++ b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/frontier_eval/copy_files.txt @@ -0,0 +1 @@ +. diff --git a/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/frontier_eval/eval_command.txt b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/frontier_eval/eval_command.txt new file mode 100644 index 00000000..527d23b5 --- /dev/null +++ b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/frontier_eval/eval_command.txt @@ -0,0 +1 @@ +{python} verification/evaluator.py {candidate} --metrics-out metrics.json --artifacts-out artifacts.json diff --git a/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/frontier_eval/eval_cwd.txt b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/frontier_eval/eval_cwd.txt new file mode 100644 index 00000000..9c558e35 --- /dev/null +++ b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/frontier_eval/eval_cwd.txt @@ -0,0 +1 @@ +. diff --git a/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/frontier_eval/initial_program.txt b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/frontier_eval/initial_program.txt new file mode 100644 index 00000000..d26dd4fb --- /dev/null +++ b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/frontier_eval/initial_program.txt @@ -0,0 +1 @@ +scripts/init.cpp diff --git a/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/frontier_eval/readonly_files.txt b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/frontier_eval/readonly_files.txt new file mode 100644 index 00000000..c287d998 --- /dev/null +++ b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/frontier_eval/readonly_files.txt @@ -0,0 +1,6 @@ +README.md +Task.md +baseline +references +verification +frontier_eval diff --git a/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/references/problem_config.json b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/references/problem_config.json new file mode 100644 index 00000000..17f5316c --- /dev/null +++ b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/references/problem_config.json @@ -0,0 +1,56 @@ +{ + "benchmark_id": "adaptive_compressed_telemetry_execution", + "evaluation_seed": 20260713, + "schema": [ + {"column": 0, "id": "timestamp_ns", "description": "event timestamp in Unix nanoseconds"}, + {"column": 1, "id": "service_id", "description": "numeric service or tenant identifier"}, + {"column": 2, "id": "severity", "description": "OpenTelemetry-style severity number"}, + {"column": 3, "id": "duration_us", "description": "request or operation duration in microseconds"}, + {"column": 4, "id": "payload_bytes", "description": "event payload size in bytes"} + ], + "execution": { + "block_rows": 4096, + "warmup_rounds": 2, + "measured_rounds": 7, + "compile_timeout_s": 60, + "run_timeout_s": 30, + "source_limit_bytes": 1000000, + "memory_limit_mb": 1024, + "max_processes": 64, + "max_encoded_ratio": 2.0 + }, + "pricing": { + "storage_dollars_per_gib_month": 0.023, + "cpu_dollars_per_core_hour": 0.10, + "description": "Public scenario coefficients, not a claim about any specific vendor price" + }, + "scenarios": [ + { + "id": "steady_api_traffic", + "pattern": "steady", + "rows": 262144, + "seed": 1103, + "monthly_full_decodes": 80.0, + "monthly_query_batches": 240.0, + "description": "Low-cardinality services, regular timestamps, mostly informational traffic" + }, + { + "id": "bursty_multitenant_observability", + "pattern": "bursty", + "rows": 262144, + "seed": 2207, + "monthly_full_decodes": 24.0, + "monthly_query_batches": 120.0, + "description": "Rotating hot tenants, burst boundaries, higher cardinality, and local incidents" + }, + { + "id": "incident_distribution_shift", + "pattern": "incident", + "rows": 262144, + "seed": 3301, + "monthly_full_decodes": 8.0, + "monthly_query_batches": 40.0, + "description": "A workload whose service, severity, latency, and payload distributions shift during an incident" + } + ] +} diff --git a/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/scripts/init.cpp b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/scripts/init.cpp new file mode 100644 index 00000000..1e378093 --- /dev/null +++ b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/scripts/init.cpp @@ -0,0 +1,96 @@ +#include "codec_api.h" + +// EVOLVE-BLOCK-START +#include +#include + +namespace frontier_telemetry { + +namespace { + +constexpr std::size_t column_index(Column column) { + return static_cast(column); +} + +bool raw_size(std::size_t count, std::size_t &bytes) { + constexpr std::size_t width = kColumnCount * sizeof(std::uint64_t); + if (count > std::numeric_limits::max() / width) { + return false; + } + bytes = count * width; + return true; +} + +std::uint64_t load_value(const EncodedView &encoded, Column column, + std::size_t row) { + const std::size_t offset = + (column_index(column) * encoded.count + row) * sizeof(std::uint64_t); + std::uint64_t value = 0; + std::memcpy(&value, encoded.data + offset, sizeof(value)); + return value; +} + +} // namespace + +bool encode_block(const BlockView &input, std::vector &encoded) { + std::size_t bytes = 0; + if (!raw_size(input.count, bytes)) { + return false; + } + encoded.resize(bytes); + for (std::size_t column = 0; column < kColumnCount; ++column) { + if (input.count != 0 && input.columns[column] == nullptr) { + return false; + } + std::memcpy(encoded.data() + column * input.count * sizeof(std::uint64_t), + input.columns[column], input.count * sizeof(std::uint64_t)); + } + return true; +} + +bool decode_block(const EncodedView &encoded, const MutableBlock &output) { + std::size_t expected = 0; + if (encoded.count != output.count || !raw_size(encoded.count, expected) || + encoded.size != expected) { + return false; + } + for (std::size_t column = 0; column < kColumnCount; ++column) { + if (output.count != 0 && output.columns[column] == nullptr) { + return false; + } + std::memcpy(output.columns[column], + encoded.data + column * output.count * sizeof(std::uint64_t), + output.count * sizeof(std::uint64_t)); + } + return true; +} + +std::uint64_t query_block(const EncodedView &encoded, const QuerySpec &query) { + std::uint64_t result = 0; + for (std::size_t row = 0; row < encoded.count; ++row) { + const std::uint64_t filter = + load_value(encoded, query.filter_column, row); + switch (query.kind) { + case QueryKind::CountEqual: + result += static_cast(filter == query.low); + break; + case QueryKind::CountRange: + result += + static_cast(filter >= query.low && filter <= query.high); + break; + case QueryKind::SumWhereEqual: + if (filter == query.low) { + result += load_value(encoded, query.value_column, row); + } + break; + case QueryKind::SumWhereRange: + if (filter >= query.low && filter <= query.high) { + result += load_value(encoded, query.value_column, row); + } + break; + } + } + return result; +} +} // namespace frontier_telemetry +// EVOLVE-BLOCK-END diff --git a/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/verification/benchmark_driver.cpp b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/verification/benchmark_driver.cpp new file mode 100644 index 00000000..34fda38d --- /dev/null +++ b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/verification/benchmark_driver.cpp @@ -0,0 +1,597 @@ +#include "codec_api.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace ft = frontier_telemetry; + +namespace { + +constexpr std::array kRawMagic{'T', 'E', 'L', 'R', 'A', 'W', '0', '1'}; +constexpr std::array kEncodedMagic{'T', 'E', 'L', 'C', 'M', 'P', '0', '1'}; + +struct RawDataset { + std::size_t rows{}; + std::array, ft::kColumnCount> columns; +}; + +struct EncodedBlock { + std::size_t count{}; + std::vector bytes; +}; + +struct EncodedDataset { + std::size_t rows{}; + std::size_t block_rows{}; + std::vector blocks; +}; + +struct NamedQuery { + std::string name; + ft::QuerySpec spec; +}; + +using Clock = std::chrono::steady_clock; + +std::uint64_t load_le64(const std::uint8_t *p) { + std::uint64_t value = 0; + for (unsigned shift = 0; shift < 64; shift += 8) { + value |= static_cast(*p++) << shift; + } + return value; +} + +std::uint32_t load_le32(const std::uint8_t *p) { + std::uint32_t value = 0; + for (unsigned shift = 0; shift < 32; shift += 8) { + value |= static_cast(*p++) << shift; + } + return value; +} + +void write_le64(std::ostream &out, std::uint64_t value) { + std::array bytes{}; + for (unsigned index = 0; index < bytes.size(); ++index) { + bytes[index] = static_cast((value >> (index * 8)) & 0xffU); + } + out.write(bytes.data(), static_cast(bytes.size())); +} + +void write_le32(std::ostream &out, std::uint32_t value) { + std::array bytes{}; + for (unsigned index = 0; index < bytes.size(); ++index) { + bytes[index] = static_cast((value >> (index * 8)) & 0xffU); + } + out.write(bytes.data(), static_cast(bytes.size())); +} + +std::vector read_all(const std::string &path) { + std::ifstream in(path, std::ios::binary | std::ios::ate); + if (!in) { + throw std::runtime_error("cannot open input: " + path); + } + const std::streamoff end = in.tellg(); + if (end < 0) { + throw std::runtime_error("cannot determine input size: " + path); + } + std::vector bytes(static_cast(end)); + in.seekg(0); + if (!bytes.empty()) { + in.read(reinterpret_cast(bytes.data()), + static_cast(bytes.size())); + } + if (!in) { + throw std::runtime_error("failed while reading input: " + path); + } + return bytes; +} + +RawDataset read_raw(const std::string &path) { + const std::vector bytes = read_all(path); + if (bytes.size() < 16 || + !std::equal(kRawMagic.begin(), kRawMagic.end(), bytes.begin())) { + throw std::runtime_error("invalid raw dataset header"); + } + const std::uint64_t rows64 = load_le64(bytes.data() + 8); + if (rows64 > std::numeric_limits::max()) { + throw std::runtime_error("raw dataset row count is too large"); + } + const std::size_t rows = static_cast(rows64); + if (rows > (std::numeric_limits::max() - 16) / + (ft::kColumnCount * sizeof(std::uint64_t))) { + throw std::runtime_error("raw dataset size overflows"); + } + const std::size_t expected = + 16 + rows * ft::kColumnCount * sizeof(std::uint64_t); + if (bytes.size() != expected) { + throw std::runtime_error("raw dataset length mismatch"); + } + + RawDataset dataset; + dataset.rows = rows; + const std::uint8_t *cursor = bytes.data() + 16; + for (auto &column : dataset.columns) { + column.resize(rows); + for (std::size_t row = 0; row < rows; ++row) { + column[row] = load_le64(cursor); + cursor += sizeof(std::uint64_t); + } + } + return dataset; +} + +void write_raw(const std::string &path, const RawDataset &dataset) { + std::ofstream out(path, std::ios::binary | std::ios::trunc); + if (!out) { + throw std::runtime_error("cannot open decoded output: " + path); + } + out.write(kRawMagic.data(), static_cast(kRawMagic.size())); + write_le64(out, static_cast(dataset.rows)); + for (const auto &column : dataset.columns) { + if (column.size() != dataset.rows) { + throw std::runtime_error("decoded column length mismatch"); + } + for (const std::uint64_t value : column) { + write_le64(out, value); + } + } + if (!out) { + throw std::runtime_error("failed while writing decoded output"); + } +} + +void write_encoded(const std::string &path, const EncodedDataset &dataset) { + std::ofstream out(path, std::ios::binary | std::ios::trunc); + if (!out) { + throw std::runtime_error("cannot open encoded output: " + path); + } + out.write(kEncodedMagic.data(), + static_cast(kEncodedMagic.size())); + write_le64(out, static_cast(dataset.rows)); + write_le32(out, static_cast(dataset.block_rows)); + write_le32(out, static_cast(dataset.blocks.size())); + for (const EncodedBlock &block : dataset.blocks) { + write_le32(out, static_cast(block.count)); + write_le64(out, static_cast(block.bytes.size())); + if (!block.bytes.empty()) { + out.write(reinterpret_cast(block.bytes.data()), + static_cast(block.bytes.size())); + } + } + if (!out) { + throw std::runtime_error("failed while writing encoded output"); + } +} + +EncodedDataset read_encoded(const std::string &path) { + const std::vector bytes = read_all(path); + if (bytes.size() < 24 || + !std::equal(kEncodedMagic.begin(), kEncodedMagic.end(), bytes.begin())) { + throw std::runtime_error("invalid encoded dataset header"); + } + const std::uint8_t *cursor = bytes.data() + 8; + const std::uint8_t *end = bytes.data() + bytes.size(); + EncodedDataset dataset; + dataset.rows = static_cast(load_le64(cursor)); + cursor += 8; + dataset.block_rows = static_cast(load_le32(cursor)); + cursor += 4; + const std::size_t block_count = static_cast(load_le32(cursor)); + cursor += 4; + if (dataset.block_rows == 0 || block_count == 0) { + throw std::runtime_error("encoded dataset has no blocks"); + } + dataset.blocks.reserve(block_count); + std::size_t total_rows = 0; + for (std::size_t index = 0; index < block_count; ++index) { + if (static_cast(end - cursor) < 12) { + throw std::runtime_error("truncated encoded block header"); + } + EncodedBlock block; + block.count = static_cast(load_le32(cursor)); + cursor += 4; + const std::uint64_t size64 = load_le64(cursor); + cursor += 8; + if (size64 > static_cast(end - cursor)) { + throw std::runtime_error("truncated encoded block payload"); + } + const std::size_t size = static_cast(size64); + block.bytes.assign(cursor, cursor + size); + cursor += size; + if (block.count == 0 || block.count > dataset.block_rows) { + throw std::runtime_error("invalid encoded block row count"); + } + total_rows += block.count; + dataset.blocks.push_back(std::move(block)); + } + if (cursor != end || total_rows != dataset.rows) { + throw std::runtime_error("encoded dataset totals do not match header"); + } + return dataset; +} + +std::uint64_t file_size_bytes(const std::string &path) { + std::ifstream in(path, std::ios::binary | std::ios::ate); + if (!in) { + throw std::runtime_error("cannot inspect file size: " + path); + } + const std::streamoff size = in.tellg(); + if (size < 0) { + throw std::runtime_error("invalid file size: " + path); + } + return static_cast(size); +} + +double percentile_median(std::vector values) { + if (values.empty()) { + throw std::runtime_error("no timing samples were collected"); + } + std::sort(values.begin(), values.end()); + const std::size_t middle = values.size() / 2; + if (values.size() % 2 == 1) { + return values[middle]; + } + return (values[middle - 1] + values[middle]) / 2.0; +} + +long peak_rss_kib() { + rusage usage{}; + if (getrusage(RUSAGE_SELF, &usage) != 0) { + return 0; + } + return usage.ru_maxrss; +} + +void add_timing_metrics(std::map &metrics, + const std::vector &samples) { + metrics["median_ns"] = percentile_median(samples); + metrics["min_ns"] = *std::min_element(samples.begin(), samples.end()); + metrics["max_ns"] = *std::max_element(samples.begin(), samples.end()); + metrics["timing_rounds"] = static_cast(samples.size()); + for (std::size_t index = 0; index < samples.size(); ++index) { + metrics["sample_" + std::to_string(index) + "_ns"] = samples[index]; + } +} + +void write_metrics(const std::string &path, + const std::map &metrics) { + std::ofstream out(path, std::ios::trunc); + if (!out) { + throw std::runtime_error("cannot open metrics output: " + path); + } + out << std::setprecision(17); + for (const auto &[key, value] : metrics) { + if (!std::isfinite(value)) { + throw std::runtime_error("non-finite metric: " + key); + } + out << key << '=' << value << '\n'; + } +} + +std::size_t parse_size(const char *text, const std::string &label) { + try { + const unsigned long long value = std::stoull(text); + if (value == 0 || value > std::numeric_limits::max()) { + throw std::out_of_range("range"); + } + return static_cast(value); + } catch (const std::exception &) { + throw std::runtime_error("invalid " + label + ": " + text); + } +} + +double parse_positive_double(const char *text, const std::string &label) { + try { + const double value = std::stod(text); + if (!std::isfinite(value) || value <= 0.0) { + throw std::out_of_range("range"); + } + return value; + } catch (const std::exception &) { + throw std::runtime_error("invalid " + label + ": " + text); + } +} + +ft::BlockView block_view(const RawDataset &dataset, std::size_t offset, + std::size_t count) { + ft::BlockView view; + view.count = count; + for (std::size_t column = 0; column < ft::kColumnCount; ++column) { + view.columns[column] = dataset.columns[column].data() + offset; + } + return view; +} + +EncodedDataset encode_all(const RawDataset &raw, std::size_t block_rows, + double max_encoded_ratio) { + EncodedDataset result; + result.rows = raw.rows; + result.block_rows = block_rows; + for (std::size_t offset = 0; offset < raw.rows; offset += block_rows) { + const std::size_t count = std::min(block_rows, raw.rows - offset); + EncodedBlock block; + block.count = count; + if (!ft::encode_block(block_view(raw, offset, count), block.bytes)) { + throw std::runtime_error("candidate encode_block returned false"); + } + const double raw_bytes = + static_cast(count * ft::kColumnCount * sizeof(std::uint64_t)); + const double allowed = raw_bytes * max_encoded_ratio + 4096.0; + if (static_cast(block.bytes.size()) > allowed) { + throw std::runtime_error("candidate encoded block exceeds size limit"); + } + result.blocks.push_back(std::move(block)); + } + return result; +} + +int run_encode(int argc, char **argv) { + if (argc != 9) { + throw std::runtime_error( + "encode usage: driver encode RAW ENCODED METRICS BLOCK_ROWS WARMUP " + "ROUNDS MAX_RATIO"); + } + const RawDataset raw = read_raw(argv[2]); + const std::size_t block_rows = parse_size(argv[5], "block rows"); + const std::size_t warmup = parse_size(argv[6], "warmup rounds"); + const std::size_t rounds = parse_size(argv[7], "measured rounds"); + const double max_ratio = parse_positive_double(argv[8], "max encoded ratio"); + + for (std::size_t index = 0; index < warmup; ++index) { + EncodedDataset ignored = encode_all(raw, block_rows, max_ratio); + if (ignored.blocks.empty()) { + throw std::runtime_error("candidate produced no encoded blocks"); + } + } + + std::vector samples; + samples.reserve(rounds); + EncodedDataset final_result; + for (std::size_t index = 0; index < rounds; ++index) { + const auto started = Clock::now(); + EncodedDataset current = encode_all(raw, block_rows, max_ratio); + const auto finished = Clock::now(); + samples.push_back(static_cast( + std::chrono::duration_cast(finished - started) + .count())); + final_result = std::move(current); + } + write_encoded(argv[3], final_result); + + const double logical_bytes = static_cast( + raw.rows * ft::kColumnCount * sizeof(std::uint64_t)); + const double median_ns = percentile_median(samples); + std::map metrics{ + {"rows", static_cast(raw.rows)}, + {"block_count", static_cast(final_result.blocks.size())}, + {"logical_bytes", logical_bytes}, + {"encoded_bytes", static_cast(file_size_bytes(argv[3]))}, + {"throughput_gib_s", logical_bytes / (median_ns * (1ULL << 30) / 1e9)}, + {"peak_rss_kib", static_cast(peak_rss_kib())}, + }; + add_timing_metrics(metrics, samples); + write_metrics(argv[4], metrics); + return 0; +} + +RawDataset allocate_decoded(std::size_t rows) { + RawDataset output; + output.rows = rows; + for (auto &column : output.columns) { + column.resize(rows); + } + return output; +} + +void decode_all(const EncodedDataset &encoded, RawDataset &output) { + std::size_t offset = 0; + for (const EncodedBlock &block : encoded.blocks) { + ft::MutableBlock destination; + destination.count = block.count; + for (std::size_t column = 0; column < ft::kColumnCount; ++column) { + destination.columns[column] = output.columns[column].data() + offset; + } + const ft::EncodedView view{block.bytes.data(), block.bytes.size(), block.count}; + if (!ft::decode_block(view, destination)) { + throw std::runtime_error("candidate decode_block returned false"); + } + offset += block.count; + } + if (offset != output.rows) { + throw std::runtime_error("decoded row count mismatch"); + } +} + +int run_decode(int argc, char **argv) { + if (argc != 7) { + throw std::runtime_error( + "decode usage: driver decode ENCODED RAW_OUT METRICS WARMUP ROUNDS"); + } + const EncodedDataset encoded = read_encoded(argv[2]); + RawDataset output = allocate_decoded(encoded.rows); + const std::size_t warmup = parse_size(argv[5], "warmup rounds"); + const std::size_t rounds = parse_size(argv[6], "measured rounds"); + + for (std::size_t index = 0; index < warmup; ++index) { + decode_all(encoded, output); + } + std::vector samples; + samples.reserve(rounds); + for (std::size_t index = 0; index < rounds; ++index) { + const auto started = Clock::now(); + decode_all(encoded, output); + const auto finished = Clock::now(); + samples.push_back(static_cast( + std::chrono::duration_cast(finished - started) + .count())); + } + write_raw(argv[3], output); + + const double logical_bytes = static_cast( + encoded.rows * ft::kColumnCount * sizeof(std::uint64_t)); + const double median_ns = percentile_median(samples); + std::map metrics{ + {"rows", static_cast(encoded.rows)}, + {"logical_bytes", logical_bytes}, + {"throughput_gib_s", logical_bytes / (median_ns * (1ULL << 30) / 1e9)}, + {"peak_rss_kib", static_cast(peak_rss_kib())}, + }; + add_timing_metrics(metrics, samples); + write_metrics(argv[4], metrics); + return 0; +} + +ft::Column parse_column(std::uint32_t value) { + if (value >= ft::kColumnCount) { + throw std::runtime_error("query references an unknown column"); + } + return static_cast(value); +} + +std::vector read_queries(const std::string &path) { + std::ifstream in(path); + if (!in) { + throw std::runtime_error("cannot open query file: " + path); + } + std::vector queries; + std::string line; + while (std::getline(in, line)) { + if (line.empty() || line.front() == '#') { + continue; + } + std::istringstream row(line); + NamedQuery query; + std::uint32_t kind = 0; + std::uint32_t filter = 0; + std::uint32_t value = 0; + if (!(row >> query.name >> kind >> filter >> value >> query.spec.low >> + query.spec.high)) { + throw std::runtime_error("invalid query row: " + line); + } + if (kind > static_cast(ft::QueryKind::SumWhereRange)) { + throw std::runtime_error("query uses an unknown kind"); + } + query.spec.kind = static_cast(kind); + query.spec.filter_column = parse_column(filter); + query.spec.value_column = parse_column(value); + queries.push_back(std::move(query)); + } + if (queries.empty()) { + throw std::runtime_error("query file contains no queries"); + } + return queries; +} + +std::vector execute_queries( + const EncodedDataset &encoded, const std::vector &queries) { + std::vector results(queries.size(), 0); + for (std::size_t query_index = 0; query_index < queries.size(); ++query_index) { + for (const EncodedBlock &block : encoded.blocks) { + const ft::EncodedView view{block.bytes.data(), block.bytes.size(), block.count}; + results[query_index] += ft::query_block(view, queries[query_index].spec); + } + } + return results; +} + +void write_query_results(const std::string &path, + const std::vector &queries, + const std::vector &results) { + if (queries.size() != results.size()) { + throw std::runtime_error("query result count mismatch"); + } + std::ofstream out(path, std::ios::trunc); + if (!out) { + throw std::runtime_error("cannot open query result output: " + path); + } + for (std::size_t index = 0; index < queries.size(); ++index) { + out << queries[index].name << '\t' << results[index] << '\n'; + } +} + +int run_query(int argc, char **argv) { + if (argc != 8) { + throw std::runtime_error( + "query usage: driver query ENCODED QUERIES RESULTS METRICS WARMUP " + "ROUNDS"); + } + const EncodedDataset encoded = read_encoded(argv[2]); + const std::vector queries = read_queries(argv[3]); + const std::size_t warmup = parse_size(argv[6], "warmup rounds"); + const std::size_t rounds = parse_size(argv[7], "measured rounds"); + + for (std::size_t index = 0; index < warmup; ++index) { + const std::vector ignored = execute_queries(encoded, queries); + if (ignored.size() != queries.size()) { + throw std::runtime_error("candidate query warmup failed"); + } + } + std::vector samples; + samples.reserve(rounds); + std::vector final_results; + for (std::size_t index = 0; index < rounds; ++index) { + const auto started = Clock::now(); + std::vector current = execute_queries(encoded, queries); + const auto finished = Clock::now(); + samples.push_back(static_cast( + std::chrono::duration_cast(finished - started) + .count())); + final_results = std::move(current); + } + write_query_results(argv[4], queries, final_results); + + const double logical_bytes = static_cast( + encoded.rows * ft::kColumnCount * sizeof(std::uint64_t)); + const double median_ns = percentile_median(samples); + std::map metrics{ + {"rows", static_cast(encoded.rows)}, + {"logical_bytes", logical_bytes}, + {"query_count", static_cast(queries.size())}, + {"scanned_gib_s", + logical_bytes * static_cast(queries.size()) / + (median_ns * (1ULL << 30) / 1e9)}, + {"peak_rss_kib", static_cast(peak_rss_kib())}, + }; + add_timing_metrics(metrics, samples); + write_metrics(argv[5], metrics); + return 0; +} + +} // namespace + +int main(int argc, char **argv) { + try { + if (argc < 2) { + throw std::runtime_error("missing driver mode"); + } + const std::string mode = argv[1]; + if (mode == "encode") { + return run_encode(argc, argv); + } + if (mode == "decode") { + return run_decode(argc, argv); + } + if (mode == "query") { + return run_query(argc, argv); + } + throw std::runtime_error("unknown driver mode: " + mode); + } catch (const std::exception &error) { + std::cerr << "benchmark_driver: " << error.what() << '\n'; + return 2; + } +} diff --git a/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/verification/codec_api.h b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/verification/codec_api.h new file mode 100644 index 00000000..aa8cac7a --- /dev/null +++ b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/verification/codec_api.h @@ -0,0 +1,62 @@ +#pragma once + +#include +#include +#include +#include + +namespace frontier_telemetry { + +enum class Column : std::uint32_t { + TimestampNs = 0, + ServiceId = 1, + Severity = 2, + DurationUs = 3, + PayloadBytes = 4, +}; + +constexpr std::size_t kColumnCount = 5; + +struct BlockView { + std::size_t count{}; + std::array columns{}; +}; + +struct MutableBlock { + std::size_t count{}; + std::array columns{}; +}; + +struct EncodedView { + const std::uint8_t *data{}; + std::size_t size{}; + std::size_t count{}; +}; + +enum class QueryKind : std::uint32_t { + CountEqual = 0, + CountRange = 1, + SumWhereEqual = 2, + SumWhereRange = 3, +}; + +struct QuerySpec { + QueryKind kind{QueryKind::CountEqual}; + Column filter_column{Column::TimestampNs}; + Column value_column{Column::TimestampNs}; + std::uint64_t low{}; + std::uint64_t high{}; +}; + +// The candidate owns the encoded block format. The immutable driver passes the +// row count separately, so the format does not need to repeat it. +bool encode_block(const BlockView &input, std::vector &encoded); + +// Decode every value into the supplied, preallocated column buffers. +bool decode_block(const EncodedView &encoded, const MutableBlock &output); + +// Execute one exact query on an encoded block. A candidate may decode internally, +// but faster solutions can operate directly on their compressed representation. +std::uint64_t query_block(const EncodedView &encoded, const QuerySpec &query); + +} // namespace frontier_telemetry diff --git a/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/verification/evaluator.py b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/verification/evaluator.py new file mode 100644 index 00000000..b032fb82 --- /dev/null +++ b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/verification/evaluator.py @@ -0,0 +1,1063 @@ +"""Compile and benchmark adaptive compressed telemetry execution candidates.""" + +from __future__ import annotations + +import argparse +import array +import hashlib +import json +import math +import os +import random +import resource +import shutil +import struct +import subprocess +import sys +import tempfile +import time +from collections import Counter +from pathlib import Path +from typing import Any + + +TASK_DIR = Path(__file__).resolve().parents[1] +VERIFICATION_DIR = TASK_DIR / "verification" +DRIVER_SOURCE = VERIFICATION_DIR / "benchmark_driver.cpp" +API_HEADER = VERIFICATION_DIR / "codec_api.h" +BASELINE_SOURCE = TASK_DIR / "baseline" / "solution.cpp" +PROBLEM_PATH = TASK_DIR / "references" / "problem_config.json" +RAW_MAGIC = b"TELRAW01" +UINT64_BYTES = 8 +COLUMN_COUNT = 5 +TIB_BYTES = 1 << 40 +MAX_TEXT_CAPTURE = 12_000 +START_MARKER = "// EVOLVE-BLOCK-START" +END_MARKER = "// EVOLVE-BLOCK-END" + + +class EvaluationFailure(RuntimeError): + def __init__(self, message: str, *, timeout: bool = False) -> None: + super().__init__(message) + self.timeout = timeout + + +def _reject_constant(value: str) -> None: + raise ValueError(f"non-finite JSON constant is not allowed: {value}") + + +def _load_json(path: Path) -> Any: + return json.loads( + path.read_text(encoding="utf-8"), + parse_constant=_reject_constant, + ) + + +def _object(value: Any, label: str) -> dict[str, Any]: + if not isinstance(value, dict): + raise ValueError(f"{label} must be an object") + return value + + +def _positive_int(value: Any, label: str) -> int: + if isinstance(value, bool) or not isinstance(value, int) or value <= 0: + raise ValueError(f"{label} must be a positive integer") + return value + + +def _positive_number(value: Any, label: str) -> float: + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise ValueError(f"{label} must be a positive finite number") + result = float(value) + if not math.isfinite(result) or result <= 0.0: + raise ValueError(f"{label} must be a positive finite number") + return result + + +def _load_problem() -> tuple[dict[str, Any], dict[str, Any], dict[str, Any], list[dict[str, Any]]]: + problem = _object(_load_json(PROBLEM_PATH), "problem configuration") + if str(problem.get("benchmark_id", "")) != "adaptive_compressed_telemetry_execution": + raise ValueError("unexpected benchmark_id") + execution = _object(problem.get("execution"), "execution") + pricing = _object(problem.get("pricing"), "pricing") + scenarios_raw = problem.get("scenarios") + if not isinstance(scenarios_raw, list) or not scenarios_raw: + raise ValueError("scenarios must be a non-empty list") + scenarios: list[dict[str, Any]] = [] + seen: set[str] = set() + for index, raw in enumerate(scenarios_raw): + scenario = _object(raw, f"scenarios[{index}]") + scenario_id = str(scenario.get("id", "")).strip() + if not scenario_id or scenario_id in seen: + raise ValueError(f"invalid or duplicate scenario id: {scenario_id!r}") + seen.add(scenario_id) + if str(scenario.get("pattern", "")) not in {"steady", "bursty", "incident"}: + raise ValueError(f"scenario {scenario_id} has an unsupported pattern") + _positive_int(scenario.get("rows"), f"scenario {scenario_id}.rows") + _positive_int(scenario.get("seed"), f"scenario {scenario_id}.seed") + _positive_number( + scenario.get("monthly_full_decodes"), + f"scenario {scenario_id}.monthly_full_decodes", + ) + _positive_number( + scenario.get("monthly_query_batches"), + f"scenario {scenario_id}.monthly_query_batches", + ) + scenarios.append(scenario) + + for key in ( + "block_rows", + "warmup_rounds", + "measured_rounds", + "compile_timeout_s", + "run_timeout_s", + "source_limit_bytes", + "memory_limit_mb", + "max_processes", + ): + _positive_int(execution.get(key), f"execution.{key}") + _positive_number(execution.get("max_encoded_ratio"), "execution.max_encoded_ratio") + _positive_number( + pricing.get("storage_dollars_per_gib_month"), + "pricing.storage_dollars_per_gib_month", + ) + _positive_number( + pricing.get("cpu_dollars_per_core_hour"), + "pricing.cpu_dollars_per_core_hour", + ) + return problem, execution, pricing, scenarios + + +def _severity(rng: random.Random, *, incident: bool = False) -> int: + roll = rng.randrange(1000) + if incident: + if roll < 430: + return 17 + if roll < 520: + return 21 + if roll < 720: + return 13 + return 9 + if roll < 885: + return 9 + if roll < 950: + return 13 + if roll < 993: + return 17 + return 21 + + +def _generate_columns(scenario: dict[str, Any], evaluation_seed: int) -> list[array.array[int]]: + rows = _positive_int(scenario.get("rows"), "scenario.rows") + seed = _positive_int(scenario.get("seed"), "scenario.seed") ^ evaluation_seed + rng = random.Random(seed) + pattern = str(scenario["pattern"]) + columns = [array.array("Q") for _ in range(COLUMN_COUNT)] + timestamp = 1_700_000_000_000_000_000 + (seed % 10_000_000) + payload_choices = (128, 256, 384, 512, 768, 1024, 2048, 4096, 8192) + + timestamp_col, service_col, severity_col, duration_col, payload_col = columns + for row in range(rows): + if pattern == "steady": + timestamp += 850_000 + rng.randrange(300_001) + service = (row * 17 + rng.randrange(5)) % 32 + severity = _severity(rng) + duration = 75 + rng.randrange(650) + (service % 5) * 23 + if severity >= 17: + duration += 800 + rng.randrange(2200) + payload = payload_choices[(service + rng.randrange(4)) % len(payload_choices)] + elif pattern == "bursty": + burst = row // 4096 + timestamp += 90_000 + rng.randrange(180_001) + if row % 4096 == 0: + timestamp += 8_000_000 + rng.randrange(20_000_000) + hot_service = (burst * 73 + 11) % 512 + service = hot_service if rng.randrange(100) < 72 else rng.randrange(512) + severity = _severity(rng, incident=(burst % 11 == 7 and row % 17 < 5)) + duration = 110 + rng.randrange(1600) + (service % 13) * 19 + if service == hot_service: + duration //= 2 + if severity >= 17: + duration += 2500 + rng.randrange(7000) + payload = payload_choices[(service ^ burst ^ rng.randrange(8)) % len(payload_choices)] + else: + phase = (row * 10) // rows + active_incident = 5 <= phase <= 7 + timestamp += 600_000 + rng.randrange(800_001) + if row % 16384 == 0: + timestamp += 50_000_000 + rng.randrange(100_000_000) + base_service = (row // 128 + phase * 29) % 96 + if active_incident and rng.randrange(100) < 62: + service = 77 + else: + service = (base_service + rng.randrange(9)) % 96 + severity = _severity(rng, incident=active_incident and service == 77) + duration = 60 + rng.randrange(900) + (service % 8) * 31 + if active_incident and service == 77: + duration += 8_000 + rng.randrange(45_000) + elif severity >= 17: + duration += 1_000 + rng.randrange(5_000) + payload = payload_choices[(phase + service + rng.randrange(5)) % len(payload_choices)] + + timestamp_col.append(timestamp) + service_col.append(service) + severity_col.append(severity) + duration_col.append(duration) + payload_col.append(payload) + return columns + + +def _quantile(values: array.array[int], fraction: float) -> int: + ordered = sorted(values) + index = min(len(ordered) - 1, max(0, int((len(ordered) - 1) * fraction))) + return int(ordered[index]) + + +def _build_queries(columns: list[array.array[int]]) -> list[dict[str, Any]]: + timestamp, service, severity, duration, _payload = columns + hot_service = int(Counter(service).most_common(1)[0][0]) + max_service = int(max(service)) + queries = [ + { + "name": "hot_service_count", + "kind": 0, + "filter_column": 1, + "value_column": 1, + "low": hot_service, + "high": hot_service, + }, + { + "name": "error_count", + "kind": 1, + "filter_column": 2, + "value_column": 2, + "low": 17, + "high": 24, + }, + { + "name": "hot_service_duration", + "kind": 2, + "filter_column": 1, + "value_column": 3, + "low": hot_service, + "high": hot_service, + }, + { + "name": "error_payload", + "kind": 3, + "filter_column": 2, + "value_column": 4, + "low": 17, + "high": 24, + }, + { + "name": "middle_window_count", + "kind": 1, + "filter_column": 0, + "value_column": 0, + "low": int(timestamp[len(timestamp) // 4]), + "high": int(timestamp[(3 * len(timestamp)) // 4]), + }, + { + "name": "low_service_duration", + "kind": 3, + "filter_column": 1, + "value_column": 3, + "low": 0, + "high": max(1, max_service // 8), + }, + { + "name": "middle_latency_count", + "kind": 1, + "filter_column": 3, + "value_column": 3, + "low": _quantile(duration, 0.40), + "high": _quantile(duration, 0.75), + }, + { + "name": "window_payload", + "kind": 3, + "filter_column": 0, + "value_column": 4, + "low": int(timestamp[len(timestamp) // 3]), + "high": int(timestamp[(2 * len(timestamp)) // 3]), + }, + ] + for query in queries: + filter_values = columns[int(query["filter_column"])] + value_values = columns[int(query["value_column"])] + kind = int(query["kind"]) + low = int(query["low"]) + high = int(query["high"]) + result = 0 + if kind == 0: + result = sum(1 for value in filter_values if value == low) + elif kind == 1: + result = sum(1 for value in filter_values if low <= value <= high) + elif kind == 2: + result = sum( + int(value_values[index]) + for index, value in enumerate(filter_values) + if value == low + ) + elif kind == 3: + result = sum( + int(value_values[index]) + for index, value in enumerate(filter_values) + if low <= value <= high + ) + query["expected"] = int(result) + return queries + + +def _write_raw(path: Path, columns: list[array.array[int]]) -> None: + rows = len(columns[0]) + if any(len(column) != rows for column in columns): + raise ValueError("generated columns have different lengths") + with path.open("wb") as handle: + handle.write(RAW_MAGIC) + handle.write(struct.pack(" None: + lines = [ + "\t".join( + str(value) + for value in ( + query["name"], + query["kind"], + query["filter_column"], + query["value_column"], + query["low"], + query["high"], + ) + ) + for query in queries + ] + path.write_text("\n".join(lines) + "\n", encoding="utf-8") + + +def _immutable_parts(source: str) -> tuple[str, str]: + if source.count(START_MARKER) != 1 or source.count(END_MARKER) != 1: + raise EvaluationFailure("candidate must contain exactly one EVOLVE marker pair") + start = source.index(START_MARKER) + start_line_end = source.find("\n", start) + if start_line_end < 0: + raise EvaluationFailure("EVOLVE-BLOCK-START must end with a newline") + editable_start = start_line_end + 1 + end = source.index(END_MARKER, editable_start) + if end <= editable_start: + raise EvaluationFailure("EVOLVE markers are out of order") + return source[:editable_start], source[end:] + + +def _validate_candidate_shell(candidate_path: Path, max_bytes: int) -> bytes: + if not candidate_path.is_file(): + raise EvaluationFailure(f"candidate source not found: {candidate_path}") + if candidate_path.stat().st_size > max_bytes: + raise EvaluationFailure(f"candidate source exceeds {max_bytes} bytes") + candidate = candidate_path.read_bytes() + baseline = BASELINE_SOURCE.read_bytes() + try: + candidate_text = candidate.decode("utf-8") + baseline_text = baseline.decode("utf-8") + except UnicodeDecodeError as exc: + raise EvaluationFailure(f"candidate source is not UTF-8: {exc}") from exc + if _immutable_parts(candidate_text) != _immutable_parts(baseline_text): + raise EvaluationFailure( + "code outside the EVOLVE block differs from the frozen shell" + ) + return candidate + + +def _compile( + compiler: str, + source: Path, + output: Path, + timeout_s: int, +) -> dict[str, Any]: + command = [ + compiler, + "-std=c++20", + "-O3", + "-DNDEBUG", + "-march=native", + "-I", + str(VERIFICATION_DIR), + str(DRIVER_SOURCE), + str(source), + "-o", + str(output), + ] + started = time.perf_counter() + try: + process = subprocess.run( + command, + cwd=str(TASK_DIR), + capture_output=True, + text=True, + timeout=timeout_s, + check=False, + ) + except subprocess.TimeoutExpired as exc: + raise EvaluationFailure( + f"compilation timed out after {timeout_s}s", + timeout=True, + ) from exc + result = { + "command": command, + "returncode": process.returncode, + "runtime_s": time.perf_counter() - started, + "stdout": process.stdout[-MAX_TEXT_CAPTURE:], + "stderr": process.stderr[-MAX_TEXT_CAPTURE:], + } + if process.returncode != 0 or not output.is_file(): + raise EvaluationFailure( + "candidate compilation failed:\n" + process.stderr[-MAX_TEXT_CAPTURE:] + ) + return result + + +def _preexec_limits( + memory_limit_mb: int, + maximum_processes: int, + cpu: int | None, + timeout_s: int, +) -> Any: + def apply() -> None: + memory_bytes = memory_limit_mb * 1024 * 1024 + resource.setrlimit(resource.RLIMIT_AS, (memory_bytes, memory_bytes)) + cpu_limit = max(2, int(math.ceil(timeout_s))) + resource.setrlimit(resource.RLIMIT_CPU, (cpu_limit, cpu_limit + 1)) + file_limit = 512 * 1024 * 1024 + resource.setrlimit(resource.RLIMIT_FSIZE, (file_limit, file_limit)) + try: + resource.setrlimit( + resource.RLIMIT_NPROC, (maximum_processes, maximum_processes) + ) + except (AttributeError, ValueError, OSError): + pass + if cpu is not None and hasattr(os, "sched_setaffinity"): + os.sched_setaffinity(0, {cpu}) + + return apply + + +def _parse_driver_metrics(path: Path) -> dict[str, float]: + if not path.is_file(): + raise EvaluationFailure("benchmark driver did not write its metrics file") + metrics: dict[str, float] = {} + for raw_line in path.read_text(encoding="utf-8").splitlines(): + if not raw_line.strip(): + continue + if "=" not in raw_line: + raise EvaluationFailure(f"invalid driver metric row: {raw_line}") + key, raw_value = raw_line.split("=", 1) + try: + value = float(raw_value) + except ValueError as exc: + raise EvaluationFailure(f"invalid driver metric {key}") from exc + if not math.isfinite(value): + raise EvaluationFailure(f"non-finite driver metric {key}") + metrics[key] = value + for required in ("median_ns", "min_ns", "max_ns", "logical_bytes"): + if metrics.get(required, 0.0) <= 0.0: + raise EvaluationFailure(f"driver metric {required} is missing or non-positive") + return metrics + + +def _run_driver( + binary: Path, + arguments: list[str], + metrics_path: Path, + *, + cwd: Path, + timeout_s: int, + memory_limit_mb: int, + maximum_processes: int, + cpu: int | None, +) -> tuple[dict[str, float], dict[str, Any]]: + cwd.mkdir(parents=True, exist_ok=True) + command = [str(binary), *arguments] + environment = os.environ.copy() + environment.update( + { + "OMP_NUM_THREADS": "1", + "OPENBLAS_NUM_THREADS": "1", + "MKL_NUM_THREADS": "1", + "NUMEXPR_NUM_THREADS": "1", + } + ) + started = time.perf_counter() + try: + process = subprocess.run( + command, + cwd=str(cwd), + capture_output=True, + text=True, + timeout=timeout_s, + check=False, + env=environment, + preexec_fn=_preexec_limits( + memory_limit_mb, maximum_processes, cpu, timeout_s + ), + ) + except subprocess.TimeoutExpired as exc: + raise EvaluationFailure( + f"benchmark driver timed out after {timeout_s}s", + timeout=True, + ) from exc + record = { + "command": command, + "returncode": process.returncode, + "runtime_s": time.perf_counter() - started, + "stdout": process.stdout[-MAX_TEXT_CAPTURE:], + "stderr": process.stderr[-MAX_TEXT_CAPTURE:], + } + if process.returncode != 0: + raise EvaluationFailure( + "benchmark driver failed:\n" + process.stderr[-MAX_TEXT_CAPTURE:] + ) + return _parse_driver_metrics(metrics_path), record + + +def _read_query_results(path: Path) -> dict[str, int]: + if not path.is_file(): + raise EvaluationFailure("benchmark driver did not write query results") + results: dict[str, int] = {} + for line in path.read_text(encoding="utf-8").splitlines(): + parts = line.split("\t") + if len(parts) != 2 or not parts[0] or parts[0] in results: + raise EvaluationFailure(f"invalid query result row: {line}") + try: + results[parts[0]] = int(parts[1]) + except ValueError as exc: + raise EvaluationFailure(f"invalid query result value: {line}") from exc + return results + + +def _economic_cost( + encode: dict[str, float], + decode: dict[str, float], + query: dict[str, float], + scenario: dict[str, Any], + pricing: dict[str, Any], +) -> dict[str, float]: + logical_bytes = encode["logical_bytes"] + if ( + abs(decode["logical_bytes"] - logical_bytes) > 0.5 + or abs(query["logical_bytes"] - logical_bytes) > 0.5 + ): + raise EvaluationFailure("driver phases disagree on logical byte count") + encoded_bytes = encode.get("encoded_bytes", 0.0) + if encoded_bytes <= 0.0: + raise EvaluationFailure("encoded byte count is missing") + scale_to_tib = TIB_BYTES / logical_bytes + ns_to_core_hours_per_tib = scale_to_tib / (1e9 * 3600.0) + compression_ratio = encoded_bytes / logical_bytes + encode_hours = encode["median_ns"] * ns_to_core_hours_per_tib + decode_hours = decode["median_ns"] * ns_to_core_hours_per_tib + query_hours = query["median_ns"] * ns_to_core_hours_per_tib + storage_cost = ( + 1024.0 + * compression_ratio + * _positive_number( + pricing.get("storage_dollars_per_gib_month"), + "storage_dollars_per_gib_month", + ) + ) + cpu_price = _positive_number( + pricing.get("cpu_dollars_per_core_hour"), + "cpu_dollars_per_core_hour", + ) + cpu_cost = cpu_price * ( + encode_hours + + _positive_number( + scenario.get("monthly_full_decodes"), "monthly_full_decodes" + ) + * decode_hours + + _positive_number( + scenario.get("monthly_query_batches"), "monthly_query_batches" + ) + * query_hours + ) + total = storage_cost + cpu_cost + if not math.isfinite(total) or total <= 0.0: + raise EvaluationFailure("economic cost is not positive and finite") + return { + "monthly_cost_per_logical_tib": total, + "storage_cost": storage_cost, + "cpu_cost": cpu_cost, + "compression_ratio": compression_ratio, + "encode_core_hours_per_tib": encode_hours, + "decode_core_hours_per_tib": decode_hours, + "query_batch_core_hours_per_tib": query_hours, + } + + +def _invalid_metrics(started: float, *, timeout: bool = False) -> dict[str, float]: + return { + "combined_score": 0.0, + "valid": 0.0, + "correctness": 0.0, + "timeout": 1.0 if timeout else 0.0, + "runtime_s": time.perf_counter() - started, + } + + +def evaluate( + candidate_path: Path, + *, + seed_override: int | None = None, + timeout_override: int | None = None, +) -> tuple[dict[str, float], dict[str, Any]]: + started = time.perf_counter() + artifacts: dict[str, Any] = { + "benchmark_id": "adaptive_compressed_telemetry_execution", + "candidate_path": str(candidate_path), + } + try: + problem, execution, pricing, scenarios = _load_problem() + if not candidate_path.is_file(): + raise EvaluationFailure(f"candidate source not found: {candidate_path}") + if not BASELINE_SOURCE.is_file() or not DRIVER_SOURCE.is_file() or not API_HEADER.is_file(): + raise EvaluationFailure("immutable benchmark source is missing") + source_limit = _positive_int( + execution.get("source_limit_bytes"), "execution.source_limit_bytes" + ) + candidate_source = _validate_candidate_shell(candidate_path, source_limit) + compiler = shutil.which("g++") + if not compiler: + raise EvaluationFailure("g++ is required but was not found") + + candidate_matches_baseline = candidate_source == BASELINE_SOURCE.read_bytes() + artifacts["candidate_source_matches_baseline"] = candidate_matches_baseline + evaluation_seed = ( + int(seed_override) + if seed_override is not None + else _positive_int(problem.get("evaluation_seed"), "evaluation_seed") + ) + artifacts["evaluation_seed"] = evaluation_seed + available_cpus: list[int] = [] + if hasattr(os, "sched_getaffinity"): + available_cpus = sorted(os.sched_getaffinity(0)) + pinned_cpu = available_cpus[0] if available_cpus else None + artifacts["pinned_cpu"] = pinned_cpu + + compile_timeout = _positive_int( + execution.get("compile_timeout_s"), "execution.compile_timeout_s" + ) + run_timeout = ( + int(timeout_override) + if timeout_override is not None + else _positive_int(execution.get("run_timeout_s"), "execution.run_timeout_s") + ) + if run_timeout <= 0: + raise EvaluationFailure("run timeout must be positive") + memory_limit_mb = _positive_int( + execution.get("memory_limit_mb"), "execution.memory_limit_mb" + ) + maximum_processes = _positive_int( + execution.get("max_processes"), "execution.max_processes" + ) + block_rows = _positive_int(execution.get("block_rows"), "execution.block_rows") + warmup_rounds = _positive_int( + execution.get("warmup_rounds"), "execution.warmup_rounds" + ) + measured_rounds = _positive_int( + execution.get("measured_rounds"), "execution.measured_rounds" + ) + max_encoded_ratio = _positive_number( + execution.get("max_encoded_ratio"), "execution.max_encoded_ratio" + ) + + with tempfile.TemporaryDirectory(prefix="adaptive_telemetry_eval_") as tmp_text: + tmp = Path(tmp_text) + candidate_copy = tmp / "candidate.cpp" + candidate_copy.write_bytes(candidate_source) + candidate_binary = tmp / "candidate_driver" + artifacts["candidate_compile"] = _compile( + compiler, candidate_copy, candidate_binary, compile_timeout + ) + baseline_binary = candidate_binary + if not candidate_matches_baseline: + baseline_binary = tmp / "baseline_driver" + artifacts["baseline_compile"] = _compile( + compiler, BASELINE_SOURCE, baseline_binary, compile_timeout + ) + binaries = { + "candidate": candidate_binary, + "baseline": baseline_binary, + } + + scenario_artifacts: dict[str, Any] = {} + ratios: list[float] = [] + total_logical_bytes = 0.0 + candidate_encoded_bytes = 0.0 + candidate_encode_ns = 0.0 + candidate_decode_ns = 0.0 + candidate_query_ns = 0.0 + baseline_total_cost = 0.0 + candidate_total_cost = 0.0 + failed_scenarios: list[str] = [] + failure_summaries: list[str] = [] + any_timeout = False + for scenario_index, scenario in enumerate(scenarios): + scenario_id = str(scenario["id"]) + scenario_result: dict[str, Any] = { + "pattern": scenario["pattern"], + "rows": scenario["rows"], + } + scenario_errors: list[str] = [] + try: + scenario_dir = tmp / scenario_id + scenario_dir.mkdir() + columns = _generate_columns(scenario, evaluation_seed) + queries = _build_queries(columns) + raw_path = scenario_dir / "input.raw" + query_path = scenario_dir / "queries.tsv" + _write_raw(raw_path, columns) + _write_queries(query_path, queries) + raw_payload = raw_path.read_bytes() + raw_sha256 = hashlib.sha256(raw_payload).hexdigest() + expected_queries = { + str(query["name"]): int(query["expected"]) + for query in queries + } + scenario_result["dataset_sha256"] = raw_sha256 + scenario_result["query_expected"] = expected_queries + del columns + except Exception as exc: + message = str(exc) or type(exc).__name__ + scenario_result["scenario_error"] = message + scenario_result["error"] = f"scenario setup: {message}" + scenario_artifacts[scenario_id] = scenario_result + failed_scenarios.append(scenario_id) + failure_summaries.append( + f"{scenario_id}: {scenario_result['error']}" + ) + any_timeout = any_timeout or ( + isinstance(exc, EvaluationFailure) and exc.timeout + ) or "timed out" in message.lower() + continue + + labels = ( + ["candidate"] + if candidate_matches_baseline + else ["baseline", "candidate"] + ) + if not candidate_matches_baseline and scenario_index % 2 == 1: + labels.reverse() + phase_metrics: dict[str, dict[str, dict[str, float]]] = {} + phase_runs: dict[str, dict[str, Any]] = {} + encoded_paths: dict[str, Path] = {} + for label in labels: + label_dir = scenario_dir / label + label_dir.mkdir() + encoded_path = label_dir / "encoded.bin" + encode_metrics_path = label_dir / "encode.metrics" + try: + encode_metrics, encode_run = _run_driver( + binaries[label], + [ + "encode", + str(raw_path), + str(encoded_path), + str(encode_metrics_path), + str(block_rows), + str(warmup_rounds), + str(measured_rounds), + str(max_encoded_ratio), + ], + encode_metrics_path, + cwd=label_dir / "encode_work", + timeout_s=run_timeout, + memory_limit_mb=memory_limit_mb, + maximum_processes=maximum_processes, + cpu=pinned_cpu, + ) + phase_metrics[label] = {"encode": encode_metrics} + phase_runs[label] = {"encode": encode_run} + encoded_paths[label] = encoded_path + except Exception as exc: + message = str(exc) or type(exc).__name__ + scenario_result[f"{label}_error"] = message + scenario_errors.append(f"{label}: {message}") + any_timeout = any_timeout or ( + isinstance(exc, EvaluationFailure) and exc.timeout + ) or "timed out" in message.lower() + + # Decoding happens in a fresh process after the original input has + # been removed, so candidate globals or file references cannot serve + # as a substitute for a self-contained encoded representation. + try: + raw_path.unlink() + except Exception as exc: + message = str(exc) or type(exc).__name__ + scenario_result["scenario_error"] = message + scenario_errors.append(f"raw input removal: {message}") + for label in labels: + if label not in encoded_paths or "scenario_error" in scenario_result: + continue + label_dir = scenario_dir / label + decoded_path = label_dir / "decoded.raw" + decode_metrics_path = label_dir / "decode.metrics" + try: + decode_metrics, decode_run = _run_driver( + binaries[label], + [ + "decode", + str(encoded_paths[label]), + str(decoded_path), + str(decode_metrics_path), + str(warmup_rounds), + str(measured_rounds), + ], + decode_metrics_path, + cwd=label_dir / "decode_work", + timeout_s=run_timeout, + memory_limit_mb=memory_limit_mb, + maximum_processes=maximum_processes, + cpu=pinned_cpu, + ) + phase_metrics[label]["decode"] = decode_metrics + phase_runs[label]["decode"] = decode_run + if ( + not decoded_path.is_file() + or decoded_path.read_bytes() != raw_payload + ): + raise EvaluationFailure( + f"{label} decode does not exactly match input" + ) + decoded_path.unlink() + + query_results_path = label_dir / "query_results.tsv" + query_metrics_path = label_dir / "query.metrics" + query_metrics, query_run = _run_driver( + binaries[label], + [ + "query", + str(encoded_paths[label]), + str(query_path), + str(query_results_path), + str(query_metrics_path), + str(warmup_rounds), + str(measured_rounds), + ], + query_metrics_path, + cwd=label_dir / "query_work", + timeout_s=run_timeout, + memory_limit_mb=memory_limit_mb, + maximum_processes=maximum_processes, + cpu=pinned_cpu, + ) + actual_queries = _read_query_results(query_results_path) + if actual_queries != expected_queries: + mismatches = { + key: { + "expected": expected_queries.get(key), + "actual": actual_queries.get(key), + } + for key in sorted( + set(expected_queries) | set(actual_queries) + ) + if expected_queries.get(key) + != actual_queries.get(key) + } + raise EvaluationFailure( + f"{label} compressed query mismatch: {mismatches}" + ) + phase_metrics[label]["query"] = query_metrics + phase_runs[label]["query"] = query_run + except Exception as exc: + message = str(exc) or type(exc).__name__ + scenario_result[f"{label}_error"] = message + scenario_errors.append(f"{label}: {message}") + any_timeout = any_timeout or ( + isinstance(exc, EvaluationFailure) and exc.timeout + ) or "timed out" in message.lower() + + if candidate_matches_baseline and not scenario_errors: + phase_metrics["baseline"] = phase_metrics["candidate"] + phase_runs["baseline"] = phase_runs["candidate"] + + if not scenario_errors: + try: + costs = { + label: _economic_cost( + phase_metrics[label]["encode"], + phase_metrics[label]["decode"], + phase_metrics[label]["query"], + scenario, + pricing, + ) + for label in ("baseline", "candidate") + } + ratio = ( + costs["baseline"]["monthly_cost_per_logical_tib"] + / costs["candidate"]["monthly_cost_per_logical_tib"] + ) + if not math.isfinite(ratio) or ratio <= 0.0: + raise EvaluationFailure("invalid score ratio") + ratios.append(ratio) + baseline_total_cost += costs["baseline"][ + "monthly_cost_per_logical_tib" + ] + candidate_total_cost += costs["candidate"][ + "monthly_cost_per_logical_tib" + ] + + candidate_encode = phase_metrics["candidate"]["encode"] + candidate_decode = phase_metrics["candidate"]["decode"] + candidate_query = phase_metrics["candidate"]["query"] + total_logical_bytes += candidate_encode["logical_bytes"] + candidate_encoded_bytes += candidate_encode["encoded_bytes"] + candidate_encode_ns += candidate_encode["median_ns"] + candidate_decode_ns += candidate_decode["median_ns"] + candidate_query_ns += candidate_query["median_ns"] + scenario_result["score_ratio"] = ratio + scenario_result["baseline"] = { + "measurements": phase_metrics["baseline"], + "economic_cost": costs["baseline"], + } + scenario_result["candidate"] = { + "measurements": phase_metrics["candidate"], + "economic_cost": costs["candidate"], + } + except Exception as exc: + message = str(exc) or type(exc).__name__ + scenario_result["score_error"] = message + scenario_errors.append(f"score: {message}") + any_timeout = any_timeout or ( + isinstance(exc, EvaluationFailure) and exc.timeout + ) or "timed out" in message.lower() + + scenario_result["driver_runs"] = phase_runs + if scenario_errors: + scenario_result["error"] = "; ".join(scenario_errors) + failed_scenarios.append(scenario_id) + failure_summaries.append( + f"{scenario_id}: {scenario_result['error']}" + ) + else: + scenario_result["status"] = "ok" + scenario_artifacts[scenario_id] = scenario_result + + partial_combined_score = ( + math.exp(sum(math.log(value) for value in ratios) / len(ratios)) + if ratios + else 0.0 + ) + all_scenarios_valid = len(ratios) == len(scenarios) + seconds_per_ns = 1e-9 + gib = 1 << 30 + metrics = { + "combined_score": ( + partial_combined_score if all_scenarios_valid else 0.0 + ), + "partial_combined_score": partial_combined_score, + "valid": 1.0 if all_scenarios_valid else 0.0, + "correctness": 1.0 if all_scenarios_valid else 0.0, + "timeout": 1.0 if any_timeout else 0.0, + "runtime_s": time.perf_counter() - started, + "scenario_count": float(len(scenarios)), + "successful_scenario_count": float(len(ratios)), + "failed_scenario_count": float(len(scenarios) - len(ratios)), + "mean_score_ratio": sum(ratios) / len(ratios) if ratios else 0.0, + "min_score_ratio": min(ratios) if ratios else 0.0, + "candidate_total_monthly_cost_per_tib": candidate_total_cost, + "baseline_total_monthly_cost_per_tib": baseline_total_cost, + "candidate_compression_ratio": ( + candidate_encoded_bytes / total_logical_bytes + if total_logical_bytes > 0.0 + else 0.0 + ), + "candidate_encode_gib_s": ( + total_logical_bytes + / gib + / (candidate_encode_ns * seconds_per_ns) + if candidate_encode_ns > 0.0 + else 0.0 + ), + "candidate_decode_gib_s": ( + total_logical_bytes + / gib + / (candidate_decode_ns * seconds_per_ns) + if candidate_decode_ns > 0.0 + else 0.0 + ), + "candidate_query_batch_gib_s": ( + total_logical_bytes + / gib + / (candidate_query_ns * seconds_per_ns) + if candidate_query_ns > 0.0 + else 0.0 + ), + "candidate_source_matches_baseline": 1.0 + if candidate_matches_baseline + else 0.0, + } + artifacts["scenario_results"] = scenario_artifacts + artifacts["pricing"] = pricing + artifacts["execution"] = execution + artifacts["timing_note"] = ( + "All candidate functions were timed in compiled, CPU-pinned child " + "processes; file loading, result writing, and compilation were excluded." + ) + if failure_summaries: + artifacts["failed_scenarios"] = failed_scenarios + artifacts["failure_summary"] = "\n".join(failure_summaries) + artifacts["error_message"] = ( + f"{len(failed_scenarios)} of {len(scenarios)} scenarios failed; " + "see failure_summary and per-scenario errors" + ) + return metrics, artifacts + except EvaluationFailure as exc: + artifacts["error_message"] = str(exc) + return _invalid_metrics(started, timeout=exc.timeout), artifacts + except Exception as exc: + artifacts["error_message"] = f"evaluator error: {type(exc).__name__}: {exc}" + return _invalid_metrics(started), artifacts + + +def _write_json(path_text: str | None, payload: dict[str, Any]) -> None: + if not path_text: + return + path = Path(path_text).expanduser().resolve() + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + json.dumps(payload, indent=2, sort_keys=True, allow_nan=False) + "\n", + encoding="utf-8", + ) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("candidate", help="candidate C++ source") + parser.add_argument("--seed", type=int, default=None) + parser.add_argument("--timeout-s", type=int, default=None) + parser.add_argument("--metrics-out", default=None) + parser.add_argument("--artifacts-out", default=None) + return parser.parse_args() + + +def main() -> int: + args = parse_args() + metrics, artifacts = evaluate( + Path(args.candidate).expanduser().resolve(), + seed_override=args.seed, + timeout_override=args.timeout_s, + ) + _write_json(args.metrics_out, metrics) + _write_json(args.artifacts_out, artifacts) + print(json.dumps(metrics, sort_keys=True, allow_nan=False)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/verification/requirements.txt b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/verification/requirements.txt new file mode 100644 index 00000000..f11e1fb5 --- /dev/null +++ b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/verification/requirements.txt @@ -0,0 +1 @@ +# Standard-library Python evaluator; requires a C++20-capable g++ compiler. diff --git a/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/verification/test_evaluator.py b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/verification/test_evaluator.py new file mode 100644 index 00000000..c6b7e2a7 --- /dev/null +++ b/benchmarks/ComputerSystems/AdaptiveCompressedTelemetryExecution/verification/test_evaluator.py @@ -0,0 +1,115 @@ +"""Regression tests for the adaptive telemetry evaluator.""" + +from __future__ import annotations + +import sys +import tempfile +import unittest +from pathlib import Path +from unittest import mock + + +VERIFICATION_DIR = Path(__file__).resolve().parent +if str(VERIFICATION_DIR) not in sys.path: + sys.path.insert(0, str(VERIFICATION_DIR)) + +import evaluator # noqa: E402 + + +class AdaptiveTelemetryEvaluatorTests(unittest.TestCase): + def test_baseline_encode_decode_and_query_pipeline(self) -> None: + metrics, artifacts = evaluator.evaluate( + evaluator.TASK_DIR / "scripts" / "init.cpp" + ) + + self.assertEqual(metrics["valid"], 1.0) + self.assertAlmostEqual(metrics["combined_score"], 1.0) + self.assertEqual(metrics["successful_scenario_count"], 3.0) + self.assertEqual(metrics["failed_scenario_count"], 0.0) + self.assertNotIn("failure_summary", artifacts) + for result in artifacts["scenario_results"].values(): + self.assertEqual(result["status"], "ok") + self.assertEqual( + set(result["candidate"]["measurements"]), + {"encode", "decode", "query"}, + ) + self.assertEqual( + set(result["query_expected"]), + { + "error_count", + "error_payload", + "hot_service_count", + "hot_service_duration", + "low_service_duration", + "middle_latency_count", + "middle_window_count", + "window_payload", + }, + ) + + def test_evolve_block_boundary_violation_is_rejected(self) -> None: + source = (evaluator.TASK_DIR / "scripts" / "init.cpp").read_text( + encoding="utf-8" + ) + with tempfile.TemporaryDirectory(prefix="telemetry_shell_test_") as temporary: + candidate = Path(temporary) / "candidate.cpp" + candidate.write_text("// forbidden shell edit\n" + source, encoding="utf-8") + metrics, artifacts = evaluator.evaluate(candidate) + + self.assertEqual(metrics["valid"], 0.0) + self.assertEqual(metrics["combined_score"], 0.0) + self.assertIn("outside the EVOLVE block", artifacts["error_message"]) + self.assertNotIn("candidate_compile", artifacts) + + def test_candidate_scenario_failure_is_isolated(self) -> None: + problem, _execution, _pricing, scenarios = evaluator._load_problem() + evaluation_seed = int(problem["evaluation_seed"]) + failed_scenario = scenarios[1] + columns = evaluator._generate_columns(failed_scenario, evaluation_seed) + sentinel_timestamp = int(columns[0][0]) + + source = (evaluator.TASK_DIR / "scripts" / "init.cpp").read_text( + encoding="utf-8" + ) + function_start = ( + "bool encode_block(const BlockView &input, " + "std::vector &encoded) {\n" + ) + injected_start = function_start + ( + " if (input.count != 0 && input.columns[0][0] == " + f"{sentinel_timestamp}ULL) return false;\n" + ) + self.assertIn(function_start, source) + source = source.replace(function_start, injected_start, 1) + + with tempfile.TemporaryDirectory(prefix="telemetry_isolation_test_") as temporary: + candidate = Path(temporary) / "candidate.cpp" + candidate.write_text(source, encoding="utf-8") + metrics, artifacts = evaluator.evaluate(candidate) + + failed_id = str(failed_scenario["id"]) + self.assertEqual(metrics["valid"], 0.0) + self.assertEqual(metrics["combined_score"], 0.0) + self.assertGreater(metrics["partial_combined_score"], 0.0) + self.assertEqual(metrics["successful_scenario_count"], 2.0) + self.assertEqual(metrics["failed_scenario_count"], 1.0) + self.assertEqual(artifacts["failed_scenarios"], [failed_id]) + failed = artifacts["scenario_results"][failed_id] + self.assertIn("encode_block returned false", failed["candidate_error"]) + self.assertIn("candidate:", failed["error"]) + self.assertEqual( + artifacts["scenario_results"]["incident_distribution_shift"]["status"], + "ok", + ) + + @unittest.skipUnless( + hasattr(evaluator.resource, "RLIMIT_NPROC"), "RLIMIT_NPROC is unavailable" + ) + def test_process_limit_is_applied(self) -> None: + with mock.patch.object(evaluator.resource, "setrlimit") as setrlimit: + evaluator._preexec_limits(1024, 64, None, 30)() + setrlimit.assert_any_call(evaluator.resource.RLIMIT_NPROC, (64, 64)) + + +if __name__ == "__main__": + unittest.main() diff --git a/benchmarks/ComputerSystems/README.md b/benchmarks/ComputerSystems/README.md index 411962ac..12043306 100644 --- a/benchmarks/ComputerSystems/README.md +++ b/benchmarks/ComputerSystems/README.md @@ -3,5 +3,11 @@ Includes computer-systems engineering optimization tasks: - `MallocLab`: dynamic memory allocation. - `DuckDBWorkloadOptimization`: analytical SQL workload tuning (index/materialized-view selection + query rewrite). +- `AdaptiveCompressedTelemetryExecution`: CPU-measured co-design of telemetry compression and compressed query execution. + +`AdaptiveCompressedTelemetryExecution` is an execution-backed C++ task: candidate +codec and query code is compiled, correctness-checked, and timed on a pinned CPU. +It reports raw throughput and compression metrics as well as a storage/CPU economic +objective. Note for contributors: ensure the evolved baseline source file contains `EVOLVE-BLOCK-START` / `EVOLVE-BLOCK-END` markers (use `// ...` in C/C++). diff --git a/benchmarks/ComputerSystems/README_zh-CN.md b/benchmarks/ComputerSystems/README_zh-CN.md index 0d12dfc1..1c7009a4 100644 --- a/benchmarks/ComputerSystems/README_zh-CN.md +++ b/benchmarks/ComputerSystems/README_zh-CN.md @@ -3,5 +3,10 @@ 包含以下计算机系统工程优化任务： - `MallocLab`：动态内存分配。 - `DuckDBWorkloadOptimization`：分析型 SQL 负载调优（索引/物化视图选择 + 查询改写）。 +- `AdaptiveCompressedTelemetryExecution`：实测 CPU 遥测数据压缩与压缩态查询协同设计。 + +`AdaptiveCompressedTelemetryExecution` 是一个执行驱动的 C++ 任务：候选编解码 +与查询代码会被编译、逐值校验，并在绑定的单个 CPU 上计时；评测同时报告原始 +吞吐量、压缩率以及存储与 CPU 综合经济目标。 贡献提示：请确保被 evolve 的 baseline 源码文件包含 `EVOLVE-BLOCK-START` / `EVOLVE-BLOCK-END` 标记（C/C++ 中使用 `// ...`）。