网络与 RPC：tonic/gRPC

节点之间怎么说话:从你 tcpdump 过的流量说起

运维 Ceph 时你抓过 messenger 的包,调过心跳超时,也一定处理过「OSD 之间通信异常」的告警。 分布式存储的第一块积木不是什么高深算法,是节点间通信:消息怎么编码、连接怎么管、 超时了算成功还是失败。这节课把 forge 的内部协议立起来 —— 从 L0 就占着位的 forge-proto 终于开工,同时新增 forge-node:节点守护进程,你以后 systemctl start 的那个东西。

选 gRPC/tonic 而不是自己搞 TCP 私有协议,理由和 L2 选 io_uring 而不是 SPDK 一样: 同一个问题的低一档解法,机制完整,工程量可控。Weka 用 DPDK 自研 RDMA 风格协议, 我们用 tonic 走内核 TCP —— 慢一档,但连接管理、多路复用、流控这些课一节不少。

protobuf 与协议演进:字段号就是盘上格式的兄弟

L1 设计盘上格式时你体会过「格式定了就改不动」的敬畏感。网络协议是同一个问题的镜像: 盘上格式要兼容过去的自己(老数据还在盘上),网络协议要兼容别的节点—— 滚动升级时,新老版本的 forge-node 会同时在线,谁也不能拒收谁的消息。

protobuf 的答案是字段号:每个字段绑定一个永不复用的编号,编码时只写编号不写名字。 新版本加字段,老节点解码时跳过不认识的编号;老版本缺字段,新节点读到默认值。

syntax = "proto3";
package forge.v1;

message PutBlobRequest {
  string key = 1;
  bytes  data = 2;
  uint32 crc32c = 3;   // 端到端校验:客户端算,服务端验,复用 forge-util
}

message PutBlobResponse {
  uint64 cluster_epoch = 1;  // 服务端当时的集群配置版本,placement 课兑现
}

message GetBlobRequest {
  string key = 1;
}

message GetBlobResponse {
  bytes  data = 1;
  uint32 crc32c = 2;
}

service BlobService {
  rpc PutBlob(PutBlobRequest) returns (PutBlobResponse);
  rpc GetBlob(GetBlobRequest) returns (GetBlobResponse);
}

// 成员管理协议,下一课实现;先把字段号占住
service GossipService {
  rpc Exchange(GossipRequest) returns (GossipResponse);
}

协议演进的三条铁律,和盘上格式的版本号规则一一对应:

1. 字段号只增不改不复用。删掉的字段用 reserved 3; 立墓碑,防止后人误用
2. 只加字段,不改类型。把 uint32 改成 uint64 在 proto3 里有些组合恰好兼容,但别赌
3. 语义变更就加新字段,老字段留着,让新代码同时认两种 —— 和盘上格式升级一个套路

tonic 服务端与客户端骨架

forge-proto 用 build script 在编译期把 .proto 生成 Rust 代码:

// crates/forge-proto/build.rs
fn main() -> Result<(), Box<dyn std::error::Error>> {
    tonic_build::configure()
        .compile_protos(&["proto/forge.proto"], &["proto"])?;
    Ok(())
}

服务端在新 crate forge-node 里,把 L1 的 BlobStore trait 挂到网络上 —— L2 做过的「同步核心 + 异步外壳」改造在这里兑现:

use forge_proto::forge::v1::blob_service_server::BlobService;
use forge_proto::forge::v1::{PutBlobRequest, PutBlobResponse, GetBlobRequest, GetBlobResponse};
use tonic::{Request, Response, Status};

pub struct BlobSvc<S> {
    store: std::sync::Arc<tokio::sync::Mutex<S>>,
}

#[tonic::async_trait]
impl<S: forge_store::BlobStore + Send + 'static> BlobService for BlobSvc<S> {
    async fn put_blob(
        &self,
        req: Request<PutBlobRequest>,
    ) -> Result<Response<PutBlobResponse>, Status> {
        let msg = req.into_inner();
        // 红线代码:先验校验和再落盘,网络上翻转的位不许进引擎
        if forge_util::crc32c_of(&msg.data) != msg.crc32c {
            return Err(Status::data_loss("crc32c mismatch, refusing to write"));
        }
        let mut store = self.store.lock().await;
        store
            .put(msg.key.as_bytes(), &msg.data)
            .map_err(|e| Status::internal(e.to_string()))?;
        Ok(Response::new(PutBlobResponse { cluster_epoch: 0 }))
    }

    async fn get_blob(
        &self,
        req: Request<GetBlobRequest>,
    ) -> Result<Response<GetBlobResponse>, Status> {
        // 读路径对称:从引擎读出后重算 crc 再回包
        // ...
        todo!()
    }
}

注意错误的翻译:引擎的 StoreError 到网络边界要映射成 gRPC 的 Status。 crc 不匹配用 data_loss,key 不存在用 not_found,盘写失败用 internal —— 客户端靠这个分类决定重试还是报警,别全都 internal 糊过去。

超时、重试、幂等:分布式第一课

