[Refactor] Refactor fused_moe_triton tuning tools: extract shared utils, add EP/MLLM support, reduce overhead (#12440)
Co-authored-by: xu-yfei <xu-yfei@users.noreply.github.com> Co-authored-by: Yongfei Xu <xuyongfei.xyf@antgroup.com>
This commit is contained in:
Xiaoyu Zhang
co-authored by
xu-yfei
Yongfei Xu
xu-yfei
Yongfei Xu
parent
8be0e1bc9c
commit
fc84b0730c
@@ -1,22 +1,28 @@
|
||||
# Adapted from https://github.com/vllm-project/vllm/blob/main/benchmarks/kernels/benchmark_moe.py
|
||||
import argparse
|
||||
import json
|
||||
import time
|
||||
from contextlib import nullcontext
|
||||
from datetime import datetime
|
||||
from typing import Any, Dict, List, Tuple, TypedDict
|
||||
from typing import Any, Dict, List, Tuple
|
||||
|
||||
import ray
|
||||
import torch
|
||||
import triton
|
||||
from common_utils import (
|
||||
BenchmarkConfig,
|
||||
get_config_filename,
|
||||
get_configs_compute_bound,
|
||||
get_default_batch_sizes,
|
||||
get_model_config,
|
||||
save_configs,
|
||||
sort_config,
|
||||
)
|
||||
from ray.experimental.tqdm_ray import tqdm
|
||||
from transformers import AutoConfig
|
||||
|
||||
from sglang.srt.layers.moe.fused_moe_triton import override_config
|
||||
from sglang.srt.layers.moe.fused_moe_triton.fused_moe import fused_moe
|
||||
from sglang.srt.layers.moe.fused_moe_triton.fused_moe_triton_config import (
|
||||
get_config_dtype_str,
|
||||
get_config_file_name,
|
||||
get_default_config,
|
||||
get_moe_configs,
|
||||
)
|
||||
@@ -27,15 +33,6 @@ from sglang.srt.utils import is_hip
|
||||
_is_hip = is_hip()
|
||||
|
||||
|
||||
class BenchmarkConfig(TypedDict):
|
||||
BLOCK_SIZE_M: int
|
||||
BLOCK_SIZE_N: int
|
||||
BLOCK_SIZE_K: int
|
||||
GROUP_SIZE_M: int
|
||||
num_warps: int
|
||||
num_stages: int
|
||||
|
||||
|
||||
def benchmark_config(
|
||||
config: BenchmarkConfig,
|
||||
num_tokens: int,
|
||||
@@ -173,74 +170,28 @@ def benchmark_config(
|
||||
graph.replay()
|
||||
torch.cuda.synchronize()
|
||||
|
||||
start_event = torch.cuda.Event(enable_timing=True)
|
||||
end_event = torch.cuda.Event(enable_timing=True)
|
||||
# Flush L2 cache with 256 MB data
|
||||
cache_flush = torch.empty(int(256e6 // 4), dtype=torch.int, device="cuda")
|
||||
cache_flush.zero_()
|
||||
|
||||
start_events = [torch.cuda.Event(enable_timing=True) for _ in range(num_iters)]
|
||||
end_events = [torch.cuda.Event(enable_timing=True) for _ in range(num_iters)]
|
||||
|
||||
for i in range(num_iters):
|
||||
prepare(i)
|
||||
start_events[i].record()
|
||||
graph.replay()
|
||||
end_events[i].record()
|
||||
torch.cuda.synchronize()
|
||||
|
||||
latencies: List[float] = []
|
||||
for i in range(num_iters):
|
||||
prepare(i)
|
||||
torch.cuda.synchronize()
|
||||
|
||||
start_event.record()
|
||||
graph.replay()
|
||||
end_event.record()
|
||||
end_event.synchronize()
|
||||
latencies.append(start_event.elapsed_time(end_event))
|
||||
latencies.append(start_events[i].elapsed_time(end_events[i]))
|
||||
avg = sum(latencies) / (num_iters * 10) * 1000 # us
|
||||
graph.reset()
|
||||
return avg
|
||||
|
||||
|
||||
def get_rocm_configs_compute_bound() -> List[Dict[str, int]]:
|
||||
configs: List[BenchmarkConfig] = []
|
||||
waves_per_eu_range = 0
|
||||
for num_stages in [2]:
|
||||
for block_m in [32, 64, 128, 256]:
|
||||
for block_k in [32, 64, 128, 256]:
|
||||
for block_n in [16, 32, 64, 128, 256]:
|
||||
for num_warps in [1, 2, 4, 8]:
|
||||
for group_size in [1, 4, 8, 16, 32]:
|
||||
configs.append(
|
||||
{
|
||||
"BLOCK_SIZE_M": block_m,
|
||||
"BLOCK_SIZE_N": block_n,
|
||||
"BLOCK_SIZE_K": block_k,
|
||||
"GROUP_SIZE_M": group_size,
|
||||
"num_warps": num_warps,
|
||||
"num_stages": num_stages,
|
||||
"waves_per_eu": waves_per_eu_range,
|
||||
}
|
||||
)
|
||||
return configs
|
||||
|
||||
|
||||
def get_configs_compute_bound() -> List[Dict[str, int]]:
|
||||
# Reduced search space for faster tuning.
|
||||
# TODO(woosuk): Increase the search space and use a performance model to
|
||||
# prune the search space.
|
||||
configs: List[BenchmarkConfig] = []
|
||||
if _is_hip:
|
||||
configs = get_rocm_configs_compute_bound()
|
||||
else:
|
||||
for num_stages in [2, 3, 4, 5]:
|
||||
for block_m in [16, 32, 64, 128, 256]:
|
||||
for block_k in [64, 128, 256]:
|
||||
for block_n in [32, 64, 128, 256]:
|
||||
for num_warps in [4, 8]:
|
||||
for group_size in [1, 16, 32, 64]:
|
||||
configs.append(
|
||||
{
|
||||
"BLOCK_SIZE_M": block_m,
|
||||
"BLOCK_SIZE_N": block_n,
|
||||
"BLOCK_SIZE_K": block_k,
|
||||
"GROUP_SIZE_M": group_size,
|
||||
"num_warps": num_warps,
|
||||
"num_stages": num_stages,
|
||||
}
|
||||
)
|
||||
return configs
|
||||
|
||||
|
||||
@ray.remote(num_gpus=1)
|
||||
class BenchmarkWorker:
|
||||
|
||||
@@ -360,189 +311,27 @@ class BenchmarkWorker:
|
||||
return best_config
|
||||
|
||||
|
||||
def sort_config(config: BenchmarkConfig) -> BenchmarkConfig:
|
||||
return {
|
||||
"BLOCK_SIZE_M": config["BLOCK_SIZE_M"],
|
||||
"BLOCK_SIZE_N": config["BLOCK_SIZE_N"],
|
||||
"BLOCK_SIZE_K": config["BLOCK_SIZE_K"],
|
||||
"GROUP_SIZE_M": config["GROUP_SIZE_M"],
|
||||
"num_warps": config["num_warps"],
|
||||
"num_stages": config["num_stages"],
|
||||
**(
|
||||
{"waves_per_eu": config["waves_per_eu"]} if "waves_per_eu" in config else {}
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def save_configs(
|
||||
configs: Dict[int, BenchmarkConfig],
|
||||
filename: str,
|
||||
) -> None:
|
||||
print(f"Writing best config to {filename}...")
|
||||
with open(filename, "w") as f:
|
||||
json.dump(configs, f, indent=4)
|
||||
f.write("\n")
|
||||
|
||||
|
||||
def get_filename(
|
||||
num_experts: int,
|
||||
shard_intermediate_size: int,
|
||||
hidden_size: int,
|
||||
topk: int,
|
||||
dtype: torch.dtype,
|
||||
use_fp8_w8a8: bool,
|
||||
use_int8_w8a8: bool,
|
||||
use_int8_w8a16: bool,
|
||||
per_channel_quant: bool,
|
||||
block_shape: List[int],
|
||||
) -> None:
|
||||
dtype_str = get_config_dtype_str(
|
||||
dtype,
|
||||
use_int8_w8a16=use_int8_w8a16,
|
||||
use_fp8_w8a8=use_fp8_w8a8,
|
||||
use_int8_w8a8=use_int8_w8a8,
|
||||
)
|
||||
|
||||
# NOTE(woosuk): The current naming convention uses w2.shape[2], which
|
||||
# is the intermediate size after silu_and_mul.
|
||||
filename = get_config_file_name(
|
||||
num_experts,
|
||||
shard_intermediate_size // 2,
|
||||
dtype_str,
|
||||
block_shape,
|
||||
per_channel_quant,
|
||||
)
|
||||
|
||||
return filename
|
||||
|
||||
|
||||
def main(args: argparse.Namespace):
|
||||
print(args)
|
||||
|
||||
def _calculate_shard_intermediate_size(intermediate_size: int) -> int:
|
||||
# In EP mode, use original intermediate_size; otherwise apply TP sharding
|
||||
return (
|
||||
intermediate_size
|
||||
if args.ep_size > 1
|
||||
else 2 * intermediate_size // args.tp_size
|
||||
)
|
||||
model_config = get_model_config(
|
||||
args.model, args.tp_size, args.ep_size, args.disable_shared_experts_fusion
|
||||
)
|
||||
|
||||
# Check EP mode constraint: tp_size must be 1 when ep_size > 1
|
||||
if args.ep_size > 1 and args.tp_size != 1:
|
||||
raise ValueError(
|
||||
f"When using Expert Parallelism (ep_size={args.ep_size}), "
|
||||
f"tp_size must be set to 1, but got tp_size={args.tp_size}. "
|
||||
f"Please set --tp-size 1 when using --ep-size > 1."
|
||||
)
|
||||
E = model_config["num_experts"]
|
||||
topk = model_config["topk"]
|
||||
hidden_size = model_config["hidden_size"]
|
||||
shard_intermediate_size = model_config["shard_intermediate_size"]
|
||||
dtype = model_config["dtype"]
|
||||
block_shape = model_config["block_shape"]
|
||||
|
||||
config = AutoConfig.from_pretrained(args.model, trust_remote_code=True)
|
||||
|
||||
# Determine block shape for quantization
|
||||
block_shape = None
|
||||
if (
|
||||
hasattr(config, "quantization_config")
|
||||
and "weight_block_size" in config.quantization_config
|
||||
):
|
||||
block_shape = config.quantization_config["weight_block_size"]
|
||||
assert len(block_shape) == 2
|
||||
|
||||
architecture = config.architectures[0]
|
||||
# replace config with text_config for encoder-decoder models after getting block_shape and architecture
|
||||
if hasattr(config, "text_config"):
|
||||
config = config.get_text_config()
|
||||
|
||||
if architecture == "DbrxForCausalLM":
|
||||
E = config.ffn_config.moe_num_experts
|
||||
topk = config.ffn_config.moe_top_k
|
||||
intermediate_size = config.ffn_config.ffn_hidden_size
|
||||
shard_intermediate_size = _calculate_shard_intermediate_size(intermediate_size)
|
||||
elif architecture == "JambaForCausalLM":
|
||||
E = config.num_experts
|
||||
topk = config.num_experts_per_tok
|
||||
intermediate_size = config.intermediate_size
|
||||
shard_intermediate_size = _calculate_shard_intermediate_size(intermediate_size)
|
||||
elif architecture in [
|
||||
"Qwen2MoeForCausalLM",
|
||||
"Qwen3MoeForCausalLM",
|
||||
"Qwen3NextForCausalLM",
|
||||
"Qwen3VLMoeForConditionalGeneration",
|
||||
]:
|
||||
E = config.num_experts // args.ep_size
|
||||
topk = config.num_experts_per_tok
|
||||
intermediate_size = config.moe_intermediate_size
|
||||
shard_intermediate_size = _calculate_shard_intermediate_size(intermediate_size)
|
||||
elif architecture in ["DeepseekV2ForCausalLM", "DeepseekV3ForCausalLM"]:
|
||||
E = (
|
||||
config.n_routed_experts + (0 if args.disable_shared_experts_fusion else 1)
|
||||
if architecture == "DeepseekV3ForCausalLM"
|
||||
else config.n_routed_experts
|
||||
)
|
||||
topk = config.num_experts_per_tok
|
||||
intermediate_size = config.moe_intermediate_size
|
||||
shard_intermediate_size = _calculate_shard_intermediate_size(intermediate_size)
|
||||
elif architecture == "Llama4ForConditionalGeneration":
|
||||
E = config.num_local_experts + (0 if args.disable_shared_experts_fusion else 1)
|
||||
topk = config.num_experts_per_tok
|
||||
intermediate_size = config.intermediate_size
|
||||
shard_intermediate_size = _calculate_shard_intermediate_size(intermediate_size)
|
||||
elif architecture in [
|
||||
"Grok1ForCausalLM",
|
||||
"Grok1ImgGen",
|
||||
"Grok1AForCausalLM",
|
||||
]:
|
||||
E = config.num_local_experts // args.ep_size
|
||||
topk = config.num_experts_per_tok
|
||||
intermediate_size = config.moe_intermediate_size
|
||||
shard_intermediate_size = _calculate_shard_intermediate_size(intermediate_size)
|
||||
elif architecture in [
|
||||
"BailingMoEForCausalLM",
|
||||
"BailingMoeForCausalLM",
|
||||
"BailingMoeV2ForCausalLM",
|
||||
]:
|
||||
E = config.num_experts // args.ep_size
|
||||
topk = config.num_experts_per_tok
|
||||
intermediate_size = config.moe_intermediate_size
|
||||
shard_intermediate_size = _calculate_shard_intermediate_size(intermediate_size)
|
||||
elif architecture in ["Glm4MoeForCausalLM"]:
|
||||
E = config.n_routed_experts
|
||||
topk = config.num_experts_per_tok
|
||||
intermediate_size = config.moe_intermediate_size
|
||||
shard_intermediate_size = _calculate_shard_intermediate_size(intermediate_size)
|
||||
else:
|
||||
# Default: Mixtral
|
||||
E = config.num_local_experts // args.ep_size
|
||||
topk = config.num_experts_per_tok
|
||||
intermediate_size = config.intermediate_size
|
||||
shard_intermediate_size = _calculate_shard_intermediate_size(intermediate_size)
|
||||
|
||||
hidden_size = config.hidden_size
|
||||
dtype = config.torch_dtype
|
||||
use_fp8_w8a8 = args.dtype == "fp8_w8a8"
|
||||
use_int8_w8a8 = args.dtype == "int8_w8a8"
|
||||
use_int8_w8a16 = args.dtype == "int8_w8a16"
|
||||
per_channel_quant = args.per_channel_quant
|
||||
|
||||
if args.batch_size is None:
|
||||
batch_sizes = [
|
||||
1,
|
||||
2,
|
||||
4,
|
||||
8,
|
||||
16,
|
||||
24,
|
||||
32,
|
||||
48,
|
||||
64,
|
||||
96,
|
||||
128,
|
||||
256,
|
||||
512,
|
||||
1024,
|
||||
1536,
|
||||
2048,
|
||||
3072,
|
||||
4096,
|
||||
]
|
||||
batch_sizes = get_default_batch_sizes()
|
||||
else:
|
||||
batch_sizes = [args.batch_size]
|
||||
|
||||
@@ -571,7 +360,7 @@ def main(args: argparse.Namespace):
|
||||
if block_k % config["BLOCK_SIZE_K"] == 0
|
||||
]
|
||||
|
||||
filename = get_filename(
|
||||
filename = get_config_filename(
|
||||
E,
|
||||
shard_intermediate_size,
|
||||
hidden_size,
|
||||
|
||||
Reference in New Issue
Block a user