[HiCache] Yield idle scheduler so storage workers can drain (#38504)

python/sglang/srt/managers/scheduler.py

@@ -4767,7 +4767,7 @@ class Scheduler(
         # (queues parked under KV pressure / disagg transfer) has no
         # process_batch_result to publish the growing gauge, and gating here
         # froze /get_loads, DP balancing, and the LoadStat for the stall. This
-        # path spins without sleeping, so a wall-clock floor bounds the
+        # path is polled repeatedly, so a wall-clock floor bounds the
         # O(queue) get_loads for both sinks; the fully-idle publish runs
         # post-flush below.
         fully_idle = self.is_fully_idle()
@@ -4780,6 +4780,10 @@ class Scheduler(
                 self.load_publisher.publish_load_stat(
                     self.load_inquirer.get_loads, force=True, snapshot=snapshot
                 )
+            if self.enable_hicache_storage:
+                # Storage workers need the GIL between I/O calls. Yield while
+                # there is no GPU batch so polling cannot starve their acks.
+                time.sleep(0)
             return
         self.metrics_reporter.record_scheduler_idle()
 

test/registered/unit/managers/test_scheduler_on_idle_load.py

@@ -1,6 +1,6 @@
 """on_idle's stalled-path load publish is wall-clock bounded.
 
-A no-batch-but-not-idle stall spins on_idle without sleeping, so the gate must
+A non-storage no-batch stall spins on_idle without sleeping, so the gate must
 cap the O(queue) get_loads for both the DP-balancing writer and the load
 socket. CPU-only: builds a bare Scheduler with mocked collaborators, like
 test_scheduler_flush_cache.
@@ -23,6 +23,7 @@ class TestOnIdleStallPublish(CustomTestCase):
     def _stalled_scheduler(self) -> Scheduler:
         s = Scheduler.__new__(Scheduler)
         s.scheduler_stage_metrics = None
+        s.enable_hicache_storage = False
         s.maybe_send_health_check_signal = MagicMock()
         s.is_fully_idle = MagicMock(return_value=False)  # stalled, not idle
         s.publish_load_snapshot = MagicMock(return_value=None)