Implementation:LMCache LMCache Distributed Storage Manager

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

Overview

Provides the high-level multi-tier storage manager for the distributed KV cache, orchestrating write reservations, prefetch tasks, read operations, and eviction.

Description

The StorageManager class is the primary interface for the LMCache distributed storage system in multiprocess mode. It wraps an L1Manager for managing the L1 cache tier and an EvictionController that runs background eviction based on configured policies. The manager exposes a write path (reserve_write to allocate writable memory, finish_write to commit), a prefetch/read path (submit_prefetch_task to check L1 availability and add read locks, read_prefetched_results as a context manager to access prefetched data, finish_read_prefetched to release read locks), and lifecycle methods (clear, close, memcheck). The PrefetchHandle dataclass tracks how many prefix chunks were found in L1.

Usage

Use StorageManager as the main entry point for the serving engine integration code to store and retrieve KV cache chunks. Call reserve_write before writing new data, finish_write after data is written, submit_prefetch_task to check cache availability before inference, and read_prefetched_results to consume the cached data.

Code Reference

Source Location

• Repository: LMCache
• File: lmcache/v1/distributed/storage_manager.py
• Lines: 1-241

Signature

@dataclass(frozen=True)
class PrefetchHandle:
    prefix_hit_chunks: int

class StorageManager:
    def __init__(self, config: StorageManagerConfig) -> None: ...
    def reserve_write(self, keys: list[ObjectKey], layout_desc: MemoryLayoutDesc,
                      mode: Literal["new", "update", "all"]) -> dict[ObjectKey, MemoryObj]: ...
    def finish_write(self, keys: list[ObjectKey]) -> None: ...
    def read_prefetched_results(self, keys: list[ObjectKey]) -> Iterator[list[MemoryObj] | None]: ...
    def finish_read_prefetched(self, keys: list[ObjectKey]) -> None: ...
    def submit_prefetch_task(self, keys: list[ObjectKey]) -> PrefetchHandle: ...
    def query_prefetch_status(self, handle: PrefetchHandle) -> int | None: ...
    def clear(self) -> None: ...
    def close(self) -> None: ...
    def memcheck(self) -> None: ...

Import

from lmcache.v1.distributed.storage_manager import StorageManager, PrefetchHandle

I/O Contract

Inputs

Name	Type	Required	Description
config	StorageManagerConfig	Yes	Full storage manager configuration (L1 + eviction)
keys	list[ObjectKey]	Yes	Object keys to operate on
layout_desc	MemoryLayoutDesc	Yes	Memory layout description for write reservations
mode	Literal["new", "update", "all"]	Yes	Write reservation mode: new-only, update-only, or all
handle	PrefetchHandle	Yes	Handle returned from submit_prefetch_task for status queries

Outputs

Name	Type	Description
reserved	dict[ObjectKey, MemoryObj]	Mapping of successfully reserved keys to their memory objects
PrefetchHandle	PrefetchHandle	Handle containing prefix_hit_chunks count
prefetched_data	list[MemoryObj] or None	List of memory objects from L1 (None if any object missing)
prefix_hit_chunks	int or None	Number of prefix-matching chunks found; None if still in progress

Usage Examples

from lmcache.v1.distributed.storage_manager import StorageManager
from lmcache.v1.distributed.config import StorageManagerConfig

storage = StorageManager(config)

# Write path
reserved = storage.reserve_write(keys, layout_desc, mode="new")
for key, mem_obj in reserved.items():
    # Copy data into mem_obj.tensor
    mem_obj.tensor.copy_(source_data)
storage.finish_write(list(reserved.keys()))

# Prefetch and read path
handle = storage.submit_prefetch_task(read_keys)
hit_count = storage.query_prefetch_status(handle)
with storage.read_prefetched_results(read_keys[:hit_count]) as mem_objs:
    if mem_objs is not None:
        for obj in mem_objs:
            # Process cached KV data
            process(obj.tensor)
storage.finish_read_prefetched(read_keys[:hit_count])

# Cleanup
storage.close()