Metadata-Version: 2.4
Name: p2pstore
Version: 0.1.15.post4
Summary: P2P Store - High-performance distributed data transfer system built on RDMA, supports Paddle Tensor, NumPy array transfer with Redis metadata backends
Author: Paddle
License-Expression: Apache-2.0
Keywords: p2p,rdma,distributed,tensor,transfer
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Operating System :: POSIX :: Linux
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: System :: Distributed Computing
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: numpy>=1.20.0
Requires-Dist: safetensors>=0.3.0
Requires-Dist: redis>=6.4.0
Provides-Extra: mooncake
Requires-Dist: mooncake-transfer-engine<0.4.0,>=0.3.7; extra == "mooncake"
Provides-Extra: mooncake-cuda13
Requires-Dist: mooncake-transfer-engine-cuda13<0.4.0,>=0.3.7; extra == "mooncake-cuda13"
Provides-Extra: paddle
Requires-Dist: paddlepaddle>=2.5.0; extra == "paddle"
Provides-Extra: redis
Requires-Dist: redis>=6.4.0; extra == "redis"
Provides-Extra: dev
Requires-Dist: pytest>=7.0.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.20.0; extra == "dev"
Requires-Dist: black; extra == "dev"
Requires-Dist: ruff; extra == "dev"
Provides-Extra: benchmark
Requires-Dist: paddlepaddle-gpu>=2.5.0; extra == "benchmark"
Dynamic: license-file

# 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](#client-api)
- [环境变量](#环境变量)
- [详细文档](#详细文档)

## 安装

### 1. 系统依赖（RDMA 库）

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

### 2. 安装 p2pstore

```bash
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 集中维护）**

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

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

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

```bash
# 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

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

## 快速开始

### 1. 准备 Redis（任意一种）

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

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

### 2. 最小 put / get 示例

```python
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`](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,...` |

### 配置示例

```python
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` 事件循环中使用。

### 初始化

```python
import asyncio
from p2pstore import P2PClient, P2PConfig

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

### put / get（异步）

```python
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())
```

### 删除 / 批量删除

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

### 清空所有数据

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

### 检查存在 / 列出

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

### 批量 cache（高级）

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

### 资源释放

```python
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`](docs/用户使用文档.md)
- 配置类源码（最权威）：[`src/p2pstore/utils/config.py`](src/p2pstore/utils/config.py)
