Implementation:LMCache LMCache Cluster Executor

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

Overview

LMCacheClusterExecutor dispatches cache operations to individual worker sockets across the cluster and aggregates their results.

Description

The LMCacheClusterExecutor class acts as the execution layer between the KV controller and the actual LMCache worker processes. When an orchestration-level operation (clear, pin, compress, decompress, move, health check) is requested, the executor looks up all workers for the target instance, creates per-worker messages with unique event IDs, sends them in parallel via ZMQ async sockets, and collects the results. It validates consistency across workers (e.g., all workers must report the same token count) and returns aggregated responses. Communication is serialized using msgspec msgpack encoding.

Usage

This class is instantiated internally by the cache controller during initialization. It is passed to KVController.post_init() and invoked by the KV controller when handling orchestrator requests. It is not typically imported directly by users.

Code Reference

Source Location

• Repository: LMCache
• File: lmcache/v1/cache_controller/executor.py
• Lines: 1-463

Signature

class LMCacheClusterExecutor:
    def __init__(self, reg_controller) -> None: ...

    async def clear(self, msg: ClearMsg) -> Union[ClearRetMsg, ErrorMsg]: ...
    async def pin(self, msg: PinMsg) -> Union[PinRetMsg, ErrorMsg]: ...
    async def compress(self, msg: CompressMsg) -> Union[CompressRetMsg, ErrorMsg]: ...
    async def decompress(self, msg: DecompressMsg) -> Union[DecompressRetMsg, ErrorMsg]: ...
    async def move(self, msg: MoveMsg) -> Union[MoveRetMsg, ErrorMsg]: ...
    async def health(self, msg: HealthMsg) -> Union[HealthRetMsg, ErrorMsg]: ...
    async def check_finish(self, msg: CheckFinishMsg) -> Union[CheckFinishRetMsg, ErrorMsg]: ...
    async def execute(self, operation: str, msg: MsgBase) -> MsgBase: ...
    async def execute_workers(
        self,
        sockets: list[zmq.asyncio.Socket],
        serialized_msgs: list[bytes],
    ) -> list[bytes]: ...

Import

from lmcache.v1.cache_controller.executor import LMCacheClusterExecutor

I/O Contract

Inputs

Name	Type	Required	Description
reg_controller	RegistrationController	Yes	Registration controller providing access to worker sockets and peer URLs
msg	MsgBase subclass	Yes	The operation message (ClearMsg, PinMsg, CompressMsg, etc.)
operation	str	Yes	Name of the method to invoke (e.g., "clear", "pin", "compress")

Outputs

Name	Type	Description
ClearRetMsg	ClearRetMsg	Clear result with event_id and num_tokens cleared
PinRetMsg	PinRetMsg	Pin result with event_id and num_tokens pinned
CompressRetMsg	CompressRetMsg	Compress result with event_id and num_tokens compressed
DecompressRetMsg	DecompressRetMsg	Decompress result with event_id and num_tokens decompressed
MoveRetMsg	MoveRetMsg	Move result with event_id and num_tokens moved
HealthRetMsg	HealthRetMsg	Health check result with per-worker error codes
ErrorMsg	ErrorMsg	Error message if worker not registered or operation fails

Usage Examples

from lmcache.v1.cache_controller.executor import LMCacheClusterExecutor
from lmcache.v1.cache_controller.message import ClearMsg

executor = LMCacheClusterExecutor(reg_controller)

# Execute a clear operation via the generic dispatch
clear_msg = ClearMsg(event_id="evt_001", instance_id="inst_0", location="hot_cache")
result = await executor.execute("clear", clear_msg)

# Or call the specific method directly
result = await executor.clear(clear_msg)