基于生产级 RL 分布式训练系统的实践经验，深入剖析”控制面/数据面分离”的架构设计原则。本文适合对分布式训练系统设计感兴趣的工程师阅读，尤其适合正在使用或评估 Ray 框架的 RL Infra 开发者。
TL;DR
模型权重走 Ray Object Store 实现零拷贝分发，采样数据走原生 TCP Socket 实现高吞吐流式传输——这不是历史遗留，而是基于数据量级差异、传输效率、控制面负载隔离和固定拓扑优势四个维度的刻意设计。 控制面（编排、调度、容错）和数据面（流式传输）分离，是分布式系统中最经典的设计模式之一。理解这条架构边界，是设计大规模分布式 RL 训练系统的关键。
一、问题引入
强化学习训练系统的分布式化是一个经典的工程问题：训练 Learner 在 GPU 节点上持续更新模型，数千个采样 Actor 在 CPU 节点上并行采集游戏轨迹。数据如何在 Learner 和 Actor 之间高效流转，直接影响训练的吞吐和稳定性。
在一个典型的分布式 RL 训练系统中，存在两条明显不同的数据传输通路：
| 数据类型 | 传输路径 | 传输机制 |
|---|---|---|
| 模型权重 | LearnerServer → ray.put() → ObjectRef → Actor |
Ray Object Store |
| 采样数据 | Actor → socket.send() → BatchReceiver → ReplayBuffer → Learner |
原生 TCP Socket |
核心问题：既然 Ray 已经接管了模型权重的分发，为什么采样数据要绕开 Ray、另起一条 TCP Socket 通路？这是一种刻意的架构设计，还是历史妥协？
二、控制面与数据面：完整架构全景
2.1 控制面 (Control Plane) —— Ray 负责
控制面负责系统的编排、调度、状态监控和故障恢复，数据量小、频率低，但对可靠性和一致性要求高。
┌─────────────────────────┐
│ Ray GCS │
│ (命名 Actor 注册、 │
│ 集群拓扑、资源管理) │
└───────────┬─────────────┘
│
┌───────────────────────────┼───────────────────────────┐
│ │ │
▼ ▼ ▼
┌─────────────────┐ ┌─────────────────────────┐ ┌─────────────────┐
│ LearnerServer │ │ ModelDispatcher │ │ Sampling Actor │
│ (GPU 节点) │ │ (调度节点) │ │ (CPU 节点 × N) │
│ │ │ │ │ │
│ • 训练模型 │◄──►│ • 权重缓存与分发 │◄──►│ • 环境交互 │
│ • ray.put 权重 │Ray │ • Actor 生命周期管理 │Ray │ • RL 推理 │
│ • 回放缓冲区 │ │ • 节点拓扑感知 │ │ • 轨迹采集 │
└─────────────────┘ │ • 故障检测与恢复 │ └─────────────────┘
└─────────────────────────┘
控制面数据流 (Ray):
LearnerServer ──ray.put(weights)──► ObjectRef ──actor.act.remote(ObjectRef)──► Actor
│
ray.get(ObjectRef) ◄─────┘
数据特征: 低频 (每 3 秒一次), 中数据量 (50-200MB/次), 全集群共享一份
控制面通过 Ray 承载的具体功能：
| 功能 | 实现机制 |
|---|---|
| 模型权重零拷贝分发 | ray.put() + ObjectRef 传递 |
| 任务调度与异步编排 | actor.act.remote() + ray.wait() |
| 节点亲和性绑定 | NodeAffinitySchedulingStrategy |
| Actor 生命周期管理 | ray.kill() + 异常捕获重建 |
| 集群拓扑感知 | ray.nodes() 查询 |
| 命名服务与单例 | Named Actor + lifetime="detached" |
2.2 数据面 (Data Plane) —— Socket 负责
数据面负责采样数据的流式传输，数据量大、频率高，追求低延迟和高吞吐。
┌──────────────────┐ ┌──────────────────────┐
│ Sampling Actor │ │ LearnerServer 节点 │
│ (CPU 节点) │ │ │
│ │ TCP Socket │ BatchReceiverServer │
│ act() 执行: │ ═══════════════════►│ (多进程 TCP 服务) │
│ ┌────────────┐ │ 采样数据流 │ │ │
│ │ 采集 chunk │ │ zlib 压缩 │ ▼ │
│ │ trajectory │──┼──────────────────►│ BatchMaker │
│ │ 打包压缩 │ │ pickle 序列化 │ (解压→组batch→ │
│ └────────────┘ │ │ 共享内存) │
│ │ │ │ │
│ conns[name] │ │ ▼ │
│ .send(data) │ │ ReplayBufferServer │
│ │ │ (环形缓冲区) │
│ │ │ │ │
│ │ │ ▼ │
│ │ │ Learner │
│ │ │ (GPU 训练) │
└──────────────────┘ └──────────────────────┘
数据面数据流 (Socket):
Actor ──pickle+zlib──► conn.send() ──TCP──► BatchReceiver.recv()
│
BatchMaker (解压+组batch)
│
ReplayBuffer (环形缓冲区, maxlen=133120)
│
Learner.forward_loss() (GPU 训练)
数据特征: 高频 (数千 Actor 并发), 大数据量 (每秒数 GB), 流式点对点传输
数据面 Socket 通信的关键实现：
class Actor4Sample:
def __init__(self, addresses, packed_len=1):
# 初始化时建立到 BatchReceiver 的 TCP 长连接
self.conns = {
name: connect_socket_connection(address[0], address[1])
for name, address in addresses.items()
}
def act(self, model_id_weights):
moms, sample_info_dict = self.sample(model_id_weights)
# 采样数据直接通过 socket 发送，不经 Ray
packed_moms = [zlib.compress(pickle.dumps(moms[i:i+self.packed_len]))
for i in range(0, len(moms), self.packed_len)]
self.conns[self.name].send(packed_moms)
return sample_info_dict
三、为什么采样数据不走 Ray？逐层分析
3.1 数据量级差异：模型权重 vs 采样数据
| 维度 | 模型权重 | 采样数据 |
|---|---|---|
| 单次数据量 | 50-200MB | 单个 chunk 压缩后约 10-50MB |
| 产生频率 | 每 100 Learner step（约 0.5-1s） | 每 15-30s（一个 chunk） |
| 并发 Producer 数 | 1（LearnerServer） | 数千（采样 Actor） |
| 集群总吞吐 | ~50-200MB/s | 数千 Actor × 1-2MB/s ≈ 数 GB/s |
| 共享模式 | 一份权重，全集群 Actor 共享 | 每个 Actor 产生独立数据流 |
如果采样数据走 Ray Object Store（Plasma）：
- Plasma 容量问题：Ray 的 Plasma object store 默认与本节点共享内存大小绑定（通常几十 GB）。数 GB/s 的采样数据灌入，会在几秒内打满 Plasma。
- Eviction 连锁反应：Plasma LRU eviction 触发时，模型权重 ObjectRef 可能被踢出，导致 Actor 的
ray.get(weights)回退到从远端 owner 重新拉取，阻塞加剧。 - Spilling 开销：Object store 溢出到磁盘时，I/O 延迟远高于内存，采样数据的时效性严重受损。
3.2 传输路径复杂度对比
Socket 直连路径（当前方案）：
Actor.sample() → pickle + zlib → conn.send(buffer)
│
TCP 直连, 一次 send/recv
│
BatchReceiver.recv() → BatchMaker → ReplayBuffer
Ray 通路路径（假设方案）：
Actor.sample() → ray.put(samples) → ObjectRef
│ │
│ ① GCS 元数据广播（owner, size, location）
│ ② 数据写入本地 Plasma
│ ③ 引用计数管理
└─────────────────────────────────────┘
ModelDispatcher.ray.wait(futures) → ray.get(ObjectRef)
│ │
│ ① 反序列化 trajectory 数据
│ ② 再次 ray.put 到 ReplayBuffer 所在节点
│ ③ 跨节点数据传输（可能再次）
└─────────────────────────────────────┘
ReplayBuffer.ray.get(ObjectRef) → 解压 → 入队
每一步 Ray 都要做引用计数管理、owner 节点 pinning、元数据广播到 GCS。对于大流量流式数据，这些开销是不必要的复杂度。
3.3 控制面负载隔离
这是最关键的架构理由。如果把数据面也塞进 Ray：
控制面 (Ray GCS) 需要处理:
• 每秒成千上万个 ObjectRef 的创建/销毁元数据广播
• Object 的分布式寻址（哪个节点有这份数据？）
• 引用计数维护（owner 死了怎么办？）
• Plasma 内存分配/回收
结果: GCS 被数据面流量淹没
→ 命名 Actor 注册延迟飙升（影响推理服务 get-or-create）
→ ray.wait() 轮询延迟增加（影响采样调度）
→ 故障恢复响应变慢（ActorDied 检测延迟）
控制面和数据面的流量特征完全不同：
| 特征 | 控制面 | 数据面 |
|---|---|---|
| 流量模式 | 低频、突发（权重同步、Actor 创建/销毁） | 高频、持续（每秒数 GB） |
| 延迟要求 | 中等（秒级可接受） | 低（采样→训练的端到端延迟影响训练效果） |
| 可靠性要求 | 高（丢失 = 训练中断） | 中（个别样本丢失可接受，buffer 有冗余） |
| 拓扑结构 | 一对多广播 | 固定点对点流 |
3.4 固定拓扑的优势
Actor → BatchReceiver 是固定拓扑：
# 典型配置示例
class LearnerServerCFG:
num_replay_buffer_servers: int = 16 # 多个 ReplayBuffer
batch_receiver_num: int = 1 # BatchReceiver 数量
remote_replay_buffer_server: bool = True # 远程部署
class ModelDispatcherCFG:
num_actors: int = 5000+ # 大规模采样 Actor
数千个 Actor → BatchReceiver → 多个 ReplayBuffer → Learner。这种固定拓扑下，Socket 长连接是最优解：
- 零寻址开销：Actor 初始化时就连到固定的
(ip, port)，不需要分布式对象寻址。 - 零额外序列化：
pickle.dumps+zlib.compress直接写入 socket buffer，Ray 的 ObjectRef 传递反而要多一层序列化。 - 背压自然：TCP 自带流控，BatchReceiver 的
mp.Queue(maxsize=2)满了就 drop 数据，不会像 Ray Object Store 那样触发连锁 eviction。
四、什么情况下应该切换到 Ray 通路？
Ray Object Store 适合的场景特征：
| 特征 | 适合 Ray | 适合 Socket |
|---|---|---|
| 数据共享模式 | 一对多（一份数据多消费者） | 一对一（点对点流） |
| 数据量级 | 中（MB 级） | 大（GB/s 级） |
| 频率 | 低频（秒级） | 高频（持续） |
| 拓扑 | 动态（消费者动态加入） | 静态固定 |
| 持久化需求 | 需要 Ray 的分布式对象语义 | 不需要，直接消费即可 |
应该走 Ray 通路的数据：
- ✅ 模型权重（一对多共享，当前方案）
- ✅ 训练 metrics / tensorboard 日志（小数据，需要聚合）
- ✅ 小型配置同步（低频，需要一致性保证）
不应该走 Ray 通路的数据：
- ❌ 采样 trajectory（大数据量、高频、固定拓扑）
- ❌ 推理中间结果（实时性要求高，Socket/共享内存更优）
- ❌ 日志流（量大且不需要 Ray 的分布式语义）
五、这种设计的局限性
诚实地说，控制面/数据面分离也带来了额外的复杂度和边界条件：
5.1 运维复杂度
需要同时维护两套通信基础设施：Ray 集群的健康监控（GCS、raylet）和 Socket 连接的健康监控（TCP 长连接保活、断线重连）。当出现网络分区时，两条通路的故障模式不同——控制面可能通过 Ray 的 ActorDiedError 检测到异常，但数据面的 Socket 连接可能处于半开状态（TCP half-open），需要额外的应用层心跳。
5.2 BatchReceiver 单点瓶颈
当前架构中数千个 Actor → 少量 BatchReceiver，虽然 Receiver 是多进程 + 多线程架构，但在极端吞吐下（所有 Actor 同时完成 chunk），mp.Queue(maxsize=2) 的背压策略会主动丢数据。这在当前 data_reuse=1 的配置下可接受（丢弃个别样本不影响训练），但如果未来需要样本全量保留（如训练 offline RL），就需要对 BatchReceiver 做水平扩展。
5.3 跨机房部署的挑战
当 Actor 和 BatchReceiver 跨机房部署时，TCP 长连接的质量取决于跨机房专线带宽和延迟。如果网络抖动导致 TCP 重传，采样数据的端到端延迟会增加。相比之下，Ray 的 object store 在跨机房场景下有更成熟的数据传输调度机制。这种情况下可能需要重新评估 Socket 和 Ray 的取舍。
六、总结：控制面/数据面分离的设计原则
┌──────────────────────────────────────────────────────────────┐
│ 控制面 (Ray) │
│ ┌──────────┐ ┌──────────────┐ ┌─────────────────────┐ │
│ │ 权重分发 │ │ Actor 编排 │ │ 故障检测与恢复 │ │
│ │ ray.put │ │ .remote() │ │ ActorDiedError 处理 │ │
│ │ ObjectRef│ │ ray.wait() │ │ ray.kill() + 重建 │ │
│ └──────────┘ └──────────────┘ └─────────────────────┘ │
│ │
│ 特点: 低频、低数据量、高可靠性要求、动态拓扑 │
└──────────────────────────────────────────────────────────────┘
│
│ 架构边界清晰
│
┌──────────────────────────────────────────────────────────────┐
│ 数据面 (Socket) │
│ ┌──────────┐ ┌──────────────┐ ┌─────────────────────┐ │
│ │ 采样数据 │ │ Batch 组装 │ │ ReplayBuffer 存储 │ │
│ │ conn.send│ │ pickle+zlib │ │ 环形缓冲区 133120 │ │
│ │ TCP 直连 │ │ BatchMaker │ │ Learner GPU 训练 │ │
│ └──────────┘ └──────────────┘ └─────────────────────┘ │
│ │
│ 特点: 高频、大数据量、低延迟要求、固定拓扑 │
└──────────────────────────────────────────────────────────────┘
核心原则：
- 数据量决定通道：GB/s 级的持续流式数据不走对象存储。
- 拓扑决定协议：固定点对点拓扑用 Socket，动态多消费者用 Ray ObjectRef。
- 控制面不承载数据面流量：GCS 的职责是编排和调度，不是数据传输。
- 各层做各层最擅长的事：Ray 擅长分布式编排，TCP Socket 擅长流式传输。
这不是历史遗留，是刻意的架构选择——控制面/数据面分离是分布式系统中最经典的设计模式之一。
参考文献
- Ray Architecture Whitepaper — Ray 官方架构文档，包含 GCS、Object Store、Core Worker 的设计细节
- Ray Object Store (Plasma) Documentation — ObjectRef 引用计数、ownership model、内存管理机制
- Control Plane vs Data Plane in Distributed Systems — 控制面/数据面分离的经典设计模式