diff --git a/.github/workflows/performance.yml b/.github/workflows/performance.yml new file mode 100644 index 0000000000..baa047a62b --- /dev/null +++ b/.github/workflows/performance.yml @@ -0,0 +1,151 @@ +name: Performance regression + +on: + pull_request: + branches: [main, 'release/**'] + types: [opened, synchronize, reopened] + workflow_dispatch: + inputs: + baseline_ref: + description: Baseline commit or ref (use the candidate SHA for an A/A calibration) + required: true + default: main + candidate_ref: + description: Candidate commit or ref (empty uses the selected workflow ref) + required: false + +permissions: + contents: read + +concurrency: + group: performance-${{ github.event.pull_request.number || github.run_id }} + cancel-in-progress: true + +jobs: + tests: + name: Performance gate tests + runs-on: ubuntu-22.04 + steps: + - uses: actions/checkout@v4 + with: + persist-credentials: false + - run: python3 -m unittest discover -s scripts/ci/tests -p 'test_*perf*.py' -v + + performance: + name: Performance Gate + needs: tests + if: ${{ always() }} + # Fork code always runs on a disposable hosted VM. A repository-owned PR + # can use the optional dedicated Linux runner label after A/A calibration. + runs-on: ${{ github.event_name == 'pull_request' && github.event.pull_request.head.repo.fork && 'ubuntu-22.04' || vars.PERF_RUNNER || 'ubuntu-22.04' }} + timeout-minutes: 75 + env: + PERF_ROOT: ${{ github.workspace }}/performance-${{ github.run_id }}-${{ github.run_attempt }} + steps: + - name: Initialize run directory + run: mkdir -p "$PERF_ROOT/results" + + - name: Require harness tests + env: + TEST_RESULT: ${{ needs.tests.result }} + run: test "$TEST_RESULT" = success + + - name: Checkout candidate + uses: actions/checkout@v4 + with: + ref: ${{ inputs.candidate_ref || github.sha }} + path: candidate + fetch-depth: 2 + submodules: true + persist-credentials: false + + - name: Checkout baseline + uses: actions/checkout@v4 + with: + ref: ${{ inputs.baseline_ref || github.event.pull_request.base.sha }} + path: baseline + submodules: true + persist-credentials: false + + - name: Freeze the benchmark and policy + env: + EVENT_NAME: ${{ github.event_name }} + run: | + set -euo pipefail + baseline_sha=$(git -C baseline rev-parse HEAD) + candidate_sha=$(git -C candidate rev-parse HEAD) + if [ "$EVENT_NAME" = pull_request ]; then + test "$(git -C candidate rev-parse HEAD^1)" = "$baseline_sha" + fi + harness_source=baseline + if [ ! -f baseline/scripts/ci/performance/perf_gate.py ]; then + harness_source=candidate + echo '::warning::Bootstrapping the performance harness from the candidate; review the new policy before merging.' + fi + mkdir -p "$PERF_ROOT/harness" + cp -R "$harness_source/scripts/ci/performance/." "$PERF_ROOT/harness/" + cp baseline/mooncake-store/benchmarks/offset_allocator_concurrency_bench.cpp \ + candidate/mooncake-store/benchmarks/offset_allocator_concurrency_bench.cpp + python3 "$PERF_ROOT/harness/freeze_build.py" \ + --baseline baseline --candidate candidate \ + --manifest "$PERF_ROOT/results/build-config.json" + echo "BASELINE_SHA=$baseline_sha" >> "$GITHUB_ENV" + echo "CANDIDATE_SHA=$candidate_sha" >> "$GITHUB_ENV" + echo "PERF_HARNESS_SHA=$(git -C "$harness_source" rev-parse HEAD)" >> "$GITHUB_ENV" + + - name: Install hosted runner dependencies + if: runner.environment == 'github-hosted' + timeout-minutes: 10 + working-directory: baseline + run: sudo bash dependencies.sh -y + + - name: Build both revisions with the same Release configuration + timeout-minutes: 35 + run: | + set -euo pipefail + for revision in baseline candidate; do + cmake -S "$revision" -B "$revision/build-perf" -G Ninja \ + -DCMAKE_BUILD_TYPE=Release -DCMAKE_CXX_COMPILER=c++ \ + -DBUILD_UNIT_TESTS=OFF -DBUILD_EXAMPLES=OFF -DBUILD_BENCHMARK=ON \ + -DWITH_STORE_RUST=OFF -DUSE_CUDA=OFF -DUSE_ETCD=OFF \ + -DSTORE_USE_ETCD=OFF -DENABLE_DEBUG_SYMBOLS=OFF \ + 2>&1 | tee "$PERF_ROOT/results/$revision-configure.log" + cmake --build "$revision/build-perf" --parallel 2 \ + --target offset_allocator_concurrency_bench \ + 2>&1 | tee "$PERF_ROOT/results/$revision-build.log" + done + + - name: Collect alternating baseline and candidate samples + timeout-minutes: 16 + run: | + python3 "$PERF_ROOT/harness/perf_gate.py" collect \ + --baseline baseline/build-perf/mooncake-store/benchmarks/offset_allocator_concurrency_bench \ + --candidate candidate/build-perf/mooncake-store/benchmarks/offset_allocator_concurrency_bench \ + --baseline-sha "$BASELINE_SHA" --candidate-sha "$CANDIDATE_SHA" \ + --collection-timeout-seconds 900 \ + --output "$PERF_ROOT/results" + + - name: Publish comparison and enforce thresholds + if: ${{ always() }} + run: | + mkdir -p "$PERF_ROOT/results" + if [ ! -f "$PERF_ROOT/harness/perf_gate.py" ]; then + echo '## Performance Gate: ERROR — setup or harness tests failed' > "$PERF_ROOT/results/summary.md" + cat "$PERF_ROOT/results/summary.md" >> "$GITHUB_STEP_SUMMARY" + echo '::error::Performance gate setup failed; no valid measurement is available.' + exit 2 + fi + cp "$PERF_ROOT/harness/policy.json" "$PERF_ROOT/results/policy.json" + python3 "$PERF_ROOT/harness/perf_gate.py" compare \ + --baseline "$PERF_ROOT/results/baseline.json" --candidate "$PERF_ROOT/results/candidate.json" \ + --baseline-sha "${BASELINE_SHA:-}" --candidate-sha "${CANDIDATE_SHA:-}" \ + --output "$PERF_ROOT/results" --summary "$GITHUB_STEP_SUMMARY" + + - name: Preserve raw samples and diagnostics + if: ${{ always() }} + uses: actions/upload-artifact@v4 + with: + name: performance-results + path: ${{ env.PERF_ROOT }}/results/ + if-no-files-found: error + retention-days: 14 diff --git a/.github/workflows/rdma-performance.yml b/.github/workflows/rdma-performance.yml new file mode 100644 index 0000000000..c6c51d5a9f --- /dev/null +++ b/.github/workflows/rdma-performance.yml @@ -0,0 +1,89 @@ +name: RDMA performance regression + +on: + pull_request: + branches: [main, 'release/**'] + types: [opened, synchronize, reopened] + workflow_dispatch: + inputs: + baseline_sha: + description: Full baseline commit SHA (same as candidate for A/A calibration) + required: true + candidate_sha: + description: Full candidate commit SHA + required: true + workflow_call: + inputs: + baseline_sha: + required: true + type: string + candidate_sha: + required: true + type: string + secrets: + TONE_USER_NAME: + required: true + TONE_USER_TOKEN: + required: true + +permissions: + contents: read + +# One fixed hardware pair; retain up to 100 waiting runs without interrupting it. +concurrency: + group: rdma-performance-pair + cancel-in-progress: false + queue: max + +jobs: + rdma: + name: RDMA Performance Gate + if: ${{ github.event_name != 'pull_request' || vars.RDMA_PERF_ENABLED == 'true' }} + runs-on: ubuntu-22.04 + timeout-minutes: 60 + steps: + - name: Checkout the reviewed harness + uses: actions/checkout@v4 + with: + # Never execute candidate code in the credential-bearing controller. + # Manual bootstrap uses the selected workflow revision, reviewed by its dispatcher. + ref: ${{ github.event.pull_request.base.sha || github.workflow_sha }} + persist-credentials: false + + - name: Submit, collect, and enforce RDMA thresholds + env: + BASELINE_SHA: ${{ inputs.baseline_sha || github.event.pull_request.base.sha }} + CANDIDATE_SHA: ${{ inputs.candidate_sha || github.sha }} + TONE_USER_NAME: ${{ secrets.TONE_USER_NAME }} + TONE_USER_TOKEN: ${{ secrets.TONE_USER_TOKEN }} + TONE_WORKSPACE: ${{ vars.TONE_WORKSPACE }} + TONE_PROJECT: ${{ vars.TONE_PROJECT }} + TONE_RDMA_TEMPLATE: ${{ vars.TONE_RDMA_TEMPLATE }} + TONE_RESULT_ORIGIN: ${{ vars.TONE_RESULT_ORIGIN }} + RUN_ID: ${{ github.run_id }}-${{ github.run_attempt }} + run: | + python3 scripts/ci/performance/tone_rdma.py \ + --baseline-sha "$BASELINE_SHA" --candidate-sha "$CANDIDATE_SHA" \ + --harness-sha "$(git rev-parse HEAD)" --run-id "$RUN_ID" \ + --output rdma-performance-results + + - name: Publish setup errors + if: ${{ always() }} + run: | + if [ ! -f rdma-performance-results/summary.md ]; then + mkdir -p rdma-performance-results + echo '## RDMA Performance Gate: ERROR — no complete report was returned' \ + > rdma-performance-results/summary.md + cat rdma-performance-results/summary.md >> "$GITHUB_STEP_SUMMARY" + echo '::error::RDMA setup failed or the run was interrupted.' + exit 2 + fi + + - name: Preserve results even on failure + if: ${{ always() }} + uses: actions/upload-artifact@v4 + with: + name: rdma-performance-results + path: rdma-performance-results/ + if-no-files-found: error + retention-days: 14 diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 871667cddb..744194ecd6 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -130,6 +130,10 @@ git commit -m "wip: skipping hooks" --no-verify But please avoid using `--no-verify` for routine commits to keep code quality high. #### CI Integration +The [performance regression gate](scripts/ci/performance/README.md) compares +baseline and candidate CPU benchmarks, publishes a result table in the Actions +summary, and fails on regressions beyond the configured thresholds. + GitHub pull-request and push checks validate only added or modified C/C++ line ranges relative to the selected base revision. This avoids failing a focused change solely because an otherwise untouched part of the same file has older diff --git a/mooncake-transfer-engine/benchmark/tent_backend.cpp b/mooncake-transfer-engine/benchmark/tent_backend.cpp index 18b7c67b23..90cf2dad9c 100644 --- a/mooncake-transfer-engine/benchmark/tent_backend.cpp +++ b/mooncake-transfer-engine/benchmark/tent_backend.cpp @@ -163,6 +163,13 @@ static int resolveSegTypeParams(const std::string& seg_type, if (seg_type == "DRAM" || seg_type == "dram") { device_prefix = "cpu"; num_buffers = numa_num_configured_nodes(); + if (XferBenchConfig::dram_numa_node != -1) { + start_idx = XferBenchConfig::dram_numa_node; + LOG_ASSERT(start_idx >= 0 && start_idx <= numa_max_node() && + numa_bitmask_isbitset(numa_all_nodes_ptr, start_idx)) + << "dram_numa_node is not an available NUMA node"; + num_buffers = 1; + } #if defined(USE_CUDA) || defined(USE_SUNRISE) } else if (seg_type == "VRAM" || seg_type == "vram") { device_prefix = "cuda"; diff --git a/mooncake-transfer-engine/benchmark/utils.cpp b/mooncake-transfer-engine/benchmark/utils.cpp index 235d31efec..772fa5d7dd 100644 --- a/mooncake-transfer-engine/benchmark/utils.cpp +++ b/mooncake-transfer-engine/benchmark/utils.cpp @@ -99,6 +99,8 @@ DEFINE_bool(deadline_bw_arbitration, false, DEFINE_int32(local_gpu_id, 0, "Local device ordinal for VRAM, -1 for all visible devices. " "Selects an MLU device with --backend=tent in USE_MLU builds."); +DEFINE_int32(dram_numa_node, -1, + "TENT DRAM NUMA node, -1 allocates on all configured nodes"); DEFINE_int32(target_gpu_id, 0, "Offset for round-robin assignment across target buffers; " "-1 uses 0."); @@ -182,6 +184,7 @@ std::string XferBenchConfig::tent_transport_hint; std::string XferBenchConfig::tent_intent_type; int XferBenchConfig::local_gpu_id = 0; +int XferBenchConfig::dram_numa_node = -1; int XferBenchConfig::target_gpu_id = 0; namespace { @@ -274,6 +277,7 @@ void XferBenchConfig::loadFromFlags() { tent_intent_type = FLAGS_tent_intent_type; local_gpu_id = FLAGS_local_gpu_id; + dram_numa_node = FLAGS_dram_numa_node; target_gpu_id = FLAGS_target_gpu_id; } diff --git a/mooncake-transfer-engine/benchmark/utils.h b/mooncake-transfer-engine/benchmark/utils.h index f676becbc9..d472f6f3d0 100644 --- a/mooncake-transfer-engine/benchmark/utils.h +++ b/mooncake-transfer-engine/benchmark/utils.h @@ -104,6 +104,7 @@ struct XferBenchConfig { static std::string tent_intent_type; static int local_gpu_id; + static int dram_numa_node; static int target_gpu_id; }; diff --git a/scripts/ci/performance/README.md b/scripts/ci/performance/README.md new file mode 100644 index 0000000000..162d660445 --- /dev/null +++ b/scripts/ci/performance/README.md @@ -0,0 +1,127 @@ +# CI performance regression gate + +The `Performance regression` workflow builds the PR base and test merge commit +in Release mode on the same Linux runner. It runs the existing +`offset_allocator_concurrency_bench` with the **base revision's benchmark source +and tracked CMake recipes** in both builds. Target definitions, parent compile +flags and included `.cmake` files are frozen; candidate-only recipes are removed +from the disposable checkout. The baseline's policy and Python harness are also +used. The first PR introducing +this directory necessarily bootstraps the harness from the candidate and emits +a warning. Benchmark, build-recipe and policy changes take effect for subsequent +PRs after merge. Candidate implementation sources still change. A candidate +requiring an incompatible build layout fails explicitly; this gate compares +source changes under the baseline build configuration, not build-system changes. +This workload control is not a sandbox for arbitrary candidate C++ code. + +The initial suite covers successful allocate/free, capacity rejection, and +fragmentation rejection with one and two threads. It measures allocator CPU +performance; it does not validate master RPC, Store end-to-end, RDMA, or GPUs. +Those need their own workload adapters and, for RDMA, a fixed pair of machines. + +The opt-in [T-One RDMA gate](tone/mooncake-rdma-perf/README.md) adds that fixed-pair +adapter for TENT DRAM read/write. It has a separate workflow and policy; the CPU +suite does not require RDMA credentials or hardware. Deploy and calibrate the +T-One template before enabling `RDMA_PERF_ENABLED`. + +## Results and failure behavior + +Open **Performance Gate → Details → Summary** to see each case's baseline and +candidate medians, regression percentage, threshold, and measurement spread. +The `performance-results` artifact contains raw CSV/stderr for every invocation, +the exact policy, frozen CMake file hashes (`build-config.json`), build logs, +binary hashes, commit SHAs, CPU affinity, compiler, +environment, normalized samples, `comparison.json`, and `summary.md`. + +`policy.json` specifies five measured runs per revision. Each invocation performs +100,000 warmup operations per thread, then separate throughput and latency phases +of 2,000,000 operations per thread. Measurement order alternates A/B, B/A. +All subprocesses inherit the same two-CPU affinity. No benchmark invocations run +concurrently. The default limits are: + +| Metric | Better | Fail when regression exceeds | Maximum relative MAD | +|---|---|---:|---:| +| `qps` | Higher | 5% | 5% | +| `p50_ns` | Lower | 5% | 5% | +| `p99_ns` | Lower | 10% | 10% | + +A case can override individual limits, for example +`"max_regression_percent": {"qps": 8}` inside that case's configuration. +Other metrics retain their default limits. Threshold changes require review. + +For baseline median B and candidate median C, regression is `100 * (B-C)/B` +for throughput and `100 * (C-B)/B` for latency. Equal-to-threshold changes pass. +Every metric must pass; improvements do not cancel other regressions. +Relative MAD is `100 * median(abs(sample - median)) / median` for each revision. +A noisy metric is an ERROR, not a confirmed code regression or a passing check. + +The comparator writes its reports **before** exiting: 0 = passed, 1 = regression, +2 = invalid/incomplete/unstable measurement. Missing cases, units or SHAs that do +not match, non-finite or non-positive values, benchmark errors, and incomplete +sample counts fail closed. Expected allocation rejection is part of the capacity +and fragmentation workloads; any *unexpected* result fails correctness checks. +Setup/build failures also produce an ERROR summary. Reports are uploaded even on +failure. Collection stops at the first process, timeout or parsing failure after +saving partial samples. A 15-minute total collection budget also bounds sequences +of slow successful samples. Dependency installation and building have separate +10/35-minute step limits within the 75-minute job, leaving time for reports and +uploads. Owned benchmark process groups are stopped on failure or cancellation. +Re-running creates a new artifact; there is no retry-until-green logic. + +## Runner calibration and branch protection + +The default runner is `ubuntu-22.04`. Both revisions share the same VM, but hosted +machines can still have host contention and frequency variation. Start with A/A +calibration: manually dispatch the workflow with **the same full commit SHA** for +both `baseline_ref` and `candidate_ref`. Run across multiple times of day before +choosing thresholds or making the check required. MAD detects within-run spread; +it cannot establish that the host is free of systematic bias. + +For repository-owned PRs and manual runs, `PERF_RUNNER` can name an exclusive +Linux runner label. Provision the normal Mooncake dependencies, CMake, Ninja, +Python 3, and a C++20 compiler in advance; dependency installation is automatic +only on hosted runners. Use one runner process per machine and reserve its CPUs +for this suite. Fork PRs always use disposable hosted VMs. A dedicated runner +must isolate and discard PR build state; do not give it production credentials. +Calibrate hosted and dedicated environments separately. + +After validation, add **Performance Gate** to required status checks in the +repository ruleset / branch protection. The workflow has no path filters, and +the named gate runs even if the harness-test job fails. This PR does not modify +repository protection settings. Only make it required after calibration. + +For release-to-release comparisons or an A/A run, use `workflow_dispatch` with +explicit baseline/candidate SHAs. The baseline must contain the selected +benchmark and support the same flags. Old or incompatible baselines produce an +ERROR; they are never silently replaced. Each PR's base is resolved anew, so a +separate fixed release anchor is needed to detect cumulative small regressions. + +## Local reproduction + +Use disposable checkouts. Freeze the baseline benchmark source and build recipes, +then build both revisions with the workflow's CMake flags: + +```bash +cp /path/to/base/mooncake-store/benchmarks/offset_allocator_concurrency_bench.cpp \ + /path/to/candidate/mooncake-store/benchmarks/offset_allocator_concurrency_bench.cpp +python3 scripts/ci/performance/freeze_build.py \ + --baseline /path/to/base --candidate /path/to/candidate --manifest build-config.json +``` + +On Linux, collect results into a fresh directory: + +```bash +python3 scripts/ci/performance/perf_gate.py collect \ + --baseline /path/to/base/build-perf/mooncake-store/benchmarks/offset_allocator_concurrency_bench \ + --candidate /path/to/candidate/build-perf/mooncake-store/benchmarks/offset_allocator_concurrency_bench \ + --baseline-sha FULL_BASE_SHA --candidate-sha FULL_CANDIDATE_SHA --output results +python3 scripts/ci/performance/perf_gate.py compare \ + --baseline results/baseline.json --candidate results/candidate.json \ + --baseline-sha FULL_BASE_SHA --candidate-sha FULL_CANDIDATE_SHA --output results +python3 -m unittest discover -s scripts/ci/tests -p 'test_*perf*.py' -v +``` + +Use the same policy and benchmark source for both revisions. The comparison +command also runs on non-Linux machines. Unit tests use controlled fixture +executables to verify reporting and exit status; they do not measure Mooncake +performance or replace a native A/A calibration. diff --git a/scripts/ci/performance/freeze_build.py b/scripts/ci/performance/freeze_build.py new file mode 100644 index 0000000000..42d9db3501 --- /dev/null +++ b/scripts/ci/performance/freeze_build.py @@ -0,0 +1,74 @@ +#!/usr/bin/env python3 +"""Use the baseline's tracked CMake recipes in disposable benchmark checkouts.""" + +import argparse +import hashlib +import json +from pathlib import Path +import subprocess + + +def build_files(root): + tracked = ( + subprocess.check_output(["git", "-C", str(root), "ls-files", "-z"], timeout=30) + .decode() + .split("\0") + ) + return { + name + for name in tracked + if Path(name).name == "CMakeLists.txt" or name.endswith(".cmake") + } + + +def checked_path(root, name): + path = root / name + if not path.resolve().is_relative_to(root) or any( + item.is_symlink() for item in (path, *path.parents) if item != root + ): + raise ValueError( + f"build configuration uses a symlink or escapes its checkout: {name}" + ) + return path + + +def freeze(baseline, candidate): + baseline, candidate = Path(baseline).resolve(), Path(candidate).resolve() + if baseline == candidate: + raise ValueError("baseline and candidate must be separate checkouts") + names = build_files(baseline) + if "CMakeLists.txt" not in names: + raise ValueError("baseline has no tracked CMake project") + sources = { + name: checked_path(baseline, name).read_bytes() for name in sorted(names) + } + destinations = { + name: checked_path(candidate, name) for name in names | build_files(candidate) + } + # Remove newly added recipes too: a frozen glob/include could otherwise load them. + for name, path in destinations.items(): + if name not in sources: + path.unlink(missing_ok=True) + for name, content in sources.items(): + path = destinations[name] + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(content) + files = { + name: hashlib.sha256(content).hexdigest() for name, content in sources.items() + } + return { + "files": files, + "sha256": hashlib.sha256( + json.dumps(files, sort_keys=True).encode() + ).hexdigest(), + } + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--baseline", required=True) + parser.add_argument("--candidate", required=True) + parser.add_argument("--manifest", required=True) + args = parser.parse_args() + manifest = freeze(args.baseline, args.candidate) + Path(args.manifest).write_text(json.dumps(manifest, indent=2) + "\n") diff --git a/scripts/ci/performance/perf_gate.py b/scripts/ci/performance/perf_gate.py new file mode 100644 index 0000000000..96f70e9f01 --- /dev/null +++ b/scripts/ci/performance/perf_gate.py @@ -0,0 +1,480 @@ +#!/usr/bin/env python3 +"""Collect paired CPU benchmarks and publish a fail-closed performance gate.""" + +import argparse +import csv +import hashlib +import io +import json +import math +import os +from pathlib import Path +import platform +import re +import signal +import statistics +import subprocess +import sys +import time + + +def require(condition, message): + if not condition: + raise ValueError(message) + + +def number(value, minimum=0): + try: + return type(value) in (int, float) and math.isfinite(value) and value > minimum + except OverflowError: + return False + + +def load_policy(path): + policy = json.loads(Path(path).read_text()) + require(isinstance(policy, dict), "policy must be an object") + require(policy["schema_version"] == 1, "unsupported policy version") + for field in ("arguments", "cases", "metrics"): + require( + isinstance(policy[field], dict) and policy[field], + f"{field} must be a nonempty object", + ) + require( + type(policy["repetitions"]) is int and policy["repetitions"] >= 5, + "at least five repetitions are required", + ) + require( + type(policy["cpu_count"]) is int and policy["cpu_count"] > 0, + "invalid CPU count", + ) + require(number(policy["timeout_seconds"]), "invalid timeout") + rdma = policy.get("suite", "cpu") == "rdma" + require(policy.get("suite", "cpu") in ("cpu", "rdma"), "unknown suite") + if not rdma: + require( + type(policy["arguments"]["iterations"]) is int + and policy["arguments"]["iterations"] > 0, + "invalid operation count", + ) + else: + for key in ("duration", "total_buffer_size"): + require( + type(policy["arguments"][key]) is int and policy["arguments"][key] > 0, + f"invalid {key}", + ) + require( + policy["arguments"]["duration"] + 1 < policy["timeout_seconds"] <= 60, + "RDMA timeout must include warmup/setup and fit the 60-second sample watchdog", + ) + for name, case in policy["cases"].items(): + require(re.fullmatch(r"[a-z0-9_-]+", name), "invalid case name") + require(isinstance(case, dict), f"{name}: case must be an object") + require( + isinstance(case.get("max_regression_percent", {}), dict), + f"{name}: threshold overrides must be an object", + ) + require( + type(case["threads"]) is int and 0 < case["threads"] <= policy["cpu_count"], + "invalid thread count", + ) + if rdma: + require( + isinstance(case.get("metrics"), list) + and case["metrics"] + and all(isinstance(key, str) for key in case["metrics"]) + and len(set(case["metrics"])) == len(case["metrics"]), + f"{name}: metrics must be a nonempty list of distinct names", + ) + require(case["op_type"] in ("read", "write"), "invalid RDMA operation") + for key in ("block_size", "batch_size"): + require(type(case[key]) is int and case[key] > 0, f"invalid {key}") + require( + case["block_size"] * case["batch_size"] * case["threads"] + <= policy["arguments"]["total_buffer_size"], + "RDMA buffer too small", + ) + else: + require("metrics" not in case, "CPU collector uses all policy metrics") + require( + case["layer"] in ("offset_allocator", "put_allocation"), "invalid layer" + ) + require( + case["scenario"] in ("success", "capacity", "fragmentation", "mixed"), + "invalid scenario", + ) + require( + case.get("metrics", list(policy["metrics"])) + and set(case.get("metrics", policy["metrics"])) <= set(policy["metrics"]), + "invalid case metrics", + ) + for key, limit in case.get("max_regression_percent", {}).items(): + require( + key in policy["metrics"] and number(limit, -1) and limit >= 0, + "invalid case threshold override", + ) + for name, metric in policy["metrics"].items(): + require(isinstance(metric, dict), f"{name}: metric must be an object") + allowed = ("bandwidth_gbs", "p99_us") if rdma else ("qps", "p50_ns", "p99_ns") + require(name in allowed, "unsupported benchmark metric") + require( + metric["direction"] in ("higher", "lower") + and isinstance(metric["unit"], str) + and metric["unit"], + "invalid metric direction or unit", + ) + require( + number(metric["max_regression_percent"], -1) + and metric["max_regression_percent"] >= 0 + and number(metric["max_mad_percent"]), + "invalid threshold", + ) + return policy + + +def policy_digest(policy): + return hashlib.sha256(json.dumps(policy, sort_keys=True).encode()).hexdigest() + + +def write_json(path, value): + Path(path).write_text(json.dumps(value, indent=2, allow_nan=False) + "\n") + + +def annotation(message): + escaped = message.replace("%", "%25").replace("\r", "%0D").replace("\n", "%0A") + print(f"::error title=Performance Gate::{escaped}") + + +def stop_process(process): + """Stop only the process group started for this invocation, including descendants.""" + if process is None: + return + try: + os.killpg(process.pid, signal.SIGTERM) + except ProcessLookupError: + pass # The group already exited; its direct child still needs to be reaped. + try: + process.wait(timeout=10) + except subprocess.TimeoutExpired: + try: + os.killpg(process.pid, signal.SIGKILL) + except ProcessLookupError: + pass # It exited between the wait and the signal. + process.wait(timeout=5) + try: + os.killpg(process.pid, signal.SIGKILL) + except ProcessLookupError: + pass # No descendants remain after the parent was reaped. + + +def environment(cpu_count): + cpus = sorted(os.sched_getaffinity(0))[:cpu_count] + require(len(cpus) == cpu_count, "not enough CPUs for the configured suite") + os.sched_setaffinity(0, cpus) + cpu_info = Path("/proc/cpuinfo").read_text() + model = next( + ( + line.split(":", 1)[1].strip() + for line in cpu_info.splitlines() + if line.startswith(("model name", "Hardware")) + ), + platform.machine(), + ) + return { + "hostname": platform.node(), + "platform": platform.platform(), + "cpu_model": model, + "cpu_affinity": cpus, + "compiler": subprocess.check_output( + ["c++", "--version"], text=True + ).splitlines()[0], + "harness_sha": os.environ.get("PERF_HARNESS_SHA", "local"), + } + + +def parse_sample(output, case, policy): + rows = list(csv.DictReader(io.StringIO(output))) + require(len(rows) == 1, "expected exactly one benchmark result") + row = rows[0] + require( + row["layer"] == case["layer"] and row["scenario"] == case["result_scenario"], + "benchmark returned a different case", + ) + require( + int(row["threads"]) == case["threads"] + and int(row["iterations_per_thread"]) == policy["arguments"]["iterations"], + "benchmark workload differs from the policy", + ) + require( + int(row["operations"]) == case["threads"] * policy["arguments"]["iterations"], + "benchmark did not complete all operations", + ) + require( + int(row["throughput_unexpected_results"]) == 0 + and int(row["latency_unexpected_results"]) == 0, + "benchmark correctness failure", + ) + require(number(float(row["throughput_seconds"])), "invalid measurement duration") + sample = {name: float(row[name]) for name in policy["metrics"]} + require( + all(number(value) for value in sample.values()), "invalid benchmark measurement" + ) + return sample + + +def collect(args): + policy = load_policy(args.policy) + out = Path(args.output) + out.mkdir(parents=True, exist_ok=True) + # Build metadata is prepared before collection; prior measurements are not reusable. + require( + not any(path.name != "build-config.json" for path in out.glob("*.json")), + "output directory already contains results", + ) + env = environment(policy["cpu_count"]) + budget = getattr(args, "collection_timeout_seconds", 900) + require(number(budget), "invalid collection time budget") + deadline = time.monotonic() + budget + results = {} + for side in ("baseline", "candidate"): + binary = Path(getattr(args, side)).resolve() + results[side] = { + "schema_version": 1, + "revision": getattr(args, f"{side}_sha"), + "environment": env, + "policy_sha256": policy_digest(policy), + "binary_sha256": hashlib.sha256(binary.read_bytes()).hexdigest(), + "cases": {}, + "errors": [], + } + for name, case in policy["cases"].items(): + flags = { + **policy["arguments"], + **{k: case[k] for k in ("layer", "scenario", "threads")}, + } + for repeat in range(policy["repetitions"]): + # Reverse each pair to reduce bias from temperature / temporal drift. + order = ( + ("baseline", "candidate") + if repeat % 2 == 0 + else ("candidate", "baseline") + ) + for side in order: + log = out / f"{name}-{repeat}-{side}" + command = [str(Path(getattr(args, side)).resolve())] + command += [f"--{key}={value}" for key, value in flags.items()] + print(f"{name} round {repeat + 1}: {side}", flush=True) + try: + remaining = deadline - time.monotonic() + require(remaining > 0, "collection time budget exhausted") + with log.with_suffix(".csv").open("w") as stdout, log.with_suffix( + ".log" + ).open("w") as stderr: + process = subprocess.Popen( + command, + stdout=stdout, + stderr=stderr, + start_new_session=True, + ) + try: + code = process.wait( + timeout=min(policy["timeout_seconds"], remaining) + ) + if code: + raise subprocess.CalledProcessError(code, command) + finally: + stop_process(process) + sample = parse_sample( + log.with_suffix(".csv").read_text(), case, policy + ) + values = results[side]["cases"].setdefault( + name, + { + key: {"unit": metric["unit"], "samples": []} + for key, metric in policy["metrics"].items() + }, + ) + for key, value in sample.items(): + values[key]["samples"].append(value) + except ( + OSError, + ValueError, + KeyError, + TypeError, + RuntimeError, + subprocess.SubprocessError, + ) as exc: + results[side]["errors"].append(f"{name} round {repeat + 1}: {exc}") + # A failed invocation invalidates the run. Save diagnostics below, + # then leave time for comparison/upload instead of retrying it. + return 2 + finally: + for result_side, result in results.items(): + write_json(out / f"{result_side}.json", result) + return 0 + + +def compare(policy, baseline, candidate, baseline_sha, candidate_sha): + for result, sha in ((baseline, baseline_sha), (candidate, candidate_sha)): + require(re.fullmatch(r"[0-9a-f]{40}", sha or ""), "expected a full commit SHA") + require( + result["schema_version"] == 1 and result["revision"] == sha, + "result version or commit SHA mismatch", + ) + require( + result["policy_sha256"] == policy_digest(policy), "result policy mismatch" + ) + require( + not result["errors"], "benchmark errors: " + "; ".join(result["errors"]) + ) + require( + set(result["cases"]) == set(policy["cases"]), + "missing or unexpected benchmark cases", + ) + require( + baseline["environment"] and baseline["environment"] == candidate["environment"], + "baseline and candidate environments differ", + ) + rows = [] + for name in policy["cases"]: + require( + set(baseline["cases"][name]) + == set(candidate["cases"][name]) + == set(policy["cases"][name].get("metrics", policy["metrics"])), + f"{name}: missing or unexpected metrics", + ) + for key in policy["cases"][name].get("metrics", policy["metrics"]): + metric = policy["metrics"][key] + medians, noise = [], [] + for result in (baseline, candidate): + data = result["cases"][name][key] + samples = data["samples"] + require(data["unit"] == metric["unit"], f"{name}/{key}: unit mismatch") + require( + len(samples) == policy["repetitions"] + and all(number(x) for x in samples), + f"{name}/{key}: incomplete or invalid samples", + ) + median = statistics.median(samples) + medians.append(median) + noise.append( + 100 * statistics.median(abs(x - median) for x in samples) / median + ) + base, current = medians + regression = 100 * (current - base) / base + if metric["direction"] == "higher": + regression = -regression + require( + math.isfinite(regression) and all(math.isfinite(x) for x in noise), + f"{name}/{key}: measurement ratio overflow", + ) + limit = ( + policy["cases"][name] + .get("max_regression_percent", {}) + .get(key, metric["max_regression_percent"]) + ) + status = "PASS" + if regression > limit and not math.isclose(regression, limit, abs_tol=1e-9): + status = "FAIL" + if max(noise) > metric["max_mad_percent"]: + status = "ERROR" + rows.append( + { + "case": name, + "metric": key, + "unit": metric["unit"], + "baseline": base, + "candidate": current, + "regression_percent": regression, + "limit_percent": limit, + "mad_percent": noise, + "status": status, + } + ) + return rows + + +def report(args): + out = Path(args.output) + out.mkdir(parents=True, exist_ok=True) + rows, errors, revisions = [], [], [] + try: + policy = load_policy(args.policy) + base = json.loads(Path(args.baseline).read_text()) + candidate = json.loads(Path(args.candidate).read_text()) + rows = compare(policy, base, candidate, args.baseline_sha, args.candidate_sha) + revisions = [base["revision"], candidate["revision"]] + except (OSError, ValueError, KeyError, TypeError) as exc: + errors.append(str(exc)) + code = ( + 2 + if errors or any(row["status"] == "ERROR" for row in rows) + else int(any(row["status"] == "FAIL" for row in rows)) + ) + status = ("PASSED", "FAILED", "ERROR")[code] + lines = [f"## Performance Gate: {status}", ""] + if revisions: + lines += [f"Baseline: `{revisions[0]}` · Candidate: `{revisions[1]}`", ""] + lines += [ + "Positive change means a regression. Values are medians; MAD is median absolute deviation / median.", + "", + "| Case | Metric | Baseline | Candidate | Regression | Limit | MAD base / candidate | Result |", + "|---|---|---:|---:|---:|---:|---:|---|", + ] + for row in rows: + lines.append( + f"| {row['case']} | {row['metric']} ({row['unit']}) | {row['baseline']:.3f} | " + f"{row['candidate']:.3f} | {row['regression_percent']:+.2f}% | {row['limit_percent']:g}% | " + f"{row['mad_percent'][0]:.2f}% / {row['mad_percent'][1]:.2f}% | {row['status']} |" + ) + if row["status"] != "PASS": + reason = ( + "unstable measurements" + if row["status"] == "ERROR" + else f"regressed {row['regression_percent']:.2f}% (limit {row['limit_percent']:g}%)" + ) + annotation(f"{row['case']} {row['metric']}: {reason}") + for error in errors: + lines += ["", "ERROR: " + error.replace("\n", " ").replace("|", "\\|")] + annotation(error) + lines += [ + "", + f"Normalized samples, environment and comparison.json are in the {getattr(args, 'artifact_name', 'performance-results')} artifact. " + "Consult the job logs for raw benchmark diagnostics.", + ] + markdown = "\n".join(lines) + "\n" + (out / "summary.md").write_text(markdown) + write_json( + out / "comparison.json", {"status": status, "rows": rows, "errors": errors} + ) + if args.summary: + with Path(args.summary).open("a") as summary: + summary.write(markdown) + print(markdown) + return code + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("mode", choices=("collect", "compare")) + parser.add_argument( + "--policy", default=str(Path(__file__).with_name("policy.json")) + ) + parser.add_argument("--baseline", required=True) + parser.add_argument("--candidate", required=True) + parser.add_argument("--baseline-sha", required=True) + parser.add_argument("--candidate-sha", required=True) + parser.add_argument("--output", required=True) + parser.add_argument("--collection-timeout-seconds", type=float, default=900) + parser.add_argument("--summary", default=os.environ.get("GITHUB_STEP_SUMMARY")) + args = parser.parse_args() + return collect(args) if args.mode == "collect" else report(args) + + +if __name__ == "__main__": + + def interrupted(signum, frame): + raise RuntimeError(f"interrupted by signal {signum}") + + signal.signal(signal.SIGTERM, interrupted) + signal.signal(signal.SIGINT, interrupted) + sys.exit(main()) diff --git a/scripts/ci/performance/policy.json b/scripts/ci/performance/policy.json new file mode 100644 index 0000000000..595f8ac1a8 --- /dev/null +++ b/scripts/ci/performance/policy.json @@ -0,0 +1,30 @@ +{ + "schema_version": 1, + "repetitions": 5, + "cpu_count": 2, + "timeout_seconds": 120, + "arguments": { + "iterations": 2000000, + "warmup_iterations": 100000, + "pool_size_mb": 1024, + "failed_request_mb": 128, + "success_request_kb": 4, + "fragmentation_block_mb": 4, + "fragmentation_stride": 5, + "high_water_ratio": 0.9, + "mixed_failure_ratio": 0.05 + }, + "cases": { + "offset-success-1t": {"layer": "offset_allocator", "scenario": "success", "threads": 1, "result_scenario": "successful_allocate_free"}, + "offset-success-2t": {"layer": "offset_allocator", "scenario": "success", "threads": 2, "result_scenario": "successful_allocate_free"}, + "offset-capacity-1t": {"layer": "offset_allocator", "scenario": "capacity", "threads": 1, "result_scenario": "capacity_failed"}, + "offset-capacity-2t": {"layer": "offset_allocator", "scenario": "capacity", "threads": 2, "result_scenario": "capacity_failed"}, + "offset-fragmentation-1t": {"layer": "offset_allocator", "scenario": "fragmentation", "threads": 1, "result_scenario": "fragmentation_failed"}, + "offset-fragmentation-2t": {"layer": "offset_allocator", "scenario": "fragmentation", "threads": 2, "result_scenario": "fragmentation_failed"} + }, + "metrics": { + "qps": {"unit": "ops/s", "direction": "higher", "max_regression_percent": 5, "max_mad_percent": 5}, + "p50_ns": {"unit": "ns", "direction": "lower", "max_regression_percent": 5, "max_mad_percent": 5}, + "p99_ns": {"unit": "ns", "direction": "lower", "max_regression_percent": 10, "max_mad_percent": 10} + } +} diff --git a/scripts/ci/performance/rdma-policy.json b/scripts/ci/performance/rdma-policy.json new file mode 100644 index 0000000000..9c13ba2a6f --- /dev/null +++ b/scripts/ci/performance/rdma-policy.json @@ -0,0 +1,18 @@ +{ + "schema_version": 1, + "suite": "rdma", + "repetitions": 5, + "cpu_count": 4, + "timeout_seconds": 60, + "arguments": {"duration": 10, "total_buffer_size": 268435456}, + "cases": { + "read-4k": {"op_type": "read", "block_size": 4096, "batch_size": 1, "threads": 1, "metrics": ["p99_us"]}, + "write-4k": {"op_type": "write", "block_size": 4096, "batch_size": 1, "threads": 1, "metrics": ["p99_us"]}, + "read-1m": {"op_type": "read", "block_size": 1048576, "batch_size": 16, "threads": 4, "metrics": ["bandwidth_gbs"]}, + "write-1m": {"op_type": "write", "block_size": 1048576, "batch_size": 16, "threads": 4, "metrics": ["bandwidth_gbs"]} + }, + "metrics": { + "bandwidth_gbs": {"unit": "GB/s", "direction": "higher", "max_regression_percent": 5, "max_mad_percent": 5}, + "p99_us": {"unit": "us", "direction": "lower", "max_regression_percent": 10, "max_mad_percent": 10} + } +} diff --git a/scripts/ci/performance/rdma_collect.py b/scripts/ci/performance/rdma_collect.py new file mode 100644 index 0000000000..7cb1d81c1d --- /dev/null +++ b/scripts/ci/performance/rdma_collect.py @@ -0,0 +1,377 @@ +#!/usr/bin/env python3 +"""Collect paired TENT DRAM measurements on an exclusively leased RDMA pair.""" + +import argparse +from concurrent.futures import ThreadPoolExecutor +import hashlib +import ipaddress +import json +import os +from pathlib import Path +import re +import select +import shlex +import signal +import subprocess +import sys + +from perf_gate import load_policy, number, policy_digest, report, require, write_json + + +def benchmark_digest(directory): + fingerprint = hashlib.sha256() + paths = sorted(Path(directory).rglob("*")) + require(any(path.is_file() for path in paths), "missing benchmark source") + for path in paths: + if path.is_file(): + fingerprint.update( + str(path.relative_to(directory)).encode() + b"\0" + path.read_bytes() + ) + return fingerprint.hexdigest() + + +def validate_hosts(hosts): + require( + isinstance(hosts, list) and len(hosts) == 2, "exactly two hosts are required" + ) + for host in hosts: + require( + re.fullmatch( + r"[a-zA-Z0-9_][a-zA-Z0-9_.-]*@[a-zA-Z0-9][a-zA-Z0-9.-]*", host["ssh"] + ), + "invalid SSH destination", + ) + ipaddress.IPv4Address(host["address"]) + require(re.fullmatch(r"[a-zA-Z0-9_]+", host["device"]), "invalid RDMA device") + for key, minimum in (("numa", 0), ("port", 1), ("gid_index", 0)): + require(type(host[key]) is int and host[key] >= minimum, f"invalid {key}") + require( + hosts[0]["address"] != hosts[1]["address"] + and hosts[0]["ssh"] != hosts[1]["ssh"], + "hosts must differ", + ) + return hosts + + +def parse_sample(text, case, target, duration): + records = [json.loads(line) for line in text.splitlines() if line.strip()] + require(len(records) == 1, "expected one tebench JSONL record") + row = records[0] + expected = { + "schema_version": 1, + "record_type": "target_metrics", + "backend": "tent", + "op_type": case["op_type"], + "block_size": case["block_size"], + "batch_size": case["batch_size"], + "num_threads": case["threads"], + } + require( + all(row[key] == value for key, value in expected.items()), + "benchmark workload mismatch", + ) + require(len(row["targets"]) == 1, "expected exactly one RDMA target") + peer = row["targets"][0] + require( + peer["segment_name"] == target and peer["threads"] == case["threads"], + "target mismatch", + ) + count = row["aggregate_operations"] + require( + type(count) is int and count > 0 and peer["operations"] == count, + "invalid operation count", + ) + require( + row["aggregate_transferred_bytes"] + == peer["transferred_bytes"] + == count * case["block_size"] * case["batch_size"], + "incomplete transferred bytes", + ) + # target_metrics reports the average wall duration across its threads. + require( + number(peer["total_duration_us"]) + and peer["total_duration_us"] >= duration * 1e6, + "measurement ended before the requested duration", + ) + # tebench's *_gbps fields are decimal GB/s, not gigabits/s. + sample = { + "bandwidth_gbs": row["aggregate_throughput_gbps"], + "p99_us": peer["p99_us"], + } + require(all(number(value) for value in sample.values()), "invalid RDMA measurement") + return {key: sample[key] for key in case["metrics"]} + + +class Remote: + def __init__(self, host, benchmark, output, index): + self.destination = host["ssh"] + self.options = [ + "-o", + "BatchMode=yes", + "-o", + "StrictHostKeyChecking=yes", + "-o", + "ConnectTimeout=15", + ] + self.output = output / f"host-{index}" + self.output.mkdir() + self.root = None + self.process = None + self.stderr = (self.output / "ssh.log").open("w") + try: + self.root = self.ssh( + ["mktemp", "-d", "/tmp/mooncake-rdma-perf.XXXXXXXX"] + ).strip() + require( + re.fullmatch(r"/tmp/mooncake-rdma-perf\.[a-zA-Z0-9]+", self.root), + "invalid remote work directory", + ) + files = [ + str(Path(__file__).with_name(name)) + for name in ("rdma_worker.py", "perf_gate.py", "freeze_build.py") + ] + subprocess.run( + [ + "scp", + *self.options, + "-r", + *files, + str(benchmark), + f"{self.destination}:{self.root}/", + ], + check=True, + timeout=120, + stdout=self.stderr, + stderr=self.stderr, + ) + command = [ + "timeout", + "--kill-after=15s", + "2700s", + "python3", + "-u", + self.root + "/rdma_worker.py", + self.root, + ] + self.process = subprocess.Popen( + [ + "ssh", + *self.options, + "-o", + "ServerAliveInterval=15", + "-o", + "ServerAliveCountMax=2", + self.destination, + shlex.join(command), + ], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=self.stderr, + text=True, + ) + except BaseException: + self.close() + raise + + def ssh(self, command): + return subprocess.check_output( + ["ssh", *self.options, self.destination, shlex.join(command)], + text=True, + timeout=30, + stderr=self.stderr, + ) + + def request(self, op, **args): + self.process.stdin.write(json.dumps({"op": op, **args}) + "\n") + self.process.stdin.flush() + ready, _, _ = select.select( + [self.process.stdout], [], [], 2400 if op == "prepare" else 90 + ) + require(ready, "SSH worker response timed out") + response = json.loads(self.process.stdout.readline()) + require("error" not in response, response.get("error", "worker error")) + return response["value"] + + def close(self): + errors = [] + try: + if self.process: + self.process.stdin.close() + try: + self.process.wait(timeout=20) + except subprocess.TimeoutExpired: + self.process.terminate() + self.process.wait(timeout=10) + raise RuntimeError( + "SSH worker did not release its lease; remote watchdog bounds cleanup" + ) + if self.process.returncode != 0: + errors.append("SSH worker failed; see ssh.log") + if self.root: + try: + subprocess.run( + [ + "scp", + *self.options, + "-r", + f"{self.destination}:{self.root}/logs", + str(self.output), + ], + check=True, + timeout=120, + stdout=self.stderr, + stderr=self.stderr, + ) + except (OSError, subprocess.SubprocessError): + errors.append("could not retrieve all remote logs") + finally: + self.ssh(["rm", "-rf", "--", self.root]) + finally: + self.stderr.close() + require(not errors, "; ".join(errors)) + + +def collect(args, remote_class=Remote): + args.artifact_name = "T-One job" + output = Path(args.output) + output.mkdir(parents=True, exist_ok=False) + policy = load_policy(args.policy) + require(policy["suite"] == "rdma", "expected RDMA policy") + hosts = validate_hosts(json.loads(Path(args.hosts).read_text())) + for sha in (args.baseline_sha, args.candidate_sha, args.harness_sha): + require(re.fullmatch(r"[0-9a-f]{40}", sha), "expected full commit SHA") + benchmark = Path(args.benchmark).resolve() + environment = { + "hosts": [], + "harness_sha": args.harness_sha, + "benchmark_sha256": benchmark_digest(benchmark), + "run_id": args.run_id, + "host_config": hosts, + } + results = { + side: { + "schema_version": 1, + "revision": getattr(args, side + "_sha"), + "environment": environment, + "policy_sha256": policy_digest(policy), + "builds": [], + "cases": {}, + "errors": [], + } + for side in ("baseline", "candidate") + } + remotes = [] + try: + for index, host in enumerate(hosts): + remotes.append(remote_class(host, benchmark, output, index)) + with ThreadPoolExecutor(max_workers=2) as pool: + futures = [ + pool.submit( + remote.request, + "prepare", + host=host, + policy=policy, + baseline_sha=args.baseline_sha, + candidate_sha=args.candidate_sha, + ) + for remote, host in zip(remotes, hosts) + ] + prepared = [future.result() for future in futures] + environment["hosts"] = [item["environment"] for item in prepared] + require( + len({item["boot_id"] for item in environment["hosts"]}) == 2, + "both SSH destinations resolve to one host", + ) + for side in results: + results[side]["builds"] = [item["builds"][side] for item in prepared] + for name, case in policy["cases"].items(): + for repeat in range(policy["repetitions"]): + for side in ( + ("baseline", "candidate") + if repeat % 2 == 0 + else ("candidate", "baseline") + ): + tag = f"{name}-{repeat}-{side}" + print(tag, flush=True) + target = remotes[1].request("launch", side=side, tag=tag, case=name) + try: + if repeat == 0: + remotes[0].request( + "launch", + side=side, + tag=tag + "-verify", + case=name, + target=target, + verify=True, + ) + text = remotes[0].request( + "launch", side=side, tag=tag, case=name, target=target + ) + (output / (tag + ".jsonl")).write_text(text) + sample = parse_sample( + text, case, target, policy["arguments"]["duration"] + ) + values = results[side]["cases"].setdefault( + name, + { + key: { + "unit": policy["metrics"][key]["unit"], + "samples": [], + } + for key in case["metrics"] + }, + ) + for key, value in sample.items(): + values[key]["samples"].append(value) + finally: + remotes[1].request("stop") + for index, remote in enumerate(remotes): + require( + remote.request("probe") == environment["hosts"][index], + "RDMA environment changed during measurements", + ) + except Exception as exc: + for result in results.values(): + result["errors"].append(str(exc)) + finally: + for remote in reversed(remotes): + try: + remote.close() + except Exception as exc: + for result in results.values(): + result["errors"].append(f"cleanup: {exc}") + for side, result in results.items(): + write_json(output / (side + ".json"), result) + # T-One indexes .log files; this file contains JSON, never console text. + write_json(output / "rdma-results.log", {"run_id": args.run_id, **results}) + args.baseline, args.candidate = ( + str(output / "baseline.json"), + str(output / "candidate.json"), + ) + return report(args) + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description=__doc__) + for name in ( + "hosts", + "benchmark", + "baseline-sha", + "candidate-sha", + "harness-sha", + "run-id", + "output", + ): + parser.add_argument("--" + name, required=True) + parser.add_argument( + "--policy", default=str(Path(__file__).with_name("rdma-policy.json")) + ) + parser.add_argument("--summary", default=os.environ.get("GITHUB_STEP_SUMMARY")) + + def interrupted(signum, frame): + raise RuntimeError(f"interrupted by signal {signum}") + + signal.signal(signal.SIGTERM, interrupted) + signal.signal(signal.SIGINT, interrupted) + signal.signal(signal.SIGHUP, interrupted) + sys.exit(collect(parser.parse_args())) diff --git a/scripts/ci/performance/rdma_worker.py b/scripts/ci/performance/rdma_worker.py new file mode 100644 index 0000000000..9fe3e17be8 --- /dev/null +++ b/scripts/ci/performance/rdma_worker.py @@ -0,0 +1,354 @@ +#!/usr/bin/env python3 +"""Bounded SSH worker. One lease owns one host for the entire paired run.""" + +import hashlib +import json +import os +from pathlib import Path +import platform +import re +import resource +import shutil +import signal +import subprocess +import sys +import time + +from freeze_build import freeze +from perf_gate import require, stop_process as stop, write_json + + +def digest(path): + return hashlib.sha256(Path(path).read_bytes()).hexdigest() + + +class Worker: + def __init__(self, root): + self.root = Path(root).resolve() + self.logs = self.root / "logs" + self.logs.mkdir() + self.target = None + self.active = None + # Do not pass T-One credentials or inherited transport overrides to PR code. + self.env = { + key: os.environ[key] + for key in ("PATH", "HOME", "LANG") + if key in os.environ + } + + def command(self, argv, log, timeout=900, cwd=None, capture=False, env=None): + with (self.logs / log).open("a") as output: + self.active = subprocess.Popen( + ["timeout", "--kill-after=10s", f"{timeout}s", *argv], + cwd=cwd, + env=env or self.env, + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE if capture else output, + stderr=output, + text=True, + start_new_session=True, + ) + try: + stdout, _ = self.active.communicate(timeout=timeout + 15) + require(self.active.returncode == 0, f"command failed; see {log}") + return stdout + finally: + stop(self.active) + self.active = None + + def probe(self, host, policy): + device = Path("/sys/class/infiniband") / host["device"] + require(device.exists(), "configured RDMA device is absent") + require( + "/virtual/" not in str(device.resolve()), + "software RDMA is not a performance runner", + ) + port = device / "ports" / str(host["port"]) + require( + (port / "state").read_text().strip().startswith("4:"), + "RDMA port is not ACTIVE", + ) + node = Path(f"/sys/devices/system/node/node{host['numa']}") + cpus = set() + for item in (node / "cpulist").read_text().strip().split(","): + ends = [int(x) for x in item.split("-")] + cpus.update(range(ends[0], ends[-1] + 1)) + cpus &= os.sched_getaffinity(0) + require( + len(cpus) >= policy["cpu_count"], "NUMA node has too few available CPUs" + ) + soft, _ = resource.getrlimit(resource.RLIMIT_MEMLOCK) + require( + soft == resource.RLIM_INFINITY + or soft >= 2 * policy["arguments"]["total_buffer_size"], + "memlock limit is too small", + ) + info = self.command( + ["ibv_devinfo", "-d", host["device"], "-i", str(host["port"])], + "preflight.log", + timeout=15, + capture=True, + ) + cpuinfo = { + key.strip(): value.strip() + for line in Path("/proc/cpuinfo").read_text().split("\n\n")[0].splitlines() + if ":" in line + for key, value in [line.split(":", 1)] + if key.strip() in ("model name", "vendor_id", "microcode", "flags") + } + gid = (port / "gids" / str(host["gid_index"])).read_text().strip() + require(gid.replace(":", "").strip("0"), "configured GID is empty") + return { + "hostname": platform.node(), + "boot_id": Path("/proc/sys/kernel/random/boot_id").read_text().strip(), + "platform": platform.platform(), + "cpuinfo": cpuinfo, + "cpus": sorted(cpus), + "numa": host["numa"], + "device": host["device"], + "device_path": str(device.resolve()), + "port": host["port"], + "gid_index": host["gid_index"], + "gid": gid, + "rate": (port / "rate").read_text().strip(), + "ibv_devinfo": info, + "compiler": self.command( + ["c++", "--version"], "preflight.log", capture=True + ).splitlines()[0], + } + + def prepare(self, request): + self.host, self.policy = request["host"], request["policy"] + before = self.probe(self.host, self.policy) + builds = {} + for side in ("baseline", "candidate"): + sha = request[side + "_sha"] + require(re.fullmatch(r"[0-9a-f]{40}", sha), "invalid revision") + source = self.root / side + source.mkdir() + log = side + "-build.log" + for argv in ( + ["git", "init", "-q"], + [ + "git", + "remote", + "add", + "origin", + "https://github.com/kvcache-ai/Mooncake.git", + ], + ["git", "fetch", "--depth=1", "origin", sha], + ["git", "checkout", "--detach", "FETCH_HEAD"], + ["git", "submodule", "update", "--init", "--recursive", "--depth=1"], + ): + self.command(argv, log, cwd=source) + actual = self.command( + ["git", "rev-parse", "HEAD"], log, cwd=source, capture=True + ).strip() + require(actual == sha, "checkout SHA mismatch") + # Both revisions use the benchmark from the reviewed harness bundle. + benchmark = source / "mooncake-transfer-engine/benchmark" + shutil.rmtree(benchmark) + shutil.copytree(self.root / "benchmark", benchmark) + manifest = freeze(self.root / "baseline", self.root / "candidate") + write_json(self.logs / "build-config.json", manifest) + for side in ("baseline", "candidate"): + source = self.root / side + log = side + "-build.log" + self.command( + [ + "cmake", + "-S", + str(source), + "-B", + str(source / "build"), + "-G", + "Ninja", + "-DCMAKE_BUILD_TYPE=Release", + "-DCMAKE_CXX_COMPILER=c++", + "-DUSE_TENT=ON", + "-DUSE_CUDA=OFF", + "-DUSE_ETCD=OFF", + "-DWITH_STORE=OFF", + "-DWITH_STORE_RUST=OFF", + "-DBUILD_UNIT_TESTS=OFF", + "-DBUILD_EXAMPLES=OFF", + "-DBUILD_BENCHMARK=ON", + ], + log, + ) + self.command( + [ + "cmake", + "--build", + str(source / "build"), + "--parallel", + str(self.policy["cpu_count"]), + "--target", + "tebench", + ], + log, + ) + binary = source / "build/mooncake-transfer-engine/benchmark/tebench" + linked = self.command(["ldd", str(binary)], log, capture=True) + require("not found" not in linked, "missing benchmark shared library") + libraries = { + path: digest(path) + for path in re.findall(r"(?:=>\s+|^\s*)(/\S+)", linked, re.M) + } + builds[side] = { + "revision": request[side + "_sha"], + "build_config_sha256": manifest["sha256"], + "binary_sha256": digest(binary), + "libraries": libraries, + } + require( + before == self.probe(self.host, self.policy), "host changed during build" + ) + self.builds = builds + return {"environment": before, "builds": builds} + + def launch(self, request): + side, tag = request["side"], request["tag"] + require( + side in ("baseline", "candidate") and re.fullmatch(r"[a-z0-9_-]+", tag), + "invalid invocation", + ) + case = self.policy["cases"][request["case"]] + host = self.host + flags = { + "backend": "tent", + "xport_type": "rdma", + "tent_transport_hint": "rdma", + "seg_type": "DRAM", + "dram_numa_node": host["numa"], + "metadata_type": "p2p", + "seg_name": host["address"], + "op_type": case["op_type"], + **self.policy["arguments"], + } + for key in ("block_size", "batch_size"): + flags["start_" + key] = flags["max_" + key] = case[key] + flags["start_num_threads"] = flags["max_num_threads"] = case["threads"] + env = { + **self.env, + "MC_TENT_CONF": json.dumps({"rpc_server_hostname": host["address"]}), + "MC_TE_FILTERS": host["device"], + "MC_IB_PORT": str(host["port"]), + "MC_GID_INDEX": str(host["gid_index"]), + } + # A separate integrity smoke is excluded from performance samples. + if request.get("verify"): + flags.update(check_consistency="true", duration=1) + output = self.logs / (tag + ".jsonl") + require(not output.exists(), "refusing stale benchmark output") + if request.get("target"): + flags.update( + target_seg_name=request["target"], result_output_jsonl=str(output) + ) + argv = [ + "numactl", + f"--cpunodebind={host['numa']}", + f"--membind={host['numa']}", + str(self.root / side / "build/mooncake-transfer-engine/benchmark/tebench"), + ] + argv += [f"--{key}={value}" for key, value in flags.items()] + log = tag + ".log" + if request.get("target"): + self.command(argv, log, timeout=self.policy["timeout_seconds"], env=env) + return output.read_text() + require(self.target is None, "a target is already running") + with (self.logs / log).open("w") as stream: + # The worker watchdog bounds target lifetime. A timeout wrapper would + # forward stop()'s group SIGTERM a second time, forcing TENT to exit(1). + self.target = subprocess.Popen( + argv, + env=env, + stdin=subprocess.DEVNULL, + stdout=stream, + stderr=stream, + start_new_session=True, + ) + deadline = time.monotonic() + 30 + while time.monotonic() < deadline: + require(self.target.poll() is None, "RDMA target exited before readiness") + text = (self.logs / log).read_text() + match = re.search(r"--target_seg_name=([^\s\x1b]+)", text) + if match: + require( + re.fullmatch(re.escape(host["address"]) + r":[0-9]+", match[1]), + "target advertised a different RPC address", + ) + return match[1] + time.sleep(0.1) + raise TimeoutError("RDMA target readiness timed out") + + def handle(self, request): + if request["op"] == "prepare": + return self.prepare(request) + if request["op"] == "launch": + return self.launch(request) + if request["op"] == "probe": + for side, build in self.builds.items(): + binary = ( + self.root + / side + / "build/mooncake-transfer-engine/benchmark/tebench" + ) + require( + digest(binary) == build["binary_sha256"], + "benchmark binary changed during run", + ) + require( + all( + digest(path) == value + for path, value in build["libraries"].items() + ), + "linked library changed during run", + ) + return self.probe(self.host, self.policy) + if request["op"] == "stop": + require( + self.target is not None and self.target.poll() is None, + "RDMA target died during sample", + ) + stop(self.target) + require( + self.target.returncode == 0, "RDMA target did not shut down cleanly" + ) + self.target = None + return None + raise ValueError("unknown worker operation") + + +def serve(root): + import fcntl + + worker = None + # Stable path across runs. All templates using this pair must use the same account. + lock = Path.home() / ".mooncake-rdma-perf.lock" + with lock.open("a") as lease: + fcntl.flock(lease, fcntl.LOCK_EX | fcntl.LOCK_NB) + worker = Worker(root) + try: + for line in sys.stdin: + try: + result = {"value": worker.handle(json.loads(line))} + except Exception as exc: + result = {"error": str(exc)} + print(json.dumps(result, allow_nan=False), flush=True) + if "error" in result: + break + finally: + stop(worker.active) + stop(worker.target) + + +if __name__ == "__main__": + + def interrupted(signum, frame): + raise SystemExit(128 + signum) + + signal.signal(signal.SIGTERM, interrupted) + signal.signal(signal.SIGINT, interrupted) + signal.signal(signal.SIGHUP, interrupted) + serve(sys.argv[1]) diff --git a/scripts/ci/performance/tone/hosts.example.json b/scripts/ci/performance/tone/hosts.example.json new file mode 100644 index 0000000000..dd63702cb9 --- /dev/null +++ b/scripts/ci/performance/tone/hosts.example.json @@ -0,0 +1,4 @@ +[ + {"ssh": "benchmark@192.0.2.10", "address": "192.0.2.10", "device": "erdma_0", "port": 1, "gid_index": 0, "numa": 0}, + {"ssh": "benchmark@192.0.2.11", "address": "192.0.2.11", "device": "erdma_0", "port": 1, "gid_index": 0, "numa": 0} +] diff --git a/scripts/ci/performance/tone/mooncake-rdma-perf.conf b/scripts/ci/performance/tone/mooncake-rdma-perf.conf new file mode 100644 index 0000000000..f4d2a5036a --- /dev/null +++ b/scripts/ci/performance/tone/mooncake-rdma-perf.conf @@ -0,0 +1,2 @@ +test +paired diff --git a/scripts/ci/performance/tone/mooncake-rdma-perf/README.md b/scripts/ci/performance/tone/mooncake-rdma-perf/README.md new file mode 100644 index 0000000000..bbe845b936 --- /dev/null +++ b/scripts/ci/performance/tone/mooncake-rdma-perf/README.md @@ -0,0 +1,149 @@ +# T-One RDMA performance gate + +This is a tone-cli **functional** suite whose Pass/Fail result enforces a +performance policy. Version 1 measures TENT host DRAM RDMA read/write on a fixed +pair. It does not measure GPU Direct, Store RPC, or model-serving performance. +The suite is shipped with Mooncake so its adapter and controller can be reviewed +together. Installing these files and registering a T-One template are separate +deployment steps; adding the GitHub workflow does not provision hardware. + +## Provision and install + +1. Reserve two Linux RDMA hosts exclusively for this template. Install matching + RDMA kernel/userspace drivers, Mooncake build dependencies, C++20, CMake, + Ninja, Python 3.10+, Git, numactl, rdma-core utilities, GNU timeout, and OpenSSH. + Use a fixed image/toolchain. eRDMA hosts need a mode supporting Mooncake's OOB + QP setup; validate connectivity first. Keep eRDMA/RoCE/IB hardware profiles + and thresholds separate. RXE/SIW are not accepted as hardware measurements. +2. Give the T-One controller key-based SSH access to **both** hosts using the + same dedicated benchmark account for every job. Populate known_hosts in + advance; host-key verification is mandatory. Allow the pair's RDMA traffic + and p2p metadata RPC ports. Configure unlimited memlock (at least twice the + 256 MiB buffer is checked). Do not share these hosts with inference jobs. +3. Copy `scripts/ci/performance/tone/hosts.example.json` to + `/etc/mooncake-rdma/hosts.json` on the controller. Replace the documentation + addresses, device names, ports, GID indices, and NUMA nodes with the actual + pair. The first host is the initiator; the second is the target. Validate + with `ibv_devinfo -d DEVICE -i PORT` and `numactl --hardware`. This file is + operator-controlled, not supplied by the PR. +4. From a reviewed Mooncake checkout, install the adapter into tone-cli: + + ```bash + cp -R scripts/ci/performance/tone/mooncake-rdma-perf "$TONE_ROOT/tests/" + cp scripts/ci/performance/tone/mooncake-rdma-perf.conf "$TONE_ROOT/conf/functional/" + ``` + +5. Register `mooncake-rdma-perf` as a **functional** suite with the single + `paired` case. Create a template named `mooncake-rdma-perf` in workspace + `mooncake_test`, project `mooncake-ci`. Bind the reserved controller/pair and + set `RDMA_HOSTS_FILE=/etc/mooncake-rdma/hosts.json`. Do not reuse the SGLang + template's case list or mutable container images. Grant the existing T-One + CI account create/query/stop/log-read access to this template. +6. Verify `/api/job/log/query/` exports the suite's `rdma-results.log` and that + its download URL is reachable from GitHub over HTTPS. Set the GitHub variable + `TONE_RESULT_ORIGIN` to that exact origin (scheme + hostname + optional port, + no path). Redirects and other download origins are rejected. If the deployed + T-One storage endpoint only supports HTTP or private addresses, provision an + HTTPS download endpoint before enabling the gate. No credentials are sent + to the result-download URL. + +No changes to the existing `mooncake-ci-test` suite or its SGLang/vLLM jobs are +needed. The public tone-web API implementation is the integration contract; +confirm the deployed service's response shape during commissioning. + +## Inputs and execution + +The GitHub controller injects `BASELINE_SHA`, `CANDIDATE_SHA`, `HARNESS_SHA` +(all full 40-character commit IDs in kvcache-ai/Mooncake) and `RDMA_RUN_ID`. +No branch-tip clone or PR patch application is used. `install.sh` fetches the +exact harness revision. `RDMA_HOSTS_FILE` belongs to the template configuration. + +Each host acquires `~/.mooncake-rdma-perf.lock` for the entire run. A busy host +fails immediately; all suites sharing the pair must honor this lock/account. +The GitHub workflow serializes the fixed pair with `queue: max`, retaining up to +100 pending runs instead of replacing the previous pending run. Additional runs +are canceled when the queue is full. This queue does not cover jobs submitted +directly to T-One; the host leases still reject those if the pair is busy. +The two hosts build both revisions using the **same harness benchmark source**, +baseline tracked CMake recipes and Release flags, with CUDA/Store disabled and +TENT enabled. Candidate-only CMake recipes are removed from disposable checkouts; +target definitions and parent flags therefore cannot change the paired workload. +This compares implementation changes under a fixed build configuration, not +candidate build-system changes, and is not a sandbox for candidate C++ code. +Building on the +pair keeps the eRDMA provider and runtime libraries consistent; it does not +depend on `tebench` being present in a wheel. Builds finish before measurements. +Build-configuration, binary and linked-library hashes, compiler, device/port/GID, kernel, boot IDs, +NUMA CPU set, policy digest, harness identity and both revisions are recorded. + +`tebench` is forced to TENT + RDMA with an explicit RDMA transport hint and NIC +allowlist. `--dram_numa_node` selects one DRAM node, with CPU/memory NUMA binding +on each host. The node's available CPUs are used; this is not a promise of +single-core affinity. The default cases are 4 KiB read/write P99 (one thread, +batch 1) and 1 MiB read/write bandwidth (four threads, batch 16). + +An integrity smoke with `--check_consistency` runs before each case/version's +first measurement and is excluded from the samples. Every measured invocation +has tebench's built-in one-second warmup followed by ten seconds of measurement. +Five pairs alternate A/B, B/A, with both hosts switching to the same revision. +The policy gates bandwidth loss above 5% and P99 growth above 10%; excessive +relative MAD is an ERROR. These initial limits require hardware A/A calibration. + +## Results, cancellation and rollout + +`rdma-results.log` contains a JSON envelope with the run ID and both normalized +result sets. `.log` is intentional: tone-web's log API indexes that suffix. +`parse()` copies all results and native/build/SSH logs into +`TONE_CURRENT_RESULT_DIR`, then emits `rdma_perf: Pass` or `rdma_perf: Fail`. +Missing reports are Fail. An ERROR is also Fail in T-One; its reason is in the +comparison report. + +GitHub downloads this envelope and recomputes the verdict using its trusted +policy. It does not trust a returned Pass/Fail string or merely a completed job. +The Actions artifact contains the envelope, normalized samples, identity/build +manifests, job status, policy and comparison. The full raw native/build logs +remain in the T-One job artifacts. In both systems, 0 means pass, 1 regression, +and 2 invalid/incomplete/unstable execution. Summary tables and errors survive +failures. A 45-minute controller timeout or a cancellation requests T-One stop; +workers also have a 45-minute watchdog. SSH EOF normally releases each host +lease and stops only that job's processes. Network partitions/build timeouts can +delay cleanup until the worker timeout. Do not override a busy lease. + +Targets run directly in their own process group, without an inner `timeout` +wrapper that could forward the group's SIGTERM a second time. Normal stop sends +one SIGTERM and waits up to ten seconds for a clean exit before escalating to +SIGKILL; an already-dead target or a nonzero exit remains an ERROR. Initiator +commands retain their per-sample timeout, and the worker watchdog bounds target +lifetime if the controller disconnects without delivering EOF. + +The GitHub workflow uses existing `TONE_USER_NAME`/`TONE_USER_TOKEN` secrets. +Optional variables `TONE_WORKSPACE`, `TONE_PROJECT`, and `TONE_RDMA_TEMPLATE` +override the defaults. PR runs remain disabled until `RDMA_PERF_ENABLED=true`. +Manually dispatch the workflow first with the same full SHA for both inputs, +using a reviewed workflow revision that contains this harness. Native A/A +calibration, an injected regression, lost-target, and cancellation tests must +pass before making **RDMA Performance Gate** required. + +PR runs use the base revision's harness/policy/benchmark and compare the base +against the PR test merge commit. Candidate code runs only on the isolated pair. +Fork PRs do not receive GitHub secrets; if automatic gating is enabled they fail +closed without submitting a T-One job. Maintainers can run reviewed comparisons +manually; such a manual run is not automatically a required check on the fork PR. +Choose a fork-approval/check-publication policy before enabling a required check +for all contributors. This change does not alter branch protection or grant +fork code credentials. + +For direct controller reproduction (no T-One or GitHub credentials needed): + +```bash +python3 scripts/ci/performance/rdma_collect.py \ + --hosts /etc/mooncake-rdma/hosts.json \ + --benchmark mooncake-transfer-engine/benchmark \ + --baseline-sha FULL_BASE_SHA --candidate-sha FULL_HEAD_SHA \ + --harness-sha FULL_REVIEWED_HARNESS_SHA --run-id local-1 \ + --output fresh-rdma-results +``` + +Old revisions that cannot compile the frozen benchmark fail explicitly. They +are never replaced with a newer baseline. CPU/fixture tests validate the control +path only; they are not evidence of RDMA bandwidth, latency or A/A stability. diff --git a/scripts/ci/performance/tone/mooncake-rdma-perf/install.sh b/scripts/ci/performance/tone/mooncake-rdma-perf/install.sh new file mode 100644 index 0000000000..d3a2b4abe8 --- /dev/null +++ b/scripts/ci/performance/tone/mooncake-rdma-perf/install.sh @@ -0,0 +1,21 @@ +#!/bin/bash +# Install this directory as tone-cli/tests/mooncake-rdma-perf. +# Dependencies (including the matching RDMA provider) are provisioned on the pair. +DEP_PKG_LIST="" + +extract_src() ( + set -euo pipefail + [[ ${HARNESS_SHA:-} =~ ^[0-9a-f]{40}$ ]] + [[ ${RDMA_RUN_ID:-} =~ ^[a-zA-Z0-9_-]+$ ]] + echo 2 > "$TONE_BM_RUN_DIR/rdma-exit-$RDMA_RUN_ID" + source_dir="$TONE_BM_RUN_DIR/rdma-$RDMA_RUN_ID" + mkdir "$source_dir" + git -C "$source_dir" init -q + git -C "$source_dir" remote add origin https://github.com/kvcache-ai/Mooncake.git + git -C "$source_dir" fetch --depth=1 origin "$HARNESS_SHA" + git -C "$source_dir" checkout --detach FETCH_HEAD + test "$(git -C "$source_dir" rev-parse HEAD)" = "$HARNESS_SHA" +) + +build() { :; } +install() { :; } diff --git a/scripts/ci/performance/tone/mooncake-rdma-perf/run.sh b/scripts/ci/performance/tone/mooncake-rdma-perf/run.sh new file mode 100644 index 0000000000..f802c14a3a --- /dev/null +++ b/scripts/ci/performance/tone/mooncake-rdma-perf/run.sh @@ -0,0 +1,58 @@ +#!/bin/bash + +setup() { + echo 2 > "$TONE_BM_RUN_DIR/rdma-exit-$RDMA_RUN_ID" + test -f "${RDMA_HOSTS_FILE:-}" && + test -f "$TONE_BM_RUN_DIR/rdma-$RDMA_RUN_ID/scripts/ci/performance/rdma_collect.py" +} + +run() { + local source_dir="$TONE_BM_RUN_DIR/rdma-$RDMA_RUN_ID" + timeout --signal=TERM --kill-after=45s 3000s \ + python3 -u "$source_dir/scripts/ci/performance/rdma_collect.py" \ + --hosts "$RDMA_HOSTS_FILE" --benchmark "$source_dir/mooncake-transfer-engine/benchmark" \ + --baseline-sha "$BASELINE_SHA" --candidate-sha "$CANDIDATE_SHA" \ + --harness-sha "$HARNESS_SHA" --run-id "$RDMA_RUN_ID" \ + --output "$TONE_BM_RUN_DIR/results-$RDMA_RUN_ID" & + rdma_collector_pid=$! + local status=0 + wait "$rdma_collector_pid" || status=$? + echo "$status" > "$TONE_BM_RUN_DIR/rdma-exit-$RDMA_RUN_ID" + if ! kill -0 "$rdma_collector_pid" 2>/dev/null; then + rdma_collector_pid="" + fi + return "$status" +} + +teardown() { + if [[ -n ${rdma_collector_pid:-} ]]; then + kill -TERM "$rdma_collector_pid" 2>/dev/null || true + wait "$rdma_collector_pid" 2>/dev/null || true + rdma_collector_pid="" + fi +} + +parse() { + local results="$TONE_BM_RUN_DIR/results-$RDMA_RUN_ID" + if [[ -d "$results" ]]; then + cp -a "$results/." "$TONE_CURRENT_RESULT_DIR/" || return 1 + fi + python3 - "$results/comparison.json" "$TONE_BM_RUN_DIR/rdma-exit-$RDMA_RUN_ID" \ + "$TONE_CURRENT_RESULT_DIR/rdma-results.log" <<'PY' +import json +from pathlib import Path +import sys +try: + with open(sys.argv[1]) as source: + status = json.load(source)["status"] + expected = {"PASSED": 0, "FAILED": 1, "ERROR": 2}[status] + valid = int(Path(sys.argv[2]).read_text()) == expected + passed = valid and status == "PASSED" +except (OSError, ValueError, KeyError, TypeError): + valid = passed = False +if not valid: + # Never publish an earlier successful envelope after a failed setup/retry. + Path(sys.argv[3]).unlink(missing_ok=True) +print("rdma_perf: " + ("Pass" if passed else "Fail")) +PY +} diff --git a/scripts/ci/performance/tone_rdma.py b/scripts/ci/performance/tone_rdma.py new file mode 100644 index 0000000000..1554005e40 --- /dev/null +++ b/scripts/ci/performance/tone_rdma.py @@ -0,0 +1,285 @@ +#!/usr/bin/env python3 +"""Submit a pinned T-One RDMA job and independently enforce its measurements.""" + +import argparse +import base64 +import json +import os +from pathlib import Path +import re +import signal +import sys +import time +import urllib.error +import urllib.parse +import urllib.request + +from perf_gate import compare, load_policy, policy_digest, report, require, write_json +from rdma_collect import benchmark_digest + + +class NoRedirect(urllib.request.HTTPRedirectHandler): + def redirect_request(self, req, fp, code, msg, headers, newurl): + return None + + +class Tone: + def __init__(self): + self.username = os.environ["TONE_USER_NAME"] + self.token = os.environ["TONE_USER_TOKEN"] + require(self.username and self.token, "T-One credentials are required") + self.opener = urllib.request.build_opener(NoRedirect) + + def read(self, request, limit=4 * 1024 * 1024): + try: + with self.opener.open(request, timeout=30) as response: + body = response.read(limit + 1) + require(len(body) <= limit, "T-One response exceeds size limit") + return json.loads(body) + except urllib.error.URLError: + # A stop request's URL contains the signature; never print the exception URL. + raise RuntimeError("T-One HTTP request failed") from None + + def api(self, method, **fields): + signature = base64.b64encode( + f"{self.username}|{self.token}|{time.time()}".encode() + ).decode() + body = {"username": self.username, "signature": signature, **fields} + url = "https://tone.openanolis.cn/api/job/" + method + "/" + if method == "stop": + request = urllib.request.Request(url + "?" + urllib.parse.urlencode(body)) + else: + request = urllib.request.Request( + url, + data=json.dumps(body).encode(), + headers={"Content-Type": "application/json"}, + ) + response = self.read(request) + require(response.get("code") == 200, f"T-One {method} request was rejected") + return response.get("data") + + def result(self, job_id, origin): + records = self.api("log/query", job_id=job_id, page_size=100, page_num=1) + urls = { + url + for record in records + if record["test_suite"] == "mooncake-rdma-perf" + for url in record.get("logs", []) + if urllib.parse.urlsplit(url).path.endswith("/rdma-results.log") + } + require(len(urls) == 1, "expected exactly one rdma-results.log from T-One") + url = urllib.parse.urljoin(origin + "/", urls.pop()) + parsed = urllib.parse.urlsplit(url) + allowed = urllib.parse.urlsplit(origin) + require( + (parsed.scheme, parsed.netloc) == (allowed.scheme, allowed.netloc) + and not parsed.username + and not parsed.password, + "untrusted result download origin", + ) + return self.read(urllib.request.Request(url)) + + +def import_results(bundle, args, policy): + require(bundle["run_id"] == args.run_id, "stale T-One result") + build_configs = set() + for side in ("baseline", "candidate"): + result = bundle[side] + require( + isinstance(result["errors"], list) + and all(isinstance(error, str) for error in result["errors"]), + "invalid collection errors", + ) + env = result["environment"] + require( + env["run_id"] == args.run_id and env["harness_sha"] == args.harness_sha, + "result harness or run identity mismatch", + ) + require( + env["benchmark_sha256"] == benchmark_digest(args.benchmark), + "benchmark source mismatch", + ) + if not result["errors"]: + require( + len(env["hosts"]) == 2 + and len({host["boot_id"] for host in env["hosts"]}) == 2, + "missing independent host identities", + ) + require(len(result["builds"]) == 2, "missing build manifests") + for build in result["builds"]: + require( + build["revision"] == getattr(args, side + "_sha") + and re.fullmatch(r"[0-9a-f]{64}", build["build_config_sha256"]) + and re.fullmatch(r"[0-9a-f]{64}", build["binary_sha256"]) + and build["libraries"] + and all( + re.fullmatch(r"[0-9a-f]{64}", value) + for value in build["libraries"].values() + ), + "invalid build provenance", + ) + build_configs.add(build["build_config_sha256"]) + write_json(Path(args.output) / (side + ".json"), result) + require( + len(build_configs) <= 1, "baseline and candidate build configurations differ" + ) + return compare( + policy, + bundle["baseline"], + bundle["candidate"], + args.baseline_sha, + args.candidate_sha, + ) + + +def run(args, client=None): + args.artifact_name = "rdma-performance-results" + output = Path(args.output) + output.mkdir(parents=True, exist_ok=False) + args.baseline, args.candidate = ( + str(output / "baseline.json"), + str(output / "candidate.json"), + ) + job_id, finished, error, policy = None, False, None, None + try: + for sha in (args.baseline_sha, args.candidate_sha, args.harness_sha): + require( + re.fullmatch(r"[0-9a-f]{40}", sha), + "full baseline, candidate and harness SHAs are required", + ) + require(re.fullmatch(r"[a-zA-Z0-9_-]+", args.run_id), "invalid run id") + origin = urllib.parse.urlsplit(args.result_origin) + require( + origin.scheme == "https" + and origin.netloc + and origin.path in ("", "/") + and not origin.query + and not origin.fragment + and not origin.username, + "configure an HTTPS TONE_RESULT_ORIGIN", + ) + policy = load_policy(args.policy) + write_json(output / "policy.json", policy) + client = client or Tone() + env = { + "BASELINE_SHA": args.baseline_sha, + "CANDIDATE_SHA": args.candidate_sha, + "HARNESS_SHA": args.harness_sha, + "RDMA_RUN_ID": args.run_id, + } + job = client.api( + "create", + workspace=args.workspace, + project=args.project, + template=args.template, + name="mooncake-rdma-" + args.run_id, + env_ifs=" ", + env_info=" ".join(f"{key}={value}" for key, value in env.items()), + ) + job_id = job["id"] + require(str(job_id).isdigit(), "invalid T-One job id") + write_json(output / "job.json", {"id": job_id, **env}) + print(f"T-One RDMA job: {job_id}", flush=True) + deadline = time.monotonic() + args.timeout + while time.monotonic() < deadline: + status = client.api("query", job_id=job_id) + write_json(output / "job-status.json", status) + state = status["job_state"] + require( + state in ("pending", "pending_q", "running", "success", "fail", "stop"), + "unknown T-One job state", + ) + if state in ("success", "fail", "stop"): + finished = True + break + time.sleep(15) + require(finished, "T-One RDMA job timed out") + bundle = client.result(job_id, args.result_origin.rstrip("/")) + write_json(output / "rdma-results.log", bundle) + rows = import_results(bundle, args, policy) + api_failed = state != "success" or status.get("job_second_state") in ( + "fail", + "failed", + "stop", + ) + require( + not (api_failed and all(row["status"] == "PASS" for row in rows)), + "T-One execution failed despite passing measurements", + ) + except Exception as exc: + error = str(exc) + finally: + if job_id is not None and not finished: + try: + client.api("stop", job_id=job_id) + except Exception: + error = ( + error or "" + ) + "; could not cancel T-One job (remote worker has a watchdog)" + if error: + # Preserve partial normalized measurements, but never let them pass. + for side, sha in ( + ("baseline", args.baseline_sha), + ("candidate", args.candidate_sha), + ): + path = output / (side + ".json") + result = ( + json.loads(path.read_text()) + if path.exists() + else { + "schema_version": 1, + "revision": sha, + "policy_sha256": policy_digest(policy) if policy else "", + "environment": {}, + "cases": {}, + "errors": [], + } + ) + result["errors"].append(error) + write_json(path, result) + (output / "error.log").write_text(error + "\n") + code = report(args) + if job_id is not None: + note = f"\nT-One job ID: `{job_id}` (job-status.json contains the job link and execution status).\n" + for path in (output / "summary.md", args.summary): + if path: + with Path(path).open("a") as stream: + stream.write(note) + return code + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description=__doc__) + for name in ("baseline-sha", "candidate-sha", "harness-sha", "run-id", "output"): + parser.add_argument("--" + name, required=True) + parser.add_argument( + "--policy", default=str(Path(__file__).with_name("rdma-policy.json")) + ) + parser.add_argument( + "--benchmark", + default=str( + Path(__file__).resolve().parents[3] / "mooncake-transfer-engine/benchmark" + ), + ) + parser.add_argument("--summary", default=os.environ.get("GITHUB_STEP_SUMMARY")) + parser.add_argument( + "--workspace", default=os.environ.get("TONE_WORKSPACE") or "mooncake_test" + ) + parser.add_argument( + "--project", default=os.environ.get("TONE_PROJECT") or "mooncake-ci" + ) + parser.add_argument( + "--template", + default=os.environ.get("TONE_RDMA_TEMPLATE") or "mooncake-rdma-perf", + ) + parser.add_argument( + "--result-origin", default=os.environ.get("TONE_RESULT_ORIGIN", "") + ) + parser.add_argument("--timeout", type=int, default=2700) + + def interrupted(signum, frame): + raise RuntimeError(f"interrupted by signal {signum}") + + signal.signal(signal.SIGINT, interrupted) + signal.signal(signal.SIGTERM, interrupted) + sys.exit(run(parser.parse_args())) diff --git a/scripts/ci/tests/test_perf_build.py b/scripts/ci/tests/test_perf_build.py new file mode 100644 index 0000000000..00fb9ff2e4 --- /dev/null +++ b/scripts/ci/tests/test_perf_build.py @@ -0,0 +1,134 @@ +"""Exercise workload freezing against changes to CMake targets and parent flags.""" + +import importlib.util +from pathlib import Path +import shutil +import subprocess +import tempfile +import unittest + +SCRIPT = Path(__file__).resolve().parents[1] / "performance/freeze_build.py" +SPEC = importlib.util.spec_from_file_location("freeze_build", SCRIPT) +build = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(build) + + +class PerformanceBuildTest(unittest.TestCase): + def setUp(self): + temp = tempfile.TemporaryDirectory() + self.addCleanup(temp.cleanup) + self.root = Path(temp.name) + self.baseline, self.candidate = self.root / "base", self.root / "candidate" + for root, value in ((self.baseline, 11), (self.candidate, 23)): + (root / "bench").mkdir(parents=True) + (root / "cmake").mkdir() + (root / "CMakeLists.txt").write_text("""cmake_minimum_required(VERSION 3.16) +project(frozen_workload LANGUAGES CXX) +file(GLOB recipes "${CMAKE_CURRENT_SOURCE_DIR}/cmake/*.cmake") +foreach(recipe IN LISTS recipes) + include("${recipe}") +endforeach() +add_subdirectory(bench) +""") + (root / "cmake/flags.cmake").write_text( + "add_compile_definitions(WORKLOAD=7)\n" + ) + (root / "bench/CMakeLists.txt").write_text( + "add_executable(bench main.cpp ../implementation.cpp)\n" + ) + (root / "bench/main.cpp").write_text( + "#include \nint implementation();\nint main() { std::cout << WORKLOAD + implementation(); }\n" + ) + (root / "implementation.cpp").write_text( + f"int implementation() {{ return {value}; }}\n" + ) + self.git(root, "init", "-q") + self.git(root, "add", ".") + + def git(self, root, *args): + subprocess.run( + ["git", "-C", str(root), *args], check=True, capture_output=True, timeout=10 + ) + + def redirect_candidate(self): + (self.candidate / "CMakeLists.txt").write_text( + 'message(FATAL_ERROR "candidate project used")\n' + ) + (self.candidate / "bench/CMakeLists.txt").write_text( + 'message(FATAL_ERROR "candidate target used")\n' + ) + (self.candidate / "cmake/flags.cmake").write_text( + "add_compile_definitions(WORKLOAD=999)\n" + ) + (self.candidate / "cmake/extra.cmake").write_text( + 'message(FATAL_ERROR "new recipe used")\n' + ) + self.git(self.candidate, "add", ".") + + def test_freezes_parent_flags_targets_and_removes_added_recipes(self): + self.redirect_candidate() + implementation = (self.candidate / "implementation.cpp").read_bytes() + manifest = build.freeze(self.baseline, self.candidate) + self.assertEqual(len(manifest["sha256"]), 64) + self.assertEqual( + set(manifest["files"]), + {"CMakeLists.txt", "bench/CMakeLists.txt", "cmake/flags.cmake"}, + ) + for name in manifest["files"]: + self.assertEqual( + (self.baseline / name).read_bytes(), + (self.candidate / name).read_bytes(), + ) + self.assertFalse((self.candidate / "cmake/extra.cmake").exists()) + self.assertEqual( + (self.candidate / "implementation.cpp").read_bytes(), implementation + ) + + def test_refuses_symlinked_build_files_or_parent_directories(self): + for name in ("CMakeLists.txt", "cmake"): + with self.subTest(name=name): + path = self.candidate / name + saved = path.with_name(path.name + ".saved") + path.rename(saved) + path.symlink_to(self.baseline / name) + try: + with self.assertRaisesRegex(ValueError, "symlink"): + build.freeze(self.baseline, self.candidate) + finally: + path.unlink() + saved.rename(path) + + @unittest.skipUnless( + shutil.which("cmake") and shutil.which("c++"), "CMake and C++ are required" + ) + def test_native_build_uses_frozen_workload_and_candidate_implementation(self): + self.redirect_candidate() + build.freeze(self.baseline, self.candidate) + for root, expected in ((self.baseline, "18"), (self.candidate, "30")): + directory = self.root / (root.name + "-build") + for command in ( + [ + "cmake", + "-S", + str(root), + "-B", + str(directory), + "-DCMAKE_BUILD_TYPE=Release", + "-DCMAKE_CXX_COMPILER=c++", + ], + ["cmake", "--build", str(directory), "--target", "bench"], + ): + process = subprocess.run( + command, capture_output=True, text=True, timeout=60 + ) + self.assertEqual(process.returncode, 0, process.stdout + process.stderr) + self.assertEqual( + subprocess.check_output( + [str(directory / "bench/bench")], text=True, timeout=10 + ), + expected, + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/scripts/ci/tests/test_perf_gate.py b/scripts/ci/tests/test_perf_gate.py new file mode 100644 index 0000000000..6e100c7e9e --- /dev/null +++ b/scripts/ci/tests/test_perf_gate.py @@ -0,0 +1,407 @@ +import argparse +import copy +import importlib.util +import io +import json +import os +from pathlib import Path +import subprocess +import sys +import tempfile +import time +import unittest +from contextlib import redirect_stdout +from unittest.mock import patch + + +DIRECTORY = Path(__file__).resolve().parents[1] / "performance" +SPEC = importlib.util.spec_from_file_location("perf_gate", DIRECTORY / "perf_gate.py") +gate = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(gate) +BASE_SHA = "a" * 40 +CANDIDATE_SHA = "b" * 40 + + +class PerformanceGateTest(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.root = Path(self.temp.name) + self.policy = gate.load_policy(DIRECTORY / "policy.json") + self.policy_path = self.root / "policy.json" + self.baseline = self.result(BASE_SHA) + self.candidate = self.result(CANDIDATE_SHA) + + def result(self, sha): + return { + "schema_version": 1, + "revision": sha, + "environment": {"hostname": "test-host", "cpu_affinity": [0, 1]}, + "policy_sha256": gate.policy_digest(self.policy), + "binary_sha256": "c" * 64, + "errors": [], + "cases": { + name: { + key: {"unit": metric["unit"], "samples": [1000.0] * 5} + for key, metric in self.policy["metrics"].items() + } + for name in self.policy["cases"] + }, + } + + def set_samples(self, metric, values, side=None): + result = self.candidate if side is None else side + result["cases"][next(iter(self.policy["cases"]))][metric]["samples"] = values + + def run_gate(self): + for name, value in ( + ("baseline", self.baseline), + ("candidate", self.candidate), + ("policy", self.policy), + ): + (self.root / f"{name}.json").write_text(json.dumps(value)) + return subprocess.run( + [ + sys.executable, + str(DIRECTORY / "perf_gate.py"), + "compare", + "--policy", + str(self.policy_path), + "--baseline", + str(self.root / "baseline.json"), + "--candidate", + str(self.root / "candidate.json"), + "--baseline-sha", + BASE_SHA, + "--candidate-sha", + CANDIDATE_SHA, + "--output", + str(self.root / "report"), + "--summary", + str(self.root / "job-summary.md"), + ], + text=True, + capture_output=True, + timeout=10, + ) + + def test_pass_publishes_all_metrics_and_artifacts(self): + process = self.run_gate() + self.assertEqual(process.returncode, 0, process.stderr) + report = json.loads((self.root / "report/comparison.json").read_text()) + self.assertEqual(report["status"], "PASSED") + self.assertEqual(len(report["rows"]), len(self.policy["cases"]) * 3) + summary = (self.root / "job-summary.md").read_text() + self.assertIn(BASE_SHA, summary) + self.assertIn(CANDIDATE_SHA, summary) + self.assertIn("ops/s", summary) + self.assertNotIn("::error", process.stdout) + + def test_threshold_equality_passes_in_both_directions(self): + self.set_samples("qps", [950.0] * 5) + self.set_samples("p50_ns", [1050.0] * 5) + self.set_samples("p99_ns", [1100.0] * 5) + self.assertEqual(self.run_gate().returncode, 0) + + def test_one_regression_fails_despite_other_improvements(self): + self.set_samples("qps", [949.0] * 5) + self.set_samples("p50_ns", [500.0] * 5) + self.set_samples("p99_ns", [500.0] * 5) + process = self.run_gate() + self.assertEqual(process.returncode, 1, process.stderr) + self.assertIn("::error title=Performance Gate::", process.stdout) + self.assertIn("regressed 5.10%", process.stdout) + self.assertIn("FAILED", (self.root / "job-summary.md").read_text()) + self.assertTrue((self.root / "report/comparison.json").is_file()) + + def test_latency_regression_fails(self): + for metric, value in (("p50_ns", 1051), ("p99_ns", 1101)): + with self.subTest(metric=metric): + self.candidate = self.result(CANDIDATE_SHA) + self.set_samples(metric, [value] * 5) + self.assertEqual(self.run_gate().returncode, 1) + + def test_case_override_only_changes_its_own_threshold(self): + name = next(iter(self.policy["cases"])) + self.policy["cases"][name]["max_regression_percent"] = {"qps": 8} + self.baseline = self.result(BASE_SHA) + self.candidate = self.result(CANDIDATE_SHA) + self.set_samples("qps", [925] * 5) + self.assertEqual(self.run_gate().returncode, 0) + self.set_samples("p99_ns", [1110] * 5) + self.assertEqual(self.run_gate().returncode, 1) + + def test_negative_or_nonfinite_threshold_is_rejected(self): + for value in (-0.1, float("nan"), True, 10**400): + with self.subTest(value=value): + self.policy["metrics"]["qps"]["max_regression_percent"] = value + self.assertEqual(self.run_gate().returncode, 2) + + def test_improvements_and_one_outlier_pass(self): + self.set_samples("qps", [1200, 1200, 1200, 1200, 100]) + self.set_samples("p50_ns", [800] * 5) + self.set_samples("p99_ns", [800] * 5) + self.assertEqual(self.run_gate().returncode, 0) + + def test_noise_fails_as_error_in_either_revision(self): + for side in (self.baseline, self.candidate): + with self.subTest(revision=side["revision"]): + self.set_samples("qps", [700, 850, 1000, 1150, 1300], side) + process = self.run_gate() + self.assertEqual(process.returncode, 2) + self.assertIn("unstable measurements", process.stdout) + self.set_samples("qps", [1000] * 5, side) + + def test_invalid_results_fail_closed_with_report(self): + name = next(iter(self.policy["cases"])) + mutations = { + "missing case": lambda r: r["cases"].pop(name), + "extra case": lambda r: r["cases"].update(extra={}), + "missing metric": lambda r: r["cases"][name].pop("qps"), + "unit mismatch": lambda r: r["cases"][name]["qps"].update(unit="GB/s"), + "missing sample": lambda r: r["cases"][name]["qps"]["samples"].pop(), + "extra sample": lambda r: r["cases"][name]["qps"]["samples"].append(1000), + "wrong SHA": lambda r: r.update(revision=BASE_SHA), + "wrong version": lambda r: r.update(schema_version=2), + "wrong policy": lambda r: r.update(policy_sha256="bad"), + "wrong host": lambda r: r["environment"].update(hostname="another-host"), + "benchmark error": lambda r: r["errors"].append( + "unexpected allocation result" + ), + } + for description, mutate in mutations.items(): + with self.subTest(description=description): + self.candidate = self.result(CANDIDATE_SHA) + mutate(self.candidate) + process = self.run_gate() + self.assertEqual(process.returncode, 2, process.stdout + process.stderr) + self.assertIn( + "Performance Gate: ERROR", + (self.root / "report/summary.md").read_text(), + ) + + def test_nonfinite_zero_negative_and_boolean_samples_fail(self): + for value in (float("nan"), float("inf"), -float("inf"), 0, -1, True, "1000"): + with self.subTest(value=value): + self.set_samples("qps", [1000, 1000, value, 1000, 1000]) + self.assertEqual(self.run_gate().returncode, 2) + + def test_missing_input_still_writes_summary(self): + args = argparse.Namespace( + policy=DIRECTORY / "policy.json", + baseline=self.root / "missing.json", + candidate=self.root / "missing-candidate.json", + baseline_sha=BASE_SHA, + candidate_sha=CANDIDATE_SHA, + output=self.root / "report", + summary=None, + ) + with redirect_stdout(io.StringIO()): + self.assertEqual(gate.report(args), 2) + self.assertIn("missing.json", (self.root / "report/summary.md").read_text()) + + def test_malformed_json_still_writes_summary(self): + self.policy_path.write_text("{") + args = argparse.Namespace( + policy=self.policy_path, + baseline="unused", + candidate="unused", + baseline_sha=BASE_SHA, + candidate_sha=CANDIDATE_SHA, + output=self.root / "report", + summary=None, + ) + with redirect_stdout(io.StringIO()): + self.assertEqual(gate.report(args), 2) + self.assertTrue((self.root / "report/comparison.json").is_file()) + + def test_annotation_escapes_untrusted_line_breaks(self): + output = io.StringIO() + with redirect_stdout(output): + gate.annotation("bad%\r\n::notice::forged annotation") + self.assertEqual(len(output.getvalue().splitlines()), 1) + self.assertIn("%25%0D%0A", output.getvalue()) + + def test_invalid_policy_is_rejected(self): + for field, value in ( + ("repetitions", 0), + ("repetitions", True), + ("cpu_count", 0), + ("timeout_seconds", -1), + ("cases", {}), + ("metrics", {}), + ): + with self.subTest(field=field): + invalid = {**self.policy, field: value} + self.policy_path.write_text(json.dumps(invalid)) + with self.assertRaises(ValueError): + gate.load_policy(self.policy_path) + + def test_malformed_policy_containers_preserve_error_reports(self): + valid = copy.deepcopy(self.policy) + paths = ( + (), + ("arguments",), + ("cases",), + ("metrics",), + ("cases", "offset-success-1t"), + ("cases", "offset-success-1t", "max_regression_percent"), + ("metrics", "qps"), + ) + for path in paths: + for invalid in (None, [], ["not-an-object"], "not-an-object", 1): + with self.subTest(path=path, invalid=invalid): + self.policy = copy.deepcopy(valid) + if path: + parent = self.policy + for key in path[:-1]: + parent = parent[key] + parent[path[-1]] = invalid + else: + self.policy = invalid + process = self.run_gate() + self.assertEqual(process.returncode, 2, process.stderr) + comparison = json.loads( + (self.root / "report/comparison.json").read_text() + ) + self.assertEqual(comparison["status"], "ERROR") + self.assertIn( + "Performance Gate: ERROR", + (self.root / "report/summary.md").read_text(), + ) + + def fixture_binary(self, side, behavior="ok"): + path = self.root / side + path.write_text(f"""#!{sys.executable} +import csv +import os +import sys +import time +from pathlib import Path +args = dict(value[2:].split("=", 1) for value in sys.argv[1:]) +with Path(os.environ["PERF_TEST_ORDER"]).open("a") as order: + order.write({side!r} + "\\n") +if {behavior!r} == "timeout": + time.sleep(10) +if {behavior!r} == "exit": + sys.exit(7) +row = {{ + "layer": args["layer"], "scenario": "successful_allocate_free", + "threads": args["threads"], "iterations_per_thread": args["iterations"], + "operations": int(args["threads"]) * int(args["iterations"]), + "throughput_seconds": 1, "qps": 1000, "p50_ns": 100, "p99_ns": 200, + "throughput_unexpected_results": int({behavior!r} == "incorrect"), + "latency_unexpected_results": 0, +}} +if {behavior!r} == "malformed": + print("not a benchmark result") +else: + writer = csv.DictWriter(sys.stdout, fieldnames=list(row)) + writer.writeheader() + writer.writerow(row) +""") + path.chmod(0o755) + return path + + def collect_fixture(self, behavior="ok", budget=900): + self.policy["cases"] = { + "offset-success-1t": self.policy["cases"]["offset-success-1t"] + } + self.policy["arguments"]["iterations"] = 1000 + if behavior == "timeout": + self.policy["timeout_seconds"] = 1 + self.policy_path.write_text(json.dumps(self.policy)) + args = argparse.Namespace( + policy=self.policy_path, + baseline=self.fixture_binary("baseline"), + candidate=self.fixture_binary("candidate", behavior), + baseline_sha=BASE_SHA, + candidate_sha=CANDIDATE_SHA, + output=self.root / "samples", + collection_timeout_seconds=budget, + ) + args.output.mkdir() + gate.write_json(args.output / "build-config.json", {"sha256": "d" * 64}) + with patch.object( + gate, "environment", return_value={"hostname": "fixture"} + ), patch.dict( + os.environ, {"PERF_TEST_ORDER": str(self.root / "order")} + ), redirect_stdout(io.StringIO()): + code = gate.collect(args) + return code, args + + def test_collection_alternates_pairs_and_preserves_raw_results(self): + code, args = self.collect_fixture() + self.assertEqual(code, 0) + self.assertEqual( + (self.root / "order").read_text().splitlines(), + [ + "baseline", + "candidate", + "candidate", + "baseline", + "baseline", + "candidate", + "candidate", + "baseline", + "baseline", + "candidate", + ], + ) + baseline = json.loads((args.output / "baseline.json").read_text()) + candidate = json.loads((args.output / "candidate.json").read_text()) + self.assertEqual(len(list(args.output.glob("*.csv"))), 10) + self.assertEqual(len(list(args.output.glob("*.log"))), 10) + self.assertEqual( + json.loads((args.output / "build-config.json").read_text()), + {"sha256": "d" * 64}, + ) + rows = gate.compare(self.policy, baseline, candidate, BASE_SHA, CANDIDATE_SHA) + self.assertTrue(all(row["status"] == "PASS" for row in rows)) + with self.assertRaisesRegex(ValueError, "already contains results"): + gate.collect(args) + + def test_process_failures_and_corrupt_output_are_not_passes(self): + for behavior in ("exit", "incorrect", "malformed", "timeout"): + with self.subTest(behavior=behavior): + with tempfile.TemporaryDirectory() as directory: + self.root = Path(directory) + self.policy_path = self.root / "policy.json" + code, args = self.collect_fixture(behavior) + self.assertEqual(code, 2) + candidate = json.loads((args.output / "candidate.json").read_text()) + self.assertEqual(len(candidate["errors"]), 1) + self.assertEqual(candidate["cases"], {}) + self.assertEqual( + (self.root / "order").read_text().splitlines(), + ["baseline", "candidate"], + ) + self.assertEqual(len(list(args.output.glob("*.csv"))), 2) + args.baseline, args.candidate = ( + args.output / "baseline.json", + args.output / "candidate.json", + ) + args.summary = None + with redirect_stdout(io.StringIO()): + self.assertEqual(gate.report(args), 2) + self.assertEqual( + json.loads((args.output / "comparison.json").read_text())[ + "status" + ], + "ERROR", + ) + + def test_collection_budget_bounds_total_runtime(self): + start = time.monotonic() + # The first baseline consumes the budget; no subsequent sample may start. + with patch.object(gate.time, "monotonic", side_effect=[0, 0, 2]): + code, args = self.collect_fixture(budget=1) + self.assertEqual(code, 2) + self.assertLess(time.monotonic() - start, 5) + self.assertEqual((self.root / "order").read_text().splitlines(), ["baseline"]) + candidate = json.loads((args.output / "candidate.json").read_text()) + self.assertIn("collection time budget exhausted", candidate["errors"][0]) + + +if __name__ == "__main__": + unittest.main() diff --git a/scripts/ci/tests/test_rdma_perf.py b/scripts/ci/tests/test_rdma_perf.py new file mode 100644 index 0000000000..fcf192d364 --- /dev/null +++ b/scripts/ci/tests/test_rdma_perf.py @@ -0,0 +1,643 @@ +"""Protocol/fixture tests: these do not measure RDMA hardware performance.""" + +import argparse +import copy +from contextlib import redirect_stdout +import importlib +import io +import json +from pathlib import Path +import signal +import subprocess +import sys +import tempfile +import time +import unittest +from unittest.mock import Mock, patch + +DIRECTORY = Path(__file__).resolve().parents[1] / "performance" +sys.path.insert(0, str(DIRECTORY)) +gate = importlib.import_module("perf_gate") +collector = importlib.import_module("rdma_collect") +worker = importlib.import_module("rdma_worker") +tone = importlib.import_module("tone_rdma") + +BASE, HEAD, HARNESS = "a" * 40, "b" * 40, "c" * 40 +HOSTS = [ + { + "ssh": f"benchmark@192.0.2.{i}", + "address": f"192.0.2.{i}", + "device": "erdma_0", + "port": 1, + "gid_index": 0, + "numa": 0, + } + for i in (10, 11) +] + + +def native(case, target, bandwidth=20, latency=10): + size = 100 * case["block_size"] * case["batch_size"] + return { + "schema_version": 1, + "record_type": "target_metrics", + "backend": "tent", + "op_type": case["op_type"], + "block_size": case["block_size"], + "batch_size": case["batch_size"], + "num_threads": case["threads"], + "aggregate_operations": 100, + "aggregate_transferred_bytes": size, + "aggregate_throughput_gbps": bandwidth, + "targets": [ + { + "segment_name": target, + "threads": case["threads"], + "operations": 100, + "transferred_bytes": size, + "total_duration_us": 10e6, + "p99_us": latency, + } + ], + } + + +class FixtureRemote: + calls = [] + failure = None + bandwidth_ratio = 1 + + def __init__(self, host, benchmark, output, index): + self.index = index + + def request(self, op, **args): + self.calls.append((self.index, op, args)) + if op == self.failure: + raise RuntimeError("injected " + op + " failure") + environment = { + "hostname": f"host-{self.index}", + "boot_id": f"boot-{self.index}", + } + if op == "prepare": + self.policy = args["policy"] + return { + "environment": environment, + "builds": { + side: { + "revision": args[side + "_sha"], + "binary_sha256": "d" * 64, + "build_config_sha256": "f" * 64, + "libraries": {"/lib/provider.so": "e" * 64}, + } + for side in ("baseline", "candidate") + }, + } + if op == "probe": + return environment + if op == "launch": + if "target" not in args: + return "192.0.2.11:12345" + ratio = self.bandwidth_ratio if args["side"] == "candidate" else 1 + return ( + json.dumps( + native( + self.policy["cases"][args["case"]], + args["target"], + bandwidth=20 * ratio, + ) + ) + + "\n" + ) + return None + + def close(self): + self.calls.append((self.index, "close", {})) + if self.failure == "close": + raise RuntimeError("injected cleanup failure") + + +class RDMAPerformanceTest(unittest.TestCase): + def setUp(self): + temp = tempfile.TemporaryDirectory() + self.addCleanup(temp.cleanup) + self.root = Path(temp.name) + self.policy = gate.load_policy(DIRECTORY / "rdma-policy.json") + self.hosts = self.root / "hosts.json" + self.hosts.write_text(json.dumps(HOSTS)) + self.benchmark = self.root / "benchmark" + self.benchmark.mkdir() + (self.benchmark / "fixture.cpp").write_text("fixture") + self.args = argparse.Namespace( + policy=DIRECTORY / "rdma-policy.json", + hosts=self.hosts, + benchmark=self.benchmark, + baseline_sha=BASE, + candidate_sha=HEAD, + harness_sha=HARNESS, + run_id="123-1", + output=self.root / "results", + summary=self.root / "summary.md", + ) + FixtureRemote.calls = [] + FixtureRemote.failure = None + FixtureRemote.bandwidth_ratio = 1 + + def collect(self): + with redirect_stdout(io.StringIO()): + return collector.collect(self.args, FixtureRemote) + + def bundle(self): + self.assertEqual(self.collect(), 0) + return json.loads((self.args.output / "rdma-results.log").read_text()) + + def client_args(self): + return argparse.Namespace( + **{ + **vars(self.args), + "output": self.root / "github", + "workspace": "test", + "project": "test", + "template": "mooncake-rdma-perf", + "result_origin": "https://logs.example.com", + "timeout": 60, + } + ) + + def run_client(self, bundle, state="success", **changes): + client = Mock() + client.api.side_effect = [{"id": 42}, {"job_state": state}] + client.result.return_value = bundle + args = self.client_args() + for key, value in changes.items(): + setattr(args, key, value) + with redirect_stdout(io.StringIO()): + code = tone.run(args, client) + return code, args, client + + def test_paired_collection_orders_versions_checks_integrity_and_preserves_evidence( + self, + ): + self.assertEqual(self.collect(), 0) + calls = FixtureRemote.calls + launches = [ + args + for index, op, args in calls + if index == 0 and op == "launch" and not args.get("verify") + ] + self.assertEqual( + [item["side"] for item in launches[:6]], + ["baseline", "candidate", "candidate", "baseline", "baseline", "candidate"], + ) + self.assertEqual(len(launches), 40) + self.assertEqual( + sum(args.get("verify", False) for _, op, args in calls if op == "launch"), 8 + ) + self.assertEqual(sum(op == "stop" for _, op, _ in calls), 40) + self.assertEqual(sum(op == "close" for _, op, _ in calls), 2) + summary = json.loads((self.args.output / "comparison.json").read_text()) + self.assertEqual(len(summary["rows"]), 4) + self.assertEqual(summary["status"], "PASSED") + self.assertEqual(len(list(self.args.output.glob("*.jsonl"))), 40) + self.assertIn("GB/s", self.args.summary.read_text()) + + def test_regression_fails_locally_and_in_github_with_a_table(self): + FixtureRemote.bandwidth_ratio = 0.94 + self.assertEqual(self.collect(), 1) + bundle = json.loads((self.args.output / "rdma-results.log").read_text()) + code, args, _ = self.run_client(bundle, state="fail") + self.assertEqual(code, 1) + summary = (args.output / "summary.md").read_text() + self.assertIn("+6.00%", summary) + self.assertIn("FAILED", summary) + + def test_collection_failures_close_both_hosts_and_write_error_reports(self): + for failure in ("prepare", "launch", "stop", "probe", "close"): + with self.subTest(failure=failure): + FixtureRemote.failure = failure + FixtureRemote.calls = [] + self.args.output = self.root / failure + self.assertEqual(self.collect(), 2) + self.assertEqual( + sum(op == "close" for _, op, _ in FixtureRemote.calls), 2 + ) + self.assertIn("ERROR", (self.args.output / "summary.md").read_text()) + self.assertTrue((self.args.output / "rdma-results.log").exists()) + + def test_native_parser_rejects_wrong_backend_workload_target_truncated_or_invalid_data( + self, + ): + case = self.policy["cases"]["read-1m"] + good = native(case, "peer") + edits = [ + lambda x: x.update(backend="classic"), + lambda x: x.update(block_size=4096), + lambda x: x.update(num_threads=1), + lambda x: x.update(op_type="write"), + lambda x: x.update(aggregate_transferred_bytes=1), + lambda x: x.update(aggregate_operations=0), + lambda x: x.update(aggregate_throughput_gbps=float("nan")), + lambda x: x["targets"][0].update(segment_name="wrong"), + lambda x: x["targets"][0].update(total_duration_us=1), + lambda x: x.update(targets=[]), + ] + for edit in edits: + bad = copy.deepcopy(good) + edit(bad) + with self.assertRaises((ValueError, KeyError)): + collector.parse_sample(json.dumps(bad), case, "peer", 10) + with self.assertRaises(ValueError): + collector.parse_sample( + json.dumps(good) + "\n" + json.dumps(good), case, "peer", 10 + ) + self.assertEqual( + collector.parse_sample(json.dumps(good), case, "peer", 10), + {"bandwidth_gbs": 20}, + ) + + def test_hosts_validate_distinct_ips_and_reject_shell_injection(self): + self.assertEqual(collector.validate_hosts(HOSTS), HOSTS) + for key, value in ( + ("ssh", "-oProxyCommand=bad"), + ("address", "not-an-ip"), + ("device", "erdma_0;bad"), + ("numa", -1), + ("port", 0), + ("gid_index", True), + ): + bad = copy.deepcopy(HOSTS) + bad[0][key] = value + with self.assertRaises(ValueError): + collector.validate_hosts(bad) + with self.assertRaises(ValueError): + collector.validate_hosts([HOSTS[0], HOSTS[0]]) + + def test_github_recomputes_verdict_and_rejects_stale_or_mismatched_provenance(self): + bundle = self.bundle() + code, _, client = self.run_client(bundle) + self.assertEqual(code, 0) + create = client.api.call_args_list[0] + self.assertIn("HARNESS_SHA=" + HARNESS, create.kwargs["env_info"]) + edits = [ + lambda x: x.update(run_id="old"), + lambda x: x["candidate"].update(revision=BASE), + lambda x: x["candidate"]["environment"].update(harness_sha=HEAD), + lambda x: x["candidate"].update(builds=[]), + lambda x: x["candidate"]["builds"][0].update(build_config_sha256="0" * 64), + lambda x: x["candidate"]["builds"][0].pop("build_config_sha256"), + lambda x: x["baseline"].update(errors="malformed errors"), + lambda x: x["baseline"].update(errors=[{}]), + lambda x: x["candidate"].update(policy_sha256="wrong"), + lambda x: x["candidate"]["environment"].update( + hosts=[{"boot_id": "same"}] * 2 + ), + lambda x: x["candidate"]["cases"].pop("write-1m"), + ] + for index, edit in enumerate(edits): + bad = copy.deepcopy(bundle) + edit(bad) + code, args, _ = self.run_client(bad, output=self.root / f"bad-{index}") + self.assertEqual(code, 2) + self.assertIn("ERROR", (args.output / "summary.md").read_text()) + self.assertEqual( + json.loads((args.output / "comparison.json").read_text())["status"], + "ERROR", + ) + + def test_infrastructure_failure_cannot_be_masked_by_passing_measurements(self): + code, args, _ = self.run_client(self.bundle(), state="fail") + self.assertEqual(code, 2) + self.assertIn("execution failed", (args.output / "summary.md").read_text()) + + def test_timeout_and_query_error_cancel_the_remote_job(self): + for index, timeout in enumerate((0, 60)): + args = self.client_args() + args.timeout, args.output = timeout, self.root / f"timeout-{index}" + client = Mock() + client.api.side_effect = ( + [{"id": 42}, None] + if timeout == 0 + else [{"id": 42}, RuntimeError("query failed"), None] + ) + with redirect_stdout(io.StringIO()): + self.assertEqual(tone.run(args, client), 2) + self.assertEqual(client.api.call_args.args, ("stop",)) + self.assertEqual(client.api.call_args.kwargs, {"job_id": 42}) + self.assertTrue((args.output / "summary.md").exists()) + + def test_missing_credentials_or_unknown_job_state_fail_closed(self): + args = self.client_args() + with patch.dict( + "os.environ", {"TONE_USER_NAME": "", "TONE_USER_TOKEN": ""} + ), redirect_stdout(io.StringIO()): + self.assertEqual(tone.run(args), 2) + self.assertIn("credentials", (args.output / "summary.md").read_text()) + client = Mock() + client.api.side_effect = [{"id": 42}, {"job_state": "unknown"}, None] + args.output = self.root / "unknown" + with redirect_stdout(io.StringIO()): + self.assertEqual(tone.run(args, client), 2) + self.assertEqual(client.api.call_args.args, ("stop",)) + + def test_log_download_contract_rejects_missing_duplicates_and_other_origins(self): + client = tone.Tone.__new__(tone.Tone) + client.api = Mock() + client.read = Mock(return_value={"run_id": "123-1"}) + record = { + "test_suite": "mooncake-rdma-perf", + "logs": ["https://logs.example.com/job/rdma-results.log"], + } + client.api.return_value = [record] + self.assertEqual( + client.result(42, "https://logs.example.com"), {"run_id": "123-1"} + ) + self.assertNotIn("Authorization", client.read.call_args.args[0].headers) + for urls in ( + [], + ["https://other.example.com/rdma-results.log"], + [ + "https://logs.example.com/one/rdma-results.log", + "https://logs.example.com/two/rdma-results.log", + ], + ): + client.api.return_value = [{**record, "logs": urls}] + with self.assertRaises(ValueError): + client.result(42, "https://logs.example.com") + + def test_tone_parse_reports_missing_results_as_fail(self): + suite = DIRECTORY / "tone/mooncake-rdma-perf/run.sh" + result = subprocess.run( + ["bash", "-c", 'source "$1"; parse', "bash", str(suite)], + env={ + "PATH": "/usr/bin:/bin", + "TONE_BM_RUN_DIR": str(self.root), + "RDMA_RUN_ID": "absent", + "TONE_CURRENT_RESULT_DIR": str(self.root), + }, + capture_output=True, + text=True, + check=True, + ) + self.assertIn("rdma_perf: Fail", result.stdout) + + def test_worker_process_group_is_stopped(self): + process = subprocess.Popen( + [sys.executable, "-c", "import time; time.sleep(60)"], + start_new_session=True, + ) + self.addCleanup(worker.stop, process) + worker.stop(process) + self.assertIsNotNone(process.returncode) + + def test_process_group_cleanup_reaches_descendants(self): + marker = self.root / "stopped" + ready = self.root / "ready" + child = f"""import signal, time +from pathlib import Path +def terminate(signum, frame): + Path({str(marker)!r}).write_text("stopped") + raise SystemExit(0) +signal.signal(signal.SIGTERM, terminate) +Path({str(ready)!r}).write_text("ready") +while True: + time.sleep(1) +""" + parent = f"""import signal, subprocess, sys +signal.signal(signal.SIGTERM, signal.SIG_IGN) +child = subprocess.Popen([sys.executable, "-c", {child!r}]) +child.wait() +""" + process = subprocess.Popen( + [sys.executable, "-c", parent], start_new_session=True + ) + self.addCleanup(worker.stop, process) + for _ in range(100): + if ready.exists(): + break + time.sleep(0.01) + self.assertTrue(ready.exists(), "child did not become ready") + worker.stop(process) + self.assertEqual(process.returncode, 0) + self.assertEqual(marker.read_text(), "stopped") + + def test_worker_launch_pins_rdma_and_numa_and_manages_target_lifetime(self): + remote = self.root / "remote" + remote.mkdir() + with patch.dict( + "os.environ", + {"TONE_USER_TOKEN": "private-fixture-token", "MC_FORCE_TCP": "1"}, + ): + instance = worker.Worker(remote) + instance.host, instance.policy = HOSTS[0], self.policy + shim = self.root / "bin" + shim.mkdir() + numactl = shim / "numactl" + numactl.write_text('#!/bin/sh\nshift\nshift\nexec "$@"\n') + numactl.chmod(0o755) + timeout = shim / "timeout" + timeout.write_text( + f"#!{sys.executable}\n" + + """ +import signal, subprocess, sys, time +args = sys.argv[1:] +while args[0].startswith("-"): + args.pop(0) +child = subprocess.Popen(args[1:]) +def forward(signum, frame): + # Like GNU timeout, forward a signal also delivered to the child by killpg. + # Delay forwarding so the fixture observes two distinct signal deliveries. + time.sleep(0.05) + child.send_signal(signum) +signal.signal(signal.SIGTERM, forward) +sys.exit(child.wait()) +""" + ) + timeout.chmod(0o755) + instance.env["PATH"] = str(shim) + ":/usr/bin:/bin" + binary = remote / "baseline/build/mooncake-transfer-engine/benchmark/tebench" + binary.parent.mkdir(parents=True) + binary.write_text( + f"#!{sys.executable}\n" + + """ +import json, os, pathlib, signal, sys, time +signals = 0 +def terminate(signum, frame): + global signals + signals += 1 + pathlib.Path(__file__).with_suffix(".signals").write_text(str(signals)) + if signals > 1: + sys.exit(1) + # Model TENT's graceful teardown window and second-signal failure. + time.sleep(0.2) + sys.exit(0) +signal.signal(signal.SIGTERM, terminate) +flags = dict(arg[2:].split("=", 1) for arg in sys.argv[1:]) +if "target_seg_name" not in flags: + print("./tebench --target_seg_name=192.0.2.10:12345 --seg_type=DRAM", flush=True) + time.sleep(60) +else: + pathlib.Path(flags["result_output_jsonl"]).write_text(json.dumps({"flags": flags, "env": dict(os.environ)})) +""" + ) + binary.chmod(0o755) + request = {"side": "baseline", "tag": "fixture", "case": "read-4k"} + try: + target = instance.launch(request) + self.assertEqual(target, "192.0.2.10:12345") + result = json.loads( + instance.launch({**request, "tag": "client", "target": target}) + ) + self.assertEqual(result["flags"]["xport_type"], "rdma") + self.assertEqual(result["flags"]["tent_transport_hint"], "rdma") + self.assertEqual(result["flags"]["backend"], "tent") + self.assertEqual(result["flags"]["dram_numa_node"], "0") + self.assertEqual(result["flags"]["start_block_size"], "4096") + self.assertEqual(result["flags"]["max_block_size"], "4096") + self.assertEqual(result["env"]["MC_TE_FILTERS"], "erdma_0") + self.assertEqual( + json.loads(result["env"]["MC_TENT_CONF"]), + {"rpc_server_hostname": "192.0.2.10"}, + ) + self.assertNotIn("TONE_USER_TOKEN", result["env"]) + self.assertNotIn("MC_FORCE_TCP", result["env"]) + pid = instance.target + instance.handle({"op": "stop"}) + self.assertEqual(pid.returncode, 0) + self.assertEqual(binary.with_suffix(".signals").read_text(), "1") + self.assertIsNone(instance.target) + finally: + worker.stop(instance.target) + + def test_worker_stop_rejects_a_dead_or_unresponsive_target(self): + remote = self.root / "remote" + remote.mkdir() + instance = worker.Worker(remote) + dead = subprocess.Popen( + [sys.executable, "-c", "raise SystemExit(7)"], start_new_session=True + ) + self.addCleanup(worker.stop, dead) + dead.wait(timeout=5) + instance.target = dead + with self.assertRaisesRegex(ValueError, "target died during sample"): + instance.handle({"op": "stop"}) + + ready = self.root / "ready" + hung = subprocess.Popen( + [ + sys.executable, + "-c", + "import signal, time; from pathlib import Path; " + "signal.signal(signal.SIGTERM, signal.SIG_IGN); " + f"Path({str(ready)!r}).touch(); time.sleep(60)", + ], + start_new_session=True, + ) + self.addCleanup(worker.stop, hung) + deadline = time.monotonic() + 5 + while not ready.exists() and time.monotonic() < deadline: + time.sleep(0.01) + self.assertTrue(ready.exists(), "target did not become ready") + instance.target = hung + wait = hung.wait + with patch.object( + hung, "wait", side_effect=lambda timeout: wait(min(timeout, 0.2)) + ): + with self.assertRaisesRegex(ValueError, "did not shut down cleanly"): + instance.handle({"op": "stop"}) + self.assertEqual(hung.returncode, -signal.SIGKILL) + + def test_worker_lease_excludes_another_job_and_releases_on_eof(self): + import os + + first_root, second_root = self.root / "lease-first", self.root / "lease-second" + first_root.mkdir() + second_root.mkdir() + env = {**os.environ, "HOME": str(self.root)} + command = [sys.executable, str(DIRECTORY / "rdma_worker.py")] + first = subprocess.Popen( + command + [str(first_root)], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + env=env, + ) + try: + deadline = time.monotonic() + 5 + while not (first_root / "logs").exists() and time.monotonic() < deadline: + time.sleep(0.01) + self.assertTrue((first_root / "logs").exists()) + second = subprocess.run( + command + [str(second_root)], + input="", + text=True, + capture_output=True, + env=env, + timeout=5, + ) + self.assertNotEqual(second.returncode, 0) + first.communicate(timeout=5) + self.assertEqual(first.returncode, 0) + released = subprocess.run( + command + [str(second_root)], + input="", + text=True, + capture_output=True, + env=env, + timeout=5, + ) + self.assertEqual(released.returncode, 0, released.stderr) + finally: + if first.poll() is None: + first.terminate() + first.communicate(timeout=5) + + def test_tone_retry_cannot_republish_a_stale_passing_envelope(self): + import os + + result_dir, publish_dir = self.root / "results-retry", self.root / "publish" + result_dir.mkdir() + publish_dir.mkdir() + (result_dir / "comparison.json").write_text('{"status":"PASSED"}') + (result_dir / "rdma-results.log").write_text('{"run_id":"retry"}') + (self.root / "rdma-exit-retry").write_text("2") + result = subprocess.run( + [ + "bash", + "-c", + 'source "$1"; parse', + "bash", + str(DIRECTORY / "tone/mooncake-rdma-perf/run.sh"), + ], + env={ + **os.environ, + "TONE_BM_RUN_DIR": str(self.root), + "RDMA_RUN_ID": "retry", + "TONE_CURRENT_RESULT_DIR": str(publish_dir), + }, + capture_output=True, + text=True, + check=True, + ) + self.assertIn("rdma_perf: Fail", result.stdout) + self.assertFalse((publish_dir / "rdma-results.log").exists()) + + def test_http_error_redacts_signature_and_disallows_redirects(self): + client = tone.Tone.__new__(tone.Tone) + client.username, client.token = "fixture", "do-not-print-this" + client.opener = Mock() + client.opener.open.side_effect = tone.urllib.error.URLError( + "url with do-not-print-this" + ) + with self.assertRaisesRegex(RuntimeError, "^T-One HTTP request failed$"): + client.api("stop", job_id=42) + request = client.opener.open.call_args.args[0] + self.assertEqual(request.get_method(), "GET") + self.assertIn("signature=", request.full_url) + self.assertIsNone( + tone.NoRedirect().redirect_request( + None, None, 302, None, None, "https://other.example.com" + ) + ) + + +if __name__ == "__main__": + unittest.main()