[NCCL][PyTorch] Fix test-fusible-ops-file-rendezvous-bug (#3478)

* world_group() used a single hardcoded init_method="file:///tmp/rdzv",
shared across every world_size this test parametrizes over
([device_count(), 1, 2]), and neither world_group() nor
test_distributed_fuser_ops ever calls destroy_process_group() or
removes the file afterwards. A world_size=N run's leftover FileStore
content can then corrupt a differently-shaped world_size=M run that
reuses the same path in the same pytest session, surfacing as a
confusing NCCL bootstrap failure:

  torch.distributed.DistBackendError: NCCL error ...
  ncclOsSocketPollConnect: connect to <self> ... Connection refused,
  exceeded error retry count after 35 attempts

instead of a clear rendezvous error. Confirmed 100% deterministic:
running the full file fresh (no pre-existing /tmp/rdzv) still fails
test_distributed_fuser_ops[2] every time, because the [4] parametrization
(which runs first) leaves /tmp/rdzv behind for [2] to trip over.

This was previously masked by older bundled NCCL (2.30.7), which
apparently tolerated the stale/mismatched FileStore well enough to
still succeed; a newer NCCL (2.31.2) surfaces it as a hard failure.
See the investigation writeup for the full comparison:

Fix: key the rendezvous path by world_size
(file:///tmp/rdzv_test_fusible_ops_{world_size}), and defensively
remove any pre-existing file at that path in test_distributed_fuser_ops
before launching each subprocess job, to also cover a leftover file
from an earlier crashed run of the same world_size.

Verified on a GB200 node
NCCL 2.31.2, the container that previously failed 2 of 3 parametrizations):
all 3 world_size parametrizations now pass, including two runs of the
full file back to back with no manual cleanup in between.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Signed-off-by: Wei Wang <weiwan@nvidia.com>

* Use unique rendezvous files in fusible ops tests

Signed-off-by: Przemek Tredak <ptredak@nvidia.com>

---------

Signed-off-by: Wei Wang <weiwan@nvidia.com>
Signed-off-by: Przemek Tredak <ptredak@nvidia.com>
Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>
Co-authored-by: Przemek Tredak <ptredak@nvidia.com>

Author: 3 people

tests/pytorch/distributed/test_fusible_ops.py

@@ -12,6 +12,7 @@
 import pathlib
 import subprocess
 import sys
+import tempfile
 from typing import Optional
 
 import pytest
@@ -58,7 +59,8 @@ def world_group() -> torch.distributed.ProcessGroup:
     torch.cuda.set_device(rank)
     group = torch.distributed.init_process_group(
         "nccl",
-        init_method="file:///tmp/rdzv",
+        # Each parallel job must use a fresh FileStore shared by only its ranks.
+        init_method=f"file://{os.environ['NVTE_TEST_RDZV_PATH']}",
         world_size=world_size,
         rank=rank,
     )
@@ -1053,10 +1055,10 @@ def test_distributed_fuser_ops(world_size: int) -> None:
         current_file,
         "--parallel",
     ]
-    result = subprocess.run(
-        command,
-        check=True,
-    )
+    with tempfile.TemporaryDirectory(prefix="te-test-fusible-ops-") as temp_dir:
+        env = dict(os.environ)
+        env["NVTE_TEST_RDZV_PATH"] = str(pathlib.Path(temp_dir) / "rdzv")
+        subprocess.run(command, check=True, env=env)
 
 
 def main() -> None:

tests/pytorch/distributed/test_fusible_ops_with_userbuffers.py

@@ -12,6 +12,7 @@
 import pathlib
 import subprocess
 import sys
+import tempfile
 
 import pytest
 import torch
@@ -107,7 +108,8 @@ def world_group() -> torch.distributed.ProcessGroup:
     torch.cuda.set_device(local_rank)
     group = torch.distributed.init_process_group(
         "nccl",
-        init_method="file:///tmp/rdzv",
+        # Each parallel job must use a fresh FileStore shared by only its ranks.
+        init_method=f"file://{os.environ['NVTE_TEST_RDZV_PATH']}",
         world_size=world_size,
         rank=rank,
         device_id=torch.device(f"cuda:{local_rank}"),
@@ -471,7 +473,9 @@ def test_fuser_ops_with_userbuffers(
     env["NVTE_ALLOW_NONDETERMINISTIC_ALGO"] = "0"
 
     # Launch parallel job
-    run_distributed(command, env=env)
+    with tempfile.TemporaryDirectory(prefix="te-test-fusible-ops-userbuffers-") as temp_dir:
+        env["NVTE_TEST_RDZV_PATH"] = str(pathlib.Path(temp_dir) / "rdzv")
+        run_distributed(command, env=env)
 
 
 def main() -> None: