Method batch_processor

optillm/batching.py:136–179

Background thread that forms and processes batches

()

            queue_key: The key identifying the queue/model type
        """
        def batch_processor():
            """Background thread that forms and processes batches"""
            if self.enable_logging:
                logger.debug(f"Batch processor started for {queue_key}")
            
            while not self._shutdown:
                try:
                    batch = []
                    queue_obj = self.queues[queue_key]
                    deadline = time.time() + self.max_wait_seconds
                    
                    # Collect requests until batch is full or timeout
                    while len(batch) < self.max_batch_size and time.time() < deadline:
                        timeout = max(0.001, deadline - time.time())  # Minimum 1ms timeout
                        try:
                            request = queue_obj.get(timeout=timeout)
                            batch.append(request)
                            
                            if self.enable_logging and len(batch) == 1:
                                logger.debug(f"Started batch formation for {queue_key}")
                                
                        except queue.Empty:
                            break
                    
                    if batch:
                        if self.enable_logging:
                            wait_time = time.time() - batch[0].timestamp
                            logger.info(f"Processing batch of {len(batch)} requests for {queue_key} "
                                       f"(waited {wait_time*1000:.1f}ms)")
                        
                        # Update stats
                        self.stats['total_batches'] += 1
                        self.stats['total_requests'] += len(batch)
                        self.stats['avg_batch_size'] = self.stats['total_requests'] / self.stats['total_batches']
                        self.stats['total_wait_time'] += sum(time.time() - req.timestamp for req in batch)
                        
                        self._process_batch(batch)
                        
                except Exception as e:
                    logger.error(f"Error in batch processor for {queue_key}: {e}")
                    # Continue processing other batches
                    
            if self.enable_logging:
                logger.debug(f"Batch processor stopped for {queue_key}")
        
        thread = threading.Thread(target=batch_processor, daemon=True)
        thread.start()