flashinfer / wrapper71bd33

import torch
import flashinfer

_WORKSPACE_SIZE_BYTES = 128 * 1024 * 1024
_workspace_cache = {}
_wrapper_cache = {}
_plan_state = {}


def _get_workspace(device):
    key = str(device)
    buffer = _workspace_cache.get(key)
    if buffer is None or buffer.device != device or buffer.numel() < _WORKSPACE_SIZE_BYTES:
        buffer = torch.empty(_WORKSPACE_SIZE_BYTES, dtype=torch.uint8, device=device)
        _workspace_cache[key] = buffer
    return buffer


def _get_wrapper(key, device):
    wrapper = _wrapper_cache.get(key)
    if wrapper is None:
        workspace = _get_workspace(device)
        wrapper = flashinfer.BatchPrefillWithPagedKVCacheWrapper(
            workspace,
            kv_layout="NHD",
        )
        _wrapper_cache[key] = wrapper
    return wrapper


def run(q, k_cache, v_cache, qo_indptr, kv_indptr, kv_indices, sm_scale):
    total_q, num_qo_heads, head_dim = q.shape
    _, page_size, num_kv_heads, _ = k_cache.shape
    batch_size = qo_indptr.shape[0] - 1
    num_kv_indices = kv_indices.shape[0]

    device = q.device
    wrapper_key = (
        str(device),
        num_qo_heads,
        num_kv_heads,
        head_dim,
        page_size,
        q.dtype,
        k_cache.dtype,
    )

    wrapper = _get_wrapper(wrapper_key, device)
    state = _plan_state.get(wrapper_key)

    if isinstance(sm_scale, torch.Tensor):
        sm_scale_value = float(sm_scale.item())
    else:
        sm_scale_value = float(sm_scale)

    needs_plan = True
    if state is not None:
        needs_plan = (
            state.get("total_q") != total_q
            or state.get("batch_size") != batch_size
            or state.get("num_kv_indices") != num_kv_indices
            or state.get("sm_scale") != sm_scale_value
            or state.get("qo_indptr_ptr") != qo_indptr.data_ptr()
            or state.get("kv_indptr_ptr") != kv_indptr.data_ptr()
            or state.get("kv_indices_ptr") != kv_indices.data_ptr()
        )

    if needs_plan:
        kv_last_page_len = torch.ones(batch_size, dtype=torch.int32, device=device)
        wrapper.plan(
            qo_indptr=qo_indptr,
            paged_kv_indptr=kv_indptr,
            paged_kv_indices=kv_indices,
            paged_kv_last_page_len=kv_last_page_len,
            num_qo_heads=num_qo_heads,
            num_kv_heads=num_kv_heads,
            head_dim_qk=head_dim,
            page_size=page_size,
            causal=True,
            sm_scale=sm_scale,
            q_data_type=q.dtype,
            kv_data_type=k_cache.dtype,
        )
        _plan_state[wrapper_key] = {
            "total_q": total_q,
            "batch_size": batch_size,
            "num_kv_indices": num_kv_indices,
            "sm_scale": sm_scale_value,
            "qo_indptr_ptr": qo_indptr.data_ptr(),
            "kv_indptr_ptr": kv_indptr.data_ptr(),
            "kv_indices_ptr": kv_indices.data_ptr(),
        }

    output, lse = wrapper.run(
        q,
        (k_cache, v_cache),
        return_lse=True,
    )

    return output, lse