lightseekorg--tokenspeed/test/runtime/cache/test_mamba_cache_host.py

from __future__ import annotations

import torch

from tokenspeed.runtime.cache.mamba_cache_host import MambaPoolHost
from tokenspeed.runtime.cache.transfer.mamba_pool import MambaCachePool
from tokenspeed.runtime.layers.attention.backends.hybrid_linear_attn import (
    SimpleMambaPool,
)


def _new_mamba_pool(size: int = 4):
    return SimpleMambaPool(
        size=size,
        num_mamba_layers=2,
        conv_state_shape=(3,),
        temporal_state_shape=(2, 2),
        conv_dtype=torch.float32,
        ssm_dtype=torch.float32,
        mamba_layer_ids=[4, 7],
        device="cpu",
        page_size=1,
    )


def test_mamba_pool_host_alloc_free_tracks_slots():
    device_pool = _new_mamba_pool(size=3)
    host_pool = MambaPoolHost(
        device_pool, host_size_slots=2, pin_memory=False, register_host=False
    )

    first = host_pool.alloc(1)
    second = host_pool.alloc(1)
    assert first.tolist() == [0]
    assert second.tolist() == [1]
    assert host_pool.alloc(1) is None

    host_pool.free(first)
    assert host_pool.available_size() == 1
    assert host_pool.alloc(1).tolist() == [0]


class _SpyPlatform:
    def __init__(self):
        self.registered = []

    def register_host_tensor_for_gpu_access(self, tensor):
        self.registered.append(tensor)


def test_mamba_pool_host_registers_unpinned_cpu_buffers(monkeypatch):
    import tokenspeed.runtime.cache.mamba_cache_host as mamba_cache_host

    device_pool = _new_mamba_pool(size=2)
    platform = _SpyPlatform()
    monkeypatch.setattr(mamba_cache_host, "current_platform", lambda: platform)

    host_pool = MambaPoolHost(
        device_pool, host_size_slots=2, pin_memory=True, register_host=True
    )

    assert platform.registered == [host_pool.conv_buffer, host_pool.ssm_buffer]
    assert not host_pool.conv_buffer.is_pinned()
    assert not host_pool.ssm_buffer.is_pinned()


def test_mamba_pool_host_kernel_writeback_reuses_pointer_tables(monkeypatch):
    import tokenspeed.runtime.cache.mamba_cache_host as mamba_cache_host

    class PtrPlatform(_SpyPlatform):
        def device_visible_data_ptr(self, tensor):
            return tensor.data_ptr()

    calls = []

    def fake_transfer_kv_all_layer_mla(**kwargs):
        calls.append((kwargs["src_layers"], kwargs["dst_layers"]))

    monkeypatch.setattr(mamba_cache_host, "current_platform", lambda: PtrPlatform())
    monkeypatch.setattr(
        mamba_cache_host,
        "transfer_kv_all_layer_mla",
        fake_transfer_kv_all_layer_mla,
    )

    device_pool = _new_mamba_pool(size=4)
    host_pool = MambaPoolHost(
        device_pool, host_size_slots=4, pin_memory=False, register_host=False
    )
    device_indices = torch.tensor([0, 2], dtype=torch.int64)
    host_indices = torch.tensor([1, 3], dtype=torch.int64)

    for _ in range(2):
        host_pool.backup_from_device_all_layer(
            device_pool, host_indices, device_indices, io_backend="kernel"
        )

    ptrs = host_pool._kernel_ptr_tables
    assert ptrs is not None
    assert (
        calls
        == [
            (ptrs["device_conv"], ptrs["host_conv"]),
            (ptrs["device_ssm"], ptrs["host_ssm"]),
        ]
        * 2
    )


def test_mamba_pool_host_direct_roundtrip_is_layerwise():
    device_pool = _new_mamba_pool(size=4)
    host_pool = MambaPoolHost(
        device_pool, host_size_slots=4, pin_memory=False, register_host=False
    )
    device_indices = torch.tensor([0, 2], dtype=torch.int64)
    host_indices = torch.tensor([1, 3], dtype=torch.int64)

    device_pool.conv_state[:, device_indices] = torch.arange(
        device_pool.conv_state[:, device_indices].numel(), dtype=torch.float32
    ).reshape_as(device_pool.conv_state[:, device_indices])
    device_pool.ssm_state[:, device_indices] = (
        torch.arange(
            device_pool.ssm_state[:, device_indices].numel(), dtype=torch.float32
        ).reshape_as(device_pool.ssm_state[:, device_indices])
        + 1000
    )
    expected_conv = device_pool.conv_state[:, device_indices].clone()
    expected_ssm = device_pool.ssm_state[:, device_indices].clone()

    host_pool.backup_from_device_all_layer(
        device_pool,
        host_indices=host_indices,
        device_indices=device_indices,
        io_backend="direct",
    )
    device_pool.conv_state.zero_()
    device_pool.ssm_state.zero_()

    host_pool.load_to_device_per_layer(
        device_pool,
        host_indices=host_indices,
        device_indices=device_indices,
        layer_idx=0,
        io_backend="direct",
    )
    assert torch.equal(device_pool.conv_state[0, device_indices], expected_conv[0])
    assert torch.equal(device_pool.ssm_state[0, device_indices], expected_ssm[0])
    assert torch.equal(
        device_pool.conv_state[1, device_indices], torch.zeros_like(expected_conv[1])
    )
    assert torch.equal(
        device_pool.ssm_state[1, device_indices], torch.zeros_like(expected_ssm[1])
    )

    host_pool.load_to_device_per_layer(
        device_pool,
        host_indices=host_indices,
        device_indices=device_indices,
        layer_idx=1,
        io_backend="direct",
    )
    assert torch.equal(device_pool.conv_state[:, device_indices], expected_conv)
    assert torch.equal(device_pool.ssm_state[:, device_indices], expected_ssm)


def test_mamba_cache_pool_registers_layer_counter_and_delegates():
    device_pool = _new_mamba_pool(size=2)
    host_pool = MambaPoolHost(
        device_pool, host_size_slots=2, pin_memory=False, register_host=False
    )
    cache_pool = MambaCachePool(device_pool, host_pool, io_backend="direct")

    assert device_pool.layer_transfer_counter is cache_pool.get_layer_done_counter()
    assert cache_pool.kind.value == "mamba"
    assert cache_pool.page_size() == 1
    assert cache_pool.num_layers() == 2
    assert cache_pool.local_layer_idx(7) == 1


class SpyCounter:
    def __init__(self):
        self.waited = []

    def wait_until(self, layer_idx: int):
        self.waited.append(layer_idx)


def test_simple_mamba_pool_waits_for_local_layer_before_returning_params():
    device_pool = _new_mamba_pool(size=2)