diff --git a/python/sglang/srt/entrypoints/http_server.py b/python/sglang/srt/entrypoints/http_server.py index e7eb05f96..eeefe3f9b 100644 --- a/python/sglang/srt/entrypoints/http_server.py +++ b/python/sglang/srt/entrypoints/http_server.py @@ -736,9 +736,15 @@ async def classify_request(obj: EmbeddingReqInput, request: Request): async def flush_cache(timeout: float = Query(0.0, ge=0.0)): """Flush the radix cache.""" ret = await _global_state.tokenizer_manager.flush_cache(timeout_s=timeout) + if ret.success: + content = ( + "Cache flushed.\nPlease check backend logs for more details. " + "(When there are running or waiting requests, the operation will not be performed.)\n" + ) + else: + content = ret.message or "Flush cache failed.\n" return Response( - content="Cache flushed.\nPlease check backend logs for more details. " - "(When there are running or waiting requests, the operation will not be performed.)\n", + content=content, status_code=200 if ret.success else HTTPStatus.BAD_REQUEST, ) diff --git a/python/sglang/srt/managers/io_struct.py b/python/sglang/srt/managers/io_struct.py index 468143be2..139a588ae 100644 --- a/python/sglang/srt/managers/io_struct.py +++ b/python/sglang/srt/managers/io_struct.py @@ -1172,6 +1172,7 @@ class FlushCacheReqInput(BaseReq): @dataclass class FlushCacheReqOutput(BaseReq): success: bool + message: str = "" @dataclass diff --git a/python/sglang/srt/managers/scheduler.py b/python/sglang/srt/managers/scheduler.py index b050a72c9..8c1e33c19 100644 --- a/python/sglang/srt/managers/scheduler.py +++ b/python/sglang/srt/managers/scheduler.py @@ -834,7 +834,7 @@ class Scheduler( self.last_batch: Optional[ScheduleBatch] = None self.forward_ct = 0 self.return_health_check_ipcs: Deque[Optional[str]] = deque() - self._pending_flush: Deque[Tuple[FlushCacheReqInput, float]] = deque() + self._pending_flush: Optional[Tuple[FlushCacheReqInput, float]] = None self.num_retracted_reqs: int = 0 self.num_paused_reqs: int = 0 self.session_controller = SessionController(self.tree_cache) @@ -2785,38 +2785,40 @@ class Scheduler( ) def _check_pending_flush(self): - if not self._pending_flush: + if self._pending_flush is None: return + pending_req, deadline = self._pending_flush + if self.is_fully_idle(): success = self.flush_cache() - while self._pending_flush: - pending_req, _ = self._pending_flush.popleft() - self.send_to_tokenizer.send_output( - FlushCacheReqOutput(success=success), pending_req - ) + self._pending_flush = None + self.send_to_tokenizer.send_output( + FlushCacheReqOutput(success=success), pending_req + ) return - self._expire_timed_out_pending_flushes(time.monotonic()) - - def _expire_timed_out_pending_flushes(self, now: float): - remaining: Deque[Tuple[FlushCacheReqInput, float]] = deque() - while self._pending_flush: - pending_req, deadline = self._pending_flush.popleft() - if now >= deadline: - logging.warning( - "Deferred flush_cache timed out while waiting for idle state." - ) - self.send_to_tokenizer.send_output( - FlushCacheReqOutput(success=False), pending_req - ) - else: - remaining.append((pending_req, deadline)) - self._pending_flush = remaining + if time.monotonic() >= deadline: + logging.warning( + "Deferred flush_cache timed out while waiting for idle state." + ) + self._pending_flush = None + self.send_to_tokenizer.send_output( + FlushCacheReqOutput( + success=False, message="Timed out waiting for idle state." + ), + pending_req, + ) def flush_cache_wrapped( self, recv_req: FlushCacheReqInput ) -> Optional[FlushCacheReqOutput]: + if self._pending_flush is not None: + return FlushCacheReqOutput( + success=False, + message="Another flush_cache is already in progress.", + ) + timeout_s = float(recv_req.timeout_s or 0.0) if timeout_s <= 0.0: return FlushCacheReqOutput(success=self.flush_cache()) @@ -2824,7 +2826,7 @@ class Scheduler( if self.is_fully_idle(): return FlushCacheReqOutput(success=self.flush_cache()) - self._pending_flush.append((recv_req, time.monotonic() + timeout_s)) + self._pending_flush = (recv_req, time.monotonic() + timeout_s) return None def clear_hicache_storage_wrapped(self, recv_req: ClearHiCacheReqInput): diff --git a/python/sglang/test/test_utils.py b/python/sglang/test/test_utils.py index 54efe47db..44f4e010a 100644 --- a/python/sglang/test/test_utils.py +++ b/python/sglang/test/test_utils.py @@ -168,31 +168,6 @@ def download_image_with_retry(image_url: str, max_retries: int = 3) -> Image.Ima time.sleep(2**i) -def flush_cache_with_retry( - base_url: str, - timeout: float = 30.0, - poll_interval: float = 0.5, -) -> bool: - """Flush device cache, polling until success or timeout. - - flush_cache only succeeds when the scheduler is fully idle, but - HiCache async ops (write-through, backup) may still be in-flight - after a request completes. We poll with a short interval so idle - is detected quickly, while the generous timeout accommodates slow - CI environments. - """ - deadline = time.time() + timeout - while time.time() < deadline: - try: - response = requests.post(f"{base_url}/flush_cache", timeout=10) - if response.status_code == 200: - return True - except requests.RequestException: - pass - time.sleep(poll_interval) - return False - - def is_in_ci(): """Return whether it is in CI runner.""" return get_bool_env_var("SGLANG_IS_IN_CI") diff --git a/test/manual/hicache/test_disaggregation_hicache.py b/test/manual/hicache/test_disaggregation_hicache.py index 42ad6609f..fe340e6f4 100644 --- a/test/manual/hicache/test_disaggregation_hicache.py +++ b/test/manual/hicache/test_disaggregation_hicache.py @@ -1,7 +1,6 @@ import os import random import tempfile -import time import unittest from typing import Dict @@ -14,7 +13,6 @@ from sglang.test.server_fixtures.disaggregation_fixture import ( from sglang.test.test_utils import ( DEFAULT_MODEL_NAME_FOR_TEST, DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH, - flush_cache_with_retry, popen_launch_pd_server, ) @@ -116,8 +114,12 @@ class DisaggregationHiCacheBase(PDDisaggregationServerBase): self.send_request(self.gen_prompt(1), max_tokens=150) # Flush device cache to force remote storage access. - time.sleep(2) - flush_cache_with_retry(self.prefill_url) + res = requests.post( + f"{self.prefill_url}/flush_cache", + params={"timeout": 30}, + timeout=40, + ) + res.raise_for_status() class TestDisaggregationPrefillWithHiCache(DisaggregationHiCacheBase): diff --git a/test/manual/hicache/test_pp_with_hicache.py b/test/manual/hicache/test_pp_with_hicache.py index 0ed3bdeac..761c0fe46 100644 --- a/test/manual/hicache/test_pp_with_hicache.py +++ b/test/manual/hicache/test_pp_with_hicache.py @@ -18,7 +18,6 @@ from sglang.test.test_utils import ( DEFAULT_MODEL_NAME_FOR_TEST, DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH, find_available_port, - flush_cache_with_retry, popen_launch_server, ) @@ -180,8 +179,13 @@ class TestPPWithHiCache(unittest.TestCase): return False - def flush_cache(self) -> bool: - return flush_cache_with_retry(self.base_url) + def flush_cache(self): + res = requests.post( + f"{self.base_url}/flush_cache", + params={"timeout": 30}, + timeout=40, + ) + res.raise_for_status() def test_eval_accuracy(self): args = SimpleNamespace( @@ -197,8 +201,7 @@ class TestPPWithHiCache(unittest.TestCase): metrics_initial = run_eval_few_shot_gsm8k(args) self.assertGreater(metrics_initial["accuracy"], 0.6) - self.assertTrue(self.flush_cache()) - time.sleep(2) + self.flush_cache() metrics_cached = run_eval_few_shot_gsm8k(args) self.assertGreater(metrics_cached["accuracy"], 0.6) diff --git a/test/registered/4-gpu-models/test_qwen35_hicache.py b/test/registered/4-gpu-models/test_qwen35_hicache.py index 90011b401..fedf461d5 100644 --- a/test/registered/4-gpu-models/test_qwen35_hicache.py +++ b/test/registered/4-gpu-models/test_qwen35_hicache.py @@ -1,6 +1,5 @@ import shutil import tempfile -import time import unittest from types import SimpleNamespace @@ -107,8 +106,12 @@ class TestQwen35WithHiCache(CustomTestCase): ) print(f"flush cache") - time.sleep(2) - requests.post(f"{self.base_url}/flush_cache", timeout=10) + res = requests.post( + f"{self.base_url}/flush_cache", + params={"timeout": 30}, + timeout=40, + ) + res.raise_for_status() second_metrics = self._run_gsm8k() print(f"second_metrics={second_metrics}") diff --git a/test/registered/hicache/test_hicache_storage_file_backend.py b/test/registered/hicache/test_hicache_storage_file_backend.py index afdce2cd4..554dd5b6f 100644 --- a/test/registered/hicache/test_hicache_storage_file_backend.py +++ b/test/registered/hicache/test_hicache_storage_file_backend.py @@ -26,7 +26,6 @@ from sglang.test.test_utils import ( DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH, DEFAULT_URL_FOR_TEST, CustomTestCase, - flush_cache_with_retry, is_in_ci, popen_launch_server, ) @@ -170,9 +169,14 @@ class HiCacheStorageBaseMixin: meta = response_json.get("meta_info", {}) return int(meta.get("cached_tokens", 0)) - def flush_cache(self) -> bool: + def flush_cache(self): """Flush device cache to force remote storage access.""" - return flush_cache_with_retry(self.base_url) + res = requests.post( + f"{self.base_url}/flush_cache", + params={"timeout": 30}, + timeout=40, + ) + res.raise_for_status() def gen_prompt(self, token_num: int) -> str: """Generate a random prompt of specified token length using tokenizer vocabulary.""" @@ -186,8 +190,7 @@ class HiCacheStorageBaseMixin: self.send_request(self.gen_prompt(1), max_tokens=150) # Flush device cache to force remote storage access - time.sleep(2) - self.assertTrue(self.flush_cache(), "Cache flush should succeed") + self.flush_cache() def test_basic_backup_and_prefetch(self): """Test storage and retrieval of large context through remote cache""" @@ -304,8 +307,7 @@ def run_eval_accuracy_test(test_instance, accuracy_threshold: float = 0.03): # Flush cache to force remote storage access print("Phase 2: Flushing device cache...") - test_instance.assertTrue(test_instance.flush_cache(), "Cache flush should succeed") - time.sleep(2) + test_instance.flush_cache() # Second evaluation - should use remote cache print("Phase 3: Running second GSM8K evaluation using remote cache...") diff --git a/test/registered/unit/managers/test_scheduler_flush_cache.py b/test/registered/unit/managers/test_scheduler_flush_cache.py index 133716db8..76c610ea4 100644 --- a/test/registered/unit/managers/test_scheduler_flush_cache.py +++ b/test/registered/unit/managers/test_scheduler_flush_cache.py @@ -1,8 +1,7 @@ import unittest -from collections import deque from unittest.mock import MagicMock, patch -from sglang.srt.managers.io_struct import FlushCacheReqInput, FlushCacheReqOutput +from sglang.srt.managers.io_struct import FlushCacheReqInput from sglang.srt.managers.scheduler import Scheduler from sglang.test.ci.ci_register import register_cpu_ci @@ -12,13 +11,14 @@ register_cpu_ci(est_time=2, suite="stage-a-cpu-only") class TestSchedulerFlushCache(unittest.TestCase): def _new_scheduler(self) -> Scheduler: scheduler = Scheduler.__new__(Scheduler) - scheduler._pending_flush = deque() + scheduler._pending_flush = None scheduler.send_to_tokenizer = MagicMock() scheduler.flush_cache = MagicMock(return_value=True) scheduler.is_fully_idle = MagicMock(return_value=False) return scheduler - def test_flush_cache_wrapped_non_positive_timeout_immediate(self): + def test_immediate_flush_no_timeout(self): + """No timeout → flush immediately regardless of idle state.""" scheduler = self._new_scheduler() scheduler.flush_cache.return_value = False @@ -26,12 +26,11 @@ class TestSchedulerFlushCache(unittest.TestCase): scheduler, FlushCacheReqInput(timeout_s=None) ) - self.assertIsInstance(output, FlushCacheReqOutput) self.assertFalse(output.success) scheduler.flush_cache.assert_called_once() - self.assertEqual(len(scheduler._pending_flush), 0) - def test_flush_cache_wrapped_positive_timeout_idle_immediate(self): + def test_immediate_flush_when_idle(self): + """Positive timeout but already idle → flush immediately.""" scheduler = self._new_scheduler() scheduler.is_fully_idle.return_value = True @@ -39,12 +38,11 @@ class TestSchedulerFlushCache(unittest.TestCase): scheduler, FlushCacheReqInput(timeout_s=5.0) ) - self.assertIsInstance(output, FlushCacheReqOutput) self.assertTrue(output.success) scheduler.flush_cache.assert_called_once() - self.assertEqual(len(scheduler._pending_flush), 0) - def test_flush_cache_wrapped_positive_timeout_busy_enqueues(self): + def test_defers_when_busy(self): + """Positive timeout + busy → defers, returns None.""" scheduler = self._new_scheduler() req = FlushCacheReqInput(timeout_s=3.0) @@ -52,52 +50,60 @@ class TestSchedulerFlushCache(unittest.TestCase): output = Scheduler.flush_cache_wrapped(scheduler, req) self.assertIsNone(output) - self.assertEqual(len(scheduler._pending_flush), 1) - pending_req, deadline = scheduler._pending_flush[0] + pending_req, deadline = scheduler._pending_flush self.assertIs(pending_req, req) self.assertEqual(deadline, 13.0) + + def test_rejects_when_already_pending(self): + """Any new request is rejected while another is pending.""" + scheduler = self._new_scheduler() + scheduler._pending_flush = (FlushCacheReqInput(timeout_s=10.0), 999.0) + + for timeout in [None, 5.0]: + output = Scheduler.flush_cache_wrapped( + scheduler, FlushCacheReqInput(timeout_s=timeout) + ) + self.assertFalse(output.success) + self.assertIn("already in progress", output.message) + scheduler.flush_cache.assert_not_called() - def test_check_pending_flush_idle_flushes_all(self): + def test_pending_flush_completes_on_idle(self): scheduler = self._new_scheduler() scheduler.is_fully_idle.return_value = True - - req1 = FlushCacheReqInput(timeout_s=1.0) - req2 = FlushCacheReqInput(timeout_s=2.0) - scheduler._pending_flush = deque([(req1, 111.0), (req2, 222.0)]) + req = FlushCacheReqInput(timeout_s=1.0) + scheduler._pending_flush = (req, 111.0) Scheduler._check_pending_flush(scheduler) + self.assertIsNone(scheduler._pending_flush) scheduler.flush_cache.assert_called_once() - self.assertEqual(len(scheduler._pending_flush), 0) - calls = scheduler.send_to_tokenizer.send_output.call_args_list - self.assertEqual(len(calls), 2) - self.assertTrue(calls[0].args[0].success) - self.assertIs(calls[0].args[1], req1) - self.assertTrue(calls[1].args[0].success) - self.assertIs(calls[1].args[1], req2) + out = scheduler.send_to_tokenizer.send_output.call_args.args[0] + self.assertTrue(out.success) - def test_check_pending_flush_busy_expires_only_timed_out(self): + def test_pending_flush_expires_on_timeout(self): scheduler = self._new_scheduler() - scheduler.is_fully_idle.return_value = False - - expired_req = FlushCacheReqInput(timeout_s=1.0) - alive_req = FlushCacheReqInput(timeout_s=5.0) - scheduler._pending_flush = deque([(expired_req, 99.0), (alive_req, 101.0)]) + req = FlushCacheReqInput(timeout_s=1.0) + scheduler._pending_flush = (req, 99.0) with patch("sglang.srt.managers.scheduler.time.monotonic", return_value=100.0): Scheduler._check_pending_flush(scheduler) + self.assertIsNone(scheduler._pending_flush) scheduler.flush_cache.assert_not_called() - self.assertEqual(len(scheduler._pending_flush), 1) - pending_req, deadline = scheduler._pending_flush[0] - self.assertIs(pending_req, alive_req) - self.assertEqual(deadline, 101.0) + out = scheduler.send_to_tokenizer.send_output.call_args.args[0] + self.assertFalse(out.success) - calls = scheduler.send_to_tokenizer.send_output.call_args_list - self.assertEqual(len(calls), 1) - self.assertFalse(calls[0].args[0].success) - self.assertIs(calls[0].args[1], expired_req) + def test_pending_flush_survives_before_deadline(self): + scheduler = self._new_scheduler() + req = FlushCacheReqInput(timeout_s=5.0) + scheduler._pending_flush = (req, 101.0) + + with patch("sglang.srt.managers.scheduler.time.monotonic", return_value=100.0): + Scheduler._check_pending_flush(scheduler) + + self.assertIsNotNone(scheduler._pending_flush) + scheduler.send_to_tokenizer.send_output.assert_not_called() if __name__ == "__main__":