作者：屈鵬

本篇 TiKV 原始碼解析將為大家介紹 TiKV 的另一週邊元件—— grpc-rs。grpc-rs 是 PingCAP 實現的一個 gRPC 的 Rust 繫結，其 Server/Client 端的程式碼框架都基於 Future，事件驅動的 EventLoop 被隱藏在了庫的內部，所以非常易於使用。本文將以一個簡單的 gRPC 服務作為例子，展示 grpc-rs 會生成的服務端程式碼框架和需要服務的實現者填寫的內容，然後會深入介紹伺服器在啟動時如何將後臺的事件迴圈與這個框架掛鉤，並在後臺執行緒中執行實現者的程式碼。

基本的程式碼生成及服務端 API

gRPC 使用 protobuf 定義一個服務，之後呼叫相關的程式碼生成工具就可以生成服務端、客戶端的程式碼框架了，這個過程可以參考我們的

官方文件。客戶端可以直接呼叫這些生成的程式碼，向服務端傳送請求並接收響應，而服務端則需要服務的實現者自己來定製對請求的處理邏輯，生成響應併發回給客戶端。舉一個例子：

#[derive(Clone)]
struct MyHelloService {}
impl Hello for MyHelloService {
    // trait 中的函式簽名由 grpc-rs 生成，內部實現需要使用者自己填寫
    fn hello(&mut self, ctx: RpcContext, req: HelloRequest, sink: UnarySink<HelloResponse>) {
        let mut resp = HelloResponse::new();
        resp.set_to(req.get_from());
        ctx.spawn(
            sink.success(resp)
                .map(|_| println!("send hello response back success"))
                .map_err(|e| println!("send hello response back fail: {}", e))
        );
    }
}
 

我們定義了一個名為 Hello 的服務，裡面只有一個名為 hello 的 RPC。grpc-rs 會為服務生成一個 trait，裡面的方法就是這個服務包含的所有 RPC。在這個例子中唯一的 RPC 中，我們從 HelloRequest 中拿到客戶端的名字，然後再將這個名字放到 HelloResponse 中發回去，非常簡單，只是展示一下函式簽名中各個引數的用法。

然後，我們需要考慮的是如何把這個服務執行起來，監聽一個埠，真正能夠響應客戶端的請求呢？下面的程式碼片段展示瞭如何執行這個服務：

fn main() {
    // 建立一個 Environment，裡面包含一個 Completion Queue
    let env = Arc::new(EnvBuilder::new().cq_count(4).build());
    let channel_args = ChannelBuilder::new(env.clone()).build_args();
    let my_service = MyHelloWorldService::new();
    let mut server = ServerBuilder::new(env.clone())
        // 使用 MyHelloWorldService 作為服務端的實現，註冊到 gRPC server 中
        .register_service(create_hello(my_service))
        .bind("0.0.0.0", 44444)
        .channel_args(channel_args)
        .build()
        .unwrap();
    server.start();
    thread::park();
}
 

以上程式碼展示了 grpc-rs 的足夠簡潔的 API 介面，各行程式碼的意義如其註釋所示。

Server 的建立和啟動

下面我們來看一下這個 gRPC server 是如何接收客戶端的請求，並路由到我們實現的服務端程式碼中進行後續的處理的。

第一步我們初始化一個 Environment，並設定 Completion Queue（完成佇列）的個數為 4 個。完成佇列是 gRPC 的一個核心概念，grpc-rs 為每一個完成佇列建立一個執行緒，並在執行緒中執行一個事件迴圈，類似於 Linux 網路程式設計中不斷地呼叫 epoll_wait 來獲取事件，進行處理：

// event loop
fn poll_queue(cq: Arc<CompletionQueueHandle>) {
    let id = thread::current().id();
    let cq = CompletionQueue::new(cq, id);
    loop {
        let e = cq.next();
        match e.event_type {
            EventType::QueueShutdown => break,
            EventType::QueueTimeout => continue,
            EventType::OpComplete => {}
        }
        let tag: Box<CallTag> = unsafe { Box::from_raw(e.tag as _) };
        tag.resolve(&cq, e.success != 0);
    }
}

事件被封裝在 Tag 中。我們暫時忽略對事件的具體處理邏輯，目前我們只需要知道，當這個 Environment 被建立好之後，這些後臺執行緒便開始運行了。那麼剩下的任務就是監聽一個埠，將網路上的事件路由到這幾個事件迴圈中。這個過程在 Server 的 start 方法中：

/// Start the server.
pub fn start(&mut self) {
    unsafe {
        grpc_sys::grpc_server_start(self.core.server);
        for cq in self.env.completion_queues() {
            let registry = self
                .handlers
                .iter()
                .map(|(k, v)| (k.to_owned(), v.box_clone()))
                .collect();
            let rc = RequestCallContext {
                server: self.core.clone(),
                registry: Arc::new(UnsafeCell::new(registry)),
            };
            for _ in 0..self.core.slots_per_cq {
                request_call(rc.clone(), cq);
            }
        }
    }
}

首先呼叫 grpc_server_start 來啟動這個 Server，然後對每一個完成佇列，複製一份 handler 字典。這個字典的 key 是一個字串，而 value 是一個函式指標，指向對這個型別的請求的處理函式——其實就是前面所述的服務的具體實現邏輯。key 的構造方式其實就是 /<ServiceName>/<RpcName>，實際上就是 HTTP/2 中頭部欄位中的 path 的值。我們知道 gRPC 是基於 HTTP/2 的，關於 gRPC 的請求、響應是如何裝進 HTTP/2 的幀中的，更多的細節可以參考 官方文件，這裡就不贅述了。

接著我們建立一個 RequestCallContext，然後對每個完成佇列呼叫幾次 request_call。這個函式會往完成佇列中註冊若干個 Call，相當於用 epoll_ctl 往一個 epoll fd 中註冊一些事件的關注。Call 是 gRPC 在進行遠端過程呼叫時的基本單元，每一個 RPC 在建立的時候都會從完成佇列裡取出一個 Call 物件，後者會在這個 RPC 結束時被回收。因此，在 start 函式中每一個完成佇列上註冊的 Call 個數決定了這個完成佇列上可以併發地處理多少個 RPC，在 grpc-rs 中預設的值是 1024 個。

小結

以上程式碼基本都在 grpc-rs 倉庫中的 src/server.rs 檔案中。在 start 函式返回之後，服務端的初始化及啟動過程便結束了。現在，可以快速地用幾句話回顧一下：首先建立一個 Environment，內部會為每一個完成佇列啟動一個執行緒；接著建立 Server 物件，繫結埠，並將一個或多個服務註冊到這個 Server 上；最後呼叫 Server 的 start 方法，將服務的具體實現關聯到若干個 Call 上，並塞進所有的完成佇列中。在這之後，網路上新來的 RPC 請求便可以在後臺的事件迴圈中被取出，並根據具體實現的字典分別執行了。最後，不要忘記 start 是一個非阻塞的方法，呼叫它的主執行緒在之後可以繼續執行別的邏輯或者掛起。

本篇原始碼解析就到這裡，下篇關於 grpc-rs 的文章我們會進一步介紹一個 Call 或者 RPC 的生命週期，以及每一階段在 Server 端的完成佇列中對應哪一種事件、會被如何處理，這一部分是 grpc-rs 的核心程式碼，敬請期待！

原文連結：https://www.pingcap.com/blog-cn/tikv-source-code-reading-7/

相關推薦