Implementation:LMCache LMCache Base Connector

Knowledge Sources	LMCache
Domains	Caching, Storage Connectors, Abstract Interfaces
Last Updated	2026-02-09 00:00 GMT

Overview

RemoteConnector is the abstract base class that defines the interface contract for all remote storage connectors used in LMCache's KV cache storage system.

Description

The RemoteConnector class, along with the NotAudit decorator, establishes the foundational interface for communicating with remote KV cache storage backends. The base class initializes common metadata attributes derived from the LMCacheMetadata including tensor shapes, dtypes, memory format, full chunk size, single token size, and remote metadata byte count. It declares abstract methods for core operations (exists, get, put, list, close) and provides default implementations for optional capabilities such as batched_get, batched_put, batched_async_contains, batched_get_non_blocking, ping, and remove_sync. The reshape_partial_chunk utility method handles reshaping memory objects for partial (non-full) chunks.

Usage

Use RemoteConnector as the base class when implementing a new remote storage connector (e.g., Redis, filesystem, Valkey). All concrete connector implementations must subclass RemoteConnector and implement the abstract methods. The NotAudit decorator can be applied to methods that should bypass audit logging when wrapped by AuditConnector.

Code Reference

Source Location

• Repository: LMCache
• File: lmcache/v1/storage_backend/connector/base_connector.py
• Lines: 1-354

Signature

def NotAudit(func): ...

class RemoteConnector(metaclass=abc.ABCMeta):
    def __init__(self, config: LMCacheEngineConfig, metadata: Optional[LMCacheMetadata]): ...
    def reshape_partial_chunk(self, memory_obj: MemoryObj, bytes_read: int) -> MemoryObj: ...
    def post_init(self): ...
    async def exists(self, key: CacheEngineKey) -> bool: ...
    def exists_sync(self, key: CacheEngineKey) -> bool: ...
    async def get(self, key: CacheEngineKey) -> Optional[MemoryObj]: ...
    async def put(self, key: CacheEngineKey, memory_obj: MemoryObj): ...
    async def list(self) -> List[str]: ...
    async def close(self): ...
    def support_ping(self) -> bool: ...
    async def ping(self) -> int: ...
    def support_batched_get(self) -> bool: ...
    async def batched_get(self, keys: List[CacheEngineKey]) -> List[Optional[MemoryObj]]: ...
    def support_batched_put(self) -> bool: ...
    async def batched_put(self, keys: List[CacheEngineKey], memory_objs: List[MemoryObj]): ...
    def support_batched_async_contains(self) -> bool: ...
    async def batched_async_contains(self, lookup_id: str, keys: List[CacheEngineKey], pin: bool = False) -> int: ...
    def support_batched_get_non_blocking(self) -> bool: ...
    async def batched_get_non_blocking(self, lookup_id: str, keys: List[CacheEngineKey]) -> List[MemoryObj]: ...
    def remove_sync(self, key: CacheEngineKey) -> bool: ...
    def batched_contains(self, keys: List[CacheEngineKey]) -> int: ...
    def support_batched_contains(self) -> bool: ...

Import

from lmcache.v1.storage_backend.connector.base_connector import RemoteConnector, NotAudit

I/O Contract

Inputs

Name	Type	Required	Description
config	LMCacheEngineConfig	Yes	LMCache engine configuration with chunk and remote settings
metadata	Optional[LMCacheMetadata]	Yes	Metadata describing tensor shapes, dtypes, chunk size, and MLA mode (must not be None)

Outputs

Name	Type	Description
RemoteConnector	ABC instance	An initialized abstract connector with precomputed metadata attributes (meta_shapes, meta_dtypes, meta_fmt, full_chunk_size, single_token_size, remote_metadata_bytes)

Usage Examples

from lmcache.v1.storage_backend.connector.base_connector import RemoteConnector

# Subclassing to create a custom connector
class MyCustomConnector(RemoteConnector):
    def __init__(self, config, metadata):
        super().__init__(config, metadata)

    async def exists(self, key):
        # Check existence in custom storage
        ...

    async def get(self, key):
        # Retrieve from custom storage
        ...

    async def put(self, key, memory_obj):
        # Store to custom storage
        ...

    async def list(self):
        return []

    async def close(self):
        pass

    def exists_sync(self, key):
        return False