P2P Store — 基于 RDMA 的分布式数据传输 SDK

项目概述

P2P Store 是一个基于 RDMA 的高性能分布式数据传输 SDK，专为大规模 ML/RLHF 训练场景设计，用于在节点之间点对点传输 Tensor、文件和字节数据。

核心特性

• RDMA 零拷贝传输：基于 mooncake-transfer-engine，单卡可打满 200Gbps NIC
• Redis 元数据后端：支持 standalone / sentinel / cluster 三种部署模式
• 多 Transport 负载均衡：单 client 可绑定多个 RDMA QP，自带 round-robin / sticky-hash 策略
• 零拷贝 Paddle/NumPy 适配：put(key, paddle.Tensor) / put(key, np.ndarray) 不发生数据复制
• 批量 cache 与 streaming：内置 cache_all、delete_batch、P2P_STREAM_* 异步事件流

核心概念

• P2PClient：客户端，提供 put / get / list / delete / clear / cache_all 等 API
• Metaserver：元数据服务器，由 Redis 担任（独立部署或复用现有 Redis 集群）
• P2PConfig：统一配置类，Server 和 Client 共用
• Transport：传输层，支持 RDMA（生产）和 TCP（调试）

数据流概览

                    ┌──────────────────┐
                    │   Redis Cluster  │  ← 元数据（key → provider addr）
                    │  (metaserver)    │
                    └──┬───────────┬───┘
                       │           │
              register │           │ query
                       │           │
         ┌─────────────▼──┐    ┌───▼──────────────┐
         │  P2PClient A   │    │   P2PClient B    │
         │  (provider)    │    │   (consumer)     │
         └─────┬──────────┘    └──────────┬───────┘
               │                          │
               │     RDMA 直连传输         │
               └──────── (零拷贝) ────────┘

目录

• 安装
• 快速开始
• 配置说明
• Client API
• 环境变量
• 详细文档

安装

1. 系统依赖（RDMA 库）

apt install -y rdma-core libibverbs1 libmlx5-1 libibverbs-dev

2. 安装 p2pstore

pip install p2pstore

核心依赖（numpy>=1.20.0、redis>=6.4.0、xxhash>=3.0.0,<4.0.0）会随 pip install 自动拉取。

3. 单独安装 mooncake-transfer-engine（按 CUDA 主版本二选一）

从 0.1.15.post1 起，mooncake-transfer-engine 不再写进 p2pstore 的核心 dependencies，原因是它按 CUDA 主版本拆成了两个 PyPI 包（包名不同、Python namespace 都是 mooncake），统一声明会在错误的 CUDA 环境下拉到不能用的二进制。请在构建虚拟环境的脚本里按 CUDA 版本显式安装，两种方式任选其一：

方式 A：通过 extras（推荐，被 pyproject.toml 声明、版本随 p2pstore 集中维护）

# CUDA < 13
pip install "p2pstore[mooncake]"

# CUDA >= 13
pip install "p2pstore[mooncake-cuda13]"

方式 B：直接装 mooncake 包（脚本里显式指定，绕开 extras）

# CUDA < 13
pip install "mooncake-transfer-engine>=0.3.7,<0.4.0"

# CUDA >= 13
pip install "mooncake-transfer-engine-cuda13>=0.3.7,<0.4.0"

⚠️ 没有装任一 mooncake wheel 时，p2pstore 可以被 import，但 P2PClient 在初始化 transport 时会因 from mooncake.engine import TransferEngine 失败而报 ModuleNotFoundError。

4. Paddle Tensor 与 Safetensors 支持

Paddle Tensor 与 Safetensors Paddle 解码采用延迟导入，不属于核心依赖。需要这些能力时，业务环境应提供与 CUDA 版本匹配的 paddlepaddle 或 paddlepaddle-gpu，以及 safetensors。仅使用 bytes 或 ndarray 时无需安装它们。

快速开始

1. 准备 Redis（任意一种）

# 最简单：单机
docker run -d --name p2p-redis -p 6379:6379 redis:7

