From 63c4996fefc519ecf5a28c7aa9c945bb0b6b1dac Mon Sep 17 00:00:00 2001 From: ybyang <10629930+whybeyoung@users.noreply.github.com> Date: Sun, 5 Jul 2026 03:09:41 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20populate=20batch=20req=20rids=20and=20pe?= =?UTF-8?q?r-request=20http=5Fworker=5Fipc=20for=20mult=E2=80=A6=20(#29882?= =?UTF-8?q?)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- python/sglang/srt/managers/io_struct.py | 5 +++ .../sglang/srt/managers/tokenizer_manager.py | 5 +++ .../tokenizer/test_multi_tokenizer.py | 40 ++++++++++++++++++- 3 files changed, 48 insertions(+), 2 deletions(-) diff --git a/python/sglang/srt/managers/io_struct.py b/python/sglang/srt/managers/io_struct.py index e7f1c2665..8435da43c 100644 --- a/python/sglang/srt/managers/io_struct.py +++ b/python/sglang/srt/managers/io_struct.py @@ -86,6 +86,9 @@ class BaseBatchReq(msgspec.Struct, tag=True, kw_only=True, array_like=True): """Base for batched IPC payloads.""" rids: Optional[List[str]] = None + # Used by batch messages whose items are parallel arrays, such as scheduler + # outputs. Tokenized input batches store routing on batch[i].http_worker_ipc + # because the scheduler unpacks them into single-request handlers. http_worker_ipcs: Optional[List[Optional[str]]] = None @classmethod @@ -881,6 +884,7 @@ class TokenizedGenerateReqInput(BaseReq, kw_only=True): class BatchTokenizedGenerateReqInput(BaseBatchReq, kw_only=True): # The batch of tokenized requests + # Routing for request i is batch[i].http_worker_ipc, not http_worker_ipcs[i]. batch: List[TokenizedGenerateReqInput] def __len__(self): @@ -1166,6 +1170,7 @@ class TokenizedEmbeddingReqInput(BaseReq, kw_only=True): class BatchTokenizedEmbeddingReqInput(BaseBatchReq, kw_only=True): # The batch of tokenized embedding requests + # Routing for request i is batch[i].http_worker_ipc, not http_worker_ipcs[i]. batch: List[TokenizedEmbeddingReqInput] def __len__(self): diff --git a/python/sglang/srt/managers/tokenizer_manager.py b/python/sglang/srt/managers/tokenizer_manager.py index 29d4b0e2e..74688606f 100644 --- a/python/sglang/srt/managers/tokenizer_manager.py +++ b/python/sglang/srt/managers/tokenizer_manager.py @@ -3153,5 +3153,10 @@ class SignalHandler: def stamp_http_worker_ipc(obj: Any, ipc_name: str) -> None: if isinstance(obj, BaseReq): obj.http_worker_ipc = ipc_name + elif isinstance( + obj, (BatchTokenizedGenerateReqInput, BatchTokenizedEmbeddingReqInput) + ): + for req in obj: + req.http_worker_ipc = ipc_name elif isinstance(obj, BaseBatchReq): obj.http_worker_ipcs = [ipc_name] * len(obj.rids) diff --git a/test/registered/tokenizer/test_multi_tokenizer.py b/test/registered/tokenizer/test_multi_tokenizer.py index 1454679d1..a49cf57d0 100644 --- a/test/registered/tokenizer/test_multi_tokenizer.py +++ b/test/registered/tokenizer/test_multi_tokenizer.py @@ -1,5 +1,7 @@ import unittest +import requests + from sglang.srt.utils import kill_process_tree from sglang.test.ci.ci_register import register_amd_ci, register_cuda_ci from sglang.test.kits.eval_accuracy_kit import MMLUMixin @@ -17,8 +19,8 @@ from sglang.test.test_utils import ( write_github_step_summary, ) -register_cuda_ci(est_time=211, stage="base-b", runner_config="1-gpu-large") -register_amd_ci(est_time=345, suite="stage-b-test-1-gpu-small-amd") +register_cuda_ci(est_time=220, stage="base-b", runner_config="1-gpu-large") +register_amd_ci(est_time=355, suite="stage-b-test-1-gpu-small-amd") class TestMultiTokenizer(CustomTestCase, MMLUMixin): @@ -75,6 +77,40 @@ class TestMultiTokenizer(CustomTestCase, MMLUMixin): self.assertLess(res["median_ttft_ms"], 130 if is_in_amd_ci() else 86) self.assertLess(res["median_itl_ms"], 10) + def test_batch_input_ids_routing(self): + # Regression guard for sgl-project/sglang#29878 (introduced by #29214). + # + # A batch of pre-tokenized `input_ids` (no text / multimodal) is the one + # case that takes the batch-tokenization path (_send_batch_request -> + # BatchTokenizedGenerateReqInput). In multi-tokenizer mode this batch + # must stamp each sub-request's `http_worker_ipc` so the scheduler can + # route every reply back to its owning tokenizer worker. If it is missing, + # the requests hang forever. + # + # The existing ttft test only sends *text*, so it never exercises this + # path — this case does, and uses a short timeout so a routing hang + # fails fast instead of stalling until the server launch timeout. + batch_input_ids = [ + [1, 2, 3, 4, 5], + [10, 11, 12, 13, 14], + [20, 21, 22, 23, 24], + [30, 31, 32, 33, 34], + ] + response = requests.post( + self.base_url + "/generate", + json={ + "input_ids": batch_input_ids, + "sampling_params": {"max_new_tokens": 8, "temperature": 0}, + }, + timeout=60, + ) + self.assertEqual(response.status_code, 200, response.text) + results = response.json() + # Every batched request must get its reply routed back — not hang. + self.assertEqual(len(results), len(batch_input_ids)) + for result in results: + self.assertIn("text", result) + if __name__ == "__main__": unittest.main()