介紹

在上篇文章中，我們講到 TiKV 為了支持 gRPC，我們造了個輪子 gRPC-rs，這篇文章簡要地介紹一下這個庫。首先我們來聊聊什么是 gRPC。gRPC 是 Google 推出的基于 HTTP2 的開源 RPC 框架，希望通過它使得各種微服務之間擁有統(tǒng)一的 RPC 基礎設施。它不僅支持常規(guī)的平臺如 Linux，Windows，還支持移動設備和 IoT，現(xiàn)有十幾種語言的實現(xiàn)，現(xiàn)在又多了一種語言 Rust。

gRPC 之所以有如此多的語言支持，是因為它有一個 C 寫的核心庫(gRPC core)，因此只要某個語言兼容 C ABI，那么就可以通過封裝，寫一個該語言的 gRPC 庫。Rust 對 C 有良好的支持，gRPC-rs 就是對 gRPC core ABI 的 Rust 封裝。

Core 能異步處理 RPC 請求，在考慮到 Rust 中已有較為成熟的異步框架 Futures，我們決定將 API 設計成 Future 模式。

gRPC-rs 架構圖

我們將根據架構圖從底向上地講一下，在上一篇文章中已經討論過傳輸層和協(xié)議，在這就不再贅述。

gRPC Core

Core 中有幾個比較重要的對象：

• Call 以及 4 種類型 RPC： Call 代表了一次 RPC，可以派生出四種類型 RPC，

  • Unary： 這是最簡單的一種 RPC 模式，即一問一答，客戶端發(fā)送一個請求，服務端返回一個回復，該輪 RPC 結束。
  • Client streaming： 這類的 RPC 會創(chuàng)建一個客戶端到服務端的流，客戶端可以通過這個流，向服務端發(fā)送多個請求，而服務端只會返回一個回復。
  • Server streaming： 與上面的類似，不過它會創(chuàng)建一個服務端到客戶端的流，服務端可以發(fā)送多個回復，

  • Bidirectional streaming： 如果說上面兩類是單工，那么這類就是雙工了，客戶端和服務端可以同時向對方發(fā)送消息。

    值得一提的是由于 gRPC 基于 HTTP2，它利用了 HTTP2 多路復用特性，使得一個 TCP 連接上可以同時進行多個 RPC，一次 RPC 即為 HTTP2 中的一個 Stream。

• Channel： 它是對底層鏈接的抽象，具體來說一個 Channel 就是一條連著遠程服務器的 TCP 鏈接。

• Server： 顧名思義，它就是 gRPC 服務端封裝，可以在上面注冊我們的服務。
• Completion queue: 它是 gRPC 完成事件隊列，事件可以是收到新的回復，可以是新來的請求。

簡要介紹一下 Core 庫的實現(xiàn)，Core 中有一個 Combiner 的概念，Combiner 中一個函數(shù)指針或稱組合子（Combinator）隊列。每個組合子都有特定的功能，通過不同的組合可以實現(xiàn)不同的功能。下面的偽碼大概說明了 Combiner 的工作方式。

