[Bugfix][Distributed] Pass --distributed-timeout-seconds to the NCCL subgroups

tests/distributed/test_group_coordinator_timeout.py

@@ -0,0 +1,73 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""The NCCL subgroups of a GroupCoordinator follow --distributed-timeout-seconds.
+
+CPU-only: ``torch.distributed.new_group`` is replaced by a recorder, so no
+process group is created. Only the arguments handed to it are checked.
+"""
+
+from datetime import timedelta
+from unittest.mock import MagicMock
+
+import pytest
+import torch
+
+from vllm.config import VllmConfig, set_current_vllm_config
+from vllm.distributed.parallel_state import GroupCoordinator
+
+
+@pytest.fixture
+def new_group_calls(monkeypatch):
+    calls: list[dict] = []
+
+    def record(ranks, **kwargs):
+        calls.append({"ranks": ranks, **kwargs})
+        return MagicMock(name=f"pg[{kwargs.get('backend')}]")
+
+    monkeypatch.setattr(torch.distributed, "new_group", record)
+    monkeypatch.setattr(torch.distributed, "get_rank", lambda: 0)
+    return calls
+
+
+def _build_coordinator() -> GroupCoordinator:
+    return GroupCoordinator(
+        group_ranks=[[0]],
+        local_rank=0,
+        torch_distributed_backend="nccl",
+        use_device_communicator=False,
+    )
+
+
+def test_device_group_follows_distributed_timeout(new_group_calls):
+    config = VllmConfig()
+    config.parallel_config.distributed_timeout_seconds = 3600
+    config.parallel_config.cpu_distributed_timeout_seconds = 120
+
+    with set_current_vllm_config(config):
+        _build_coordinator()
+
+    device_call, cpu_call = new_group_calls
+    assert device_call["backend"] == "nccl"
+    assert device_call["timeout"] == timedelta(seconds=3600)
+    assert cpu_call["backend"] == "gloo"
+    assert cpu_call["timeout"] == timedelta(seconds=120)
+
+
+def test_unset_timeout_keeps_pytorch_default(new_group_calls):
+    config = VllmConfig()
+    assert config.parallel_config.distributed_timeout_seconds is None
+
+    with set_current_vllm_config(config):
+        _build_coordinator()
+
+    device_call, cpu_call = new_group_calls
+    assert device_call["timeout"] is None
+    assert cpu_call["timeout"] is None
+
+
+def test_without_config_keeps_pytorch_default(new_group_calls):
+    _build_coordinator()
+
+    device_call, cpu_call = new_group_calls
+    assert device_call["timeout"] is None
+    assert cpu_call["timeout"] is None

vllm/distributed/parallel_state.py

@@ -486,13 +486,25 @@ def __init__(
         self_device_group = None
         self_cpu_group = None
 
-        from vllm.distributed.utils import get_cpu_distributed_timeout_or_none
+        from vllm.distributed.utils import (
+            get_cpu_distributed_timeout_or_none,
+            get_distributed_timeout_or_none,
+        )
 
         timeout = get_cpu_distributed_timeout_or_none()
+        # --distributed-timeout-seconds reaches init_process_group (the world
+        # group) and the gloo CPU groups, but the NCCL subgroups were created
+        # without it, so TP and PP silently kept PyTorch's 600 s default. On a
+        # cold boot under pipeline parallelism the first stage compiles Triton
+        # kernels for minutes while the next stage sits in its receive; the
+        # watchdog then kills the waiting rank and the boot dies at a timeout
+        # instead of an error. The subgroups follow the configured value;
+        # unset keeps PyTorch's default.
+        device_timeout = get_distributed_timeout_or_none()
 
         for ranks in group_ranks:
             device_group = torch.distributed.new_group(
-                ranks, backend=torch_distributed_backend
+                ranks, backend=torch_distributed_backend, timeout=device_timeout
             )
             # a group with `gloo` backend, to allow direct coordination between
             # processes through the CPU.

vllm/distributed/utils.py

@@ -516,6 +516,17 @@ def get_cpu_distributed_timeout_or_none() -> timedelta | None:
     return timedelta(seconds=timeout_seconds) if timeout_seconds is not None else None
 
 
+def get_distributed_timeout_or_none() -> timedelta | None:
+    """Timeout for the device (NCCL) subgroups, from the current vLLM config."""
+    from vllm.config import get_current_vllm_config_or_none
+
+    vllm_config = get_current_vllm_config_or_none()
+    if vllm_config is None:
+        return None
+    timeout_seconds = vllm_config.parallel_config.distributed_timeout_seconds
+    return timedelta(seconds=timeout_seconds) if timeout_seconds is not None else None
+
+
 def init_gloo_process_group(
     prefix_store: PrefixStore,
     group_rank: int,