diff --git a/.github/workflows/asv-benchmarking-pr.yml b/.github/workflows/asv-benchmarking-pr.yml index 66eca2401..3c1af3ca0 100644 --- a/.github/workflows/asv-benchmarking-pr.yml +++ b/.github/workflows/asv-benchmarking-pr.yml @@ -12,28 +12,47 @@ on: pull_request: types: [opened, reopened, synchronize, labeled] workflow_dispatch: + inputs: + base_ref: + description: >- + Branch, tag or commit to benchmark against. The comparison is against + its merge-base with the dispatched ref, so a topic branch should name + the branch it will merge into rather than main. + required: false + default: main + shards: + description: >- + How many runners to spread the suite over. Each shard pays the conda + restore, the wheel install and the ~57s the suite's numba warmups + cost at import, so past about four the added runners mostly pay that + floor again. + required: false + default: "4" env: PR_HEAD_LABEL: ${{ github.event.pull_request.head.label }} + ASV_DIR: "./benchmarks" + CONDA_ENV_FILE: ci/environment.yml + # One machine name for every shard. + ASV_MACHINE: gh-linux-x64 jobs: - benchmark: + setup: + name: Setup if: ${{ contains(github.event.pull_request.labels.*.name, 'run-benchmark') && github.event_name == 'pull_request' || github.event_name == 'workflow_dispatch' }} - name: Linux runs-on: ubuntu-latest - env: - ASV_DIR: "./benchmarks" - CONDA_ENV_FILE: ci/environment.yml - + outputs: + base: ${{ steps.base.outputs.sha }} + shards: ${{ steps.plan.outputs.shards }} + count: ${{ steps.plan.outputs.count }} steps: - uses: actions/checkout@v7 with: fetch-depth: 0 - - name: Record CPU topology + # A diagnostic; never fail the job over it. run: | - # A diagnostic; never fail the job over it. lscpu | grep -E 'Model name|^CPU\(s\):|Thread\(s\) per core|Core\(s\) per socket|Socket\(s\)|CPU max MHz' || lscpu || true - name: Set up Conda environment @@ -42,29 +61,328 @@ jobs: environment-file: ${{env.CONDA_ENV_FILE}} cache-environment: true environment-name: uxarray_build - cache-environment-key: "${{runner.os}}-${{runner.arch}}-py${{env.PYTHON_VERSION}}-${{env.TODAY}}-${{hashFiles(env.CONDA_ENV_FILE)}}-benchmark" + cache-environment-key: "${{runner.os}}-${{runner.arch}}-${{hashFiles(env.CONDA_ENV_FILE)}}-benchmark" create-args: >- asv python-build mamba - - name: Run Benchmarks + - name: Resolve the baseline commit + id: base + # Only a pull_request event carries a pull_request payload, so on a + # manual run every one of these expressions is the empty string. + run: | + set -ex + BASE="${{ github.event.pull_request.base.sha }}" + if [ -z "$BASE" ]; then + # The merge-base rather than the base branch's tip + BASE_REF="${{ github.event.inputs.base_ref }}" + BASE_REF="${BASE_REF:-main}" + git rev-parse --verify -q "origin/$BASE_REF" >/dev/null \ + || git fetch -q origin "$BASE_REF:refs/remotes/origin/$BASE_REF" + BASE=$(git merge-base "$GITHUB_SHA" "origin/$BASE_REF") + fi + if [ "$BASE" = "$(git rev-parse "$GITHUB_SHA")" ]; then + # Dispatched from main itself: compare it against the commit before. + BASE=$(git rev-parse "$GITHUB_SHA^") + fi + # Fail loudly rather than leaving asv to infer a baseline of its own. + test -n "$BASE" || { echo "could not determine a baseline commit" >&2; exit 1; } + echo "sha=$BASE" >> "$GITHUB_OUTPUT" + + # asv builds its own conda environment under ``benchmarks/env` + - name: Cache asv's benchmark environment + uses: actions/cache@v6 + with: + path: ${{ env.ASV_DIR }}/env + key: asv-env-${{ runner.os }}-${{ runner.arch }}-${{ hashFiles('benchmarks/asv.conf.json') }} + restore-keys: | + asv-env-${{ runner.os }}-${{ runner.arch }}- + + - name: Cache the benchmark fixtures + uses: actions/cache@v6 + with: + path: | + ${{ env.ASV_DIR }}/oQU*.nc + ${{ env.ASV_DIR }}/_io_cache + key: asv-fixtures-${{ runner.os }}-${{ hashFiles('benchmarks/helpers/_fixtures.py') }} + restore-keys: | + asv-fixtures-${{ runner.os }}- + + # ``_partition`` weights each benchmark by the duration asv recorded for + # it, and falls back to the median of the ones it knows when it knows + # none -- which, with no results on a fresh runner, is every one of them, + # making the split by count. Restoring the last run's tree is what turns + # the packing back into a cost-balanced one. Restore-only here; the merge + # job saves the updated tree under a key of its own. + # The environment cache key has no commit in it -- deliberately, since the + # conda environment does not depend on one -- so on a hit ``actions/cache`` + # saves nothing at the end of the job and the wheels built above would go + # with it, leaving every shard to build both commits again. The wheels get + # their own cache, keyed on the pair, and it is a couple of MB against the + # environment's 568, so keying it per run costs the cache budget little. + - name: Cache the built wheels + uses: actions/cache@v6 + with: + path: ${{ env.ASV_DIR }}/env/*/asv-build-cache + key: asv-wheels-${{ runner.os }}-${{ steps.base.outputs.sha }}-${{ github.sha }} + + - name: Restore recorded durations + uses: actions/cache/restore@v6 + with: + path: ${{ env.ASV_DIR }}/results + key: asv-results-${{ runner.os }}-${{ github.run_id }} + restore-keys: | + asv-results-${{ runner.os }}- + + - name: Pre-build and discover shell: bash -l {0} - id: benchmark + working-directory: ${{ env.ASV_DIR }} + env: + BASE: ${{ steps.base.outputs.sha }} + # ``--bench just-discover`` is asv's own discovery-only mode + # (``commands/run.py``): it creates the environments, builds the project + # for the commit it discovers from, writes ``results/benchmarks.json`` + # and returns 0 without running a benchmark. Done once here rather than + # once per shard, a cold cache costs one conda solve instead of four, + # and the shards restore an environment whose build cache already holds + # both commits' wheels. On a shared filesystem this is also what stops + # concurrent shards racing to install into one ``env/``; separate + # runners have no such race, but they do have the cost. run: | - set -x + set -ex # Fill the fixture cache before asv preimports the suite, which would otherwise build it serially in the forkserver parent (cd .. && python -m benchmarks.helpers._fixtures) - # ID this runner asv machine --yes - echo "Baseline: ${{ github.event.pull_request.base.sha }} (${{ github.event.pull_request.base.label }})" - echo "Contender: ${GITHUB_SHA} ($PR_HEAD_LABEL)" - # Run benchmarks for current commit against base - ASV_OPTIONS="--split --show-stderr" - asv continuous $ASV_OPTIONS ${{ github.event.pull_request.base.sha }} ${GITHUB_SHA} - # Save compare results - asv compare --split ${{ github.event.pull_request.base.sha }} ${GITHUB_SHA} > asv_compare_results.txt + (cd .. && python -m benchmarks.helpers._machine --name "$ASV_MACHINE") + asv run --bench just-discover "${BASE}^!" + asv run --bench just-discover "${GITHUB_SHA}^!" + + - name: Plan the shards + id: plan + shell: bash -l {0} + working-directory: ${{ env.ASV_DIR }} + env: + SHARDS: ${{ github.event.inputs.shards || '4' }} + run: | + set -ex + # The report goes in the log so a lopsided split is visible without + # opening four shard jobs to find which one ran long. + PYTHONPATH=.. python -m benchmarks.helpers._partition --shards "$SHARDS" + echo "count=$SHARDS" >> "$GITHUB_OUTPUT" + python -c "import json, os; print('shards=' + json.dumps(list(range(int(os.environ['SHARDS'])))))" \ + >> "$GITHUB_OUTPUT" + + # The whole tree rather than ``benchmarks.json`` alone: the shards + # partition from this, and a partition is only reproducible if they weigh + # the same durations setup weighed. + - name: Upload the discovered suite + uses: actions/upload-artifact@v7 + with: + name: asv-plan + path: ${{ env.ASV_DIR }}/results + + benchmark: + name: Shard ${{ matrix.shard }} + needs: setup + runs-on: ubuntu-latest + strategy: + # Every shard is independent, and a shard that dies still leaves the rest + # worth merging. + fail-fast: false + matrix: + shard: ${{ fromJSON(needs.setup.outputs.shards) }} + steps: + - uses: actions/checkout@v7 + with: + fetch-depth: 0 + + - name: Set up Conda environment + uses: mamba-org/setup-micromamba@v3 + with: + environment-file: ${{env.CONDA_ENV_FILE}} + cache-environment: true + environment-name: uxarray_build + cache-environment-key: "${{runner.os}}-${{runner.arch}}-${{hashFiles(env.CONDA_ENV_FILE)}}-benchmark" + create-args: >- + asv + python-build + mamba + + - name: Cache asv's benchmark environment + uses: actions/cache@v6 + with: + path: ${{ env.ASV_DIR }}/env + key: asv-env-${{ runner.os }}-${{ runner.arch }}-${{ hashFiles('benchmarks/asv.conf.json') }} + restore-keys: | + asv-env-${{ runner.os }}-${{ runner.arch }}- + + - name: Cache the benchmark fixtures + uses: actions/cache@v6 + with: + path: | + ${{ env.ASV_DIR }}/oQU*.nc + ${{ env.ASV_DIR }}/_io_cache + key: asv-fixtures-${{ runner.os }}-${{ hashFiles('benchmarks/helpers/_fixtures.py') }} + restore-keys: | + asv-fixtures-${{ runner.os }}- + + # The environment cache key has no commit in it -- deliberately, since the + # conda environment does not depend on one -- so on a hit ``actions/cache`` + # saves nothing at the end of the job and the wheels built above would go + # with it, leaving every shard to build both commits again. The wheels get + # their own cache, keyed on the pair, and it is a couple of MB against the + # environment's 568, so keying it per run costs the cache budget little. + - name: Cache the built wheels + uses: actions/cache@v6 + with: + path: ${{ env.ASV_DIR }}/env/*/asv-build-cache + key: asv-wheels-${{ runner.os }}-${{ needs.setup.outputs.base }}-${{ github.sha }} + + - name: Download the discovered suite + uses: actions/download-artifact@v8 + with: + name: asv-plan + path: plan + + - name: Run shard + shell: bash -l {0} + working-directory: ${{ env.ASV_DIR }} + env: + BASE: ${{ needs.setup.outputs.base }} + SHARDS: ${{ needs.setup.outputs.count }} + SHARD: ${{ matrix.shard }} + # Partitioned from setup's ``benchmarks.json`` rather than from whatever + # this runner's cache happens to hold, so every shard splits the same + # suite the same way. Nothing checks that the shards tile it: were they + # to disagree, benchmarks would simply go unrun and no step would say so. + run: | + set -ex + (cd .. && python -m benchmarks.helpers._fixtures) + asv machine --yes + (cd .. && python -m benchmarks.helpers._machine --name "$ASV_MACHINE") + ASV_ARGS=$(PYTHONPATH=.. python -m benchmarks.helpers._partition \ + --shards "$SHARDS" --shard "$SHARD" --results ../plan \ + --config asv.conf.json --asv-args) + echo "Baseline: $BASE" + echo "Contender: ${GITHUB_SHA} (${PR_HEAD_LABEL:-$GITHUB_REF_NAME})" + # asv returns 2 when a benchmark failed rather than when the run + # broke (``commands/run.py``: ``if failures: return 2``), and this + # shard's other results are complete and worth merging. The old + # single-job workflow tolerated it by accident -- ``asv compare`` ran + # last in the same step and its status was the step's -- so make it + # deliberate here, loudly, rather than letting one long-broken + # benchmark redden every shard that happens to hold it. + status=0 + asv continuous --split --show-stderr -m "$ASV_MACHINE" $ASV_ARGS \ + "$BASE" "${GITHUB_SHA}" || status=$? + if [ "$status" -eq 2 ]; then + echo "::warning title=Benchmark failures in shard ${SHARD}::asv exited 2; see the failed entries above" + elif [ "$status" -ne 0 ]; then + exit "$status" + fi + + - name: Upload shard results + if: always() + uses: actions/upload-artifact@v7 + with: + name: asv-shard-${{ matrix.shard }} + path: ${{ env.ASV_DIR }}/results.shard${{ matrix.shard }} + if-no-files-found: warn + + merge: + name: Merge and compare + needs: [setup, benchmark] + # Runs on a partial fan-out too: a merged tree missing one shard's rows is + # still worth reading, and the artifact is the only place a failed shard's + # half of the run can be seen. + if: ${{ always() && needs.setup.result == 'success' }} + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v7 + with: + fetch-depth: 0 + + - name: Set up Conda environment + uses: mamba-org/setup-micromamba@v3 + with: + environment-file: ${{env.CONDA_ENV_FILE}} + cache-environment: true + environment-name: uxarray_build + cache-environment-key: "${{runner.os}}-${{runner.arch}}-${{hashFiles(env.CONDA_ENV_FILE)}}-benchmark" + create-args: >- + asv + python-build + mamba + + # Without merge-multiple each artifact lands in its own + # ``shards/asv-shard-N/``, which is what the merge needs: every shard + # named its results file identically, and telling them apart is the point. + - name: Download the shards + uses: actions/download-artifact@v8 + with: + pattern: asv-shard-* + path: shards + + - name: Merge and compare + shell: bash -l {0} working-directory: ${{ env.ASV_DIR }} + env: + BASE: ${{ needs.setup.outputs.base }} + run: | + set -ex + asv machine --yes + (cd .. && python -m benchmarks.helpers._machine --name "$ASV_MACHINE") + (cd .. && python -m benchmarks.helpers._merge --out benchmarks/results shards/asv-shard-*) + asv compare --split --machine "$ASV_MACHINE" "$BASE" "${GITHUB_SHA}" \ + > asv_compare_results.txt + cat asv_compare_results.txt + + # asv records a duration per benchmark, plus ```` and + # ```` entries, in the results file it writes. Printing + # them is what tells us where a run's wall clock actually went. Two things + # to read them with: a per-benchmark duration is the final round's only, + # since asv assigns rather than accumulates it, and ```` and + # ```` are the slowest shard's, since every shard paid them. + - name: Report where the time went + if: always() + shell: bash -l {0} + run: | + python - <<'PY' || true + import glob, json, os + + for path in sorted(glob.glob("benchmarks/results/*/*.json")): + if os.path.basename(path) in ("machine.json", "benchmarks.json"): + continue + data = json.load(open(path)) + columns = data.get("result_columns") or [] + if "duration" not in columns: + continue + index = columns.index("duration") + rows = sorted( + (float(row[index]), name) + for name, row in data["results"].items() + if len(row) > index and row[index] is not None + ) + total = sum(duration for duration, _ in rows) or 1.0 + params = data.get("params", {}) + print(f"\n=== {data['commit_hash'][:8]} on {params.get('cpu', '?')} " + f"({params.get('num_cpu', '?')} cpu) ===") + for key, value in sorted(data.get("durations", {}).items()): + print(f" {value:8.1f}s {key}") + print(f" {total:8.1f}s all {len(rows)} benchmarks ({total / 60:.1f} min)") + for duration, name in reversed(rows[-15:]): + print(f" {duration:8.1f}s {100 * duration / total:5.1f}% {name}") + PY + + # Keyed on the run id so every run stores its own entry and setup's + # prefixed restore picks up the newest; there is nothing to restore here. + - name: Save recorded durations for the next run + if: always() + uses: actions/cache/save@v6 + with: + path: ${{ env.ASV_DIR }}/results + key: asv-results-${{ runner.os }}-${{ github.run_id }} - name: Save PR number if: always() @@ -73,7 +391,7 @@ jobs: - uses: actions/upload-artifact@v7 if: always() with: - name: asv-benchmark-results-${{ runner.os }} + name: asv-benchmark-results-Linux path: | ${{ env.ASV_DIR }}/results ${{ env.ASV_DIR }}/asv_compare_results.txt diff --git a/.github/workflows/asv-benchmarking.yml b/.github/workflows/asv-benchmarking.yml index f5f751cf0..bc9a8a3a0 100644 --- a/.github/workflows/asv-benchmarking.yml +++ b/.github/workflows/asv-benchmarking.yml @@ -9,6 +9,9 @@ on: jobs: benchmark: runs-on: ubuntu-latest + # Without this the platform's 6h cap is the only limit, which is how a + # mis-skipped range quietly turns into a 453-commit build. + timeout-minutes: 180 defaults: run: shell: bash -el {0} @@ -73,15 +76,43 @@ jobs: cp -r uxarray-asv/results benchmarks/ fi + - name: Cache asv's benchmark environment + uses: actions/cache@v6 + with: + path: ${{ env.ASV_DIR }}/env + key: asv-env-${{ runner.os }}-${{ runner.arch }}-${{ hashFiles('benchmarks/asv.conf.json') }} + restore-keys: | + asv-env-${{ runner.os }}-${{ runner.arch }}- + + - name: Cache the benchmark fixtures + uses: actions/cache@v6 + with: + path: | + ${{ env.ASV_DIR }}/oQU*.nc + ${{ env.ASV_DIR }}/_io_cache + key: asv-fixtures-${{ runner.os }}-${{ hashFiles('benchmarks/helpers/_fixtures.py') }} + restore-keys: | + asv-fixtures-${{ runner.os }}- + - name: Run benchmarks shell: bash -l {0} id: benchmark run: | - # Fill the fixture cache before asv preimports the suite, which would otherwise build it serially in the forkserver parent - python -m benchmarks.helpers._fixtures + # Fill the fixture cache before asv preimports the suite, which would + # otherwise build it serially in the forkserver parent. Guarded because + # this workflow checks out main whatever ref it was dispatched from + if [ -f benchmarks/helpers/_fixtures.py ]; then + python -m benchmarks.helpers._fixtures + else + echo "no benchmarks/helpers/_fixtures.py on this ref; nothing to prime" + fi cd benchmarks asv machine --machine GH-Actions --os ubuntu-latest --arch x64 --cpu "2-core unknown" --ram 7GB - asv run v2024.02.0..main --skip-existing --parallel || true + # ``--skip-existing-commits``, not ``--skip-existing``: the latter keys + # its skip set by (commit, env), so asv's per-commit check never + # matches and it builds and installs every commit in the range before + # skipping the benchmarks it already has. + asv run main^! --skip-existing-commits --parallel || true - name: Commit and push benchmark results run: | diff --git a/.gitignore b/.gitignore index 5c52cd586..724dfa80b 100644 --- a/.gitignore +++ b/.gitignore @@ -164,3 +164,6 @@ benchmarks/env benchmarks/results benchmarks/html benchmarks/_io_cache +# Generated per shard by benchmarks/helpers/_partition.py --asv-args +benchmarks/asv.conf*.shard*.json +benchmarks/results.shard* diff --git a/benchmarks/asv.conf.hpc.json b/benchmarks/asv.conf.hpc.json new file mode 100644 index 000000000..62f3d9bcb --- /dev/null +++ b/benchmarks/asv.conf.hpc.json @@ -0,0 +1,83 @@ +{ + // Thread-scaling variant of ``asv.conf.json``, for a machine with cores to + // spare. Run it explicitly: + // + // asv run --config benchmarks/asv.conf.hpc.json + // asv compare --config benchmarks/asv.conf.hpc.json -E conda:3.11 A B + // + // Why a second file rather than an environment variable: asv records the + // ``env_nobuild`` matrix into every results file it writes and folds it + // into the environment name, so each thread count gets its own result set + // and ``asv compare`` keeps them apart. A count exported in the shell is + // invisible to asv -- runs at 1 and 64 threads land in the *same* file for + // a commit and compare as though they measured the same thing. See + // ``benchmarks/helpers/_threads.py`` for the shell route, which is the + // right one for "just run this faster" and the wrong one for a study. + // + // The two mechanisms are mutually exclusive: asv layers ``env_nobuild`` + // over the inherited environment, so ``NUMBA_NUM_THREADS`` below wins over + // any export. + // + // KEEP IN SYNC with asv.conf.json: ``pythons``, ``environment_type``, + // ``conda_channels``, ``matrix.req`` and ``build_command`` decide what gets + // installed. Let them drift and these numbers stop being comparable to the + // ones the normal config produces, silently. + + "version": 1, + "project": "uxarray", + "project_url": "https://github.com/UXARRAY/uxarray", + "repo": "..", + "branches": ["main"], + "dvcs": "git", + "environment_type": "conda", + "conda_channels": ["conda-forge"], + "install_timeout": 600, + "launch_method": "forkserver", + "show_commit_url": "https://github.com/UXARRAY/uxarray/commit/", + "pythons": ["3.11"], + "benchmark_dir": ".", + "build_command": [ + "python -mpip wheel --no-deps --no-build-isolation --no-index -w {build_cache_dir} {build_dir}" + ], + "build_cache_size": 4, + + // A parallel kernel held at one thread takes roughly as long as the core + // count it would otherwise have used, so the 360s the normal config allows + // is not enough at the bottom of a scaling sweep. + "default_benchmark_timeout": 1800, + + "matrix": { + "req": { + "setuptools_scm": [""], + "xarray": [""], + "netcdf4": [""], + "pip+pyfma": [""], + "tbb": [""] + }, + + // One result set per value, so the run costs its length: four values is + // four passes over the suite. Edit for the node -- powers of two up to + // its physical core count is the usual shape, and + // ``python -m benchmarks.helpers._threads`` reports that count. + // Values above the core count are allowed and simply oversubscribe. + // One value each: a matrix entry with several values is a separate + // environment per value, and asv runs the whole suite in every one of + // them. Only 24 of the 86 benchmarks reach a ``parallel=True`` kernel, + // so a four-value sweep ran the other 62 four times over to print the + // same number four times. ``NUMBA_NUM_THREADS`` is the one knob here; + // to sweep it, see ``--env`` in ``benchmarks/helpers/_partition.py``. + // + // The BLAS variables are pinned because they are what made the old + // sweep unreadable: numpy would take the whole node in every + // environment, so the ``NUMBA_NUM_THREADS=1`` column was never a + // single-threaded baseline and the scaling curve it implied was + // measuring contention. Pinned, numba's threads are the only ones. + "env_nobuild": { + "NUMBA_THREADING_LAYER": ["forksafe"], + "NUMBA_NUM_THREADS": ["8"], + "OMP_NUM_THREADS": ["1"], + "MKL_NUM_THREADS": ["1"], + "OPENBLAS_NUM_THREADS": ["1"] + } + } +} diff --git a/benchmarks/asv.conf.json b/benchmarks/asv.conf.json index 9b3852f53..57114b6ad 100644 --- a/benchmarks/asv.conf.json +++ b/benchmarks/asv.conf.json @@ -122,12 +122,27 @@ // Belt to that brace: if TBB is ever unavailable, pick the other // fork-safe layer rather than quietly falling back to the one that // breaks. ``forksafe`` raises if no fork-safe layer exists at all. - "env_nobuild": {"NUMBA_THREADING_LAYER": ["forksafe"]} + // + // The BLAS variables are pinned so numpy cannot take the runner's four + // vCPUs out from under whatever is being measured. Unpinned it competes + // with numba for the same cores, which shows up as run-to-run spread in + // every benchmark that touches an array, not just the threaded ones. + // ``NUMBA_NUM_THREADS`` is deliberately left alone: numba's default is + // the whole machine, and with BLAS out of the way that is now an + // unambiguous four rather than four contended with numpy's. + "env_nobuild": { + "NUMBA_THREADING_LAYER": ["forksafe"], + "OMP_NUM_THREADS": ["1"], + "MKL_NUM_THREADS": ["1"], + "OPENBLAS_NUM_THREADS": ["1"] + } }, + // Just the one command: ``python -m build`` put an sdist and a wheel in + // {build_dir}/dist, which asv never reads, and then this rebuilt the wheel + // into the directory asv actually installs from. "build_command": [ - "python -m build", "python -mpip wheel --no-deps --no-build-isolation --no-index -w {build_cache_dir} {build_dir}" ], diff --git a/benchmarks/helpers/_fixtures.py b/benchmarks/helpers/_fixtures.py index 9ab574197..df3e875ab 100644 --- a/benchmarks/helpers/_fixtures.py +++ b/benchmarks/helpers/_fixtures.py @@ -14,11 +14,13 @@ everything the reader produced from ``Grid.open_grid`` and ``Grid.open_dataset`` -Artifacts are keyed on both the uxarray build and the files, because an -artifact is one version's reader output and ASV diffs commits. Likewise, there's a -fresh read per commit. ``prime`` covers every source that is readable here, -and there is a CLI (``python -m benchmarks.helpers._fixtures``) to fill the -cache from a batch script instead of from inside a benchmark. +Artifacts are keyed on the source files and nothing else, so they persist +across runs -- and across the commits ASV diffs, which means a benchmark on the +cached flavors measures its own subject against one fixed reader output rather +than against a per-commit re-read. A source replaced in place misses rather than +being served something stale. ``prime`` covers every source that is readable +here, and there is a CLI (``python -m benchmarks.helpers._fixtures``) to fill +the cache from a batch script instead of from inside a benchmark. """ import hashlib diff --git a/benchmarks/helpers/_machine.py b/benchmarks/helpers/_machine.py new file mode 100644 index 000000000..439ae0205 --- /dev/null +++ b/benchmarks/helpers/_machine.py @@ -0,0 +1,166 @@ +"""Pinning the machine name asv records results under. + +asv keys results on a machine name and defaults it to the hostname +(``Machine.get_defaults``), which on a hosted runner is fresh for every job -- +``runnervmgx7h7`` on one run, something else on the next. A name that never +repeats cannot be compared across runs, and once the suite is sharded it cannot +even be merged within one run: the file asv writes is +``results//-.json``, so every shard has to agree on +```` or there is nothing for :mod:`_merge` to line up. + +``asv machine --machine NAME`` will not do it on its own. That command stores +only the fields that differ from the ones it detected and then skips filling the +rest in (``commands/machine.py``), so naming the machine is precisely what drops +``cpu``, ``num_cpu`` and ``ram`` -- the fields that say what the timings were +measured on, and the ones the duration report prints. Detect first with ``asv +machine --yes``, rename after, which is what this does. + +Idempotent, so a job that runs it twice, or a machine file that arrives already +pinned, is fine. + +Usage:: + + asv machine --yes + python -m benchmarks.helpers._machine --name gh-Linux-X64 +""" + +import argparse +import json +import os +import platform +import re +import sys +from pathlib import Path + +__all__ = ["pin"] + +_VERSION_KEY = "version" + + +def default_name(): + """A machine name that survives landing on a different node next time. + + The scheduler puts you on ``derecho3`` one day and ``crhtc70`` the next, and + asv keys results on ``platform.uname``'s node name, so left alone it records + a new machine every login and the results scatter across all of them. + + ``NCAR_HOST`` is the reliable answer where it is set -- it names the cluster + rather than the node, which is the granularity results want. Failing that, + the node name with its trailing digits removed, which folds ``derecho3`` and + ``derecho5`` together but *not* ``derecho3`` and ``crhtc70``: login and + compute nodes of one cluster do not share a stem. Set ``ASV_MACHINE`` + yourself if you move between them without ``NCAR_HOST``. + """ + for variable in ("ASV_MACHINE", "NCAR_HOST"): + value = os.environ.get(variable) + if value: + return value, variable + node = platform.node().split(".")[0] + return re.sub(r"[-_]?\d+$", "", node) or node, None + + +def default_path(): + """Where asv keeps its machine file (``MachineCollection.get_machine_file_path``).""" + return Path.home() / ".asv-machine.json" + + +def pin(name, path=None, hostname=None, sole=False): + """Renames the machine file's freshly detected entry to ``name``. + + Returns its details. Entries for other machines are left alone -- a runner + has only the one, but a login node that has recorded every compute node it + ever landed on should not lose them to a benchmark run. + + The fresh entry is the one keyed by this host's name, since that is what + ``asv machine --yes`` writes (``Machine.get_defaults`` takes it from + ``platform.uname``). Renaming it is the whole point: several nodes of one + cluster should file their results under one machine, or a sharded run has + nothing to merge. Falls back to a lone entry whatever its name, for a runner + whose hostname has already been renamed away by an earlier call. + """ + path = Path(path) if path is not None else default_path() + hostname = hostname if hostname is not None else platform.node() + stored = json.loads(path.read_text()) + version = stored.pop(_VERSION_KEY, None) + + if hostname in stored: + detected = stored.pop(hostname) + elif name in stored: + detected = stored[name] + elif len(stored) == 1: + (only,) = stored + detected = stored.pop(only) + else: + raise ValueError( + f"{path} holds {len(stored)} machines ({', '.join(sorted(stored))}), none of " + f"them this host ({hostname!r}) and none of them {name!r}; cannot tell which " + f"describes the machine this is running on. Run ``asv machine --yes`` first, " + f"or pass --name one of the recorded machines" + ) + + detected["machine"] = name + if sole: + stored = {} + stored[name] = detected + if version is not None: + stored[_VERSION_KEY] = version + path.write_text(json.dumps(stored, indent=4)) + return detected + + +def main(argv=None): + parser = argparse.ArgumentParser( + prog="python -m benchmarks.helpers._machine", + description="Rename asv's detected machine entry to a fixed name.", + ) + parser.add_argument( + "--name", + default=None, + help="Machine name to pin to. Defaults to $ASV_MACHINE, then $NCAR_HOST, " + "then this host's name with trailing digits removed.", + ) + parser.add_argument("--path", default=None, help="Machine file (default ~/.asv-machine.json).") + parser.add_argument( + "--hostname", default=None, help="Host whose entry to rename (default this one)." + ) + parser.add_argument( + "--sole", + action="store_true", + help="Drop every other machine from the file. asv falls back to a lone entry " + "whatever the hostname, so this makes bare ``asv run``/``asv show`` work from " + "any node without -m. Use it where you only ever benchmark one machine.", + ) + parser.add_argument( + "--quiet", action="store_true", help="Print only the pinned name, for capturing." + ) + parser.add_argument( + "--print", + dest="print_only", + action="store_true", + help="Print the name that would be pinned and change nothing.", + ) + args = parser.parse_args(argv) + + name, source = (args.name, "--name") if args.name else default_name() + if args.print_only: + print(name) + return 0 + detected = pin(name, args.path, args.hostname, sole=args.sole) + if args.quiet: + print(name) + return 0 + print( + f"{name}: {detected.get('cpu', '?')} " + f"({detected.get('num_cpu', '?')} cpu, {detected.get('os', '?')})" + ) + if source is None: + print( + f" note: {name!r} came from this host's name. A cluster's login and compute " + f"nodes do not share a stem, so set ASV_MACHINE (or rely on NCAR_HOST) if you " + f"benchmark from both, or the results will still split in two.", + ) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/benchmarks/helpers/_merge.py b/benchmarks/helpers/_merge.py new file mode 100644 index 000000000..638751d8e --- /dev/null +++ b/benchmarks/helpers/_merge.py @@ -0,0 +1,252 @@ +"""Merging a sharded run's results back into one tree. + +asv reads its results file once before running a benchmark set and writes it +once after (``Results.load_data`` then ``Results.save`` in ``commands/run.py``), +and the name it writes is ``results//-.json`` -- one file +per commit and environment, whatever subset of benchmarks the run measured. So +shards sharing a results directory each rewrite that whole file from what they +alone measured, and the last one to finish wins. Each shard therefore gets its +own directory (``_partition --config-out``) and they are combined here. + +The combination is a union rather than an element-wise reconciliation, which is +what the by-whole-benchmark split buys: a row is keyed on the benchmark name and +carries its whole parameter sweep inside, so every row is owned by exactly one +shard. Splitting inside a benchmark would have put two shards in one row. + +Order is restored rather than preserved. ``results`` is a JSON object, asv writes +it with ``compact=True`` -- which disables sorting, so key order is the order asv +appended to it -- and for an unsharded run that order is ``sorted(benchmarks)`` +grouped by ``setup_cache_key`` (``runner.py``, ``iter_run_items``). Shards finish +in whatever order the queue hands back, so :func:`canonical_order` recovers the +order the same suite would have produced serially and every merged file is +written in it. + +Idempotent, and indifferent to shards that have not landed: merging the three +directories that exist gives a valid tree, and merging again when the fourth +arrives puts its rows in their proper place. That is what makes it safe to run +from a polling loop as jobs come back rather than only after a barrier. + +Usage:: + + python -m benchmarks.helpers._merge --out results results.shard* + python -m benchmarks.helpers._merge --out results --quiet results.shard* +""" + +import argparse +import json +import shutil +import sys +from pathlib import Path + +__all__ = ["canonical_order", "merge", "merge_benchmarks", "merge_result_files"] + +BENCHMARK_DIR = Path(__file__).resolve().parents[1] + +MACHINE_FILE = "machine.json" +BENCHMARKS_FILE = "benchmarks.json" +_SPECIAL_FILES = frozenset({MACHINE_FILE, BENCHMARKS_FILE}) + +# asv stores its own format version alongside the data in both files. +_VERSION_KEY = "version" + + +def _load(path): + with open(path) as handle: + return json.load(handle) + + +def _dump(path, data): + """Writes ``data`` the way asv writes a results file. + + ``util.write_json(..., compact=True)`` disables both sorting and + indentation; the sorting is the part that matters, because key order is the + only place a results file records what ran when. + """ + path.parent.mkdir(parents=True, exist_ok=True) + with open(path, "w", encoding="utf-8") as handle: + json.dump(data, handle) + + +def canonical_order(benchmarks): + """Benchmark names in the order an unsharded ``asv run`` would produce them. + + Mirrors ``runner.run_benchmarks``: it walks ``sorted(benchmarks.items())`` + building ``benchmark_order``, a dict keyed on ``setup_cache_key``, then runs + each of those groups in turn. So the order is by name within a group, and + groups in the order their first member is reached by name. + """ + groups = {} + for name in sorted(benchmarks): + key = benchmarks[name].get("setup_cache_key") + groups.setdefault(key, []).append(name) + return [name for group in groups.values() for name in group] + + +def merge_benchmarks(shard_dirs): + """Union of the shards' ``benchmarks.json``. + + A shard discovers under its own ``--bench`` patterns, so each file holds + only that shard's benchmarks and the full set exists nowhere until here. + ``_partition.load_benchmarks`` needs that full set to plan the next run. + """ + merged, version = {}, None + for shard_dir in shard_dirs: + path = Path(shard_dir) / BENCHMARKS_FILE + if not path.is_file(): + continue + data = _load(path) + version = data.get(_VERSION_KEY, version) + for name, value in data.items(): + if name != _VERSION_KEY: + merged[name] = value + if version is not None: + merged[_VERSION_KEY] = version + return merged + + +def _pick(name, existing, candidate, report): + """Which of two rows for one benchmark to keep. + + Only reachable when a name landed in more than one shard, which the + partition does not do -- so it means the plan the shards ran was not the one + that produced them. Preferring a row that has a result over one that does + not, then the later ``started_at``, keeps a re-run over the run it replaced + instead of picking on file order. + """ + if existing == candidate: + return existing + + def rank(row): + return (row.get("result") is not None, row.get("started_at") or 0) + + keep, drop = (candidate, existing) if rank(candidate) > rank(existing) else (existing, candidate) + report( + f"{name}: found in more than one shard with different data; keeping the " + f"row started at {keep.get('started_at')} over {drop.get('started_at')}" + ) + return keep + + +def merge_result_files(datas, order, report): + """One results file from several shards' versions of it. + + ``datas`` are the parsed files, in shard order; ``order`` is the name order + to write. Every field outside ``results`` and ``durations`` describes the + commit and environment rather than the run, and is identical across shards + by construction, so the first shard's copy carries over untouched. + """ + merged = dict(datas[0]) + columns = list(merged.get("result_columns") or []) + + rows, durations = {}, {} + for data in datas: + # Read each row against its own file's columns. Identical in practice -- + # one asv builds every shard -- but a row is a bare list, so aligning it + # to the wrong header would silently shift every value. + shard_columns = data.get("result_columns") or columns + for name, row in (data.get("results") or {}).items(): + values = dict(zip(shard_columns, row)) + rows[name] = ( + _pick(name, rows[name], values, report) if name in rows else values + ) + # ``durations`` holds only the ```` and ```` + # entries; a benchmark's own duration lives in its row. Every shard pays + # both, so the max is the one a single run would have reported, and the + # sum would describe work no single wall clock ever saw. + for key, value in (data.get("durations") or {}).items(): + durations[key] = max(durations.get(key, 0.0), float(value)) + + known = [name for name in order if name in rows] + extra = sorted(name for name in rows if name not in set(order)) + if extra: + report(f"{len(extra)} row(s) not in benchmarks.json, appended: {', '.join(extra[:3])}...") + + results = {} + for name in known + extra: + row = [rows[name].get(column) for column in columns] + # asv drops trailing nulls when it writes a row; keeping that keeps the + # merged file the same size as the one a serial run would have written. + while row and row[-1] is None: + row.pop() + results[name] = row + + merged["results"] = results + merged["durations"] = durations + return merged + + +def merge(shard_dirs, out_dir, report=lambda message: None): + """Merges ``shard_dirs`` into ``out_dir``. Returns a per-file row count.""" + shard_dirs = [Path(d) for d in shard_dirs] + out_dir = Path(out_dir) + resolved_out = out_dir.resolve() + if any(d.resolve() == resolved_out for d in shard_dirs): + raise ValueError(f"--out {out_dir} is also a shard directory; refusing to merge in place") + + present = [d for d in shard_dirs if d.is_dir()] + for missing in [d for d in shard_dirs if not d.is_dir()]: + report(f"{missing}: not there yet, skipped") + if not present: + raise ValueError("no shard directories to merge") + + benchmarks = merge_benchmarks(present) + order = canonical_order({k: v for k, v in benchmarks.items() if k != _VERSION_KEY}) + out_dir.mkdir(parents=True, exist_ok=True) + if len(benchmarks) > (1 if _VERSION_KEY in benchmarks else 0): + _dump(out_dir / BENCHMARKS_FILE, benchmarks) + + # One group per (machine, results file): a shard writes the same file name as + # every other shard of its commit and environment, which is the collision + # this module exists to undo. + groups = {} + for shard_dir in present: + for path in sorted(shard_dir.glob("*/*.json")): + if path.name in _SPECIAL_FILES: + continue + groups.setdefault((path.parent.name, path.name), []).append(path) + for machine_path in sorted(shard_dir.glob(f"*/{MACHINE_FILE}")): + target = out_dir / machine_path.parent.name / MACHINE_FILE + if target.is_file() and _load(target) != _load(machine_path): + report(f"{machine_path}: disagrees with the machine.json already merged") + else: + target.parent.mkdir(parents=True, exist_ok=True) + shutil.copyfile(machine_path, target) + + counts = {} + for (machine, filename), paths in sorted(groups.items()): + merged = merge_result_files( + [_load(p) for p in paths], + order, + lambda message, f=filename: report(f"{f}: {message}"), + ) + _dump(out_dir / machine / filename, merged) + counts[f"{machine}/{filename}"] = (len(merged["results"]), len(paths)) + return counts + + +def main(argv=None): + parser = argparse.ArgumentParser( + prog="python -m benchmarks.helpers._merge", + description="Merge a sharded run's results directories into one tree.", + ) + parser.add_argument("shards", nargs="+", help="Shard results directories to merge.") + parser.add_argument( + "--out", + default=str(BENCHMARK_DIR / "results"), + help="Directory to write the merged tree to (default benchmarks/results).", + ) + parser.add_argument("--quiet", action="store_true", help="Suppress per-file notes.") + args = parser.parse_args(argv) + + def report(message): + if not args.quiet: + print(f" {message}", file=sys.stderr) + + counts = merge(args.shards, args.out, report) + for name, (rows, shards) in sorted(counts.items()): + print(f"{name}: {rows} benchmarks from {shards} shard(s)") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/benchmarks/helpers/_partition.py b/benchmarks/helpers/_partition.py new file mode 100644 index 000000000..f24e75d3e --- /dev/null +++ b/benchmarks/helpers/_partition.py @@ -0,0 +1,367 @@ +"""Splitting the suite into shards of roughly equal cost. + +asv runs one benchmark at a time. ``--parallel`` builds environments in +parallel and nothing else -- its own help text is "Build (but don't benchmark) +in parallel" -- so cutting the wall clock means running several ``asv`` +processes, and a ``time_*`` result is only worth having if nothing else is +competing for the machine while it is measured. That points at one shard per +runner rather than several per runner, and at this module, whose whole job is to +decide which benchmarks each of those runners should claim. + +The split is by whole class, not by benchmark and not by parameter +combination. asv matches ``--bench`` against the expanded +``name(param0, param1)``, so a much finer cut is available, but a class's +benchmarks share the kernels its first one compiles and splitting them makes +every shard pay that compile again. Measured: ``face_bounds.FaceBounds``'s four +benchmarks cost ~59s together in one process, where ``time_face_bounds`` paid +the bounds compile and the three ``track_*`` variants rode on it warm at under +7s each; scattered one per shard across four runners they cost 221s, every one +of them paying the compile alone. ``cache=True`` does not save this -- asv +reinstalls the wheel for each commit and the on-disk cache goes with it. The +suite stays flat enough at class granularity for greedy longest-first packing +to land close to a perfect split. + +Going finer than a whole benchmark would also put two shards in one row of one +results file, with no sane way to reconcile them. Whole benchmarks leave every +row owned by exactly one shard, which is what lets +:mod:`benchmarks.helpers._merge` rebuild the tree by union. + +Shards still need somewhere separate to write, because asv names its results +file per commit and environment rather than per run and rewrites the whole thing +at the end of a set. ``--config-out`` emits a copy of the config with +``results_dir`` pointed at this shard's own directory, which is what the merge +then reads. + +Weights come from the ``duration`` asv records per benchmark in the results file +it writes (``Results.save``), so a partition improves as results accumulate +rather than needing a cost model. Benchmarks with no recorded duration -- new +ones, mostly -- get the median of the ones that have, which is a better guess +than either zero or the mean of a long-tailed distribution. + +Usage:: + + python -m benchmarks.helpers._partition --shards 4 + asv run $(python -m benchmarks.helpers._partition --shards 4 --shard 0 \ + --config asv.conf.hpc.json --asv-args) + + # A thread sweep, for whoever wants one. ``--shards 1`` is the whole suite; + # add ``--bench`` to hold it to the benchmarks that can actually respond. + for n in 1 2 4 8; do + asv run $(python -m benchmarks.helpers._partition --shards 1 --shard 0 \ + --config asv.conf.hpc.json --env NUMBA_NUM_THREADS=$n --asv-args) + done +""" + +import argparse +import json +import os +import re +import statistics +import sys +from pathlib import Path + +__all__ = [ + "bench_regexes", + "load_benchmarks", + "load_weights", + "plan", + "shard_results_dir", + "write_shard_config", +] + +BENCHMARK_DIR = Path(__file__).resolve().parents[1] + +# ``setup_cache`` groups whose members may be split across shards. Anything else +# is paid once per shard that holds any of its benchmarks, so those move +# together. ``CachedFixtures.setup_cache`` is just ``prime()``, a stat per file +# once the cache is warm (1.2s in CI), and ``None`` is no setup_cache at all. +SPLITTABLE_PREFIX = "helpers._fixtures:" + +_SKIP_FILES = frozenset({"machine.json", "benchmarks.json"}) + + +def _splittable(setup_cache_key): + """Whether benchmarks sharing this ``setup_cache_key`` may land in different shards.""" + return setup_cache_key is None or str(setup_cache_key).startswith(SPLITTABLE_PREFIX) + + +def _owner(name): + """The class -- or the module, for a bare function -- a benchmark belongs to.""" + return name.rsplit(".", 1)[0] + + +def _units(benchmarks): + """Benchmarks that have to ride together, as ``{root name: [names]}``. + + Two constraints, unioned so a ``setup_cache`` group spanning classes pulls + those classes together rather than contradicting them: + + ``class`` + Its benchmarks share whatever its first one compiles (see the module + docstring for what splitting one measured). + ``setup_cache`` + Anything sharing a ``setup_cache`` expensive enough to matter, which asv + would otherwise run once in every shard holding a member. + + Roots are the lexicographically smallest member, so the grouping is + deterministic and a shard can still work out its own membership. + """ + parent = {} + + def find(x): + parent.setdefault(x, x) + while parent[x] != x: + parent[x] = parent[parent[x]] + x = parent[x] + return x + + def union(a, b): + ra, rb = find(a), find(b) + if ra != rb: + parent[max(ra, rb)] = min(ra, rb) + + groups = {} + for name, benchmark in benchmarks.items(): + find(name) + groups.setdefault(("class", _owner(name)), []).append(name) + key = benchmark.get("setup_cache_key") + if not _splittable(key): + groups.setdefault(("setup_cache", key), []).append(name) + for members in groups.values(): + for other in members[1:]: + union(members[0], other) + + units = {} + for name in benchmarks: + units.setdefault(find(name), []).append(name) + return units + + +def load_benchmarks(results_dir): + """The discovered benchmarks, as asv wrote them to ``benchmarks.json``.""" + path = Path(results_dir) / "benchmarks.json" + with open(path) as handle: + discovered = json.load(handle) + # asv stores its own format version alongside the benchmarks. + return {name: value for name, value in discovered.items() if name != "version"} + + +def load_weights(results_dirs): + """Mean recorded duration per benchmark, in seconds, over the files on disk. + + The mean rather than the latest: a benchmark's first run on a cold numba + cache can cost hundreds of times its warm cost (a 9.6s bounds compile + against 13ms warm, in one observed run), and a partition built from one + such outlier sends a whole shard chasing work that is not there. + """ + samples = {} + for results_dir in results_dirs: + root = Path(results_dir) + if not root.is_dir(): + continue + for path in sorted(root.glob("*/*.json")): + if path.name in _SKIP_FILES: + continue + try: + with open(path) as handle: + data = json.load(handle) + except (OSError, ValueError): + continue + columns = data.get("result_columns") or [] + if "duration" not in columns: + continue + index = columns.index("duration") + for name, row in (data.get("results") or {}).items(): + if len(row) <= index or row[index] is None: + continue + samples.setdefault(name, []).append(float(row[index])) + return {name: statistics.fmean(values) for name, values in samples.items()} + + +def plan(benchmarks, n_shards, weights=None): + """Partitions ``benchmarks`` into ``n_shards`` lists of names. + + Greedy longest-first onto the lightest shard so far -- the standard LPT + heuristic, which on a distribution this flat is within about 1% of optimal + and, unlike anything smarter, is obvious enough to debug from the report. + + Deterministic: equal weights are broken by name, so the same inputs always + give the same shards and a shard can compute its own membership without + being told. + """ + if n_shards < 1: + raise ValueError(f"n_shards must be at least 1, got {n_shards}") + weights = dict(weights or {}) + known = [value for value in weights.values() if value > 0] + default = statistics.median(known) if known else 1.0 + + units = _units(benchmarks) + + costs = { + unit: sum(weights.get(name, default) for name in names) + for unit, names in units.items() + } + + shards = [[] for _ in range(n_shards)] + loads = [0.0] * n_shards + for unit in sorted(units, key=lambda u: (-costs[u], u)): + target = min(range(n_shards), key=lambda i: (loads[i], i)) + shards[target].extend(sorted(units[unit])) + loads[target] += costs[unit] + return shards + + +def bench_regexes(names): + """``--bench`` patterns selecting exactly ``names`` and nothing else. + + asv filters a parameterized benchmark on ``name(param0, param1)`` and an + unparameterized one on ``name`` (``Benchmarks.__init__``), so the trailing + group has to admit both an open parenthesis and end-of-string. Without it + ``^name$`` silently matches none of a parameterized benchmark's + combinations, and the shard runs nothing. + """ + return [f"^{re.escape(name)}($|\\()" for name in names] + + +def shard_config_path(base_config, shard): + """Where shard ``shard``'s generated config goes, beside ``base_config``.""" + base = Path(base_config) + return base.with_name(f"{base.stem}.shard{shard}{base.suffix}") + + +def shard_results_dir(results_dir, shard): + """Where shard ``shard`` writes, given the run's ordinary ``results_dir``.""" + return f"{results_dir}.shard{shard}" + + +def write_shard_config(base_config, out_path, shard, env=None): + """Writes a copy of ``base_config`` that writes results where ``shard`` should. + + ``env`` overrides ``env_nobuild`` variables. An override lands in the + environment's name, so asv files those results under a name of their own and + a sweep's runs do not overwrite one another. + + Returns the shard's results directory. + """ + # asv's loader, because an asv config is JSON with javascript comments and + # ``json`` cannot read one. Imported here so the rest of the module stays + # runnable without asv installed. + from asv import util + + config = util.load_json(str(base_config), js_comments=True) + results_dir = shard_results_dir(config.get("results_dir", "results"), shard) + config["results_dir"] = results_dir + if env: + matrix = config.setdefault("matrix", {}).setdefault("env_nobuild", {}) + # One value per variable: a list of several is a separate environment + # per value, and asv would run the whole suite in each of them. + matrix.update({key: [value] for key, value in env.items()}) + with open(out_path, "w") as handle: + json.dump(config, handle, indent=4) + return results_dir + + +def _report(benchmarks, shards, weights): + known = [value for value in weights.values() if value > 0] + default = statistics.median(known) if known else 1.0 + total = sum(weights.get(name, default) for name in benchmarks) + print( + f"{len(benchmarks)} benchmarks, {len(weights)} with recorded durations, " + f"{total / 60:.1f} min of work; median fallback {default:.1f}s" + ) + loads = [sum(weights.get(name, default) for name in shard) for shard in shards] + ideal = total / len(shards) if shards else 0.0 + for index, (shard, load) in enumerate(zip(shards, loads)): + drift = 100 * (load - ideal) / ideal if ideal else 0.0 + print(f" shard {index}: {len(shard):3} benchmarks {load / 60:5.1f} min {drift:+5.1f}%") + if loads and ideal: + print( + f" slowest shard {max(loads) / 60:.1f} min against an ideal " + f"{ideal / 60:.1f}; speedup {total / max(loads):.2f}x of a possible {len(shards)}x" + ) + + +def main(argv=None): + parser = argparse.ArgumentParser( + prog="python -m benchmarks.helpers._partition", + description="Split the benchmark suite into shards of roughly equal cost.", + ) + parser.add_argument("--shards", type=int, default=4, help="Number of shards (default 4).") + parser.add_argument( + "--shard", type=int, default=None, help="Report only this shard, by index." + ) + parser.add_argument( + "--results", + action="append", + default=None, + help="Results directory to read durations and benchmarks.json from. " + "Repeatable; defaults to benchmarks/results.", + ) + parser.add_argument( + "--bench-args", + action="store_true", + help="Print the shard's --bench arguments for asv, rather than a report.", + ) + parser.add_argument( + "--config", + default=str(BENCHMARK_DIR / "asv.conf.json"), + help="Base asv config for --config-out (default benchmarks/asv.conf.json).", + ) + parser.add_argument( + "--config-out", + default=None, + help="Where --asv-args writes the shard config (default: beside --config).", + ) + parser.add_argument( + "--env", + action="append", + default=[], + metavar="KEY=VALUE", + help="Override an env_nobuild variable in the generated config. Repeatable; " + "use it to sweep a variable the config pins to one value, e.g. " + "--env NUMBA_NUM_THREADS=4.", + ) + parser.add_argument( + "--asv-args", + action="store_true", + help="Write this shard's config and print every argument its asv run " + "needs, so launching a shard is one substitution. Needs --shard.", + ) + args = parser.parse_args(argv) + + results_dirs = args.results or [str(BENCHMARK_DIR / "results")] + benchmarks = load_benchmarks(results_dirs[0]) + weights = load_weights(results_dirs) + shards = plan(benchmarks, args.shards, weights) + + if args.bench_args or args.asv_args: + if args.shard is None: + parser.error("--bench-args and --asv-args need --shard") + if args.asv_args: + # Written here rather than by a call of its own: a shard that is + # told which benchmarks to run has to be told where to put them, + # and splitting that across two commands is two chances to pass + # one shard's benchmarks with another's results directory. + env = {} + for entry in args.env: + key, sep, value = entry.partition("=") + if not sep: + parser.error(f"--env wants KEY=VALUE, got {entry!r}") + env[key] = value + config_out = args.config_out or shard_config_path(args.config, args.shard) + write_shard_config(args.config, config_out, args.shard, env) + print("--config", config_out) + for pattern in bench_regexes(shards[args.shard]): + print("--bench", pattern) + return 0 + + if args.shard is None: + _report(benchmarks, shards, weights) + else: + for name in shards[args.shard]: + print(name) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/benchmarks/helpers/_threads.py b/benchmarks/helpers/_threads.py new file mode 100644 index 000000000..14ce8315c --- /dev/null +++ b/benchmarks/helpers/_threads.py @@ -0,0 +1,117 @@ +"""Choosing how many threads a benchmark run may use. + +numba reads ``NUMBA_NUM_THREADS`` once, when it brings up its threading layer, +and treats it as a ceiling rather than a setting: ``set_num_threads`` can hold +the pool lower for a block -- which is what :func:`~benchmarks.helpers._peakmem.numba_threads` +does while tracing -- but asking for more than the ceiling raises +``ValueError``. So a run's thread count has to be decided before the benchmark +process imports numba, which means the environment, which is what this module +resolves. + +asv copies ``os.environ`` into the processes it launches +(``Environment.run_executable``), so exporting the variable ahead of ``asv run`` +is enough:: + + export NUMBA_NUM_THREADS=$(python -m benchmarks.helpers._threads) + asv run ... + +One asymmetry to know about: asv layers the ``env_nobuild`` matrix *over* the +inherited environment, not under it, so a variable named in ``asv.conf.json`` +overrides the shell. ``NUMBA_THREADING_LAYER`` is named there because +fork-safety is not negotiable. The thread count deliberately is not, so a node +with more cores than a CI runner can decide for itself. + +The default is physical cores rather than ``os.cpu_count()``. These kernels are +floating-point and memory-bound, and a second hardware thread per core tends to +cost more in contention than it recovers in latency hiding: CI runners with 2 +physical cores plus SMT measured about 1.28x slower per thread on the same +scalar kernels than runners with 4 real cores. + +``UXARRAY_BENCH_THREADS`` overrides the default -- an integer, or ``physical`` +or ``logical`` to name a rule rather than a number. +""" + +import os +import subprocess +import sys + +__all__ = ["logical_cores", "physical_cores", "resolve"] + +_ENV_VAR = "UXARRAY_BENCH_THREADS" + + +def logical_cores(): + """Schedulable CPUs, honouring any affinity mask this process was given. + + ``os.cpu_count()`` reports the machine; ``os.sched_getaffinity`` reports + what this process may actually use, which is the smaller and more useful + number under a batch scheduler or a ``taskset``. + """ + if hasattr(os, "sched_getaffinity"): + return len(os.sched_getaffinity(0)) + return os.cpu_count() or 1 + + +def physical_cores(): + """Cores rather than hardware threads, or the logical count if unknown. + + Deliberately shells out rather than adding a dependency on ``psutil``: the + benchmark environment asv builds is defined by ``asv.conf.json``'s matrix, + and a helper that has to run in it is not worth an entry there. + """ + try: + if sys.platform == "darwin": + out = subprocess.run( + ["sysctl", "-n", "hw.physicalcpu"], + capture_output=True, text=True, timeout=5, check=True, + ).stdout + return max(1, int(out.strip())) + if sys.platform.startswith("linux"): + # One line per logical CPU, ",,,..."; distinct + # (socket, core) pairs are the physical cores. Counting distinct + # core ids alone would collapse two sockets into one. + out = subprocess.run( + ["lscpu", "-p=core,socket"], + capture_output=True, text=True, timeout=5, check=True, + ).stdout + pairs = { + line for line in (l.strip() for l in out.splitlines()) + if line and not line.startswith("#") + } + if pairs: + return max(1, len(pairs)) + except (OSError, ValueError, subprocess.SubprocessError): + pass + return logical_cores() + + +def resolve(spec=None): + """The thread count to run with. + + ``spec`` defaults to ``$UXARRAY_BENCH_THREADS``, and that to ``physical``. + Anything unrecognised falls back to the physical core count rather than + failing: this sits in front of a benchmark run, and refusing to start + because a variable is misspelt costs more than quietly doing the sensible + thing. The resolved number is echoed to stderr so it appears in the log. + """ + if spec is None: + spec = os.environ.get(_ENV_VAR, "").strip() + spec = (spec or "physical").lower() + + if spec == "logical": + return logical_cores() + if spec != "physical": + try: + return max(1, int(spec)) + except ValueError: + print( + f"{_ENV_VAR}={spec!r} is not an integer, 'physical' or 'logical'; " + "using the physical core count", + file=sys.stderr, + ) + # Never hand back more than this process may schedule on. + return min(physical_cores(), logical_cores()) + + +if __name__ == "__main__": + print(resolve()) diff --git a/benchmarks/hpc/local.sh b/benchmarks/hpc/local.sh new file mode 100755 index 000000000..52a62878b --- /dev/null +++ b/benchmarks/hpc/local.sh @@ -0,0 +1,108 @@ +#!/usr/bin/env bash +# Runs the sharded suite on the node you are already sitting on -- an +# interactive PBS session, typically -- with the shards concurrent rather than +# queued as separate jobs. +# +# This only makes sense because a derecho CPU node has 128 cores and a shard at +# NUMBA_NUM_THREADS=8 wants nine of them. Each shard is pinned to its own slice +# so they cannot land on each other's cores; what they do still share is memory +# bandwidth and last-level cache, and for the grid operations in this suite that +# is not nothing. So: the BASE-vs-HEAD ratios ``asv compare`` reports stay +# usable, since both sides of a comparison run inside the same shard under the +# same contention, but absolute timings come out noisier than a run that had a +# node to itself. Take those from one-shard-per-node (``submit.sh``). +# +# Usage, from the repository root: +# +# ./benchmarks/hpc/local.sh +# SHARDS=8 THREADS=4 ./benchmarks/hpc/local.sh +# REV=main^! ./benchmarks/hpc/local.sh +# +set -euo pipefail + +REPO="${REPO:-$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)}" +SHARDS="${SHARDS:-4}" +THREADS="${THREADS:-8}" +export REPO SHARDS THREADS +export REV="${REV:-HEAD^!}" +export CONFIG="${CONFIG:-asv.conf.hpc.json}" +# Left empty on purpose: stage.pbs derives it, so this works unchanged on +# derecho and casper both. +export ASV_MACHINE="${ASV_MACHINE:-}" +export ASV_ACTIVATE="${ASV_ACTIVATE:-true}" +# Also evaluated here, not just in the stages, so the core count below can be +# read with python rather than with a shell tool that lies about it. +eval "$ASV_ACTIVATE" + +STAGE_SCRIPT="$REPO/benchmarks/hpc/stage.pbs" +LOGS="${LOGS:-$REPO/benchmarks/hpc/logs}" +mkdir -p "$LOGS" + +# Neither PBS's NCPUS nor ``nproc`` can be trusted for this. NCPUS is the ncpus +# *requested per chunk*, 1 for a plain ``qsub -I``, and says nothing about the +# node. And GNU ``nproc`` honours OMP_NUM_THREADS and OMP_THREAD_LIMIT, so in a +# session that sets either it reports the OpenMP thread limit -- 1, or 2 -- and +# not the machine's cores at all. +# +# The affinity mask is the real answer: the CPUs this process may actually run +# on. It respects a cpuset the scheduler imposed and ignores OpenMP entirely. +# Override with CORES to hold the run to fewer than the mask allows. +CORES="${CORES:-$(python -c ' +import os +try: + print(len(os.sched_getaffinity(0))) +except AttributeError: # not Linux + print(os.cpu_count() or 1) +')}" +if ! [ "$CORES" -ge 1 ] 2>/dev/null; then + echo "could not work out a core count (got ${CORES:-empty}); set CORES" >&2 + exit 1 +fi +PER=$((CORES / SHARDS)) +if [ "$PER" -lt "$((THREADS + 1))" ]; then + echo "warning: $CORES cores over $SHARDS shards is $PER each, under the" >&2 + echo " $THREADS threads a shard wants; they will oversubscribe" >&2 +fi + +PIN="" +if command -v taskset >/dev/null; then + PIN="taskset" +else + echo "warning: no taskset, shards will not be pinned and will drift across cores" >&2 +fi + +echo "== setup ==" +STAGE=setup bash "$STAGE_SCRIPT" 2>&1 | tee "$LOGS/setup.log" + +echo "== $SHARDS shards, $PER cores each, $THREADS threads each ==" +pids=() +for S in $(seq 0 $((SHARDS - 1))); do + lo=$((S * PER)) + hi=$((lo + PER - 1)) + if [ -n "$PIN" ]; then + SHARD=$S STAGE=shard taskset -c "$lo-$hi" bash "$STAGE_SCRIPT" \ + >"$LOGS/shard$S.log" 2>&1 & + else + SHARD=$S STAGE=shard bash "$STAGE_SCRIPT" >"$LOGS/shard$S.log" 2>&1 & + fi + pid=$! + pids+=("$pid") + echo " shard $S -> cores $lo-$hi, pid $pid, log $LOGS/shard$S.log" +done + +# Every shard is waited on and its status reported, but a failure does not stop +# the merge: a tree missing one shard's rows is still worth having, same as the +# ``afteranyarray`` dependency the PBS path uses. +failed=0 +for S in $(seq 0 $((SHARDS - 1))); do + if wait "${pids[$S]}"; then + echo " shard $S ok" + else + echo " shard $S FAILED (see $LOGS/shard$S.log)" >&2 + failed=$((failed + 1)) + fi +done + +echo "== merge ==" +STAGE=merge bash "$STAGE_SCRIPT" 2>&1 | tee "$LOGS/merge.log" +[ "$failed" -eq 0 ] || echo "$failed shard(s) failed; merged what landed" >&2 diff --git a/benchmarks/hpc/stage.pbs b/benchmarks/hpc/stage.pbs new file mode 100644 index 000000000..1d2c1074d --- /dev/null +++ b/benchmarks/hpc/stage.pbs @@ -0,0 +1,98 @@ +#!/bin/bash +# One script, three stages of a sharded asv run. Submitted by ``submit.sh``, +# which sets STAGE and the rest of the environment; not meant to be qsub'd by +# hand. Defaults here are only so a stage is runnable outside PBS for debugging. +# +#PBS -N asv +#PBS -j oe +#PBS -l select=1:ncpus=128 +#PBS -l walltime=12:00:00 + +set -euo pipefail + +STAGE="${STAGE:?STAGE must be setup, shard or merge}" +SHARDS="${SHARDS:-4}" +REV="${REV:-HEAD^!}" +REPO="${REPO:?REPO must be the repository root}" +CONFIG="${CONFIG:-asv.conf.hpc.json}" +# One name for every shard: asv files results under +# ``results//-.json``, so shards that disagree leave +# nothing for the merge to line up. Derived rather than assumed, because the +# scheduler hands you a different node each login and asv would otherwise +# record each one as a machine of its own -- see ``_machine.default_name``. +ASV_MACHINE="${ASV_MACHINE:-$(cd "$REPO" && python -m benchmarks.helpers._machine --print)}" +# Shared, so the shards read the dyamond grids off campaign storage once +# between them rather than once each; the setup stage fills it. Optional, +# because running the stages locally has nothing to share and +# ``_fixtures.cache_dir`` already defaults to the checkout. +if [ -n "${UXARRAY_BENCH_CACHE_DIR:-}" ]; then + export UXARRAY_BENCH_CACHE_DIR +fi + +eval "${ASV_ACTIVATE:-true}" +cd "$REPO/benchmarks" + +case "$STAGE" in +setup) + # Everything that must happen exactly once, because every shard shares the + # filesystem it happens on: the fixture cache, the machine file, and the + # asv environments plus the wheel. Four shards building into one + # ``env/`` concurrently is the race this stage exists to prevent. + (cd .. && python -m benchmarks.helpers._fixtures) + asv machine --yes --config "$CONFIG" + (cd .. && python -m benchmarks.helpers._machine --name "$ASV_MACHINE") + # asv's own discovery-only mode: creates the environments, builds the + # project, writes results/benchmarks.json, runs no benchmark. The four + # NUMBA_NUM_THREADS environments share one build directory -- ``env_nobuild`` + # variables are omitted from the name ``dir_name`` hashes -- so this one + # build serves all of them. + # ``--config`` on every asv call, and ``-m`` on every one after the pin. + # Without ``--config`` asv falls back to ``asv.conf.json`` in the working + # directory -- the CI config -- so this stage would discover, create the + # environment and build against a different matrix from the one the shards + # then run under. The two happen to share a ``req`` matrix, and so an + # environment directory, which is why it worked at all rather than failing; + # it just meant the pre-build was warming the wrong config's environment. + # + # ``-m`` on every asv call from here on. The pin above renamed this host's + # entry to $ASV_MACHINE, so the hostname asv would otherwise look itself up + # under no longer exists in the machine file -- and once that file holds + # more than one machine, asv's fall-back to a lone entry does not apply + # either. Without it this fails with "No information stored about machine + # ''" immediately after the pin reports success. + asv run --bench just-discover --config "$CONFIG" -m "$ASV_MACHINE" "$REV" + PYTHONPATH=.. python -m benchmarks.helpers._partition --shards "$SHARDS" + ;; +shard) + SHARD="${PBS_ARRAY_INDEX:-${SHARD:?}}" + # ``--asv-args`` writes this shard's config -- a results_dir of its own, + # plus any --env override -- and prints it with the --bench patterns. + # THREADS is optional: unset, the config's own NUMBA_NUM_THREADS stands. + # Set, it overrides it, and lands in the environment name, so a run at one + # thread count files its results separately from a run at another. + THREAD_ARG="" + if [ -n "${THREADS:-}" ]; then + THREAD_ARG="--env NUMBA_NUM_THREADS=$THREADS" + fi + ASV_ARGS=$(PYTHONPATH=.. python -m benchmarks.helpers._partition \ + --shards "$SHARDS" --shard "$SHARD" --config "$CONFIG" $THREAD_ARG --asv-args) + status=0 + asv run --show-stderr -m "$ASV_MACHINE" $ASV_ARGS "$REV" || status=$? + # 2 is "a benchmark failed", not "the shard broke": its other results are + # complete and the merge should still get them. Anything else is real. + if [ "$status" -eq 2 ]; then + echo "asv exited 2: some benchmarks failed, see above" >&2 + elif [ "$status" -ne 0 ]; then + exit "$status" + fi + ;; +merge) + (cd .. && python -m benchmarks.helpers._merge \ + --out benchmarks/results benchmarks/results.shard*) + asv show --config "$CONFIG" -m "$ASV_MACHINE" || true + ;; +*) + echo "unknown STAGE $STAGE" >&2 + exit 2 + ;; +esac diff --git a/benchmarks/hpc/submit.sh b/benchmarks/hpc/submit.sh new file mode 100755 index 000000000..801c71f59 --- /dev/null +++ b/benchmarks/hpc/submit.sh @@ -0,0 +1,85 @@ +#!/usr/bin/env bash +# Submits a sharded asv run on derecho as three chained PBS jobs. +# +# setup one node, once: fixture cache, machine file, asv environments and +# the wheel, then the shard plan. Everything the shards would +# otherwise race each other to create on the filesystem they share. +# shards a job array, one node each, exclusive. A ``time_*`` result is only +# worth having if nothing else is competing for the machine, so one +# shard per node rather than several. +# merge one node, once: combines the shards' results directories and shows +# the run. +# +# The merge depends on ``afteranyarray`` rather than ``afterokarray`` on +# purpose: a shard that fails still leaves results worth merging, and a suite +# with one long-broken benchmark should not cost you the other eighty. +# +# Usage: +# +# PBS_ACCOUNT=UXXX0001 ./benchmarks/hpc/submit.sh +# PBS_ACCOUNT=UXXX0001 THREADS=8 ./benchmarks/hpc/submit.sh +# PBS_ACCOUNT=UXXX0001 SHARDS=8 REV=main^! ./benchmarks/hpc/submit.sh +# +# THREADS overrides the config's NUMBA_NUM_THREADS and is recorded in the +# environment name, so runs at different thread counts do not overwrite each +# other's results. Leave it unset to take the config's value. +# +set -euo pipefail + +: "${PBS_ACCOUNT:?set PBS_ACCOUNT to your project code, e.g. PBS_ACCOUNT=UXXX0001}" + +REPO="${REPO:-$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)}" +SHARDS="${SHARDS:-4}" +# A single commit by default. ``asv run`` takes a range, so ``main^!`` is main's +# tip alone and ``base..head`` is every commit between. +REV="${REV:-HEAD^!}" +QUEUE="${QUEUE:-main}" +CONFIG="${CONFIG:-asv.conf.hpc.json}" +# Empty means stage.pbs derives it from NCAR_HOST or the node name. +ASV_MACHINE="${ASV_MACHINE:-}" +WALLTIME="${WALLTIME:-12:00:00}" +SETUP_WALLTIME="${SETUP_WALLTIME:-02:00:00}" +# Off /glade/derecho/scratch so every shard shares one fixture cache. The +# default in ``_fixtures.cache_dir`` follows the checkout, which would give a +# second working tree its own empty cache and re-read every source. +CACHE_DIR="${UXARRAY_BENCH_CACHE_DIR:-/glade/derecho/scratch/$USER/uxarray-bench}" +# Submit from a shell that already has ``asv`` on PATH -- ``-V`` below carries +# that environment into all three jobs, which is both simpler and less brittle +# than reactivating inside them. Set ASV_ACTIVATE only if you would rather the +# jobs do it themselves; it is eval'd once per stage. +export ASV_ACTIVATE="${ASV_ACTIVATE:-true}" +command -v asv >/dev/null || [ "$ASV_ACTIVATE" != "true" ] || { + echo "asv is not on PATH; activate your environment first, or set ASV_ACTIVATE" >&2 + exit 1 +} + +STAGE_SCRIPT="$REPO/benchmarks/hpc/stage.pbs" +# Passed through the environment rather than in ``-v``, whose value list is +# comma-separated and so cannot hold an activation command or a path with a +# comma in it. +export REPO SHARDS REV CONFIG ASV_MACHINE +export THREADS="${THREADS:-}" +export UXARRAY_BENCH_CACHE_DIR="$CACHE_DIR" + +mkdir -p "$CACHE_DIR" + +setup=$(qsub -A "$PBS_ACCOUNT" -q "$QUEUE" -N asv-setup \ + -l select=1:ncpus=128 -l walltime="$SETUP_WALLTIME" \ + -V -v "STAGE=setup" "$STAGE_SCRIPT") +echo "setup $setup" + +shards=$(qsub -A "$PBS_ACCOUNT" -q "$QUEUE" -N asv-shard \ + -J "0-$((SHARDS - 1))" \ + -l select=1:ncpus=128 -l walltime="$WALLTIME" \ + -W "depend=afterok:$setup" \ + -V -v "STAGE=shard" "$STAGE_SCRIPT") +echo "shards $shards ($SHARDS of them)" + +merge=$(qsub -A "$PBS_ACCOUNT" -q "$QUEUE" -N asv-merge \ + -l select=1:ncpus=1 -l walltime=00:30:00 \ + -W "depend=afteranyarray:$shards" \ + -V -v "STAGE=merge" "$STAGE_SCRIPT") +echo "merge $merge" +echo +echo "watch with: qstat -u $USER -t" +echo "results in: $REPO/benchmarks/results/$ASV_MACHINE/"