feat(hicache): add KVCR L3 storage with router-guided peer restore

docs/docs/advanced_features/server_arguments.mdx

@@ -780,6 +780,12 @@ Please consult the documentation below and [server_args.py](https://github.com/s
       <td style={{padding: "9px 12px", backgroundColor: "rgba(255,255,255,0.02)"}}>`False`</td>
       <td style={{padding: "9px 12px", backgroundColor: "rgba(255,255,255,0.05)"}}>bool flag (set to enable)</td>
     </tr>
+    <tr>
+      <td style={{padding: "9px 12px", fontWeight: 500, backgroundColor: "rgba(255,255,255,0.02)"}}>`--python-thread-switch-interval`</td>
+      <td style={{padding: "9px 12px", backgroundColor: "rgba(255,255,255,0.05)"}}>Optional positive, finite Python thread switch interval in seconds for each spawned scheduler process. Affects all Python threads in that process. Unset preserves interpreter defaults; CPU affinity and thread priority are unchanged.</td>
+      <td style={{padding: "9px 12px", backgroundColor: "rgba(255,255,255,0.02)"}}>`None`</td>
+      <td style={{padding: "9px 12px", backgroundColor: "rgba(255,255,255,0.05)"}}>Type: float</td>
+    </tr>
     <tr>
       <td style={{padding: "9px 12px", fontWeight: 500, backgroundColor: "rgba(255,255,255,0.02)"}}>`--batch-notify-size`</td>
       <td style={{padding: "9px 12px", backgroundColor: "rgba(255,255,255,0.05)"}}>Number of streaming notifications to batch before yielding to the event loop. Reduces asyncio wakeup overhead under high concurrency.</td>

python/sglang/srt/arg_groups/field_order.py

@@ -521,5 +521,18 @@
     "enable_decoder_swa_bounded_replay",
     "enable_response_store",
     "disaggregation_decode_host_receive_threshold",
+    # Freeze the previously appended fields before adding another namespace leaf.
+    "qsa_indexer_dtype",
+    "disaggregation_decode_allocation_policy",
+    "speculative_boundary_reduction",
+    "speculative_use_block_verification",
+    "speculative_draft_sink_size",
+    "disable_aiter_allreduce_fusion_in_prefill",
+    "disable_aiter_allreduce_fusion_in_decode",
+    "dsa_triton_union",
+    "enable_dsa_fused_indexer",
+    "enable_cp_tp_group_sharing",
+    "grpc_response_timeout_secs",
+    "python_thread_switch_interval",
 )
 # fmt: on

python/sglang/srt/arg_groups/fields/device.py

@@ -9,7 +9,6 @@
 )
 
 import msgspec
-
 from sglang.srt.arg_groups.arg_utils import A, Derived
 
 
@@ -72,6 +71,10 @@ class Device(msgspec.Struct):
         "Set soft watchdog timeout in seconds. If a forward batch takes longer than this, the server will dump information for debugging.",
     ] = None
     sleep_on_idle: A[bool, "Reduce CPU usage when sglang is idle."] = False
