diff --git a/src/code/issue4/bitwise_diagnostics/.gitignore b/src/code/issue4/bitwise_diagnostics/.gitignore new file mode 100644 index 0000000..3717294 --- /dev/null +++ b/src/code/issue4/bitwise_diagnostics/.gitignore @@ -0,0 +1,3 @@ +__pycache__/ +diagnostic-output/ +sweep-output/ diff --git a/src/code/issue4/bitwise_diagnostics/EXPERIMENT.md b/src/code/issue4/bitwise_diagnostics/EXPERIMENT.md new file mode 100644 index 0000000..f56a7d4 --- /dev/null +++ b/src/code/issue4/bitwise_diagnostics/EXPERIMENT.md @@ -0,0 +1,100 @@ +# Four-GPU reproducibility experiment + +Date: 2026-07-23 + +The machine-readable summary is +[`results/autodl-4x3090.json`](results/autodl-4x3090.json). + +## Environment + +- 4 × NVIDIA GeForce RTX 3090 24 GiB, driver 580.105.08 +- PyTorch 2.8.0+cu128, CUDA 12.8, NCCL 2.27.3 +- GPU 0↔1 and GPU 2↔3: `PXB`; traffic between the pairs: `SYS` +- two NUMA nodes; no NVLink reported by `nvidia-smi topo -m` +- fixed seed 2026 and rank-distinct float32 inputs + +Each reported run is a fresh four-rank `torchrun --standalone` launch. This +matters: repeated calls inside one communicator do not test communicator +initialization or run-to-run selection. + +## Same-configuration results + +At 1 MiB input per rank, AllReduce covered nine +`NCCL_ALGO × NCCL_PROTO` configurations: + +| Algorithm | Protocols | Independent runs | Calls/run | Result | +|---|---|---:|---:|---| +| automatic | automatic, Simple, LL | 5 each | 20 | bitwise identical | +| Ring | automatic, Simple, LL | 5 each | 20 | bitwise identical | +| Tree | automatic, Simple, LL | 5 each | 20 | bitwise identical | + +Reduce-Scatter used the same input size and repetition count: + +| Algorithm | Protocols | Result | +|---|---|---| +| automatic | automatic, Simple, LL | bitwise identical | +| Ring | automatic, Simple, LL | bitwise identical | +| Tree | automatic, Simple, LL | rejected with `ncclInvalidUsage` | + +Tree is therefore not presented as a Reduce-Scatter mitigation on this NCCL +version. Importantly, the runner did not silently fall back to Ring. + +No same-configuration run-to-run divergence was observed. This is a bounded +negative result for this exact software and PCIe topology, not a universal +claim that these algorithms are deterministic. + +## Granularity + +NCCL automatic selection was also checked across message sizes: + +| Bytes/rank | Independent runs | Calls/run | Result | +|---:|---:|---:|---| +| 1 KiB | 5 | 50 | bitwise identical | +| 64 KiB | 5 | 30 | bitwise identical | +| 1 MiB | 5 | 20 | bitwise identical | +| 16 MiB | 5 | 3 | bitwise identical | + +The call count is lower for large messages because the diagnostic retains raw +bytes until comparison. Every row still compares independent launches with +the same call count within the row. + +## Reproducible algorithm-change case + +To verify localization and demonstrate the numerical consequence of reduction +order, one factor was changed from Ring to Tree while seed, rank inputs, +hardware, dtype, message size, and call sequence stayed fixed. + +The first difference appeared at call 0, rank 0, byte 24 (float32 element 6). +For that 1 MiB output: + +- 94,656 bytes / 192,767 bits changed; +- maximum absolute error was `9.5367431640625e-07`; +- maximum reported ULP distance was 49,152; +- Ring SHA-256: + `c5797c4ea5b9d3a387fab7d4e3a6405fa60f3397d19a050070e896433b99d513`; +- Tree SHA-256: + `3f3470571b53caf023710336ca21a493c7a27e27e40871280a5c85a44d44a991`. + +This is a controlled configuration-change case, **not** evidence of +nondeterminism within either fixed configuration. It demonstrates why +algorithm/topology selection must remain stable when bitwise continuity +between jobs is required. + +## Conclusions by acceptance dimension + +1. **Granularity:** no run-to-run difference was observed from 1 KiB through + 16 MiB under automatic selection. Larger captures cost proportionally more + memory; sampling call count is an explicit experimental trade-off. +2. **Hardware:** results apply to a four-GPU, dual-NUMA PCIe host. The `SYS` + boundary between GPU pairs is materially different from NVLink/NVSwitch; + NVLS was therefore not tested or recommended. +3. **Software flow:** fresh process groups, fixed rank mapping, fixed inputs, + and explicit algorithm/protocol settings produced stable bytes. Switching + Ring to Tree immediately changed results. For reproducible jobs, pin the + tested software stack and selection controls, but only use algorithms that + support the target collective. + +The evidence supports fixed Ring (including Simple or LL on this host) as a +measured reproducible option for both tested collectives. This is deliberately +scoped to the recorded environment rather than framed as a universal NCCL +guarantee. diff --git a/src/code/issue4/bitwise_diagnostics/README.md b/src/code/issue4/bitwise_diagnostics/README.md new file mode 100644 index 0000000..b96f6ef --- /dev/null +++ b/src/code/issue4/bitwise_diagnostics/README.md @@ -0,0 +1,121 @@ +# NCCL bitwise reproducibility diagnostics + +This tool launches the same collective workload in **separate `torchrun` +process groups**, captures every output as raw bytes, and compares all ranks +and calls against the first run. A JSON report identifies the first divergent +call, rank, byte/element offset, changed bits, absolute error, and ULP error. + +Unlike an arithmetic simulation, this exercises the installed PyTorch, NCCL, +CUDA runtime, topology discovery, and actual communication path. + +## Requirements + +- Linux, Python 3.10+, PyTorch with NCCL, and at least two CUDA GPUs +- All tested ranks must see the same GPUs in every independent run +- For multi-node experiments, invoke `worker.py` with the site's normal + rendezvous command and compare its capture files with `core.compare_runs` + +No claim of determinism is inferred from an algorithm name. The report only +describes the measured hardware/software configuration. + +See [`EXPERIMENT.md`](EXPERIMENT.md) for measured four-GPU results covering +algorithm/protocol selection, message granularity, both collectives, and a +controlled Ring-versus-Tree difference. + +## Quick start + +```bash +cd src/code/issue4/bitwise_diagnostics + +# NCCL automatic selection +python diagnose.py --nproc-per-node 8 --runs 5 \ + --op all_reduce --elements 1048576 --calls 20 \ + --output-dir results/default + +# Pin one supported algorithm/protocol for an A/B comparison +python diagnose.py --nproc-per-node 8 --runs 5 \ + --algo Ring --proto Simple --op all_reduce \ + --elements 1048576 --calls 20 --output-dir results/ring-simple + +# Unified A/B matrix; unsupported combinations are recorded, not hidden +python sweep.py --nproc-per-node 8 --runs 5 \ + --algos default,Ring,Tree --protos default,Simple,LL +``` + +Exit status is `0` for bitwise-identical runs, `2` when a divergence is +detected, and non-zero on invalid configuration or launch failure. Use +`--keep-payloads` when forensic inspection is needed; otherwise multi-MB raw +captures are deleted after `report.json` is written. + +Existing captures can also be compared offline (including captures copied +from different nodes): + +```bash +python core.py run-a.json run-b.json --output comparison.json +``` + +Run CPU-only unit tests with: + +```bash +python -m unittest -v test_core.py +``` + +## Controlled experiment matrix + +Change one factor at a time and retain `report.json` for each row: + +1. **Message granularity:** 1 KiB, 64 KiB, 1 MiB, 16 MiB, 128 MiB. Since + raw captures scale as `elements × dtype bytes × calls × ranks × runs`, + reduce `--calls` for the largest cases and keep the value identical across + compared configurations. +2. **Collective:** `all_reduce` and `reduce_scatter`. +3. **Selection:** NCCL default, then supported `NCCL_ALGO` values with + `Simple`, `LL`, and (only on supported platforms) `LL128`. +4. **Resources:** record GPU model/count, NVLink/NVSwitch, NIC, node count, + PyTorch/CUDA/NCCL versions, and topology. Worker metadata records the + software versions and effective experiment overrides automatically. +5. **Stability:** at least five independent launches and 50 calls per launch. + +Do not force unsupported combinations. NCCL 2.24+ fails on invalid algorithm +tokens, and NVIDIA warns that forcing LL128 on unsupported platforms can cause +data corruption. `NCCL_ALGO` and `NCCL_PROTO` are diagnostic controls, not +universal production recommendations. + +## Interpreting results across the three acceptance dimensions + +- **Granularity:** small messages emphasize launch/protocol behavior; large + messages exercise more chunks/channels and expose more reduction sites. + Compare divergence rate and first-call position alongside latency measured + by a dedicated benchmark such as `nccl-tests`. +- **Hardware resources:** rank count and topology determine valid algorithms + and reduction paths. A result on NVSwitch must not be generalized to PCIe or + multi-node fabrics. A pinned `NCCL_TOPO_FILE` is useful only when it + accurately represents the tested system. +- **Software flow:** fresh process groups test run-to-run reproducibility, + which repeated calls inside one communicator cannot establish. Pin package + versions and seeds; serialize collectives in the same order on every rank. + PyTorch deterministic-algorithm settings and + `CUBLAS_WORKSPACE_CONFIG=:4096:8` matter for surrounding compute, but they do + not constitute evidence that a collective is bitwise reproducible. + +## Reproducible non-determinism case + +The tool deliberately does not fabricate nondeterminism. To document a case: + +1. run the default selection matrix on the target cluster; +2. retain the first report with `bitwise_identical=false`; +3. repeat with one variable pinned at a time; +4. report a mitigation only if repeated measurements turn identical; +5. attach NCCL `INFO` logs to show the selected algorithm/topology. + +This separates observed evidence from assumptions about Ring, Tree, PAT, or +NVLS internals and avoids recommending undocumented environment variables. + +## References + +- NVIDIA NCCL environment variables: + https://docs.nvidia.com/deeplearning/nccl/user-guide/docs/env.html +- PyTorch reproducibility: + https://docs.pytorch.org/docs/stable/notes/randomness.html +- PyTorch deterministic algorithms: + https://docs.pytorch.org/docs/stable/generated/torch.use_deterministic_algorithms.html diff --git a/src/code/issue4/bitwise_diagnostics/core.py b/src/code/issue4/bitwise_diagnostics/core.py new file mode 100644 index 0000000..cc27ed3 --- /dev/null +++ b/src/code/issue4/bitwise_diagnostics/core.py @@ -0,0 +1,184 @@ +"""Pure-Python comparison and reporting primitives for NCCL run captures.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import math +import struct +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Any, Iterable + + +SCHEMA_VERSION = 1 + + +@dataclass(frozen=True) +class TensorCapture: + call: int + rank: int + dtype: str + shape: list[int] + payload_hex: str + + @property + def payload(self) -> bytes: + return bytes.fromhex(self.payload_hex) + + @property + def sha256(self) -> str: + return hashlib.sha256(self.payload).hexdigest() + + +@dataclass(frozen=True) +class Difference: + run: int + call: int + rank: int + first_byte: int + first_element: int + changed_bytes: int + changed_bits: int + max_abs_error: float | None + max_ulp_error: int | None + baseline_sha256: str + candidate_sha256: str + + +def _float_format(dtype: str) -> tuple[str, str, int] | None: + return { + "torch.float16": ("e", "H", 2), + "torch.float32": ("f", "I", 4), + "torch.float64": ("d", "Q", 8), + }.get(dtype) + + +def _ordered_int(bits: int, width: int) -> int: + """Map IEEE sign-magnitude bit patterns to monotonically ordered integers.""" + sign = 1 << (width * 8 - 1) + mask = (1 << (width * 8)) - 1 + return (~bits & mask) if bits & sign else (bits | sign) + + +def compare_capture( + baseline: TensorCapture, candidate: TensorCapture, run: int +) -> Difference | None: + if (baseline.dtype, baseline.shape) != (candidate.dtype, candidate.shape): + raise ValueError( + f"capture metadata changed at call={baseline.call}, rank={baseline.rank}" + ) + left, right = baseline.payload, candidate.payload + if len(left) != len(right): + raise ValueError("capture payload lengths differ") + if left == right: + return None + + xor = bytes(a ^ b for a, b in zip(left, right)) + first_byte = next(i for i, byte in enumerate(xor) if byte) + changed_bytes = sum(bool(byte) for byte in xor) + changed_bits = sum(byte.bit_count() for byte in xor) + fmt = _float_format(baseline.dtype) + max_abs: float | None = None + max_ulp: int | None = None + element_size = fmt[2] if fmt else 1 + + if fmt: + float_code, int_code, width = fmt + count = len(left) // width + left_values = struct.unpack(f"<{count}{float_code}", left) + right_values = struct.unpack(f"<{count}{float_code}", right) + left_bits = struct.unpack(f"<{count}{int_code}", left) + right_bits = struct.unpack(f"<{count}{int_code}", right) + abs_errors = [ + abs(a - b) + for a, b in zip(left_values, right_values) + if not (math.isnan(a) and math.isnan(b)) + ] + max_abs = max(abs_errors, default=0.0) + max_ulp = max( + abs(_ordered_int(a, width) - _ordered_int(b, width)) + for a, b in zip(left_bits, right_bits) + ) + + return Difference( + run=run, + call=baseline.call, + rank=baseline.rank, + first_byte=first_byte, + first_element=first_byte // element_size, + changed_bytes=changed_bytes, + changed_bits=changed_bits, + max_abs_error=max_abs, + max_ulp_error=max_ulp, + baseline_sha256=baseline.sha256, + candidate_sha256=candidate.sha256, + ) + + +def load_capture(path: Path) -> dict[str, Any]: + data = json.loads(path.read_text(encoding="utf-8")) + if data.get("schema_version") != SCHEMA_VERSION: + raise ValueError(f"unsupported capture schema in {path}") + return data + + +def compare_runs(run_paths: Iterable[Path]) -> dict[str, Any]: + paths = list(run_paths) + if len(paths) < 2: + raise ValueError("at least two independent runs are required") + runs = [load_capture(path) for path in paths] + keys = ("op", "dtype", "elements", "calls", "world_size", "seed") + baseline_meta = {key: runs[0]["metadata"][key] for key in keys} + for index, run in enumerate(runs[1:], 1): + candidate_meta = {key: run["metadata"][key] for key in keys} + if candidate_meta != baseline_meta: + raise ValueError(f"run {index} is not comparable to the baseline") + + baseline = { + (item["call"], item["rank"]): TensorCapture(**item) + for item in runs[0]["captures"] + } + differences: list[Difference] = [] + for run_index, run in enumerate(runs[1:], 1): + candidate = { + (item["call"], item["rank"]): TensorCapture(**item) + for item in run["captures"] + } + if candidate.keys() != baseline.keys(): + raise ValueError(f"run {run_index} has an incomplete capture set") + for key in sorted(baseline): + difference = compare_capture(baseline[key], candidate[key], run_index) + if difference: + differences.append(difference) + + first = min(differences, key=lambda item: (item.call, item.run, item.rank), default=None) + return { + "schema_version": SCHEMA_VERSION, + # Preserve the complete measured environment in the report while only + # using workload-defining fields above to decide comparability. + "metadata": runs[0]["metadata"], + "run_files": [str(path) for path in paths], + "bitwise_identical": not differences, + "first_divergence": asdict(first) if first else None, + "differences": [asdict(item) for item in differences], + } + + +def main() -> int: + parser = argparse.ArgumentParser(description="Compare existing NCCL capture files") + parser.add_argument("captures", type=Path, nargs="+") + parser.add_argument("--output", type=Path) + args = parser.parse_args() + report = compare_runs(args.captures) + rendered = json.dumps(report, indent=2) + if args.output: + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(rendered, encoding="utf-8") + print(rendered) + return 0 if report["bitwise_identical"] else 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/code/issue4/bitwise_diagnostics/diagnose.py b/src/code/issue4/bitwise_diagnostics/diagnose.py new file mode 100644 index 0000000..214973c --- /dev/null +++ b/src/code/issue4/bitwise_diagnostics/diagnose.py @@ -0,0 +1,86 @@ +"""Run independent NCCL jobs and compare their output bytes.""" + +from __future__ import annotations + +import argparse +import json +import os +import subprocess +import sys +from pathlib import Path + +from core import compare_runs + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--nproc-per-node", type=int, required=True) + parser.add_argument("--runs", type=int, default=3) + parser.add_argument("--op", choices=("all_reduce", "reduce_scatter"), default="all_reduce") + parser.add_argument("--elements", type=int, default=1 << 18) + parser.add_argument("--calls", type=int, default=10) + parser.add_argument("--dtype", choices=("float16", "float32", "float64"), default="float32") + parser.add_argument("--seed", type=int, default=2026) + parser.add_argument("--algo", help="NCCL_ALGO value, e.g. Ring or Tree") + parser.add_argument("--proto", help="NCCL_PROTO value, e.g. Simple, LL, or LL128") + parser.add_argument("--topo-file", type=Path) + parser.add_argument("--output-dir", type=Path, default=Path("diagnostic-output")) + parser.add_argument("--keep-payloads", action="store_true") + return parser.parse_args() + + +def main() -> int: + args = parse_args() + if args.runs < 2 or args.nproc_per_node < 2: + raise ValueError("--runs and --nproc-per-node must both be at least 2") + args.output_dir.mkdir(parents=True, exist_ok=True) + worker = Path(__file__).with_name("worker.py") + captures: list[Path] = [] + env = os.environ.copy() + for key, value in ( + ("NCCL_ALGO", args.algo), + ("NCCL_PROTO", args.proto), + ("NCCL_TOPO_FILE", str(args.topo_file.resolve()) if args.topo_file else None), + ): + if value: + env[key] = value + else: + env.pop(key, None) + + for run in range(args.runs): + capture = args.output_dir / f"run-{run:02d}.json" + cmd = [ + "torchrun", + "--standalone", + f"--nproc-per-node={args.nproc_per_node}", + str(worker), + "--output", + str(capture), + "--op", + args.op, + "--elements", + str(args.elements), + "--calls", + str(args.calls), + "--dtype", + args.dtype, + "--seed", + str(args.seed), + ] + print(f"[run {run + 1}/{args.runs}] {' '.join(cmd)}", flush=True) + subprocess.run(cmd, env=env, check=True) + captures.append(capture) + + report = compare_runs(captures) + report_path = args.output_dir / "report.json" + report_path.write_text(json.dumps(report, indent=2), encoding="utf-8") + print(json.dumps(report["first_divergence"], indent=2)) + print(f"bitwise_identical={report['bitwise_identical']} report={report_path}") + if not args.keep_payloads: + for capture in captures: + capture.unlink() + return 0 if report["bitwise_identical"] else 2 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/code/issue4/bitwise_diagnostics/results/autodl-4x3090.json b/src/code/issue4/bitwise_diagnostics/results/autodl-4x3090.json new file mode 100644 index 0000000..6a08877 --- /dev/null +++ b/src/code/issue4/bitwise_diagnostics/results/autodl-4x3090.json @@ -0,0 +1,106 @@ +{ + "schema_version": 1, + "date": "2026-07-23", + "environment": { + "gpu": "4 x NVIDIA GeForce RTX 3090 24 GiB", + "driver": "580.105.08", + "torch": "2.8.0+cu128", + "cuda": "12.8", + "nccl": "2.27.3", + "topology": { + "gpu0_gpu1": "PXB", + "gpu2_gpu3": "PXB", + "between_pairs": "SYS", + "numa_nodes": 2, + "nvlink": false + } + }, + "same_config_results": { + "all_reduce_1mib": { + "runs_per_config": 5, + "calls_per_run": 20, + "world_size": 4, + "dtype": "float32", + "configs_tested": [ + "default+default", + "default+Simple", + "default+LL", + "Ring+default", + "Ring+Simple", + "Ring+LL", + "Tree+default", + "Tree+Simple", + "Tree+LL" + ], + "bitwise_identical_configs": 9, + "divergent_configs": 0 + }, + "reduce_scatter_1mib_input": { + "runs_per_config": 5, + "calls_per_run": 20, + "world_size": 4, + "dtype": "float32", + "bitwise_identical_configs": [ + "default+default", + "default+Simple", + "default+LL", + "Ring+default", + "Ring+Simple", + "Ring+LL" + ], + "unsupported_configs": [ + "Tree+default", + "Tree+Simple", + "Tree+LL" + ], + "unsupported_error": "ncclInvalidUsage" + }, + "all_reduce_granularity": [ + { + "bytes_per_rank": 1024, + "runs": 5, + "calls_per_run": 50, + "bitwise_identical": true + }, + { + "bytes_per_rank": 65536, + "runs": 5, + "calls_per_run": 30, + "bitwise_identical": true + }, + { + "bytes_per_rank": 1048576, + "runs": 5, + "calls_per_run": 20, + "bitwise_identical": true + }, + { + "bytes_per_rank": 16777216, + "runs": 5, + "calls_per_run": 3, + "bitwise_identical": true + } + ] + }, + "controlled_algorithm_change": { + "baseline": "Ring", + "candidate": "Tree", + "op": "all_reduce", + "bytes_per_rank": 1048576, + "calls": 20, + "controlled_factors": "same seed, rank inputs, hardware, dtype and message size", + "bitwise_identical": false, + "first_divergence": { + "call": 0, + "rank": 0, + "first_byte": 24, + "first_element": 6, + "changed_bytes": 94656, + "changed_bits": 192767, + "max_abs_error": 9.5367431640625e-07, + "max_ulp_error": 49152, + "ring_sha256": "c5797c4ea5b9d3a387fab7d4e3a6405fa60f3397d19a050070e896433b99d513", + "tree_sha256": "3f3470571b53caf023710336ca21a493c7a27e27e40871280a5c85a44d44a991" + } + } +} diff --git a/src/code/issue4/bitwise_diagnostics/sweep.py b/src/code/issue4/bitwise_diagnostics/sweep.py new file mode 100644 index 0000000..18b1db6 --- /dev/null +++ b/src/code/issue4/bitwise_diagnostics/sweep.py @@ -0,0 +1,95 @@ +"""Execute a controlled NCCL algorithm/protocol matrix and summarize it.""" + +from __future__ import annotations + +import argparse +import json +import os +import subprocess +import sys +from pathlib import Path + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--nproc-per-node", type=int, required=True) + parser.add_argument("--runs", type=int, default=3) + parser.add_argument("--op", choices=("all_reduce", "reduce_scatter"), default="all_reduce") + parser.add_argument("--elements", type=int, default=1 << 18) + parser.add_argument("--calls", type=int, default=10) + parser.add_argument("--dtype", choices=("float16", "float32", "float64"), default="float32") + parser.add_argument("--algos", default="default,Ring,Tree") + parser.add_argument("--protos", default="default,Simple,LL") + parser.add_argument("--output-dir", type=Path, default=Path("sweep-output")) + return parser.parse_args() + + +def main() -> int: + args = parse_args() + driver = Path(__file__).with_name("diagnose.py") + args.output_dir.mkdir(parents=True, exist_ok=True) + rows: list[dict[str, object]] = [] + for algo in args.algos.split(","): + for proto in args.protos.split(","): + # "default + explicit" duplicates an explicit override because the + # unset dimension remains under NCCL's automatic selection. + slug = f"{algo.lower()}-{proto.lower()}" + output = args.output_dir / slug + cmd = [ + sys.executable, + str(driver), + "--nproc-per-node", + str(args.nproc_per_node), + "--runs", + str(args.runs), + "--op", + args.op, + "--elements", + str(args.elements), + "--calls", + str(args.calls), + "--dtype", + args.dtype, + "--output-dir", + str(output), + ] + if algo != "default": + cmd.extend(("--algo", algo)) + if proto != "default": + cmd.extend(("--proto", proto)) + print(f"\n=== ALGO={algo} PROTO={proto} ===", flush=True) + result = subprocess.run(cmd, env=os.environ.copy()) + report_path = output / "report.json" + row: dict[str, object] = { + "algo": algo, + "proto": proto, + "exit_code": result.returncode, + "report": str(report_path), + } + if report_path.exists(): + report = json.loads(report_path.read_text(encoding="utf-8")) + row["bitwise_identical"] = report["bitwise_identical"] + row["first_divergence"] = report["first_divergence"] + else: + row["error"] = "launch failed; inspect console/NCCL logs" + rows.append(row) + + summary = { + "experiment": { + "op": args.op, + "elements": args.elements, + "calls": args.calls, + "runs": args.runs, + "world_size": args.nproc_per_node, + "dtype": args.dtype, + }, + "results": rows, + } + summary_path = args.output_dir / "summary.json" + summary_path.write_text(json.dumps(summary, indent=2), encoding="utf-8") + print(f"\nsummary={summary_path}") + return 1 if any("error" in row for row in rows) else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/code/issue4/bitwise_diagnostics/test_core.py b/src/code/issue4/bitwise_diagnostics/test_core.py new file mode 100644 index 0000000..a8a6b01 --- /dev/null +++ b/src/code/issue4/bitwise_diagnostics/test_core.py @@ -0,0 +1,59 @@ +import json +import struct +import tempfile +import unittest +from pathlib import Path + +from core import TensorCapture, compare_capture, compare_runs + + +def capture(values, call=0, rank=0): + return TensorCapture( + call=call, + rank=rank, + dtype="torch.float32", + shape=[len(values)], + payload_hex=struct.pack(f"<{len(values)}f", *values).hex(), + ) + + +class CompareCaptureTests(unittest.TestCase): + def test_identical_payload(self): + self.assertIsNone(compare_capture(capture([1.0]), capture([1.0]), 1)) + + def test_reports_first_offset_and_ulp(self): + result = compare_capture(capture([1.0, 2.0]), capture([1.0, 2.000000238418579]), 1) + self.assertEqual(result.first_byte, 4) + self.assertEqual(result.first_element, 1) + self.assertEqual(result.max_ulp_error, 1) + self.assertGreater(result.changed_bits, 0) + + def test_negative_float_ulp_ordering(self): + result = compare_capture(capture([-1.0]), capture([-1.0000001192092896]), 1) + self.assertEqual(result.max_ulp_error, 1) + + +class CompareRunsTests(unittest.TestCase): + def test_localizes_first_call(self): + with tempfile.TemporaryDirectory() as directory: + paths = [] + for run, changed in enumerate((False, True)): + items = [capture([1.0], call=0), capture([2.0 if not changed else 3.0], call=1)] + data = { + "schema_version": 1, + "metadata": { + "op": "all_reduce", "dtype": "torch.float32", "elements": 1, + "calls": 2, "world_size": 1, "seed": 1, + }, + "captures": [item.__dict__ for item in items], + } + path = Path(directory) / f"{run}.json" + path.write_text(json.dumps(data), encoding="utf-8") + paths.append(path) + report = compare_runs(paths) + self.assertFalse(report["bitwise_identical"]) + self.assertEqual(report["first_divergence"]["call"], 1) + + +if __name__ == "__main__": + unittest.main() diff --git a/src/code/issue4/bitwise_diagnostics/worker.py b/src/code/issue4/bitwise_diagnostics/worker.py new file mode 100644 index 0000000..4eb6237 --- /dev/null +++ b/src/code/issue4/bitwise_diagnostics/worker.py @@ -0,0 +1,117 @@ +"""One independent distributed NCCL capture. Launch with torchrun.""" + +from __future__ import annotations + +import argparse +import json +import os +import platform +from pathlib import Path + +import torch +import torch.distributed as dist + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser() + parser.add_argument("--output", type=Path, required=True) + parser.add_argument("--op", choices=("all_reduce", "reduce_scatter"), default="all_reduce") + parser.add_argument("--elements", type=int, default=1 << 18) + parser.add_argument("--calls", type=int, default=10) + parser.add_argument("--dtype", choices=("float16", "float32", "float64"), default="float32") + parser.add_argument("--seed", type=int, default=2026) + return parser.parse_args() + + +def nccl_version_string() -> str: + version = torch.cuda.nccl.version() + if isinstance(version, tuple): + return ".".join(map(str, version)) + return str(version) + + +def main() -> None: + args = parse_args() + if args.elements <= 0 or args.calls <= 0: + raise ValueError("--elements and --calls must be positive") + rank = int(os.environ["RANK"]) + local_rank = int(os.environ["LOCAL_RANK"]) + world_size = int(os.environ["WORLD_SIZE"]) + if args.op == "reduce_scatter" and args.elements % world_size: + raise ValueError("--elements must be divisible by world size for reduce_scatter") + + torch.cuda.set_device(local_rank) + dist.init_process_group("nccl", device_id=torch.device("cuda", local_rank)) + # Keep capture transport off NCCL. A forced NCCL_ALGO may be valid for the + # collective under test but invalid for gather_object's internal AllGather + # (for example, Tree). A Gloo control group prevents that instrumentation + # traffic from changing or invalidating the experiment. + control_group = dist.new_group(backend="gloo") + dtype = getattr(torch, args.dtype) + generator = torch.Generator(device="cpu").manual_seed(args.seed + rank) + source = torch.randn(args.elements, dtype=dtype, generator=generator).cuda(local_rank) + captures: list[dict[str, object]] = [] + + for call in range(args.calls): + # Preserve identical per-rank inputs across independent launches while + # changing data between calls so the first divergent call is meaningful. + input_tensor = source + call * torch.finfo(dtype).eps + if args.op == "all_reduce": + output = input_tensor.clone() + dist.all_reduce(output) + else: + output = torch.empty(args.elements // world_size, dtype=dtype, device=local_rank) + dist.reduce_scatter_tensor(output, input_tensor.contiguous()) + torch.cuda.synchronize() + raw = output.detach().cpu().contiguous().view(torch.uint8).numpy().tobytes() + captures.append( + { + "call": call, + "rank": rank, + "dtype": str(dtype), + "shape": list(output.shape), + "payload_hex": raw.hex(), + } + ) + + gathered: list[list[dict[str, object]] | None] | None = ( + [None] * world_size if rank == 0 else None + ) + dist.gather_object(captures, gathered, dst=0, group=control_group) + if rank == 0: + metadata = { + "op": args.op, + "dtype": str(dtype), + "elements": args.elements, + "calls": args.calls, + "world_size": world_size, + "seed": args.seed, + "torch_version": torch.__version__, + "cuda_version": torch.version.cuda, + "nccl_version": nccl_version_string(), + "gpu": torch.cuda.get_device_name(local_rank), + "host": platform.node(), + "nccl_env": { + key: os.environ[key] + for key in ("NCCL_ALGO", "NCCL_PROTO", "NCCL_TOPO_FILE") + if key in os.environ + }, + } + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text( + json.dumps( + { + "schema_version": 1, + "metadata": metadata, + "captures": [item for rank_items in gathered or [] for item in rank_items or []], + }, + indent=2, + ), + encoding="utf-8", + ) + dist.destroy_process_group(control_group) + dist.destroy_process_group() + + +if __name__ == "__main__": + main()