diff --git a/python/sglang/srt/managers/tokenizer_manager.py b/python/sglang/srt/managers/tokenizer_manager.py index aa7a5d95a..7beaa1d42 100644 --- a/python/sglang/srt/managers/tokenizer_manager.py +++ b/python/sglang/srt/managers/tokenizer_manager.py @@ -482,6 +482,7 @@ class TokenizerManager(TokenizerCommunicatorMixin, TokenizerManagerMultiItemMixi # Normalize the request obj.normalize_batch_and_arguments() + self._validate_rid(obj) self._req_stats_init(obj, request) if self.server_args.language_only: @@ -737,6 +738,21 @@ class TokenizerManager(TokenizerCommunicatorMixin, TokenizerManagerMultiItemMixi obj, input_text, input_ids, input_embeds, mm_inputs, token_type_ids ) + def _validate_rid(self, obj: Union[GenerateReqInput, EmbeddingReqInput]) -> None: + """Validate the request ID (rid) uniqueness.""" + rid = obj.rid + if rid is None: + return + ids = rid if isinstance(rid, list) else [rid] + if len(ids) != len(set(ids)): + raise ValueError( + f"Duplicate request IDs detected within the request: {ids}" + ) + + for i in ids: + if i in self.rid_to_state: + raise ValueError(f"Duplicate request ID detected: {i}") + def _validate_one_request( self, obj: Union[GenerateReqInput, EmbeddingReqInput], input_ids: List[int] ) -> None: diff --git a/test/registered/scheduler/test_abort.py b/test/registered/scheduler/test_abort.py index 9caaa2cac..e2f5d16f5 100644 --- a/test/registered/scheduler/test_abort.py +++ b/test/registered/scheduler/test_abort.py @@ -1,4 +1,5 @@ import multiprocessing +import threading import time import unittest from concurrent.futures import ThreadPoolExecutor, as_completed @@ -125,6 +126,86 @@ class TestAbortAll(AbortAllMixin, CustomTestCase): def tearDownClass(cls): kill_process_tree(cls.process.pid) + def _generate_with_rid(self, rid, max_new_tokens=8): + return requests.post( + f"{self.base_url}/generate", + json={ + "text": "The capital of France is", + "sampling_params": { + "temperature": 0, + "max_new_tokens": max_new_tokens, + }, + "rid": rid, + }, + timeout=30, + ) + + def test_duplicate_rid_sequential_ok(self): + rid = "dup-rid-test-sequential" + resp1 = self._generate_with_rid(rid) + self.assertEqual(resp1.status_code, 200) + self.assertNotIn("error", resp1.json()) + + resp2 = self._generate_with_rid(rid) + self.assertEqual(resp2.status_code, 200) + self.assertNotIn("error", resp2.json()) + + def test_duplicate_rid_concurrent_rejected(self): + rid = "dup-rid-test-concurrent" + results = {} + + def send(key, max_tokens): + results[key] = self._generate_with_rid(rid, max_new_tokens=max_tokens) + + t1 = threading.Thread(target=send, args=("first", 512)) + t2 = threading.Thread(target=send, args=("second", 8)) + t1.start() + time.sleep(0.1) + t2.start() + t1.join(timeout=30) + t2.join(timeout=30) + + r1, r2 = results["first"], results["second"] + self.assertTrue( + r1.status_code == 400 or r2.status_code == 400, + "One of the concurrent duplicate-rid requests should be rejected", + ) + + rejected = r2 if r2.status_code == 400 else r1 + self.assertIn("Duplicate request ID", rejected.json()["error"]["message"]) + + def test_duplicate_rid_in_batch(self): + rid = "dup-rid-batch" + response = requests.post( + f"{self.base_url}/generate", + json={ + "text": ["Hello", "World"], + "sampling_params": {"temperature": 0, "max_new_tokens": 8}, + "rid": [rid, rid], + }, + timeout=30, + ) + self.assertEqual(response.status_code, 400) + self.assertIn("Duplicate request ID", response.json()["error"]["message"]) + + def test_server_healthy_after_duplicate_rid(self): + requests.post( + f"{self.base_url}/generate", + json={ + "text": ["Hello", "World"], + "sampling_params": {"temperature": 0, "max_new_tokens": 8}, + "rid": ["dup-health", "dup-health"], + }, + timeout=30, + ) + + resp = requests.get(f"{self.base_url}/health", timeout=5) + self.assertEqual(resp.status_code, 200) + + resp = self._generate_with_rid("after-dup-health") + self.assertEqual(resp.status_code, 200) + self.assertIn("text", resp.json()) + class TestAbortAllWithRetraction(CustomTestCase): @classmethod