缓冲区、原始数据与 Copy 原理详解一、什么是缓冲区缓冲区 一块连续的内存区域 用来临时存储数据 内存地址: 0x1000 0x1001 0x1002 0x1003 0x1004 内存内容: [ 72 ] [ 101] [ 108] [ 108] [ 111] H e l l o buffer bytearray(bHello) └── 就是这样一块内存 └── bytearray 是可变的字节数组 └── Python 管理这块内存的生命周期二、原始数据到缓冲区的过程原始数据Python 对象# EngineCoreOutputs 是一个Python对象outputsEngineCoreOutputs(outputs[RequestOutput(request_idreq1,token_ids[1234,5678],finishedFalse,)],scheduler_statsSchedulerStats(num_running_reqs3,num_waiting_reqs1,))# 这个对象在内存中是分散的:# outputs 对象头 → 地址 0x2000# outputs列表 → 地址 0x3000# RequestOutput对象 → 地址 0x4000# token_ids列表 → 地址 0x5000# scheduler_stats → 地址 0x6000# 各个字段分散在堆内存各处Python对象在内存中的样子: 堆内存: ┌─────────────────────────────────────────────────┐ │ 0x2000: EngineCoreOutputs │ │ ├── outputs_ptr ──────────────────────►0x3000 │ │ └── stats_ptr ───────────────────────►0x6000 │ │ │ │ 0x3000: list │ │ └── item[0]_ptr ─────────────────────►0x4000 │ │ │ │ 0x4000: RequestOutput │ │ ├── request_id req1 │ │ ├── token_ids_ptr ───────────────────►0x5000 │ │ └── finished False │ │ │ │ 0x5000: list [1234, 5678] │ │ │ │ 0x6000: SchedulerStats │ │ ├── num_running 3 │ │ └── num_waiting 1 │ └─────────────────────────────────────────────────┘ 问题: 网络传输需要连续的字节流 不能直接传输这种分散的Python对象 需要序列化成连续的字节序列化对象 → 连续字节encoderMsgpackEncoder()bufferbytearray()buffersencoder.encode_into(outputs,buffer)序列化过程: Python对象 (分散内存) buffer (连续内存) ┌──────────────────┐ ┌──────────────────────────────┐ │ 0x2000: outputs │ │ 0x7000: │ │ 0x3000: list │ encode │ [82][a7]outputs[91][...req1 │ │ 0x4000: ReqOut │ ───────► │ ...][1234][5678][a5]stats[03]│ │ 0x5000: tokens │ │ [01] │ │ 0x6000: stats │ │ ← 连续的msgpack字节流 → │ └──────────────────┘ └──────────────────────────────┘ 序列化后: └── 所有数据打包成连续字节 └── 可以直接通过网络传输 └── 接收方可以反序列化还原Python对象三、copyTrue 的原理coord_socket.send(coord_buf,copyTrue)copyTrue 的内存操作: Step1: Python 持有 buffer ┌─────────────────────────────────┐ │ Python内存 (0x7000): │ │ [82][a7][o][u][t][p][u][t][s] │ │ [91][...][1234][5678][...] │ └─────────────────────────────────┘ │ │ copyTrue ▼ ZMQ 复制一份到自己的内存 ┌─────────────────────────────────┐ │ ZMQ内存 (0x8000): │ │ [82][a7][o][u][t][p][u][t][s] │ ← 完整拷贝 │ [91][...][1234][5678][...] │ └─────────────────────────────────┘ Step2: send() 返回后 ├── Python 的 buffer 可以立即释放/复用 ├── ZMQ 用自己的副本发送数据 └── 两块内存互不影响 Step3: ZMQ 后台线程发送 ZMQ内存 (0x8000) ──网络──► API Server Step4: 发送完成 ZMQ内存 (0x8000) 释放copyTrue 优缺点: 优点: ├── 简单安全 ├── send()返回后立即可以复用buffer └── 不需要追踪发送状态 缺点: └── 额外的内存分配和数据拷贝 数据量大时影响性能 例如: 1MB的输出数据 → Python buffer 1MB → ZMQ buffer 1MB (拷贝) → 总共 2MB 内存四、copyFalse 的原理零拷贝trackersockets[client_index].send_multipart(buffers,copyFalse,trackTrue,)copyFalse 的内存操作: Step1: Python 持有 buffer ┌─────────────────────────────────┐ │ Python内存 (0x7000): │ │ [82][a7][o][u][t][p][u][t][s] │ │ [91][...][1234][5678][...] │ └─────────────────────────────────┘ Step2: send(copyFalse) 调用 ZMQ 不复制数据 只记录 Python buffer 的地址: ┌─────────────────────────────────┐ │ ZMQ 内部记录: │ │ data_ptr 0x7000 ← 指向Python │ │ data_len 1024 │ │ 直接用这个地址发送 │ └─────────────────────────────────┘ Step3: ZMQ 后台线程发送中 Python内存 (0x7000) ──网络──► API Server ↑ ★ 此时绝对不能释放或修改这块内存! ZMQ 还在读取它 Step4: 发送完成 tracker.done True └── 此刻才能安全释放/复用 Python buffercopyFalse 优缺点: 优点: ├── 零内存拷贝 ├── 节省内存 └── 提高吞吐量 例如: 1MB的输出数据 → Python buffer 1MB → ZMQ 直接用这1MB → 总共 1MB 内存 缺点: ├── 必须追踪发送状态 ├── buffer 不能提前释放 └── 代码更复杂五、tracker 和 pending 的作用trackersocket.send_multipart(buffers,copyFalse,trackTrue)ifnottracker.done:pending.appendleft((tracker,outputs,buffer))eliflen(reuse_buffers)max_reuse_bufs:reuse_buffers.append(buffer)tracker 的状态变化: T0ms: send_multipart() 调用 tracker.done False ← 发送中 buffer 不能动! pending [(tracker, outputs, buffer)] T1ms: ZMQ后台线程读取 buffer 数据 通过网络发送 tracker.done False ← 还在发 T5ms: 网络发送完成 tracker.done True ← 发完了! buffer 现在可以安全复用 下一轮循环: while pending and pending[-1][0].done: _, _, buf pending.pop() reuse_buffers.append(buf) ← 放回缓冲池pending deque 的结构: appendleft 从左边加入 (最新的在左) pop 从右边取出 (最旧的在右) 时间轴: T1: send buf0 pending [buf0] T2: send buf1 pending [buf1, buf0] T3: send buf2 pending [buf2, buf1, buf0] T4: buf0发完 pending[-1] buf0, doneTrue pending.pop() → 回收buf0 pending [buf2, buf1] T5: buf1发完 pending[-1] buf1, doneTrue pending.pop() → 回收buf1 pending [buf2]六、缓冲区复用池原理reuse_buffers:list[bytearray][]max_reuse_bufs4# 取出复用:bufferreuse_buffers.pop()ifreuse_bufferselsebytearray()# 放回复用:iflen(reuse_buffers)max_reuse_bufs:reuse_buffers.append(buffer)没有复用池时: 每次循环: T1: buffer bytearray() ← malloc 分配内存 encode_into(outputs, buffer) send(buffer) 发完后 buffer 被GC回收 ← free 释放内存 T2: buffer bytearray() ← malloc 再次分配 ... 问题: └── 频繁 malloc/free └── 内存碎片 └── GC压力大 └── 高吞吐时性能下降 有复用池时: 初始: reuse_buffers [] T1: buffer bytearray() ← 第一次才malloc encode_into(outputs, buffer) 发完后: reuse_buffers [buffer] ← 不释放留着 T2: buffer reuse_buffers.pop() ← 直接复用! encode_into(outputs, buffer) ← 覆盖写入新数据 发完后: reuse_buffers [buffer] T3: buffer reuse_buffers.pop() ← 再次复用! ... 优势: └── 只有第一次 malloc └── 之后一直复用同一块内存 └── 无GC压力 └── 高吞吐时性能稳定encode_into 覆盖写入: 第一次: buffer bytearray() (空的, 长度0) encode_into(outputs1, buffer) buffer bytearray([82, a7, ...]) (长度1024) 第二次(复用): buffer bytearray([82, a7, ...]) (上次的数据) encode_into(outputs2, buffer) → 清空buffer → 重新写入新数据 buffer bytearray([91, b3, ...]) (新数据, 覆盖了旧的)七、完整内存生命周期一次输出的完整内存流转: 主线程: outputs EngineCoreOutputs(...) ← Python对象创建 output_queue.put((client_idx, outputs)) 输出线程: output output_queue.get() ① 获取缓冲区: buffer reuse_buffers.pop() ← 复用 或 新建bytearray ② 序列化: encode_into(outputs, buffer) 此时: Python对象(分散) ──序列化──► buffer(连续字节) ③ 零拷贝发送: tracker socket.send_multipart(buffer, copyFalse) 此时: buffer内存 ──ZMQ直接引用──► 网络发送中 ↑ 不能动! ④ 追踪: if not tracker.done: pending.appendleft((tracker, outputs, buffer)) ↑ 保持buffer的引用防止被GC ⑤ 发送完成(下一轮检查): tracker.done True pending.pop() ← 移除追踪 reuse_buffers.append(buffer) ← 放回复用池 ⑥ del outputs ← Python对象可以被GC了 buffer准备好给下一次序列化使用八、一句话总结缓冲区: └── 连续内存块存放序列化后的字节数据 └── bytearray 在Python堆上管理 copyTrue: └── ZMQ复制一份数据到自己的内存 └── send()返回后buffer立即可用 └── 内存占用翻倍但简单安全 copyFalse (零拷贝): └── ZMQ直接引用Python的buffer内存 └── 不复制节省内存和时间 └── 必须等tracker.done才能复用buffer 缓冲区复用池: └── 避免频繁malloc/free └── 发送完成后buffer放回池子 └── 下次直接取出覆盖写入 └── 最多保留4个防止内存无限增长