class combiner {mpscq q; // multi-producer single-consumer queue can be made non-blockingstate s; // is it empty or executingrun(f) {if (q.push(f)) {// q.push returns true if it's the first thingwhile (q.pop(&f)) { // modulo some extra work to avoid racesf();}}} }復制代碼

Combiner 里面有一個 mpsc 的無鎖隊列 q，由于 q 只能有一個消費者，這就要求在同一時刻只能有一個線程去調用隊列里面的各個函數(shù)。調用的入口是 run() 方法，在 run() 中各個函數(shù)會被序列地執(zhí)行。當取完 q 時，該輪調用結束。假設一次 RPC 由六個函數(shù)組成，這樣的設計使這組函數(shù)（RPC）可以在不同的線程上運行，這是異步化 RPC 的基礎。

Completion queue（以下簡稱 CQ）就是一個 Combiner，它暴露出了一個 next()借口，相當于 Combiner 的 run()。由于接口的簡單，Core 內部不用開啟額外線程，只要通過外部不斷調用 next() 就能驅動整個 Core。

所有的 HTTP2 處理，Client 的 RPC 請求和 Server 的 RPC 連接全是通過一個個組合子的不同組合而構成的。下面是一次 Unary 的代碼。它由6個組合子組成，這些組合子作為一個 batch 再加上 Call 用于記錄狀態(tài)，兩者構成了這次的 RPC。

grpc_call_error grpcwarp_call_start_unary(grpc_call *call, grpcsharp_batch_context *tag) {grpc_op ops[6];ops[0].op = GRPC_OP_SEND_INITIAL_METADATA;...ops[1].op = GRPC_OP_SEND_MESSAGE;...ops[2].op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;...ops[3].op = GRPC_OP_RECV_INITIAL_METADATA;...ops[4].op = GRPC_OP_RECV_MESSAGE;...ops[5].op = GRPC_OP_RECV_STATUS_ON_CLIENT;return grpcwrap_call_start_batch(call, ops, tag); }復制代碼

用 Rust 封裝 Core

介紹完 Core，現(xiàn)在說一下如何用 Rust 封裝它。這一層封裝并不會產生額外的開銷，不像有的語言在調用 C 時會有類型的轉換或者 runtime 會有較大開銷，在 Rust 中開銷微乎其微，這得益于 Rust 用 llvm 做編譯器后端，它對 C 有良好的支持，Rust 調用 C ABI 就像調用一個普通的函數(shù)，可以做到 Zero-cost。

同時用 Rust 封裝 C ABI 是一件很簡單的事情，簡單到像黑魔法。比如封裝 CQ next():

C：

grpc_event grpc_completion_queue_next(grpc_completion_queue *cq,gpr_timespec deadline,void *reserved);復制代碼

Rust：

extern "C" {pub fn grpc_completion_queue_next(cq: *mut GrpcCompletionQueue,deadline: GprTimespec,reserved: *mut c_void)-> GrpcEvent; }復制代碼

接著我們看看如何封裝 C 的類型。繼續(xù)以 next() 為例子：

C：

// CQ 指針 grpc_completion_queue *cq;// grpc_event 結構體 struct grpc_event {grpc_completion_type type;int success;void *tag; };復制代碼

Rust：

pub enum GrpcCompletionQueue {}#[repr(C)] pub struct GrpcEvent {pub event_type: GrpcCompletionType,pub success: c_int,pub tag: *mut c_void, }復制代碼

CQ 在 Core 的 ABI 中傳遞的形式是指針，Rust Wraper 無須知道 CQ 具體的內部結構。對于這種情況，Rust 推薦用無成員的枚舉體表示，具體好處有兩個，第一，由于沒有成員，我們無法在 Rust 中構建該枚舉體的實例，第二，Type safe，當傳遞了一個錯誤類型的指針時編譯器會報錯。

#[repr(C)] 也是 Rust 的黑魔法之一。加上了這個標簽的結構體，在內存中的布局和對齊就和 C 一樣了，這樣的結構體可以安全地傳遞給 C ABI。

Futures in gRPC-rs

經過上一節(jié)的封裝，我們已經得到了一個可用但是非常裸的 Rust gRPC 庫了，grpc-sys。在實踐中，我們不推薦直接用 grpc-sys，直接用它就像在 Rust 中寫 C 一樣，事倍功半，Rust 語言的諸多特性無法得到施展，例如泛型，Trait，Ownership 等，也無法融入 Rust 社區(qū)。

上面說過 Core 能異步處理 RPC，那么如何用 Rust 來做更好的封裝呢？ Futures！它是一個成熟的異步編程庫，同時有一個活躍的社區(qū)。 Futures 非常適用于 RPC 等一些 IO 操作頻繁的場景。Futures 中也有組合子概念，和 Core 中的類似，但是使用上更加方便，也更加好理解。舉一個栗子：

use futures::{future, Future};fn double(i: i64) -> i64 { i * 2 }let ans = future::ok(1).map(double).and_then(|i| Ok(40 + i));println!("{:?}", ans.wait().unwrap());復制代碼

你覺得輸出的答案是多少呢？沒錯就是 42。在 Core 那節(jié)說過不同的組合子組織在一起可以干不同的事，在 Future 中我們可以這么理解，一件事可以分成多個步驟，每個步驟由一個組合子完成。比如上例，map 完成了翻倍的動作，and_then 將輸入加上 40。 現(xiàn)在來看看 gRPC-rs 封裝的 API。

// helloworld.proto service Greeter {// An unary RPC, sends a greetingrpc SayHello (HelloRequest) returns (HelloReply) {} }impl GreeterClient {pub fn say_hello_async(&self, req: HelloRequest) -> ClientUnaryReceiver<HelloReply> {self.client.unary_call_async(&METHOD_GREETER_SAY_HELLO, req, CallOption::default())}... }復制代碼

以 helloworld.proto 為例，GreeterClient::say_hello_async() 向遠程 Server 發(fā)送一個請求 (HelloRequest)，Server 返回給一個結果 (HelloReply)。由于是異步操作，這個函數(shù)會立即返回，返回的 ClientUnaryReceiver 實現(xiàn)了 Future，當它完成時就會得到 HelloReply。在一般的異步編程中都會有 Callback，用于處理異步的返回值，在這個 RPC 中就是 HelloReply，在 Future 中可以用組合子來寫，比如 and_then，再舉一個栗子，現(xiàn)有一次完整的 RPC 邏輯，拿到回復后打印到日志。下面就是 gRPC-rs 的具體用法。

// 同步 let resp = client.say_hello(req); println!("{:?}", resp);// 異步 let f = client.say_hello_async(req).and_then(|resp| {println！("{:?}", resp);Ok(())}); executer.spawn(f); // 類似 Combiner,// 用于異步執(zhí)行 Future，// 常用的有 tokio-core。復制代碼

Unary RPC

gRPC-rs 根據 service 在 proto 文件中的定義生成對應的代碼，包括 RPC 方法的定義（Method）、客戶端和服務端代碼，生成的代碼中會使用 gRPC-rs 的 API。那么具體是怎么做的呢？這節(jié)還是以 helloworld.proto 為例，來講講客戶端 Unary RPC 具體的實現(xiàn)。首先，SayHello 的 Method 記錄了 RPC 類型，全稱以及序列化反序列化函數(shù)。為什么要序列化反序列化函數(shù)呢？因為 Core 本身不涉及消息的序列化，這一部分交由封裝層解決。在生成的客戶端中可以會調用 gRPC-rs 的 API，根據 Method 的定義發(fā)起 RPC。

// 生成的代碼 const METHOD_GREETER_SAY_HELLO: Method<HelloRequest, HelloReply> = Method {ty: MethodType::Unary,name: "/helloworld.Greeter/SayHello",req_mar: Marshaller { ser: pb_ser, de: pb_de },resp_mar: Marshaller { ser: pb_ser, de: pb_de }, };impl GreeterClient {// An unary RPC, sends a greetingpub fn say_hello_async(&self, req: HelloRequest)-> ClientUnaryReceiver<HelloReply> {self.client.unary_call_async(&METHOD_GREETER_SAY_HELLO, req)}... }// gRPC-rs 的 API。該函數(shù)立即返回，不會等待 RPC 完成。省略部分代碼。 pub fn unary_async<P, Q>(channel: &Channel,method: &Method<P, Q>,req: P)-> ClientUnaryReceiver<Q> {let mut payload = vec![];(method.req_ser())(&req, &mut payload); // 序列化消息let call = channel.create_call(method, &opt); // 新建 Calllet cq_f = unsafe {grpc_sys::grpcwrap_call_start_unary(call.call, // 發(fā)起 RPCpayload,tag)};ClientUnaryReceiver::new(call, cq_f, method.resp_de()) // 收到回復后再反序列化 }復制代碼

寫在最后

這篇簡單介紹了 gRPC Core 的實現(xiàn)和 gRPC-rs 的封裝，詳細的用法，在這就不做過多介紹了，大家如果感興趣可以查看 examples。 gRPC-rs 深入使用了 Future，里面有很多神奇的用法，比如 Futures in gRPC-rs 那節(jié)最后的 executer， gRPC-rs 利用 CQ 實現(xiàn)了一個能并發(fā)執(zhí)行 Future 的 executer（類似 furtures-rs 中的 Executer），大幅減少 context switch，性能得到了顯著提升。如果你對 gRPC 和 rust 都很感興趣，歡迎參與開發(fā)，目前還有一些工作沒完成，詳情請點擊 github.com/pingcap/grp…

參考資料：

gRPC open-source universal RPC framework

The rust language implementation of gRPC

Hypertext Transfer Protocol Version 2 (HTTP/2)

Zero-cost Futures in Rust

深入了解 gRPC：協(xié)議

gRPC, Combiner Explanation

Rust, Representing opaque structs

Rust repr(), alternative representations

gRPC - A solution for RPCs by Google

Tokio, A platform for writing fast networking code with Rust.

作者：沈泰寧

總結

以上是生活随笔為你收集整理的gRPC-rs：从 C 到 Rust的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。