From a1cfd4ef9c5e595426842fd6fa675787f5a2c71f Mon Sep 17 00:00:00 2001 From: lcskrishna Date: Mon, 27 Apr 2026 11:47:37 +0000 Subject: [PATCH 1/2] add cluster rdma env recommender & cluster rdma bw tests --- docs/preflight.md | 14 +- primus/cli/subcommands/preflight.py | 2 + primus/tools/preflight/README.md | 14 + .../preflight/cluster_sphere/__init__.py | 11 + .../cluster_sphere/env_recommender.py | 406 ++++++++++++++++++ .../tools/preflight/cluster_sphere/paths.py | 31 ++ .../tools/preflight/cluster_sphere/rdma_bw.py | 200 +++++++++ .../tools/preflight/cluster_sphere/report.py | 90 ++++ primus/tools/preflight/preflight_args.py | 17 + primus/tools/preflight/preflight_perf_test.py | 18 + tests/test_preflight_cluster_sphere.py | 72 ++++ 11 files changed, 873 insertions(+), 2 deletions(-) create mode 100644 primus/tools/preflight/cluster_sphere/__init__.py create mode 100644 primus/tools/preflight/cluster_sphere/env_recommender.py create mode 100644 primus/tools/preflight/cluster_sphere/paths.py create mode 100644 primus/tools/preflight/cluster_sphere/rdma_bw.py create mode 100644 primus/tools/preflight/cluster_sphere/report.py create mode 100644 tests/test_preflight_cluster_sphere.py diff --git a/docs/preflight.md b/docs/preflight.md index b4290e04e..ebbbb9fb0 100644 --- a/docs/preflight.md +++ b/docs/preflight.md @@ -53,6 +53,15 @@ Selection: - `--network`: network info - `--perf-test`: run perf tests only (GEMM + comm). This is slower. +Cluster Sphere (built into Primus under `primus/tools/preflight/cluster_sphere/`): +- `--cluster-sphere`: enable **both** Cluster Sphere behaviors below when dependencies are present. +- `--cluster-sphere-env`: add **RDMA env recommendations** (NCCL/GLOO/rocSHMEM export hints from local verbs devices) to the **info** report (`preflight_report*.md`). +- `--cluster-sphere-rdma-bw`: append **Verbs `ib_write_bw`** results to the **perf** report (`*_perf.md`). Intended for **`WORLD_SIZE=2`** only (two processes, typically one per node). Requires [linux-rdma/perftest](https://github.com/linux-rdma/perftest) (`ib_write_bw` on `PATH`). Optional: set `PRIMUS_IB_WRITE_BW_PORT` (default `2000`). + +Optional override: set **`PRIMUS_CLUSTER_SPHERE_ROOT`** to a directory only if you need to point at a custom/fork copy of the integration (defaults to the in-tree package path). + +Preflight’s existing perf tests measure **PyTorch / NCCL-style** GPU communication and use sysfs link rate as a roofline; they are **not** a substitute for raw Verbs bandwidth. Cluster Sphere **`ib_write_bw`** validates the **host RDMA path** separately. + Reporting: - `--dump-path`: output directory (default: `output/preflight`) - `--report-file-name`: base report name (default: `preflight_report`) @@ -69,10 +78,11 @@ Backward compatibility: By default, outputs are written under `output/preflight`. Typical report files: -- `preflight_report.md` / `preflight_report.pdf`: **info report** (host/GPU/network) -- `preflight_report_perf.md` / `preflight_report_perf.pdf`: **perf report** (GEMM + comm tests) +- `preflight_report.md` / `preflight_report.pdf`: **info report** (host/GPU/network, plus Cluster Sphere env section when enabled) +- `preflight_report_perf.md` / `preflight_report_perf.pdf`: **perf report** (GEMM + comm tests, plus optional `ib_write_bw` section) ## Notes - For multi-node runs, use `primus-cli slurm …` (or your preferred launcher) so distributed environment variables are set correctly. - If you only want a quick environment snapshot, prefer `--host --gpu --network`. +- `--perf-test` skips the info report; use `--cluster-sphere-env` without `--perf-test` if you only need RDMA export hints. diff --git a/primus/cli/subcommands/preflight.py b/primus/cli/subcommands/preflight.py index fc3c581d8..9bb4f91e8 100644 --- a/primus/cli/subcommands/preflight.py +++ b/primus/cli/subcommands/preflight.py @@ -13,6 +13,7 @@ primus-cli preflight --gpu # GPU info only primus-cli preflight --network # Network info only primus-cli preflight --perf-test # Perf tests only (skip info) + primus-cli preflight --cluster-sphere # Optional Cluster Sphere RDMA env + ib_write_bw """ from __future__ import annotations @@ -43,6 +44,7 @@ def register_subcommand(subparsers): primus-cli preflight --gpu # GPU info only primus-cli preflight --network # Network info only primus-cli preflight --perf-test # Perf only + primus-cli preflight --cluster-sphere # Cluster Sphere env + ib_write_bw """ from primus.tools.preflight.preflight_args import add_preflight_parser diff --git a/primus/tools/preflight/README.md b/primus/tools/preflight/README.md index d7ca011f0..cbc223111 100644 --- a/primus/tools/preflight/README.md +++ b/primus/tools/preflight/README.md @@ -13,6 +13,20 @@ primus-cli preflight \ --report-file-name preflight_report ``` +### Cluster Sphere (optional) + +RDMA environment hints (NCCL/GLOO/rocSHMEM) from local InfiniBand/RoCE devices, plus optional Verbs `ib_write_bw` on two ranks: + +```bash +# Info report: export hints only +primus-cli preflight --cluster-sphere-env --dump-path output/preflight + +# Full preflight + Cluster Sphere env + ib_write_bw on the perf report (use WORLD_SIZE=2 for the Verbs test) +primus-cli slurm srun -N 2 --ntasks-per-node=1 -- primus-cli preflight --cluster-sphere +``` + +Cluster Sphere logic ships inside Primus; set **`PRIMUS_CLUSTER_SPHERE_ROOT`** only if overriding that path. Install **perftest** for `ib_write_bw`. + Slurm (multi-node example): ```bash diff --git a/primus/tools/preflight/cluster_sphere/__init__.py b/primus/tools/preflight/cluster_sphere/__init__.py new file mode 100644 index 000000000..68b0fbd27 --- /dev/null +++ b/primus/tools/preflight/cluster_sphere/__init__.py @@ -0,0 +1,11 @@ +############################################################################### +# Copyright (c) 2025, Advanced Micro Devices, Inc. All rights reserved. +# +# See LICENSE for license information. +############################################################################### + +"""Cluster Sphere RDMA helpers for Primus Preflight (in-tree under this package).""" + +from primus.tools.preflight.cluster_sphere.paths import resolve_cluster_sphere_root + +__all__ = ["resolve_cluster_sphere_root"] diff --git a/primus/tools/preflight/cluster_sphere/env_recommender.py b/primus/tools/preflight/cluster_sphere/env_recommender.py new file mode 100644 index 000000000..7f4b1d6de --- /dev/null +++ b/primus/tools/preflight/cluster_sphere/env_recommender.py @@ -0,0 +1,406 @@ +############################################################################### +# Copyright (c) 2025, Advanced Micro Devices, Inc. All rights reserved. +# +# See LICENSE for license information. +# +# RDMA device scan and NCCL environment guidance (ROCm-oriented; in-tree). +############################################################################### + +from __future__ import annotations + +import glob +import os +import re +import subprocess +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional, Set, Tuple + +from primus.tools.preflight.network.info import Finding + +LIB_SEARCH_PATHS = [ + "/usr/lib", + "/usr/lib64", + "/usr/lib/x86_64-linux-gnu", + "/usr/local/lib", + "/etc/libibverbs.d", +] + + +@dataclass +class _DeviceInfo: + rdma: str + pci: str + netdev: str + firmware: str + gid_index: str + gid_value: str + vendor: str + + +@dataclass +class ClusterSphereEnvResult: + """Structured output for preflight report and Findings.""" + + devices: List[Dict[str, Any]] = field(default_factory=list) + warnings: List[str] = field(default_factory=list) + firmware_by_version: Dict[str, List[str]] = field(default_factory=dict) + nccl_exports: List[str] = field(default_factory=list) + rocshmem_exports: List[str] = field(default_factory=list) + docker_cmd: str = "" + socket_ifname: str = "" + + +class EnvRecommenderEngine: + """ + RDMA device scan and environment recommendations (no stdout; results in + ClusterSphereEnvResult + Findings). + """ + + def __init__(self) -> None: + self._devices: List[_DeviceInfo] = [] + + @property + def devices(self) -> List[Dict[str, Any]]: + return [d.__dict__ for d in self._devices] + + def scan_rdma_devices(self) -> bool: + self._devices = [] + rdma_paths = sorted(glob.glob("/sys/class/infiniband/*")) + if not rdma_paths: + return False + + for path in rdma_paths: + rdma = os.path.basename(path) + pci = self._get_pci_device(path) + netdev = self._find_netdev_for_pci(pci) + ibv_out = self._ibv_devinfo(rdma) + firmware = self._get_firmware_version(ibv_out) + gid_index, gid_value = self._get_gid_info(ibv_out) + vendor = self._rdma_vendor_from_pci(pci) + self._devices.append( + _DeviceInfo( + rdma=rdma, + pci=pci, + netdev=netdev, + firmware=firmware, + gid_index=gid_index, + gid_value=gid_value, + vendor=vendor, + ) + ) + return bool(self._devices) + + def _get_pci_device(self, device_path: str) -> str: + try: + link = os.path.join(device_path, "device") + if os.path.islink(link): + return os.path.basename(os.readlink(link)) + except OSError: + pass + return "UNKNOWN_PCI" + + def _find_netdev_for_pci(self, target_pci: str) -> str: + for netdev in glob.glob("/sys/class/net/*"): + try: + link = os.path.join(netdev, "device") + if os.path.islink(link) and os.path.basename(os.readlink(link)) == target_pci: + return os.path.basename(netdev) + except OSError: + pass + return "NO_NETDEV" + + def get_socket_ifname_value(self) -> str: + try: + out = subprocess.check_output("ip route show default | awk '{print $5}'", shell=True, text=True).strip() + if not out: + return "NA" + ifnames = list(dict.fromkeys(out.splitlines())) + return ifnames[0] if ifnames else "NA" + except Exception: + return "NA" + + def _rdma_vendor_from_pci(self, pci: str) -> str: + try: + out = subprocess.check_output(["lspci", "-s", pci, "-nn"], text=True).lower() + if "pensando" in out: + return "AINIC" + if "broadcom" in out: + return "BNXT" + if "mellanox" in out: + return "MLNX" + except Exception: + pass + return "UNKNOWN" + + def _ibv_devinfo(self, rdma: str) -> str: + try: + result = subprocess.run( + ["ibv_devinfo", "-d", rdma, "-v"], + capture_output=True, + text=True, + timeout=10, + ) + if result.returncode == 0: + return result.stdout + except Exception: + pass + return "" + + def _get_firmware_version(self, output: str) -> str: + for line in output.splitlines(): + line = line.strip() + if line.startswith("fw_ver:"): + return line.split("fw_ver:", 1)[1].strip() + return "UNKNOWN" + + def _get_gid_info(self, output: str) -> Tuple[str, str]: + for line in output.splitlines(): + if "::ffff:" in line and "GID[" in line: + idx = re.search(r"GID\[\s*(\d+)\]", line) + ip = re.search(r"(::ffff:[0-9.]+)", line) + if idx and ip: + return idx.group(1), ip.group(1) + return "-", "N/A" + + def _find_lib(self, patterns: List[str]) -> Optional[str]: + for base in LIB_SEARCH_PATHS: + for pat in patterns: + matches = glob.glob(os.path.join(base, "**", pat), recursive=True) + if matches: + return matches[0] + return None + + def _find_all_libs(self, patterns: List[str]) -> List[str]: + found: Set[str] = set() + for base in LIB_SEARCH_PATHS: + for pat in patterns: + found.update(glob.glob(os.path.join(base, "**", pat), recursive=True)) + return sorted(found) + + def _docker_cmd_bnxt(self, warnings: List[str]) -> str: + bnxt_rdma = self._find_lib(["libbnxt_re-rdmav*.so"]) + rdmacm = self._find_lib(["librdmacm.so.1"]) + ibverbs = self._find_lib(["libibverbs.so.1"]) + libnl3 = self._find_lib(["libnl-3.so.200"]) + libnl3_router = self._find_lib(["libnl-route-3.so.200"]) + + if not bnxt_rdma: + warnings.append("Docker (BNXT): Missing libbnxt_re-rdma*.so") + if not rdmacm: + warnings.append("Docker (BNXT): Missing librdmacm.so") + if not libnl3: + warnings.append("Docker (BNXT): Missing libnl-3.so") + + lines = [ + "docker run --rm -it \\", + " --device /dev/dri \\", + " --device /dev/infiniband \\", + " --device /dev/kfd \\", + " --network host \\", + " --ipc host \\", + " --privileged \\", + " --ulimit memlock=-1:-1 \\", + " --group-add video \\", + " --cap-add SYS_PTRACE \\", + " --security-opt seccomp=unconfined \\", + " --shm-size 64G \\", + " -v /sys:/sys \\", + " -v $HOME/.ssh:/root/.ssh \\", + " -v $HOME:$HOME \\", + " -v /dev/infiniband:/dev/infiniband \\", + " -v /sys/class/infiniband:/sys/class/infiniband:ro \\", + " -v /sys/class/net:/sys/class/net:ro \\", + " -v /sys/bus/pci:/sys/bus/pci:ro \\", + " -v /etc/libibverbs.d:/etc/libibverbs.d:ro \\", + " -v /etc/rdma:/etc/rdma:ro \\", + ] + if bnxt_rdma: + lines.append(f" -v {bnxt_rdma}:{bnxt_rdma}:ro \\") + if rdmacm: + lines.append(f" -v {rdmacm}:{rdmacm}:ro \\") + if ibverbs: + lines.append(f" -v {ibverbs}:{ibverbs}:ro \\") + if libnl3: + lines.append(f" -v {libnl3}:{libnl3}:ro \\") + if libnl3_router: + lines.append(f" -v {libnl3_router}:{libnl3_router}:ro \\") + lines.append(" ") + return "\n".join(lines) + + def _docker_cmd_mlnx(self) -> str: + return ( + "docker run --rm -it \\\n" + " --device /dev/dri \\\n" + " --device /dev/infiniband \\\n" + " --device /dev/kfd \\\n" + " --network host \\\n" + " --ipc host \\\n" + " --privileged \\\n" + " --ulimit memlock=-1:-1 \\\n" + " --group-add video \\\n" + " --cap-add SYS_PTRACE \\\n" + " --security-opt seccomp=unconfined \\\n" + " --shm-size 64G \\\n" + " -v /sys:/sys \\\n" + " -v $HOME/.ssh:/root/.ssh \\\n" + " -v $HOME:$HOME \\\n" + " -v /dev/infiniband:/dev/infiniband \\\n" + " -v /sys/class/infiniband:/sys/class/infiniband:ro \\\n" + " -v /sys/class/net:/sys/class/net:ro \\\n" + " -v /sys/bus/pci:/sys/bus/pci:ro \\\n" + " " + ) + + def _docker_cmd_ionic(self, warnings: List[str]) -> str: + ionic_rdma = self._find_lib(["libionic-rdmav*.so"]) + ionic_so = self._find_all_libs(["libionic.so*"]) + ionic_driver = self._find_lib(["ionic.driver"]) + + if not ionic_rdma: + warnings.append("Docker (AINIC): Missing libionic-rdma*.so") + if not ionic_so: + warnings.append("Docker (AINIC): Missing libionic.so") + if not ionic_driver: + warnings.append("Docker (AINIC): Missing ionic.driver") + + lines = [ + "docker run --rm -it \\", + " --device /dev/dri \\", + " --device /dev/infiniband \\", + " --device /dev/kfd \\", + " --network host \\", + " --ipc host \\", + " --privileged \\", + " --ulimit memlock=-1:-1 \\", + " --group-add video \\", + " --cap-add SYS_PTRACE \\", + " --security-opt seccomp=unconfined \\", + " --shm-size 64G \\", + " -v /sys:/sys \\", + " -v $HOME/.ssh:/root/.ssh \\", + " -v $HOME:$HOME \\", + " -v /dev/infiniband:/dev/infiniband \\", + " -v /sys/class/infiniband:/sys/class/infiniband:ro \\", + " -v /sys/class/net:/sys/class/net:ro \\", + " -v /sys/bus/pci:/sys/bus/pci:ro \\", + ] + if ionic_rdma: + lines.append(f" -v {ionic_rdma}:{ionic_rdma}:ro \\") + for so_file in ionic_so: + lines.append(f" -v {so_file}:{so_file}:ro \\") + if ionic_driver: + lines.append(f" -v {ionic_driver}:{ionic_driver}:ro \\") + lines.append(" ") + return "\n".join(lines) + + def generate_docker_launch_command(self, warnings: List[str]) -> str: + vendors = {d.vendor for d in self._devices} + if len(vendors) > 1: + warnings.append("Multiple RDMA vendors detected; verify Docker snippet manually.") + + if "AINIC" in vendors: + return self._docker_cmd_ionic(warnings) + if "BNXT" in vendors: + return self._docker_cmd_bnxt(warnings) + if "MLNX" in vendors: + return self._docker_cmd_mlnx() + + warnings.append("Vendor UNKNOWN or unsupported for auto-generated Docker command.") + return "" + + def build_nccl_exports(self, warnings: List[str]) -> List[str]: + exports = ["export NCCL_IGNORE_CPU_AFFINITY=1"] + + gid_numeric: List[int] = [] + for d in self._devices: + if str(d.gid_index).isdigit(): + gid_numeric.append(int(d.gid_index)) + + unique_gid = sorted(set(gid_numeric)) + if len(unique_gid) > 1: + warnings.append("Multiple GID indices detected; verify NCCL_IB_GID_INDEX against detailed device table.") + + if unique_gid: + exports.append(f"export NCCL_IB_GID_INDEX={max(unique_gid)}") + else: + exports.append("export NCCL_IB_GID_INDEX=0") + + firmware_version = max((d.firmware for d in self._devices), default="UNKNOWN") + parts: List[str] = [] + for d in self._devices: + part = d.rdma if str(d.gid_index).isdigit() and d.firmware == firmware_version else "" + if part: + parts.append(part) + nccl_hca = ",".join(parts) if parts else ",".join(d.rdma for d in self._devices) + exports.append(f"export NCCL_IB_HCA={nccl_hca}") + + socket_ifname = self.get_socket_ifname_value() + exports.append(f"export NCCL_SOCKET_IFNAME={socket_ifname}") + exports.append(f"export GLOO_SOCKET_IFNAME={socket_ifname}") + return exports + + def build_rocshmem_exports(self) -> List[str]: + return [ + "export ROCSHMEM_HEAP_SIZE=7524589824", + "export ROCSHMEM_MAX_NUM_CONTEXTS=256", + ] + + def build_result(self) -> ClusterSphereEnvResult: + warnings: List[str] = [] + result = ClusterSphereEnvResult(warnings=list(warnings)) + + if not self.scan_rdma_devices(): + warnings.append("No RDMA devices found under /sys/class/infiniband.") + result.warnings = warnings + return result + + fw_map: Dict[str, List[str]] = {} + for d in self._devices: + fw_map.setdefault(d.firmware, []).append(d.rdma) + result.devices = self.devices + result.firmware_by_version = fw_map + if len(fw_map) > 1: + warnings.append("Multiple firmware versions detected — standardization recommended.") + + result.nccl_exports = self.build_nccl_exports(warnings) + result.rocshmem_exports = self.build_rocshmem_exports() + result.socket_ifname = self.get_socket_ifname_value() + result.docker_cmd = self.generate_docker_launch_command(warnings) + result.warnings = warnings + return result + + +def collect_cluster_sphere_env_findings() -> List[Finding]: + """Run env recommender on the local host; returns Findings for preflight aggregation.""" + engine = EnvRecommenderEngine() + result = engine.build_result() + + findings: List[Finding] = [] + + if not result.devices: + findings.append( + Finding( + "warn", + "Cluster Sphere env recommender: no RDMA devices found", + {"warnings": result.warnings}, + ) + ) + return findings + + findings.append( + Finding( + "info", + "Cluster Sphere RDMA environment recommendations", + { + "devices": result.devices, + "nccl_exports": result.nccl_exports, + "rocshmem_exports": result.rocshmem_exports, + "docker_cmd": result.docker_cmd, + "socket_ifname": result.socket_ifname, + "firmware_by_version": result.firmware_by_version, + "warnings": result.warnings, + }, + ) + ) + + return findings diff --git a/primus/tools/preflight/cluster_sphere/paths.py b/primus/tools/preflight/cluster_sphere/paths.py new file mode 100644 index 000000000..84173b2d5 --- /dev/null +++ b/primus/tools/preflight/cluster_sphere/paths.py @@ -0,0 +1,31 @@ +############################################################################### +# Copyright (c) 2025, Advanced Micro Devices, Inc. All rights reserved. +# +# See LICENSE for license information. +############################################################################### + +from __future__ import annotations + +import os +from pathlib import Path + + +def resolve_cluster_sphere_root() -> Path: + """ + Return the Primus Cluster Sphere integration root: the directory + ``primus/tools/preflight/cluster_sphere/`` (this package). + + RDMA env recommendations and ``ib_write_bw`` orchestration are implemented + in Python under that tree. + + **Override (optional):** if ``PRIMUS_CLUSTER_SPHERE_ROOT`` is set to an + existing directory, that path is returned instead (e.g. fork or vendor + staging). ``DIST_INF_COOKBOOK_ROOT`` is no longer read. + """ + env_override = os.environ.get("PRIMUS_CLUSTER_SPHERE_ROOT", "").strip() + if env_override: + p = Path(env_override).expanduser().resolve() + if p.is_dir(): + return p + + return Path(__file__).resolve().parent diff --git a/primus/tools/preflight/cluster_sphere/rdma_bw.py b/primus/tools/preflight/cluster_sphere/rdma_bw.py new file mode 100644 index 000000000..287ef852c --- /dev/null +++ b/primus/tools/preflight/cluster_sphere/rdma_bw.py @@ -0,0 +1,200 @@ +############################################################################### +# Copyright (c) 2025, Advanced Micro Devices, Inc. All rights reserved. +# +# See LICENSE for license information. +############################################################################### + +from __future__ import annotations + +import os +import re +import shutil +import subprocess +import threading +import time +from typing import Any, Dict, List, Optional + +import torch.distributed as dist + +from primus.tools.preflight.global_vars import RANK, WORLD_SIZE + + +def _first_ib_device_name() -> Optional[str]: + ib_path = "/sys/class/infiniband" + try: + names = sorted(os.listdir(ib_path)) + return names[0] if names else None + except OSError: + return None + + +def _resolve_server_host() -> str: + return os.environ.get("MASTER_ADDR", "localhost").strip() or "localhost" + + +def _parse_peak_gbps(text: str) -> Optional[float]: + """Best-effort peak Gb/sec from ib_write_bw stdout/stderr.""" + best: Optional[float] = None + for line in text.splitlines(): + for m in re.finditer(r"(\d+\.\d+)\s*Gb/sec", line, re.I): + v = float(m.group(1)) + best = v if best is None else max(best, v) + for m in re.finditer(r"(\d+\.\d+)\s*GB/sec", line): + v = float(m.group(1)) + best = v if best is None else max(best, v) + return best + + +def append_cluster_sphere_verbs_rdma_section(args: Any, markdown_file: str) -> None: + """ + Run ib_write_bw between two ranks (WORLD_SIZE == 2) and append results to the perf markdown. + Server is rank 0; client is rank 1 connecting to MASTER_ADDR. + """ + from primus.tools.preflight.cluster_sphere.report import wants_cluster_sphere_rdma_bw + + if not wants_cluster_sphere_rdma_bw(args): + return + + lines: List[str] = ["\n---\n\n## Cluster Sphere — Verbs RDMA (ib_write_bw)\n\n"] + + if WORLD_SIZE != 2: + lines.append( + f"This check requires exactly **two** distributed processes (`WORLD_SIZE=2`). " + f"Current world size: **{WORLD_SIZE}**. Skipping `ib_write_bw`.\n\n" + ) + if RANK == 0: + _append_lines(markdown_file, lines) + return + + if shutil.which("ib_write_bw") is None: + lines.append( + "`ib_write_bw` was not found in PATH (install " + "[linux-rdma/perftest](https://github.com/linux-rdma/perftest)).\n\n" + ) + if RANK == 0: + _append_lines(markdown_file, lines) + return + + ib_dev = _first_ib_device_name() + if not ib_dev: + lines.append("No InfiniBand / RDMA device under `/sys/class/infiniband`; skipping.\n\n") + if RANK == 0: + _append_lines(markdown_file, lines) + return + + port = int(os.environ.get("PRIMUS_IB_WRITE_BW_PORT", "2000")) + server_host = _resolve_server_host() + + server_holder: Dict[str, Any] = {} + + def server_main() -> None: + cmd = [ + "ib_write_bw", + "-d", + ib_dev, + "-q", + "4", + "-a", + "--report_gbits", + "-F", + "-p", + str(port), + ] + try: + proc = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=180, + ) + server_holder["out"] = (proc.stdout or "") + "\n" + (proc.stderr or "") + server_holder["rc"] = proc.returncode + except subprocess.TimeoutExpired as e: + server_holder["out"] = str(e) + server_holder["rc"] = -1 + + th: Optional[threading.Thread] = None + if RANK == 0: + th = threading.Thread(target=server_main, daemon=True) + th.start() + time.sleep(2) + + dist.barrier() + + client_payload: Optional[Dict[str, Any]] = None + if RANK == 1: + time.sleep(5) + cmd = [ + "ib_write_bw", + "-d", + ib_dev, + "-q", + "4", + "-a", + "--report_gbits", + "-F", + server_host, + "-p", + str(port), + ] + try: + proc = subprocess.run(cmd, capture_output=True, text=True, timeout=120) + out = (proc.stdout or "") + "\n" + (proc.stderr or "") + peak = _parse_peak_gbps(out) + client_payload = {"stdout": out, "rc": proc.returncode, "peak_gbps": peak} + except subprocess.TimeoutExpired as e: + client_payload = {"stdout": str(e), "rc": -1, "peak_gbps": None} + + dist.barrier() + + if RANK == 0 and th is not None: + th.join(timeout=200) + + local_obj = client_payload if RANK == 1 else None + gathered_objs: Optional[List[Any]] = None + if RANK == 0: + gathered_objs = [None] * WORLD_SIZE + dist.gather_object(local_obj, object_gather_list=gathered_objs, dst=0) + else: + dist.gather_object(local_obj, dst=0) + + if RANK == 0: + client = None + if gathered_objs is not None and len(gathered_objs) > 1: + client = gathered_objs[1] + + peak = None + client_txt = "" + if isinstance(client, dict): + peak = client.get("peak_gbps") + client_txt = str(client.get("stdout") or "") + + if peak is None: + peak = _parse_peak_gbps(client_txt) + + lines.append(f"- **Device**: `{ib_dev}`\n") + lines.append(f"- **Server**: `{server_host}:{port}` (rank 0)\n") + lines.append("- **Client**: rank 1\n") + if peak is not None: + lines.append(f"- **Peak bandwidth (parsed)**: **{peak:.2f} Gb/sec**\n") + else: + lines.append("- **Peak bandwidth**: could not parse (see raw output below)\n") + lines.append("\n") + + if client_txt: + lines.append("
Client output\n\n```\n") + lines.append(client_txt[:12000]) + lines.append("\n```\n\n
\n\n") + + so = server_holder.get("out") + if so: + lines.append("
Server output\n\n```\n") + lines.append(str(so)[:12000]) + lines.append("\n```\n\n
\n\n") + + _append_lines(markdown_file, lines) + + +def _append_lines(path: str, lines: List[str]) -> None: + with open(path, "a", encoding="utf-8") as f: + f.write("".join(lines)) diff --git a/primus/tools/preflight/cluster_sphere/report.py b/primus/tools/preflight/cluster_sphere/report.py new file mode 100644 index 000000000..da6a6763b --- /dev/null +++ b/primus/tools/preflight/cluster_sphere/report.py @@ -0,0 +1,90 @@ +############################################################################### +# Copyright (c) 2025, Advanced Micro Devices, Inc. All rights reserved. +# +# See LICENSE for license information. +############################################################################### + +from __future__ import annotations + +from typing import Any, Dict, List + + +def wants_cluster_sphere_env(args: Any) -> bool: + return bool(getattr(args, "cluster_sphere", False) or getattr(args, "cluster_sphere_env", False)) + + +def wants_cluster_sphere_rdma_bw(args: Any) -> bool: + return bool(getattr(args, "cluster_sphere", False) or getattr(args, "cluster_sphere_rdma_bw", False)) + + +def write_cluster_sphere_env_section(f, gathered: List[Dict[str, Any]]) -> None: + """Append Cluster Sphere env recommender markdown (rank0).""" + f.write("## Cluster Sphere — RDMA environment recommendations\n\n") + + blocks = 0 + for r in gathered: + host = str(r.get("host", "unknown")) + for fin in r.get("findings", []): + if not isinstance(fin, dict): + continue + msg = fin.get("message", "") + if msg == "Cluster Sphere RDMA environment recommendations": + blocks += 1 + details = fin.get("details") or {} + _write_one_host_block(f, host, details) + elif msg == "Cluster Sphere env recommender: no RDMA devices found": + blocks += 1 + f.write(f"### Host `{host}`\n\n") + f.write("_No RDMA devices detected for Cluster Sphere env recommender._\n\n") + w = (fin.get("details") or {}).get("warnings") or [] + if w: + f.write("Warnings:\n\n") + for line in w: + f.write(f"- {line}\n") + f.write("\n") + + if blocks == 0: + f.write("_No Cluster Sphere environment data was collected._\n\n") + + +def _write_one_host_block(f, host: str, details: Dict[str, Any]) -> None: + f.write(f"### Host `{host}`\n\n") + + warns = details.get("warnings") or [] + if warns: + f.write("**Warnings:**\n\n") + for w in warns: + f.write(f"- {w}\n") + f.write("\n") + + devices = details.get("devices") or [] + if devices: + f.write("| RDMA | PCI | NETDEV | Firmware | GID idx | GID | Vendor |\n") + f.write("|------|-----|--------|----------|---------|-----|--------|\n") + for d in devices: + f.write( + f"| {d.get('rdma','')} | {d.get('pci','')} | {d.get('netdev','')} | " + f"{d.get('firmware','')} | {d.get('gid_index','')} | {d.get('gid_value','')} | " + f"{d.get('vendor','')} |\n" + ) + f.write("\n") + + nccl = details.get("nccl_exports") or [] + if nccl: + f.write("**Suggested NCCL / socket exports:**\n\n```bash\n") + for line in nccl: + f.write(f"{line}\n") + f.write("```\n\n") + + roc = details.get("rocshmem_exports") or [] + if roc: + f.write("**Suggested rocSHMEM exports:**\n\n```bash\n") + for line in roc: + f.write(f"{line}\n") + f.write("```\n\n") + + docker_cmd = (details.get("docker_cmd") or "").strip() + if docker_cmd: + f.write("**Example Docker launch (vendor-specific template):**\n\n```bash\n") + f.write(docker_cmd) + f.write("\n```\n\n") diff --git a/primus/tools/preflight/preflight_args.py b/primus/tools/preflight/preflight_args.py index a816b2b47..40c0e8179 100644 --- a/primus/tools/preflight/preflight_args.py +++ b/primus/tools/preflight/preflight_args.py @@ -88,4 +88,21 @@ def add_preflight_parser(parser: argparse.ArgumentParser) -> argparse.ArgumentPa action="store_false", help="Disable PDF report generation.", ) + + # Cluster Sphere (in-tree): RDMA env recommender + optional Verbs ib_write_bw + parser.add_argument( + "--cluster-sphere", + action="store_true", + help="Enable Cluster Sphere RDMA env recommendations (info report) and Verbs ib_write_bw (perf report).", + ) + parser.add_argument( + "--cluster-sphere-env", + action="store_true", + help="Cluster Sphere only: NCCL/GLOO/rocSHMEM export hints from local RDMA devices (info report).", + ) + parser.add_argument( + "--cluster-sphere-rdma-bw", + action="store_true", + help="Cluster Sphere only: run ib_write_bw between two ranks (requires WORLD_SIZE=2; perf report).", + ) return parser diff --git a/primus/tools/preflight/preflight_perf_test.py b/primus/tools/preflight/preflight_perf_test.py index e3eb2607a..a575af51c 100644 --- a/primus/tools/preflight/preflight_perf_test.py +++ b/primus/tools/preflight/preflight_perf_test.py @@ -19,6 +19,9 @@ from primus.tools.preflight.inter_node_comm_p2p import run_inter_node_comm_p2p from primus.tools.preflight.inter_node_ring_p2p import run_inter_node_ring_p2p from primus.tools.preflight.intra_node_comm import run_intra_node_comm +from primus.tools.preflight.cluster_sphere.env_recommender import collect_cluster_sphere_env_findings +from primus.tools.preflight.cluster_sphere.paths import resolve_cluster_sphere_root +from primus.tools.preflight.cluster_sphere.report import wants_cluster_sphere_env from primus.tools.preflight.network.info import ( collect_network_info, write_network_report, @@ -99,6 +102,10 @@ def run_preflight_info(args: Any, expect_distributed: bool = True) -> int: for f in collect_network_info(expect_distributed=expect_distributed): findings.append(Finding(level=f.level, message=f.message, details=f.details)) + if wants_cluster_sphere_env(args): + for f in collect_cluster_sphere_env_findings(): + findings.append(Finding(level=f.level, message=f.message, details=f.details)) + fail_count = sum(1 for x in findings if x.level == "fail") warn_count = sum(1 for x in findings if x.level == "warn") info_count = sum(1 for x in findings if x.level == "info") @@ -165,6 +172,13 @@ def run_preflight_info(args: Any, expect_distributed: bool = True) -> int: if check_network: write_network_report(f, by_host) + from primus.tools.preflight.cluster_sphere.report import write_cluster_sphere_env_section + + if wants_cluster_sphere_env(args): + write_cluster_sphere_env_section(f, gathered) + cs_root = resolve_cluster_sphere_root() + f.write(f"\n*Cluster Sphere integration path (Primus): `{cs_root}`.*\n\n") + if save_pdf: try: from primus.tools.preflight.utility import md_to_pdf as _md_to_pdf @@ -318,6 +332,10 @@ def _append_dist_init_failure(markdown_file: str, timeout_sec: int, err: Excepti run_inter_node_comm_p2p(args) run_inter_node_ring_p2p(args) + from primus.tools.preflight.cluster_sphere.rdma_bw import append_cluster_sphere_verbs_rdma_section + + append_cluster_sphere_verbs_rdma_section(args, args.markdown_file) + if RANK == 0 and args.save_pdf: md_to_pdf(args.markdown_file, args.pdf_file) diff --git a/tests/test_preflight_cluster_sphere.py b/tests/test_preflight_cluster_sphere.py new file mode 100644 index 000000000..c5b5e86e4 --- /dev/null +++ b/tests/test_preflight_cluster_sphere.py @@ -0,0 +1,72 @@ +############################################################################### +# Copyright (c) 2025, Advanced Micro Devices, Inc. All rights reserved. +# +# See LICENSE for license information. +############################################################################### + +from __future__ import annotations + +from pathlib import Path +from unittest.mock import patch + + +def test_resolve_default_is_primus_package(): + import primus.tools.preflight.cluster_sphere.paths as paths_mod + + expected = Path(paths_mod.__file__).resolve().parent + assert paths_mod.resolve_cluster_sphere_root() == expected + + +def test_resolve_override_primus_cluster_sphere_env(tmp_path, monkeypatch): + alt = tmp_path / "override" + alt.mkdir() + monkeypatch.setenv("PRIMUS_CLUSTER_SPHERE_ROOT", str(alt.resolve())) + + from primus.tools.preflight.cluster_sphere.paths import resolve_cluster_sphere_root + + assert resolve_cluster_sphere_root() == alt.resolve() + + +def test_resolve_override_must_exist(tmp_path, monkeypatch): + monkeypatch.setenv("PRIMUS_CLUSTER_SPHERE_ROOT", str(tmp_path / "nonexistent")) + + import primus.tools.preflight.cluster_sphere.paths as paths_mod + + expected = Path(paths_mod.__file__).resolve().parent + assert paths_mod.resolve_cluster_sphere_root() == expected + + +def test_env_recommender_nccl_exports_no_devices(): + from primus.tools.preflight.cluster_sphere import env_recommender as er + + eng = er.EnvRecommenderEngine() + with patch.object(er.glob, "glob", return_value=[]): + result = eng.build_result() + assert result.devices == [] + assert "No RDMA devices" in "".join(result.warnings) + + +def test_collect_findings_warns_when_empty_sysfs(): + from primus.tools.preflight.cluster_sphere import env_recommender as er + from primus.tools.preflight.cluster_sphere.env_recommender import collect_cluster_sphere_env_findings + + with patch.object(er.glob, "glob", return_value=[]): + findings = collect_cluster_sphere_env_findings() + assert len(findings) == 1 + assert findings[0].level == "warn" + + +def test_nccl_exports_with_stub_devices(monkeypatch): + from primus.tools.preflight.cluster_sphere.env_recommender import EnvRecommenderEngine + from primus.tools.preflight.cluster_sphere.env_recommender import _DeviceInfo + + eng = EnvRecommenderEngine() + eng._devices = [ + _DeviceInfo("mlx5_0", "0000:01:00.0", "eth0", "fw1", "3", "::ffff:1.2.3.4", "MLNX") + ] + monkeypatch.setattr(EnvRecommenderEngine, "get_socket_ifname_value", lambda self: "eth0") + warns: list = [] + exports = eng.build_nccl_exports(warns) + joined = "\n".join(exports) + assert "NCCL_IB_HCA=mlx5_0" in joined + assert "NCCL_SOCKET_IFNAME=eth0" in joined From e2a1c82d4fb25fc455aba961ed5814eaf92f3baf Mon Sep 17 00:00:00 2001 From: lcskrishna Date: Tue, 28 Apr 2026 15:51:42 +0000 Subject: [PATCH 2/2] add clustersphere dist inf tools into primus --- cluster_sphere_env.md | 63 ++++++ docs/preflight.md | 59 ++++++ primus/tools/preflight/README.md | 78 +++++++- .../preflight/cluster_sphere/__main__.py | 185 ++++++++++++++++++ .../tools/preflight/cluster_sphere/rdma_bw.py | 71 ++----- .../tools/preflight/cluster_sphere/report.py | 25 +++ .../preflight/cluster_sphere/verbs_bw.py | 79 ++++++++ scripts/slurm/README.md | 26 +++ .../slurm/cluster_sphere_env_single_node.sh | 14 ++ .../cluster_sphere_ib_write_bw_two_node.sh | 50 +++++ tests/test_preflight_cluster_sphere.py | 62 +++++- 11 files changed, 653 insertions(+), 59 deletions(-) create mode 100644 cluster_sphere_env.md create mode 100644 primus/tools/preflight/cluster_sphere/__main__.py create mode 100644 primus/tools/preflight/cluster_sphere/verbs_bw.py create mode 100644 scripts/slurm/README.md create mode 100755 scripts/slurm/cluster_sphere_env_single_node.sh create mode 100755 scripts/slurm/cluster_sphere_ib_write_bw_two_node.sh diff --git a/cluster_sphere_env.md b/cluster_sphere_env.md new file mode 100644 index 000000000..74e945b32 --- /dev/null +++ b/cluster_sphere_env.md @@ -0,0 +1,63 @@ +## Cluster Sphere — RDMA environment recommendations + +### Host `useocpm2m-097-078` + +**Warnings:** + +- Multiple firmware versions detected — standardization recommended. + +| RDMA | PCI | NETDEV | Firmware | GID idx | GID | Vendor | +|------|-----|--------|----------|---------|-----|--------| +| mlx5_0 | 0000:0c:00.0 | rdma0 | 28.47.1900 | 3 | ::ffff:10.224.0.73 | MLNX | +| mlx5_1 | 0000:1f:00.0 | eth0 | 22.47.1088 | 3 | ::ffff:10.158.212.73 | MLNX | +| mlx5_2 | 0000:2a:00.0 | rdma1 | 28.47.1900 | 3 | ::ffff:10.224.4.73 | MLNX | +| mlx5_3 | 0000:41:00.0 | rdma2 | 28.47.1900 | 3 | ::ffff:10.224.8.73 | MLNX | +| mlx5_4 | 0000:58:00.0 | rdma3 | 28.47.1900 | 3 | ::ffff:10.224.12.73 | MLNX | +| mlx5_5 | 0000:86:00.0 | rdma4 | 28.47.1900 | 3 | ::ffff:10.224.16.73 | MLNX | +| mlx5_6 | 0000:9a:00.0 | eth1 | 22.47.1088 | - | N/A | MLNX | +| mlx5_7 | 0000:a5:00.0 | rdma5 | 28.47.1900 | 3 | ::ffff:10.224.20.73 | MLNX | +| mlx5_8 | 0000:bd:00.0 | rdma6 | 28.47.1900 | 3 | ::ffff:10.224.24.73 | MLNX | +| mlx5_9 | 0000:d5:00.0 | rdma7 | 28.47.1900 | 3 | ::ffff:10.224.28.73 | MLNX | + +**Suggested NCCL / socket exports:** + +```bash +export NCCL_IGNORE_CPU_AFFINITY=1 +export NCCL_IB_GID_INDEX=3 +export NCCL_IB_HCA=mlx5_0,mlx5_2,mlx5_3,mlx5_4,mlx5_5,mlx5_7,mlx5_8,mlx5_9 +export NCCL_SOCKET_IFNAME=eth0 +export GLOO_SOCKET_IFNAME=eth0 +``` + +**Suggested rocSHMEM exports:** + +```bash +export ROCSHMEM_HEAP_SIZE=7524589824 +export ROCSHMEM_MAX_NUM_CONTEXTS=256 +``` + +**Example Docker launch (vendor-specific template):** + +```bash +docker run --rm -it \ + --device /dev/dri \ + --device /dev/infiniband \ + --device /dev/kfd \ + --network host \ + --ipc host \ + --privileged \ + --ulimit memlock=-1:-1 \ + --group-add video \ + --cap-add SYS_PTRACE \ + --security-opt seccomp=unconfined \ + --shm-size 64G \ + -v /sys:/sys \ + -v $HOME/.ssh:/root/.ssh \ + -v $HOME:$HOME \ + -v /dev/infiniband:/dev/infiniband \ + -v /sys/class/infiniband:/sys/class/infiniband:ro \ + -v /sys/class/net:/sys/class/net:ro \ + -v /sys/bus/pci:/sys/bus/pci:ro \ + +``` + diff --git a/docs/preflight.md b/docs/preflight.md index ebbbb9fb0..d20eb25de 100644 --- a/docs/preflight.md +++ b/docs/preflight.md @@ -81,6 +81,65 @@ Typical report files: - `preflight_report.md` / `preflight_report.pdf`: **info report** (host/GPU/network, plus Cluster Sphere env section when enabled) - `preflight_report_perf.md` / `preflight_report_perf.pdf`: **perf report** (GEMM + comm tests, plus optional `ib_write_bw` section) +## Slurm: Cluster Sphere without preflight / torchrun + +Use this when you only need **host RDMA validation** and **NCCL-style export hints**, not full GPU preflight (no `torchrun`, no `primus-cli preflight` required). Run on **compute nodes** so `/sys/class/infiniband` and `ibv_devinfo` match real jobs. + +**Checklist** + +- Primus checkout on nodes (or bind-mount); set `PYTHONPATH` to the repo root that contains the `primus/` package. +- `ibv_devinfo` (often `rdma-core`), `lspci` for PCI vendor detection. +- For Pipeline B: [perftest](https://github.com/linux-rdma/perftest) (`ib_write_bw` on `PATH`). +- Your site may require Slurm `-t`, `-p`, `-A`, etc. + +### Pipeline A — RDMA env recommender (one node) + +Produces a NIC-oriented Markdown report (GID, per-device firmware, a grouped **firmware report** when multiple NICs exist, suggested `NCCL_*` / `GLOO_*` exports). + +```bash +export PRIMUS_ROOT=/path/to/Primus +export PYTHONPATH="${PRIMUS_ROOT}" + +srun -N 1 -n 1 -t 00:30:00 bash -lc ' + cd "${PRIMUS_ROOT}" && python3 -m primus.tools.preflight.cluster_sphere env --markdown \ + > cluster_sphere_env_${SLURM_JOB_ID:-local}.md +' +``` + +Or use the helper script: [`scripts/slurm/cluster_sphere_env_single_node.sh`](../scripts/slurm/cluster_sphere_env_single_node.sh). + +### Pipeline B — Verbs `ib_write_bw` (two nodes) + +Server listens on **node A**; client on **node B** connects to **`SERVER_RDMA_IP`** — the address of **A on the RDMA/RoCE network**, not necessarily the management DNS name. Obtain it from Pipeline A output / `ip` on the HCA netdev. + +Manual (two terminals or SSH into allocated nodes): + +```bash +# Node A +export PYTHONPATH=/path/to/Primus +python3 -m primus.tools.preflight.cluster_sphere verbs-server --device rdma0 + +# Node B (after server is up) +python3 -m primus.tools.preflight.cluster_sphere verbs-client --server-ip --device rdma0 +``` + +Port: `2000` or `PRIMUS_IB_WRITE_BW_PORT`. + +**Single `srun` (server + client)** — set `SERVER_RDMA_IP` to the **server host’s** RDMA address (task 0 / first node); task 1 runs the client after a short delay: + +```bash +export SERVER_RDMA_IP= +export PYTHONPATH=/path/to/Primus +srun -N 2 -n 2 -t 00:30:00 env PYTHONPATH="${PYTHONPATH}" SERVER_RDMA_IP="${SERVER_RDMA_IP}" \ + python3 -m primus.tools.preflight.cluster_sphere verbs-pair --device rdma0 +``` + +Automated helper (same allocation, two nodes): [`scripts/slurm/cluster_sphere_ib_write_bw_two_node.sh`](../scripts/slurm/cluster_sphere_ib_write_bw_two_node.sh) — set `SERVER_RDMA_IP` before running. + +See [`scripts/slurm/README.md`](../scripts/slurm/README.md) for examples. + +--- + ## Notes - For multi-node runs, use `primus-cli slurm …` (or your preferred launcher) so distributed environment variables are set correctly. diff --git a/primus/tools/preflight/README.md b/primus/tools/preflight/README.md index cbc223111..e6a3077e3 100644 --- a/primus/tools/preflight/README.md +++ b/primus/tools/preflight/README.md @@ -27,7 +27,83 @@ primus-cli slurm srun -N 2 --ntasks-per-node=1 -- primus-cli preflight --cluster Cluster Sphere logic ships inside Primus; set **`PRIMUS_CLUSTER_SPHERE_ROOT`** only if overriding that path. Install **perftest** for `ib_write_bw`. -Slurm (multi-node example): +### Cluster Sphere without torchrun (Slurm or local) + +Use the **module CLI** when you want RDMA checks **without** `torchrun`, `torch.distributed`, or `primus-cli preflight` (no GPU preflight / NCCL process group). Run on a **compute node** (via `srun` / `salloc`) when possible so verbs devices match real jobs—not only on the login node. + +**Setup (every invocation)** + +```bash +export PRIMUS_ROOT=/path/to/Primus # repo root containing the primus/ package +export PYTHONPATH="${PRIMUS_ROOT}${PYTHONPATH:+:${PYTHONPATH}}" +``` + +Add your site’s Slurm flags (`-t`, `-p`, `-A`, …) where shown. Optional: **`PRIMUS_CLUSTER_SPHERE_ROOT`** only if you override the in-tree integration path. + +**Pipeline A — RDMA env recommender (one host, NIC report)** + +Uses `ibv_devinfo` / sysfs (no GPU, no Torch). Writes Markdown with device table, a **firmware report** (grouped by NIC firmware version), and suggested NCCL/GLOO/rocSHMEM exports. + +```bash +# From a shell with PYTHONPATH set (local or already on a compute node): +python3 -m primus.tools.preflight.cluster_sphere env --markdown > cluster_sphere_env.md + +# Quick text summary to stderr instead of Markdown: +python3 -m primus.tools.preflight.cluster_sphere env + +# Slurm — one allocate step, write report in cwd (add -p/-A/-t as needed): +srun -N 1 -n 1 -t 00:30:00 bash -lc \ + 'cd "${HOME}" && python3 -m primus.tools.preflight.cluster_sphere env --markdown > cluster_sphere_env_${SLURM_JOB_ID:-local}.md' +``` + +Helper script (same idea; sets `PYTHONPATH` from `PRIMUS_ROOT`): +[`scripts/slurm/cluster_sphere_env_single_node.sh`](../../../scripts/slurm/cluster_sphere_env_single_node.sh) + +**Pipeline B — Verbs `ib_write_bw` (two hosts)** + +Requires [perftest](https://github.com/linux-rdma/perftest) (`ib_write_bw` on `PATH`). Default TCP port **`2000`** or set **`PRIMUS_IB_WRITE_BW_PORT`**. The client must use the server’s **`SERVER_RDMA_IP`** on the **RDMA/RoCE network** (from Pipeline A output or `ip` on the HCA netdev)—not necessarily the hostname’s management IP. + +1. Allocate two nodes (`salloc -N 2 -n 2 -t 01:00:00 …`). + +2. **Server (node A)** — blocks until the run finishes: + +```bash +python3 -m primus.tools.preflight.cluster_sphere verbs-server --device mlx5_0 +# Omit --device to pick the first device under /sys/class/infiniband +``` + +3. **Client (node B)** — after the server is listening: + +```bash +python3 -m primus.tools.preflight.cluster_sphere verbs-client \ + --server-ip "${SERVER_RDMA_IP}" --device mlx5_0 +``` + +**Single Slurm step (recommended)** — one `srun` launches server + client (`SERVER_RDMA_IP` = server node’s address on the RDMA network; task 0 is the first node in the allocation, task 1 the second): + +```bash +export SERVER_RDMA_IP= +srun -N 2 -n 2 -t 00:30:00 env PYTHONPATH="${PYTHONPATH}" SERVER_RDMA_IP="${SERVER_RDMA_IP}" \ + python3 -m primus.tools.preflight.cluster_sphere verbs-pair --device mlx5_0 +``` + +Optional: `--client-delay 15` (default) gives the server time to listen before the client connects. + +Example coordinating two separate `srun` steps (replace `node-a` / `node-b` with hosts from `scontrol show hostnames "$SLURM_JOB_NODELIST"`; add `--overlap` on the first line if your Slurm allows overlapping steps): + +```bash +srun -N 1 -n 1 -w node-a python3 -m primus.tools.preflight.cluster_sphere verbs-server --device mlx5_0 & +sleep 15 +srun -N 1 -n 1 -w node-b python3 -m primus.tools.preflight.cluster_sphere verbs-client \ + --server-ip "${SERVER_RDMA_IP}" --device mlx5_0 +``` + +Automated helper (requires **`export SERVER_RDMA_IP=…`**): +[`scripts/slurm/cluster_sphere_ib_write_bw_two_node.sh`](../../../scripts/slurm/cluster_sphere_ib_write_bw_two_node.sh) · index: [`scripts/slurm/README.md`](../../../scripts/slurm/README.md). + +More detail: **[`docs/preflight.md`](../../../docs/preflight.md)** (*Slurm: Cluster Sphere without preflight / torchrun*). + +Slurm (multi-node preflight example): ```bash NUM_NODES=8 srun -N ${NUM_NODES} --ntasks-per-node=1 --cpus-per-task=256 \ diff --git a/primus/tools/preflight/cluster_sphere/__main__.py b/primus/tools/preflight/cluster_sphere/__main__.py new file mode 100644 index 000000000..341490b8b --- /dev/null +++ b/primus/tools/preflight/cluster_sphere/__main__.py @@ -0,0 +1,185 @@ +############################################################################### +# Copyright (c) 2025, Advanced Micro Devices, Inc. All rights reserved. +# +# See LICENSE for license information. +############################################################################### + +""" +Bare-metal Cluster Sphere tools (no torchrun / torch.distributed). + +See docs: ``docs/preflight.md`` — Slurm execution pipelines. +""" + +from __future__ import annotations + +import argparse +import os +import shutil +import socket +import subprocess +import sys +import time + +from primus.tools.preflight.cluster_sphere import verbs_bw + + +def _cmd_env(args: argparse.Namespace) -> int: + from primus.tools.preflight.cluster_sphere.env_recommender import collect_cluster_sphere_env_findings + from primus.tools.preflight.cluster_sphere.report import emit_cluster_sphere_env_markdown + + hostname = os.environ.get("HOSTNAME") or socket.gethostname() + findings = collect_cluster_sphere_env_findings() + if args.markdown: + sys.stdout.write(emit_cluster_sphere_env_markdown(hostname, findings)) + return 0 + + for fin in findings: + print(f"[{fin.level}] {fin.message}", file=sys.stderr) + if fin.details: + det = fin.details + w = det.get("warnings") + if isinstance(w, list): + for line in w: + print(f" - {line}", file=sys.stderr) + return 0 + + +def _cmd_verbs_server(args: argparse.Namespace) -> int: + ib_dev = args.device or verbs_bw.first_ib_device_name() + if not ib_dev: + print("No RDMA device under /sys/class/infiniband; pass --device.", file=sys.stderr) + return 2 + port = args.port if args.port is not None else verbs_bw.default_port() + cmd = verbs_bw.ib_write_bw_server_cmd(ib_dev, port) + print(f"Running (server): {' '.join(cmd)}", file=sys.stderr) + if shutil.which("ib_write_bw") is None: + print("ib_write_bw not found; install perftest.", file=sys.stderr) + return 2 + os.execvp(cmd[0], cmd) + + +def _cmd_verbs_client(args: argparse.Namespace) -> int: + ib_dev = args.device or verbs_bw.first_ib_device_name() + if not ib_dev: + print("No RDMA device; pass --device.", file=sys.stderr) + return 2 + port = args.port if args.port is not None else verbs_bw.default_port() + cmd = verbs_bw.ib_write_bw_client_cmd(ib_dev, args.server_ip, port) + print(f"Running (client): {' '.join(cmd)}", file=sys.stderr) + if shutil.which("ib_write_bw") is None: + print("ib_write_bw not found; install perftest.", file=sys.stderr) + return 2 + + proc = subprocess.run(cmd, capture_output=True, text=True, timeout=args.timeout) + text = (proc.stdout or "") + "\n" + (proc.stderr or "") + print(text) + peak = verbs_bw.parse_peak_gbps(text) + if peak is not None: + print(f"\nParsed peak (best-effort): {peak:.2f} Gb/sec", file=sys.stderr) + return proc.returncode + + +def _cmd_verbs_pair(args: argparse.Namespace) -> int: + """ + Single Slurm step: task 0 runs ib_write_bw server, task 1 sleeps then runs client. + Requires: ``srun -N2 -n2`` (or equivalent) and ``SERVER_RDMA_IP`` for the server’s RDMA address. + """ + raw = os.environ.get("SLURM_PROCID") + if raw is None or raw == "": + print( + "verbs-pair must run under Slurm with exactly 2 tasks, e.g.\n" + " export SERVER_RDMA_IP=\n" + " srun -N2 -n2 -t 00:30:00 env PYTHONPATH=... SERVER_RDMA_IP=... \\\n" + " python3 -m primus.tools.preflight.cluster_sphere verbs-pair -d mlx5_0", + file=sys.stderr, + ) + return 2 + + try: + procid = int(raw) + except ValueError: + print(f"Invalid SLURM_PROCID={raw!r}", file=sys.stderr) + return 2 + + if procid == 0: + return _cmd_verbs_server(args) + + if procid == 1: + delay = float(getattr(args, "client_delay", 15.0)) + if delay > 0: + time.sleep(delay) + ip = os.environ.get("SERVER_RDMA_IP", "").strip() + if not ip: + print( + "Task 1: set SERVER_RDMA_IP to the server host’s RDMA-accessible address.", + file=sys.stderr, + ) + return 2 + args.server_ip = ip + return _cmd_verbs_client(args) + + print( + f"verbs-pair expects SLURM_PROCID 0 or 1 (two tasks); got {procid}.", + file=sys.stderr, + ) + return 2 + + +def main() -> int: + parser = argparse.ArgumentParser( + description="Cluster Sphere bare-metal helpers (no torchrun / no torch.distributed).", + ) + sub = parser.add_subparsers(dest="command", required=True) + + p_env = sub.add_parser("env", help="Print RDMA / NCCL export recommendations for this host.") + p_env.add_argument("--markdown", action="store_true", help="Emit Markdown section to stdout.") + p_env.set_defaults(func=_cmd_env) + + p_srv = sub.add_parser("verbs-server", help="Run ib_write_bw server (replaces current process).") + p_srv.add_argument("--device", "-d", help="RDMA device name (default: first in /sys/class/infiniband)") + p_srv.add_argument( + "--port", + "-p", + type=int, + help=f"TCP port (default: env {verbs_bw.DEFAULT_PORT_ENV} or 2000)", + ) + p_srv.set_defaults(func=_cmd_verbs_server) + + p_cli = sub.add_parser("verbs-client", help="Run ib_write_bw client toward verbs-server.") + p_cli.add_argument("--server-ip", required=True, help="Server IP/hostname on the RDMA-accessible network") + p_cli.add_argument("--device", "-d", help="RDMA device name (default: first IB device)") + p_cli.add_argument( + "--port", + "-p", + type=int, + help=f"TCP port (must match server; default env {verbs_bw.DEFAULT_PORT_ENV} or 2000)", + ) + p_cli.add_argument("--timeout", type=int, default=120, help="Run timeout (seconds)") + p_cli.set_defaults(func=_cmd_verbs_client) + + p_pair = sub.add_parser( + "verbs-pair", + help="Slurm: run server on task 0 and client on task 1 (use srun -N2 -n2; set SERVER_RDMA_IP).", + ) + p_pair.add_argument("--device", "-d", help="RDMA device name (default: first IB device)") + p_pair.add_argument( + "--port", + "-p", + type=int, + help=f"TCP port (default: env {verbs_bw.DEFAULT_PORT_ENV} or 2000)", + ) + p_pair.add_argument( + "--client-delay", + type=float, + default=15.0, + help="Seconds to wait on client task before connecting (default: 15)", + ) + p_pair.add_argument("--timeout", type=int, default=120, help="Client ib_write_bw timeout (seconds)") + p_pair.set_defaults(func=_cmd_verbs_pair) + + args = parser.parse_args() + return int(args.func(args)) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/primus/tools/preflight/cluster_sphere/rdma_bw.py b/primus/tools/preflight/cluster_sphere/rdma_bw.py index 287ef852c..f23ba1f0e 100644 --- a/primus/tools/preflight/cluster_sphere/rdma_bw.py +++ b/primus/tools/preflight/cluster_sphere/rdma_bw.py @@ -7,7 +7,6 @@ from __future__ import annotations import os -import re import shutil import subprocess import threading @@ -16,35 +15,16 @@ import torch.distributed as dist +from primus.tools.preflight.cluster_sphere.verbs_bw import ( + default_port, + first_ib_device_name, + ib_write_bw_client_cmd, + ib_write_bw_server_cmd, + parse_peak_gbps, +) from primus.tools.preflight.global_vars import RANK, WORLD_SIZE -def _first_ib_device_name() -> Optional[str]: - ib_path = "/sys/class/infiniband" - try: - names = sorted(os.listdir(ib_path)) - return names[0] if names else None - except OSError: - return None - - -def _resolve_server_host() -> str: - return os.environ.get("MASTER_ADDR", "localhost").strip() or "localhost" - - -def _parse_peak_gbps(text: str) -> Optional[float]: - """Best-effort peak Gb/sec from ib_write_bw stdout/stderr.""" - best: Optional[float] = None - for line in text.splitlines(): - for m in re.finditer(r"(\d+\.\d+)\s*Gb/sec", line, re.I): - v = float(m.group(1)) - best = v if best is None else max(best, v) - for m in re.finditer(r"(\d+\.\d+)\s*GB/sec", line): - v = float(m.group(1)) - best = v if best is None else max(best, v) - return best - - def append_cluster_sphere_verbs_rdma_section(args: Any, markdown_file: str) -> None: """ Run ib_write_bw between two ranks (WORLD_SIZE == 2) and append results to the perf markdown. @@ -75,31 +55,20 @@ def append_cluster_sphere_verbs_rdma_section(args: Any, markdown_file: str) -> N _append_lines(markdown_file, lines) return - ib_dev = _first_ib_device_name() + ib_dev = first_ib_device_name() if not ib_dev: lines.append("No InfiniBand / RDMA device under `/sys/class/infiniband`; skipping.\n\n") if RANK == 0: _append_lines(markdown_file, lines) return - port = int(os.environ.get("PRIMUS_IB_WRITE_BW_PORT", "2000")) - server_host = _resolve_server_host() + port = default_port() + server_host = os.environ.get("MASTER_ADDR", "localhost").strip() or "localhost" server_holder: Dict[str, Any] = {} def server_main() -> None: - cmd = [ - "ib_write_bw", - "-d", - ib_dev, - "-q", - "4", - "-a", - "--report_gbits", - "-F", - "-p", - str(port), - ] + cmd = ib_write_bw_server_cmd(ib_dev, port) try: proc = subprocess.run( cmd, @@ -124,23 +93,11 @@ def server_main() -> None: client_payload: Optional[Dict[str, Any]] = None if RANK == 1: time.sleep(5) - cmd = [ - "ib_write_bw", - "-d", - ib_dev, - "-q", - "4", - "-a", - "--report_gbits", - "-F", - server_host, - "-p", - str(port), - ] + cmd = ib_write_bw_client_cmd(ib_dev, server_host, port) try: proc = subprocess.run(cmd, capture_output=True, text=True, timeout=120) out = (proc.stdout or "") + "\n" + (proc.stderr or "") - peak = _parse_peak_gbps(out) + peak = parse_peak_gbps(out) client_payload = {"stdout": out, "rc": proc.returncode, "peak_gbps": peak} except subprocess.TimeoutExpired as e: client_payload = {"stdout": str(e), "rc": -1, "peak_gbps": None} @@ -170,7 +127,7 @@ def server_main() -> None: client_txt = str(client.get("stdout") or "") if peak is None: - peak = _parse_peak_gbps(client_txt) + peak = parse_peak_gbps(client_txt) lines.append(f"- **Device**: `{ib_dev}`\n") lines.append(f"- **Server**: `{server_host}:{port}` (rank 0)\n") diff --git a/primus/tools/preflight/cluster_sphere/report.py b/primus/tools/preflight/cluster_sphere/report.py index da6a6763b..67912d021 100644 --- a/primus/tools/preflight/cluster_sphere/report.py +++ b/primus/tools/preflight/cluster_sphere/report.py @@ -6,8 +6,11 @@ from __future__ import annotations +import io from typing import Any, Dict, List +from primus.tools.preflight.network.info import Finding + def wants_cluster_sphere_env(args: Any) -> bool: return bool(getattr(args, "cluster_sphere", False) or getattr(args, "cluster_sphere_env", False)) @@ -17,6 +20,17 @@ def wants_cluster_sphere_rdma_bw(args: Any) -> bool: return bool(getattr(args, "cluster_sphere", False) or getattr(args, "cluster_sphere_rdma_bw", False)) +def emit_cluster_sphere_env_markdown(hostname: str, findings: List[Finding]) -> str: + """Serialize Cluster Sphere env findings to Markdown (standalone CLI / Slurm).""" + fd: List[Dict[str, Any]] = [] + for fin in findings: + fd.append({"message": fin.message, "level": fin.level, "details": fin.details}) + gathered = [{"host": hostname, "findings": fd}] + buf = io.StringIO() + write_cluster_sphere_env_section(buf, gathered) + return buf.getvalue() + + def write_cluster_sphere_env_section(f, gathered: List[Dict[str, Any]]) -> None: """Append Cluster Sphere env recommender markdown (rank0).""" f.write("## Cluster Sphere — RDMA environment recommendations\n\n") @@ -69,6 +83,17 @@ def _write_one_host_block(f, host: str, details: Dict[str, Any]) -> None: ) f.write("\n") + fw_by = details.get("firmware_by_version") or {} + if fw_by: + f.write("**Firmware report:**\n\n") + f.write("| Firmware | RDMA devices |\n") + f.write("|----------|---------------|\n") + for fw in sorted(fw_by.keys()): + devs = fw_by[fw] + devs_s = ", ".join(devs) if isinstance(devs, list) else str(devs) + f.write(f"| {fw} | {devs_s} |\n") + f.write("\n") + nccl = details.get("nccl_exports") or [] if nccl: f.write("**Suggested NCCL / socket exports:**\n\n```bash\n") diff --git a/primus/tools/preflight/cluster_sphere/verbs_bw.py b/primus/tools/preflight/cluster_sphere/verbs_bw.py new file mode 100644 index 000000000..de774e5ab --- /dev/null +++ b/primus/tools/preflight/cluster_sphere/verbs_bw.py @@ -0,0 +1,79 @@ +############################################################################### +# Copyright (c) 2025, Advanced Micro Devices, Inc. All rights reserved. +# +# See LICENSE for license information. +############################################################################### + +"""Verbs ``ib_write_bw`` helpers (subprocess only; no PyTorch).""" + +from __future__ import annotations + +import os +import re +import subprocess +from typing import List, Optional + +DEFAULT_PORT_ENV = "PRIMUS_IB_WRITE_BW_PORT" + + +def default_port() -> int: + return int(os.environ.get(DEFAULT_PORT_ENV, "2000")) + + +def first_ib_device_name() -> Optional[str]: + ib_path = "/sys/class/infiniband" + try: + names = sorted(os.listdir(ib_path)) + return names[0] if names else None + except OSError: + return None + + +def parse_peak_gbps(text: str) -> Optional[float]: + best: Optional[float] = None + for line in text.splitlines(): + for m in re.finditer(r"(\d+\.\d+)\s*Gb/sec", line, re.I): + v = float(m.group(1)) + best = v if best is None else max(best, v) + for m in re.finditer(r"(\d+\.\d+)\s*GB/sec", line): + v = float(m.group(1)) + best = v if best is None else max(best, v) + return best + + +def ib_write_bw_server_cmd(ib_dev: str, port: int) -> List[str]: + return ["ib_write_bw", "-d", ib_dev, "-q", "4", "-a", "--report_gbits", "-F", "-p", str(port)] + + +def ib_write_bw_client_cmd(ib_dev: str, server_host: str, port: int) -> List[str]: + return [ + "ib_write_bw", + "-d", + ib_dev, + "-q", + "4", + "-a", + "--report_gbits", + "-F", + server_host, + "-p", + str(port), + ] + + +def run_ib_write_bw_server(ib_dev: str, port: int, *, timeout_sec: int = 180) -> subprocess.CompletedProcess: + return subprocess.run( + ib_write_bw_server_cmd(ib_dev, port), + capture_output=True, + text=True, + timeout=timeout_sec, + ) + + +def run_ib_write_bw_client(ib_dev: str, server_host: str, port: int, *, timeout_sec: int = 120) -> subprocess.CompletedProcess: + return subprocess.run( + ib_write_bw_client_cmd(ib_dev, server_host, port), + capture_output=True, + text=True, + timeout=timeout_sec, + ) diff --git a/scripts/slurm/README.md b/scripts/slurm/README.md new file mode 100644 index 000000000..2ba454238 --- /dev/null +++ b/scripts/slurm/README.md @@ -0,0 +1,26 @@ +# Slurm helper scripts + +## Cluster Sphere (no torchrun) + +| Script | Nodes | Purpose | +|--------|-------|---------| +| [`cluster_sphere_env_single_node.sh`](cluster_sphere_env_single_node.sh) | 1 | Writes RDMA / NCCL export recommendations to Markdown (`CLUSTER_SPHERE_OUT` or cwd). | +| [`cluster_sphere_ib_write_bw_two_node.sh`](cluster_sphere_ib_write_bw_two_node.sh) | 2 | One `srun -N2 -n2`: `verbs-pair` (server on task 0, client on task 1). Requires `SERVER_RDMA_IP` (first node’s RDMA IP). | + +See [Preflight docs](../../docs/preflight.md) for prerequisites (`ibv_devinfo`, `ib_write_bw`, `PYTHONPATH`) and manual two-terminal usage. + +Examples: + +```bash +export PRIMUS_ROOT=/path/to/Primus + +# Pipeline A — one compute node +srun -N 1 -n 1 -t 00:30:00 -- bash Primus/scripts/slurm/cluster_sphere_env_single_node.sh + +# Pipeline B — two nodes (after setting server IP from Pipeline A / `ip`) +export SERVER_RDMA_IP=10.224.0.73 +salloc -N 2 -n 2 -t 01:00:00 +./Primus/scripts/slurm/cluster_sphere_ib_write_bw_two_node.sh +``` + +Site-specific Slurm flags (`-p`, `-A`, `-t`) must be added by each cluster. diff --git a/scripts/slurm/cluster_sphere_env_single_node.sh b/scripts/slurm/cluster_sphere_env_single_node.sh new file mode 100755 index 000000000..feb5c0214 --- /dev/null +++ b/scripts/slurm/cluster_sphere_env_single_node.sh @@ -0,0 +1,14 @@ +#!/usr/bin/env bash +############################################################################### +# Cluster Sphere — RDMA env recommender on one Slurm node (no torchrun). +# Run via: srun/salloc with -N1, or sbatch. Set PRIMUS_ROOT to Primus repo root if unset. +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +PRIMUS_ROOT="${PRIMUS_ROOT:-$(cd "${SCRIPT_DIR}/../.." && pwd)}" +export PYTHONPATH="${PRIMUS_ROOT}${PYTHONPATH:+:${PYTHONPATH:-}}" + +OUT="${CLUSTER_SPHERE_OUT:-${SLURM_SUBMIT_DIR:-.}/cluster_sphere_env_${SLURM_JOB_ID:-local}.md}" + +python3 -m primus.tools.preflight.cluster_sphere env --markdown >"${OUT}" +echo "Wrote ${OUT}" >&2 diff --git a/scripts/slurm/cluster_sphere_ib_write_bw_two_node.sh b/scripts/slurm/cluster_sphere_ib_write_bw_two_node.sh new file mode 100755 index 000000000..9b205e175 --- /dev/null +++ b/scripts/slurm/cluster_sphere_ib_write_bw_two_node.sh @@ -0,0 +1,50 @@ +#!/usr/bin/env bash +############################################################################### +# Cluster Sphere — ib_write_bw across two Slurm nodes (no torchrun). +# +# Single srun step: task 0 runs ib_write_bw server, task 1 waits then client. +# Set SERVER_RDMA_IP to the **first allocated node’s** address on the RDMA/RoCE +# network (same node that runs the server — from Pipeline A or `ip` on the HCA). +# Requires a 2-node allocation: salloc -N 2 -n 2 -t ... (adjust -p/-A/-t per site.) +# +# export PRIMUS_ROOT=/path/to/Primus +# export SERVER_RDMA_IP=10.x.x.x +# ./cluster_sphere_ib_write_bw_two_node.sh +# +# Optional: IB_DEVICE (e.g. mlx5_0), PORT / PRIMUS_IB_WRITE_BW_PORT, CLIENT_DELAY +# (seconds before client connects; default 15 in verbs-pair). +############################################################################### +set -euo pipefail + +if [[ -z "${SLURM_JOB_NODELIST:-}" ]]; then + echo "error: SLURM_JOB_NODELIST not set — run inside a Slurm allocation with >= 2 nodes." >&2 + exit 1 +fi +if [[ -z "${SERVER_RDMA_IP:-}" ]]; then + echo "error: export SERVER_RDMA_IP=" >&2 + exit 1 +fi + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +PRIMUS_ROOT="${PRIMUS_ROOT:-$(cd "${SCRIPT_DIR}/../.." && pwd)}" +export PYTHONPATH="${PRIMUS_ROOT}${PYTHONPATH:+:${PYTHONPATH:-}}" + +mapfile -t HOSTS < <(scontrol show hostnames "${SLURM_JOB_NODELIST}") +if [[ -z "${HOSTS[1]:-}" ]]; then + echo "error: need at least 2 nodes in allocation." >&2 + exit 1 +fi + +PORT="${PORT:-${PRIMUS_IB_WRITE_BW_PORT:-2000}}" +PAIR_OPTS=(--port "${PORT}") +if [[ -n "${IB_DEVICE:-}" ]]; then + PAIR_OPTS+=(--device "${IB_DEVICE}") +fi +if [[ -n "${CLIENT_DELAY:-}" ]]; then + PAIR_OPTS+=(--client-delay "${CLIENT_DELAY}") +fi + +echo "verbs-pair: server on ${HOSTS[0]}, client on ${HOSTS[1]}; SERVER_RDMA_IP=${SERVER_RDMA_IP}:${PORT}" >&2 + +exec srun -N 2 -n 2 env PYTHONPATH="${PYTHONPATH}" SERVER_RDMA_IP="${SERVER_RDMA_IP}" \ + python3 -m primus.tools.preflight.cluster_sphere verbs-pair "${PAIR_OPTS[@]}" diff --git a/tests/test_preflight_cluster_sphere.py b/tests/test_preflight_cluster_sphere.py index c5b5e86e4..ffa8d9bbe 100644 --- a/tests/test_preflight_cluster_sphere.py +++ b/tests/test_preflight_cluster_sphere.py @@ -7,7 +7,9 @@ from __future__ import annotations from pathlib import Path -from unittest.mock import patch +from unittest.mock import MagicMock, patch + +from primus.tools.preflight.network.info import Finding def test_resolve_default_is_primus_package(): @@ -56,6 +58,64 @@ def test_collect_findings_warns_when_empty_sysfs(): assert findings[0].level == "warn" +def test_emit_markdown_firmware_report_section(): + from primus.tools.preflight.cluster_sphere.report import emit_cluster_sphere_env_markdown + + findings = [ + Finding( + level="info", + message="Cluster Sphere RDMA environment recommendations", + details={ + "warnings": [], + "devices": [ + { + "rdma": "mlx5_0", + "pci": "—", + "netdev": "eth0", + "firmware": "fwA", + "gid_index": "3", + "gid_value": "::", + "vendor": "MLNX", + }, + ], + "firmware_by_version": {"fwA": ["mlx5_0"], "fwB": ["mlx5_1"]}, + "nccl_exports": [], + }, + ), + ] + md = emit_cluster_sphere_env_markdown("test-host", findings) + assert "**Firmware report:**" in md + assert "| fwA | mlx5_0 |" in md + assert "| fwB | mlx5_1 |" in md + + +def test_verbs_pair_routes_by_slurm_procid(monkeypatch): + from primus.tools.preflight.cluster_sphere import __main__ as cs_main + + args = MagicMock() + args.device = None + args.port = None + args.timeout = 120 + args.client_delay = 0.0 + + monkeypatch.delenv("SLURM_PROCID", raising=False) + assert cs_main._cmd_verbs_pair(args) == 2 + + monkeypatch.setenv("SLURM_PROCID", "2") + assert cs_main._cmd_verbs_pair(args) == 2 + + monkeypatch.setenv("SLURM_PROCID", "1") + monkeypatch.delenv("SERVER_RDMA_IP", raising=False) + assert cs_main._cmd_verbs_pair(args) == 2 + + monkeypatch.setenv("SERVER_RDMA_IP", "10.0.0.1") + mock_cli = MagicMock(return_value=0) + monkeypatch.setattr(cs_main, "_cmd_verbs_client", mock_cli) + assert cs_main._cmd_verbs_pair(args) == 0 + mock_cli.assert_called_once() + assert args.server_ip == "10.0.0.1" + + def test_nccl_exports_with_stub_devices(monkeypatch): from primus.tools.preflight.cluster_sphere.env_recommender import EnvRecommenderEngine from primus.tools.preflight.cluster_sphere.env_recommender import _DeviceInfo