vLLM Code Reading (Section 2): vLLM 的 ZMQ、msgpack 和 Tensor Buffer 传输机制

引子

vLLM API Server 和 EngineCore 的跨进程通信里，一个非常值得停下来看的点是这段代码：

msg = (self.core_engine, request_type.value, *self.encoder.encode(request))

if len(msg) <= 3:
    # No auxiliary buffers => no tensor backing buffers in request.
    self.input_socket.send_multipart(msg, copy=False)
    return

tracker = self.input_socket.send_multipart(msg, copy=False, track=True)
self.add_pending_message(tracker, request)

这段代码看起来很短，但里面同时包含了 vLLM 跨进程通信的几个关键设计：

ZMQ multipart message
msgpack 主体序列化
tensor / ndarray 大 buffer 的零拷贝传输
track=True 保护 buffer 生命周期
可选的 torch shared memory tensor IPC

下面把这套通信机制拆开来看。

为什么需要这套通信层

vLLM 的 OpenAI API Server 和 EngineCore 通常不在同一个逻辑执行层里。

API Server 进程负责：

FastAPI / Uvicorn
OpenAI Chat / Responses protocol
InputProcessor
OutputProcessor
streaming response

EngineCore 进程负责：

Scheduler
KVCacheManager
Executor
Model execution control

它们之间需要传递两类消息：

API -> EngineCore:
  EngineCoreRequest
  abort request
  utility command

EngineCore -> API:
  EngineCoreOutputs
  utility result
  scheduler stats

这些消息有一个特点：绝大多数时候只是 Python 结构化数据，但在多模态、prompt embeds、pooling 等场景里，消息里可能带有比较大的 tensor 或 ndarray。

所以通信层需要同时满足：

1. 能传结构化 Python 对象
2. 能高效传大 tensor / ndarray
3. 能跨进程
4. 能支持 asyncio API server
5. 能处理一个 API server 对多个 EngineCore 的路由

vLLM 的选择是：

ZMQ for transport
msgpack for structured serialization
multipart frames for large buffers
optional torch shared memory for multimodal tensors

ZMQ socket 结构

API Server 和 EngineCore 之间有两条方向相反的通道。

输入方向：

API Server
  ROUTER socket
    -> EngineCore
       DEALER socket

输出方向：

EngineCore
  PUSH socket
    -> API Server
       PULL socket

画成图：

flowchart LR
    subgraph API["API Server process"]
        AMC["AsyncMPClient / MPClient"]
        ROUTER["ZMQ ROUTER input_socket"]
        PULL["ZMQ PULL output_socket"]
        Q["asyncio.Queue[EngineCoreOutputs]"]
    end

    subgraph CORE["EngineCore process"]
        DEALER["ZMQ DEALER input socket"]
        PUSH["ZMQ PUSH output socket"]
        IQ["input_queue"]
        OQ["output_queue"]
        EC["EngineCore busy loop"]
    end

    AMC --> ROUTER
    ROUTER -- "ADD / ABORT / UTILITY" --> DEALER
    DEALER --> IQ --> EC --> OQ --> PUSH
    PUSH -- "EngineCoreOutputs" --> PULL
    PULL --> Q --> AMC

为什么输入方向用 ROUTER / DEALER？

因为 API Server 侧可能需要和多个 EngineCore 通信。ROUTER 的第一帧可以放 engine identity，API Server 就能把请求发给指定 EngineCore。

为什么输出方向用 PUSH / PULL？

因为 EngineCore 只需要把输出推回对应 API Server client。输出消息里带有 client_index，用于多 API server 场景下把结果送回正确的前端。

一条输入消息长什么样

发送请求时，核心代码是：

1	`msg = (self.core_engine, request_type.value, *self.encoder.encode(request))`

它构造的是一个 ZMQ multipart message。

如果没有大 tensor buffer，结构是：

1
2
3

frame 0: self.core_engine
frame 1: request_type.value
frame 2: msgpack payload

也就是：

1	`len(msg) = 3`

如果有大 tensor / ndarray，被 encoder 拆成 auxiliary buffers，结构会变成：

frame 0: self.core_engine
frame 1: request_type.value
frame 2: msgpack payload
frame 3: tensor buffer 1
frame 4: tensor buffer 2
...

这时：

$$
\text{num_aux_buffers} = \text{len(msg)} - 3
$$

所以代码里的判断：

1	`if len(msg) <= 3:`

真正含义是：

