From 079e87f790f4b3179a92e9c3a924db9e38ab02db Mon Sep 17 00:00:00 2001 From: Elevatormusic <22101396+Elevatormusic@users.noreply.github.com> Date: Tue, 20 Oct 2026 17:15:22 -0700 Subject: [PATCH 1/4] Detect the RDMA GID per port, turn NCCL cuMem off, release the page cache after a weight load The GID index of each RDMA port is read from sysfs at start (lowest RoCE v2 IPv4 entry). TF_RDMA_GID_INDEX or NCCL_IB_GID_INDEX with a value of 0 or more still fixes it; unset, empty or -1 means detect. The gather takes one index per device. The engine sets NCCL_IB_GID_INDEX to the detected index when it is unset, and NCCL_CUMEM_ENABLE=0 and NCCL_CUMEM_HOST_ENABLE=0 unless the user set them. Shards.close and RankCache.close call posix_fadvise(DONTNEED) on the files they read; TF_DS_RELEASE_PAGE_CACHE=0 turns it off. --- NOTICE | 7 + THIRD_PARTY_NOTICES.md | 6 + src/tensorfold/cuda/rdma.py | 12 +- src/tensorfold/cuda/rdma_gather.cu | 18 +- src/tensorfold/cuda/roce_gid.py | 103 ++++++ .../families/deepseek_v41/cuda/engine.py | 14 + .../families/deepseek_v41/cuda/hostenv.py | 97 ++++++ .../families/deepseek_v41/cuda/weights.py | 11 + tests/test_dsv41_host_env.py | 307 ++++++++++++++++++ tools/dsv41/ATTRIBUTION.md | 1 + tools/dsv41/DESIGN.md | 10 +- 11 files changed, 576 insertions(+), 10 deletions(-) create mode 100644 src/tensorfold/cuda/roce_gid.py create mode 100644 src/tensorfold/families/deepseek_v41/cuda/hostenv.py create mode 100644 tests/test_dsv41_host_env.py diff --git a/NOTICE b/NOTICE index 6b7c65212c..bc8041e9ae 100644 --- a/NOTICE +++ b/NOTICE @@ -17,6 +17,7 @@ nodes. TensorFold's text above, LICENSE, LICENSES/ and the upstream sections of Added: src/tensorfold/families/deepseek_v41/ the model family (engine, kernels, server integration) src/tensorfold/cuda/rdma.py, src/tensorfold/cuda/rdma_gather.cu RDMA-write all-gather for TP decode + src/tensorfold/cuda/roce_gid.py RoCE v2 IPv4 GID index scan of an RDMA port (sysfs) tools/dsv41/ tests, checks, benchmark clients, the reference model, DESIGN.md, REPORT.md and ATTRIBUTION.md A "DeepSeek-V4.1-Flash on CUDA (TP2)" section in THIRD_PARTY_NOTICES.md, and a fork note at the top of README.md. @@ -49,4 +50,10 @@ Modified upstream files: src/tensorfold/server/authentication.py the no-key startup line names the CUDA server's TF_API_KEY gate .gitignore ignores macOS ._* files +Later work in the official TensorFold (https://github.com/ashhart/TensorFold, main after v0.6.6, Apache-2.0, Copyright +2026 TensorFold contributors) gave three ideas that this fork wrote again in Python, without copying its Zig code: +NCCL's cuMem allocations off by default (its pull request #995), the GID index scan of an RDMA port (`gidIndex` in +`zig/src/cuda/roce.zig`) and the release of a loaded file's page cache (its pull requests #766 and #916). They are in +src/tensorfold/families/deepseek_v41/cuda/hostenv.py, src/tensorfold/cuda/roce_gid.py and the callers of those files. + The "git log" of this branch and tools/dsv41/ATTRIBUTION.md describe the changes and their sources. diff --git a/THIRD_PARTY_NOTICES.md b/THIRD_PARTY_NOTICES.md index 72fc029523..e319e11ca5 100644 --- a/THIRD_PARTY_NOTICES.md +++ b/THIRD_PARTY_NOTICES.md @@ -94,6 +94,12 @@ This section was added by the fork that adds the `deepseek_v41` family; see `NOT - The grouped EXL3 expert kernel additions (`src/tensorfold/cuda/exl3/experts_grouped.cuh`, `experts.cu`) follow [ExLlamaV3](https://github.com/turboderp-org/exllamav3)'s EXL3 format, MIT License, Copyright (c) 2025 Turboderp (text below), as the rest of the shared EXL3 module does. +- Three host settings follow ideas of the official [TensorFold](https://github.com/ashhart/TensorFold) main branch + (Apache-2.0, Copyright 2026 TensorFold contributors), written again in Python without taking its Zig code: NCCL's + cuMem allocations off by default (`hostenv.py`; its pull request #995), the RoCE v2 IPv4 GID scan of a port + (`src/tensorfold/cuda/roce_gid.py`; the idea of `gidIndex` in `zig/src/cuda/roce.zig`) and the page cache release + after a weight load (`hostenv.py`, `weights.py`; its pull requests #766 and #916). None of its DeepSeek-V4.1 files + (`families/deepseek_v41/`, `kernels/cuda/deepseek_v41/`) was opened. - The baseline used in `tools/dsv41/REPORT.md` (a vLLM kit under AGPL-3.0) was only run as a black box over HTTP. None of its code was read or included. `tools/dsv41/ATTRIBUTION.md` lists every source in full. diff --git a/src/tensorfold/cuda/rdma.py b/src/tensorfold/cuda/rdma.py index 1bb395475a..6c346f57de 100644 --- a/src/tensorfold/cuda/rdma.py +++ b/src/tensorfold/cuda/rdma.py @@ -12,6 +12,10 @@ TF_RDMA_STAGE_FENCE / TF_RDMA_OWN_EARLY / TF_RDMA_COLLECT_UNROLL = 0 each restore c100's way. TF_RDMA_SLOTS (default 4) sets the ring slots (2 halves the pinned rings; any count >= 2 is safe: a rank stages gather q + 2 only after its own collect of q + 1, which needs the peer's data of q + 1, sent after the peer's collect of q). + +The GID index of each port is the lowest RoCE v2 IPv4 entry of its GID table, read at start (roce_gid.py): the tables of +two nodes can differ, and a link flap can move an entry. TF_RDMA_GID_INDEX or NCCL_IB_GID_INDEX with a value of 0 or more +fixes the index for every port; unset, empty or -1 means "detect". """ from __future__ import annotations @@ -24,6 +28,8 @@ import torch +from .roce_gid import resolve_gid_indices + @lru_cache(maxsize=1) def _ext(): @@ -65,7 +71,6 @@ def __init__(self, store, rank: int, world: int, *, max_bytes: int, slots: int | prefix: str = "tf_rdma", devices: list[str] | None = None, gid_index: int | None = None) -> None: ext = _ext() slots = int(os.environ.get("TF_RDMA_SLOTS") or 4) if slots is None else int(slots) - gid = int(os.environ.get("NCCL_IB_GID_INDEX", "5")) if gid_index is None else int(gid_index) cpu = int(os.environ.get("TF_RDMA_CPU", "-1")) self.ext, self.rank, self.world, self.max_bytes = ext, rank, world, int(max_bytes) self.devices = device_names() if devices is None else list(devices) @@ -73,7 +78,10 @@ def __init__(self, store, rank: int, world: int, *, max_bytes: int, slots: int | try: if slots < 2: raise ValueError(f"TF_RDMA_SLOTS must be at least 2 (got {slots})") - self.h = ext.create(rank, world, self.max_bytes, slots, self.devices, gid, cpu) + gids = resolve_gid_indices(self.devices, fixed=gid_index) + for name, (index, how) in zip(self.devices, gids): + print(f"[tensorfold] rank {rank}: RDMA GID index of {name} port 1: {index} ({how})", flush=True) + self.h = ext.create(rank, world, self.max_bytes, slots, self.devices, [g[0] for g in gids], cpu) info = pickle.dumps((ext.local_info(self.h), slots, self.max_bytes)) except Exception as exc: # noqa: BLE001 (published, so every rank refuses together) error, info = f"{type(exc).__name__}: {exc}", b"" diff --git a/src/tensorfold/cuda/rdma_gather.cu b/src/tensorfold/cuda/rdma_gather.cu index 5cac968ad3..ec598658af 100644 --- a/src/tensorfold/cuda/rdma_gather.cu +++ b/src/tensorfold/cuda/rdma_gather.cu @@ -86,6 +86,7 @@ struct Nic { ibv_mr* mr = nullptr; // the send ring's (the whole block's when the rings are cudaHostAlloc memory) ibv_mr* mr_recv = nullptr; // the receive ring's (== mr in that case) ibv_gid gid{}; + int gid_index = 0; // this port's GID table entry (RoCE v2, IPv4), chosen by rdma.py ibv_mtu mtu = IBV_MTU_4096; // the port's active MTU (RoCE: the netdev MTU less headers, rounded down) std::vector qps; // per peer std::vector psns; @@ -98,7 +99,7 @@ struct Peer { }; struct Gather { - int rank = 0, world = 1, slots = 4, port = 1, gid_index = 5, cpu = -1; + int rank = 0, world = 1, slots = 4, port = 1, cpu = -1; size_t max_bytes = 0; char* host = nullptr; // the metadata's block (host_reg 0: rings + metadata in one cudaHostAlloc) size_t host_bytes = 0; @@ -375,7 +376,7 @@ void proxy_loop(Gather* gp) { } } -Nic open_nic(const std::string& name, Gather& g, int world, int rank) { +Nic open_nic(const std::string& name, Gather& g, int world, int rank, int gid_index) { int count = 0; ibv_device** list = ibv_get_device_list(&count); TORCH_CHECK(list && count > 0, "rdma gather: no RDMA devices"); @@ -402,7 +403,9 @@ Nic open_nic(const std::string& name, Gather& g, int world, int rank) { TORCH_CHECK(nic.mr_recv, "rdma gather: ibv_reg_mr of the ", g.recv_bytes, "-byte receive ring on ", name, " failed"); } - TORCH_CHECK(ibv_query_gid(nic.ctx, g.port, g.gid_index, &nic.gid) == 0, "rdma gather: ibv_query_gid failed"); + nic.gid_index = gid_index; + TORCH_CHECK(gid_index >= 0 && ibv_query_gid(nic.ctx, g.port, gid_index, &nic.gid) == 0, + "rdma gather: ibv_query_gid failed for GID index ", gid_index, " of ", name, " port ", g.port); // The QPs take the port's active MTU, not a fixed 4096: at netdev MTU 1500 RoCE runs at 1024, and a QP set to 4096 // sends frames the link drops (every write then fails with "transport retry counter exceeded"). ibv_port_attr port{}; @@ -444,16 +447,16 @@ Nic open_nic(const std::string& name, Gather& g, int world, int rank) { } int64_t create(int64_t rank, int64_t world, int64_t max_bytes, int64_t slots, const std::vector& devices, - int64_t gid_index, int64_t cpu) { + const std::vector& gid_indices, int64_t cpu) { TORCH_CHECK(world >= 2 && rank >= 0 && rank < world, "rdma gather: rank ", rank, " of ", world); TORCH_CHECK(max_bytes > 0 && max_bytes % 64 == 0, "rdma gather: max_bytes must be a positive multiple of 64"); TORCH_CHECK(!devices.empty(), "rdma gather: name at least one RDMA device"); + TORCH_CHECK(gid_indices.size() == devices.size(), "rdma gather: one GID index a device"); auto g = std::make_unique(); g->rank = (int)rank; g->world = (int)world; g->slots = (int)slots; g->max_bytes = (size_t)max_bytes; - g->gid_index = (int)gid_index; g->cpu = (int)cpu; const size_t send_b = (size_t)slots * max_bytes, recv_b = (size_t)slots * world * max_bytes; const size_t meta_b = ((size_t)slots * world + 1 + 2 * (size_t)slots + 1) * 8; @@ -506,7 +509,8 @@ int64_t create(int64_t rank, int64_t world, int64_t max_bytes, int64_t slots, co C10_CUDA_CHECK(cudaMemset(g->staged, 0, 8)); C10_CUDA_CHECK(cudaMalloc(reinterpret_cast(&g->probe), 8 * 8)); C10_CUDA_CHECK(cudaMemset(g->probe, 0, 8 * 8)); - for (const auto& name : devices) g->nics.push_back(open_nic(name, *g, (int)world, (int)rank)); + for (size_t i = 0; i < devices.size(); ++i) + g->nics.push_back(open_nic(devices[i], *g, (int)world, (int)rank, (int)gid_indices[i])); g->peers.assign(world, Peer{}); gathers.push_back(std::move(g)); return (int64_t)gathers.size() - 1; @@ -561,7 +565,7 @@ void link_peers(int64_t h, const std::vector& remote) { rtr.min_rnr_timer = 12; rtr.ah_attr.is_global = 1; std::memcpy(rtr.ah_attr.grh.dgid.raw, peer.gid[d].data(), 16); - rtr.ah_attr.grh.sgid_index = g.gid_index; + rtr.ah_attr.grh.sgid_index = g.nics[d].gid_index; rtr.ah_attr.grh.hop_limit = 64; rtr.ah_attr.port_num = g.port; TORCH_CHECK(ibv_modify_qp(g.nics[d].qps[p], &rtr, diff --git a/src/tensorfold/cuda/roce_gid.py b/src/tensorfold/cuda/roce_gid.py new file mode 100644 index 0000000000..4878b944d1 --- /dev/null +++ b/src/tensorfold/cuda/roce_gid.py @@ -0,0 +1,103 @@ +"""Find the RoCE v2 IPv4 GID index of an RDMA port (no torch needed). + +The GID table of a CX7 port can differ from node to node, and it can change after a link flap. A fixed index +then points at the wrong entry, or at an empty one. The kernel shows the table in sysfs: +``//ports//gids/`` holds the GID and ``.../gid_attrs/types/`` holds its type. +An entry is the one we want when its type is "RoCE v2" and its GID is an IPv4-mapped address. + +The scan idea follows ``gidIndex`` in ``zig/src/cuda/roce.zig`` of the official TensorFold +(https://github.com/ashhart/TensorFold, Apache-2.0). The code here is written for this fork. +""" + +from __future__ import annotations + +import os +from collections.abc import Mapping, Sequence +from pathlib import Path + +SYSFS = Path("/sys/class/infiniband") +# The first 96 bits of an IPv4-mapped IPv6 address (::ffff:a.b.c.d), as the kernel prints a GID. +IPV4_MAPPED = "0000:0000:0000:0000:0000:ffff:" +# The variables that can fix the index. The first one that has a value decides. A value of -1 means "detect". +OVERRIDE_VARS = ("TF_RDMA_GID_INDEX", "NCCL_IB_GID_INDEX") + + +def _read(path: Path) -> str | None: + try: + return path.read_text().strip() + except OSError: # an empty table entry has no type file that can be read + return None + + +def _entries(directory: Path) -> list[int]: + try: + return sorted(int(p.name) for p in directory.iterdir() if p.name.isdigit()) + except OSError: + return [] + + +def gid_entries(device: str, port: int = 1, *, root: Path | str = SYSFS) -> list[tuple[int, str, str]]: + """Every readable entry of the GID table as ``(index, type, gid)``, in index order.""" + + base = Path(root) / device / "ports" / str(port) + out = [] + for n in _entries(base / "gids"): + kind = _read(base / "gid_attrs" / "types" / str(n)) + gid = _read(base / "gids" / str(n)) + if kind is not None and gid is not None: + out.append((n, kind, gid)) + return out + + +def detect_gid_index(device: str, port: int = 1, *, root: Path | str = SYSFS) -> int | None: + """The lowest index of a RoCE v2 entry with an IPv4 address, or None when the table has none.""" + + for n, kind, gid in gid_entries(device, port, root=root): + if not kind.startswith("RoCE v2"): + continue + gid = gid.lower() + if gid.startswith(IPV4_MAPPED) and gid[len(IPV4_MAPPED):] != "0000:0000": + return n + return None + + +def gid_override(env: Mapping[str, str] | None = None) -> int | None: + """The index that the user fixed, or None when the user did not fix one (variable unset, empty or -1).""" + + env = os.environ if env is None else env + for name in OVERRIDE_VARS: + raw = (env.get(name) or "").strip() + if not raw: + continue + try: + value = int(raw) + except ValueError: + raise ValueError(f"{name} must be a whole number (got {raw!r})") from None + return value if value >= 0 else None + return None + + +def resolve_gid_indices(devices: Sequence[str], port: int = 1, *, fixed: int | None = None, + env: Mapping[str, str] | None = None, + root: Path | str = SYSFS) -> list[tuple[int, str]]: + """One ``(index, how)`` for each device. ``how`` is "fixed" (an argument or a variable) or "detected". + + ``fixed`` (an argument) wins over the variables. A negative ``fixed`` means "detect". + Raises RuntimeError when a device has no RoCE v2 IPv4 entry and nothing fixes the index. + """ + + if fixed is not None and fixed < 0: + fixed = None + if fixed is None: + fixed = gid_override(env) + out = [] + for device in devices: + if fixed is not None: + out.append((fixed, "fixed")) + continue + found = detect_gid_index(device, port, root=root) + if found is None: + raise RuntimeError(f"{device} port {port} has no RoCE v2 IPv4 GID in {Path(root) / device}; " + f"set {OVERRIDE_VARS[0]} to the index to use") + out.append((found, "detected")) + return out diff --git a/src/tensorfold/families/deepseek_v41/cuda/engine.py b/src/tensorfold/families/deepseek_v41/cuda/engine.py index 396c6e4bc4..dca437d7c3 100644 --- a/src/tensorfold/families/deepseek_v41/cuda/engine.py +++ b/src/tensorfold/families/deepseek_v41/cuda/engine.py @@ -82,6 +82,8 @@ def __init__(self, model_dir: Path, *, rank: int, world: int, master: str, port: # gather on the CX7 link against 7.1 with NCCL's choice (decode windows go over the RDMA gather) os.environ.setdefault("NCCL_PROTO", "Simple") os.environ.setdefault("NCCL_MIN_NCHANNELS", "4") + if world > 1: + self._nccl_host_env() nccl = NCCL(rank, world, master, port) if world > 1 else None self.nccl = nccl self.w = load(model_dir, rank, world, dspark=drafts > 0) @@ -189,6 +191,18 @@ def __init__(self, model_dir: Path, *, rank: int, world: int, master: str, port: print(f"[tensorfold] DeepSeek-V4.1 engine ready: {world} rank(s), context {self.limit}, " f"{'DSpark ' + str(drafts) + ' drafts' if self.drafter else 'serial decode'}", flush=True) + def _nccl_host_env(self) -> None: + """NCCL's cuMem allocations off and its GID index set (hostenv.py); a value that the user set stays.""" + + from tensorfold.cuda.rdma import device_names + + from .hostenv import apply_nccl_defaults + + done = apply_nccl_defaults(device_names()) + if done: + print(f"[tensorfold] rank {self.rank}: NCCL settings by default: " + + " ".join(f"{k}={v}" for k, v in done.items()), flush=True) + def _memory_ceiling(self) -> None: """After the warm-up, torch may grow only while the host keeps TF_DS_MEM_FLOOR_GIB (default 2; 0: no ceiling) available: a step that would need more fails with CUDA's out-of-memory error (that request fails, the lane's diff --git a/src/tensorfold/families/deepseek_v41/cuda/hostenv.py b/src/tensorfold/families/deepseek_v41/cuda/hostenv.py new file mode 100644 index 0000000000..673d3c514a --- /dev/null +++ b/src/tensorfold/families/deepseek_v41/cuda/hostenv.py @@ -0,0 +1,97 @@ +"""Host settings of the DeepSeek-V4.1 TP path that need no torch: the NCCL defaults and the page cache release. + +NCCL cuMem off: on GB10 the driver can run out of memory when +NCCL allocates with the cuMem API under many streams, and later two-rank starts then fail until a reboot. The official +TensorFold turns these two NCCL settings off for the same reason (its pull request #995, Apache-2.0). The code here is +written for this fork. + +Page cache release: the weight loader reads about 106 GB a rank through the page cache. On GB10 the CPU and the GPU +share one memory, so those pages take memory that the serving process needs. The official TensorFold drops the pages of +a loaded file with ``fadvise(DONTNEED)`` (its pull requests #766 and #916). ``drop_page_cache`` does the same. +""" + +from __future__ import annotations + +import os +from collections.abc import Mapping, Sequence +from pathlib import Path + +from tensorfold.cuda.roce_gid import SYSFS, detect_gid_index + +NCCL_DEFAULTS = {"NCCL_CUMEM_ENABLE": "0", "NCCL_CUMEM_HOST_ENABLE": "0"} +RELEASE_VAR = "TF_DS_RELEASE_PAGE_CACHE" + + +def _is_set(env: Mapping[str, str], name: str) -> bool: + return bool((env.get(name) or "").strip()) + + +def apply_nccl_defaults(devices: Sequence[str] = (), *, env: dict | None = None, root: Path | str = SYSFS, + port: int = 1) -> dict[str, str]: + """Set what the user did not set, before NCCL starts. Returns the variables that this call set. + + * ``NCCL_CUMEM_ENABLE`` and ``NCCL_CUMEM_HOST_ENABLE`` become 0. + * ``NCCL_IB_GID_INDEX`` becomes the RoCE v2 IPv4 GID index of ``devices`` when every device has the same one. + A value that is set (also -1, which is NCCL's own detection) stays as it is. + When this call sets the index, it also sets ``TF_RDMA_GID_INDEX=-1`` (if that is unset) so that the RDMA + gather keeps detecting each port itself. + """ + + env = os.environ if env is None else env + done = {} + for name, value in NCCL_DEFAULTS.items(): + if not _is_set(env, name): + env[name] = done[name] = value + if devices and not _is_set(env, "NCCL_IB_GID_INDEX"): + found = {detect_gid_index(d, port, root=root) for d in devices} + if len(found) == 1 and None not in found: + env["NCCL_IB_GID_INDEX"] = done["NCCL_IB_GID_INDEX"] = str(found.pop()) + # the RDMA gather reads NCCL_IB_GID_INDEX as the user's choice; this marker keeps its own detection on + if not _is_set(env, "TF_RDMA_GID_INDEX"): + env["TF_RDMA_GID_INDEX"] = "-1" + return done + + +def release_enabled(env: Mapping[str, str] | None = None) -> bool: + """True unless TF_DS_RELEASE_PAGE_CACHE is 0 (or off, false, no).""" + + env = os.environ if env is None else env + return (env.get(RELEASE_VAR) or "1").strip().lower() not in ("0", "off", "false", "no") + + +def drop_page_cache(target, *, env: Mapping[str, str] | None = None) -> bool: + """Ask the kernel to drop the cached pages of a file: ``target`` is an open file descriptor or a path. + + The call is a hint: the kernel keeps dirty pages and pages that another process maps. It returns True when the + kernel took the request, and False when the switch is off, the system has no ``posix_fadvise`` or the call failed. + """ + + advise = getattr(os, "posix_fadvise", None) + flag = getattr(os, "POSIX_FADV_DONTNEED", None) + if advise is None or flag is None or not release_enabled(env): + return False + try: + if isinstance(target, int): + advise(target, 0, 0, flag) + else: + fd = os.open(target, os.O_RDONLY) + try: + advise(fd, 0, 0, flag) + finally: + os.close(fd) + except OSError: + return False + return True + + +def mem_available_gib(meminfo: str = "/proc/meminfo") -> float | None: + """MemAvailable in GiB, or None where the system has no /proc/meminfo.""" + + try: + with open(meminfo) as f: + for line in f: + if line.startswith("MemAvailable:"): + return int(line.split()[1]) / (1 << 20) + except (OSError, ValueError, IndexError): + pass + return None diff --git a/src/tensorfold/families/deepseek_v41/cuda/weights.py b/src/tensorfold/families/deepseek_v41/cuda/weights.py index 331a684e3d..6d870cde67 100644 --- a/src/tensorfold/families/deepseek_v41/cuda/weights.py +++ b/src/tensorfold/families/deepseek_v41/cuda/weights.py @@ -24,6 +24,7 @@ from tensorfold.cuda.exl3.linear import Exl3Linear from ..config import Cfg +from .hostenv import drop_page_cache, mem_available_gib, release_enabled @dataclass @@ -129,7 +130,11 @@ def get(self, name: str, rows: tuple[int, int] | None = None) -> torch.Tensor: return t.reshape(shape) def close(self) -> None: + """Close the files. Their cached pages go back to the node first (``drop_page_cache``): a load reads most of + the checkpoint once, and on GB10 the page cache takes memory that the serving process needs.""" + for f in self._files.values(): + drop_page_cache(f.fileno()) f.close() self._files.clear() @@ -248,6 +253,7 @@ def close(self, ok: bool = True) -> bool: for f in self.ahead: f.cancel() self.pool.shutdown(wait=True) + drop_page_cache(self.fd) os.close(self.fd) if ok and self.i != len(self.index): self.path.unlink(missing_ok=True) @@ -259,6 +265,7 @@ def close(self, ok: bool = True) -> bool: self.f.write(body + struct.pack(" torch.Tensor: else "the rank cache write failed and was dropped") + ")") sh.close() _trim() + free = mem_available_gib() + if free is not None: + log(f"[tensorfold] rank {rank}: weights loaded, MemAvailable {free:.1f} GiB (page cache release " + f"{'on' if release_enabled() else 'off'})") return w diff --git a/tests/test_dsv41_host_env.py b/tests/test_dsv41_host_env.py new file mode 100644 index 0000000000..3bbfae9f34 --- /dev/null +++ b/tests/test_dsv41_host_env.py @@ -0,0 +1,307 @@ +"""DeepSeek-V4.1 host settings: the RoCE GID scan, the NCCL defaults and the page cache release (CPU only). + +Run with ``python -m pytest tests/test_dsv41_host_env.py`` or ``python -m unittest``. The weight loader tests need +PyTorch and skip without it. +""" + +from __future__ import annotations + +import importlib.util +import os +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +from tensorfold.cuda import roce_gid +from tensorfold.families.deepseek_v41.cuda import hostenv + +V4 = "0000:0000:0000:0000:0000:ffff:c0a8:0a01" # 192.168.10.1 +V4_B = "0000:0000:0000:0000:0000:ffff:c0a8:0a02" +LINK_LOCAL = "fe80:0000:0000:0000:aaaa:bbff:fecc:ddee" +EMPTY = "0000:0000:0000:0000:0000:0000:0000:0000" +FADV_DONTNEED = 4 # the Linux value; Windows has no os.POSIX_FADV_DONTNEED + + +def make_port(root: Path, device: str, entries: dict, port: int = 1) -> None: + """A fake sysfs port: ``entries`` maps an index to (type, gid); a type of None leaves the type file out.""" + + base = root / device / "ports" / str(port) + (base / "gids").mkdir(parents=True) + (base / "gid_attrs" / "types").mkdir(parents=True) + for n, (kind, gid) in entries.items(): + (base / "gids" / str(n)).write_text(gid + "\n") + if kind is not None: + (base / "gid_attrs" / "types" / str(n)).write_text(kind + "\n") + + +class GidScanTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.addCleanup(self._tmp.cleanup) + self.root = Path(self._tmp.name) + + def test_takes_the_roce_v2_ipv4_entry(self): + make_port(self.root, "rocep1s0f0", { + 0: ("IB/RoCE v1", LINK_LOCAL), + 1: ("RoCE v2", LINK_LOCAL), + 2: ("IB/RoCE v1", V4), + 3: ("RoCE v2", V4), + }) + self.assertEqual(roce_gid.detect_gid_index("rocep1s0f0", root=self.root), 3) + + def test_entries_sort_by_number_not_by_text(self): + make_port(self.root, "d", {2: ("RoCE v2", LINK_LOCAL), 10: ("RoCE v2", V4), 11: ("RoCE v2", V4_B)}) + self.assertEqual(roce_gid.detect_gid_index("d", root=self.root), 10) + self.assertEqual([n for n, _, _ in roce_gid.gid_entries("d", root=self.root)], [2, 10, 11]) + + def test_entry_without_a_type_file_is_skipped(self): + # an empty slot of the table: the gid file reads as zeros and the type file cannot be read + make_port(self.root, "d", {0: (None, EMPTY), 1: (None, V4), 2: ("RoCE v2", V4)}) + self.assertEqual(roce_gid.detect_gid_index("d", root=self.root), 2) + + def test_zero_ipv4_address_is_skipped(self): + make_port(self.root, "d", {0: ("RoCE v2", "0000:0000:0000:0000:0000:ffff:0000:0000"), 1: ("RoCE v2", V4)}) + self.assertEqual(roce_gid.detect_gid_index("d", root=self.root), 1) + + def test_no_match_gives_none(self): + make_port(self.root, "d", {0: ("IB/RoCE v1", V4), 1: ("RoCE v2", LINK_LOCAL)}) + self.assertIsNone(roce_gid.detect_gid_index("d", root=self.root)) + self.assertIsNone(roce_gid.detect_gid_index("missing", root=self.root)) + + def test_port_number_is_read(self): + make_port(self.root, "d", {0: ("RoCE v2", V4)}, port=2) + self.assertIsNone(roce_gid.detect_gid_index("d", 1, root=self.root)) + self.assertEqual(roce_gid.detect_gid_index("d", 2, root=self.root), 0) + + def test_each_device_has_its_own_index(self): + make_port(self.root, "a", {5: ("RoCE v2", V4)}) + make_port(self.root, "b", {3: ("RoCE v2", V4_B)}) + got = roce_gid.resolve_gid_indices(["a", "b"], env={}, root=self.root) + self.assertEqual(got, [(5, "detected"), (3, "detected")]) + + def test_unset_empty_and_minus_one_all_mean_detect(self): + make_port(self.root, "a", {4: ("RoCE v2", V4)}) + for env in ({}, {"NCCL_IB_GID_INDEX": ""}, {"NCCL_IB_GID_INDEX": "-1"}, {"TF_RDMA_GID_INDEX": " -1 "}): + with self.subTest(env=env): + self.assertEqual(roce_gid.resolve_gid_indices(["a"], env=env, root=self.root), [(4, "detected")]) + + def test_a_fixed_index_wins_over_the_table(self): + make_port(self.root, "a", {4: ("RoCE v2", V4)}) + self.assertEqual(roce_gid.resolve_gid_indices(["a"], env={"NCCL_IB_GID_INDEX": "7"}, root=self.root), + [(7, "fixed")]) + self.assertEqual(roce_gid.resolve_gid_indices(["a"], fixed=2, env={}, root=self.root), [(2, "fixed")]) + # the argument wins over the variables; a negative argument means detect + self.assertEqual(roce_gid.resolve_gid_indices(["a"], fixed=2, env={"TF_RDMA_GID_INDEX": "9"}, + root=self.root), [(2, "fixed")]) + self.assertEqual(roce_gid.resolve_gid_indices(["a"], fixed=-1, env={}, root=self.root), [(4, "detected")]) + + def test_tf_variable_decides_before_the_nccl_variable(self): + env = {"TF_RDMA_GID_INDEX": "6", "NCCL_IB_GID_INDEX": "3"} + self.assertEqual(roce_gid.gid_override(env), 6) + # TF_RDMA_GID_INDEX=-1 asks for detection even when NCCL_IB_GID_INDEX holds an index + self.assertIsNone(roce_gid.gid_override({"TF_RDMA_GID_INDEX": "-1", "NCCL_IB_GID_INDEX": "3"})) + + def test_bad_variable_is_named(self): + with self.assertRaisesRegex(ValueError, "NCCL_IB_GID_INDEX"): + roce_gid.gid_override({"NCCL_IB_GID_INDEX": "five"}) + + def test_missing_entry_is_an_error_that_names_the_device(self): + make_port(self.root, "a", {0: ("IB/RoCE v1", V4)}) + with self.assertRaisesRegex(RuntimeError, "a port 1 has no RoCE v2 IPv4 GID"): + roce_gid.resolve_gid_indices(["a"], env={}, root=self.root) + + +class NcclDefaultsTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.addCleanup(self._tmp.cleanup) + self.root = Path(self._tmp.name) + make_port(self.root, "a", {3: ("RoCE v2", V4)}) + make_port(self.root, "b", {3: ("RoCE v2", V4_B)}) + + def test_cumem_is_off_by_default(self): + env: dict = {} + done = hostenv.apply_nccl_defaults(env=env) + self.assertEqual(env, {"NCCL_CUMEM_ENABLE": "0", "NCCL_CUMEM_HOST_ENABLE": "0"}) + self.assertEqual(done, env) + + def test_a_value_that_the_user_set_stays(self): + env = {"NCCL_CUMEM_ENABLE": "1", "NCCL_CUMEM_HOST_ENABLE": "1"} + self.assertEqual(hostenv.apply_nccl_defaults(env=env), {}) + self.assertEqual(env, {"NCCL_CUMEM_ENABLE": "1", "NCCL_CUMEM_HOST_ENABLE": "1"}) + + def test_one_set_and_one_unset(self): + env = {"NCCL_CUMEM_HOST_ENABLE": "1"} + hostenv.apply_nccl_defaults(env=env) + self.assertEqual(env, {"NCCL_CUMEM_ENABLE": "0", "NCCL_CUMEM_HOST_ENABLE": "1"}) + + def test_an_empty_value_counts_as_unset(self): + env = {"NCCL_CUMEM_ENABLE": " "} + hostenv.apply_nccl_defaults(env=env) + self.assertEqual(env["NCCL_CUMEM_ENABLE"], "0") + + def test_the_process_environment_is_the_default_target(self): + with mock.patch.dict(os.environ, {}, clear=True): + hostenv.apply_nccl_defaults() + self.assertEqual(os.environ["NCCL_CUMEM_ENABLE"], "0") + self.assertEqual(os.environ["NCCL_CUMEM_HOST_ENABLE"], "0") + + def test_gid_index_is_set_from_the_table(self): + env: dict = {} + done = hostenv.apply_nccl_defaults(["a", "b"], env=env, root=self.root) + self.assertEqual(env["NCCL_IB_GID_INDEX"], "3") + self.assertEqual(done["NCCL_IB_GID_INDEX"], "3") + # the gather keeps its own per-port detection + self.assertEqual(env["TF_RDMA_GID_INDEX"], "-1") + self.assertEqual(roce_gid.resolve_gid_indices(["a", "b"], env=env, root=self.root), + [(3, "detected"), (3, "detected")]) + + def test_gid_index_that_the_user_set_stays(self): + for value in ("5", "-1"): + with self.subTest(value=value): + env = {"NCCL_IB_GID_INDEX": value} + hostenv.apply_nccl_defaults(["a"], env=env, root=self.root) + self.assertEqual(env["NCCL_IB_GID_INDEX"], value) + self.assertNotIn("TF_RDMA_GID_INDEX", env) + + def test_user_tf_variable_stays(self): + env = {"TF_RDMA_GID_INDEX": "8"} + hostenv.apply_nccl_defaults(["a"], env=env, root=self.root) + self.assertEqual(env["TF_RDMA_GID_INDEX"], "8") + + def test_gid_index_stays_unset_when_devices_differ_or_have_no_entry(self): + make_port(self.root, "c", {6: ("RoCE v2", V4)}) + make_port(self.root, "d", {0: ("IB/RoCE v1", V4)}) + for devices in (["a", "c"], ["a", "d"], ["d"], ["missing"]): + with self.subTest(devices=devices): + env: dict = {} + hostenv.apply_nccl_defaults(devices, env=env, root=self.root) + self.assertNotIn("NCCL_IB_GID_INDEX", env) + + def test_no_devices_leave_the_gid_index_alone(self): + env: dict = {} + hostenv.apply_nccl_defaults(env=env, root=self.root) + self.assertNotIn("NCCL_IB_GID_INDEX", env) + + +class PageCacheTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.addCleanup(self._tmp.cleanup) + self.path = Path(self._tmp.name) / "weights.bin" + self.path.write_bytes(os.urandom(1 << 16)) + self.on = mock.patch.dict(os.environ, {}, clear=False) + self.on.start() + self.addCleanup(self.on.stop) + os.environ.pop(hostenv.RELEASE_VAR, None) + + def patched(self, **kwargs): + advise = mock.Mock(return_value=None, **kwargs) + return advise, (mock.patch.object(os, "posix_fadvise", advise, create=True), + mock.patch.object(os, "POSIX_FADV_DONTNEED", FADV_DONTNEED, create=True)) + + def test_switch_default_and_values(self): + self.assertTrue(hostenv.release_enabled({})) + self.assertTrue(hostenv.release_enabled({hostenv.RELEASE_VAR: "1"})) + for off in ("0", "off", "False", " no "): + self.assertFalse(hostenv.release_enabled({hostenv.RELEASE_VAR: off})) + + def test_a_file_descriptor_gets_dontneed_on_the_whole_file(self): + advise, (a, b) = self.patched() + with a, b, open(self.path, "rb") as f: + self.assertTrue(hostenv.drop_page_cache(f.fileno())) + advise.assert_called_once_with(f.fileno(), 0, 0, FADV_DONTNEED) + + def test_a_path_is_opened_advised_and_closed(self): + seen = [] + advise, (a, b) = self.patched() + advise.side_effect = lambda fd, off, n, flag: seen.append(os.fstat(fd).st_size) + with a, b: + self.assertTrue(hostenv.drop_page_cache(self.path)) + self.assertTrue(hostenv.drop_page_cache(str(self.path))) + self.assertEqual(seen, [1 << 16, 1 << 16]) + self.assertEqual(advise.call_count, 2) + + def test_the_switch_turns_it_off(self): + advise, (a, b) = self.patched() + with a, b, mock.patch.dict(os.environ, {hostenv.RELEASE_VAR: "0"}), open(self.path, "rb") as f: + self.assertFalse(hostenv.drop_page_cache(f.fileno())) + advise.assert_not_called() + + def test_a_refused_call_gives_false(self): + advise, (a, b) = self.patched() + advise.side_effect = OSError(22, "Invalid argument") + with a, b, open(self.path, "rb") as f: + self.assertFalse(hostenv.drop_page_cache(f.fileno())) + + def test_a_missing_path_gives_false(self): + advise, (a, b) = self.patched() + with a, b: + self.assertFalse(hostenv.drop_page_cache(self.path.with_name("none.bin"))) + advise.assert_not_called() + + @unittest.skipUnless(hasattr(os, "posix_fadvise"), "this system has no posix_fadvise") + def test_the_real_call_works_on_a_temp_file(self): + self.assertTrue(hostenv.drop_page_cache(self.path)) + with open(self.path, "rb") as f: + self.assertTrue(hostenv.drop_page_cache(f.fileno())) + self.assertEqual(len(f.read()), 1 << 16) # the file reads the same bytes after the drop + + def test_mem_available_reads_meminfo(self): + f = Path(self._tmp.name) / "meminfo" + f.write_text("MemTotal: 131072000 kB\nMemAvailable: 2097152 kB\nCached: 1 kB\n") + self.assertAlmostEqual(hostenv.mem_available_gib(str(f)), 2.0) + self.assertIsNone(hostenv.mem_available_gib(str(f.with_name("none")))) + + +@unittest.skipIf(importlib.util.find_spec("torch") is None, "needs PyTorch (the CUDA backend's code)") +class LoaderReleaseTests(unittest.TestCase): + """The weight loader gives its files' pages back when it closes them.""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.addCleanup(self._tmp.cleanup) + self.dir = Path(self._tmp.name) + from tensorfold.families.deepseek_v41.cuda import weights + + self.weights = weights + + def test_shards_close_drops_every_open_file(self): + import json + import struct + + for name, tensor in (("a.safetensors", "x"), ("b.safetensors", "y")): + header = json.dumps({tensor: {"dtype": "U8", "shape": [4], "data_offsets": [0, 4]}}).encode() + (self.dir / name).write_bytes(struct.pack("`, GID index 5) for prompt chunks. + CX7) for decode-sized payloads, NCCL over RoCE (the CX7 HCA, `NCCL_IB_HCA=`) for prompt chunks. Each port's GID index is the lowest RoCE v2 IPv4 entry + of its sysfs GID table, read at start (`cuda/roce_gid.py`; a node's table can differ from the other node's, and a link flap can + move an entry). `TF_RDMA_GID_INDEX` or `NCCL_IB_GID_INDEX` with a value of 0 or more fixes it; unset, empty or -1 means detect. + When `NCCL_IB_GID_INDEX` is unset, the engine sets it to the detected index, and it sets `NCCL_CUMEM_ENABLE=0` and + `NCCL_CUMEM_HOST_ENABLE=0` (cuMem allocations in NCCL ran the GB10 driver out of memory under many streams) unless the user set them. Budget ≈ 2 ms a token at 84 gathers. - Prompt chunks overlap the gather of one micro-batch with the next one's compute (as in our GLM-5.3 fork). @@ -123,5 +127,9 @@ pieces vs reference per layer. 4. TP2 serial decode end to end, oracle agreement ## 11. Risks - Memory: 97 GiB a rank leaves ~10 GiB; Engram I/O must use no pinned table memory. +- Page cache: a load reads the checkpoint (or the rank cache, about 106 GB a rank) through the page cache, and on GB10 + those pages take memory that serving needs. `Shards.close` and `RankCache.close` call `posix_fadvise(DONTNEED)` on + the files they read (`hostenv.drop_page_cache`); `TF_DS_RELEASE_PAGE_CACHE=0` turns it off. The load log ends with + the node's MemAvailable. A restart reads the files from disk again, not from the page cache. - EXL3 2/3-bit expert kernels on GB10 decide decode speed (~3.5 GB of weights read a token a rank, ~78 tok/s ceiling). - The kit may use bounded decoder replay or a different KV format; the oracle comparison will show it. From 29117570364aa6720a0ce3c8d3f7fa0902f779a2 Mon Sep 17 00:00:00 2001 From: Elevatormusic <22101396+Elevatormusic@users.noreply.github.com> Date: Tue, 20 Oct 2026 18:37:55 -0700 Subject: [PATCH 2/4] Log the page cache size beside MemAvailable after the weight load --- src/tensorfold/families/deepseek_v41/cuda/hostenv.py | 6 +++--- src/tensorfold/families/deepseek_v41/cuda/weights.py | 10 +++++----- tests/test_dsv41_host_env.py | 10 ++++++---- 3 files changed, 14 insertions(+), 12 deletions(-) diff --git a/src/tensorfold/families/deepseek_v41/cuda/hostenv.py b/src/tensorfold/families/deepseek_v41/cuda/hostenv.py index 673d3c514a..06a36bd5f1 100644 --- a/src/tensorfold/families/deepseek_v41/cuda/hostenv.py +++ b/src/tensorfold/families/deepseek_v41/cuda/hostenv.py @@ -84,13 +84,13 @@ def drop_page_cache(target, *, env: Mapping[str, str] | None = None) -> bool: return True -def mem_available_gib(meminfo: str = "/proc/meminfo") -> float | None: - """MemAvailable in GiB, or None where the system has no /proc/meminfo.""" +def meminfo_gib(field: str, meminfo: str = "/proc/meminfo") -> float | None: + """One field of /proc/meminfo (for example "MemAvailable" or "Cached") in GiB, or None when it is not there.""" try: with open(meminfo) as f: for line in f: - if line.startswith("MemAvailable:"): + if line.startswith(field + ":"): return int(line.split()[1]) / (1 << 20) except (OSError, ValueError, IndexError): pass diff --git a/src/tensorfold/families/deepseek_v41/cuda/weights.py b/src/tensorfold/families/deepseek_v41/cuda/weights.py index 6d870cde67..27e5d53db2 100644 --- a/src/tensorfold/families/deepseek_v41/cuda/weights.py +++ b/src/tensorfold/families/deepseek_v41/cuda/weights.py @@ -24,7 +24,7 @@ from tensorfold.cuda.exl3.linear import Exl3Linear from ..config import Cfg -from .hostenv import drop_page_cache, mem_available_gib, release_enabled +from .hostenv import drop_page_cache, meminfo_gib, release_enabled @dataclass @@ -575,10 +575,10 @@ def plain(name: str, dtype=None) -> torch.Tensor: else "the rank cache write failed and was dropped") + ")") sh.close() _trim() - free = mem_available_gib() - if free is not None: - log(f"[tensorfold] rank {rank}: weights loaded, MemAvailable {free:.1f} GiB (page cache release " - f"{'on' if release_enabled() else 'off'})") + free, cached = meminfo_gib("MemAvailable"), meminfo_gib("Cached") + if free is not None and cached is not None: + log(f"[tensorfold] rank {rank}: weights loaded, MemAvailable {free:.1f} GiB, page cache {cached:.1f} GiB " + f"(page cache release {'on' if release_enabled() else 'off'})") return w diff --git a/tests/test_dsv41_host_env.py b/tests/test_dsv41_host_env.py index 3bbfae9f34..bec1b99586 100644 --- a/tests/test_dsv41_host_env.py +++ b/tests/test_dsv41_host_env.py @@ -248,11 +248,13 @@ def test_the_real_call_works_on_a_temp_file(self): self.assertTrue(hostenv.drop_page_cache(f.fileno())) self.assertEqual(len(f.read()), 1 << 16) # the file reads the same bytes after the drop - def test_mem_available_reads_meminfo(self): + def test_meminfo_reads_one_field(self): f = Path(self._tmp.name) / "meminfo" - f.write_text("MemTotal: 131072000 kB\nMemAvailable: 2097152 kB\nCached: 1 kB\n") - self.assertAlmostEqual(hostenv.mem_available_gib(str(f)), 2.0) - self.assertIsNone(hostenv.mem_available_gib(str(f.with_name("none")))) + f.write_text("MemTotal: 131072000 kB\nMemAvailable: 2097152 kB\nCached: 1048576 kB\nSwapCached: 5 kB\n") + self.assertAlmostEqual(hostenv.meminfo_gib("MemAvailable", str(f)), 2.0) + self.assertAlmostEqual(hostenv.meminfo_gib("Cached", str(f)), 1.0) # not SwapCached + self.assertIsNone(hostenv.meminfo_gib("Missing", str(f))) + self.assertIsNone(hostenv.meminfo_gib("Cached", str(f.with_name("none")))) @unittest.skipIf(importlib.util.find_spec("torch") is None, "needs PyTorch (the CUDA backend's code)") From 0dce299e94f82eead5513f0816607290a358ddb9 Mon Sep 17 00:00:00 2001 From: Elevatormusic <22101396+Elevatormusic@users.noreply.github.com> Date: Tue, 20 Oct 2026 18:44:36 -0700 Subject: [PATCH 3/4] Take the NCCL GID index from NCCL's own HCA list, and let gid_index=-1 bypass the variables --- src/tensorfold/cuda/rdma.py | 13 +-- src/tensorfold/cuda/roce_gid.py | 53 +++++++++++- .../families/deepseek_v41/cuda/engine.py | 4 +- .../families/deepseek_v41/cuda/hostenv.py | 16 ++-- tests/test_dsv41_host_env.py | 84 ++++++++++++++++--- tools/dsv41/DESIGN.md | 3 +- 6 files changed, 137 insertions(+), 36 deletions(-) diff --git a/src/tensorfold/cuda/rdma.py b/src/tensorfold/cuda/rdma.py index 6c346f57de..4182534a16 100644 --- a/src/tensorfold/cuda/rdma.py +++ b/src/tensorfold/cuda/rdma.py @@ -28,7 +28,7 @@ import torch -from .roce_gid import resolve_gid_indices +from .roce_gid import active_roce_devices, resolve_gid_indices @lru_cache(maxsize=1) @@ -48,16 +48,7 @@ def device_names() -> list[str]: if named: return named first = (os.environ.get("NCCL_IB_HCA") or "").strip().lstrip("=^").split(",")[0].split(":")[0].strip() - root = Path("/sys/class/infiniband") - active = [] - for d in sorted(p.name for p in root.iterdir()) if root.exists() else []: - try: - state = (root / d / "ports" / "1" / "state").read_text() - layer = (root / d / "ports" / "1" / "link_layer").read_text().strip() - except OSError: - continue - if "ACTIVE" in state and layer == "Ethernet": - active.append(d) + active = active_roce_devices() if first in active: active.remove(first) active.insert(0, first) diff --git a/src/tensorfold/cuda/roce_gid.py b/src/tensorfold/cuda/roce_gid.py index 4878b944d1..7f844d6a16 100644 --- a/src/tensorfold/cuda/roce_gid.py +++ b/src/tensorfold/cuda/roce_gid.py @@ -61,6 +61,53 @@ def detect_gid_index(device: str, port: int = 1, *, root: Path | str = SYSFS) -> return None +def active_roce_devices(*, root: Path | str = SYSFS) -> list[str]: + """The RDMA devices whose port 1 is ACTIVE with an Ethernet link layer (RoCE), in name order.""" + + root = Path(root) + out = [] + try: + names = sorted(p.name for p in root.iterdir()) + except OSError: + return out + for d in names: + state, layer = _read(root / d / "ports" / "1" / "state"), _read(root / d / "ports" / "1" / "link_layer") + if state is not None and "ACTIVE" in state and layer == "Ethernet": + out.append(d) + return out + + +def nccl_hca_ports(env: Mapping[str, str] | None = None, *, root: Path | str = SYSFS) -> list[tuple[str, int]] | None: + """The ``(device, port)`` pairs that NCCL_IB_HCA lets NCCL use, among the active RoCE devices. + + NCCL_IB_HCA is ``[^][=][:],...``: a name matches the start of a device name, or the whole name after + ``=``. Unset or empty: every active RoCE device (port 1). A list with ``^`` excludes devices: returns None, as the + scan does not follow NCCL's rules for it. A pair without a port uses port 1. + """ + + env = os.environ if env is None else env + raw = (env.get("NCCL_IB_HCA") or "").strip() + active = active_roce_devices(root=root) + if not raw: + return [(d, 1) for d in active] + if raw.startswith("^"): + return None + exact = raw.startswith("=") + out = [] + for item in raw.lstrip("=").split(","): + name, _, port = item.strip().partition(":") + if not name: + continue + try: + number = int(port) if port.strip() else 1 + except ValueError: + return None + for d in active: + if (d == name if exact else d.startswith(name)) and (d, number) not in out: + out.append((d, number)) + return out + + def gid_override(env: Mapping[str, str] | None = None) -> int | None: """The index that the user fixed, or None when the user did not fix one (variable unset, empty or -1).""" @@ -82,14 +129,14 @@ def resolve_gid_indices(devices: Sequence[str], port: int = 1, *, fixed: int | N root: Path | str = SYSFS) -> list[tuple[int, str]]: """One ``(index, how)`` for each device. ``how`` is "fixed" (an argument or a variable) or "detected". - ``fixed`` (an argument) wins over the variables. A negative ``fixed`` means "detect". + ``fixed`` (an argument) wins over the variables. A negative ``fixed`` means "detect", and the variables are not read. Raises RuntimeError when a device has no RoCE v2 IPv4 entry and nothing fixes the index. """ - if fixed is not None and fixed < 0: - fixed = None if fixed is None: fixed = gid_override(env) + elif fixed < 0: + fixed = None out = [] for device in devices: if fixed is not None: diff --git a/src/tensorfold/families/deepseek_v41/cuda/engine.py b/src/tensorfold/families/deepseek_v41/cuda/engine.py index dca437d7c3..84f465bc0b 100644 --- a/src/tensorfold/families/deepseek_v41/cuda/engine.py +++ b/src/tensorfold/families/deepseek_v41/cuda/engine.py @@ -194,11 +194,9 @@ def __init__(self, model_dir: Path, *, rank: int, world: int, master: str, port: def _nccl_host_env(self) -> None: """NCCL's cuMem allocations off and its GID index set (hostenv.py); a value that the user set stays.""" - from tensorfold.cuda.rdma import device_names - from .hostenv import apply_nccl_defaults - done = apply_nccl_defaults(device_names()) + done = apply_nccl_defaults() if done: print(f"[tensorfold] rank {self.rank}: NCCL settings by default: " + " ".join(f"{k}={v}" for k, v in done.items()), flush=True) diff --git a/src/tensorfold/families/deepseek_v41/cuda/hostenv.py b/src/tensorfold/families/deepseek_v41/cuda/hostenv.py index 06a36bd5f1..4e78afbaa0 100644 --- a/src/tensorfold/families/deepseek_v41/cuda/hostenv.py +++ b/src/tensorfold/families/deepseek_v41/cuda/hostenv.py @@ -13,10 +13,10 @@ from __future__ import annotations import os -from collections.abc import Mapping, Sequence +from collections.abc import Mapping from pathlib import Path -from tensorfold.cuda.roce_gid import SYSFS, detect_gid_index +from tensorfold.cuda.roce_gid import SYSFS, detect_gid_index, nccl_hca_ports NCCL_DEFAULTS = {"NCCL_CUMEM_ENABLE": "0", "NCCL_CUMEM_HOST_ENABLE": "0"} RELEASE_VAR = "TF_DS_RELEASE_PAGE_CACHE" @@ -26,12 +26,13 @@ def _is_set(env: Mapping[str, str], name: str) -> bool: return bool((env.get(name) or "").strip()) -def apply_nccl_defaults(devices: Sequence[str] = (), *, env: dict | None = None, root: Path | str = SYSFS, - port: int = 1) -> dict[str, str]: +def apply_nccl_defaults(*, env: dict | None = None, root: Path | str = SYSFS) -> dict[str, str]: """Set what the user did not set, before NCCL starts. Returns the variables that this call set. * ``NCCL_CUMEM_ENABLE`` and ``NCCL_CUMEM_HOST_ENABLE`` become 0. - * ``NCCL_IB_GID_INDEX`` becomes the RoCE v2 IPv4 GID index of ``devices`` when every device has the same one. + * ``NCCL_IB_GID_INDEX`` becomes the RoCE v2 IPv4 GID index of the ports that NCCL_IB_HCA lets NCCL use (every + active RoCE port when it is unset), when each of those ports has the same index. With no such port, an + exclusion list in NCCL_IB_HCA, or ports that differ, NCCL keeps its own choice. A value that is set (also -1, which is NCCL's own detection) stays as it is. When this call sets the index, it also sets ``TF_RDMA_GID_INDEX=-1`` (if that is unset) so that the RDMA gather keeps detecting each port itself. @@ -42,8 +43,9 @@ def apply_nccl_defaults(devices: Sequence[str] = (), *, env: dict | None = None, for name, value in NCCL_DEFAULTS.items(): if not _is_set(env, name): env[name] = done[name] = value - if devices and not _is_set(env, "NCCL_IB_GID_INDEX"): - found = {detect_gid_index(d, port, root=root) for d in devices} + ports = nccl_hca_ports(env, root=root) if not _is_set(env, "NCCL_IB_GID_INDEX") else None + if ports: + found = {detect_gid_index(d, p, root=root) for d, p in ports} if len(found) == 1 and None not in found: env["NCCL_IB_GID_INDEX"] = done["NCCL_IB_GID_INDEX"] = str(found.pop()) # the RDMA gather reads NCCL_IB_GID_INDEX as the user's choice; this marker keeps its own detection on diff --git a/tests/test_dsv41_host_env.py b/tests/test_dsv41_host_env.py index bec1b99586..2f16275f0a 100644 --- a/tests/test_dsv41_host_env.py +++ b/tests/test_dsv41_host_env.py @@ -23,11 +23,14 @@ FADV_DONTNEED = 4 # the Linux value; Windows has no os.POSIX_FADV_DONTNEED -def make_port(root: Path, device: str, entries: dict, port: int = 1) -> None: +def make_port(root: Path, device: str, entries: dict, port: int = 1, *, state: str = "4: ACTIVE", + layer: str = "Ethernet") -> None: """A fake sysfs port: ``entries`` maps an index to (type, gid); a type of None leaves the type file out.""" base = root / device / "ports" / str(port) (base / "gids").mkdir(parents=True) + (base / "state").write_text(state + "\n") + (base / "link_layer").write_text(layer + "\n") (base / "gid_attrs" / "types").mkdir(parents=True) for n, (kind, gid) in entries.items(): (base / "gids" / str(n)).write_text(gid + "\n") @@ -96,6 +99,13 @@ def test_a_fixed_index_wins_over_the_table(self): root=self.root), [(2, "fixed")]) self.assertEqual(roce_gid.resolve_gid_indices(["a"], fixed=-1, env={}, root=self.root), [(4, "detected")]) + def test_a_negative_argument_bypasses_the_variables(self): + make_port(self.root, "a", {4: ("RoCE v2", V4)}) + for env in ({"NCCL_IB_GID_INDEX": "9"}, {"TF_RDMA_GID_INDEX": "9"}): + with self.subTest(env=env): + self.assertEqual(roce_gid.resolve_gid_indices(["a"], fixed=-1, env=env, root=self.root), + [(4, "detected")]) + def test_tf_variable_decides_before_the_nccl_variable(self): env = {"TF_RDMA_GID_INDEX": "6", "NCCL_IB_GID_INDEX": "3"} self.assertEqual(roce_gid.gid_override(env), 6) @@ -112,6 +122,40 @@ def test_missing_entry_is_an_error_that_names_the_device(self): roce_gid.resolve_gid_indices(["a"], env={}, root=self.root) +class DeviceListTests(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.addCleanup(self._tmp.cleanup) + self.root = Path(self._tmp.name) + make_port(self.root, "rocep1s0f0", {}, state="1: DOWN") + make_port(self.root, "rocep1s0f1", {}) + make_port(self.root, "roceP2p1s0f1", {}) + make_port(self.root, "mlx5_ib", {}, layer="InfiniBand") + + def test_active_roce_devices_are_the_active_ethernet_ports(self): + self.assertEqual(roce_gid.active_roce_devices(root=self.root), ["roceP2p1s0f1", "rocep1s0f1"]) + self.assertEqual(roce_gid.active_roce_devices(root=self.root / "none"), []) + + def test_nccl_hca_ports(self): + both = [("roceP2p1s0f1", 1), ("rocep1s0f1", 1)] + cases = [ + ({}, both), + ({"NCCL_IB_HCA": ""}, both), + ({"NCCL_IB_HCA": "rocep1s0f1"}, [("rocep1s0f1", 1)]), + ({"NCCL_IB_HCA": "=rocep1s0f1,roceP2p1s0f1"}, [("rocep1s0f1", 1), ("roceP2p1s0f1", 1)]), + ({"NCCL_IB_HCA": "rocep1s0f"}, [("rocep1s0f1", 1)]), # a prefix; f0 is down + ({"NCCL_IB_HCA": "=rocep1s0f"}, []), # "=" is an exact match + ({"NCCL_IB_HCA": "rocep1s0f1:1"}, [("rocep1s0f1", 1)]), + ({"NCCL_IB_HCA": "rocep1s0f1:2"}, [("rocep1s0f1", 2)]), + ({"NCCL_IB_HCA": "mlx5"}, []), # not RoCE + ({"NCCL_IB_HCA": "^rocep1s0f1"}, None), # an exclusion list is not followed + ({"NCCL_IB_HCA": "rocep1s0f1:x"}, None), + ] + for env, want in cases: + with self.subTest(env=env): + self.assertEqual(roce_gid.nccl_hca_ports(env, root=self.root), want) + + class NcclDefaultsTests(unittest.TestCase): def setUp(self): self._tmp = tempfile.TemporaryDirectory() @@ -120,6 +164,9 @@ def setUp(self): make_port(self.root, "a", {3: ("RoCE v2", V4)}) make_port(self.root, "b", {3: ("RoCE v2", V4_B)}) + def apply(self, env: dict) -> dict: + return hostenv.apply_nccl_defaults(env=env, root=self.root) + def test_cumem_is_off_by_default(self): env: dict = {} done = hostenv.apply_nccl_defaults(env=env) @@ -149,7 +196,7 @@ def test_the_process_environment_is_the_default_target(self): def test_gid_index_is_set_from_the_table(self): env: dict = {} - done = hostenv.apply_nccl_defaults(["a", "b"], env=env, root=self.root) + done = self.apply(env) self.assertEqual(env["NCCL_IB_GID_INDEX"], "3") self.assertEqual(done["NCCL_IB_GID_INDEX"], "3") # the gather keeps its own per-port detection @@ -161,27 +208,42 @@ def test_gid_index_that_the_user_set_stays(self): for value in ("5", "-1"): with self.subTest(value=value): env = {"NCCL_IB_GID_INDEX": value} - hostenv.apply_nccl_defaults(["a"], env=env, root=self.root) + self.apply(env) self.assertEqual(env["NCCL_IB_GID_INDEX"], value) self.assertNotIn("TF_RDMA_GID_INDEX", env) def test_user_tf_variable_stays(self): env = {"TF_RDMA_GID_INDEX": "8"} - hostenv.apply_nccl_defaults(["a"], env=env, root=self.root) + self.apply(env) self.assertEqual(env["TF_RDMA_GID_INDEX"], "8") - def test_gid_index_stays_unset_when_devices_differ_or_have_no_entry(self): + def test_gid_index_comes_from_the_hcas_that_nccl_uses(self): + make_port(self.root, "c", {6: ("RoCE v2", V4)}) + for hca, want in (("=c", "6"), ("c", "6"), ("=a", "3"), ("=a,b", "3")): + with self.subTest(hca=hca): + env = {"NCCL_IB_HCA": hca} + self.apply(env) + self.assertEqual(env["NCCL_IB_GID_INDEX"], want) + + def test_gid_index_stays_unset_when_the_ports_differ_or_cannot_be_named(self): make_port(self.root, "c", {6: ("RoCE v2", V4)}) make_port(self.root, "d", {0: ("IB/RoCE v1", V4)}) - for devices in (["a", "c"], ["a", "d"], ["d"], ["missing"]): - with self.subTest(devices=devices): - env: dict = {} - hostenv.apply_nccl_defaults(devices, env=env, root=self.root) + for hca in ("=a,c", "a,c", "=d", "^a", "=none", "a:x"): + with self.subTest(hca=hca): + env = {"NCCL_IB_HCA": hca} + self.apply(env) self.assertNotIn("NCCL_IB_GID_INDEX", env) + self.assertNotIn("TF_RDMA_GID_INDEX", env) + + def test_all_active_ports_count_when_nccl_ib_hca_is_unset(self): + make_port(self.root, "c", {6: ("RoCE v2", V4)}) + env: dict = {} + self.apply(env) # a, b and c are active: 3, 3 and 6 + self.assertNotIn("NCCL_IB_GID_INDEX", env) - def test_no_devices_leave_the_gid_index_alone(self): + def test_no_ports_leave_the_gid_index_alone(self): env: dict = {} - hostenv.apply_nccl_defaults(env=env, root=self.root) + hostenv.apply_nccl_defaults(env=env, root=self.root / "none") self.assertNotIn("NCCL_IB_GID_INDEX", env) diff --git a/tools/dsv41/DESIGN.md b/tools/dsv41/DESIGN.md index 13549d5746..f286afb79d 100644 --- a/tools/dsv41/DESIGN.md +++ b/tools/dsv41/DESIGN.md @@ -51,7 +51,8 @@ tokens; workspace and graphs ~3–4 GiB. The kit allocates ~99.5 GiB weights a r CX7) for decode-sized payloads, NCCL over RoCE (the CX7 HCA, `NCCL_IB_HCA=`) for prompt chunks. Each port's GID index is the lowest RoCE v2 IPv4 entry of its sysfs GID table, read at start (`cuda/roce_gid.py`; a node's table can differ from the other node's, and a link flap can move an entry). `TF_RDMA_GID_INDEX` or `NCCL_IB_GID_INDEX` with a value of 0 or more fixes it; unset, empty or -1 means detect. - When `NCCL_IB_GID_INDEX` is unset, the engine sets it to the detected index, and it sets `NCCL_CUMEM_ENABLE=0` and + When `NCCL_IB_GID_INDEX` is unset, the engine sets it to the detected index of the ports that `NCCL_IB_HCA` lets NCCL use + (every active RoCE port when it is unset) if they agree, and it sets `NCCL_CUMEM_ENABLE=0` and `NCCL_CUMEM_HOST_ENABLE=0` (cuMem allocations in NCCL ran the GB10 driver out of memory under many streams) unless the user set them. Budget ≈ 2 ms a token at 84 gathers. - Prompt chunks overlap the gather of one micro-batch with the next one's compute (as in our GLM-5.3 fork). From 269967ab6212ff379c0d41963a33b22ef5507fab Mon Sep 17 00:00:00 2001 From: Elevatormusic <22101396+Elevatormusic@users.noreply.github.com> Date: Tue, 20 Oct 2026 19:02:46 -0700 Subject: [PATCH 4/4] Tests: keep the cumem tests off the real sysfs --- tests/test_dsv41_host_env.py | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/tests/test_dsv41_host_env.py b/tests/test_dsv41_host_env.py index 2f16275f0a..0cea55ac92 100644 --- a/tests/test_dsv41_host_env.py +++ b/tests/test_dsv41_host_env.py @@ -163,34 +163,35 @@ def setUp(self): self.root = Path(self._tmp.name) make_port(self.root, "a", {3: ("RoCE v2", V4)}) make_port(self.root, "b", {3: ("RoCE v2", V4_B)}) + self.nothing = self.root / "none" # a sysfs root with no device: the cumem tests only def apply(self, env: dict) -> dict: return hostenv.apply_nccl_defaults(env=env, root=self.root) def test_cumem_is_off_by_default(self): env: dict = {} - done = hostenv.apply_nccl_defaults(env=env) + done = hostenv.apply_nccl_defaults(env=env, root=self.nothing) self.assertEqual(env, {"NCCL_CUMEM_ENABLE": "0", "NCCL_CUMEM_HOST_ENABLE": "0"}) self.assertEqual(done, env) def test_a_value_that_the_user_set_stays(self): env = {"NCCL_CUMEM_ENABLE": "1", "NCCL_CUMEM_HOST_ENABLE": "1"} - self.assertEqual(hostenv.apply_nccl_defaults(env=env), {}) + self.assertEqual(hostenv.apply_nccl_defaults(env=env, root=self.nothing), {}) self.assertEqual(env, {"NCCL_CUMEM_ENABLE": "1", "NCCL_CUMEM_HOST_ENABLE": "1"}) def test_one_set_and_one_unset(self): env = {"NCCL_CUMEM_HOST_ENABLE": "1"} - hostenv.apply_nccl_defaults(env=env) + hostenv.apply_nccl_defaults(env=env, root=self.nothing) self.assertEqual(env, {"NCCL_CUMEM_ENABLE": "0", "NCCL_CUMEM_HOST_ENABLE": "1"}) def test_an_empty_value_counts_as_unset(self): env = {"NCCL_CUMEM_ENABLE": " "} - hostenv.apply_nccl_defaults(env=env) + hostenv.apply_nccl_defaults(env=env, root=self.nothing) self.assertEqual(env["NCCL_CUMEM_ENABLE"], "0") def test_the_process_environment_is_the_default_target(self): with mock.patch.dict(os.environ, {}, clear=True): - hostenv.apply_nccl_defaults() + hostenv.apply_nccl_defaults(root=self.nothing) self.assertEqual(os.environ["NCCL_CUMEM_ENABLE"], "0") self.assertEqual(os.environ["NCCL_CUMEM_HOST_ENABLE"], "0")