sgl-project--sglang/python/sglang/srt/kv_canary/perturb/manager.py

from __future__ import annotations

from collections.abc import Callable
from typing import TYPE_CHECKING, Optional

from sglang.srt.kv_canary.buffer_group import CanaryBufferGroup
from sglang.srt.kv_canary.perturb import (
    real_kv_post_forward,
    real_kv_unused_cache,
    real_kv_used,
    req_to_token,
)
from sglang.srt.kv_canary.perturb.config import PerturbConfig
from sglang.srt.kv_canary.perturb.utils import WarmupGate

if TYPE_CHECKING:
    from sglang.srt.mem_cache.base_prefix_cache import BasePrefixCache
    from sglang.srt.mem_cache.memory_pool import ReqToTokenPool
    from sglang.srt.model_executor.forward_batch_info import ForwardBatch


class PerturbManager:
    def __init__(
        self,
        *,
        config: PerturbConfig,
        req_to_token_pool: ReqToTokenPool,
        buffer_groups: tuple[CanaryBufferGroup, ...],
        outer_step_counter_getter: Callable[[], int],
        swa_window_size: int = 0,
        sweep_interval: int = 0,
    ) -> None:
        self._config = config
        self._req_to_token_pool = req_to_token_pool
        self._buffer_groups = buffer_groups
        self._outer_step_counter_getter = outer_step_counter_getter
        self._swa_window_size = swa_window_size
        self._sweep_interval = sweep_interval
        self._radix_cache: Optional[BasePrefixCache] = None
        self._warmup_gate = WarmupGate(
            config=config, outer_step_counter_getter=outer_step_counter_getter
        )

    def attach_radix_cache(self, radix_cache: BasePrefixCache) -> None:
        self._radix_cache = radix_cache

    def perturb(
        self,
        *,
        maybe_inaccurate_forward_batch: Optional[ForwardBatch],
    ) -> None:
        self.perturb_req_to_token(maybe_inaccurate_forward_batch)
        self.perturb_real_kv_used(maybe_inaccurate_forward_batch)
        self.perturb_real_kv_unused_cache(maybe_inaccurate_forward_batch)

    def perturb_post_forward(
        self,
        *,
        maybe_inaccurate_forward_batch: Optional[ForwardBatch],
    ) -> None:
        self.perturb_real_kv_post_forward(maybe_inaccurate_forward_batch)

    def perturb_req_to_token(
        self, maybe_inaccurate_forward_batch: Optional[ForwardBatch]
    ) -> None:
        req_to_token.run(
            maybe_inaccurate_forward_batch=maybe_inaccurate_forward_batch,
            config=self._config,
            req_to_token_pool=self._req_to_token_pool,
            warmup_gate=self._warmup_gate,
        )

    def perturb_real_kv_used(
        self, maybe_inaccurate_forward_batch: Optional[ForwardBatch]
    ) -> None:
        real_kv_used.run(
            maybe_inaccurate_forward_batch=maybe_inaccurate_forward_batch,
            config=self._config,
            req_to_token_pool=self._req_to_token_pool,
            buffer_groups=self._buffer_groups,
            swa_window_size=self._swa_window_size,
            warmup_gate=self._warmup_gate,
        )

    def perturb_real_kv_unused_cache(
        self, maybe_inaccurate_forward_batch: Optional[ForwardBatch]
    ) -> None:
        real_kv_unused_cache.run(
            maybe_inaccurate_forward_batch=maybe_inaccurate_forward_batch,
            config=self._config,
            buffer_groups=self._buffer_groups,
            radix_cache=self._radix_cache,
            swa_window_size=self._swa_window_size,
            sweep_interval=self._sweep_interval,
            outer_step_counter=self._outer_step_counter_getter(),
            warmup_gate=self._warmup_gate,
        )

    def perturb_real_kv_post_forward(
        self, maybe_inaccurate_forward_batch: Optional[ForwardBatch]
    ) -> None:
        real_kv_post_forward.run(
            maybe_inaccurate_forward_batch=maybe_inaccurate_forward_batch,
            config=self._config,
            buffer_groups=self._buffer_groups,
            warmup_gate=self._warmup_gate,
        )