From 3355b6e21b8c031fc2dd9bd446c83c8bc92e34ef Mon Sep 17 00:00:00 2001 From: StonyPort <157573149+zhooooong@users.noreply.github.com> Date: Fri, 16 Jan 2026 17:55:09 +0800 Subject: [PATCH] feat: add request queued timeout (#17143) Co-authored-by: qiuxuan.lzw Co-authored-by: Xiaoyu Zhang <35585791+BBuf@users.noreply.github.com> --- docs/references/environment_variables.md | 1 + python/sglang/srt/environ.py | 1 + python/sglang/srt/managers/scheduler.py | 28 ++++++++++++++ python/sglang/srt/utils/request_logger.py | 2 +- test/registered/scheduler/test_abort.py | 47 +++++++++++++++++++++++ 5 files changed, 78 insertions(+), 1 deletion(-) diff --git a/docs/references/environment_variables.md b/docs/references/environment_variables.md index a24eac334..200fb6000 100644 --- a/docs/references/environment_variables.md +++ b/docs/references/environment_variables.md @@ -16,6 +16,7 @@ SGLang supports various environment variables that can be used to configure its | `SGLANG_HEALTH_CHECK_TIMEOUT` | Timeout for health check in seconds | `20` | | `SGLANG_EPLB_HEATMAP_COLLECTION_INTERVAL` | The interval of passes to collect the metric of selected count of physical experts on each layer and GPU rank. 0 means disabled. | `0` | | `SGLANG_FORWARD_UNKNOWN_TOOLS` | Forward unknown tool calls to clients instead of dropping them | `false` (drop unknown tools) | +| `SGLANG_QUEUED_TIMEOUT_MS` | Timeout (in ms) for requests in the waiting queue | `-1` | ## Performance Tuning diff --git a/python/sglang/srt/environ.py b/python/sglang/srt/environ.py index f2883dc4a..e2bbbed2b 100644 --- a/python/sglang/srt/environ.py +++ b/python/sglang/srt/environ.py @@ -242,6 +242,7 @@ class Envs: SGLANG_PREFILL_DELAYER_MAX_DELAY_PASSES = EnvInt(30) SGLANG_PREFILL_DELAYER_TOKEN_USAGE_LOW_WATERMARK = EnvFloat(None) SGLANG_DATA_PARALLEL_BUDGET_INTERVAL = EnvInt(1) + SGLANG_QUEUED_TIMEOUT_MS = EnvInt(-1) # Test: pd-disaggregation SGLANG_TEST_PD_DISAGG_BACKEND = EnvStr("mooncake") diff --git a/python/sglang/srt/managers/scheduler.py b/python/sglang/srt/managers/scheduler.py index 89e222109..812d72859 100644 --- a/python/sglang/srt/managers/scheduler.py +++ b/python/sglang/srt/managers/scheduler.py @@ -1671,6 +1671,33 @@ class Scheduler( ) return req_to_abort.rid == recv_req.rid + def _abort_on_queued_timeout(self): + if (timeout_ms := envs.SGLANG_QUEUED_TIMEOUT_MS.get()) <= 0: + return + + deleted_reqs = set() + deadline = time.perf_counter() - (timeout_ms / 1000.0) + for req in self.waiting_queue: + entry_time = req.time_stats.wait_queue_entry_time + if 0 < entry_time < deadline: + self.send_to_tokenizer.send_output( + AbortReq( + finished_reason={ + "type": "abort", + "status_code": HTTPStatus.SERVICE_UNAVAILABLE, + "message": "Request queue timeout reached.", + }, + rid=req.rid, + ), + req, + ) + deleted_reqs.add(req) + + if deleted_reqs: + self.waiting_queue = [ + req for req in self.waiting_queue if req not in deleted_reqs + ] + def handle_embedding_request( self, recv_req: TokenizedEmbeddingReqInput, @@ -1739,6 +1766,7 @@ class Scheduler( self.handle_embedding_request(tokenized_req) def get_next_batch_to_run(self) -> Optional[ScheduleBatch]: + self._abort_on_queued_timeout() if self.dllm_config is not None: if self.chunked_req is not None and self.chunked_req.finished(): self.chunked_req = None diff --git a/python/sglang/srt/utils/request_logger.py b/python/sglang/srt/utils/request_logger.py index fca783c30..5b8aee550 100644 --- a/python/sglang/srt/utils/request_logger.py +++ b/python/sglang/srt/utils/request_logger.py @@ -130,7 +130,7 @@ class RequestLogger: if not self.log_requests: return - e2e_latency_ms = out["meta_info"]["e2e_latency"] * 1000 + e2e_latency_ms = out["meta_info"].get("e2e_latency", 0) * 1000 if self.log_exceeded_ms > 0 and e2e_latency_ms < self.log_exceeded_ms: return diff --git a/test/registered/scheduler/test_abort.py b/test/registered/scheduler/test_abort.py index b8e6b691d..466891608 100644 --- a/test/registered/scheduler/test_abort.py +++ b/test/registered/scheduler/test_abort.py @@ -191,5 +191,52 @@ class TestAbortAllWithRetraction(CustomTestCase): print("Finished test_abort_all_with_retraction") +class TestAbortWithQueueTimeout(CustomTestCase): + @classmethod + def setUpClass(cls): + cls.model = DEFAULT_MODEL_NAME_FOR_TEST + cls.base_url = DEFAULT_URL_FOR_TEST + with envs.SGLANG_QUEUED_TIMEOUT_MS.override(1): + cls.process = popen_launch_server( + cls.model, + cls.base_url, + timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH, + other_args=[ + "--max-running-requests=1", + ], + ) + + @classmethod + def tearDownClass(cls): + kill_process_tree(cls.process.pid) + + def _run_decode(self): + response = requests.post( + self.base_url + "/generate", + json={ + "text": "Today is ", + "sampling_params": { + "temperature": 0, + "max_new_tokens": 512, + "ignore_eos": True, + }, + }, + ) + return response.json() + + def test_queue_timeout(self): + num_requests = 2 + with ThreadPoolExecutor(num_requests) as executor: + futures = [executor.submit(self._run_decode) for _ in range(num_requests)] + + error_count = 0 + for future in as_completed(futures): + result = future.result() + if result.get("object") == "error": + error_count += 1 + self.assertEqual(result["code"], 503) + self.assertEqual(error_count, 1) + + if __name__ == "__main__": unittest.main()