[HiCache] Check in-flight async ops in is_fully_idle() before attach/detach (#20746)

python/sglang/srt/managers/scheduler.py

@@ -2674,8 +2674,8 @@ class Scheduler(
         idle &= len(self.waiting_queue) == 0
 
         if not for_health_check:
-            # Grammar queue and prefill inflight queue may not produce batch results
-            # instantly, but they still indicate the server is not fully idle.
+            # Grammar queue and prefill inflight queue may not produce batch
+            # results instantly, but they still indicate the server is not idle.
             idle &= len(self.grammar_manager.grammar_queue) == 0
             if self.disaggregation_mode == DisaggregationMode.PREFILL:
                 idle &= len(self.disagg_prefill_inflight_queue) == 0
@@ -2685,6 +2685,16 @@ class Scheduler(
                 idle &= len(self.disagg_decode_prealloc_queue.queue) == 0
                 idle &= len(self.disagg_decode_transfer_queue.queue) == 0
 
+            # HiCache: in-flight async ops (GPU↔Host↔L3) must drain before
+            # destructive operations like attach/detach/flush_cache.
+            if self.enable_hierarchical_cache:
+                tc = self.tree_cache
+                idle &= len(tc.ongoing_write_through) == 0
+                idle &= len(tc.ongoing_load_back) == 0
+                if tc.enable_storage:
+                    idle &= len(tc.ongoing_prefetch) == 0
+                    idle &= len(tc.ongoing_backup) == 0
+
         return idle
 
     def attach_hicache_storage_wrapped(

python/sglang/test/test_utils.py

@@ -167,6 +167,25 @@ 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, retries: int = 5, interval: float = 2.0
+) -> bool:
+    """Flush device cache with retry.
+
+    The scheduler may still have in-flight HiCache async ops (GPU↔Host↔L3)
+    that prevent is_fully_idle() from returning True, so we retry.
+    """
+    for _ in range(retries):
+        try:
+            response = requests.post(f"{base_url}/flush_cache", timeout=10)
+            if response.status_code == 200:
+                return True
+        except requests.RequestException:
+            pass
+        time.sleep(interval)
+    return False
+
+
 def is_in_ci():
     """Return whether it is in CI runner."""
     return get_bool_env_var("SGLANG_IS_IN_CI")