[AMD] Add SGLANG_MORI_MOE_MAX_INPUT_TOKENS to truncate dispatch before MoE. (#22952)

docs/references/environment_variables.md

@@ -84,6 +84,7 @@ SGLang supports various environment variables that can be used to configure its
 | `SGLANG_MORI_NUM_MAX_DISPATCH_TOKENS_PER_RANK` | Maximum number of dispatch tokens per rank for MORI-EP buffer allocation | `4096` |
 | `SGLANG_MORI_DISPATCH_INTER_KERNEL_SWITCH_THRESHOLD` | Threshold for switching between `InterNodeV1` and `InterNodeV1LL` kernel types. `InterNodeV1LL` is used if `SGLANG_MORI_NUM_MAX_DISPATCH_TOKENS_PER_RANK` is less than or equal to this threshold; otherwise, `InterNodeV1` is used. | `256` |
 | `SGLANG_MORI_PREALLOC_MAX_RECV_TOKENS` | This argument devives `SGLANG_MORI_NUM_MAX_DISPATCH_TOKENS_PER_RANK` which indicates customized amount of tokens preallocated for a rank, valid range from 1 to world_size*SGLANG_MORI_NUM_MAX_DISPATCH_TOKENS_PER_RANK, by default `0` means maximum. Setting a smaller value will reduce memory footprint but too small value could cause buffer overflow. | `0` |
+| `SGLANG_MORI_MOE_MAX_INPUT_TOKENS` | Truncate the dispatch buffer to this many rows before MoE computation, reducing kernel overhead on padding tokens. The value must be >= the actual number of received tokens (`totalRecvTokenNum`); setting it too small causes incorrect results. `0` disables truncation (use full buffer). | `0` |
 | `SGLANG_MORI_QP_PER_TRANSFER` | Number of RDMA Queue Pairs (QPs) used per transfer operation | `1` |
 | `SGLANG_MORI_POST_BATCH_SIZE` | Number of RDMA work requests posted in a single batch to each QP | `-1` |
 | `SGLANG_MORI_NUM_WORKERS` | Number of worker threads in the RDMA executor thread pool | `1` |

python/sglang/srt/layers/moe/ep_moe/layer.py

@@ -39,7 +39,7 @@ from sglang.srt.layers.quantization.fp8 import Fp8Config, Fp8MoEMethod
 from sglang.srt.layers.quantization.fp8_kernel import is_fp8_fnuz
 from sglang.srt.layers.quantization.quark.schemes import QuarkW4A4MXFp4MoE
 from sglang.srt.layers.quantization.w4afp8 import W4AFp8Config, W4AFp8MoEMethod
-from sglang.srt.utils import get_bool_env_var, is_hip, is_npu
+from sglang.srt.utils import get_bool_env_var, get_int_env_var, is_hip, is_npu
 
 if TYPE_CHECKING:
     from sglang.srt.layers.moe.token_dispatcher import (
@@ -635,6 +635,10 @@ class MoriEPMoE(DeepEPMoE):
         expert_end_idx = expert_start_idx + self.num_local_experts
         self.expert_mask[expert_start_idx:expert_end_idx] = 1
 
+        self.mori_moe_max_input_tokens = get_int_env_var(
+            "SGLANG_MORI_MOE_MAX_INPUT_TOKENS", 0
+        )
+
     def forward(
         self,
         hidden_states: torch.Tensor,
@@ -681,6 +685,19 @@ class MoriEPMoE(DeepEPMoE):
             dispatch_output.out_dtype,
         )
 
+        # Truncate dispatch tensors to reduce MoE computation on padding rows.
+        # dispatch_a1 has shape (M, hidden_size) where M is the full buffer size,
+        # but only the first dispatch_recv_token_num rows are valid.
+        # mori combine only reads [0, totalRecvTokenNum), so the truncated
+        # output can be passed directly without padding back.
+        if self.mori_moe_max_input_tokens > 0:
+            limit = self.mori_moe_max_input_tokens
+            dispatch_a1 = dispatch_a1[:limit]
+            if dispatch_scale is not None:
+                dispatch_scale = dispatch_scale[:limit]
+            dispatch_ids = dispatch_ids[:limit]
+            dispatch_weights = dispatch_weights[:limit]
+
         w13_weight = self.w13_weight
         w2_weight = self.w2_weight