+    python_thread_switch_interval: A[
+        Optional[float],
+        "Optional positive, finite Python thread switch interval in seconds for each scheduler process. Affects all Python threads in that process; unset preserves interpreter defaults. Does not change CPU affinity or thread priority.",
+    ] = None
     use_ray: A[
         bool,
         "Use Ray actors for scheduler process management.",

python/sglang/srt/arg_groups/fields/memory.py

@@ -145,11 +145,12 @@ class Memory(msgspec.Struct):
     hicache_storage_backend: A[
         Optional[str],
         Arg(
-            help="The storage backend for hierarchical KV cache. Built-in backends: file, mooncake, npu_memcache, hf3fs, nixl, aibrix, tensorcast, seaweedfs. For dynamic backend, use --hicache-storage-backend-extra-config to specify: backend_name (custom name), module_path (Python module path), class_name (backend class name).",
+            help="The storage backend for hierarchical KV cache. Built-in backends: file, mooncake, kvcr, npu_memcache, hf3fs, nixl, aibrix, tensorcast, seaweedfs. For dynamic backend, use --hicache-storage-backend-extra-config to specify: backend_name (custom name), module_path (Python module path), class_name (backend class name).",
             choices=[
                 "file",
                 "sim",
                 "mooncake",
+                "kvcr",
                 "npu_memcache",
                 "hf3fs",
                 "nixl",

python/sglang/srt/arg_groups/hicache_hook.py

@@ -179,7 +179,7 @@ def resolve_layout_io_compatibility(server_args: Any):
 def resolve_storage_layout_compatibility(server_args: Any):
     cfg = resolving_view(server_args)
     if (
-        cfg.hicache_storage_backend not in ("mooncake", "npu_memcache")
+        cfg.hicache_storage_backend not in ("mooncake", "npu_memcache", "kvcr")
         or cfg.hicache_mem_layout != "layer_first"
     ):
         return

python/sglang/srt/arg_groups/pipeline.py

@@ -79,11 +79,13 @@ def run_resolution_pipeline(server_args: Any) -> None:
         default_unset_prefill_decode_interval,
         validate_experimental_sgl_marlin,
         validate_prefill_decode_interval,
+        validate_python_thread_scheduling,
         validate_response_store,
         validate_sampling_mask_max_tokens,
     )
 
     run_hook(validate_prefill_decode_interval, server_args)
+    run_hook(validate_python_thread_scheduling, server_args)
     run_hook(validate_response_store, server_args)
     run_hook(validate_sampling_mask_max_tokens, server_args)
 

python/sglang/srt/arg_groups/validation_hook.py

@@ -25,6 +25,16 @@
 logger = logging.getLogger(__name__)
 
 
+def validate_python_thread_scheduling(server_args: Any) -> None:
+    from sglang.srt.managers.scheduler_components.thread_scheduling import (
+        validate_python_thread_switch_interval,
+    )
+
+    validate_python_thread_switch_interval(
+        resolving_view(server_args).python_thread_switch_interval
+    )
+
+
 def validate_response_store(server_args: Any) -> None:
     cfg = resolving_view(server_args)
     if cfg.enable_response_store and cfg.disaggregation_mode != "null":

python/sglang/srt/disaggregation/decode_hicache_mixin.py

@@ -95,6 +95,7 @@ def _build_decode_prefix_match(
                     prefix_keys,
                     extra_key=req.extra_key,
                     cache_salt=req.cache_salt,
+                    kv_hints=req.kv_hints,
                 )
 
         return DecodePrefixMatch(
@@ -139,6 +140,7 @@ def _start_hicache_prefetch(
                 prefix_keys,
                 extra_key=req.extra_key,
                 cache_salt=req.cache_salt,
+                kv_hints=req.kv_hints,
             )
             prefix_match.prefetch_registered = self.tree_cache.has_ongoing_prefetch(
                 req.cache_request_handle

python/sglang/srt/managers/cache_controller.py

@@ -22,11 +22,11 @@
 from typing import TYPE_CHECKING, Callable, List, NamedTuple, Optional
 
 import torch
-
 from sglang.srt.mem_cache.hicache_storage import (
     STORAGE_BATCH_SIZE,
     HiCacheStorageConfig,
     HiCacheStorageExtraInfo,
+    PoolHitPolicy,
     PoolName,
     PoolTransfer,
     count_pool_hits,
@@ -234,13 +234,15 @@ def __init__(
         last_hash: Optional[str] = None,
         hash_value: Optional[List[str]] = None,
         prefix_keys: Optional[List[str]] = None,
+        kv_hints=None,
     ):
         self.host_indices = host_indices
         self.token_ids = token_ids
         self.last_hash = last_hash
         self.completed_tokens = 0
         self.hash_value = hash_value if hash_value is not None else []
         self.prefix_keys = prefix_keys
+        self.kv_hints = kv_hints
         # Full queried page-hash chain, set by _storage_hit_query before
         # hash_value is truncated to the hit boundary; the tail is the
         # absence signal that invalidates buffer-mode existence beliefs.
@@ -562,13 +564,6 @@ def attach_storage_backend(
         self.storage_config = self._generate_storage_config(
             model_name, storage_backend_extra_config
         )
-        # for MLA models, only one rank needs to backup the KV cache
-        self.backup_skip = (
-            self.storage_config.is_mla_model
-            # todo: load balancing
-            and self.storage_config.tp_rank != 0
-        )
-
         # Use storage backend factory for dynamic backend creation
         from sglang.srt.mem_cache.storage import StorageBackendFactory
 
@@ -577,6 +572,13 @@ def attach_storage_backend(
                 storage_backend, self.storage_config, self.storage_host_pool
             )
             self.storage_backend.register_mem_pool_host(self.storage_host_pool)
+            # Rank-local stores need every MLA writer. A backend that explicitly
+            # provides shared replicated storage can use the TP0-only path.
+            self.backup_skip = (
+                self.storage_config.is_mla_model
+                and self.storage_config.tp_rank != 0
+                and self.storage_backend.requires_rank_local_backup is not True
+            )
 
             self.enable_storage = True
             # todo: threshold policy for prefetching
@@ -608,6 +610,7 @@ def attach_storage_backend(
                 in [
                     "hf3fs",
                     "mooncake",
+                    "kvcr",
                     "npu_memcache",
                     "eic",
                     "nixl",
@@ -1094,6 +1097,7 @@ def _generic_page_get(
 
     def _page_transfer(self, operation: PrefetchOperation) -> int:
         # Transfer batch by batch
+        batch_pages = self.storage_backend.prefetch_batch_pages
         prefix_keys = operation.prefix_keys
         kv_derived_transfers = [
             transfer
@@ -1102,20 +1106,23 @@ def _page_transfer(self, operation: PrefetchOperation) -> int:
         ]
         all_success = True
         completed_pages = 0
-        for i in range(0, len(operation.hash_value), STORAGE_BATCH_SIZE):
+        for i in range(0, len(operation.hash_value), batch_pages):
             # When an error is occurred, we should keep looping and produce the same number of
             # PrefetchAck as other ranks do, because prefetch_sync_thread (i.e. consumer of
             # prefetch_sync_queue) perform reduce on the results.  This is so tricky.
             if all_success and operation.is_terminated():
                 all_success = False
             if all_success:
-                batch_hashes = operation.hash_value[i : i + STORAGE_BATCH_SIZE]
+                batch_hashes = operation.hash_value[i : i + batch_pages]
                 batch_host_indices = operation.host_indices[
                     i * self.page_size : (i + len(batch_hashes)) * self.page_size
                 ]
 
                 # Get one batch token, and update the completed_tokens if succeed
-                extra_info = HiCacheStorageExtraInfo(prefix_keys=prefix_keys)
+                extra_info = HiCacheStorageExtraInfo(
+                    prefix_keys=prefix_keys,
+                    extra_info={"kv_hints": operation.kv_hints},
+                )
 
                 try:
                     hit_pages = self._page_transfer_kv_batch(
@@ -1162,6 +1169,55 @@ def _page_transfer_kv_batch(
 
         Here, "batch" means a single unit of L3 read, not a "batch" in model forward.
         """
+        if (
+            kv_derived_transfers
+            and getattr(self.storage_backend, "supports_combined_page_reads", False)
+            is True
+            and all(
+                transfer.hit_policy == PoolHitPolicy.ALL_PAGES
+                for transfer in kv_derived_transfers
+            )
+        ):
+            transfers = [
+                PoolTransfer(
+                    name=name, host_indices=batch_host_indices, keys=batch_hashes
+                )
+                for name in [PoolName.KV, *(t.name for t in kv_derived_transfers)]
+            ]
+            required_names = {str(transfer.name) for transfer in transfers}
+            if len(required_names) != len(transfers):
+                logger.error(
+                    "Duplicate combined storage pools for request %s",
+                    operation.request_id,
+                )
+                return 0
+            results = self.storage_backend.batch_get_v2(
+                transfers, extra_info=extra_info
+            )
+            # Missing or truncated sidecar results must not silently drop a
+            # required pool from the prefix intersection. A malformed response
+            # is a miss; keep producing the normal number of cross-rank acks.
+            if not isinstance(results, dict) or any(
+                name not in results
+                or not isinstance(results[name], list)
+                or len(results[name]) != len(batch_hashes)
+                or any(type(value) is not bool for value in results[name])
+                for name in required_names
+            ):
+                logger.error(
+                    "Malformed combined storage result for request %s",
+                    operation.request_id,
+                )
+                return 0
+            hits = count_pool_hits({name: results[name] for name in required_names})
+            completed = min(hits.values())
+            if completed < len(batch_hashes):
+                logger.warning(
+                    "Prefetch operation %s failed to retrieve page %s.",
+                    operation.request_id,
+                    batch_hashes[completed],
+                )
+            return completed
         # Read from KV pool.
         kv_hits = self.page_get_func(
             operation, batch_hashes, batch_host_indices, extra_info
@@ -1258,9 +1314,12 @@ def _storage_hit_query(self, operation) -> tuple[list[str], int]:
         )
         operation.all_hash_values = page_hashes
 
-        for start in range(0, len(page_hashes), STORAGE_BATCH_SIZE):
-            batch_hashes = page_hashes[start : start + STORAGE_BATCH_SIZE]
-            extra_info = HiCacheStorageExtraInfo(prefix_keys=prefix_keys)
+        batch_pages = self.storage_backend.query_batch_pages
+        for start in range(0, len(page_hashes), batch_pages):
+            batch_hashes = page_hashes[start : start + batch_pages]
+            extra_info = HiCacheStorageExtraInfo(
+                prefix_keys=prefix_keys, extra_info={"kv_hints": operation.kv_hints}
+            )
             hit_page_num = self.storage_backend.batch_exists(batch_hashes, extra_info)
             hash_value.extend(batch_hashes[:hit_page_num])
             storage_query_count += hit_page_num * self.page_size

python/sglang/srt/managers/scheduler.py

@@ -276,6 +276,9 @@
 from sglang.srt.managers.scheduler_components.request_receiver import (
     SchedulerRequestReceiver,
 )
+from sglang.srt.managers.scheduler_components.thread_scheduling import (
+    configure_python_thread_scheduling,
+)
 from sglang.srt.managers.scheduler_components.weight_updater import (
     SchedulerWeightUpdaterManager,
 )
@@ -3185,6 +3188,7 @@ def _prefetch_kvcache(self, req: Req, storage_hit_end: Optional[int] = None):
                     extra_key=req.extra_key,
                     cache_salt=req.cache_salt,
                     storage_hit_end=storage_hit_end,
+                    kv_hints=req.kv_hints,
                 )
 
     def _process_storage_prefetch_retries(self):
@@ -5786,6 +5790,13 @@ def maybe_sleep_on_idle(self):
     def handle_freeze_gc(self, recv_req: FreezeGCReq):
         """Handle freeze_gc request: freeze scheduler's GC and forward to detokenizer."""
         freeze_gc("Scheduler")
+        logger.info(
+            "GC freeze completed: pid=%d tp=%d cp=%d pp=%d",
+            os.getpid(),
+            get_parallel().tp_rank,
+            get_parallel().attn_cp_rank,
+            get_parallel().pp_rank,
+        )
         self.ipc_channels.send_to_detokenizer.send_output(recv_req, recv_req)
         return None
 
@@ -5919,6 +5930,7 @@ def configure_scheduler_process(
     # Configure the logger
     configure_logger(server_args, prefix=prefix)
     suppress_other_loggers()
+    configure_python_thread_scheduling(get_device().python_thread_switch_interval)
 
     # Set cpu affinity to this gpu process
     if envs.SGLANG_SET_CPU_AFFINITY.get():

python/sglang/srt/managers/scheduler_components/thread_scheduling.py

@@ -0,0 +1,39 @@
+"""Explicit, process-local Python thread scheduling for scheduler workers."""
+
+import logging
+import math
+import os
+import sys
+from typing import Optional
+
+logger = logging.getLogger(__name__)
+
+
+def validate_python_thread_switch_interval(value: Optional[float]) -> None:
+    if value is not None and (
+        isinstance(value, bool)
+        or not isinstance(value, (int, float))
+        or not math.isfinite(value)
+        or value <= 0
+    ):
+        raise ValueError("--python-thread-switch-interval must be positive and finite")
+
+
+def configure_python_thread_scheduling(interval: Optional[float]) -> None:
+    """Leave interpreter defaults untouched unless explicitly requested.
+
+    Call in the spawned scheduler process, after runtime-context publication;
+    setting a parent interpreter does not configure spawned children. This is
+    process-wide, not a thread priority or a CPU-affinity control.
+    """
+    validate_python_thread_switch_interval(interval)
+    if interval is None:
+        return
+    sys.setswitchinterval(interval)
+    logger.info(
+        "Python scheduler thread switch interval pid=%d requested_seconds=%.9g "
+        "actual_seconds=%.9g",
+        os.getpid(),
+        interval,
+        sys.getswitchinterval(),
+    )