Repository: LMCache/LMCache
File: lmcache/v1/storage_backend/batched_message_sender.py (Python)

Code before:
        self,
        metadata: LMCacheEngineMetadata,
        config: LMCacheEngineConfig,
        location: str,
        lmcache_worker: "LMCacheWorker",
    ):
        self.batch_size = config.get_extra_config_value("kv_msg_batch_size", 50)
        self.batch_timeout = config.get_extra_config_value("kv_msg_batch_timeout", 0.01)
        self.lmcache_worker = lmcache_worker

        # Common fields shared by all operations in the batch
        self.instance_id = config.lmcache_instance_id
        self.worker_id = metadata.worker_id
        self.location = location

        # Use thread-safe queue for producer-consumer pattern
        self.message_queue: queue.Queue[KVOpEvent] = queue.Queue()
        self.sequence_number = 0
        self.sequence_lock = threading.Lock()

        # Condition variable for coordinating producer and consumer
        self.cv = threading.Condition()
        self.running = False
        self.thread: Optional[threading.Thread] = None

        # Lock to prevent concurrent drain operations
        self.drain_lock = threading.Lock()

        self._start_background_thread()

        self._setup_metrics()

    def _setup_metrics(self):
        """Setup metrics for monitoring queue size."""
        prometheus_logger = PrometheusLogger.GetInstanceOrNone()
        if prometheus_logger is not None:
            prometheus_logger.kv_msg_queue_size.set_function(
                lambda: self.message_queue.qsize()
            )

    def _start_background_thread(self):
        """Start background thread for periodic flushing."""
        self.running = True
        self.thread = threading.Thread(target=self._consumer_loop, daemon=True)
        self.thread.start()

    def _consumer_loop(self):
        """Consumer loop that drains queue and sends batched messages."""
        while self.running:
            with self.cv:
                # Wait for timeout or notification from producer

Review comment (performance):
@maobaolong Is the modification of `batched_message_sender.py` related to the refactor of `controller`?

Code after:
        self,
        metadata: LMCacheEngineMetadata,
        config: LMCacheEngineConfig,
        location: str,
        lmcache_worker: "LMCacheWorker",
    ):
        self.batch_size = config.get_extra_config_value("kv_msg_batch_size", 50)
        self.batch_timeout = config.get_extra_config_value("kv_msg_batch_timeout", 0.01)
        self.lmcache_worker = lmcache_worker

        # Common fields shared by all operations in the batch
        self.instance_id = config.lmcache_instance_id
        self.worker_id = metadata.worker_id
        self.location = location

        # Use thread-safe queue for producer-consumer pattern
        self.message_queue: queue.Queue[KVOpEvent] = queue.Queue()
        self.sequence_number = 0
        self.sequence_lock = threading.Lock()

        # Condition variable for coordinating producer and consumer
        self.cv = threading.Condition()
        self.running = False
        self.thread: Optional[threading.Thread] = None

        self._start_background_thread()

        self._setup_metrics()

    def _setup_metrics(self):
        """Setup metrics for monitoring queue size."""
        prometheus_logger = PrometheusLogger.GetInstanceOrNone()
        if prometheus_logger is not None:
            prometheus_logger.kv_msg_queue_size.set_function(
                lambda: self.message_queue.qsize()
            )

    def _start_background_thread(self):
        """Start background thread for periodic flushing."""
        self.running = True
        self.thread = threading.Thread(target=self._consumer_loop, daemon=True)
        self.thread.start()

    def _consumer_loop(self):
        """Consumer loop that drains queue and sends batched messages."""
        while self.running:
            with self.cv:
                # Wait for timeout or notification from producer
                self.cv.wait(timeout=self.batch_timeout)

                # Check if we have messages to process while holding the lock