From 70088aa5dbb77a7e70a16a4fb1a106fc7e8b2764 Mon Sep 17 00:00:00 2001 From: Mohammad Miadh Angkad <176301910+mmangkad@users.noreply.github.com> Date: Sat, 29 Aug 2026 03:57:07 +0800 Subject: [PATCH] Fix KeyError on batch requests whose state is freed before it is read (#36638) --- .../sglang/srt/managers/tokenizer_manager.py | 13 +++++++-- .../test_tokenizer_manager_rid_cleanup.py | 28 +++++++++++++++++++ 2 files changed, 39 insertions(+), 2 deletions(-) diff --git a/python/sglang/srt/managers/tokenizer_manager.py b/python/sglang/srt/managers/tokenizer_manager.py index 0277a955b..f826be496 100644 --- a/python/sglang/srt/managers/tokenizer_manager.py +++ b/python/sglang/srt/managers/tokenizer_manager.py @@ -1688,13 +1688,22 @@ class TokenizerManager(TokenizerControlMixin, TokenizerManagerScoreMixin): return None - async def _wait_one_response( + def _wait_one_response( self, obj: Union[GenerateReqInput, EmbeddingReqInput], request: Optional[fastapi.Request] = None, ): - """Wait for the response of one request.""" + # Batch dispatch builds every waiter before advancing any. + # Both removers append the output after the del, so the ReqState stays valid. state = self.rid_to_state[obj.rid] + return self._stream_one_response(obj=obj, state=state, request=request) + + async def _stream_one_response( + self, + obj: Union[GenerateReqInput, EmbeddingReqInput], + state: ReqState, + request: Optional[fastapi.Request] = None, + ): # Not all request types have `stream` (e.g., EmbeddingReqInput). Default to non-streaming. is_stream = getattr(obj, "stream", False) while True: diff --git a/test/registered/unit/managers/test_tokenizer_manager_rid_cleanup.py b/test/registered/unit/managers/test_tokenizer_manager_rid_cleanup.py index b44f172e8..93a3f8bf2 100644 --- a/test/registered/unit/managers/test_tokenizer_manager_rid_cleanup.py +++ b/test/registered/unit/managers/test_tokenizer_manager_rid_cleanup.py @@ -613,5 +613,33 @@ class TestGenerateRequestCleanupOnDispatchFailure(CustomTestCase): self.assertFalse(tm.rid_to_state) +class TestWaitOneResponseAfterStateFreed(CustomTestCase): + """A waiter built before its request finishes must still deliver the output. + + Batch dispatch builds every waiter before advancing any, and the + scheduler-response path drops rid_to_state as soon as a request finishes. + """ + + def test_generator_built_before_finish_still_delivers_output(self): + tm = _make_tokenizer_manager(self) + tm.request_logger = Mock() + tm.request_metrics_exporter_manager = MagicMock() + tm.request_metrics_exporter_manager.exporter_enabled.return_value = False + rid = "freed_state_rid" + state = _make_req_state(rid) + state.obj.background = True # skip the fastapi disconnect probe + tm.rid_to_state[rid] = state + + async def drive(): + waiter = tm._wait_one_response(state.obj, None) + await tm._handle_batch_output(_make_batch_str_output(rid)) + self.assertNotIn(rid, tm.rid_to_state) + return await waiter.__anext__() + + out = asyncio.run(drive()) + self.assertEqual(out["meta_info"]["id"], rid) + self.assertEqual(out["text"], "hello") + + if __name__ == "__main__": unittest.main(verbosity=2)