From 3e255fd49371a810e4fc900b95c7b60f6876ed38 Mon Sep 17 00:00:00 2001
From: Yujun Dong <118669451+cs-cat@users.noreply.github.com>
Date: Fri, 29 May 2026 07:28:33 +0800
Subject: [PATCH] fix: Graceful fallback to CustomAllReduce when full_nvlink is
 not True (#25650)

---
 .../device_communicators/custom_all_reduce.py | 15 +++++++---
 .../custom_all_reduce_v2.py                   | 29 +++++++++++++------
 .../sglang/srt/distributed/parallel_state.py  |  5 +++-
 3 files changed, 35 insertions(+), 14 deletions(-)

diff --git a/python/sglang/srt/distributed/device_communicators/custom_all_reduce.py b/python/sglang/srt/distributed/device_communicators/custom_all_reduce.py
index ac3afb8ad..09b981f9b 100644
--- a/python/sglang/srt/distributed/device_communicators/custom_all_reduce.py
+++ b/python/sglang/srt/distributed/device_communicators/custom_all_reduce.py
@@ -338,7 +338,10 @@ class CustomAllreduce:
         self.close()
 
 
-def dispatch_custom_allreduce():
+def dispatch_custom_allreduce(
+    group: ProcessGroup,
+    device: torch.device,
+):
     """Return the CustomAllreduce class to use (aiter on ROCm if enabled).
 
     On AMD with 1-stage AR enabled, use sglang's CustomAllreduce.
@@ -350,10 +353,14 @@ def dispatch_custom_allreduce():
     ``nnodes > 1`` since custom AR is intra-node only.
     """
     if _is_cuda and envs.SGLANG_OPT_USE_CUSTOM_ALL_REDUCE_V2.get():
-        from .custom_all_reduce_v2 import CustomAllReduceV2
+        from .custom_all_reduce_v2 import (
+            CustomAllReduceV2,
+            can_use_custom_all_reduce_v2,
+        )
 
-        logger.debug("[AR] Using CustomAllReduceV2 (JIT-compiled)")
-        return CustomAllReduceV2
+        if can_use_custom_all_reduce_v2(group=group, device=device):
+            logger.debug("[AR] Using CustomAllReduceV2 (JIT-compiled)")
+            return CustomAllReduceV2
 
     if _is_cuda or _is_musa:
         return CustomAllreduce
diff --git a/python/sglang/srt/distributed/device_communicators/custom_all_reduce_v2.py b/python/sglang/srt/distributed/device_communicators/custom_all_reduce_v2.py
index 01f162ff7..454090547 100644
--- a/python/sglang/srt/distributed/device_communicators/custom_all_reduce_v2.py
+++ b/python/sglang/srt/distributed/device_communicators/custom_all_reduce_v2.py
@@ -38,15 +38,9 @@ class CustomAllReduceV2:
         max_pull_blocks: Optional[int] = None,
         max_push_blocks: Optional[int] = None,
     ) -> None:
-        _init_config()
+        _maybe_init_config()
         self.disabled = True
-        full_nvlink = can_use_custom_all_reduce_with_nvlink(
-            group=group,
-            device=device,
-            supported_world_size=list(THRESHOLD_2_SHOT_MAP.keys()),
-            cls_name="CustomAllReduceV2",
-        )
-        if full_nvlink != True:
+        if not can_use_custom_all_reduce_v2(group=group, device=device):
             return
 
         self.group = group
@@ -172,8 +166,10 @@ class CustomAllReduceV2:
         self.close()
 
 
-def _init_config():
+def _maybe_init_config():
     global THRESHOLD_2_SHOT_MAP
+    if THRESHOLD_2_SHOT_MAP:
+        return
     KB, MB = 1024, 1024 * 1024
 
     if is_sm100_supported():
@@ -201,4 +197,19 @@ def _init_config():
     # TODO: tune on more GPUs, e.g A100
 
 
+def can_use_custom_all_reduce_v2(
+    group: ProcessGroup,
+    device: torch.device,
+) -> bool:
+    # call _maybe_init_config() to ensure THRESHOLD_2_SHOT_MAP is initialized, since can_use_custom_all_reduce_v2 can be called before CustomAllReduceV2 is initialized
+    _maybe_init_config()
+    full_nvlink = can_use_custom_all_reduce_with_nvlink(
+        group=group,
+        device=device,
+        supported_world_size=list(THRESHOLD_2_SHOT_MAP.keys()),
+        cls_name="CustomAllReduceV2",
+    )
+    return full_nvlink is True
+
+
 THRESHOLD_2_SHOT_MAP: Dict[int, ModeConfig] = {}
diff --git a/python/sglang/srt/distributed/parallel_state.py b/python/sglang/srt/distributed/parallel_state.py
index 863f5f1a7..e2362ab5b 100644
--- a/python/sglang/srt/distributed/parallel_state.py
+++ b/python/sglang/srt/distributed/parallel_state.py
@@ -384,7 +384,10 @@ class GroupCoordinator:
         if use_custom_allreduce and self.world_size > 1:
             # Initialize a custom fast all-reduce implementation.
             try:
-                CAClass = dispatch_custom_allreduce()
+                CAClass = dispatch_custom_allreduce(
+                    group=self.cpu_group,
+                    device=self.device,
+                )
                 self.ca_comm = CAClass(
                     group=self.cpu_group,
                     device=self.device,