fix: populate batch req rids and per-request http_worker_ipc for mult… (#29882)

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):

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)

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_cuda_ci(est_time=220, stage="base-b", runner_config="1-gpu-large")
-register_amd_ci(est_time=345, suite="stage-b-test-1-gpu-small-amd")
+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()