From 17d7a8680e5184eb8b70b1af2bc10d6cbcd974d5 Mon Sep 17 00:00:00 2001 From: Donglai Dai Date: Fri, 18 Sep 2026 20:02:03 -0400 Subject: [PATCH] rccl-tests: share GIN-SDMA pytest harness (AICOMRCCL-2221) AllGather, AllToAll, and Broadcast were each carrying their own MPI launcher and connectivity-gate retry. One module keeps those from drifting, and the GPU-free guards now cover rocminfo failure and a data error that also looks like a connectivity abort. Co-authored-by: Cursor --- projects/rccl-tests/test/gin_sdma_harness.py | 396 ++++++++++++++++ projects/rccl-tests/test/test_AllGather.py | 141 ++---- projects/rccl-tests/test/test_AllToAll.py | 162 +++---- projects/rccl-tests/test/test_Broadcast.py | 458 ++++++------------- 4 files changed, 639 insertions(+), 518 deletions(-) create mode 100644 projects/rccl-tests/test/gin_sdma_harness.py diff --git a/projects/rccl-tests/test/gin_sdma_harness.py b/projects/rccl-tests/test/gin_sdma_harness.py new file mode 100644 index 000000000000..28ca7adad745 --- /dev/null +++ b/projects/rccl-tests/test/gin_sdma_harness.py @@ -0,0 +1,396 @@ +################################################################################# +# Copyright (c) 2026 Advanced Micro Devices, Inc. All rights reserved. +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell cop- +# ies of the Software, and to permit persons to whom the Software is furnished +# to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in all +# copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IM- +# PLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS +# FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR +# COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER +# IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNE- +# CTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. +################################################################################ + +"""Shared helpers for GIN-SDMA rccl-tests pytest harnesses (AllGather, AllToAll, +Broadcast). MPI launcher assembly, connectivity-gate retry, and (for Broadcast) +NCCL debug-file capture live here so the per-collective test modules stay thin.""" + +import glob +import os +import re +import shlex +import shutil +import signal +import subprocess +import tempfile +import time + +import pytest + +MiB = 1024 * 1024 +GiB = 1024 * MiB + +# Intermittent gfx950 cuMem-VMM connectivity-gate abort (not a data error). +CONN_GATE_RE = re.compile( + r"LSA signal connectivity gate failed|unhandled system error", re.I +) +DATA_FAIL_RE = re.compile( + r"#wrong\s*=\s*[1-9]|mismatch|check.*fail|Out of bounds values\s*:\s*[1-9]", + re.I, +) + + +def detect_ngpus(): + """Return the visible GPU count. + + Prefer ROCR_VISIBLE_DEVICES / HIP_VISIBLE_DEVICES. Otherwise require + rocminfo on PATH: a missing binary used to look like zero GPUs because + `rocminfo | grep | wc -l` still exits 0, and callers that clamp with + max(1, ...) then silently collect a one-GPU matrix. + """ + if os.environ.get("ROCR_VISIBLE_DEVICES") is not None: + return len(os.environ["ROCR_VISIBLE_DEVICES"].split(",")) + if os.environ.get("HIP_VISIBLE_DEVICES") is not None: + return len(os.environ["HIP_VISIBLE_DEVICES"].split(",")) + rocminfo = shutil.which("rocminfo") + if rocminfo is None: + raise RuntimeError( + "rocminfo not on PATH; cannot detect GPU count. Set " + "ROCR_VISIBLE_DEVICES, HIP_VISIBLE_DEVICES, or RCCL_TESTS_*_NP." + ) + try: + out = subprocess.check_output( + [rocminfo], universal_newlines=True, stderr=subprocess.DEVNULL + ) + except (OSError, subprocess.SubprocessError) as exc: + raise RuntimeError( + "rocminfo failed; cannot detect GPU count. Set " + "ROCR_VISIBLE_DEVICES, HIP_VISIBLE_DEVICES, or RCCL_TESTS_*_NP." + ) from exc + return sum( + 1 for line in out.splitlines() if re.search(r"Device Type:\s*GPU", line) + ) + + +def env_int(name, default): + """Parse an integer env var; bad values fall back so collection stays usable.""" + raw = os.environ.get(name) + if raw is None or raw == "": + return default + try: + return int(raw) + except ValueError: + return default + + +def mpi_launch_prefix(request, launcher, np, mpi_opts): + """Assemble [launcher, -np, np, ...mpi_opts, (-host hostfile)?].""" + hostfile = request.config.getoption("--hostfile") + launch = [launcher, "-np", str(np)] + list(mpi_opts) + if hostfile: + launch += ["-host", hostfile] + return launch + + +def gin_env_xflags(kv_list): + """Turn ["K=V", ...] into mpirun -x flags.""" + flags = [] + for kv in kv_list: + flags += ["-x", kv] + return flags + + +def gin_perf_argv(exe, size, dtype, ctas, extra=None): + """Fixed-size GIN -D 3 perf argv. + + `extra` is inserted after `-D 3` and before `-V`, so the AllToAll vs + AllGather/Broadcast `-A 1` gap is an explicit extra rather than a + copy-paste drift. AllToAll omits `-A 1`; the others pass extra=["-A", "1"]. + """ + args = [ + exe, + "-b", + str(size), + "-e", + str(size), + "-f", + "2", + "-g", + "1", + "-R", + "2", + "-D", + "3", + ] + args += list(extra or []) + args += [ + "-V", + str(ctas), + "-d", + dtype, + "-c", + "1", + "-w", + "1", + "-n", + "3", + ] + return args + + +def gin_hang_msg(collective, timeout_s, size, dtype, size_desc): + """Timeout diagnostic; size_desc is e.g. '256 MiB/rank' or '256 MiB'.""" + return ( + "{} GIN-SDMA HANG: no completion within {}s at {} bytes " + "({}), dtype={}. Output tail:\n{{}}".format( + collective, timeout_s, size, size_desc, dtype + ) + ) + + +def launch_mpi_shell(cmd, timeout_s, hang_msg): + """Run a shell MPI command in a new session; kill the process group on timeout.""" + print(cmd) + proc = subprocess.Popen( + cmd, + shell=True, + universal_newlines=True, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + start_new_session=True, + ) + try: + out, _ = proc.communicate(timeout=timeout_s) + except subprocess.TimeoutExpired: + try: + os.killpg(os.getpgid(proc.pid), signal.SIGKILL) + except ProcessLookupError: + pass + out, _ = proc.communicate() + pytest.fail(hang_msg.format((out or "")[-2000:])) + print(out) + return proc.returncode, out + + +def run_with_conn_gate_retry( + launch_fn, conn_retries, settle_s=3, is_data_failure=None +): + """Retry launch_fn on connectivity-gate aborts; never retry genuine data failures. + + launch_fn must return (returncode, stdout).""" + if is_data_failure is None: + is_data_failure = lambda out: bool(DATA_FAIL_RE.search(out or "")) + + rc, out = 1, "" + for attempt in range(1, max(1, conn_retries) + 1): + rc, out = launch_fn() + if rc == 0: + return rc, out + if is_data_failure(out): + return rc, out + if CONN_GATE_RE.search(out or "") and attempt < conn_retries: + print( + "=== connectivity-gate abort (attempt {}/{}); " + "re-launching after settle ===".format(attempt, conn_retries) + ) + time.sleep(settle_s) + continue + return rc, out + return rc, out + + +def read_nccl_debug_logs(debug_dir): + """Merged per-rank NCCL debug output written under debug_dir.""" + chunks = [] + for log_path in sorted(glob.glob(os.path.join(debug_dir, "nccl-debug.*"))): + try: + with open(log_path, errors="replace") as fh: + chunks.append(fh.read()) + except OSError: + pass + return "\n".join(chunks) + + +# --- Broadcast-specific parsing / assertions --------------------------------- + +PLUGIN_RE = re.compile(r"gin-anvil-sdma", re.I) + +_NUM = r"[-+]?(?:\d+\.?\d*|\.\d+)(?:[eE][-+]?\d+)?" +_WRONG = r"(?:{}|N/A)".format(_NUM) +# Deliberately not end-anchored: rccl-tests may append an algo/proto/nchannels +# group and a timestamp, and concurrent rank output can splice onto the line. +BCAST_ROW_RE = re.compile( + r"^\s*(\d+)\s+(\d+)\s+\S+\s+\S+\s+(\d+)" + r"\s+{n}\s+{n}\s+{n}\s+({w})" + r"\s+{n}\s+{n}\s+{n}\s+({w})".format(n=_NUM, w=_WRONG) +) +BCAST_OOB_RE = re.compile(r"Out of bounds values\s*:\s*(\d+)") +# Matches the tier emitted by bcastReportTier() in src/broadcast.cu. +BCAST_TIER_RE = re.compile(r"^#\[bcast-tier\]\s+(\S+)", re.M) + + +def bcast_tiers(out): + """Set of -D 3 tier names the run reported.""" + return set(BCAST_TIER_RE.findall(out or "")) + + +def bcast_rows(out): + """Measured rows as (size, root, wrong_out_of_place, wrong_in_place).""" + rows = [] + for line in (out or "").splitlines(): + m = BCAST_ROW_RE.match(line) + if m: + rows.append((int(m.group(1)), int(m.group(3)), m.group(4), m.group(5))) + return rows + + +def bcast_wrong_count(field): + """Numeric #wrong, or None when the column is N/A because checks were off.""" + try: + return float(field) + except ValueError: + return None + + +def bcast_data_failed(out): + """True when the run produced demonstrably wrong results. + + Deliberately not a text match on "Wrong": rccl-tests prints a "#wrong" column + header on every run that reaches the results table.""" + for _, _, oop, ip in bcast_rows(out): + if any(bcast_wrong_count(f) not in (None, 0.0) for f in (oop, ip)): + return True + m = BCAST_OOB_RE.search(out or "") + return bool(m and m.group(1) != "0") + + +def assert_bcast_ok(rc, out, debug, ctx, expect_tier=None): + """Assert the run bound GIN-SDMA, measured something, and checked clean.""" + tail = (out or "")[-2000:] + assert rc == 0, "Broadcast {} failed (exit {}). Output tail:\n{}".format( + ctx, rc, tail + ) + assert PLUGIN_RE.search(debug or "") or PLUGIN_RE.search(out or ""), ( + "Broadcast {} never bound the GIN Anvil-SDMA backend, so the run proves " + "nothing about the GIN path. Output tail:\n{}".format(ctx, tail) + ) + rows = bcast_rows(out) + assert rows, "Broadcast {} produced no measured rows. Output tail:\n{}".format( + ctx, tail + ) + unchecked = [ + (s, r) + for s, r, oop, ip in rows + if bcast_wrong_count(oop) is None or bcast_wrong_count(ip) is None + ] + assert not unchecked, ( + "Broadcast {} reported #wrong as N/A at (size, root) {}, so the data check " + "never ran and the result is unverified.".format(ctx, unchecked) + ) + bad = [ + (s, r, oop, ip) + for s, r, oop, ip in rows + if bcast_wrong_count(oop) != 0.0 or bcast_wrong_count(ip) != 0.0 + ] + assert ( + not bad + ), "Broadcast {} reported nonzero #wrong (size, root, oop, ip): {}".format( + ctx, bad + ) + m = BCAST_OOB_RE.search(out or "") + assert m and m.group(1) == "0", "Broadcast {} out-of-bounds count is {}".format( + ctx, m.group(1) if m else "absent" + ) + if expect_tier is not None: + allowed = ( + {expect_tier} if isinstance(expect_tier, str) else set(expect_tier) + ) + tiers = bcast_tiers(out) + assert tiers, ( + "Broadcast {} reported no #[bcast-tier] line, so which tier ran is " + "unverified. Is RCCL_TESTS_BCAST_TIER_LOG set for this launch, and is " + "the binary new enough to emit it? Output tail:\n{}".format(ctx, tail) + ) + assert len(tiers) == 1 and tiers <= allowed, ( + "Broadcast {} ran tier(s) {} but expected exactly one of {}. The tier " + "gate moved, or the tier under test fell through to another " + "kernel.".format(ctx, sorted(tiers), sorted(allowed)) + ) + + +def launch_bcast_gin_sdma( + request, + msg_bytes, + dtype, + *, + exe, + np, + launcher, + mpi_opts, + xenv, + ctas, + gin_type, + timeout_s, + force_flat_gin=False, + force_sag_gin=False, +): + """Launch broadcast_perf -D 3; returns (returncode, stdout, debug_text).""" + size = str(int(msg_bytes)) + debug_dir = tempfile.mkdtemp(prefix="bcast-gin-dbg-") + gin_env = [ + "NCCL_GIN_ENABLE=1", + "NCCL_GIN_TYPE={}".format(gin_type), + "NCCL_GIN_ANVIL_SDMA_THRESHOLD=0", + "NCCL_GIN_ANVIL_SDMA_THRESHOLD_BROADCAST=0", + "NCCL_DEBUG=INFO", + "NCCL_DEBUG_SUBSYS=INIT,NET", + "NCCL_DEBUG_FILE={}/nccl-debug.%h.%p.log".format(debug_dir), + "RCCL_TESTS_BCAST_TIER_LOG=1", + ] + if force_flat_gin: + gin_env += [ + "NCCL_GIN_ANVIL_BCAST_SCATTER_AG_MIN_BYTES=0", + "NCCL_GIN_ANVIL_BCAST_RING_MIN_BYTES=0", + ] + elif force_sag_gin: + gin_env += ["NCCL_GIN_ANVIL_BCAST_RING_MIN_BYTES=0"] + gin_env += list(xenv) + + args = ( + mpi_launch_prefix(request, launcher, np, mpi_opts) + + gin_env_xflags(gin_env) + + gin_perf_argv(exe, size, dtype, ctas, extra=["-A", "1"]) + ) + cmd = " ".join(shlex.quote(a) for a in args) + hang_msg = gin_hang_msg( + "Broadcast", timeout_s, size, dtype, "{} MiB".format(msg_bytes // MiB) + ) + try: + rc, out = launch_mpi_shell(cmd, timeout_s, hang_msg) + debug = read_nccl_debug_logs(debug_dir) + finally: + shutil.rmtree(debug_dir, ignore_errors=True) + return rc, out, debug + + +def run_bcast_with_conn_gate_retry(launch_fn, conn_retries, settle_s=3): + """Connectivity-gate retry for Broadcast launches that return debug text too.""" + state = {"debug": ""} + + def attempt(): + rc, out, debug = launch_fn() + state["debug"] = debug + return rc, out + + rc, out = run_with_conn_gate_retry( + attempt, conn_retries, settle_s=settle_s, is_data_failure=bcast_data_failed + ) + return rc, out, state["debug"] diff --git a/projects/rccl-tests/test/test_AllGather.py b/projects/rccl-tests/test/test_AllGather.py index 96af959c736c..91f9f87482fb 100644 --- a/projects/rccl-tests/test/test_AllGather.py +++ b/projects/rccl-tests/test/test_AllGather.py @@ -22,21 +22,30 @@ import os import re import shlex -import signal import subprocess -import time import itertools import math import pytest -ngpus = 0 -if os.environ.get('ROCR_VISIBLE_DEVICES') is not None: - ngpus = len(os.environ['ROCR_VISIBLE_DEVICES'].split(",")) -elif os.environ.get('HIP_VISIBLE_DEVICES') is not None: - ngpus = len(os.environ['HIP_VISIBLE_DEVICES'].split(",")) -else: - ngpus = int(subprocess.check_output("rocminfo | grep \"Device Type:.\s*.GPU\" | wc -l",shell=True)) +from .gin_sdma_harness import ( + GiB, + MiB, + detect_ngpus, + env_int, + gin_env_xflags, + gin_hang_msg, + gin_perf_argv, + launch_mpi_shell, + mpi_launch_prefix, + run_with_conn_gate_retry, +) + +try: + _detected_ngpus = detect_ngpus() +except RuntimeError: + _detected_ngpus = 0 +ngpus = max(1, _detected_ngpus) log_ngpus = int(math.log2(ngpus)) nthreads = ["1"] @@ -126,16 +135,15 @@ def test_AllGatherSingleProcess(nthreads, ngpus_single, byte_range, op, step_fac # 256/512 MiB per-rank (2/4 segments) and 2 GiB total all pass with #wrong=0, # no hang (busbw ~421-427 GB/s). -MiB = 1024 * 1024 -GiB = 1024 * MiB - _ag_enabled = os.environ.get("RCCL_TESTS_GIN_SDMA_AG", "") not in ("", "0", "false", "False") -AG_NP = int(os.environ.get("RCCL_TESTS_AG_NP", "0")) or ngpus +AG_NP = env_int("RCCL_TESTS_AG_NP", 0) or ( + detect_ngpus() if _ag_enabled else _detected_ngpus +) AG_LAUNCHER = os.environ.get("RCCL_TESTS_MPI_LAUNCHER", "mpirun") AG_CTAS = os.environ.get("RCCL_TESTS_AG_CTAS", "8") -AG_TIMEOUT_S = int(os.environ.get("RCCL_TESTS_AG_TIMEOUT_S", "900")) -AG_CONN_RETRIES = int(os.environ.get("RCCL_TESTS_AG_CONN_RETRIES", "5")) +AG_TIMEOUT_S = env_int("RCCL_TESTS_AG_TIMEOUT_S", 900) +AG_CONN_RETRIES = env_int("RCCL_TESTS_AG_CONN_RETRIES", 5) AG_MPI_OPTS = shlex.split(os.environ.get("RCCL_TESTS_MPI_OPTS", "")) AG_XENV = shlex.split(os.environ.get("RCCL_TESTS_AG_XENV", "")) AG_EXE = os.environ.get( @@ -147,98 +155,45 @@ def test_AllGatherSingleProcess(nthreads, ngpus_single, byte_range, op, step_fac "a GIN-SDMA-capable (e.g. 8x MI355X) node to enable.") -# Intermittent gfx950 cuMem-VMM connectivity-gate abort (not a data error): the -# LSA signal connectivity self-test fails to map a peer aperture and the run -# aborts before any collective work. Re-launched after a short settle, exactly -# as the shell harness does. A genuine wrong-element data check is never retried. -_CONN_GATE_RE = re.compile(r"LSA signal connectivity gate failed|unhandled system error", re.I) -_DATA_FAIL_RE = re.compile( - r"#wrong\s*=\s*[1-9]|mismatch|check.*fail|Out of bounds values\s*:\s*[1-9]", re.I) _AG_DEVTIME_TIER_RE = re.compile(r"#\[ag-devtime\].*tier\s+(LSA|SDMA)", re.I) def _launch_ag_gin_sdma(request, total_bytes, dtype, force_sdma_tier=True, device_timing=False): - """Launch all_gather_perf -D 3 once at a fixed TOTAL gathered size (per-rank - chunk = total/NP). When force_sdma_tier is True (default), threshold=0 pins the - SDMA tier for every size. Returns (returncode, stdout). A timeout (hang) fails - the test immediately.""" + """Launch all_gather_perf -D 3 once at a fixed TOTAL gathered size.""" size = str(int(total_bytes)) - gin_env = [] - for kv in ["NCCL_GIN_ENABLE=1", "NCCL_GIN_TYPE=7"] + AG_XENV: - gin_env += ["-x", kv] + gin_kv = ["NCCL_GIN_ENABLE=1", "NCCL_GIN_TYPE=7"] + AG_XENV if force_sdma_tier: - for kv in ["NCCL_GIN_ANVIL_SDMA_THRESHOLD=0", - "NCCL_GIN_ANVIL_SDMA_THRESHOLD_ALLGATHER=0"]: - gin_env += ["-x", kv] - - hostfile = request.config.getoption("--hostfile") - launch = [AG_LAUNCHER, "-np", str(AG_NP)] + AG_MPI_OPTS - if hostfile: - launch += ["-host", hostfile] + gin_kv += [ + "NCCL_GIN_ANVIL_SDMA_THRESHOLD=0", + "NCCL_GIN_ANVIL_SDMA_THRESHOLD_ALLGATHER=0", + ] - args = launch + gin_env + [ - AG_EXE, - "-b", size, "-e", size, - "-f", "2", - "-g", "1", - "-R", "2", - "-D", "3", # GinHybridAllGatherKernel - "-A", "1", # emit algo/proto/nchannels columns - "-V", AG_CTAS, - "-d", dtype, - "-c", "1", # data check: nonzero exit on any wrong element - "-w", "1", - "-n", "3", - ] + extra = ["-A", "1"] + args = ( + mpi_launch_prefix(request, AG_LAUNCHER, AG_NP, AG_MPI_OPTS) + + gin_env_xflags(gin_kv) + + gin_perf_argv(AG_EXE, size, dtype, AG_CTAS, extra=extra) + ) if device_timing: - args += ["-B", "1"] # augment stdout with #[ag-devtime] tier line + args += ["-B", "1"] cmd = " ".join(shlex.quote(a) for a in args) - print(cmd) - # New session so a hang can be killed as a whole process group -- otherwise a - # wedged launcher leaves orphaned ranks pinning the GPUs (and exhausting SDMA - # queues for later tests). - proc = subprocess.Popen(cmd, shell=True, universal_newlines=True, - stdout=subprocess.PIPE, stderr=subprocess.STDOUT, - start_new_session=True) - try: - out, _ = proc.communicate(timeout=AG_TIMEOUT_S) - except subprocess.TimeoutExpired: - try: - os.killpg(os.getpgid(proc.pid), signal.SIGKILL) - except ProcessLookupError: - pass - out, _ = proc.communicate() - pytest.fail( - "AllGather GIN-SDMA HANG: no completion within {}s at total={} bytes " - "({} MiB/rank), dtype={}. Output tail:\n{}".format( - AG_TIMEOUT_S, size, total_bytes // AG_NP // MiB, dtype, (out or "")[-2000:])) - print(out) - return proc.returncode, out + hang_msg = gin_hang_msg( + "AllGather", + AG_TIMEOUT_S, + size, + dtype, + "{} MiB/rank".format(total_bytes // AG_NP // MiB), + ) + return launch_mpi_shell(cmd, AG_TIMEOUT_S, hang_msg) def _run_ag_gin_sdma(request, total_bytes, dtype, **launch_kw): - """Launch with connectivity-gate retry. Returns (returncode, stdout) of the - first run that either succeeds, hits a genuine data-check failure, or fails - for a non-connectivity-gate reason; otherwise the last attempt.""" if AG_NP < 2: pytest.skip("need >= 2 ranks/GPUs for GIN-SDMA AllGather") - - rc, out = 1, "" - for attempt in range(1, max(1, AG_CONN_RETRIES) + 1): - rc, out = _launch_ag_gin_sdma(request, total_bytes, dtype, **launch_kw) - if rc == 0: - return rc, out - # A real wrong-element mismatch must fail now, never retried. - if _DATA_FAIL_RE.search(out or ""): - return rc, out - # Intermittent gfx950 cuMem-VMM connectivity-gate abort: settle + retry. - if _CONN_GATE_RE.search(out or "") and attempt < AG_CONN_RETRIES: - print("=== connectivity-gate abort (attempt {}/{}); re-launching after settle ===".format( - attempt, AG_CONN_RETRIES)) - time.sleep(3) - continue - return rc, out - return rc, out + return run_with_conn_gate_retry( + lambda: _launch_ag_gin_sdma(request, total_bytes, dtype, **launch_kw), + AG_CONN_RETRIES, + ) # (items 1 + 2) Multi-segment loop and the 1 GiB single-copy boundary. per_rank diff --git a/projects/rccl-tests/test/test_AllToAll.py b/projects/rccl-tests/test/test_AllToAll.py index f2af20d8520c..8bd2a65341ce 100644 --- a/projects/rccl-tests/test/test_AllToAll.py +++ b/projects/rccl-tests/test/test_AllToAll.py @@ -46,9 +46,10 @@ # RCCL_TESTS_MPI_LAUNCHER launcher binary (default: mpirun) # RCCL_TESTS_MPI_OPTS extra launcher opts (e.g. --allow-run-as-root -mca ...) # RCCL_TESTS_A2A_XENV extra "-x K=V" env the backend needs on this cluster -# (e.g. HSA_FORCE_FINE_GRAIN_PCIE=1 NCCL_CUMEM_ENABLE=1) # RCCL_TESTS_A2A_EXE path to alltoall_perf (default: ../build/alltoall_perf) +# RCCL_TESTS_A2A_CTAS device CTA count (-V) (default: 16) # RCCL_TESTS_A2A_TIMEOUT_S per-run hang timeout in seconds (default: 900) +# RCCL_TESTS_A2A_CONN_RETRIES connectivity-gate retries (default: 5) # # Verified on 8x MI355X (ROCm 7.13, NCCL_GIN_TYPE=7, force1ch): 256 MiB/peer and # 2 GiB/peer int32 and the 2 GiB-total guard all pass with #wrong=0, no hang @@ -56,128 +57,99 @@ import os import shlex -import signal -import subprocess import pytest -MiB = 1024 * 1024 -GiB = 1024 * MiB +from .gin_sdma_harness import ( + GiB, + MiB, + detect_ngpus, + env_int, + gin_env_xflags, + gin_hang_msg, + gin_perf_argv, + launch_mpi_shell, + mpi_launch_prefix, + run_with_conn_gate_retry, +) path = os.path.dirname(os.path.abspath(__file__)) -# Path to alltoall_perf. Override with RCCL_TESTS_A2A_EXE when the binary is not -# at the default build/ location (e.g. an installed/container path). executable = os.environ.get( - "RCCL_TESTS_A2A_EXE", os.path.join(path, "..", "build", "alltoall_perf")) + "RCCL_TESTS_A2A_EXE", os.path.join(path, "..", "build", "alltoall_perf") +) -# Opt-in: only run where the GIN-SDMA backend + hardware are available. -_enabled = os.environ.get("RCCL_TESTS_GIN_SDMA_A2A", "") not in ("", "0", "false", "False") +_enabled = os.environ.get("RCCL_TESTS_GIN_SDMA_A2A", "") not in ( + "", + "0", + "false", + "False", +) - -def _detect_ngpus(): - if os.environ.get("ROCR_VISIBLE_DEVICES") is not None: - return len(os.environ["ROCR_VISIBLE_DEVICES"].split(",")) - if os.environ.get("HIP_VISIBLE_DEVICES") is not None: - return len(os.environ["HIP_VISIBLE_DEVICES"].split(",")) - try: - out = subprocess.check_output( - 'rocminfo | grep "Device Type:.\\s*.GPU" | wc -l', shell=True) - return int(out) - except Exception: - return 0 - - -NP = int(os.environ.get("RCCL_TESTS_A2A_NP", "0")) or _detect_ngpus() +NP = env_int("RCCL_TESTS_A2A_NP", 0) or (detect_ngpus() if _enabled else 0) LAUNCHER = os.environ.get("RCCL_TESTS_MPI_LAUNCHER", "mpirun") CTAS = os.environ.get("RCCL_TESTS_A2A_CTAS", "16") -# Generous per-call timeout; a real hang spins forever, so a finite cap is what -# converts item (3) into a detectable failure. Scaled up for the multi-GiB case. -TIMEOUT_S = int(os.environ.get("RCCL_TESTS_A2A_TIMEOUT_S", "900")) - -# Extra launcher options (e.g. "--allow-run-as-root -mca pml ob1 ...") and any -# deployment-specific env the GIN-SDMA backend needs beyond the essentials below -# (e.g. HSA_FORCE_FINE_GRAIN_PCIE=1 NCCL_CUMEM_ENABLE=1 RCCL_ENABLE_INTRANET=1). -# GIN enablement itself is site-specific, so it is injected here rather than -# hard-coded, keeping the test portable across clusters. +TIMEOUT_S = env_int("RCCL_TESTS_A2A_TIMEOUT_S", 900) +CONN_RETRIES = env_int("RCCL_TESTS_A2A_CONN_RETRIES", 5) MPI_OPTS = shlex.split(os.environ.get("RCCL_TESTS_MPI_OPTS", "")) XENV = shlex.split(os.environ.get("RCCL_TESTS_A2A_XENV", "")) pytestmark = pytest.mark.skipif( not _enabled, reason="GIN-SDMA AllToAll tests are opt-in; set RCCL_TESTS_GIN_SDMA_A2A=1 on " - "a GIN-SDMA-capable (e.g. 8x MI355X) node to enable.") + "a GIN-SDMA-capable (e.g. 8x MI355X) node to enable.", +) + + +def _launch_a2a(request, total_bytes, dtype): + """Launch alltoall_perf -D 3 once at a fixed total (per-rank) size.""" + size = str(int(total_bytes)) + gin_env = gin_env_xflags( + [ + "NCCL_GIN_ENABLE=1", + "NCCL_GIN_TYPE=7", + "NCCL_GIN_ANVIL_SDMA_THRESHOLD=0", + "NCCL_GIN_ANVIL_SDMA_THRESHOLD_ALLTOALL=0", + ] + + XENV + ) + + args = ( + mpi_launch_prefix(request, LAUNCHER, NP, MPI_OPTS) + + gin_env + + gin_perf_argv(executable, size, dtype, CTAS) + ) + cmd = " ".join(shlex.quote(a) for a in args) + hang_msg = gin_hang_msg( + "AllToAll", + TIMEOUT_S, + size, + dtype, + "{} MiB/peer".format(total_bytes // NP // MiB), + ) + return launch_mpi_shell(cmd, TIMEOUT_S, hang_msg) def _run_a2a(request, total_bytes, dtype): - """Run one all-to-all at a fixed total (per-rank) size, forcing the GIN-SDMA - tier. Returns (returncode, stdout). A timeout (hang) fails the test.""" + """Run one all-to-all with connectivity-gate retry on gfx950 VMM aborts.""" if NP < 2: pytest.skip("need >= 2 ranks/GPUs for AllToAll") + return run_with_conn_gate_retry( + lambda: _launch_a2a(request, total_bytes, dtype), CONN_RETRIES + ) - size = str(int(total_bytes)) - # Essentials to exercise the GIN-SDMA put path; force the SDMA (large) tier - # for every size. Deployment-specific env comes from RCCL_TESTS_A2A_XENV. - gin_env = [] - for kv in ["NCCL_GIN_ENABLE=1", "NCCL_GIN_TYPE=7", - "NCCL_GIN_ANVIL_SDMA_THRESHOLD=0", - "NCCL_GIN_ANVIL_SDMA_THRESHOLD_ALLTOALL=0"] + XENV: - gin_env += ["-x", kv] - - hostfile = request.config.getoption("--hostfile") - launch = [LAUNCHER, "-np", str(NP)] + MPI_OPTS - if hostfile: - launch += ["-host", hostfile] - - args = launch + gin_env + [ - executable, - "-b", size, "-e", size, - "-f", "2", - "-g", "1", - "-R", "2", - "-D", "3", - "-V", CTAS, - "-d", dtype, - "-c", "1", # data check: nonzero exit on any wrong element - "-w", "1", - "-n", "3", - ] - cmd = " ".join(shlex.quote(a) for a in args) - print(cmd) - # Run in a new session so a hang can be killed as a whole process group -- - # otherwise a wedged launcher leaves orphaned ranks pinning the GPUs (and - # exhausting SDMA queues for later tests). - proc = subprocess.Popen(cmd, shell=True, universal_newlines=True, - stdout=subprocess.PIPE, stderr=subprocess.STDOUT, - start_new_session=True) - try: - out, _ = proc.communicate(timeout=TIMEOUT_S) - except subprocess.TimeoutExpired: - try: - os.killpg(os.getpgid(proc.pid), signal.SIGKILL) - except ProcessLookupError: - pass - out, _ = proc.communicate() - pytest.fail( - "AllToAll GIN-SDMA HANG: no completion within {}s at total={} bytes " - "({} MiB/peer), dtype={}. Output tail:\n{}".format( - TIMEOUT_S, size, total_bytes // NP // MiB, dtype, (out or "")[-2000:])) - print(out) - return proc.returncode, out - - -# (item 1 + 2) Multi-segment loop and the 1 GiB truncation boundary. per_peer is -# the transfer to each peer; total per-rank bytes = per_peer * NP. -@pytest.mark.parametrize("per_peer_mib", [256, 2048]) # 256 MiB (2 seg), 2 GiB (16 seg) + +@pytest.mark.parametrize("per_peer_mib", [256, 2048]) @pytest.mark.parametrize("dtype", ["int32", "int64", "uint8"]) def test_AllToAllGinSdmaLargeSegmented(request, per_peer_mib, dtype): total = per_peer_mib * MiB * NP rc, _ = _run_a2a(request, total, dtype) - assert rc == 0, "AllToAll data check failed (nonzero exit) at {} MiB/peer, dtype={}".format( - per_peer_mib, dtype) + assert rc == 0, ( + "AllToAll data check failed (nonzero exit) at {} MiB/peer, dtype={}".format( + per_peer_mib, dtype + ) + ) -# (item 3) The original 2 GiB-total hang repro, as a completion guard. At 8 ranks -# this is 256 MiB/peer -- the size that hung as a single unclamped put. def test_AllToAllGinSdma2GiBTotalHangGuard(request): rc, _ = _run_a2a(request, 2 * GiB, "int32") assert rc == 0, "AllToAll 2 GiB-total data check failed (nonzero exit)" diff --git a/projects/rccl-tests/test/test_Broadcast.py b/projects/rccl-tests/test/test_Broadcast.py index 6e18da5aa227..1f9fa0434ada 100644 --- a/projects/rccl-tests/test/test_Broadcast.py +++ b/projects/rccl-tests/test/test_Broadcast.py @@ -20,30 +20,33 @@ ################################################################################ import os -import re -import glob import shlex -import shutil -import signal import subprocess import itertools import math -import tempfile -import time import pytest -ngpus = 0 -if os.environ.get("ROCR_VISIBLE_DEVICES") is not None: - ngpus = len(os.environ["ROCR_VISIBLE_DEVICES"].split(",")) -elif os.environ.get("HIP_VISIBLE_DEVICES") is not None: - ngpus = len(os.environ["HIP_VISIBLE_DEVICES"].split(",")) -else: - ngpus = int( - subprocess.check_output( - 'rocminfo | grep "Device Type:.\\s*.GPU" | wc -l', shell=True - ) - ) +from .gin_sdma_harness import ( + GiB, + MiB, + assert_bcast_ok, + bcast_data_failed, + bcast_rows, + bcast_tiers, + detect_ngpus, + env_int, + gin_perf_argv, + launch_bcast_gin_sdma, + run_bcast_with_conn_gate_retry, + run_with_conn_gate_retry, +) + +try: + _detected_ngpus = detect_ngpus() +except RuntimeError: + _detected_ngpus = 0 +ngpus = max(1, _detected_ngpus) log_ngpus = int(math.log2(ngpus)) nthreads = ["1"] @@ -144,9 +147,6 @@ def test_BroadcastSingleProcess( # - 6 : #9924 (131884a2ba) [NCCL v2.30.7] to #10785 (d0f7d1966a) [NCCL v2.31.2] # - 5 : #7826 (bc6a304a9c) [SDMA support] to #9924 (131884a2ba) [NCCL v2.30.7] -MiB = 1024 * 1024 -GiB = 1024 * MiB - _bcast_enabled = os.environ.get("RCCL_TESTS_GIN_SDMA_BCAST", "") not in ( "", "0", @@ -154,19 +154,9 @@ def test_BroadcastSingleProcess( "False", ) - -def _env_int(name, default): - """Parse an integer env var; bad values fall back so collection stays usable.""" - raw = os.environ.get(name) - if raw is None or raw == "": - return default - try: - return int(raw) - except ValueError: - return default - - -BCAST_NP = _env_int("RCCL_TESTS_BCAST_NP", 0) or ngpus +BCAST_NP = env_int("RCCL_TESTS_BCAST_NP", 0) or ( + detect_ngpus() if _bcast_enabled else _detected_ngpus +) BCAST_LAUNCHER = os.environ.get("RCCL_TESTS_MPI_LAUNCHER", "mpirun") BCAST_CTAS = os.environ.get("RCCL_TESTS_BCAST_CTAS", "8") BCAST_GIN_TYPE = os.environ.get("RCCL_TESTS_BCAST_GIN_TYPE", "7") @@ -174,8 +164,8 @@ def _env_int(name, default): # HW_CASES * retries * TIMEOUT_S stays under GIN_PYTEST_TIMEOUT; otherwise # GNU timeout kills pytest and leaves the mpirun session (start_new_session) # orphaned because killpg only runs in this TimeoutExpired handler. -BCAST_TIMEOUT_S = _env_int("RCCL_TESTS_BCAST_TIMEOUT_S", 900) -BCAST_CONN_RETRIES = _env_int("RCCL_TESTS_BCAST_CONN_RETRIES", 5) +BCAST_TIMEOUT_S = env_int("RCCL_TESTS_BCAST_TIMEOUT_S", 900) +BCAST_CONN_RETRIES = env_int("RCCL_TESTS_BCAST_CONN_RETRIES", 5) BCAST_MPI_OPTS = shlex.split(os.environ.get("RCCL_TESTS_MPI_OPTS", "")) BCAST_XENV = shlex.split(os.environ.get("RCCL_TESTS_BCAST_XENV", "")) BCAST_EXE = os.environ.get( @@ -188,267 +178,18 @@ def _env_int(name, default): "a GIN-SDMA-capable (e.g. 8x MI355X) node to enable.", ) -_CONN_GATE_RE = re.compile( - r"LSA signal connectivity gate failed|unhandled system error", re.I -) - -# The GIN Anvil-SDMA backend has to actually bind for any of this to mean -# something: a run that falls back to another transport still exits 0 and still -# prints a full results table. NCCL_DEBUG_SUBSYS=INIT,NET below is what puts the -# bind line in the captured output. -_PLUGIN_RE = re.compile(r"gin-anvil-sdma", re.I) - -# One measured rccl-tests results row: -# size count type redop root | time algbw busbw #wrong | time algbw busbw #wrong -# out-of-place columns first, then in-place. Timings come from getFloatStr, which -# falls back to scientific notation when a value will not fit its width, so the -# numeric fields have to admit exponents. The row is not end-anchored: an algo / -# proto / nchannels group and a timestamp may follow the in-place columns, and -# concurrent rank output can splice itself onto the end of the line. -_NUM = r"[-+]?(?:\d+\.?\d*|\.\d+)(?:[eE][-+]?\d+)?" -_WRONG = r"(?:{}|N/A)".format(_NUM) -_ROW_RE = re.compile( - r"^\s*(\d+)\s+(\d+)\s+\S+\s+\S+\s+(\d+)" - r"\s+{n}\s+{n}\s+{n}\s+({w})" - r"\s+{n}\s+{n}\s+{n}\s+({w})".format(n=_NUM, w=_WRONG) -) -_OOB_RE = re.compile(r"Out of bounds values\s*:\s*(\d+)") - -# "#[bcast-tier] ", emitted once per distinct tier by BroadcastRunColl when -# RCCL_TESTS_BCAST_TIER_LOG is set. Named tiers are the contract between the -D 3 -# dispatch and the tier assertions below; see bcastReportTier in broadcast.cu. -_TIER_RE = re.compile(r"^#\[bcast-tier\]\s+(\S+)", re.M) - - -def _bcast_tiers(out): - """Set of -D 3 tier names the run reported.""" - return set(_TIER_RE.findall(out or "")) - - -def _bcast_rows(out): - """Measured rows as (size, root, wrong_out_of_place, wrong_in_place).""" - rows = [] - for line in (out or "").splitlines(): - m = _ROW_RE.match(line) - if m: - rows.append((int(m.group(1)), int(m.group(3)), m.group(4), m.group(5))) - return rows - - -def _wrong_count(field): - """Numeric #wrong, or None when the column is N/A because checks were off.""" - try: - return float(field) - except ValueError: - return None - -def _data_failed(out): - """True when the run produced demonstrably wrong results. - - Deliberately not a text match on "Wrong": rccl-tests prints a "#wrong" column - header on every run that reaches the results table, so a bare pattern matched - every healthy run and made the connectivity retry below unreachable. N/A is - not a failure here -- it means checking was off, which _assert_bcast_ok - rejects separately rather than burning retries on. - """ - for _, _, oop, ip in _bcast_rows(out): - if any(_wrong_count(f) not in (None, 0.0) for f in (oop, ip)): - return True - m = _OOB_RE.search(out or "") - return bool(m and m.group(1) != "0") - - -def _read_debug_logs(debug_dir): - """Merged per-rank NCCL debug output written under debug_dir.""" - chunks = [] - for log_path in sorted(glob.glob(os.path.join(debug_dir, "nccl-debug.*"))): - try: - with open(log_path, errors="replace") as fh: - chunks.append(fh.read()) - except OSError: - pass - return "\n".join(chunks) - - -def _assert_bcast_ok(rc, out, debug, ctx, expect_tier=None): - """Assert the run bound GIN-SDMA, measured something, and checked clean. - - Exit status alone is not enough: it stays 0 for a run that binds a different - backend, and for one that produces no measured rows at all. - - Nor is the data check enough to say which tier ran. Every -D 3 tier produces - a correct broadcast, so #wrong stays 0 when a tier is deleted and its - messages fall through to the next one -- the test keeps passing while the - kernel it named is no longer reached. expect_tier pins the tier that has to - have run, so removing it fails here instead of silently going uncovered. - """ - tail = (out or "")[-2000:] - assert rc == 0, "Broadcast {} failed (exit {}). Output tail:\n{}".format( - ctx, rc, tail - ) - assert _PLUGIN_RE.search(debug or "") or _PLUGIN_RE.search(out or ""), ( - "Broadcast {} never bound the GIN Anvil-SDMA backend, so the run proves " - "nothing about the GIN path. Output tail:\n{}".format(ctx, tail) - ) - rows = _bcast_rows(out) - assert rows, "Broadcast {} produced no measured rows. Output tail:\n{}".format( - ctx, tail - ) - unchecked = [ - (s, r) - for s, r, oop, ip in rows - if _wrong_count(oop) is None or _wrong_count(ip) is None - ] - assert not unchecked, ( - "Broadcast {} reported #wrong as N/A at (size, root) {}, so the data check " - "never ran and the result is unverified.".format(ctx, unchecked) - ) - bad = [ - (s, r, oop, ip) - for s, r, oop, ip in rows - if _wrong_count(oop) != 0.0 or _wrong_count(ip) != 0.0 - ] - assert ( - not bad - ), "Broadcast {} reported nonzero #wrong (size, root, oop, ip): {}".format(ctx, bad) - m = _OOB_RE.search(out or "") - assert m and m.group(1) == "0", "Broadcast {} out-of-bounds count is {}".format( - ctx, m.group(1) if m else "absent" - ) - if expect_tier is not None: - # A string pins one tier; a collection allows a set of equivalent ones. - # The ring tier needs the latter: it reports ring-table where an - # edge-disjoint decomposition exists and ring-multi where none does - # (N=4 and N=6), and BCAST_NP follows whatever GPUs the host has. - # - # The ring also requires the LSA team to cover the world, so a multi-node - # launch takes scatter+allgather instead and fails here. That is the - # intent: such a run is not exercising the ring the caller asked for. - allowed = ( - {expect_tier} if isinstance(expect_tier, str) else set(expect_tier) - ) - tiers = _bcast_tiers(out) - assert tiers, ( - "Broadcast {} reported no #[bcast-tier] line, so which tier ran is " - "unverified. Is RCCL_TESTS_BCAST_TIER_LOG set for this launch, and is " - "the binary new enough to emit it? Output tail:\n{}".format(ctx, tail) - ) - # Exactly one tier, and an allowed one. These launches pin a single size, - # so a second tier means some roots took a different kernel; merely - # intersecting `allowed` would let that through. - assert len(tiers) == 1 and tiers <= allowed, ( - "Broadcast {} ran tier(s) {} but expected exactly one of {}. The tier " - "gate moved, or the tier under test fell through to another " - "kernel.".format(ctx, sorted(tiers), sorted(allowed)) - ) - - -def _launch_bcast_gin_sdma( - request, msg_bytes, dtype, *, force_flat_gin=False, force_sag_gin=False -): - """Launch broadcast_perf -D 3 at a fixed message size. When force_flat_gin, - disable the SAG/ring large tiers so the root flat gin.put() path is used. - When force_sag_gin, disable only the ring tier so scatter+allgather runs. - - Returns (returncode, stdout, debug_text); debug_text is the merged per-rank - NCCL debug log, kept off stdout so the results table stays parseable.""" - size = str(int(msg_bytes)) - debug_dir = tempfile.mkdtemp(prefix="bcast-gin-dbg-") - gin_env = [ - "NCCL_GIN_ENABLE=1", - "NCCL_GIN_TYPE={}".format(BCAST_GIN_TYPE), - "NCCL_GIN_ANVIL_SDMA_THRESHOLD=0", - "NCCL_GIN_ANVIL_SDMA_THRESHOLD_BROADCAST=0", - # Debug output is what lets _PLUGIN_RE confirm which backend actually - # bound. It goes to per-rank files rather than stdout because ranks - # interleave: on stdout an INFO line splices itself into the middle of a - # results row, which corrupts the very table we need to parse. - "NCCL_DEBUG=INFO", - "NCCL_DEBUG_SUBSYS=INIT,NET", - "NCCL_DEBUG_FILE={}/nccl-debug.%h.%p.log".format(debug_dir), - # Makes BroadcastRunColl name the tier it launched, which is what lets - # _assert_bcast_ok tell the -D 3 tiers apart. One line from the main - # proc, so the results table stays parseable. - "RCCL_TESTS_BCAST_TIER_LOG=1", - ] - if force_flat_gin: - gin_env += [ - "NCCL_GIN_ANVIL_BCAST_SCATTER_AG_MIN_BYTES=0", - "NCCL_GIN_ANVIL_BCAST_RING_MIN_BYTES=0", - ] - elif force_sag_gin: - # Ring is checked before SAG; disable ring only so SAG is exercised. - gin_env += ["NCCL_GIN_ANVIL_BCAST_RING_MIN_BYTES=0"] - gin_env += BCAST_XENV - gin_env = sum([["-x", kv] for kv in gin_env], []) - - hostfile = request.config.getoption("--hostfile") - launch = [BCAST_LAUNCHER, "-np", str(BCAST_NP)] + BCAST_MPI_OPTS - if hostfile: - launch += ["-host", hostfile] - - args = ( - launch - + gin_env - + [ - BCAST_EXE, - "-b", - size, - "-e", - size, - "-f", - "2", - "-g", - "1", - "-R", - "2", - "-D", - "3", # GinHybridBroadcastKernel - "-A", - "1", - "-V", - BCAST_CTAS, - "-d", - dtype, - "-c", - "1", - "-w", - "1", - "-n", - "3", - ] - ) - cmd = " ".join(shlex.quote(a) for a in args) - print(cmd) - try: - proc = subprocess.Popen( - cmd, - shell=True, - universal_newlines=True, - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, - start_new_session=True, - ) - try: - out, _ = proc.communicate(timeout=BCAST_TIMEOUT_S) - except subprocess.TimeoutExpired: - try: - os.killpg(os.getpgid(proc.pid), signal.SIGKILL) - except ProcessLookupError: - pass - out, _ = proc.communicate() - pytest.fail( - "Broadcast GIN-SDMA HANG: no completion within {}s at msg={} bytes " - "({} MiB), dtype={}. Output tail:\n{}".format( - BCAST_TIMEOUT_S, size, msg_bytes // MiB, dtype, (out or "")[-2000:] - ) - ) - debug = _read_debug_logs(debug_dir) - finally: - shutil.rmtree(debug_dir, ignore_errors=True) - print(out) - return proc.returncode, out, debug +def _bcast_launch_kwargs(): + return { + "exe": BCAST_EXE, + "np": BCAST_NP, + "launcher": BCAST_LAUNCHER, + "mpi_opts": BCAST_MPI_OPTS, + "xenv": BCAST_XENV, + "ctas": BCAST_CTAS, + "gin_type": BCAST_GIN_TYPE, + "timeout_s": BCAST_TIMEOUT_S, + } def _run_bcast_gin_sdma( @@ -456,30 +197,17 @@ def _run_bcast_gin_sdma( ): if BCAST_NP < 2: pytest.skip("need >= 2 ranks/GPUs for GIN-SDMA Broadcast") - - rc, out, debug = 1, "", "" - for attempt in range(1, max(1, BCAST_CONN_RETRIES) + 1): - rc, out, debug = _launch_bcast_gin_sdma( + return run_bcast_with_conn_gate_retry( + lambda: launch_bcast_gin_sdma( request, msg_bytes, dtype, force_flat_gin=force_flat_gin, force_sag_gin=force_sag_gin, - ) - if rc == 0: - return rc, out, debug - if _data_failed(out): - return rc, out, debug - if _CONN_GATE_RE.search(out or "") and attempt < BCAST_CONN_RETRIES: - print( - "=== connectivity-gate abort (attempt {}/{}); re-launching after settle ===".format( - attempt, BCAST_CONN_RETRIES - ) - ) - time.sleep(3) - continue - return rc, out, debug - return rc, out, debug + **_bcast_launch_kwargs(), + ), + BCAST_CONN_RETRIES, + ) @_bcast_skip @@ -490,7 +218,7 @@ def test_BroadcastGinSdmaLargeSegmented(request, msg_mib, dtype): rc, out, debug = _run_bcast_gin_sdma( request, msg_mib * MiB, dtype, force_flat_gin=True ) - _assert_bcast_ok( + assert_bcast_ok( rc, out, debug, @@ -505,7 +233,7 @@ def test_BroadcastGinSdmaScatterAllgather(request): rc, out, debug = _run_bcast_gin_sdma( request, 256 * MiB, "int32", force_sag_gin=True ) - _assert_bcast_ok( + assert_bcast_ok( rc, out, debug, "SAG 256 MiB", expect_tier="scatter-allgather" ) @@ -516,7 +244,7 @@ def test_BroadcastGinSdma2GiBHangGuard(request): rc, out, debug = _run_bcast_gin_sdma( request, 2 * GiB, "int32", force_flat_gin=False ) - _assert_bcast_ok( + assert_bcast_ok( rc, out, debug, @@ -531,7 +259,7 @@ def test_BroadcastGinSdma4GiBHangGuard(request): rc, out, debug = _run_bcast_gin_sdma( request, 4 * GiB, "int32", force_flat_gin=False ) - _assert_bcast_ok( + assert_bcast_ok( rc, out, debug, @@ -551,7 +279,7 @@ def test_BroadcastGinSdmaRowRegexAcceptsScientificNotation(): " 1.23e+02 1.00e+02 1.00e+02 0" " 1.23e+02 1.00e+02 1.00e+02 0" ) - rows = _bcast_rows(line) + rows = bcast_rows(line) assert rows == [(134217728, 0, "0", "0")] @@ -560,7 +288,7 @@ def test_BroadcastGinSdmaDataFailedIgnoresColumnHeader(): "# size count type redop root" " time algbw busbw #wrong" ) - assert not _data_failed(header) + assert not bcast_data_failed(header) def test_BroadcastGinSdmaDataFailedDetectsNonzeroWrong(): @@ -569,7 +297,7 @@ def test_BroadcastGinSdmaDataFailedDetectsNonzeroWrong(): " 12.34 1.00 1.00 1" " 12.34 1.00 1.00 0" ) - assert _data_failed(line) + assert bcast_data_failed(line) def test_BroadcastGinSdmaDataFailedTreatsNaAsUncheckedNotFailed(): @@ -578,7 +306,77 @@ def test_BroadcastGinSdmaDataFailedTreatsNaAsUncheckedNotFailed(): " 12.34 1.00 1.00 N/A" " 12.34 1.00 1.00 0" ) - assert not _data_failed(line) + assert not bcast_data_failed(line) + + +def test_BroadcastGinSdmaConnRetryReturnsCleanPassWithoutRetry(): + calls = [] + + def launch(): + calls.append(None) + return 0, "clean" + + assert run_with_conn_gate_retry(launch, 5, settle_s=0) == (0, "clean") + assert len(calls) == 1 + + +def test_BroadcastGinSdmaConnRetryExhaustsConnectivityAborts(): + calls = [] + + def launch(): + calls.append(None) + return 1, "LSA signal connectivity gate failed" + + assert run_with_conn_gate_retry(launch, 3, settle_s=0) == ( + 1, + "LSA signal connectivity gate failed", + ) + assert len(calls) == 3 + + +def test_BroadcastGinSdmaConnRetryNeverRetriesDataFailure(): + calls = [] + + def launch(): + calls.append(None) + return 1, "unhandled system error: #wrong = 1" + + assert run_with_conn_gate_retry(launch, 5, settle_s=0) == ( + 1, + "unhandled system error: #wrong = 1", + ) + assert len(calls) == 1 + + +def test_BroadcastGinSdmaConnRetryDoesNotRetryNonGateError(): + calls = [] + + def launch(): + calls.append(None) + return 1, "some other error" + + assert run_with_conn_gate_retry(launch, 5, settle_s=0) == ( + 1, + "some other error", + ) + assert len(calls) == 1 + + +def test_BroadcastGinSdmaDetectNgpusRequiresRocminfo(monkeypatch): + monkeypatch.delenv("ROCR_VISIBLE_DEVICES", raising=False) + monkeypatch.delenv("HIP_VISIBLE_DEVICES", raising=False) + from . import gin_sdma_harness as harness + + monkeypatch.setattr(harness.shutil, "which", lambda _cmd: None) + with pytest.raises(RuntimeError, match="rocminfo"): + harness.detect_ngpus() + + +def test_BroadcastGinSdmaPerfArgvMakesAverageCountGapExplicit(): + ag = gin_perf_argv("all_gather_perf", 8, "int32", "8", extra=["-A", "1"]) + a2a = gin_perf_argv("alltoall_perf", 8, "int32", "16") + assert ag[ag.index("-D") + 2 : ag.index("-V")] == ["-A", "1"] + assert a2a[a2a.index("-D") + 2 : a2a.index("-V")] == [] def _clean_bcast_output(tier): @@ -599,16 +397,16 @@ def _clean_bcast_output(tier): # expands its args unquoted, so a filter with a space in it is not available. def test_BroadcastGinSdmaTierGuardLineIsNotAResultsRow(): """The tier line shares stdout with the table it must not corrupt.""" - assert _bcast_rows("#[bcast-tier] scatter-allgather") == [] - assert _bcast_tiers(_clean_bcast_output("ring-table")) == {"ring-table"} - assert _bcast_rows(_clean_bcast_output("ring-table")) + assert bcast_rows("#[bcast-tier] scatter-allgather") == [] + assert bcast_tiers(_clean_bcast_output("ring-table")) == {"ring-table"} + assert bcast_rows(_clean_bcast_output("ring-table")) def test_BroadcastGinSdmaTierGuardAcceptsExpectedTier(): out = _clean_bcast_output("scatter-allgather") - _assert_bcast_ok(0, out, "gin-anvil-sdma", "guard", expect_tier="scatter-allgather") + assert_bcast_ok(0, out, "gin-anvil-sdma", "guard", expect_tier="scatter-allgather") # Ring passes against either decomposition outcome. - _assert_bcast_ok( + assert_bcast_ok( 0, _clean_bcast_output("ring-multi"), "gin-anvil-sdma", @@ -626,9 +424,9 @@ def test_BroadcastGinSdmaTierGuardRejectsFallThrough(): would go green while covering nothing. """ fell_through = _clean_bcast_output("flat-gin") - _assert_bcast_ok(0, fell_through, "gin-anvil-sdma", "guard", expect_tier="flat-gin") + assert_bcast_ok(0, fell_through, "gin-anvil-sdma", "guard", expect_tier="flat-gin") with pytest.raises(AssertionError, match="expected exactly one of"): - _assert_bcast_ok( + assert_bcast_ok( 0, fell_through, "gin-anvil-sdma", "guard", expect_tier="scatter-allgather" ) @@ -637,7 +435,7 @@ def test_BroadcastGinSdmaTierGuardRejectsMissingTierLine(): """A binary predating the tier log must fail loudly, not pass vacuously.""" no_tier = "\n".join(_clean_bcast_output("flat-gin").splitlines()[1:]) with pytest.raises(AssertionError, match="no .*bcast-tier.* line"): - _assert_bcast_ok( + assert_bcast_ok( 0, no_tier, "gin-anvil-sdma", "guard", expect_tier="flat-gin" )