Implementation:LMCache LMCache Batched Message Sender

Knowledge Sources	LMCache
Domains	Message Batching, Cache Controller
Last Updated	2026-02-09 00:00 GMT

Overview

BatchedMessageSender accumulates KV cache admit and evict operation messages and sends them in batches to reduce communication overhead with the cache controller.

Description

This class implements a producer-consumer pattern where KV operation events (ADMIT or EVICT) are added to a thread-safe queue by callers and drained periodically by a background consumer thread. Messages are flushed when either the batch size threshold (default 50) or the timeout period (default 0.01 seconds) is reached. Each message receives a monotonically increasing sequence number at drain time to ensure strict ordering, which is critical for operations like admit-evict-admit on the same key. A single queue is used for both operation types to maintain order consistency. The batched message is sent to the LMCacheWorker via put_msg. Prometheus metrics for queue size are exported. The class provides flush() for synchronous draining and close() for graceful shutdown.

Usage

Use this sender within storage backends that need to communicate cache state changes (admits and evictions) to the cache controller worker process. It is initialized with metadata, configuration, location, and a reference to the worker.

Code Reference

Source Location

• Repository: LMCache
• File: lmcache/v1/storage_backend/batched_message_sender.py
• Lines: 1-222

Signature

class BatchedMessageSender:
    def __init__(
        self,
        metadata: LMCacheMetadata,
        config: LMCacheEngineConfig,
        location: str,
        lmcache_worker: "LMCacheWorker",
    ): ...

    def add_kv_op(self, op_type: OpType, key: int): ...
    def flush(self): ...
    def close(self): ...

Import

from lmcache.v1.storage_backend.batched_message_sender import BatchedMessageSender

I/O Contract

Inputs

Name	Type	Required	Description
metadata	LMCacheMetadata	Yes	Worker metadata providing worker_id
config	LMCacheEngineConfig	Yes	Configuration with batch_size, batch_timeout, and instance_id
location	str	Yes	Storage backend location identifier
lmcache_worker	LMCacheWorker	Yes	The cache controller worker to send batched messages to
op_type	OpType	Yes	Operation type (ADMIT or EVICT)
key	int	Yes	Chunk hash key for the operation

Outputs

Name	Type	Description
BatchedKVOperationMsg	message	Batched message sent to worker containing instance_id, worker_id, location, and list of operations

Design Details

• Single queue: Both ADMIT and EVICT operations share a single queue to preserve strict operation ordering
• Sequence numbers: Assigned at drain time (not enqueue time) for monotonic ordering
• Threshold flushing: Consumer is notified when queue size reaches batch_size
• Timeout flushing: Consumer wakes up every batch_timeout seconds regardless
• Thread safety: Uses queue.Queue, threading.Lock, and threading.Condition

Usage Examples

from lmcache.v1.storage_backend.batched_message_sender import BatchedMessageSender
from lmcache.v1.cache_controller.message import OpType

sender = BatchedMessageSender(
    metadata=metadata,
    config=config,
    location="LocalCPUBackend",
    lmcache_worker=worker,
)

# Add operations (non-blocking)
sender.add_kv_op(OpType.ADMIT, key=12345)
sender.add_kv_op(OpType.EVICT, key=67890)

# Force flush all pending messages
sender.flush()

# Shutdown
sender.close()