Refactor the Cute-DSL AR fusion to support DeepseekV2 archs (GLM-5.3, etc.)

python/sglang/srt/arg_groups/fields/exec_.py

@@ -637,7 +637,7 @@ class ExecComm(msgspec.Struct):
         "Enforce disable FlashInfer allreduce fusion.",
     ] = False
     flashinfer_allreduce_fusion_backend: A[
-        Optional[Literal["auto", "trtllm", "mnnvl"]],
+        Optional[Literal["auto", "trtllm", "mnnvl", "cutedsl"]],
         Arg(
             help=(
                 "Enable FlashInfer allreduce fusion and choose backend. "
@@ -648,6 +648,9 @@ class ExecComm(msgspec.Struct):
                 "'trtllm': available on single-node systems only. "
                 "'mnnvl': available on SM90 single-node systems and SM100/SM103 "
                 "single-node or multi-node systems via MNNVL fabric. "
+                "'cutedsl': Blackwell-only bf16 MNNVL CuTe DSL backend; also "
+                "fuses the MoE finalize and the shared-expert add into the "
+                "collective when the MoE runner can defer them. "
                 "Fuses allreduce with Residual + RMSNorm for supported MoE models."
             ),
             resolvable=True,

python/sglang/srt/arg_groups/moe_hook.py

@@ -202,6 +202,7 @@ def validate_flashinfer_megamoe_model(server_args: Any) -> None:
         "DeepseekV32ForCausalLM",
         "DeepseekV4ForCausalLM",
         "Glm4MoeForCausalLM",
+        "GlmMoeDsaForCausalLM",
         "NemotronHForCausalLM",
         "NemotronHPuzzleForCausalLM",
         "Qwen2MoeForCausalLM",

python/sglang/srt/arg_groups/overrides.py

@@ -1025,20 +1025,6 @@ def _flashinfer_allreduce_fusion_auto_enable(view: Any) -> dict:
         and view.nnodes == 1
         and not view.disable_custom_all_reduce
     )
-    if envs.SGLANG_FLASHINFER_MNNVL_CUTEDSL_AR_FUSION.get() and model_arch in {
-        "Qwen3_5MoeForCausalLM",
-        "Qwen3_5MoeForConditionalGeneration",
-    }:
-        # The Qwen backend owns one workspace for ordinary AR and MoE finalize;
-        # do not allocate or fall back to the legacy TRTLLM/MNNVL workspace.
-        if view.flashinfer_allreduce_fusion_backend is not None:
-            logger.warning(
-                "SGLANG_FLASHINFER_MNNVL_CUTEDSL_AR_FUSION owns both Qwen3.5 "
-                "AllReduce fusion patterns; suppressing the separately configured "
-                "--flashinfer-allreduce-fusion-backend=%s",
-                view.flashinfer_allreduce_fusion_backend,
-            )
-        return {"flashinfer_allreduce_fusion_backend": None}
     if (
         view.flashinfer_allreduce_fusion_backend is None
         and model_arch in _FLASHINFER_ALLREDUCE_FUSION_ARCHS

python/sglang/srt/distributed/bootstrap.py

@@ -201,8 +201,12 @@ def _set_all_reduce_flags() -> None:
     set_custom_all_reduce(not get_exec().comm.disable_custom_all_reduce)
     set_mscclpp_all_reduce(get_exec().comm.enable_mscclpp)
     set_torch_symm_mem_all_reduce(get_exec().comm.enable_torch_symm_mem)
+    from sglang.srt.layers.flashinfer_comm_fusion import uses_cutedsl_ar_fusion
+
     set_flashinfer_allreduce_only(
         get_exec().comm.flashinfer_allreduce_fusion_backend is not None
+        # cutedsl has no legacy workspace for tagged groups to reduce over.
+        and not uses_cutedsl_ar_fusion()
     )
 
 

python/sglang/srt/environ.py

@@ -1245,6 +1245,9 @@ class Envs:
     SGLANG_NIXL_EP_NUM_MAX_DISPATCH_TOKENS_PER_RANK = EnvInt(128)
     SGLANG_PPLX_NUM_MAX_DISPATCH_TOKENS_PER_RANK = EnvInt(128)
     SGLANG_ENABLE_MOE_DEFERRED_FINALIZE = EnvBool(True)
+    # Deferring costs an HBM round trip of [M*top_k, hidden], so it pays only
+    # at small M; 0 disables. Measured neutral at 192 on GLM-5.2 H=6144 TP8.
+    SGLANG_MOE_DEFERRED_FINALIZE_MAX_TOKENS = EnvInt(192)
     # DeepSeek/GLM MoE (deepseek_v2.py): quantize the (dp-gathered) MoE input
     # to per-token-group-128 fp8 ONCE and feed both the fused shared-expert
     # GEMM (cutlass w8a8 linear) and the routed experts' triton fused runner,
@@ -1765,13 +1768,6 @@ class Envs:
 
     # Qwen3.5 and GDN
     SGLANG_ENABLE_GDN_DECODE_FUSED_PROJ_CONV = EnvBool(True)
-    SGLANG_TRACE_QWEN35_FINAL_NORM = EnvBool(False)
-    SGLANG_QWEN35_NATIVE_FINAL_NORM = EnvBool(False)
-    # One switch enables deferred MoE finalize and AR + residual + RMSNorm.
-    SGLANG_FLASHINFER_MNNVL_CUTEDSL_AR_FUSION = EnvBool(False)
-    # Distinct workspace configurations allowed in one process. Production
-    # uses one model/configuration per rank, so fail closed on accidental reuse.
-    SGLANG_FLASHINFER_MNNVL_CUTEDSL_AR_FUSION_MAX_INSTANCES = EnvInt(1)
 
     # ===================================================================
     # Plugin system
@@ -1918,6 +1914,18 @@ def _invert_bool(value: str) -> str:
         transform=_ms_to_s,
         note="Note the unit change: milliseconds -> seconds.",
     ),
+    # Migrated to a CLI flag: the backend argument now selects the fusion, so
+    # there is no env var left to forward the value to.
+    "SGLANG_FLASHINFER_MNNVL_CUTEDSL_AR_FUSION": _DeprecatedEnv(
+        note=(
+            "Pass --flashinfer-allreduce-fusion-backend cutedsl instead. "
+            "Without it an eligible model auto-enables the legacy mnnvl "
+            "backend rather than the CuTe DSL fusion."
+        )
+    ),
+    "SGLANG_FLASHINFER_MNNVL_CUTEDSL_AR_FUSION_MAX_INSTANCES": _DeprecatedEnv(
+        note="One workspace per process is now an invariant, not a limit."
+    ),
     # Removed without replacement.
     "SGLANG_ENABLE_CP_V2": _DeprecatedEnv(
         note="Strategy-based prefill context parallelism is now the only generic implementation."
@@ -1926,6 +1934,9 @@ def _invert_bool(value: str) -> str:
     # Superseded by the unified JIT per_token_group_quant, the default CUDA path.
     "SGLANG_OPT_USE_JIT_PER_TOKEN_GROUP_QUANT": _DeprecatedEnv(),
     "SGLANG_MASKED_GEMM_FAST_ACT": _DeprecatedEnv(),
+    # Qwen3.5 final-norm debug scaffolding, deleted with the fusion refactor.
+    "SGLANG_TRACE_QWEN35_FINAL_NORM": _DeprecatedEnv(),
+    "SGLANG_QWEN35_NATIVE_FINAL_NORM": _DeprecatedEnv(),
     # The unified free list is kept unsorted between flushes by design; the
     # sort-after-merge A/B knob never left its off default and is gone.
     "SGLANG_SORT_FREE_LIST_AFTER_MERGE": _DeprecatedEnv(),

python/sglang/srt/layers/communicator.py

@@ -56,7 +56,10 @@
     is_enable_moe_cp_allgather,
     moe_cp_all_gather_into_tensor,
 )
-from sglang.srt.layers.flashinfer_comm_fusion import is_flashinfer_allreduce_unavailable
+from sglang.srt.layers.flashinfer_comm_fusion import (
+    is_flashinfer_allreduce_unavailable,
+    uses_cutedsl_ar_fusion,
+)
 from sglang.srt.layers.moe import (
     can_merge_post_experts_all_reduce,
     deferred_post_experts_all_reduce,
@@ -196,6 +199,8 @@ def apply_flashinfer_allreduce_fusion(batch_size: int):
         and _is_flashinfer_available
         and not is_dp_attention_enabled()
         and get_exec().comm.flashinfer_allreduce_fusion_backend is not None
+        # cutedsl runs its own fused path from the fusion communicator.
+        and not uses_cutedsl_ar_fusion()
         and not is_flashinfer_allreduce_unavailable()
         # Symbolic size checks stay last: under Dynamo tracing they guard on
         # the dynamic token dim, so statically-off configs must short-circuit
@@ -840,6 +845,10 @@ def prepare_attn(
                             post_residual_addition,
                         )
 
+        return self._finish_prepare_attn(hidden_states, residual, forward_batch)
+
+    def _finish_prepare_attn(self, hidden_states, residual, forward_batch):
+        """Tail every prepare_attn path must run, or ``attn_inputs`` is unset."""
         hidden_states = self._communicate_simple_fn(
             hidden_states=hidden_states,
             forward_batch=forward_batch,
@@ -926,6 +935,12 @@ def should_use_reduce_scatter(self, forward_batch: ForwardBatch):
             return True
         return False
 
+    def should_defer_moe_finalize(
+        self, forward_batch: ForwardBatch, m: int | None = None
+    ) -> bool:
+        """Whether the MoE may hand an unfinalized output to the next layer."""
+        return False
+
     # NOTE: This function will cause torch recompilation
     def should_fuse_mlp_allreduce_with_next_layer(
         self, forward_batch: ForwardBatch

python/sglang/srt/layers/flashinfer_comm_fusion.py

@@ -47,6 +47,14 @@ def _resolve_backend(backend: str, is_multi_node: bool = False) -> str:
             "FlashInfer allreduce fusion requires SM90 or SM10X NVIDIA GPUs."
         )
 
+    if backend == "cutedsl":
+        if not get_platform().is_sm100:
+            raise ValueError(
+                "FlashInfer allreduce fusion cutedsl backend requires a "
+                "Blackwell system."
+            )
+        return backend
+
     if backend == "auto":
         if is_multi_node:
             if get_platform().is_sm100:
@@ -72,6 +80,11 @@ def _resolve_backend(backend: str, is_multi_node: bool = False) -> str:
     return backend
 
 
+def uses_cutedsl_ar_fusion() -> bool:
+    """Selected CuTe DSL owns both patterns, so the legacy workspace stands down."""
+    return get_exec().comm.flashinfer_allreduce_fusion_backend == "cutedsl"
+
+
 def resolve_flashinfer_allreduce_fusion_backend() -> Optional[str]:
     """The fusion backend for this process, or None when fusion is off.
 
@@ -713,7 +726,7 @@ def ensure_workspace_initialized(
     use_attn_tp_group: bool = True,
 ):
     """Ensure workspace is initialized."""
-    if _flashinfer_allreduce_unavailable:
+    if _flashinfer_allreduce_unavailable or uses_cutedsl_ar_fusion():
         return False
 
     if not is_flashinfer_available() or _flashinfer_comm is None: