feat: add request queued timeout (#17143)

Co-authored-by: qiuxuan.lzw <qiuxuan.lzw@alibaba-inc.com>
Co-authored-by: Xiaoyu Zhang <35585791+BBuf@users.noreply.github.com>

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
 

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

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

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
 

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