Implementation:LMCache LMCache Registration Controller

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

Overview

Manages the registration, deregistration, heartbeat, and querying of LMCache worker instances within the cache controller.

Description

The RegistrationController class serves as the central authority for managing worker lifecycle within the LMCache cache controller. It maintains a RegistryTree data structure that maps instance IDs to worker nodes, each associated with ZMQ sockets for communication. The controller handles worker registration (creating ZMQ REQ sockets to connect to workers), deregistration (closing sockets and removing entries), heartbeat processing (updating timestamps and triggering re-registration or full syncs when needed), and worker info queries. It also integrates with Prometheus metrics for monitoring registered worker counts.

Usage

Use this class within the LMCache cache controller to manage the lifecycle of distributed workers. The controller is instantiated at startup and then wired with post_init to receive references to the KV controller and cluster executor. Workers register themselves via RegisterMsg, send periodic heartbeats via HeartbeatMsg, and can be queried for status via QueryWorkerInfoMsg.

Code Reference

Source Location

• Repository: LMCache
• File: lmcache/v1/cache_controller/controllers/registration_controller.py
• Lines: 1-282

Signature

class RegistrationController:
    def __init__(self) -> None: ...
    def post_init(self, kv_controller, cluster_executor) -> None: ...
    def get_socket(self, instance_id: str, worker_id: int) -> Optional[zmq.asyncio.Socket]: ...
    def get_peer_init_url(self, instance_id: str, worker_id: int) -> Optional[str]: ...
    def get_workers(self, instance_id: str) -> list[int]: ...
    async def get_instance_id(self, msg: QueryInstMsg) -> QueryInstRetMsg: ...
    async def register(self, msg: RegisterMsg, extra_config: Optional[dict[str, str]] = None) -> RegisterRetMsg: ...
    async def deregister(self, msg: DeRegisterMsg) -> None: ...
    async def health(self, msg: HealthMsg) -> HealthRetMsg: ...
    async def heartbeat(self, msg: HeartbeatMsg) -> HeartbeatRetMsg: ...
    async def query_worker_info(self, msg: QueryWorkerInfoMsg) -> QueryWorkerInfoRetMsg: ...

Import

from lmcache.v1.cache_controller.controllers.registration_controller import RegistrationController

I/O Contract

Inputs

Name	Type	Required	Description
msg (register)	RegisterMsg	Yes	Contains instance_id, worker_id, ip, port, and optional peer_init_url
extra_config	Optional[dict[str, str]]	No	Extra config to return to worker during registration (e.g., heartbeat_url)
msg (deregister)	DeRegisterMsg	Yes	Contains instance_id and worker_id to deregister
msg (heartbeat)	HeartbeatMsg	Yes	Contains instance_id, worker_id, ip, port, peer_init_url for heartbeat
msg (query_worker_info)	QueryWorkerInfoMsg	Yes	Contains event_id, instance_id (or "all"), and optional worker_ids filter
msg (get_instance_id)	QueryInstMsg	Yes	Contains ip and event_id to look up instance by IP

Outputs

Name	Type	Description
RegisterRetMsg	RegisterRetMsg	Registration acknowledgment, optionally carrying extra_config for the worker
HeartbeatRetMsg	HeartbeatRetMsg	Heartbeat response with optional commands (e.g., FullSyncCommand) for the worker
QueryWorkerInfoRetMsg	QueryWorkerInfoRetMsg	List of worker info objects matching the query
QueryInstRetMsg	QueryInstRetMsg	Instance ID associated with the queried IP address

Usage Examples

from lmcache.v1.cache_controller.controllers.registration_controller import RegistrationController
from lmcache.v1.cache_controller.message import RegisterMsg, HeartbeatMsg

controller = RegistrationController()
controller.post_init(kv_controller=kv_ctrl, cluster_executor=executor)

# Register a worker
register_msg = RegisterMsg(
    instance_id="instance-1",
    worker_id=0,
    ip="192.168.1.10",
    port=5555,
    peer_init_url="tcp://192.168.1.10:6000",
)
ret = await controller.register(register_msg)

# Process a heartbeat
heartbeat_msg = HeartbeatMsg(
    instance_id="instance-1",
    worker_id=0,
    ip="192.168.1.10",
    port=5555,
    peer_init_url="tcp://192.168.1.10:6000",
)
heartbeat_ret = await controller.heartbeat(heartbeat_msg)