diff --git a/.github/workflows/eliza-vulkan-validation.yml b/.github/workflows/eliza-vulkan-validation.yml index 2847ff3df..7a9908f06 100644 --- a/.github/workflows/eliza-vulkan-validation.yml +++ b/.github/workflows/eliza-vulkan-validation.yml @@ -16,6 +16,8 @@ on: types: [opened, synchronize, reopened] paths: - '.github/workflows/eliza-vulkan-validation.yml' + - 'scripts/ci_verify_vulkan.py' + - 'scripts/test_ci_verify_vulkan.py' - 'ggml/src/ggml-vulkan/**' - 'ggml/include/**' - 'src/**' @@ -34,9 +36,12 @@ env: LLAMA_LOG_TIMESTAMPS: 1 jobs: - ubuntu-vulkan-llvmpipe-validation: + build-vulkan: runs-on: ubuntu-24.04 - timeout-minutes: 60 + timeout-minutes: 25 + outputs: + head: ${{ steps.package.outputs.head }} + build-sha: ${{ steps.package.outputs.build-sha }} steps: - name: Clone @@ -80,7 +85,9 @@ jobs: - name: Configure run: | source ./vulkan_sdk/setup-env.sh + # This runtime is shared across runners with different CPU features. cmake -B build-vulkan \ + -DGGML_NATIVE=OFF \ -DGGML_VULKAN=ON \ -DGGML_METAL=OFF \ -DLLAMA_CURL=OFF \ @@ -90,6 +97,7 @@ jobs: - name: Build run: | source ./vulkan_sdk/setup-env.sh + grep -q "^GGML_NATIVE:BOOL=OFF$" build-vulkan/CMakeCache.txt cmake --build build-vulkan -j $(nproc) \ --target llama-cli llama-completion \ test-backend-ops test-quantize-fns @@ -99,120 +107,149 @@ jobs: source ./vulkan_sdk/setup-env.sh vulkaninfo --summary | head -40 - - name: test-quantize-fns (CPU; smoke for Eliza custom quants) + - name: Check result verifier and CPU quantizers + shell: bash run: | - ./build-vulkan/bin/test-quantize-fns | tee /tmp/quantize-fns.log - # Must mention the Eliza-divergent quant types. + set -euo pipefail + python3 scripts/test_ci_verify_vulkan.py + ./build-vulkan/bin/test-quantize-fns | tee quantize-fns.log for t in qjl1_256 q4_polar tbq3_0 tbq4_0 tbq3_tcq; do - grep -q "Testing $t" /tmp/quantize-fns.log \ - || (echo "MISSING quant: $t" && exit 1) + grep -q "Testing $t" quantize-fns.log done - - name: test-backend-ops on Vulkan (filtered subset) - env: - GGML_VK_VISIBLE_DEVICES: '0' - GGML_VK_DISABLE_F16: '1' - GGML_VK_DISABLE_COOPMAT: '1' + - name: Package one runtime for every shard + id: package + shell: bash run: | - source ./vulkan_sdk/setup-env.sh - # Filter to ops that matter most for inference correctness. - # Full sweep is too slow on llvmpipe (~hours). - for op in MUL_MAT MUL_MAT_ID ADD MUL SOFT_MAX RMS_NORM CPY ROPE FLASH_ATTN_EXT; do - echo "=== $op ===" - timeout 600 ./build-vulkan/bin/test-backend-ops \ - -b Vulkan0 -o "$op" \ - 2>&1 | tee -a /tmp/backend-ops.log || true - done - # Summary lines for the artifact viewer. - echo "=== FAIL SUMMARY ===" - grep -E "Backend Vulkan0:.*FAIL|tests passed|Failing tests:" \ - /tmp/backend-ops.log || true - - - name: test-backend-ops Eliza custom-quant parity (Vulkan) - env: - GGML_VK_VISIBLE_DEVICES: '0' - GGML_VK_DISABLE_F16: '1' - GGML_VK_DISABLE_COOPMAT: '1' + set -euo pipefail + tar -czf vulkan-runtime.tar.gz -C build-vulkan bin + sha256sum vulkan-runtime.tar.gz > vulkan-runtime.sha256 + echo "head=$(git rev-parse HEAD)" >> "$GITHUB_OUTPUT" + echo "build-sha=$(cut -d ' ' -f 1 vulkan-runtime.sha256)" >> "$GITHUB_OUTPUT" + + - uses: actions/upload-artifact@v4 + with: + name: vulkan-validation-runtime + path: | + vulkan-runtime.tar.gz + vulkan-runtime.sha256 + quantize-fns.log + retention-days: 14 + + # MUL_MAT and FLASH_ATTN_EXT are split across 128 workers. test-backend-ops + # assigns every matching occurrence to one worker by estimated cost (longest + # first onto the lightest worker), so the large hsk=512/576 FLASH_ATTN_EXT + # cases no longer cluster on one worker the way index-modulo assignment let + # them (shard 86 exceeded the 600 s limit on llvmpipe). Sharded invocations + # get 1200 s; worker 0 additionally runs every unsharded operation. + numerical-shards: + needs: build-vulkan + runs-on: ubuntu-24.04 + timeout-minutes: 75 + strategy: + fail-fast: false + max-parallel: 4 + matrix: + shard: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72, 73, 74, 75, 76, 77, 78, 79, 80, 81, 82, 83, 84, 85, 86, 87, 88, 89, 90, 91, 92, 93, 94, 95, 96, 97, 98, 99, 100, 101, 102, 103, 104, 105, 106, 107, 108, 109, 110, 111, 112, 113, 114, 115, 116, 117, 118, 119, 120, 121, 122, 123, 124, 125, 126, 127] + env: + GGML_VK_VISIBLE_DEVICES: '0' + GGML_VK_DISABLE_F16: '1' + GGML_VK_DISABLE_COOPMAT: '1' + SHARD: ${{ matrix.shard }} + QUALIFIED_HEAD: ${{ needs.build-vulkan.outputs.head }} + BUILD_SHA: ${{ needs.build-vulkan.outputs.build-sha }} + steps: + - uses: actions/checkout@v6 + - name: System Vulkan runtime run: | - source ./vulkan_sdk/setup-env.sh - # Eliza custom quants (Q1_0_g32/g128, TBQ3_0/4_0, TBQ3_K/4_K, - # QJL1_256, Q4_POLAR, TBQ3_TCQ). These have CPU+Metal kernels - # but no Vulkan kernels yet. The harness emits one - # "not supported" line per (op,type,shape) — that is the - # intended baseline. If a future PR adds a Vulkan kernel for - # any of these types, this step will start emitting real - # PASS/FAIL lines automatically because supports_op() will - # return true and the test harness will compare against CPU. - # - # We do NOT fail the job on "not supported" lines (expected). - # We DO fail on any "FAIL" line for these types — that would - # signal a Vulkan kernel that diverges from the CPU reference. - for op in GET_ROWS CPY MUL_MAT \ - ATTN_SCORE_TBQ ATTN_SCORE_POLAR \ - ISTFT; do - echo "=== $op (Eliza custom quants) ===" - timeout 300 ./build-vulkan/bin/test-backend-ops \ - -b Vulkan0 -o "$op" \ - 2>&1 | tee -a /tmp/eliza-custom-quants.log || true - done - # Count skip/pass/fail among lines mentioning custom quant - # type names. The grep patterns target the type prefixes used - # in the test_case::vars() output. - echo "=== Eliza custom-quant summary ===" - for t in q1_0_g32 q1_0_g128 tbq3_0 tbq4_0 tbq3_k tbq4_k qjl1_256 q4_polar tbq3_tcq; do - n_skip=$(grep -c "type=$t.*not supported" /tmp/eliza-custom-quants.log || true) - n_fail=$(grep -c "type=$t.*FAIL" /tmp/eliza-custom-quants.log || true) - n_pass=$(grep -c "type=$t.*OK" /tmp/eliza-custom-quants.log || true) - echo " $t: skip=$n_skip pass=$n_pass fail=$n_fail" - if [ "$n_fail" -gt 0 ]; then - echo "::error::Vulkan kernel divergence for type=$t (see /tmp/eliza-custom-quants.log)" - exit 1 - fi - done - # ATTN_SCORE_TBQ / ATTN_SCORE_POLAR: CPU oracle landed at - # 4ab3e26f6. Vulkan kernels were corrected at 2e5094b2c - # (turbo4 uncondition) and 1cbc7e0f8 (turbo3 layout + codebook - # + uncondition); all 3 TBQ shapes now pass parity vs CPU. - # Promoted from warning to hard-fail so future shader changes - # cannot silently regress attention scores. - echo "=== Eliza custom-attention (Vulkan) ===" - for op_name in ATTN_SCORE_TBQ ATTN_SCORE_POLAR; do - n_fail=$(grep -c "$op_name.*FAIL" /tmp/eliza-custom-quants.log || true) - n_skip=$(grep -c "$op_name.*not supported" /tmp/eliza-custom-quants.log || true) - n_pass=$(grep -c "$op_name.*OK" /tmp/eliza-custom-quants.log || true) - echo " $op_name: skip=$n_skip pass=$n_pass fail=$n_fail" - if [ "$n_fail" -gt 0 ]; then - echo "::error::Vulkan kernel diverges from CPU oracle for $op_name (see /tmp/eliza-custom-quants.log)" - exit 1 - fi - done - # ISTFT: Vulkan kernel at vulkan-shaders/istft.comp + ggml_vk_istft - # dispatch in ggml-vulkan.cpp. All 4 parity cases against the CPU - # reference must PASS — any FAIL or "not supported" line indicates - # a regression in the Vulkan iSTFT dispatch path or shader. - echo "=== ISTFT (Vulkan, gated) ===" - for op_name in ISTFT; do - n_fail=$(grep -c "$op_name.*FAIL" /tmp/eliza-custom-quants.log || true) - n_skip=$(grep -c "$op_name.*not supported" /tmp/eliza-custom-quants.log || true) - n_pass=$(grep -c "$op_name.*OK" /tmp/eliza-custom-quants.log || true) - echo " $op_name: skip=$n_skip pass=$n_pass fail=$n_fail" - if [ "$n_fail" -gt 0 ]; then - echo "::error::Vulkan iSTFT diverges from CPU reference (see ggml-cpu/ops.cpp:ggml_compute_forward_istft_f32)" - exit 1 - fi - if [ "$n_skip" -gt 0 ]; then - echo "::error::Vulkan unexpectedly reports ISTFT as not supported (regression in supports_op)" - exit 1 + sudo add-apt-repository -y ppa:kisak/kisak-mesa + sudo apt-get update -y + sudo apt-get install -y mesa-vulkan-drivers libvulkan1 + - uses: actions/download-artifact@v4 + with: + name: vulkan-validation-runtime + - name: Verify and unpack shared build + shell: bash + run: | + set -euo pipefail + test "$(git rev-parse HEAD)" = "$QUALIFIED_HEAD" + test "$(cut -d ' ' -f 1 vulkan-runtime.sha256)" = "$BUILD_SHA" + sha256sum --check vulkan-runtime.sha256 + mkdir build-vulkan + tar -xzf vulkan-runtime.tar.gz -C build-vulkan + mkdir results + lscpu > results/cpu.txt + : > results/ledger.jsonl + - name: Complete assigned numerical occurrences + shell: bash + run: | + set -euo pipefail + for group in broad custom; do + if [ "$group" = broad ]; then + ops="MUL_MAT MUL_MAT_ID ADD MUL SOFT_MAX RMS_NORM CPY ROPE FLASH_ATTN_EXT" + limit=600 + else + ops="GET_ROWS CPY MUL_MAT ATTN_SCORE_TBQ ATTN_SCORE_POLAR ISTFT" + limit=300 fi + for op in $ops; do + shard_args=() + verify_args=() + op_limit="$limit" + if [ "$op" = MUL_MAT ] || [ "$op" = FLASH_ATTN_EXT ]; then + shard_args=(--test-shard "$SHARD/128") + verify_args=(--shard "$SHARD") + op_limit=1200 + elif [ "$SHARD" != 0 ]; then + continue + fi + log="results/$group-$op.log" + start=$(date +%s) + timeout "$op_limit" ./build-vulkan/bin/test-backend-ops \ + -b Vulkan0 -o "$op" "${shard_args[@]}" \ + 2>"results/$group-$op.stderr" | tee "$log" + echo "::notice::$group/$op shard $SHARD finished in $(( $(date +%s) - start ))s of ${op_limit}s" + python3 scripts/ci_verify_vulkan.py record "$group" "$op" "$log" results/ledger.jsonl \ + --head "$QUALIFIED_HEAD" --build-sha "$BUILD_SHA" "${verify_args[@]}" + done done - - - name: Upload logs + - uses: actions/upload-artifact@v4 if: always() - uses: actions/upload-artifact@v4 with: - name: eliza-vulkan-validation-logs - path: | - /tmp/quantize-fns.log - /tmp/backend-ops.log - /tmp/eliza-custom-quants.log + name: vulkan-shard-${{ matrix.shard }} + path: results/ + if-no-files-found: error + retention-days: 14 + + ubuntu-vulkan-llvmpipe-validation: + if: always() + needs: [build-vulkan, numerical-shards] + runs-on: ubuntu-24.04 + timeout-minutes: 5 + env: + BUILD_RESULT: ${{ needs.build-vulkan.result }} + SHARD_RESULT: ${{ needs.numerical-shards.result }} + QUALIFIED_HEAD: ${{ needs.build-vulkan.outputs.head }} + BUILD_SHA: ${{ needs.build-vulkan.outputs.build-sha }} + steps: + - name: Require every producer to finish successfully + run: | + test "$BUILD_RESULT" = success + test "$SHARD_RESULT" = success + - uses: actions/checkout@v6 + - uses: actions/download-artifact@v4 + with: + pattern: vulkan-shard-* + path: shard-results + - name: Reconcile all occurrences against the complete inventories + shell: bash + run: | + set -euo pipefail + test "$(git rev-parse HEAD)" = "$QUALIFIED_HEAD" + python3 scripts/ci_verify_vulkan.py aggregate shard-results \ + --head "$QUALIFIED_HEAD" --build-sha "$BUILD_SHA" | tee vulkan-coverage.json + - uses: actions/upload-artifact@v4 + with: + name: vulkan-validation-coverage + path: vulkan-coverage.json retention-days: 14 diff --git a/scripts/ci_verify_vulkan.py b/scripts/ci_verify_vulkan.py new file mode 100644 index 000000000..d474d84fa --- /dev/null +++ b/scripts/ci_verify_vulkan.py @@ -0,0 +1,167 @@ +#!/usr/bin/env python3 +"""Verify complete Vulkan results and the union of deterministic test shards. + +test-backend-ops assigns each matching occurrence to one shard by estimated cost +(longest first onto the lightest shard) and prints that assignment in every +worker's inventory; all workers must agree on it and their union must cover +every occurrence exactly once. + +Unsupported outcomes remain explicit and never count as numerical passes. +Occurrence indices preserve duplicate cases and whole-graph tests; support-mode +inventories are deliberately not used because they omit whole-graph tests. +""" +import argparse +from collections import Counter +import hashlib +import json +from pathlib import Path +import re +import sys + +ANSI = re.compile(r"\x1b\[[0-9;]*m") +LANES = { + "broad": ["MUL_MAT", "MUL_MAT_ID", "ADD", "MUL", "SOFT_MAX", "RMS_NORM", "CPY", "ROPE", "FLASH_ATTN_EXT"], + "custom": ["GET_ROWS", "CPY", "MUL_MAT", "ATTN_SCORE_TBQ", "ATTN_SCORE_POLAR", "ISTFT"], +} +SHARDED = {("broad", "MUL_MAT"), ("broad", "FLASH_ATTN_EXT"), ("custom", "MUL_MAT")} +SHARDS = 128 + + +def sha(path): + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def verify(text, operation, shard=None): + text = ANSI.sub("", text) + results = re.findall(r"^ ([A-Z0-9_]+)\((.*)\): (.*)$", text, re.M) + passed, unsupported, observed = [], [], [] + for op, params, status in results: + if op != operation: + raise ValueError("Unexpected operation: " + op) + observed.append(params) + if status.strip() == "OK": + passed.append(params) + elif status.strip() == "not supported [Vulkan0]": + unsupported.append(params) + else: + raise ValueError("Nonpassing or incomplete outcome: " + op + "(" + params + "): " + status) + inventory = [] + assignment = [] + selected = [] + inventory_rows = re.findall(r"^Shard inventory (\d+) -> (\d+): ([A-Z0-9_]+)\((.*)\)$", text, re.M) + shard_rows = re.findall(r"^Shard (\d+)/(\d+): (\d+) selected of (\d+) matching cases$", text, re.M) + if shard is not None: + index, count = shard + if not 0 <= index < count <= 256: + raise ValueError("Invalid shard index/count") + for ordinal, (actual, owner, op, params) in enumerate(inventory_rows): + if int(actual) != ordinal or op != operation: + raise ValueError("Noncontiguous or unexpected inventory") + if not 0 <= int(owner) < count: + raise ValueError("Occurrence assigned outside the shard range") + inventory.append(params) + assignment.append(int(owner)) + if not inventory: + raise ValueError("Missing full occurrence inventory") + selected = [i for i, owner in enumerate(assignment) if owner == index] + if shard_rows != [(str(index), str(count), str(len(selected)), str(len(inventory)))]: + raise ValueError("Missing or inconsistent shard completion") + if observed != [inventory[i] for i in selected]: + raise ValueError("Shard results do not equal assigned occurrence sequence") + elif inventory_rows or shard_rows or not passed: + raise ValueError("Unexpected shard output or no numerical cases") + summaries = re.findall(r"^ (\d+)/(\d+) tests passed$", text, re.M) + if summaries != [(str(len(passed)), str(len(passed)))]: + raise ValueError("Missing, repeated or inconsistent numerical completion summary") + if len(re.findall(r"^ Backend Vulkan0: OK$", text, re.M)) != 1: + raise ValueError("Vulkan backend did not complete exactly once") + if operation == "ISTFT" and (unsupported or len(passed) < 4): + raise ValueError("All four required ISTFT cases must execute numerically") + return {"operation": operation, "numericalPasses": len(passed), + "unsupportedCases": unsupported, "numericalCases": passed, + "inventory": inventory, "assignment": assignment, "selectedIndices": selected, + "shard": list(shard) if shard is not None else None} + + +def aggregate(records, head, build_sha, shards=SHARDS): + expected = {(group, op, index if (group, op) in SHARDED else None) + for group, ops in LANES.items() for op in ops + for index in (range(shards) if (group, op) in SHARDED else [0])} + found = {} + for r in records: + if r["head"] != head or r["buildSha256"] != build_sha: + raise ValueError("Source or build artifact mismatch") + shard = r["shard"] + key = (r["group"], r["operation"], shard[0] if shard else None) + if key not in expected or key in found or (shard and shard[1] != shards): + raise ValueError("Unexpected, duplicate or misconfigured invocation") + found[key] = r + if set(found) != expected: + raise ValueError("Missing terminal invocations") + totals = [] + for group, ops in LANES.items(): + for op in ops: + family = [r for (g, o, _), r in found.items() if (g, o) == (group, op)] + numeric = sum(r["numericalPasses"] for r in family) + if numeric == 0: + raise ValueError("No numerical execution for " + group + "/" + op) + if (group, op) in SHARDED: + inventory = family[0]["inventory"] + assignment = family[0]["assignment"] + indices = [] + for r in family: + if r["inventory"] != inventory: + raise ValueError("Workers generated different complete inventories") + if r["assignment"] != assignment: + raise ValueError("Workers computed different shard assignments") + if r["selectedIndices"] != [i for i, owner in enumerate(assignment) if owner == r["shard"][0]]: + raise ValueError("Shard ran occurrences it was not assigned") + indices.extend(r["selectedIndices"]) + if Counter(indices) != Counter(range(len(inventory))): + raise ValueError("Shard union omits or repeats occurrences") + totals.append({"group": group, "operation": op, "numericalPasses": numeric, + "explicitUnsupported": sum(len(r["unsupportedCases"]) for r in family)}) + return totals + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + sub = parser.add_subparsers(dest="mode", required=True) + record = sub.add_parser("record") + record.add_argument("group", choices=LANES) + record.add_argument("operation") + record.add_argument("log", type=Path) + record.add_argument("ledger", type=Path) + record.add_argument("--shard", type=int) + collect = sub.add_parser("aggregate") + collect.add_argument("root", type=Path) + for command in (record, collect): + command.add_argument("--head", required=True) + command.add_argument("--build-sha", required=True) + args = parser.parse_args() + if args.mode == "record": + shard = (args.shard, SHARDS) if args.shard is not None else None + r = verify(args.log.read_text(), args.operation, shard) + r.update(group=args.group, head=args.head, buildSha256=args.build_sha, + log=args.log.name, logSha256=sha(args.log)) + with args.ledger.open("a") as output: + output.write(json.dumps(r) + "\n") + # stdout is the machine-readable CLI result; failures retain stderr and nonzero exit. + sys.stdout.write(json.dumps({k: r[k] for k in ("group", "operation", "shard", "numericalPasses")}) + "\n") + else: + records = [] + for ledger in sorted(args.root.glob("*/ledger.jsonl")): + for line in ledger.read_text().splitlines(): + r = json.loads(line) + log = ledger.parent / r["log"] + if log.name != r["log"] or sha(log) != r["logSha256"]: + raise ValueError("Log path or digest mismatch") + verified = verify(log.read_text(), r["operation"], r["shard"]) + if any(r[key] != value for key, value in verified.items()): + raise ValueError("Receipt does not match raw output") + records.append(r) + sys.stdout.write(json.dumps(aggregate(records, args.head, args.build_sha), indent=2) + "\n") + + +if __name__ == "__main__": + main() diff --git a/scripts/test_ci_verify_vulkan.py b/scripts/test_ci_verify_vulkan.py new file mode 100644 index 000000000..a955e6587 --- /dev/null +++ b/scripts/test_ci_verify_vulkan.py @@ -0,0 +1,171 @@ +#!/usr/bin/env python3 +"""Exercise the backend console boundary: incomplete and unsupported runs cannot qualify.""" +import copy +import hashlib +import json +from pathlib import Path +import subprocess +import sys +import tempfile +import unittest +from ci_verify_vulkan import LANES, SHARDED, SHARDS, aggregate, verify + + +def completed(operation="ADD", statuses=("OK",)): + rows = [f" {operation}(case={i}): {status}" for i, status in enumerate(statuses)] + count = statuses.count("OK") + return "\n".join(rows + [f" {count}/{count} tests passed", " Backend Vulkan0: OK", ""]) + + +class VulkanCompletionTests(unittest.TestCase): + def test_counts_numeric_and_explicit_unsupported_separately(self): + text = completed(statuses=("OK", "not supported [Vulkan0]")) + result = verify(text.replace("OK", "\x1b[32mOK\x1b[0m"), "ADD") + self.assertEqual(result["numericalPasses"], 1) + self.assertEqual(result["unsupportedCases"], ["case=1"]) + + def test_rejects_invalid_or_incomplete_backend_output(self): + valid = completed() + invalid = { + "empty": "", + "unsupported-only": completed(statuses=("not supported [Vulkan0]",)), + "divergence": completed(statuses=("FAIL",)), + "partial-result": " ADD(case=0): ", + "missing-summary": valid.replace(" 1/1 tests passed\n", ""), + "repeated-summary": valid + " 1/1 tests passed\n", + "wrong-count": valid.replace("1/1", "2/2"), + "missing-backend": valid.replace(" Backend Vulkan0: OK\n", ""), + "wrong-operation": completed("MUL"), + "unexpected-status": completed(statuses=("skipping large tensors for speed",)), + } + for name, text in invalid.items(): + with self.subTest(name=name), self.assertRaises(ValueError): + verify(text, "ADD") + + def test_istft_requires_four_numeric_cases_without_unsupported(self): + self.assertEqual(verify(completed("ISTFT", ("OK",) * 4), "ISTFT")["numericalPasses"], 4) + for statuses in [("OK",) * 3, ("OK",) * 4 + ("not supported [Vulkan0]",)]: + with self.subTest(statuses=statuses), self.assertRaises(ValueError): + verify(completed("ISTFT", statuses), "ISTFT") + + +def balanced(n, count): + # Not round-robin: the verifier must follow the printed assignment. + return [(count - 1 - i) % count for i in range(n)] + + +def shard_output(op, index, count, inventory, assignment=None): + assignment = balanced(len(inventory), count) if assignment is None else assignment + rows = [f"Shard inventory {i} -> {owner}: {op}({params})" + for i, (params, owner) in enumerate(zip(inventory, assignment))] + selected = [params for params, owner in zip(inventory, assignment) if owner == index] + rows += [f" {op}({params}): OK" for params in selected] + rows += [f"Shard {index}/{count}: {len(selected)} selected of {len(inventory)} matching cases", + f" {len(selected)}/{len(selected)} tests passed", " Backend Vulkan0: OK", ""] + return "\n".join(rows) + + +class ShardCoverageTests(unittest.TestCase): + def records(self): + records = [] + for group, ops in LANES.items(): + for op in ops: + for index in (range(2) if (group, op) in SHARDED else [None]): + if index is None: + r = verify(completed(op, ("OK",) * (4 if op == "ISTFT" else 1)), op) + else: + # Duplicate identities and a whole-graph case must remain occurrences. + r = verify(shard_output(op, index, 2, ["o=1", "o=1", "o=3"]), op, (index, 2)) + r.update(group=group, head="head", buildSha256="build") + records.append(r) + return records + + def test_union_preserves_duplicate_and_whole_graph_occurrences(self): + result = aggregate(self.records(), "head", "build", shards=2) + for r in result: + if (r["group"], r["operation"]) in SHARDED: + self.assertEqual(r["numericalPasses"], 3) + + def test_rejects_incomplete_or_inconsistent_worker_sets(self): + original = self.records() + controls = {} + controls["missing-shard"] = original[1:] + controls["duplicate-shard"] = original + [original[0]] + for name, key, value in [ + ("wrong-source", "head", "other"), + ("wrong-build", "buildSha256", "other"), + ("wrong-inventory", "inventory", ["different"]), + ("duplicate-index", "selectedIndices", [0, 0]), + ("different-assignment", "assignment", [0, 0, 0]), + ("wrong-count", "shard", [0, 3]), + ]: + records = copy.deepcopy(original) + records[0][key] = value + controls[name] = records + records = copy.deepcopy(original) + for r in records: + if r["operation"] == "MUL_MAT": + r["numericalPasses"] = 0 + controls["all-unsupported-family"] = records + for name, records in controls.items(): + with self.subTest(name=name), self.assertRaises(ValueError): + aggregate(records, "head", "build", shards=2) + + def test_rejects_missing_extra_or_reordered_occurrence_output(self): + valid = shard_output("MUL_MAT", 0, 2, ["o=1", "o=1", "o=3"]) + # balanced(3, 2) == [1, 0, 1]: shard 0 owns only the second occurrence. + self.assertEqual(verify(valid, "MUL_MAT", (0, 2))["selectedIndices"], [1]) + for text in [valid.replace(" MUL_MAT(o=1): OK\n", ""), + valid.replace(" MUL_MAT(o=1): OK", " MUL_MAT(o=3): OK"), + valid.replace("Shard inventory 1 ->", "Shard inventory 2 ->"), + valid.replace("Shard inventory 2 -> 1:", "Shard inventory 2 -> 0:"), + valid.replace("Shard inventory 2 -> 1:", "Shard inventory 2 -> 2:"), + valid.replace("Shard inventory 2 -> 1:", "Shard inventory 2:"), + valid.replace("1 selected of 3", "2 selected of 3")]: + with self.subTest(text=text), self.assertRaises(ValueError): + verify(text, "MUL_MAT", (0, 2)) + for shard in [(-1, 2), (2, 2), (0, 0), (0, 257)]: + with self.subTest(shard=shard), self.assertRaises(ValueError): + verify(valid, "MUL_MAT", shard) + + +class AggregateCommandTests(unittest.TestCase): + def test_cli_rechecks_files_and_rejects_changed_or_missing_artifacts(self): + script = Path(__file__).with_name("ci_verify_vulkan.py") + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + for index in range(SHARDS): + folder = root / f"worker-{index}" + folder.mkdir() + records = [] + for group, ops in LANES.items(): + for op in ops: + sharded = (group, op) in SHARDED + if not sharded and index != 0: + continue + # Some shards have no assigned occurrences; the complete family must still execute. + text = (shard_output(op, index, SHARDS, ["o=1", "o=1", "o=3"]) + if sharded else completed(op, ("OK",) * (4 if op == "ISTFT" else 1))) + log = folder / f"{group}-{op}.log" + log.write_text(text) + record = verify(text, op, (index, SHARDS) if sharded else None) + record.update(group=group, head="head", buildSha256="build", log=log.name, + logSha256=hashlib.sha256(log.read_bytes()).hexdigest()) + records.append(record) + (folder / "ledger.jsonl").write_text("".join(json.dumps(r) + "\n" for r in records)) + command = [sys.executable, str(script), "aggregate", str(root), "--head", "head", "--build-sha", "build"] + result = subprocess.run(command, capture_output=True, text=True) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(len(json.loads(result.stdout)), sum(map(len, LANES.values()))) + log = root / "worker-0" / "broad-MUL_MAT.log" + original = log.read_bytes() + log.write_text("changed artifact") + self.assertNotEqual(subprocess.run(command, capture_output=True).returncode, 0) + log.write_bytes(original) + ledger = root / "worker-1" / "ledger.jsonl" + ledger.unlink() + self.assertNotEqual(subprocess.run(command, capture_output=True).returncode, 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test-backend-ops.cpp b/tests/test-backend-ops.cpp index eddea9555..aa3d88452 100644 --- a/tests/test-backend-ops.cpp +++ b/tests/test-backend-ops.cpp @@ -51,13 +51,18 @@ static int64_t g_perf_duration_usec = 1000000; // default: 1 second in microseconds static int g_n_threads = -1; // -1 means use backend default (N_THREADS) +static bool g_test_shard_requested = false; +static size_t g_test_shard_index = 0; +static size_t g_test_shard_count = 1; +// Shard owning each matching case, indexed by occurrence (see assign_test_shards). +static std::vector g_test_shard_assignment; static void init_tensor_uniform(ggml_tensor * tensor, float min = -1.0f, float max = 1.0f) { size_t nels = ggml_nelements(tensor); std::vector data(nels); { // parallel initialization - static const size_t n_threads = N_THREADS; + static const size_t n_threads = g_n_threads > 0 ? (size_t) g_n_threads : N_THREADS; // static RNG initialization (revisit if n_threads stops being constant) static std::vector generators = []() { std::random_device rd; @@ -1300,10 +1305,49 @@ struct test_case { } } + // Estimated relative cost of one numerical case, used only to balance --test-shard + // partitions. Builds the same metadata-only graph as eval() and applies the same op + // filter; returns false when the case does not match. The estimate is the op's + // FLOP count plus a per-byte and per-case term, so FLOP-free ops still weigh in. + bool shard_cost(const char * op_names_filter, double & cost) { + mode = MODE_SUPPORT; // no sentinels: eval() rebuilds the graph from scratch + + ggml_init_params params = { + /* .mem_size = */ ggml_tensor_overhead()*128 + ggml_graph_overhead(), + /* .mem_base = */ NULL, + /* .no_alloc = */ true, + }; + ggml_context * ctx = ggml_init(params); + GGML_ASSERT(ctx); + gf = ggml_new_graph(ctx); + + ggml_tensor * out = build_graph(ctx); + const bool match = matches_filter(out, op_names_filter); + if (match) { + ggml_build_forward_expand(gf, out); + double bytes = 0.0; + for (int i = 0; i < ggml_graph_n_nodes(gf); ++i) { + const ggml_tensor * node = ggml_graph_node(gf, i); + bytes += (double) ggml_nbytes(node); + for (int j = 0; j < GGML_MAX_SRC; ++j) { + if (node->src[j]) { + bytes += (double) ggml_nbytes(node->src[j]); + } + } + } + cost = (double) op_flops(out) + 16.0*bytes + 1e7; + } + ggml_free(ctx); + gf = nullptr; + return match; + } + test_status_t eval(ggml_backend_t backend1, ggml_backend_t backend2, const char * op_names_filter, - printer * output_printer) { + printer * output_printer, + size_t & matching_cases, + size_t & selected_cases) { mode = MODE_TEST; ggml_init_params params = { @@ -1329,6 +1373,21 @@ struct test_case { return test_status_t::SKIPPED; } + // Partition after the normal op filter, before capability checks or execution. + // Whole-graph cases and duplicate parameter strings retain distinct indices. + const size_t case_index = matching_cases++; + size_t case_shard = 0; + if (g_test_shard_requested) { + GGML_ASSERT(case_index < g_test_shard_assignment.size()); + case_shard = g_test_shard_assignment[case_index]; + printf("Shard inventory %zu -> %zu: %s(%s)\n", case_index, case_shard, current_op_name.c_str(), vars().c_str()); + } + if (case_shard != g_test_shard_index) { + ggml_free(ctx); + return test_status_t::SKIPPED; + } + selected_cases++; + // check if the backends support the ops bool supported = true; for (ggml_backend_t backend : {backend1, backend2}) { @@ -10002,6 +10061,32 @@ static std::vector> make_test_cases_from_file(const c return test_cases; } +// Deterministic cost-balanced partition of the matching cases for --test-shard: longest +// estimated case first, each to the currently lightest shard (ties: lower case index, +// lower shard). Every worker computes the same assignment from the same inventory, and +// each occurrence (including duplicates and whole-graph cases) belongs to one shard. +static void assign_test_shards(std::vector> & test_cases, const char * op_names_filter) { + std::vector costs; + for (auto & test : test_cases) { + double cost = 0.0; + if (test->shard_cost(op_names_filter, cost)) { + costs.push_back(cost); + } + } + std::vector order(costs.size()); + for (size_t i = 0; i < order.size(); ++i) { + order[i] = i; + } + std::stable_sort(order.begin(), order.end(), [&](size_t a, size_t b) { return costs[a] > costs[b]; }); + std::vector load(g_test_shard_count, 0.0); + g_test_shard_assignment.assign(costs.size(), 0); + for (size_t i : order) { + const size_t shard = (size_t) (std::min_element(load.begin(), load.end()) - load.begin()); + g_test_shard_assignment[i] = shard; + load[shard] += costs[i]; + } +} + static bool test_backend(ggml_backend_t backend, test_mode mode, const char * op_names_filter, const char * params_filter, printer * output_printer, const char * test_file_path) { auto filter_test_cases = [](std::vector> & test_cases, const char * params_filter) { @@ -10051,16 +10136,27 @@ static bool test_backend(ggml_backend_t backend, test_mode mode, const char * op // Use reference implementation on the CPU backend for comparison using ggml_backend_cpu_set_use_ref_t = void (*)(ggml_backend_t, bool); auto * reg = ggml_backend_dev_backend_reg(ggml_backend_get_device(backend_cpu)); + if (g_n_threads > 0) { + auto * set_n_threads = (ggml_backend_set_n_threads_t) ggml_backend_reg_get_proc_address(reg, "ggml_backend_set_n_threads"); + GGML_ASSERT(set_n_threads != nullptr); + set_n_threads(backend_cpu, g_n_threads); + } auto * set_use_ref = (ggml_backend_cpu_set_use_ref_t) ggml_backend_reg_get_proc_address(reg, "ggml_backend_cpu_set_use_ref"); if (set_use_ref) { set_use_ref(backend_cpu, true); } + if (g_test_shard_requested) { + assign_test_shards(test_cases, op_names_filter); + } + size_t n_ok = 0; size_t tests_run = 0; + size_t matching_cases = 0; + size_t selected_cases = 0; std::vector failed_tests; for (auto & test : test_cases) { - test_status_t status = test->eval(backend, backend_cpu, op_names_filter, output_printer); + test_status_t status = test->eval(backend, backend_cpu, op_names_filter, output_printer, matching_cases, selected_cases); if (status == test_status_t::SKIPPED || status == test_status_t::NOT_SUPPORTED) { continue; } @@ -10071,6 +10167,11 @@ static bool test_backend(ggml_backend_t backend, test_mode mode, const char * op failed_tests.push_back(test->current_op_name + "(" + test->vars() + ")"); } } + if (g_test_shard_requested) { + GGML_ASSERT(matching_cases == g_test_shard_assignment.size()); + printf("Shard %zu/%zu: %zu selected of %zu matching cases\n", + g_test_shard_index, g_test_shard_count, selected_cases, matching_cases); + } output_printer->print_summary(test_summary_info(n_ok, tests_run, false)); output_printer->print_failed_tests(failed_tests); @@ -10218,7 +10319,7 @@ static void show_test_coverage() { static void usage(char ** argv) { printf("Usage: %s [mode] [-o ] [-b ] [-p ] [--output ] [--list-ops]", argv[0]); - printf(" [--show-coverage] [--test-file ] [--perf-duration ] [--threads ]\n"); + printf(" [--show-coverage] [--test-file ] [--perf-duration ] [--threads ] [--test-shard ]\n"); printf(" valid modes:\n"); printf(" - test (default, compare with CPU backend for correctness)\n"); printf(" - grad (compare gradients from backpropagation with method of finite differences)\n"); @@ -10308,6 +10409,21 @@ int main(int argc, char ** argv) { usage(argv); return 1; } + } else if (strcmp(argv[i], "--test-shard") == 0) { + std::cmatch parts; + if (g_test_shard_requested || i + 1 >= argc || + !std::regex_match(argv[i + 1], parts, std::regex("([0-9]{1,3})/([0-9]{1,3})"))) { + fprintf(stderr, "error: --test-shard requires one zero-based index/count pair\n"); + return 1; + } + g_test_shard_index = std::stoul(parts[1].str()); + g_test_shard_count = std::stoul(parts[2].str()); + if (g_test_shard_count == 0 || g_test_shard_count > 256 || g_test_shard_index >= g_test_shard_count) { + fprintf(stderr, "error: --test-shard requires index < count and count in 1..256\n"); + return 1; + } + g_test_shard_requested = true; + ++i; } else if (strcmp(argv[i], "--threads") == 0 || strcmp(argv[i], "-t") == 0) { if (i + 1 < argc) { g_n_threads = atoi(argv[++i]); @@ -10326,6 +10442,11 @@ int main(int argc, char ** argv) { } } + if (g_test_shard_requested && (mode != MODE_TEST || output_format != output_formats::CONSOLE)) { + fprintf(stderr, "error: --test-shard requires test mode and console output\n"); + return 1; + } + // load and enumerate backends ggml_backend_load_all();