[Core] Release NCCL communicator memory in sleep mode

vllm/distributed/device_communicators/base_device_communicator.py

@@ -321,6 +321,12 @@ def broadcast(self, tensor: torch.Tensor, src: int = 0) -> torch.Tensor:
     def destroy(self):
         pass
 
+    def suspend(self) -> None:
+        """Release reclaimable communicator memory (default: no-op)."""
+
+    def resume(self) -> None:
+        """Restore memory released by ``suspend`` (default: no-op)."""
+
     def prepare_communication_buffer_for_model(self, model: torch.nn.Module) -> None:
         """
         Prepare the communication buffer for the model.

vllm/distributed/device_communicators/cuda_communicator.py

@@ -516,6 +516,14 @@ def destroy(self):
             self.all2all_manager.destroy()
             self.all2all_manager = None  # type: ignore[assignment]
 
+    def suspend(self) -> None:
+        if self.pynccl_comm is not None:
+            self.pynccl_comm.suspend()
+
+    def resume(self) -> None:
+        if self.pynccl_comm is not None:
+            self.pynccl_comm.resume()
+
     def all_gatherv(
         self,
         input_: torch.Tensor | list[torch.Tensor],

vllm/distributed/device_communicators/pynccl.py

@@ -27,6 +27,11 @@
 
 _NCCL_SYMM_OPS_REGISTERED = False
 
+# ncclCommSuspend flag: release dynamic GPU memory, keep topology/connection.
+_NCCL_SUSPEND_MEM = 0x01
+# ncclCommSuspend / ncclCommResume were introduced in NCCL 2.29.7 (22907).
+_NCCL_SUSPEND_MIN_VERSION = 22907
+
 
 def register_nccl_symmetric_ops(pynccl_comm):
     from vllm.distributed.device_communicators.pynccl_allocator import (
@@ -419,6 +424,22 @@ def register_comm_window_raw(self, ptr: int, size: int):
     def deregister_comm_window(self, window):
         return self.nccl.ncclCommWindowDeregister(self.comm, window)
 
+    def suspend(self):
+        """Release dynamic GPU memory, keeping topology/connection state.
+
+        Collective across the group's ranks (ncclCommSuspend has an internal
+        cross-rank barrier). No-op on NCCL < 2.29.7.
+        """
+        if self.disabled or self.nccl_version < _NCCL_SUSPEND_MIN_VERSION:
+            return
+        self.nccl.ncclCommSuspend(self.comm, _NCCL_SUSPEND_MEM)
+
+    def resume(self):
+        """Restore a suspended communicator (collective). No-op on NCCL < 2.29.7."""
+        if self.disabled or self.nccl_version < _NCCL_SUSPEND_MIN_VERSION:
+            return
+        self.nccl.ncclCommResume(self.comm)
+
     def batch_isend_irecv(self, p2p_ops: list, stream=None):
         if self.disabled:
             return

vllm/distributed/device_communicators/pynccl_wrapper.py

@@ -296,6 +296,10 @@ class NCCLLibrary:
         # shutdown when peer ranks may already be gone.
         # ncclResult_t  ncclCommAbort(ncclComm_t comm);
         Function("ncclCommAbort", ncclResult_t, [ncclComm_t]),
+        # ncclResult_t ncclCommSuspend(ncclComm_t comm, int flags);
+        Function("ncclCommSuspend", ncclResult_t, [ncclComm_t, ctypes.c_int]),
+        # ncclResult_t ncclCommResume(ncclComm_t comm);
+        Function("ncclCommResume", ncclResult_t, [ncclComm_t]),
         # ncclResult_t ncclGroupStart();
         Function("ncclGroupStart", ncclResult_t, []),
         # ncclResult_t ncclGroupEnd();
@@ -358,6 +362,16 @@ def __init__(self, so_file: str | None = None):
                     f.argtypes = func.argtypes
                     _funcs[func.name] = f
                 except AttributeError:
+                    if func.name in ("ncclCommSuspend", "ncclCommResume"):
+                        # Only present in NCCL >= 2.29.7. Skip on older NCCL;
+                        # sleep-mode NCCL memory release degrades to a no-op.
+                        logger.warning_once(
+                            "NCCL communicator suspend/resume not found in %s "
+                            "(needs NCCL >= 2.29.7); sleep-mode NCCL memory "
+                            "release is disabled.",
+                            so_file,
+                        )
+                        continue
                     if func.name in [
                         "ncclCommWindowRegister",
                         "ncclCommWindowDeregister",
@@ -557,6 +571,12 @@ def ncclCommDestroy(self, comm: ncclComm_t) -> None:
     def ncclCommAbort(self, comm: ncclComm_t) -> None:
         self.NCCL_CHECK(self._funcs["ncclCommAbort"](comm))
 
+    def ncclCommSuspend(self, comm: ncclComm_t, flags: int) -> None:
+        self.NCCL_CHECK(self._funcs["ncclCommSuspend"](comm, flags))
+
+    def ncclCommResume(self, comm: ncclComm_t) -> None:
+        self.NCCL_CHECK(self._funcs["ncclCommResume"](comm))
+
     def ncclGroupStart(self) -> None:
         self.NCCL_CHECK(self._funcs["ncclGroupStart"]())
 

vllm/distributed/parallel_state.py

@@ -127,6 +127,56 @@ def _register_group(group: "GroupCoordinator") -> None:
     _groups[group.unique_name] = weakref.ref(group)
 
 
+def _apply_to_device_comms(label: str, action: Callable[[Any], None]) -> None:
+    """Apply ``action`` to every group's device communicator, collectively.
+
+    Walks the registered parallel groups and skips those without a device
+    communicator (absent at ``world_size == 1``). Each communicator's
+    ``suspend``/``resume`` is a no-op unless it holds releasable device memory
+    (see ``DeviceCommunicatorBase``). Collective across ranks and synchronous
+    on return, so no extra sync is needed.
+    """
+    comms = []
+    for group_ref in _groups.values():
+        group = group_ref()
+        if group is None:
+            continue
+        dc = group.device_communicator
+        if dc is None:
+            continue
+        comms.append(dc)
+    if not comms:
+        return
+
+    free_before = torch.accelerator.get_memory_info()[0]
+    for dc in comms:
+        action(dc)
+    delta = torch.accelerator.get_memory_info()[0] - free_before
+    direction = "freed" if delta > 0 else "allocated"
+    logger.info(
+        "device-comm %s: %d comms, %.1f MiB %s",
+        label,
+        len(comms),
+        abs(delta) / 1024**2,
+        direction,
+    )
+
+
+def suspend_device_comms() -> None:
+    """Release idle device communicator memory on every group (collective).
+
+    Must run on every rank with communicators idle. Communicator suspend hooks
+    (e.g. NCCL's ncclCommSuspend) have internal cross-rank barriers; a no-op
+    where unsupported (older NCCL, world_size 1, non-CUDA).
+    """
+    _apply_to_device_comms("suspend", lambda c: c.suspend())
+
+
+def resume_device_comms() -> None:
+    """Restore all suspended device communicators before reuse (collective)."""
+    _apply_to_device_comms("resume", lambda c: c.resume())
+
+
 def all_reduce(tensor: torch.Tensor, group_name: str) -> torch.Tensor:
     assert group_name in _groups, f"Group {group_name} is not found."
     group = _groups[group_name]()

vllm/v1/worker/gpu_worker.py

@@ -43,6 +43,8 @@
     Handle,
     get_pp_group,
     get_tp_group,
+    resume_device_comms,
+    suspend_device_comms,
 )
 from vllm.distributed.weight_transfer import (
     WeightTransferEngine,
@@ -197,6 +199,11 @@ def sleep(self, level: int = 1) -> None:
 
         self._get_sleep_mode_backend().suspend(level)
 
+        # Release idle device communicator memory (e.g. NCCL >= 2.29.7).
+        # Collective across ranks; no-op where unsupported. Done after the
+        # cumem unmap so the reported freed bytes include the comm release.
+        suspend_device_comms()
+
         torch.accelerator.synchronize()
         deadline = time.monotonic() + (5.0 if current_platform.is_rocm() else 0)
         while True:
@@ -217,6 +224,10 @@ def sleep(self, level: int = 1) -> None:
     def wake_up(self, tags: list[str] | None = None) -> None:
         self._get_sleep_mode_backend().resume(tags)
 
+        # Restore device communicator memory before any collective runs again.
+        # Collective across ranks; no-op where unsupported.
+        resume_device_comms()
+
         # Restore the buffers after level 2 sleep
         if len(self._sleep_saved_buffers):
             model = self.model_runner.model