1	`没有被拆出来的 tensor / ndarray backing buffer`

而不是“消息很短所以不重要”。

MsgpackEncoder 的核心设计

vLLM 的 MsgpackEncoder 在：

1	`vllm/v1/serial_utils.py`

它不是直接用 vanilla msgpack，而是在 msgpack encoder 上挂了一个 enc_hook，专门处理 torch tensor、numpy ndarray、多模态结构等对象。

最关键的代码结构是：

def encode(self, obj: Any) -> Sequence[bytestr]:
    try:
        if self.oob_tensor_consumer is not None:
            self.oob_tensor_consumer.new_message()
        self.aux_buffers = bufs = [b""]
        bufs[0] = self.encoder.encode(obj)
        return bufs
    finally:
        self.aux_buffers = None

这里有一个容易误解的地方：

1	`self.aux_buffers = bufs = [b""]`

这不是创建了两个 list。self.aux_buffers 和局部变量 bufs 指向同一个 list。

可以理解为：

bufs ----------+
               |
               v
self.aux_buffers ---> [b""]

编码过程中，如果 msgpack 遇到 tensor，会进入：

1
2
3

def enc_hook(self, obj: Any) -> Any:
    if isinstance(obj, torch.Tensor):
        return self._encode_tensor(obj)

_encode_tensor() 可能会 append 到 self.aux_buffers：

1
2
3

data = len(self.aux_buffers)
self.aux_buffers.append(tensor_data(obj))
return dtype, obj.shape, data

因为 self.aux_buffers 和 bufs 是同一个 list，所以 append 后，最终返回的 bufs 也包含这些 tensor buffers。

为什么要先放 b”” 占位

这句写法很有意思：

1 2	`self.aux_buffers = bufs = [b""] bufs[0] = self.encoder.encode(obj)`

为什么不写成：

1 2	`main = self.encoder.encode(obj) bufs = [main]`

原因是：在 self.encoder.encode(obj) 执行过程中，enc_hook 就可能遇到 tensor，并且需要把 tensor buffer 加到 aux_buffers 里。同时，主 msgpack payload 必须放在 frame 0。

所以它先创建：

1	`bufs[0] = b"" # placeholder for main payload`

这样第一个 tensor buffer append 时：

1	`data = len(self.aux_buffers)`

此时 len(self.aux_buffers) == 1，所以第一个 tensor 的 data index 是 1。

编码结束后：

1	`bufs[0] = self.encoder.encode(obj)`

把真正的主 msgpack payload 填回 frame 0。

最终结构是：

1
2
3

bufs[0] = main msgpack payload
bufs[1] = tensor buffer 1
bufs[2] = tensor buffer 2

这是一种很小但很巧的设计：主 payload 和 aux buffers 在同一个 list 里构建，但主 payload 的位置始终稳定是 0。

主 payload 和 aux buffer 如何对应

假设 EngineCoreRequest 里有一个 prompt_embeds：

1	`prompt_embeds: torch.Tensor(shape=[10, 4096], dtype=float16)`

编码结果可以抽象成：

bufs[0] = msgpack({
  "prompt_embeds": {
    "dtype": "float16",
    "shape": [10, 4096],
    "data": 1
  },
  ...
})

bufs[1] = raw bytes of prompt_embeds

主 msgpack payload 里不直接放 tensor 内容，只放：

1
2
3

dtype
shape
data index

真正的 tensor bytes 在 bufs[data] 里。

这个设计的好处是避免把大 tensor 拷贝进 msgpack 主体。

没有 aux buffer 的路径：

tensor memory
  -> copy into msgpack bytes
  -> ZMQ send
  -> decode

有 aux buffer 的路径：

tensor memory
  -> memoryview / backing buffer frame
  -> ZMQ multipart send
  -> receiver frombuffer decode

核心收益是少拷贝。

aux_buffers 是共享内存吗

不是。

aux_buffers 本身只是一个 Python list，里面放的是准备交给 ZMQ multipart 发送的 frame。

它不是 IPC 机制，也不是共享内存。真正发送它的是：

1	`self.input_socket.send_multipart(msg, copy=False)`

也就是说：

1
2
3

aux_buffers
  -> msg tuple
  -> ZMQ multipart frames

默认路径是 ZMQ 发送，不是共享内存。

只有启用：

1	`--mm-tensor-ipc torch_shm`

或者配置里的：

1	`mm_tensor_ipc = "torch_shm"`

才会走 torch shared memory tensor IPC。

