p2pstore 0.1.15.post3

P2P Store - High-performance distributed data transfer system built on RDMA, supports Paddle Tensor, NumPy array transfer with Redis metadata backends

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、safetensors>=0.3.0、redis>=6.4.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。

3. 可选 extras

pip install "p2pstore[paddle]"     # 启用 Paddle Tensor 支持

快速开始

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	Redis 元数据后端地址，必填。格式见下表	—
local_host	本地 IP（用于 RDMA 绑定）	从 POD_IP 读取，否则 127.0.0.1
protocol	传输协议（rdma / tcp）	"rdma"
device	RDMA 设备名，环境变量 P2P_RDMA_DEVICE 可覆盖	"mlx5_3"
max_retries	最大重试次数	3
retry_interval	重试间隔（秒）	5
enable_watch	是否启用 Redis Pub/Sub 监听	True
redis_key_ttl	Redis key TTL（秒），0 = 永不过期	0
redis_provider_ttl	Provider TTL（秒），用于异常退出自清理	3600
redis_key_prefix	Redis key 前缀	"p2p:"
redis_socket_timeout	Redis 连接超时（秒）	10.0
put_enable_auto_cache	put 时是否自动写本地缓存	False

metadata_server 支持的三种格式

模式	URL 格式
单机	redis://[[user:]password@]host:port[/db][?param=value]
Sentinel	redis+sentinel://[[user:]password@]host:port,.../service_name[/db]
Cluster	redis+cluster://[[user:]password@]host:port,host:port,...

配置示例

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)                # 同步

批量 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 默认值。

基础

变量	作用	默认
POD_IP	本节点 IP（RDMA 绑定）	127.0.0.1
POD_0_IP	部分示例脚本使用的 metaserver 节点 IP	127.0.0.1

RDMA / Transport

变量	作用	默认
P2P_RDMA_DEVICE	指定 RDMA 设备名，覆盖 config.device	—
P2P_RDMA_INSTANCES	每个 client 启动的 transport 实例数（1-64），多实例分散单 QP 压力	1
MC_GID_INDEX	mooncake GID 索引	—

负载均衡

变量	作用	默认
P2P_LB_SEND_STRATEGY	PUT 路由策略：random / round_robin	random
P2P_LB_RECV_STRATEGY	GET 路由策略：sticky_hash / round_robin	sticky_hash

重试与超时

变量	作用
P2P_RDMA_TIMEOUT_MULTIPLIER	RDMA 超时乘数
P2P_RDMA_MAX_RETRIES	RDMA 单次操作最大重试次数
P2P_RDMA_BASE_RETRY_INTERVAL	指数退避基础间隔
P2P_RDMA_RETRY_MULTIPLIER	指数退避乘数

Redis / 缓存

变量	作用
P2P_REDIS_KEY_TTL	Redis key TTL（秒），覆盖 config.redis_key_ttl
P2P_QUERY_TIMEOUT	元数据查询超时
P2P_LOCAL_CACHE_DTYPE	本地缓存默认 dtype

Streaming Cache（高级）

变量	作用
P2P_STREAM_ENABLED	是否启用 Redis Stream 事件流
P2P_STREAM_KEY	Stream 的 key 名
P2P_STREAM_MAXLEN	Stream 最大长度
P2P_STREAM_BATCH_SIZE	单批拉取大小
P2P_STREAM_BLOCK_MS	阻塞等待毫秒数
P2P_CACHE_CONCURRENCY	Cache 并发度
P2P_CACHE_SAVE_DIR	Cache 落盘目录
P2P_CACHE_VERIFY	是否校验缓存内容

详细文档

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