单机上函数调用只有两种结果:成功或失败。跨网络多了第三种:不知道。 请求超时,可能是没送到,也可能是服务端已经写成功了、只是响应丢了 —— 你在运维里早见过这个:命令超时,重跑之前先查一眼上次到底执行没执行。

代码里的对策是一套组合拳,每个数字都要有出处:

• 连接超时 1s:同机房 TCP 握手是亚毫秒级,1s 还连不上就是节点或网络有病,别等
• 请求超时按数据量算:1MiB 对象在 10GbE 上传输约 1ms,加上服务端 fsync 几毫秒, 给 500ms 已经宽到荒唐 —— 超过它,等待的代价高于重试
• 只重试可重试的错误:超时、unavailable 重试;data_loss、参数错误重试一万次也是错
• 指数退避:50ms 起步,每次翻倍,上限 1s,最多 3 次 —— 防止全体客户端同步重试打死刚恢复的节点(你见过重试风暴)

而这一切的前提是幂等:重试等于同一个请求执行两次,结果必须一样。 PutBlob 按 key 覆盖写,天然幂等;将来要是加「追加写」这种非幂等接口, 就得带上客户端请求 ID 做去重 —— 现在协议里没有,是因为我们刻意让每个 RPC 都幂等。

大对象传输:流式接口与 zero-copy 的取舍

gRPC 单条消息默认上限 4MiB,而且整条消息要在内存里攒齐才能解码 —— put 一个 1GiB 对象显然不能塞一条消息。答案是流式 RPC: rpc PutBlobStream(stream PutChunk) returns (PutBlobResponse), 客户端切成 1MiB 一片连续发,服务端边收边写,内存占用恒定。

顺便把 HTTP/2 多路复用对存储流量的利弊说清,这是面试和设计评审都躲不开的问题:

• 利:一条 TCP 连接跑多个并发流,连接数从「客户端数 × 节点数 × 并发」降到「客户端数 × 节点数」, 连接管理和 keepalive 成本大降
• 弊:TCP 层的队头阻塞 —— 丢一个包,这条连接上所有流一起等重传; 大对象数据流还会挤占同连接上的小控制消息
• 对策:控制面和数据面用不同的连接(甚至不同端口)。你在 Ceph 里配 public network 和 cluster network 分离,是同一个思想的物理版

至于 zero-copy:prost 解码时会把 bytes 字段拷贝进新分配的 Vec,一次 1MiB 拷贝 在现代 CPU 上约 20~50µs,而网络往返是毫秒级 —— 这份拷贝暂时不值得优化。 Weka 用 DPDK 把这层也省了;我们记下这笔账,L5 调优时再看值不值得追。

目标:让 forge 的节点第一次通过网络开口说话 —— 起一个节点守护进程对外提供对象的存取服务,命令行客户端能把本地文件写进去、再原样读回来。

行为规格:
1. 协议用 gRPC + protobuf 定义。先立两个接口:写对象(带 key、数据、以及客户端自己算好的 crc32c)和读对象(按 key 取回数据与 crc32c)。写响应里预留一个「集群配置版本号」字段,放置那一课要用,现在恒为 0;
2. 字段号纪律:每个字段绑定一个永不复用的编号,废弃字段立墓碑,滚动升级时新老版本节点必须能互相收发,谁也不能拒收谁;
3. 服务端在落盘之前重算 crc32c 与客户端送来的值比对,不一致就拒绝写入 —— 网络上翻转的位不许进引擎;
4. 错误必须分类,不能全部糊成「内部错误」:数据校验失败、key 不存在、盘写失败要是三种不同的错误类别,客户端靠这个分类决定重试还是报警;
5. 客户端的超时与重试:连接超时 1 秒,单次请求超时 500 毫秒;只对超时和「服务不可用」重试,最多 3 次、50 毫秒起指数退避、上限 1 秒;数据校验失败和参数错误一次都不许重试;
6. 写接口必须幂等:同一个 key 重复写同样内容,和写一次的结果完全一样 —— 因为超时之后客户端无法区分「没写进去」和「写进去了但响应丢了」;
7. 大对象走流式分片传输,1MiB 一片边收边写,服务端内存占用不随对象大小增长;
8. 任何预期内的失败都要变成明确的错误返回给调用方,而不是让进程直接崩溃。

交付:节点守护进程(监听地址与数据目录用启动参数指定)+ 命令行的写入 / 读取子命令 + 集成测试(起服务端,写进去再读出来逐字节一致;故意送错 crc 时服务端拒绝)+ 说明你选了哪些库、为什么(生成代码的工具链、命令行参数解析这些你定)。

小结

• 字段号是网络协议的「盘上格式」:只增不改不复用,滚动升级时新老节点才能互认
• tonic 骨架把 L1 的 BlobStore 挂上网络;错误在边界处翻译成可分类的 Status,别糊成一团
• 分布式第一课:超时是第三种结果(不知道),重试的前提是幂等,退避防止重试风暴
• 大对象走流式分片;控制面和数据面分连接,对付 HTTP/2 的队头阻塞
• 节点能说话了,下一课解决更难的问题:怎么知道对方还活着