From a4f019fa25daf422d36804fac1eac1ebebf21d9f Mon Sep 17 00:00:00 2001 From: jialoop-git Date: Tue, 7 Jul 2026 10:52:51 +0800 Subject: [PATCH] fix(distributed): propagate distributed_timeout_seconds to NCCL device groups (#45159) Signed-off-by: jialoop-git Co-authored-by: Claude Opus 4.8 (1M context) --- vllm/distributed/parallel_state.py | 25 ++++++++++++++++++++++--- vllm/distributed/utils.py | 14 ++++++++++++++ 2 files changed, 36 insertions(+), 3 deletions(-) diff --git a/vllm/distributed/parallel_state.py b/vllm/distributed/parallel_state.py index 11b9e24e86..162ed03d23 100644 --- a/vllm/distributed/parallel_state.py +++ b/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 diff --git a/vllm/distributed/utils.py b/vllm/distributed/utils.py index ef3c11ff64..eec9890f02 100644 --- a/vllm/distributed/utils.py +++ b/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(