探索 Tokio Runtime丨Fabarta 技术专栏

标签：异步 Queue Fabarta Future Tokio Runtime Rust

导读本文将带您探索 Tokio Runtime 的核心组件，同时会介绍其在 ArcGraph 项目中的应用。阅读本文后，您将会对 Tokio Runtime 的基本原理有一个清晰的了解。此外，您也会对如何在项目中用好 Tokio 有很好的认知。

01 Tokio 概述

Rust 是一门新兴的系统编程语言，它的独特之处在于成功地解决了传统系统编程语言中常见的内存安全问题。与很多主流语言不一样的是，Rust 虽然在语言层面上提供了对异步编程的支持，但是它并没有内置异步 Runtime，因为它认为把 Runtime 这种核心的东西交给社区，可以使得自己的标准库更加轻量，同时也可以保证开发者在 Runtime 选择上的灵活性。基于 Rust 没有内置 Runtime 的现实和大家对于异步编程的强烈需求，两个优秀的程序员 Carl Lerche 和 Sean McArthur，开始探索如何在 Rust 中构建一个现代的异步框架，以支持高性能、高并发的异步程序。他们的努力最终形成了 Tokio。

源于社区的力量与 Rust 语言在各领域的快速发展，当前 Tokio 已经成为了 Rust 异步编程领域的首选框架。目前围绕 Tokio 已经形成了一个完整的生态，除了核心的 Tokio Runtime，还有许多与 Tokio 集成的扩展、工具等，可以相互配合来更好地满足开发者不同的异步编程需求、帮助他们写出更健壮的异步代码。Tokio 的社区非常活跃，Tokio 的核心开发团队也在不断努力优化框架的性能，以确保它在高并发、高吞吐量等方面保持卓越表现；社区的其他成员为 Tokio 提供了大量的使用文档和示例，这些都让 Tokio 变得越来越流行。目前，Tokio 的整个生态如下图所示：

从上图可以看到，Tokio Runtime 在整个 Tokio 生态中起着重要的作用。本文将带您探索 Tokio Runtime 的核心组件，同时会介绍其在 ArcGraph 项目中的应用。阅读本文后，您将会对 Tokio Runtime 的基本原理有一个清晰的了解。此外，您也会对如何在项目中用好 Tokio 有很好的认知。

02 Rust 语言对异步的支持

在介绍 Tokio Runtime 的核心组件之前，我们先来看一下 Rust 在语言层面上对异步编程提供的支持。Rust 异步编程依赖于两个关键概念：Future和async/await。

什么是 Future 呢？Future 是一个个的结构体，它们封装了异步运算的逻辑，Rust 规定所有的 Future 都必须实现 Future Trait，从而第三方的 Runtime 或者其他框架能够面向 Trait 对 Future 做一致性的处理。Future Trait 长这样：

pub trait Future {
    type Output;
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>;
}

我们以一个 Runtime 中的调用链路为例，来加深一下对 Future 的理解：

从图中我们可以看到，Future 是 Task 的核心成员，Runtime 通过调用 Future 的 poll 方法，驱动异步运算逻辑的运行，并根据 poll 方法返回的状态对 Task 进行后续的处理。我们也可以看到，一个 Task 包含一个主 Future 实例，主 Future 实例又可以包含多个子 Future 实例（当然子 Future 实例又可以包含多个子 Future 实例）。主 Future 实例的状态由所有子 Future 实例的状态共同决定，Task 的状态由主 Future 实例的状态决定。当 Task 返回 Pending 状态时，Process 会把它包成 Waker，放入到 Reactor 中进行后续处理。

那什么是 async/await？Rust 叫它们语法糖，目的是使异步代码的编写更加方便直观。如果没有 async/await，为了做异步的能力，我们必须实现 Future Trait，并在 poll 函数里面实现自己的业务逻辑，我们以一个简单的异步加法举例，直接使用 Future 需要这么写：

struct AddFuture {
    a: i32,
    b: i32,
    ready_time: Instant,
}

impl Future for AddFuture {
    type Output = i32;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        if Instant::now() >= self.ready_time {
            Poll::Ready(self.a + self.b)
        } else {
            cx.waker().wake_by_ref();
            Poll::Pending
        }
    }
}

这个 Future 只是用来模拟异步计算，其功能是等待一定的时间后，返回 a+b 的结果。我们可以看到，为了实现一个简单的异步功能，涉及到很多繁琐的操作，比如定义一个 struct，定义 type Output，实现 poll 逻辑；同时，这样的写法，跟我们的编程习惯也不相符，如果按照这样的做法，要手动实现一个个 Future 来推动业务功能的展开，这个是非常麻烦的。因此 Rust 引入了 async/await。我们先来看看如何用 async/await 实现相同的功能：

