refactor(moe): centralize post-experts all-reduce skip predicate (#23748)

Co-authored-by: Byron Hsu <byron@periodiclabs.ai>
Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Byron Hsu
2026-04-26 20:29:59 -07:00
committed by GitHub
co-authored by Byron Hsu Claude Opus 4.7
parent da175b964d
commit 85376a6119
17 changed files with 134 additions and 132 deletions
+2
View File
@@ -10,6 +10,7 @@ from sglang.srt.layers.moe.utils import (
get_tbo_token_distribution_threshold, get_tbo_token_distribution_threshold,
initialize_moe_config, initialize_moe_config,
is_tbo_enabled, is_tbo_enabled,
should_skip_post_experts_all_reduce,
should_use_dp_reduce_scatterv, should_use_dp_reduce_scatterv,
should_use_flashinfer_cutlass_moe_fp4_allgather, should_use_flashinfer_cutlass_moe_fp4_allgather,
) )
@@ -24,6 +25,7 @@ __all__ = [
"get_moe_a2a_backend", "get_moe_a2a_backend",
"get_moe_runner_backend", "get_moe_runner_backend",
"get_deepep_mode", "get_deepep_mode",
"should_skip_post_experts_all_reduce",
"should_use_dp_reduce_scatterv", "should_use_dp_reduce_scatterv",
"should_use_flashinfer_cutlass_moe_fp4_allgather", "should_use_flashinfer_cutlass_moe_fp4_allgather",
"is_tbo_enabled", "is_tbo_enabled",
+33
View File
@@ -346,6 +346,39 @@ def should_use_dp_reduce_scatterv():
) )
def should_skip_post_experts_all_reduce(
*,
is_tp_path: bool,
use_reduce_scatter: bool = False,
should_allreduce_fusion: bool = False,
) -> bool:
"""Whether to skip the post-experts all-reduce (EP or TP) because a
downstream component will fuse, replace, or absorb it.
Skip reasons, in order:
- ``should_allreduce_fusion``: LayerCommunicator will fuse the all-reduce
with the next layer's residual all-reduce.
- ``use_reduce_scatter``: LayerCommunicator's post-attention scatter will
do reduce-scatter, which would double-reduce on top of an all-reduce.
- ``should_use_dp_reduce_scatterv()``: the standard dispatcher's combine
path replaces the all-reduce with a reduce-scatterv.
- ``should_use_flashinfer_cutlass_moe_fp4_allgather()`` (TP path only):
the flashinfer cutlass FP4 kernel performs an all-gather that absorbs
the post-experts TP all-reduce. Not relevant to the EP all-reduce.
The first two args are layer-context flags from ``LayerCommunicator`` and
default to ``False`` for models that don't use it. Pass ``is_tp_path=True``
for the post-experts TP all-reduce, ``False`` for the EP all-reduce.
"""
if should_allreduce_fusion or use_reduce_scatter:
return True
if should_use_dp_reduce_scatterv():
return True
if is_tp_path and should_use_flashinfer_cutlass_moe_fp4_allgather():
return True
return False
@contextmanager @contextmanager
def speculative_moe_backend_context(): def speculative_moe_backend_context():
""" """
+5 -8
View File
@@ -58,8 +58,7 @@ from sglang.srt.layers.logits_processor import LogitsProcessor
from sglang.srt.layers.moe import ( from sglang.srt.layers.moe import (
get_deepep_mode, get_deepep_mode,
get_moe_a2a_backend, get_moe_a2a_backend,
should_use_dp_reduce_scatterv, should_skip_post_experts_all_reduce,
should_use_flashinfer_cutlass_moe_fp4_allgather,
) )
from sglang.srt.layers.moe.ep_moe.layer import get_moe_impl_class from sglang.srt.layers.moe.ep_moe.layer import get_moe_impl_class
from sglang.srt.layers.moe.fused_moe_triton.layer import FusedMoE from sglang.srt.layers.moe.fused_moe_triton.layer import FusedMoE
@@ -382,12 +381,10 @@ class BailingMoESparseMoeBlock(nn.Module):
if self.num_shared_experts > 0: if self.num_shared_experts > 0:
final_hidden_states = final_hidden_states + shared_output final_hidden_states = final_hidden_states + shared_output
if ( if self.tp_size > 1 and not should_skip_post_experts_all_reduce(
self.tp_size > 1 is_tp_path=True,
and not should_allreduce_fusion use_reduce_scatter=use_reduce_scatter,
and not use_reduce_scatter should_allreduce_fusion=should_allreduce_fusion,
and not should_use_flashinfer_cutlass_moe_fp4_allgather()
and not should_use_dp_reduce_scatterv()
): ):
final_hidden_states = tensor_model_parallel_all_reduce(final_hidden_states) final_hidden_states = tensor_model_parallel_all_reduce(final_hidden_states)
return final_hidden_states.view(num_tokens, hidden_size) return final_hidden_states.view(num_tokens, hidden_size)
@@ -34,7 +34,7 @@ from sglang.srt.layers.linear import (
RowParallelLinear, RowParallelLinear,
) )
from sglang.srt.layers.logits_processor import LogitsProcessor from sglang.srt.layers.logits_processor import LogitsProcessor
from sglang.srt.layers.moe import should_use_dp_reduce_scatterv from sglang.srt.layers.moe import should_skip_post_experts_all_reduce
from sglang.srt.layers.moe.ep_moe.layer import DeepEPMoE, get_moe_impl_class from sglang.srt.layers.moe.ep_moe.layer import DeepEPMoE, get_moe_impl_class
from sglang.srt.layers.moe.fused_moe_triton.layer import FusedMoE from sglang.srt.layers.moe.fused_moe_triton.layer import FusedMoE
from sglang.srt.layers.moe.topk import TopK from sglang.srt.layers.moe.topk import TopK
@@ -348,11 +348,10 @@ class BailingMoE(nn.Module):
if self.num_shared_experts > 0: if self.num_shared_experts > 0:
final_hidden_states = final_hidden_states + shared_output final_hidden_states = final_hidden_states + shared_output
if ( if self.tp_size > 1 and not should_skip_post_experts_all_reduce(
self.tp_size > 1 is_tp_path=True,
and not use_reduce_scatter use_reduce_scatter=use_reduce_scatter,
and not should_allreduce_fusion should_allreduce_fusion=should_allreduce_fusion,
and not should_use_dp_reduce_scatterv()
): ):
final_hidden_states = tensor_model_parallel_all_reduce(final_hidden_states) final_hidden_states = tensor_model_parallel_all_reduce(final_hidden_states)
return final_hidden_states return final_hidden_states
+9 -13
View File
@@ -85,7 +85,7 @@ from sglang.srt.layers.logits_processor import LogitsProcessor
from sglang.srt.layers.moe import ( from sglang.srt.layers.moe import (
get_moe_a2a_backend, get_moe_a2a_backend,
get_moe_runner_backend, get_moe_runner_backend,
should_use_dp_reduce_scatterv, should_skip_post_experts_all_reduce,
should_use_flashinfer_cutlass_moe_fp4_allgather, should_use_flashinfer_cutlass_moe_fp4_allgather,
) )
from sglang.srt.layers.moe.ep_moe.layer import get_moe_impl_class from sglang.srt.layers.moe.ep_moe.layer import get_moe_impl_class
@@ -651,12 +651,10 @@ class DeepseekV2MoE(nn.Module):
current_stream.wait_stream(self.alt_stream) current_stream.wait_stream(self.alt_stream)
final_hidden_states += shared_output final_hidden_states += shared_output
if ( if self.tp_size > 1 and not should_skip_post_experts_all_reduce(
self.tp_size > 1 is_tp_path=True,
and not should_allreduce_fusion use_reduce_scatter=use_reduce_scatter,
and not use_reduce_scatter should_allreduce_fusion=should_allreduce_fusion,
and not should_use_flashinfer_cutlass_moe_fp4_allgather()
and not should_use_dp_reduce_scatterv()
): ):
final_hidden_states = tensor_model_parallel_all_reduce(final_hidden_states) final_hidden_states = tensor_model_parallel_all_reduce(final_hidden_states)
return final_hidden_states return final_hidden_states
@@ -741,12 +739,10 @@ class DeepseekV2MoE(nn.Module):
final_hidden_states *= self.routed_scaling_factor final_hidden_states *= self.routed_scaling_factor
if shared_output is not None: if shared_output is not None:
final_hidden_states += shared_output final_hidden_states += shared_output
if ( if self.tp_size > 1 and not should_skip_post_experts_all_reduce(
self.tp_size > 1 is_tp_path=True,
and not should_allreduce_fusion use_reduce_scatter=use_reduce_scatter,
and not use_reduce_scatter should_allreduce_fusion=should_allreduce_fusion,
and not should_use_flashinfer_cutlass_moe_fp4_allgather()
and not should_use_dp_reduce_scatterv()
): ):
final_hidden_states = tensor_model_parallel_all_reduce(final_hidden_states) final_hidden_states = tensor_model_parallel_all_reduce(final_hidden_states)
return final_hidden_states return final_hidden_states
+7 -5
View File
@@ -47,7 +47,10 @@ from sglang.srt.layers.linear import (
RowParallelLinear, RowParallelLinear,
) )
from sglang.srt.layers.logits_processor import LogitsProcessor, LogitsProcessorOutput from sglang.srt.layers.logits_processor import LogitsProcessor, LogitsProcessorOutput
from sglang.srt.layers.moe import get_moe_a2a_backend, should_use_dp_reduce_scatterv from sglang.srt.layers.moe import (
get_moe_a2a_backend,
should_skip_post_experts_all_reduce,
)
from sglang.srt.layers.moe.ep_moe.layer import get_moe_impl_class from sglang.srt.layers.moe.ep_moe.layer import get_moe_impl_class
from sglang.srt.layers.moe.fused_moe_triton import FusedMoE from sglang.srt.layers.moe.fused_moe_triton import FusedMoE
from sglang.srt.layers.moe.topk import TopK from sglang.srt.layers.moe.topk import TopK
@@ -300,10 +303,9 @@ class ExaoneMoESparseMoEBlock(nn.Module):
if shared_output is not None: if shared_output is not None:
final_hidden_states = final_hidden_states + shared_output final_hidden_states = final_hidden_states + shared_output
if ( if self.tp_size > 1 and not should_skip_post_experts_all_reduce(
self.tp_size > 1 is_tp_path=True,
and not use_reduce_scatter use_reduce_scatter=use_reduce_scatter,
and not should_use_dp_reduce_scatterv()
): ):
final_hidden_states = tensor_model_parallel_all_reduce(final_hidden_states) final_hidden_states = tensor_model_parallel_all_reduce(final_hidden_states)
+9 -13
View File
@@ -61,7 +61,7 @@ from sglang.srt.layers.linear import (
from sglang.srt.layers.logits_processor import LogitsProcessor from sglang.srt.layers.logits_processor import LogitsProcessor
from sglang.srt.layers.moe import ( from sglang.srt.layers.moe import (
get_moe_a2a_backend, get_moe_a2a_backend,
should_use_dp_reduce_scatterv, should_skip_post_experts_all_reduce,
should_use_flashinfer_cutlass_moe_fp4_allgather, should_use_flashinfer_cutlass_moe_fp4_allgather,
) )
from sglang.srt.layers.moe.ep_moe.layer import get_moe_impl_class from sglang.srt.layers.moe.ep_moe.layer import get_moe_impl_class
@@ -594,12 +594,10 @@ class Glm4MoeSparseMoeBlock(nn.Module):
current_stream.wait_stream(self.alt_stream) current_stream.wait_stream(self.alt_stream)
final_hidden_states += shared_output final_hidden_states += shared_output
if ( if self.tp_size > 1 and not should_skip_post_experts_all_reduce(
self.tp_size > 1 is_tp_path=True,
and not should_allreduce_fusion use_reduce_scatter=use_reduce_scatter,
and not use_reduce_scatter should_allreduce_fusion=should_allreduce_fusion,
and not should_use_flashinfer_cutlass_moe_fp4_allgather()
and not should_use_dp_reduce_scatterv()
): ):
final_hidden_states = tensor_model_parallel_all_reduce(final_hidden_states) final_hidden_states = tensor_model_parallel_all_reduce(final_hidden_states)
return final_hidden_states return final_hidden_states
@@ -629,12 +627,10 @@ class Glm4MoeSparseMoeBlock(nn.Module):
final_hidden_states_out = torch.empty_like(final_hidden_states) final_hidden_states_out = torch.empty_like(final_hidden_states)
torch.add(final_hidden_states, shared_output, out=final_hidden_states_out) torch.add(final_hidden_states, shared_output, out=final_hidden_states_out)
final_hidden_states = final_hidden_states_out final_hidden_states = final_hidden_states_out
if ( if self.tp_size > 1 and not should_skip_post_experts_all_reduce(
self.tp_size > 1 is_tp_path=True,
and not should_allreduce_fusion use_reduce_scatter=use_reduce_scatter,
and not use_reduce_scatter should_allreduce_fusion=should_allreduce_fusion,
and not should_use_flashinfer_cutlass_moe_fp4_allgather()
and not should_use_dp_reduce_scatterv()
): ):
final_hidden_states = tensor_model_parallel_all_reduce(final_hidden_states) final_hidden_states = tensor_model_parallel_all_reduce(final_hidden_states)
return final_hidden_states return final_hidden_states
+13 -7
View File
@@ -34,7 +34,7 @@ from sglang.srt.layers.linear import (
RowParallelLinear, RowParallelLinear,
) )
from sglang.srt.layers.logits_processor import LogitsProcessor from sglang.srt.layers.logits_processor import LogitsProcessor
from sglang.srt.layers.moe import should_use_dp_reduce_scatterv from sglang.srt.layers.moe import should_skip_post_experts_all_reduce
from sglang.srt.layers.moe.fused_moe_triton.layer import FusedMoE from sglang.srt.layers.moe.fused_moe_triton.layer import FusedMoE
from sglang.srt.layers.moe.topk import TopK from sglang.srt.layers.moe.topk import TopK
from sglang.srt.layers.quantization.base_config import QuantizationConfig from sglang.srt.layers.quantization.base_config import QuantizationConfig
@@ -192,11 +192,14 @@ class HYV3MoEFused(nn.Module):
hidden_states=hidden_states, topk_output=topk_output hidden_states=hidden_states, topk_output=topk_output
) )
skip_post_reduce = should_use_dp_reduce_scatterv() if self.ep_size > 1 and not should_skip_post_experts_all_reduce(
if self.ep_size > 1 and not skip_post_reduce: is_tp_path=False,
):
final_hidden_states = moe_expert_parallel_all_reduce(final_hidden_states) final_hidden_states = moe_expert_parallel_all_reduce(final_hidden_states)
if self.tp_size > 1 and not skip_post_reduce: if self.tp_size > 1 and not should_skip_post_experts_all_reduce(
is_tp_path=True,
):
final_hidden_states = moe_tensor_model_parallel_all_reduce( final_hidden_states = moe_tensor_model_parallel_all_reduce(
final_hidden_states final_hidden_states
) )
@@ -224,11 +227,14 @@ class HYV3MoEFused(nn.Module):
current_stream.wait_stream(self.alt_stream) current_stream.wait_stream(self.alt_stream)
final_hidden_states = final_hidden_states + shared_output final_hidden_states = final_hidden_states + shared_output
skip_post_reduce = should_use_dp_reduce_scatterv() if self.ep_size > 1 and not should_skip_post_experts_all_reduce(
if self.ep_size > 1 and not skip_post_reduce: is_tp_path=False,
):
final_hidden_states = moe_expert_parallel_all_reduce(final_hidden_states) final_hidden_states = moe_expert_parallel_all_reduce(final_hidden_states)
if self.tp_size > 1 and not skip_post_reduce: if self.tp_size > 1 and not should_skip_post_experts_all_reduce(
is_tp_path=True,
):
final_hidden_states = moe_tensor_model_parallel_all_reduce( final_hidden_states = moe_tensor_model_parallel_all_reduce(
final_hidden_states final_hidden_states
) )
+4 -5
View File
@@ -58,7 +58,7 @@ from sglang.srt.layers.logits_processor import LogitsProcessor
from sglang.srt.layers.moe import ( from sglang.srt.layers.moe import (
get_deepep_mode, get_deepep_mode,
get_moe_a2a_backend, get_moe_a2a_backend,
should_use_dp_reduce_scatterv, should_skip_post_experts_all_reduce,
) )
from sglang.srt.layers.moe.ep_moe.layer import get_moe_impl_class from sglang.srt.layers.moe.ep_moe.layer import get_moe_impl_class
from sglang.srt.layers.moe.fused_moe_triton.layer import FusedMoE from sglang.srt.layers.moe.fused_moe_triton.layer import FusedMoE
@@ -383,10 +383,9 @@ class LLaDA2MoeSparseMoeBlock(nn.Module):
if self.num_shared_experts > 0: if self.num_shared_experts > 0:
final_hidden_states = final_hidden_states + shared_output final_hidden_states = final_hidden_states + shared_output
if ( if self.tp_size > 1 and not should_skip_post_experts_all_reduce(
self.tp_size > 1 is_tp_path=True,
and not use_reduce_scatter use_reduce_scatter=use_reduce_scatter,
and not should_use_dp_reduce_scatterv()
): ):
final_hidden_states = tensor_model_parallel_all_reduce(final_hidden_states) final_hidden_states = tensor_model_parallel_all_reduce(final_hidden_states)
return final_hidden_states.view(num_tokens, hidden_size) return final_hidden_states.view(num_tokens, hidden_size)
+4 -5
View File
@@ -39,7 +39,7 @@ from sglang.srt.layers.linear import (
ReplicatedLinear, ReplicatedLinear,
RowParallelLinear, RowParallelLinear,
) )
from sglang.srt.layers.moe import should_use_dp_reduce_scatterv from sglang.srt.layers.moe import should_skip_post_experts_all_reduce
from sglang.srt.layers.moe.fused_moe_triton import FusedMoE from sglang.srt.layers.moe.fused_moe_triton import FusedMoE
from sglang.srt.layers.moe.topk import TopK from sglang.srt.layers.moe.topk import TopK
from sglang.srt.layers.quantization.base_config import QuantizationConfig from sglang.srt.layers.quantization.base_config import QuantizationConfig
@@ -146,10 +146,9 @@ class Llama4MoE(nn.Module):
out_aD = routed_out + shared_out out_aD = routed_out + shared_out
if ( if self.tp_size > 1 and not should_skip_post_experts_all_reduce(
self.tp_size > 1 is_tp_path=True,
and not use_reduce_scatter use_reduce_scatter=use_reduce_scatter,
and not should_use_dp_reduce_scatterv()
): ):
out_aD = tensor_model_parallel_all_reduce(out_aD) out_aD = tensor_model_parallel_all_reduce(out_aD)
+5 -8
View File
@@ -51,8 +51,7 @@ from sglang.srt.layers.logits_processor import LogitsProcessor
from sglang.srt.layers.moe import ( from sglang.srt.layers.moe import (
get_moe_a2a_backend, get_moe_a2a_backend,
get_moe_runner_backend, get_moe_runner_backend,
should_use_dp_reduce_scatterv, should_skip_post_experts_all_reduce,
should_use_flashinfer_cutlass_moe_fp4_allgather,
) )
from sglang.srt.layers.moe.ep_moe.layer import DeepEPMoE, get_moe_impl_class from sglang.srt.layers.moe.ep_moe.layer import DeepEPMoE, get_moe_impl_class
from sglang.srt.layers.moe.topk import TopK, TopKOutputFormat from sglang.srt.layers.moe.topk import TopK, TopKOutputFormat
@@ -298,12 +297,10 @@ class MiMoV2MoE(nn.Module):
final_hidden_states = self.experts(hidden_states, topk_output) final_hidden_states = self.experts(hidden_states, topk_output)
if ( if self.tp_size > 1 and not should_skip_post_experts_all_reduce(
self.tp_size > 1 is_tp_path=True,
and not should_allreduce_fusion use_reduce_scatter=use_reduce_scatter,
and not use_reduce_scatter should_allreduce_fusion=should_allreduce_fusion,
and not should_use_flashinfer_cutlass_moe_fp4_allgather()
and not should_use_dp_reduce_scatterv()
): ):
final_hidden_states = tensor_model_parallel_all_reduce(final_hidden_states) final_hidden_states = tensor_model_parallel_all_reduce(final_hidden_states)
+5 -8
View File
@@ -61,8 +61,7 @@ from sglang.srt.layers.linear import (
from sglang.srt.layers.logits_processor import LogitsProcessor from sglang.srt.layers.logits_processor import LogitsProcessor
from sglang.srt.layers.moe import ( from sglang.srt.layers.moe import (
get_moe_a2a_backend, get_moe_a2a_backend,
should_use_dp_reduce_scatterv, should_skip_post_experts_all_reduce,
should_use_flashinfer_cutlass_moe_fp4_allgather,
) )
from sglang.srt.layers.moe.ep_moe.layer import get_moe_impl_class from sglang.srt.layers.moe.ep_moe.layer import get_moe_impl_class
from sglang.srt.layers.moe.fused_moe_triton.layer import FusedMoE from sglang.srt.layers.moe.fused_moe_triton.layer import FusedMoE
@@ -552,12 +551,10 @@ class MiniMaxM2MoE(nn.Module):
topk_output = self.topk.empty_topk_output(hidden_states.device) topk_output = self.topk.empty_topk_output(hidden_states.device)
final_hidden_states = self.experts(hidden_states, topk_output) final_hidden_states = self.experts(hidden_states, topk_output)
if ( if self.tp_size > 1 and not should_skip_post_experts_all_reduce(
self.tp_size > 1 is_tp_path=True,
and not should_allreduce_fusion use_reduce_scatter=use_reduce_scatter,
and not use_reduce_scatter should_allreduce_fusion=should_allreduce_fusion,
and not should_use_flashinfer_cutlass_moe_fp4_allgather()
and not should_use_dp_reduce_scatterv()
): ):
final_hidden_states = tensor_model_parallel_all_reduce(final_hidden_states) final_hidden_states = tensor_model_parallel_all_reduce(final_hidden_states)
+5 -8
View File
@@ -60,8 +60,7 @@ from sglang.srt.layers.linear import (
from sglang.srt.layers.logits_processor import LogitsProcessor from sglang.srt.layers.logits_processor import LogitsProcessor
from sglang.srt.layers.moe import ( from sglang.srt.layers.moe import (
get_moe_a2a_backend, get_moe_a2a_backend,
should_use_dp_reduce_scatterv, should_skip_post_experts_all_reduce,
should_use_flashinfer_cutlass_moe_fp4_allgather,
) )
from sglang.srt.layers.moe.ep_moe.layer import get_moe_impl_class from sglang.srt.layers.moe.ep_moe.layer import get_moe_impl_class
from sglang.srt.layers.moe.fused_moe_triton import FusedMoE from sglang.srt.layers.moe.fused_moe_triton import FusedMoE
@@ -467,12 +466,10 @@ class Qwen2MoeSparseMoeBlock(nn.Module):
# An out-of-place add would allocate a new tensor outside symm # An out-of-place add would allocate a new tensor outside symm
# memory, breaking subsequent symmetric collective operations. # memory, breaking subsequent symmetric collective operations.
final_hidden_states += shared_output final_hidden_states += shared_output
if ( if self.tp_size > 1 and not should_skip_post_experts_all_reduce(
self.tp_size > 1 is_tp_path=True,
and not should_allreduce_fusion use_reduce_scatter=use_reduce_scatter,
and not use_reduce_scatter should_allreduce_fusion=should_allreduce_fusion,
and not should_use_flashinfer_cutlass_moe_fp4_allgather()
and not should_use_dp_reduce_scatterv()
): ):
final_hidden_states = tensor_model_parallel_all_reduce(final_hidden_states) final_hidden_states = tensor_model_parallel_all_reduce(final_hidden_states)
+9 -13
View File
@@ -50,8 +50,7 @@ from sglang.srt.layers.linear import (
from sglang.srt.layers.logits_processor import LogitsProcessor from sglang.srt.layers.logits_processor import LogitsProcessor
from sglang.srt.layers.moe import ( from sglang.srt.layers.moe import (
get_moe_a2a_backend, get_moe_a2a_backend,
should_use_dp_reduce_scatterv, should_skip_post_experts_all_reduce,
should_use_flashinfer_cutlass_moe_fp4_allgather,
) )
from sglang.srt.layers.moe.ep_moe.layer import get_moe_impl_class from sglang.srt.layers.moe.ep_moe.layer import get_moe_impl_class
from sglang.srt.layers.moe.fused_moe_triton.layer import FusedMoE from sglang.srt.layers.moe.fused_moe_triton.layer import FusedMoE
@@ -332,20 +331,17 @@ class Qwen3MoeSparseMoeBlock(nn.Module):
topk_output = self.topk(hidden_states, router_logits) topk_output = self.topk(hidden_states, router_logits)
final_hidden_states = self.experts(hidden_states, topk_output) final_hidden_states = self.experts(hidden_states, topk_output)
if ( if self.ep_size > 1 and not should_skip_post_experts_all_reduce(
self.ep_size > 1 is_tp_path=False,
and not should_allreduce_fusion use_reduce_scatter=use_reduce_scatter,
and not use_reduce_scatter should_allreduce_fusion=should_allreduce_fusion,
and not should_use_dp_reduce_scatterv()
): ):
final_hidden_states = moe_expert_parallel_all_reduce(final_hidden_states) final_hidden_states = moe_expert_parallel_all_reduce(final_hidden_states)
if ( if self.tp_size > 1 and not should_skip_post_experts_all_reduce(
self.tp_size > 1 is_tp_path=True,
and not should_allreduce_fusion use_reduce_scatter=use_reduce_scatter,
and not use_reduce_scatter should_allreduce_fusion=should_allreduce_fusion,
and not should_use_flashinfer_cutlass_moe_fp4_allgather()
and not should_use_dp_reduce_scatterv()
): ):
final_hidden_states = moe_tensor_model_parallel_all_reduce( final_hidden_states = moe_tensor_model_parallel_all_reduce(
final_hidden_states final_hidden_states
+9 -16
View File
@@ -39,10 +39,7 @@ from sglang.srt.layers.linear import (
RowParallelLinear, RowParallelLinear,
) )
from sglang.srt.layers.logits_processor import LogitsProcessor, LogitsProcessorOutput from sglang.srt.layers.logits_processor import LogitsProcessor, LogitsProcessorOutput
from sglang.srt.layers.moe import ( from sglang.srt.layers.moe import should_skip_post_experts_all_reduce
should_use_dp_reduce_scatterv,
should_use_flashinfer_cutlass_moe_fp4_allgather,
)
from sglang.srt.layers.moe.ep_moe.layer import get_moe_impl_class from sglang.srt.layers.moe.ep_moe.layer import get_moe_impl_class
from sglang.srt.layers.moe.fused_moe_triton.layer import FusedMoE from sglang.srt.layers.moe.fused_moe_triton.layer import FusedMoE
from sglang.srt.layers.moe.topk import TopK from sglang.srt.layers.moe.topk import TopK
@@ -373,12 +370,10 @@ class SarvamMoESparseMoeBlock(nn.Module):
final_hidden_states = final_hidden_states * self.routed_scaling_factor final_hidden_states = final_hidden_states * self.routed_scaling_factor
current_stream.wait_stream(self.alt_stream) current_stream.wait_stream(self.alt_stream)
final_hidden_states = final_hidden_states + shared_out final_hidden_states = final_hidden_states + shared_out
if ( if self.tp_size > 1 and not should_skip_post_experts_all_reduce(
self.tp_size > 1 is_tp_path=True,
and not should_allreduce_fusion use_reduce_scatter=use_reduce_scatter,
and not use_reduce_scatter should_allreduce_fusion=should_allreduce_fusion,
and not should_use_flashinfer_cutlass_moe_fp4_allgather()
and not should_use_dp_reduce_scatterv()
): ):
final_hidden_states = tensor_model_parallel_all_reduce(final_hidden_states) final_hidden_states = tensor_model_parallel_all_reduce(final_hidden_states)
return final_hidden_states.view(num_tokens, hidden_dim) return final_hidden_states.view(num_tokens, hidden_dim)
@@ -417,12 +412,10 @@ class SarvamMoESparseMoeBlock(nn.Module):
elif self.routed_scaling_factor != 1.0: elif self.routed_scaling_factor != 1.0:
final_hidden_states = final_hidden_states * self.routed_scaling_factor final_hidden_states = final_hidden_states * self.routed_scaling_factor
if ( if self.tp_size > 1 and not should_skip_post_experts_all_reduce(
self.tp_size > 1 is_tp_path=True,
and not should_allreduce_fusion use_reduce_scatter=use_reduce_scatter,
and not use_reduce_scatter should_allreduce_fusion=should_allreduce_fusion,
and not should_use_flashinfer_cutlass_moe_fp4_allgather()
and not should_use_dp_reduce_scatterv()
): ):
final_hidden_states = tensor_model_parallel_all_reduce(final_hidden_states) final_hidden_states = tensor_model_parallel_all_reduce(final_hidden_states)
+5 -9
View File
@@ -34,8 +34,7 @@ from sglang.srt.layers.linear import (
from sglang.srt.layers.logits_processor import LogitsProcessor from sglang.srt.layers.logits_processor import LogitsProcessor
from sglang.srt.layers.moe import ( from sglang.srt.layers.moe import (
get_moe_a2a_backend, get_moe_a2a_backend,
should_use_dp_reduce_scatterv, should_skip_post_experts_all_reduce,
should_use_flashinfer_cutlass_moe_fp4_allgather,
) )
from sglang.srt.layers.moe.ep_moe.layer import get_moe_impl_class from sglang.srt.layers.moe.ep_moe.layer import get_moe_impl_class
from sglang.srt.layers.moe.fused_moe_triton.layer import FusedMoE from sglang.srt.layers.moe.fused_moe_triton.layer import FusedMoE
@@ -161,13 +160,10 @@ class SDARMoeSparseMoeBlock(nn.Module):
topk_output = self.topk(hidden_states, router_logits) topk_output = self.topk(hidden_states, router_logits)
out = self.experts(hidden_states, topk_output) # (T, H) out = self.experts(hidden_states, topk_output) # (T, H)
# TP all-reduce (unless fused / reduce_scatter / fp4 allgather / dp reduce_scatterv path) if self.tp_size > 1 and not should_skip_post_experts_all_reduce(
if ( is_tp_path=True,
self.tp_size > 1 use_reduce_scatter=use_reduce_scatter,
and not should_allreduce_fusion should_allreduce_fusion=should_allreduce_fusion,
and not use_reduce_scatter
and not should_use_flashinfer_cutlass_moe_fp4_allgather()
and not should_use_dp_reduce_scatterv()
): ):
out = tensor_model_parallel_all_reduce(out) out = tensor_model_parallel_all_reduce(out)
+5 -8
View File
@@ -31,8 +31,7 @@ from sglang.srt.layers.linear import (
from sglang.srt.layers.logits_processor import LogitsProcessor from sglang.srt.layers.logits_processor import LogitsProcessor
from sglang.srt.layers.moe import ( from sglang.srt.layers.moe import (
get_moe_a2a_backend, get_moe_a2a_backend,
should_use_dp_reduce_scatterv, should_skip_post_experts_all_reduce,
should_use_flashinfer_cutlass_moe_fp4_allgather,
) )
from sglang.srt.layers.moe.ep_moe.layer import get_moe_impl_class from sglang.srt.layers.moe.ep_moe.layer import get_moe_impl_class
from sglang.srt.layers.moe.fused_moe_triton.layer import FusedMoE from sglang.srt.layers.moe.fused_moe_triton.layer import FusedMoE
@@ -233,12 +232,10 @@ class Step3p5MoEMLP(nn.Module):
router_logits=topk_output.router_logits, router_logits=topk_output.router_logits,
) )
final_hidden_states = self.experts(hidden_states, topk_output) final_hidden_states = self.experts(hidden_states, topk_output)
if ( if self.tp_size > 1 and not should_skip_post_experts_all_reduce(
self.tp_size > 1 is_tp_path=True,
and not should_allreduce_fusion use_reduce_scatter=use_reduce_scatter,
and not use_reduce_scatter should_allreduce_fusion=should_allreduce_fusion,
and not should_use_flashinfer_cutlass_moe_fp4_allgather()
and not should_use_dp_reduce_scatterv()
): ):
final_hidden_states = tensor_model_parallel_all_reduce(final_hidden_states) final_hidden_states = tensor_model_parallel_all_reduce(final_hidden_states)