# 已有 Redis Cluster / Sentinel：直接复用即可

2. 最小 put / get 示例

import asyncio
import numpy as np
from p2pstore import P2PClient, P2PConfig

async def main():
    config = P2PConfig(
        metadata_server="redis://localhost:6379/0",
        protocol="rdma",
        device="mlx5_3",
    )
    client = P2PClient(config)

    # provider 端：写入数据
    await client.put("hello", np.ones((1024, 1024), dtype=np.float32))

    # consumer 端（另一进程/节点）：拉取数据
    arr = await client.get("hello")
    print(arr.shape, arr.dtype)

    client.close()

asyncio.run(main())

完整的命令行参数和典型场景示例参见 docs/用户使用文档.md。

配置说明

P2P Store 使用 P2PConfig（src/p2pstore/utils/config.py）进行配置管理，Server 和 Client 共用同一配置类。

主要配置项

metadata_server 支持的三种格式

配置示例

from p2pstore import P2PConfig

# Redis 单机
config = P2PConfig(
    metadata_server="redis://localhost:6379/0",
    protocol="rdma",
    device="mlx5_3",
)

# Redis Sentinel
config = P2PConfig(
    metadata_server="redis+sentinel://10.0.0.1:26379,10.0.0.2:26379/mymaster/0",
)

# Redis Cluster
config = P2PConfig(
    metadata_server="redis+cluster://10.0.0.1:6379,10.0.0.2:6379,10.0.0.3:6379",
)

Client API

P2PClient 的写入/读取/删除接口均为 async，需要在 asyncio 事件循环中使用。

初始化

import asyncio
from p2pstore import P2PClient, P2PConfig

config = P2PConfig(metadata_server="redis://10.0.0.1:6379/0")
client = P2PClient(config)

put / get（异步）

import paddle
import numpy as np

async def main():
    # 注册（put）：相同 key 会先删旧再写新
    tensor = paddle.to_tensor(np.random.randn(128, 128), dtype="float32")
    await client.put("my_tensor", tensor)

    # 拉取（get）：返回 Tensor 或写文件
    result = await client.get("my_tensor")
    await client.get("my_tensor", output_path="/tmp/save.bin")

asyncio.run(main())

删除 / 批量删除

await client.delete("my_tensor")
results = await client.delete_batch(["k1", "k2", "k3"])

清空所有数据

result = await client.clear()
# result: {"success": bool, "cleared": int, "failed": list[str]}

检查存在 / 列出

exists = await client.exists("my_tensor")       # async
files = client.list(prefix=None)                # 同步

exists 兼容三段式 data_id:gen_id:segment_id 和四段式 data_id:gen_id:turn_id:segment_id。四段式中的 turn_id 位于下标 2， 支持轮次 0；查询时去掉这一段，检查对应三段式存储记录已发布的 turn_id >= 请求轮次。轮次不足、缺失或无效，以及分片尚未齐全时，返回 False。 三段式仍只判断记录是否可见、分片是否完整，不要求元数据带有 turn_id。 其他不符合四段式轮次格式的普通 key 仍按原样查询。

exists = await client.exists("data:gen:segment")    # 兼容旧调用，不检查轮次
ready = await client.exists("data:gen:0:segment")   # 等待第 0 轮或更高轮次发布

批量 cache（高级）

# 把一组 prefix 下的 key 批量拉取到本地缓存
await client.cache_all(
    prefix="ckpt/",
    save_dir="/local/cache",
    concurrency=16,
    rank=0,
    world_size=8,
)

资源释放

client.close()  # 释放所有 transport、metadata client 和 buffer

环境变量

下表是运行期可调旋钮，按子系统分组。所有变量均为可选，未设置则使用 P2PConfig 默认值。

基础

RDMA / Transport

负载均衡

重试与超时

Redis / 缓存

Streaming Cache（高级）

详细文档

• 完整使用说明：docs/用户使用文档.md
• 配置类源码（最权威）：src/p2pstore/utils/config.py