为什么 len(msg) > 3 要 track=True

这里是这段代码最关键的原因：

1 2	`tracker = self.input_socket.send_multipart(msg, copy=False, track=True) self.add_pending_message(tracker, request)`

因为 copy=False。

当 ZMQ copy=False 时，发送接口可能不会立刻复制 buffer 内容，而是异步地引用调用方传进去的 buffer。

如果 aux frame 来自：

1
2
3

request.prompt_embeds
request.mm_features 中的 tensor
某个 ndarray 的底层 data buffer

那么一旦函数返回，Python 层如果释放了 request，这些 tensor / ndarray 的 backing memory 也可能被释放。

但 ZMQ 可能还没有真正发送完这些 bytes。

所以 track=True 返回一个 MessageTracker。vLLM 把：

1 2	`MessageTracker request object`

一起放到 pending_messages。

这样在 ZMQ 发送完成之前，request 对象还被强引用着，里面的 tensor backing memory 不会被释放。

生命周期可以画成：

sequenceDiagram
    participant API as API Server
    participant ENC as MsgpackEncoder
    participant ZMQ as ZMQ
    participant P as pending_messages
    participant CORE as EngineCore

    API->>ENC: encode(EngineCoreRequest)
    ENC-->>API: [main payload, tensor buffer]
    API->>ZMQ: send_multipart(copy=False, track=True)
    ZMQ-->>API: MessageTracker
    API->>P: keep (tracker, request)
    ZMQ-->>CORE: multipart frames
    ZMQ-->>API: tracker.done = True
    API->>P: release request reference

如果没有这一步，就可能出现：

ZMQ 还在用 buffer
request 已经被 GC
tensor backing memory 被释放
receiver 读到非法或损坏数据

所以 track=True 不是为了功能正确地“发送出去”，而是为了在零拷贝发送时保证内存生命周期正确。

EngineCore 侧怎么 decode

EngineCore input thread 中会创建 decoder：

add_request_decoder = MsgpackDecoder(
    EngineCoreRequest,
    oob_tensor_provider=self.tensor_ipc_receiver
)

收到 multipart 后：

1 2	`type_frame, *data_frames = input_socket.recv_multipart(copy=False) req = add_request_decoder.decode(data_frames)`

注意 data_frames 不包括 ZMQ identity 和 request type。它对应的是：

1
2
3

data_frames[0] = main msgpack payload
data_frames[1] = tensor buffer 1
data_frames[2] = tensor buffer 2

decoder 先设置：

1	`self.aux_buffers = bufs`

然后解码主 payload：

1	`return self.decoder.decode(bufs[0])`

当主 payload 里遇到 tensor 描述：

1	`dtype, shape, data`

如果 data 是 int，就从 aux_buffers[data] 拿对应 raw buffer：

1
2
3

buffer = self.aux_buffers[data]
arr = torch.frombuffer(buffer, dtype=torch.uint8)
return arr.view(torch_dtype).view(shape)

这正好对应 encoder 里记录的 index。

普通文本 Chat 会走哪条路径

普通 /v1/chat/completions 请求进入这段代码时，已经不是原始 OpenAI messages 了。

它已经被转换为：

1	`EngineCoreRequest`

纯文本 Chat 通常包含：

request_id: str
prompt_token_ids: list[int]
sampling_params: SamplingParams
arrival_time: float
lora_request: optional
trace_headers: optional

一般不会有：

1
2
3

prompt_embeds
mm_features tensor
large ndarray

所以普通文本 Chat 的 encoder 结果通常是：

1	`self.encoder.encode(request) -> [main_msgpack_payload]`

最终：

1 2	`msg = (engine_identity, ADD, main_msgpack_payload) len(msg) = 3`

走的是简单路径：

1	`self.input_socket.send_multipart(msg, copy=False)`

多模态 Chat 会走哪条路径

多模态请求可能带：

image tensor
audio tensor
video tensor
precomputed prompt embeds

这些 tensor 会出现在 EngineCoreRequest 的：

1 2	`mm_features prompt_embeds`

如果是默认 direct_rpc，大 tensor 会变成 ZMQ aux frames：

msg = (
  engine_identity,
  ADD,
  main_msgpack_payload,
  tensor_buffer_1,
  tensor_buffer_2,
  ...
)

于是：

1	`len(msg) > 3`

走：

1 2	`send_multipart(copy=False, track=True) add_pending_message(tracker, request)`

torch_shm 路径