async fn add(a: i32, b: i32, secs: u64) -> i32 {
    sleep(Duration::from_secs(secs)).await;
    a + b
}

就是这么简单！用 async/await 写的代码，相比手写 Future，不用再关注业务逻辑以外的代码，也和实现同步代码的习惯相符。Rust 编译器看到 async/await 时会把相应的代码转换为 Future。它的基本原理如下：

async函数：使用async关键字标记的函数会被编译器识别为异步函数。异步函数内部可以包含await关键字，表示异步操作的等待。编译器会将异步函数转换为一个实现了Future trait 的类型。
await表达式：await关键字用于等待异步操作完成。当遇到await表达式时，编译器会生成状态机代码，将当前函数的执行挂起，并将控制权交还给调用者。在异步操作完成后，状态机会恢复函数的执行。
状态机：编译器会根据异步函数的结构生成状态机代码。状态机中包含状态变量、存储异步操作的结果、等待的条件等。当遇到await表达式时，状态机会记录等待的状态，并返回一个未完成的Future。
Future 实现：异步函数会被编译为实现了Future trait 的类型，其中Output类型表示异步函数的返回类型。状态机的状态转换和异步操作的逻辑会在poll方法中实现。

03 Tokio 核心组件详解

上一节介绍了 Rust 语言对异步的支持，总的来说就是 Rust 在语言层面上，为开发者提供了方便和规范地生成一个个 Future 的手段。那怎么让这些生成的 Future 跑起来呢？这就得借助于 Runtime。Runtime 在异步程序中的位置如下图所示：

我们可以看到，Runtime 的核心组件是任务调度系统和 Reactor 模型，接下来我们分别进行详细介绍。

Tokio 任务调度系统

Tokio 的任务调度是其异步编程能力的核心，它能够高效地管理和调度大量的异步任务，实现并发执行。接下来我们看看 Tokio 调度系统是怎么工作的。首先是调度系统的大图：

在上图中，Local Run Queue、 LIFO Slot、 Global Queue 都用于存储待处理的任务。Tokio Runtime 可以包含多个 Processor，每个 Processor 都有自己的 Local Run Queue（Local Run Queue 的大小是固定的）和 LIFO Slot（目前 LIFO Slot 只能存放一个任务），所有的 Processor 共享 Global Queue。

Processor 获取 task 后，会开始执行这个 task，在 task 执行过程中，可能会产生很多新的 task，第一个新 task 会被放到 LIFO Slot 中，其他新 task 会被放到 Local Run Queue 中，因为 Local Run Queue 的大小是固定的，如果它满了，剩余的 task 会被放到 Global Queue 中。

Processor 运行完当前 task 后，会尝试按照以下顺序获取新的 task 并继续运行：

LIFO Slot.
Local Run Queue.
Global Queue.
其他 Processor 的 Local Run Queue。

如果 Processor 获取不到 task 了，那么其对应的线程就会休眠，等待下次唤醒。

在这个机制中，采用 Local Run Queue 和 LIFO Slot 的目的是为了避免过多的线程之间的锁竞争，并保证一定的 CPU 本地性。Global Queue 的目的是为了放置更多的任务，当然它是有锁的，不可避免的会因为锁竞争带来一些性能损耗。

Tokio Reactor 模型

let mut cnt = 1000;
while cnt > 0 {
    cnt -=1;
}

这类任务叫做同步任务。

另一类是把事情交给操作系统或其他线程去做，当前线程获取他们的处理结果后才能继续往下走的任务，常见的比如网络请求、读写文件等等。对于这类任务，如果当前线程一直等待操作系统的返回，显然是比较浪费的，因为在操作系统返回数据之前，线程完全可以去处理其他任务。这类任务叫做异步任务。那么，tokio 是怎么实现异步任务的执行的呢？答案是通过 Reactor 模型。下面我们来看看 Reactor 的组成，以及它是怎么跟 Tokio 的任务调度系统协作的。

从图中我们可以看到，Reactor 主要由三部分构成：

多路复用器：多路复用器用于监听操作系统的多个文件描述符，比如我们的读文件操作，读网络请求等，都会对应一个文件描述符。当文件描述符产生事件时，多路复用器会把事件以及对应的 Waker 放到事件队列中。
事件循环：事件循环会不停地轮询事件队列，当事件队列中有事件时，它会拿出这个事件，并执行这个事件对应的 Waker 的 wake 方法。当 wake 方法被调用后，Waker 对应的 task 又会被加入到 Task Queue 中。
事件队列：用于存放多路复用器分发过来的事件，供事件循环轮询。

从图中也可以看到，一个异步任务的处理过程为：

