from __future__ import annotations from dataclasses import dataclass from typing import TYPE_CHECKING import torch if TYPE_CHECKING: from sglang.srt.model_executor.forward_batch_info import ForwardBatch @dataclass(frozen=True, slots=True, kw_only=True) class PlanInput: """Pre-staged input to launch_canary_plan_kernels for the per-forward path. All tensors live on device. Fields: req_pool_indices: Per-row ReqToTokenPool row index, shape [bs_capacity], int64. 0 = padding sentinel. prefix_lens: Per-req prefix length already written before this step, shape [bs_capacity], int64. Extend → extend_prefix_lens; decode → seq_lens - 1. extend_seq_lens: Per-req tokens being written this step, shape [bs_capacity], int64. Extend length or all-ones for decode. req_to_verify_expected_tokens_valid_lens: Per-req snapshot length on the verify-token pool, shape [bs_capacity], int64. Equals ``len(req.origin_input_ids) + len(req.output_ids)`` at the moment ``ForwardBatch`` was built. The plan kernel uses ``valid_lens[req_id]`` as the upper bound on ``sot_pos`` when gathering the expected token; everything past the snapshot (e.g. EAGLE draft / verify positions, or stale residue from a longer recycled slot owner) returns the ``-1`` sentinel and the verify kernel skips the check. Set only when ``CanaryConfig.enable_verify_token_assert`` is on. Allocated fresh per forward by :class:`SingleForwardManager`. The boundary ForwardBatch token/position/slot tensors must already be int64 contiguous (upstream phase-1 hook is responsible). """ req_pool_indices: torch.Tensor prefix_lens: torch.Tensor extend_seq_lens: torch.Tensor req_to_verify_expected_tokens_valid_lens: torch.Tensor def zero_(self) -> None: self.req_pool_indices.zero_() self.prefix_lens.zero_() self.extend_seq_lens.zero_() self.req_to_verify_expected_tokens_valid_lens.zero_() @classmethod def allocate( cls, *, bs_capacity: int, device: torch.device, ) -> PlanInput: return cls( req_pool_indices=torch.zeros(bs_capacity, dtype=torch.int64, device=device), prefix_lens=torch.zeros(bs_capacity, dtype=torch.int64, device=device), extend_seq_lens=torch.zeros(bs_capacity, dtype=torch.int64, device=device), req_to_verify_expected_tokens_valid_lens=torch.zeros( bs_capacity, dtype=torch.int64, device=device ), ) def fill_from_forward_batch(self, *, forward_batch: ForwardBatch) -> None: req_pool_indices = forward_batch.req_pool_indices bs = int(req_pool_indices.shape[0]) capacity = int(self.req_pool_indices.shape[0]) if bs > capacity: raise RuntimeError( f"kv-canary: per-forward batch size {bs} exceeds static capacity {capacity}; " "raise the buffer size in CanaryLaunchCapacities" ) self.zero_() self.req_pool_indices[:bs].copy_(req_pool_indices) _extract_prefix_lens_and_extend_seq_lens( forward_batch=forward_batch, out_prefix_lens=self.prefix_lens[:bs], out_extend_seq_lens=self.extend_seq_lens[:bs], bs=bs, ) req_all_ids_lens = forward_batch.req_all_ids_lens if req_all_ids_lens is not None: self.req_to_verify_expected_tokens_valid_lens[:bs].copy_( req_all_ids_lens.to(torch.int64), non_blocking=True ) def _extract_prefix_lens_and_extend_seq_lens( *, forward_batch: ForwardBatch, out_prefix_lens: torch.Tensor, out_extend_seq_lens: torch.Tensor, bs: int, ) -> None: # TODO: once ForwardMode is refactored upstream so every mode ships a canonical # (prefix_lens, extend_seq_lens) pair on forward_batch, collapse this back to a single # unconditional copy. forward_mode = forward_batch.forward_mode spec_info = forward_batch.spec_info if forward_mode.is_decode_or_idle(): # Anchor on ``positions`` (canonical write position) — eagle draft leaves seq_lens # pre-bump so deriving prefix_lens from seq_lens is off-by-one. Padding tail (positions # shorter than bs under cuda-graph padding) keeps whatever stale data it had; the offsets # kernel masks those rows via ``is_active`` before using prefix_lens. positions = forward_batch.positions out_prefix_lens[: positions.shape[0]].copy_(positions.to(torch.int64)) out_extend_seq_lens.fill_(1) elif forward_mode.is_target_verify(): # Evidence: EagleVerifyInputV2Mixin.prepare_for_verify assigns out_cache_loc in # [seq_lens, seq_lens + draft_token_num) without bumping seq_lens. The target-verify # branch in TRTLLMHAAttnBackend.init_forward_metadata uses seq_lens as the prefix and # tokens_per_req as the query length, so mirror that as seq_lens plus draft_token_num. out_prefix_lens.copy_(forward_batch.seq_lens[:bs].to(torch.int64)) out_extend_seq_lens.fill_(int(spec_info.draft_token_num)) elif forward_mode.is_draft_extend_v2(): # Evidence: EagleDraftWorkerBase.prepare_for_draft_extend bumps # seq_lens by num_draft_tokens. FlashAttentionBackend.init_forward_metadata reads the # draft-extend-v2 query length from spec_info.extend_seq_lens_tensor when available. # CUDA-graph replay passes extend_seq_lens but omits extend_prefix_lens, so derive the # prefix as seq_lens - extend_seq_lens. extend_seq_lens = forward_batch.extend_seq_lens[:bs].to(torch.int64) out_extend_seq_lens.copy_(extend_seq_lens) out_prefix_lens.copy_( forward_batch.seq_lens[:bs].to(torch.int64) - extend_seq_lens ) elif forward_mode.is_extend(): # Evidence: ForwardBatch.init_new copies batch.prefix_lens and batch.extend_lens into # extend_prefix_lens / extend_seq_lens for non-decode, non-idle modes, matching regular # extend metadata builders that consume those tensors directly. out_prefix_lens.copy_(forward_batch.extend_prefix_lens[:bs].to(torch.int64)) out_extend_seq_lens.copy_(forward_batch.extend_seq_lens[:bs].to(torch.int64)) else: raise NotImplementedError( f"Unsupported forward mode for kv-canary: {forward_mode}" )