fix(distributed): propagate distributed_timeout_seconds to NCCL device groups (#45159)

Signed-off-by: jialoop-git <joane8913456@gmail.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

vllm/distributed/parallel_state.py

@@ -269,11 +269,17 @@ def _create_subgroups_split_group(
     must enter with the same ``split_ranks`` definition. Each rank receives
     the subgroup it belongs to.
     """
+    from vllm.distributed.utils import (
+        get_cpu_distributed_timeout_or_none,
+        get_distributed_timeout_or_none,
+    )
+
     device_backend_str = _device_backend_str(torch_distributed_backend)
     self_device_group = torch.distributed.split_group(
         split_ranks=group_ranks,
         group_desc=f"{group_name}:device",
         backend=device_backend_str,
+        timeout=get_distributed_timeout_or_none(),
     )
     # CPU subgroup: split_group requires the requested backend filter to
     # include the parent's default device type (= the device the parent PG
@@ -284,6 +290,7 @@ def _create_subgroups_split_group(
         split_ranks=group_ranks,
         group_desc=f"{group_name}:cpu",
         backend=f"cpu:gloo,{device_backend_str}",
+        timeout=get_cpu_distributed_timeout_or_none(),
     )
     return self_device_group, self_cpu_group
 
@@ -417,13 +424,19 @@ class GroupCoordinator:
                     self.rank_in_group = ranks.index(self.rank)
                     break
         else:
-            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()
+            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.
@@ -504,10 +517,16 @@ class GroupCoordinator:
         This is a collective call: every world rank must invoke it. Used where we
         want to issue ops that can run concurrently with ops on `device_group`.
         """
+        from vllm.distributed.utils import get_distributed_timeout_or_none
+
+        device_timeout = get_distributed_timeout_or_none()
         sibling: ProcessGroup | None = None
         for ranks in self.group_ranks:
             pg = torch.distributed.new_group(
-                ranks, backend=self.torch_distributed_backend, group_desc=group_desc
+                ranks,
+                backend=self.torch_distributed_backend,
+                group_desc=group_desc,
+                timeout=device_timeout,
             )
             if self.rank in ranks:
                 sibling = pg

vllm/distributed/utils.py

@@ -533,6 +533,16 @@ 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:
+    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,
@@ -616,6 +626,10 @@ def stateless_init_torch_distributed_process_group(
         gloo_timeout = get_cpu_distributed_timeout_or_none()
         if gloo_timeout is not None:
             timeout = gloo_timeout
+    else:
+        device_timeout = get_distributed_timeout_or_none()
+        if device_timeout is not None:
+            timeout = device_timeout
 
     if listen_socket is not None:
         store = create_tcp_store(