Implementation:LMCache LMCache Abstract Backend

Knowledge Sources	LMCache
Domains	Storage Backend, KV Cache Management, Abstract Interface
Last Updated	2026-02-09 00:00 GMT

Overview

The abstract backend module defines the three-tier interface hierarchy for LMCache storage backends: StorageBackendInterface for basic cache operations, AllocatorBackendInterface for memory-allocating backends, and StoragePluginInterface for configurable plug-and-play backends.

Description

StorageBackendInterface is the foundational abstract class that specifies the contract for all storage backends in LMCache. It defines abstract methods for contains (key existence check with optional pinning), get_blocking (synchronous retrieval), batched_submit_put_task (async batch store with optional per-key completion callbacks), pin/unpin (eviction control), remove/batched_remove (cache entry deletion), and close (resource cleanup). It also provides default implementations for batched_get_blocking, batched_contains (prefix-match based), and async variants. AllocatorBackendInterface extends this with allocate, batched_allocate, initialize_allocator, and get_memory_allocator for backends that actively manage memory allocation (e.g., LocalCPUBackend). StoragePluginInterface (aliased as ConfigurableStorageBackendInterface) adds constructor parameters for config, metadata, local CPU backend, and event loop, enabling dynamic backend loading from configuration files.

Usage

Use these interfaces when implementing a new storage backend for LMCache. Implement StorageBackendInterface for basic backends, AllocatorBackendInterface for backends that allocate memory, or StoragePluginInterface for backends loaded dynamically from configuration.

Code Reference

Source Location

Repository: LMCache
File: lmcache/v1/storage_backend/abstract_backend.py
Lines: 1-431

Signature

class StorageBackendInterface(metaclass=abc.ABCMeta):
    def __init__(self, dst_device: str = "cuda") -> None: ...
    @abc.abstractmethod
    def contains(self, key: CacheEngineKey, pin: bool = False) -> bool: ...
    @abc.abstractmethod
    def exists_in_put_tasks(self, key: CacheEngineKey) -> bool: ...
    @abc.abstractmethod
    def batched_submit_put_task(self, keys: Sequence[CacheEngineKey],
        objs: List[MemoryObj], transfer_spec: Any = None,
        on_complete_callback: Optional[Callable[[CacheEngineKey], None]] = None
    ) -> Union[List[Future], None]: ...
    @abc.abstractmethod
    def get_blocking(self, key: CacheEngineKey) -> Optional[MemoryObj]: ...
    def batched_get_blocking(self, keys: List[CacheEngineKey]) -> List[Optional[MemoryObj]]: ...
    @abc.abstractmethod
    def pin(self, key: CacheEngineKey) -> bool: ...
    @abc.abstractmethod
    def unpin(self, key: CacheEngineKey) -> bool: ...
    @abc.abstractmethod
    def remove(self, key: CacheEngineKey, force: bool = True) -> bool: ...
    def batched_remove(self, keys: list[CacheEngineKey], force: bool = True) -> int: ...
    def batched_contains(self, keys: List[CacheEngineKey], pin: bool = False) -> int: ...
    @abc.abstractmethod
    def get_allocator_backend(self) -> "AllocatorBackendInterface": ...
    @abc.abstractmethod
    def close(self) -> None: ...

class AllocatorBackendInterface(StorageBackendInterface):
    @abc.abstractmethod
    def initialize_allocator(self, config: LMCacheEngineConfig,
        metadata: LMCacheMetadata) -> MemoryAllocatorInterface: ...
    @abc.abstractmethod
    def get_memory_allocator(self) -> MemoryAllocatorInterface: ...
    @abc.abstractmethod
    def allocate(self, shapes: Union[torch.Size, list[torch.Size]],
        dtypes: Union[torch.dtype, list[torch.dtype]],
        fmt: MemoryFormat = MemoryFormat.KV_2LTD,
        eviction: bool = True, busy_loop: bool = True) -> Optional[MemoryObj]: ...
    @abc.abstractmethod
    def batched_allocate(self, shapes: Union[torch.Size, list[torch.Size]],
        dtypes: Union[torch.dtype, list[torch.dtype]], batch_size: int,
        fmt: MemoryFormat = MemoryFormat.KV_2LTD,
        eviction: bool = True, busy_loop: bool = True) -> Optional[list[MemoryObj]]: ...

class StoragePluginInterface(StorageBackendInterface):
    def __init__(self, dst_device: str = "cuda",
        config: Optional[LMCacheEngineConfig] = None,
        metadata: Optional[LMCacheMetadata] = None,
        local_cpu_backend: Optional["LocalCPUBackend"] = None,
        loop: Optional[asyncio.AbstractEventLoop] = None) -> None: ...

Import

from lmcache.v1.storage_backend.abstract_backend import (
    StorageBackendInterface,
    AllocatorBackendInterface,
    StoragePluginInterface,
)

I/O Contract

Inputs

Name	Type	Required	Description
dst_device	str	No	Target device for tensor operations: "cpu", "cuda", "cuda:0", etc. (default: "cuda")
key	CacheEngineKey	Yes	Cache key identifying a specific KV cache chunk
keys	List[CacheEngineKey]	Yes (for batched ops)	List of cache keys for batch operations
objs	List[MemoryObj]	Yes (for put)	Memory objects to store
pin	bool	No	Whether to pin the key to prevent eviction (default: False)
shapes	Union[torch.Size, list[torch.Size]]	Yes (for allocate)	Shape(s) of the tensor(s) to allocate
dtypes	Union[torch.dtype, list[torch.dtype]]	Yes (for allocate)	Data type(s) of the tensor(s) to allocate
fmt	MemoryFormat	No	Memory format for allocation (default: KV_2LTD)
on_complete_callback	Optional[Callable]	No	Per-key callback invoked after a put operation completes
config	LMCacheEngineConfig	No (for plugin)	Engine configuration for plugin backends
metadata	LMCacheMetadata	No (for plugin)	Engine metadata for plugin backends

Outputs

Name	Type	Description
contains()	bool	True if the key exists in the backend
get_blocking()	Optional[MemoryObj]	The retrieved memory object, or None if not found
batched_submit_put_task()	Union[List[Future], None]	Futures for async puts, or None for sync/skipped operations
pin()	bool	True if pinning was successful
unpin()	bool	True if unpinning was successful
remove()	bool	True if removal was successful
batched_contains()	int	Number of consecutive prefix-matched keys found
allocate()	Optional[MemoryObj]	Allocated memory object, or None on failure
batched_allocate()	Optional[list[MemoryObj]]	List of allocated memory objects, or None on failure

Usage Examples

from lmcache.v1.storage_backend.abstract_backend import (
    StorageBackendInterface,
    AllocatorBackendInterface,
)
from lmcache.utils import CacheEngineKey

# Implementing a custom storage backend
class MyBackend(StorageBackendInterface):
    def contains(self, key: CacheEngineKey, pin: bool = False) -> bool:
        return key in self._store

    def get_blocking(self, key: CacheEngineKey):
        return self._store.get(key)

    def batched_submit_put_task(self, keys, objs, transfer_spec=None,
                                 on_complete_callback=None):
        for key, obj in zip(keys, objs):
            self._store[key] = obj
            if on_complete_callback:
                on_complete_callback(key)
        return None

    # ... implement remaining abstract methods ...

# Using batched_contains for prefix-match lookup
backend = MyBackend(dst_device="cpu")
hit_count = backend.batched_contains(keys, pin=True)
print(f"Prefix match: {hit_count} consecutive hits out of {len(keys)}")

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment