[Qwen3-next] Add PD disaggregation support for mamba with extra_buffer (#15180)

Signed-off-by: Shangming Cai <csmthu@gmail.com>
Co-authored-by: ybyang <10629930+whybeyoung@users.noreply.github.com>
This commit is contained in:
Shangming Cai
2025-12-16 14:36:00 +08:00
committed by GitHub
co-authored by ybyang
parent 6292d97135
commit 36fcf71fff
5 changed files with 79 additions and 7 deletions
+1 -1
View File
@@ -157,7 +157,7 @@ suites = {
"per-commit-8-gpu-h200": [
TestFile("test_deepseek_v3_basic.py", 275),
TestFile("test_deepseek_v3_mtp.py", 275),
TestFile("test_disaggregation_hybrid_attention.py", 200),
TestFile("test_disaggregation_hybrid_attention.py", 600),
TestFile("models/test_kimi_k2_models.py", 200),
TestFile("test_deepseek_v32_basic.py", 275),
TestFile("test_deepseek_v32_mtp.py", 275),
@@ -79,5 +79,77 @@ class TestDisaggregationHybridAttentionMamba(PDDisaggregationServerBase):
self.assertGreater(metrics["accuracy"], 0.93)
class TestDisaggregationHybridAttentionMambaExtraBuffer(PDDisaggregationServerBase):
@classmethod
def setUpClass(cls):
super().setUpClass()
cls.model = "Qwen/Qwen3-Next-80B-A3B-Instruct"
# Non blocking start servers
cls.start_prefill()
cls.start_decode()
# Block until both
cls.wait_server_ready(cls.prefill_url + "/health")
cls.wait_server_ready(cls.decode_url + "/health")
cls.launch_lb()
@classmethod
def start_prefill(cls):
prefill_args = [
"--trust-remote-code",
"--disaggregation-mode",
"prefill",
"--tp",
"4",
"--mamba-scheduler-strategy",
"extra_buffer",
]
prefill_args += cls.transfer_backend + cls.rdma_devices
cls.process_prefill = popen_launch_pd_server(
cls.model,
cls.prefill_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=prefill_args,
)
@classmethod
def start_decode(cls):
decode_args = [
"--trust-remote-code",
"--disaggregation-mode",
"decode",
"--tp",
"4",
"--base-gpu-id",
"4",
"--mamba-scheduler-strategy",
"extra_buffer",
]
decode_args += cls.transfer_backend + cls.rdma_devices
cls.process_decode = popen_launch_pd_server(
cls.model,
cls.decode_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=decode_args,
)
def test_gsm8k(self):
args = SimpleNamespace(
num_shots=5,
data_path=None,
num_questions=200,
max_new_tokens=512,
parallel=128,
host=f"http://{self.base_host}",
port=int(self.lb_port),
)
metrics = run_eval_few_shot_gsm8k(args)
print(f"Evaluation metrics: {metrics}")
self.assertGreater(metrics["accuracy"], 0.93)
if __name__ == "__main__":
unittest.main()