如果配置了 mm_tensor_ipc = "torch_shm"，MPClient 会创建：

1 2	`tensor_ipc_sender = TensorIpcSender(tensor_queue) self.encoder = MsgpackEncoder(oob_tensor_consumer=tensor_ipc_sender)`

这时 _encode_tensor() 会先尝试：

1	`data = oob_consumer(obj)`

TensorIpcSender 会：

if not tensor.is_shared():
    tensor = tensor.share_memory_()

self.queue.put(TensorIpcData(..., tensor=tensor))
return metadata

也就是说 tensor 本体走：

1	`torch.multiprocessing.Queue + shared memory`

msgpack 主 payload 里只放：

1
2
3

sender_id
message_id
tensor_id

EngineCore 侧的 TensorIpcReceiver 根据这些 metadata 从 queue 里拿 tensor。

这条路径和 aux buffer 路径的区别是：

direct_rpc:
  tensor bytes as ZMQ multipart aux frames

torch_shm:
  tensor through torch shared memory queue
  ZMQ payload only carries tensor metadata

为什么不所有 tensor 都用共享内存

因为共享内存不是万能的。

默认 ZMQ direct RPC 更通用：

1. 可以跨进程，也可以跨节点走 TCP
2. 不依赖 torch shared memory queue
3. 对纯文本请求足够简单
4. 对普通 CPU tensor / ndarray 可以 multipart 发送

而 torch shared memory 更适合：

1
2
3

1. 同机进程间
2. 大多模态 tensor
3. 想避免经过 ZMQ 传大 tensor bytes

多机情况下，共享内存不能跨机器，因此仍然需要网络传输或其他跨节点 tensor/KV transport 机制。

这套机制的核心公式

对于输入消息：

1	`msg = (engine_identity, request_type, *encoded_frames)`

如果：

1	`encoded_frames = [main_payload]`

则：

$$
\text{len(msg)} = 2 + 1 = 3
$$

如果有 $k$ 个 aux buffers：

1	`encoded_frames = [main_payload, aux_1, aux_2, ..., aux_k]`

则：

$$
\text{len(msg)} = 2 + 1 + k = 3 + k
$$

所以：

$$
k = \text{len(msg)} - 3
$$

代码里的判断正是：

if k == 0:
    send without tracking
else:
    send with MessageTracker and keep request alive

机制小结

这套 ZMQ/msgpack 机制可以总结成几句话。

第一，aux_buffers 不是共享内存，它只是 encoder 返回的一组 multipart frames。

第二，主 msgpack payload 永远在 bufs[0]，tensor / ndarray 的 raw buffer 从 bufs[1] 开始。

第三，主 payload 里不会直接塞大 tensor，而是记录 dtype、shape 和 data index。

第四，len(msg) > 3 表示存在 ZMQ aux buffer。由于 copy=False，必须用 track=True，并且保留 request 引用，防止 backing memory 提前释放。

第五，torch_shm 是另一条可选路径：它用 torch shared memory queue 传 tensor，本体不作为 ZMQ aux frame 发送。

整体结构如下：

flowchart TB
    R["EngineCoreRequest"] --> E["MsgpackEncoder"]
    E --> P["bufs[0]: main msgpack payload"]
    E --> B1["bufs[1]: tensor buffer"]
    E --> B2["bufs[2]: ndarray buffer"]

    P --> Z["ZMQ multipart"]
    B1 --> Z
    B2 --> Z

    Z --> D["MsgpackDecoder"]
    D --> R2["EngineCoreRequest reconstructed"]

    Z -. "copy=False + aux buffers" .-> T["MessageTracker"]
    T --> K["pending_messages keeps request alive"]

这段设计体现了 vLLM 的工程取向：普通文本请求保持简单路径，多模态和 tensor 场景才启用复杂的 buffer 生命周期管理。它不是为了抽象漂亮，而是为了让高吞吐服务里的跨进程传输少一次拷贝，同时不牺牲内存安全。

#vLLM #code-reading #LLM serving #systems

vLLM Code Reading (Section 2): vLLM 的 ZMQ、msgpack 和 Tensor Buffer 传输机制

https://jeremyguo.space/2026/06/24/vllm-code-reading-section-2-zmq-msgpack-transport/

作者

郭俊毅 / JeremyGuo

发布于

2026年6月24日

许可协议

vLLM Code Reading (Section 3): Ray Core for vLLM 上一篇

vLLM Code Reading (Section 1): vLLM API Server 如何启动 EngineCore 下一篇