Processor 从 Task Queue 中获取一个 Task。
Processor 判断这个 Task 为异步任务（Task.future.poll（）返回 Pending），则将 Task 封装为 Waker，交给多路复用器。
多路复用器监听系统事件。
当事件出现时，多路复用器将事件和该事件对应的 Waker 交给事件队列。
事件循环处理事件，调用 Waker.wake（），将 Task 重新放入 TaskQueue 中，待 Processor 处理。

04 Tokio Runtime 在 ArcGraph 中的应用

前面一章介绍了 Tokio Runtime 的核心组件，从中我们对其基本工作原理有了一定的了解。接下去，我们简单介绍一下 Tokio Runtime 在 ArcGraph 中的应用。**ArcGraph 是多模态智能引擎 ArcNeural 的一个重要组成部分，是由 Fabarta 公司基于 Rust 语言、完全自主设计和研发的一款分布式、云原生、支持图的存查分析一体化的高性能图数据库。**在这个数据库中，我们用 Tokio Runtime 进行任务的调度和管理，包括 CPU 任务，网络通信任务，数据库读写任务等等。在使用 Tokio Runtime 的过程中，我们经历了两版架构（在以下架构中，每个方框表示一个 Runtime）。

第一版架构是这样的：

在这一版架构中，各个 Runtime 的责任定义如下：

QueryServiceExecutor：负责接收客户端发过来的 query/dml/ddl 语句，生成物理执行计划。
QueryEngineExecutor：用于发送/接收计算层的 grpc 通信。
Concurrency：用于处理计算层的 CPU 任务。
RaftExecutor：用于处理 Raft 消息。
StorageExecutor：用于写数据，同时后台任务的处理也依赖于它。

从图中我们可以看到，各个 runtime 之间的交互是非常多的，这就带来了很多的上下文切换，很影响系统的性能。

第二版架构是这样的：

在这一版架构中，各个 Runtime 的责任定义如下：

QueryServiceExecutor: 负责 query/dml 业务逻辑和 query/dml 相关的通信。
RaftExecutor：负责 Raft 业务逻辑和 Raft 通信。
BackgroundExecutor：负责后台任务的处理，job 逻辑，job 分发，ddl 执行，以及其他非 query/dml/raft 的任务。

基于上一版的教训，我们简化了系统 Runtime。在这个架构中，Runtime 的工作更加内聚，Runtime 之间的交互大大减少，带来的明显的好处是上下文切换明显下降，系统性能得到了很好的提升。

05 未来展望

以上是关于 Tokio Runtime 及其在我们项目中应用的介绍。通过本文的分析，相信读者对 Tokio Runtime 有了初步的了解。然而，Tokio 实际上是一个庞大且持续发展的生态系统，我们当前的探索也只是其中的一小部分。

在未来的发展中，我们将深入学习和探索 Tokio 的更多特性和功能。我们计划将 Tokio 的优秀特性融入到我们的项目中，比如利用 Tokio Tracing 来实现更详尽的跟踪和日志记录，以便更好地理解和调试异步操作的行为。我们还将考虑采用 Tokio Console 等工具，来提高系统的可观测性。此外，我们也会持续研究同步线程和异步 Runtime 并存的策略。我们将努力找到最佳的结合方式，使得同步和异步代码能够在项目中和谐共存，充分发挥各自的优势。通过优化系统的整体性能，我们期待为用户提供更出色的体验。

总之，我们对 Tokio Runtime 的探索和应用并未止步于此，我们将不断扩展我们的知识，融合新的技术，使得我们的项目能够在图数据库领域保持领先地位，为用户创造更大的价值。

本文作者

胡焰

Fabarta 高级技术专家

曾就职于网易、IBM、蚂蚁金服，具有丰富的系统架构和软件开发经验，尤其擅长基于微服务的系统架构，同时对自然语言处理相关技术有深入的实践。现就职于 Fabarta，从事新一代图数据库内核引擎的研发。

关于 Fabarta

Fabarta 是一家 AI 基础设施公司，通过探索和联结数据资源，助力企业实现智能驱动的持续创新。在引擎层，打造面向 AI 的技术基础设施，提供支持图、向量和 AI 推理能力融合的 ArcNeural 多模态智能引擎；在平台层，通过企业智能分析平台加速可解释图智能和新一代AI技术在企业场景的落地，同时利用多模态数据编织平台帮助企业梳理多模态的数据资产，让企业充分发挥数据流动带来的价值；此外，Fabarta 可以基于多模态智能引擎、企业智能分析平台和多模态数据编织平台与客户和伙伴一同构建行业应用，加速企业数智化转型和 AI 技术的落地。

标签：异步,Queue,Fabarta,Future,Tokio,Runtime,Rust
From： https://blog.51cto.com/fabarta/8633530