Implementation:LMCache LMCache L1 Manager

Knowledge Sources	LMCache
Domains	Distributed Caching, Memory Management
Last Updated	2026-02-09 00:00 GMT

Overview

L1Manager implements a thread-safe object lifecycle state machine for the L1 (first-level) distributed cache, managing read/write locks with TTL and memory allocation.

Description

The L1Manager class manages objects in the L1 cache through a well-defined state machine: objects transition between None, write_locked, ready, and read_locked states. Write locks are acquired via reserve_write (allocating memory if needed) and released via finish_write. Read locks use reference counting, acquired via reserve_read and released via finish_read. Both lock types have configurable TTLs for automatic expiration. The manager supports temporary objects that are automatically deleted when their last read lock is released. All state-mutating operations are atomic across lists of keys using a global lock (via the @l1_mgr_synchronized decorator). The manager delegates memory allocation and deallocation to an L1MemoryManager and notifies registered L1ManagerListener instances of lifecycle events. A companion L1ObjectState dataclass tracks per-object state including the memory object, write lock, read lock, and temporary flag.

Usage

The L1Manager is instantiated by the distributed cache layer with an L1ManagerConfig. It is used by P2P transfer handlers and local cache operations to safely manage concurrent access to shared KV cache objects.

Code Reference

Source Location

• Repository: LMCache
• File: lmcache/v1/distributed/l1_manager.py
• Lines: 1-540

Signature

@dataclass
class L1ObjectState:
    memory_obj: MemoryObj
    write_lock: TTLLock
    read_lock: TTLLock
    is_temporary: bool
    def available_for_read(self) -> bool: ...
    def available_for_write(self) -> bool: ...

L1OperationResult = tuple[L1Error, MemoryObj | None]

class L1Manager:
    def __init__(self, config: L1ManagerConfig) -> None: ...

    def register_listener(self, listener: L1ManagerListener) -> None: ...

    def reserve_read(self, keys: list[ObjectKey]) -> dict[ObjectKey, L1OperationResult]: ...

    def unsafe_read(self, keys: list[ObjectKey]) -> dict[ObjectKey, L1OperationResult]: ...

    def finish_read(self, keys: list[ObjectKey]) -> dict[ObjectKey, L1Error]: ...

    def reserve_write(
        self,
        keys: list[ObjectKey],
        is_temporary: list[bool],
        layout_desc: MemoryLayoutDesc,
        mode: Literal["new", "update", "all"] = "all",
    ) -> dict[ObjectKey, L1OperationResult]: ...

    def finish_write(self, keys: list[ObjectKey]) -> dict[ObjectKey, L1Error]: ...

    def delete(self, keys: list[ObjectKey]) -> dict[ObjectKey, L1Error]: ...

    def clear(self) -> None: ...

    def get_memory_usage(self) -> tuple[int, int]: ...

    def close(self) -> None: ...

    def get_object_state(self, key: ObjectKey) -> L1ObjectState | None: ...

    def memcheck(self) -> None: ...

Import

from lmcache.v1.distributed.l1_manager import L1Manager, L1ObjectState, L1OperationResult

I/O Contract

Inputs

Name	Type	Required	Description
config	L1ManagerConfig	Yes	Configuration with memory settings and TTL values for read/write locks
keys	list[ObjectKey]	Yes	List of object keys to operate on (all operations are batched)
is_temporary	list[bool]	Yes (reserve_write)	Whether each key should be marked as temporary (auto-deleted after last read)
layout_desc	MemoryLayoutDesc	Yes (reserve_write)	Memory layout description for allocating new objects
mode	Literal["new", "update", "all"]	No	Write reservation mode: "new" for only new keys, "update" for existing, "all" for both
listener	L1ManagerListener	No	Event listener for key lifecycle callbacks

Outputs

Name	Type	Description
dict[ObjectKey, L1OperationResult]	dict	For reserve_read/reserve_write: maps each key to (L1Error, Optional[MemoryObj])
dict[ObjectKey, L1Error]	dict	For finish_read/finish_write/delete: maps each key to an L1Error status
tuple[int, int]	tuple	Memory usage as (used_bytes, total_bytes)
L1ObjectState or None	L1ObjectState	Internal state of a specific object (for debugging)

Usage Examples

from lmcache.v1.distributed.l1_manager import L1Manager
from lmcache.v1.distributed.config import L1ManagerConfig
from lmcache.v1.distributed.api import ObjectKey, MemoryLayoutDesc

config = L1ManagerConfig(
    memory_config=memory_config,
    write_ttl_seconds=10.0,
    read_ttl_seconds=5.0,
)
manager = L1Manager(config)

# Reserve write for new objects
keys = [ObjectKey("model", 0, 12345), ObjectKey("model", 0, 67890)]
is_temp = [False, False]
results = manager.reserve_write(keys, is_temp, layout_desc, mode="new")

for key, (error, mem_obj) in results.items():
    if error == L1Error.SUCCESS:
        # Write data to mem_obj.tensor
        pass

# Finish write
manager.finish_write(keys)

# Reserve read
read_results = manager.reserve_read(keys)
for key, (error, mem_obj) in read_results.items():
    if error == L1Error.SUCCESS:
        # Read from mem_obj.tensor
        pass

# Finish read
manager.finish_read(keys)

# Check memory usage
used, total = manager.get_memory_usage()