一个client建立之后就会建立一个core engine，这些配置会通过QMZ IPC发送给core engine。

Core engine Architecture

Worker and Executor

MultiprocExecutor

在MultiprocExecutor类中，可以清晰的找到三部曲：

1、创建RPC消息队列

# Initialize worker and set up message queues for SchedulerOutputs
# and ModelRunnerOutputs
max_chunk_bytes = envs.VLLM_MQ_MAX_CHUNK_BYTES_MB * 1024 * 1024
self.rpc_broadcast_mq = MessageQueue(self.world_size,
                                     self.world_size,
                                     max_chunk_bytes=max_chunk_bytes)
scheduler_output_handle = self.rpc_broadcast_mq.export_handle()

2、初始化worker和开始工作

for rank in range(self.world_size):
    unready_workers.append(
        WorkerProc.make_worker_process(
            vllm_config=self.vllm_config,
            local_rank=rank,
            rank=rank,
            distributed_init_method=distributed_init_method,
            input_shm_handle=scheduler_output_handle,
        ))
Workers must be created before wait_for_ready to avoid
deadlock, since worker.init_device() does a device sync.
self.workers = WorkerProc.wait_for_ready(unready_workers)
Ensure message queues are ready. Will deadlock if re-ordered
Must be kept consistent with the WorkerProc.
self.rpc_broadcast_mq.wait_until_ready()
for w in self.workers:
w.worker_response_mq.wait_until_ready()
self.start_worker_monitor()
success = True

3、执行模型，拿到输出

# For pipeline parallel, we use a thread pool for asynchronous

execute_model.
if self.max_concurrent_batches > 1:
# Note: must use only 1 IO thread to keep dequeue sequence
# from the response queue
# _async_aggregate_workers_output also assumes a single IO thread
self.io_thread_pool = ThreadPoolExecutor(
max_workers=1, thread_name_prefix=“mp_exec_io”)
self.output_rank = self._get_output_rank()
self.has_connector = self.vllm_config.kv_transfer_config is not None
self.kv_output_aggregator = KVOutputAggregator(
self.parallel_config.world_size)

由上可知，整个执行依赖于WorkerProc和MessageQueue，接下来一一解析

WorkerProc

初始化的时候会把消息队列进行一对一的配对

# Initialize MessageQueue for receiving SchedulerOutput
self.rpc_broadcast_mq = MessageQueue.create_from_handle(
input_shm_handle, self.worker.rank)

Initializes a message queue for sending the model output
self.worker_response_mq = MessageQueue(1, 1)

最主要的就是worker_main函数

def monitor_parent_death():
try:
# This will block until parent process exits (pipe closes)
death_pipe.recv()
except EOFError:
# Parent process has exited, terminate this worker
logger.info(“Parent process exited, terminating worker”)
# Send signal to self to trigger clean shutdown
shutdown_event.set()
except Exception as e:
logger.warning(“Death monitoring error: %s”, e)

death_monitor = Thread(target=monitor_parent_death,
daemon=True,
name=“WorkerDeathMonitor”)
death_monitor.start()

reader.close()
worker = WorkerProc(*args, **kwargs)

Send READY once we know everything is loaded
ready_writer.send({
“status”:
WorkerProc.READY_STR,
“handle”:
worker.worker_response_mq.export_handle(),
})
Ensure message queues are ready. Will deadlock if re-ordered.
Must be kept consistent with the Executor
worker.rpc_broadcast_mq.wait_until_ready()
worker.worker_response_mq.wait_until_ready()

最后：

finally:
if ready_writer is not None:
ready_writer.close()
if death_pipe is not None:
death_pipe.close()
# Clean up once worker exits busy loop
if worker is not None:
worker.shutdown()

是通过

参考文献：

deepwiki —— vllm