[HiCache] Run the host->device load-back enqueue on a loader thread (opt-in)

[qidaye]

Motivation

Closes #38470

HiCacheController.start_loading submits the whole host->device load-back burst inline on the scheduler thread. With --hicache-io-backend direct and the page_first_direct layout every per-layer load_to_device_per_layer builds a batched-copy descriptor on the CPU (~2.4 ms per layer in our profile), so one burst on a 78-layer model with two host pools keeps the scheduler thread busy for ~375 ms. By the time the batch is built and the forward launches, every copy has already landed and the layer-wise pipeline (LayerDoneCounter.wait_until in the attention layers) overlaps ~0.04 % of the copy time.

This PR adds SGLANG_HICACHE_ASYNC_LOAD_ENQUEUE (default off). With it on, start_loading only does the producer-slot hand-out and start_event.record() on the scheduler thread, then hands the index move, the per-layer enqueue and the ack to a single FIFO daemon thread (hicache-load-enqueue). With it off the call sequence is byte-for-byte the previous one (the unit test asserts the exact stream/event sequence).

Modifications

python/sglang/srt/managers/cache_controller.py

• LayerLoadingEvent(num_layers, async_enqueue=False): per-layer CPU-side recorded flags. Waiting on a device event that was never recorded is a no-op, so wait() first blocks on the flag (logging every 30 s if it is still unset), then waits on the device event. A per-slot enqueue_done flag is the rotation's backpressure: update_producer waits on it before checking finish_event, because while a burst is still queued the slot's finish_event is the previous rotation's and query() would pass. rearm(), is_enqueue_done(), mark_enqueue_done(), disable_async_handshake().
• HiCacheController: _init_load_enqueue_thread, _load_enqueue_thread_func (single consumer; the H2D stream ordering, the 3-slot rotation and ack_load_queue popped in order are all FIFO), _abort_load_burst, _drain_load_enqueue, stop_load_enqueue_thread (joins, then falls back to the inline path and drops the handshake). Startup logs HiCache async load enqueue: enabled/disabled (<ClassName>).
• start_loading split into the scheduler-thread half and _enqueue_load_burst(op, producer_event, start_event_recorded, fence_event):
  • inline: index move, start_event.record(), fence, submit, ack, exactly as before;
  • loader thread: op.device_indices is written by the allocator on the compute stream, so the index move runs under the H2D stream after start_event.wait(h2d_stream); the overlap-scheduling fence is captured on the scheduler thread as an event recorded on load_fence_stream at hand-over time and waited on by the loader before submitting.
• Loader failure: every layer event is completed on the H2D stream and a zero-token ack is published, so a forward parked on wait_until and the later acks behind this one cannot hang. The error is logged with the node ids and latched in load_enqueue_error; the KV for those nodes is stale.
• reset() drains the loader first, and _append_load_ack takes _ack_load_lock so a loader-thread append cannot land after reset()'s clear().

python/sglang/srt/mem_cache/hybrid_cache/hybrid_cache_controller.py: the replacement LayerDoneCounter built for transfer_layer_id_max carries async_enqueue.

python/sglang/srt/mem_cache/unified_radix_cache.py: is_load_back_event_done returns False while the slot's burst is still queued (the getattr keeps the FlexKV counter, which has its own event class, untouched).

python/sglang/srt/environ.py: SGLANG_HICACHE_ASYNC_LOAD_ENQUEUE = EnvBool(False).

Tests

• New test/registered/unit/mem_cache/test_hicache_async_load_enqueue.py (CPU, base-a-test-cpu): swaps the device_module of the controller and of L2TransferEngine for a fake that logs the exact stream/event call sequence with the calling thread. Covers: gate off creates no thread and keeps the pre-split call sequence (including the fence wait_stream); producer hand-out is synchronous and the enqueue is not; loader binds to the scheduler's device; ack only after the last layer; every layer completes so a consumer cannot hang; bursts complete in submission order; slot reuse blocks on a queued burst; reset() does not resurrect an ack; the fence event is recorded at hand-over and waited on the loader; the index move runs under the H2D stream after the start_event wait (with a sentinel that returns garbage if read early); loader failure logs, releases the consumer and still acks, and the next burst is served; shutdown joins and falls back to inline; is_load_back_event_done guard; hybrid burst with a sidecar pool inline and on the loader thread; activation log line and transfer_layer_id_max counter carry-over.
• test_hicache_staged_write_back_dispatch.py::test_hybrid_load_forwards_merged_pool_transfers builds its controller with mock.Mock(spec=...); it now wires the inline burst body so the split does not turn it into a no-op.

Accuracy Tests

No numerical change: the same copies are issued on the same stream, only the thread that issues them changes. Gate off is the existing path.

Speed Tests and Profiling

Prefill-only H100 TP=8 deployment, hierarchical cache, io_backend=direct, page_first_direct, 500 requests, same prompt set, gate off vs on:

metric	change
throughput	+8.8 %
wall time	-8.1 %
TTFT mean	-7.9 %
TTFT p50	-10.0 %
TTFT p95	22.12 s -> 20.41 s
TTFT p99	flat

Per-burst start_loading on the scheduler thread drops from ~375 ms to the slot hand-out.

Checklist

• Format your code according to the Code Formatting with Pre-Commit.
• Add unit tests as outlined in the Running Unit Tests.
• Update documentation / docstrings / example tutorials as needed, according to Writing Documentation.
• Provide throughput / latency benchmark results and accuracy evaluation results as needed, according to Benchmark and Profiling and Accuracy Results.
• For reviewers: If you haven't made any contributions to this PR and are only assisting with merging the main branch, please remove yourself as a co-author when merging the PR.
• Please feel free to join our Slack channel at https://slack.sglang.io to discuss your PR.

---

CI States

Latest PR Test (Base): ❌ Run #34208615291
Latest PR Test (Extra): ❌ Run #34208615111
Latest PR Test (AMD ROCm 7.2): ❌ Run #34208615213