Tokio，Rust异步编程实践之路

迈不过友情╰ 2022-11-11 14:56 321阅读 0赞

# 缘起 #

在许多编程语言里，我们都非常乐于去研究在这个语言中所使用的异步网络编程的框架，比如说Python的 Gevent、asyncio，Nginx 和 OpenResty，Go 等，今年年初我开始接触 Rust，并被其无 GC、内存安全、极小的运行时等特性所吸引，经过一段时间的学习，开始寻找构建实际项目的解决方案，很快 mio、tokio 等框架进入了我的视野，于是开始从更加底层的 mio 出发实验。

> [https://github.com/Hevienz/mio\_test/blob/master/src/main.rs][https_github.com_Hevienz_mio_test_blob_master_src_main.rs]

可以看到 mio 是一个非常底层的异步编程的框架，这意味着如果我们要在实际的项目开发中使用它时，就不得不从 event loop 开始编写我们的软件，这并不是我们所期望的，于是我们需要一个更高层次抽象的框架，这便是本文要为大家讲述的 tokio。

# tokio #

tokio 是 Rust 中的异步编程框架，它将复杂的异步编程抽象为 Futures、Tasks 和 Executor，并提供了 Timers 等基础设施，下文中我们将一一展开。

## 运行时模型 ##

tokio 是一个基于轮训的模型。比如我们要在 tokio 上调度我们的 task，我们需要为其实现 `Future` trait。比如下面的例子中，我们想要得到一个 widget，但它有可能还没有准备好，这时候我们调用 poll 的结果就是 `Ok(Async::NotReady)`，Executor 会负责重复的调用 `poll`，直到 widget 准备好，返回`Ok(Async::Ready(()))`。

/// A task that polls a single widget and writes it to STDOUT.
    pub struct MyTask;
    
    impl Future for MyTask {
        type Item = ();
        type Error = ();
    
        fn poll(&mut self) -> Result<Async<()>, ()> {
            match poll_widget() {
                Async::Ready(widget) => {
                    println!("widget={:?}", widget);
                    Ok(Async::Ready(()))
                }
                Async::NotReady => {
                    return Ok(Async::NotReady);
                }
            }
        }
    }

在最简单的情况下，Executor 可能会长这样。(注：这不是真实的实现，只是用来说明概念)

pub struct SpinExecutor {
        tasks: VecDeque<Box<Future<Item = (), Error = ()>>>,
    }
    
    impl SpinExecutor {
        pub fn spawn<T>(&mut self, task: T)
        where T: Future<Item = (), Error = ()> + 'static
        {
            self.tasks.push_back(Box::new(task));
        }
    
        pub fn run(&mut self) {
            while let Some(mut task) = self.tasks.pop_front() {
                match task.poll().unwrap() {
                    Async::Ready(_) => {}
                    Async::NotReady => {
                        self.tasks.push_back(task);
                    }
                }
            }
        }
    }

Executor 频繁地轮询所有 task，即使某些 task 仍然会以 `NotReady` 返回。  
理想情况下，Executor 应该可以通过某种方式知道哪些 task 恰好转变为 “就绪” 状态。这正是 futures 任务模型的核心。

## Futures ##

future 是对一个未来事件的抽象。比如你可以将各种事件抽象为 future:

*  在线程池中执行的数据库查询。当数据库查询完成时，future 完成，其值是查询的结果。
 *  对服务器的 RPC 调用。当服务器回复时，future 完成，其值是服务器的响应。
 *  超时事件。当时间到了，future 就完成了，它的值是 `()`。
 *  在线程池上运行的长时间运行的 CPU 密集型任务。任务完成后，future 完成，其值为任务的返回值。

这里我们举一个例子：

extern crate futures;
    extern crate tokio;
    extern crate tokio_core;
    
    use std::error::Error;
    use futures::Future;
    use futures::future::{ok, done};
    use tokio_core::reactor::Core;
    
    fn my_fn_squared(i: u32) -> Result<u32, Box<Error>> {
        Ok(i * i)
    }
    
    fn my_fut_squared(i: u32) -> impl Future<Item = u32, Error = Box<Error + 'static>> {
        ok(i * i)
    }
    
    fn my_fut() -> impl Future<Item = u32, Error = Box<Error + 'static>> {
        ok(10)
    }
    
    fn main() {
        let mut reactor = Core::new().unwrap();
    
        let chained_future = my_fut().and_then(|retval| {
            done(my_fn_squared(retval)).and_then(|retval2| my_fut_squared(retval2))
        });
        let retval3 = reactor.run(chained_future).unwrap();
        println!("{:?}", retval3);
    }

这里，我们的 `my_fut` 的返回值实现了 Future，我们知道它被 Executor 执行完成后，会返回一个 `u32` 或者 一个 `Box<Error + 'static>`，而现在我们就可以通过 `.and_then` 来处理这个 `u32` 的值，而最终我们将我们的 future 链接了起来，交给 Executor 执行。

## Tasks ##

Tasks 是应用程序的 “逻辑单元”。他们以 Future trait 来表示。一旦 task 完成处理，task 的 future 实现将以值 `()` 返回。

Tasks 被传递给 Executor，Executor 处理 task 的调度。Executor 通常在一组或一组线程中调度许多 task。task 不得执行计算繁重的逻辑，否则将阻止其他 task 执行。

Tasks 既可以通过实现 Future trait 来实现，也可以通过使用 `futures` 和 `tokio` crates 中的各种组合器函数来构建 future 来实现。

## I/O ##

`tokio` crate 也提供了 TCP、UDP 的支持，不像 `std` 中的实现，tokio 的网络类型是基于 poll 模型的，并且当他们的 “就绪” 状态改变时会通知 task executors。在 `tokio::net` 模块中你将会找到像 TcpListener、TcpStream、UdpSocket 这些类型。

所有这些类型都提供了 `future` 的 API 以及 `poll` API。

Tokio 网络类型被一个基于 `mio` 的 reactor 所驱动，默认情况下，它在后台线程上启动。

### 使用 future API ###

一些帮助使用 future API 的函数包括：

*  `incoming`：入站 TCP 连接的 Stream。
 *  `read_exact`：将`n`字节准确读入缓冲区。
 *  `read_to_end`：将所有字节读入缓冲区。
 *  `write_all`：写缓冲区的全部内容。
 *  `copy`：将字节从一个 I/O 句柄复制到另一个。

这些函数中的许多都是源于 `AsyncRead` 和 `AsyncWrite` trait 的。这些 trait 类似于 `std` 中的 `Read` 和 `Write`，但仅仅用于具有 `future aware` 的类型，例如符合下面的特征：

*  调用 `read` 或 `write` 是非阻塞的，他们从不阻塞调用线程。
 *  如果一个调用会以其他方式阻塞，那么会返回一个错误 WouldBlock。如果发生这种情况，则当前 future 的task 将在 I/O 再次准备就绪时被调度。

注意 `AsyncRead` 和 `AsyncWrite` 类型的用户应该使用 `poll_read` 和 `poll_write` 代替直接调用 `read` 和 `write`。

例如，以下是如何接受连接，从中读取5个字节，然后将5个字节写回 socket 的例子：

### 使用 Poll API ###

当手动实现 Future 时，需要使用基于 Poll 的 API，并且你需要返回 `Async`。当您需要实现自己的处理自定义逻辑的组合器时，这非常有用。

例如，这就是如何为 TcpStream 实现 `read_exact` future 的例子。

pub struct ReadExact {
        state: State,
    }
    
    enum State {
        Reading {
            stream: TcpStream,
            buf: Vec<u8>,
            pos: usize,
        },
        Empty,
    }
    
    impl Future for ReadExact {
        type Item = (TcpStream, Vec<u8>);
        type Error = io::Error;
    
        fn poll(&mut self) -> Result<Async<Self::Item>, io::Error> {
            match self.state {
                State::Reading {
                    ref mut stream,
                    ref mut buf,
                    ref mut pos
                } => {
                    while *pos < buf.len() {
                        let n = try_ready!({
                            stream.poll_read(&mut buf[*pos..])
                        });
                        *pos += n;
                        if n == 0 {
                            let err = io::Error::new(
                                io::ErrorKind::UnexpectedEof,
                                "early eof");
    
                            return Err(err)
                        }
                    }
                }
                State::Empty => panic!("poll a ReadExact after it's done"),
            }
    
            match mem::replace(&mut self.state, State::Empty) {
                State::Reading { stream, buf, .. } => {
                    Ok(Async::Ready((stream, buf)))
                }
                State::Empty => panic!(),
            }
        }
    }

### 数据报 ###

UdpSocket 类型提供了许多方便的方法：

*  `send_dgram` 允许您将发送数据报作为 future，如果无法立即发送整个数据报，则返回错误。
 *  `recv_dgram` 表示将数据报读入缓冲区。

## 示例 ##

#[macro_use]
    extern crate log;
    extern crate futures;
    extern crate pretty_env_logger;
    extern crate tokio;
    
    use futures::future::{done, ok};
    use futures::{Future, Stream};
    use tokio::io::{self as tio, AsyncRead};
    use tokio::net::{TcpListener, TcpStream};
    
    use std::error;
    use std::fmt;
    use std::io;
    
    fn client_fut(socket: TcpStream) -> impl Future<Item = (), Error = ()> + 'static + Send {
        futures::lazy(move || match socket.peer_addr() {
            Ok(peer) => {
                info!("Tcp connection [{:?}] connected to server", peer);
                Ok((socket, peer))
            }
            Err(err) => {
                error!("Fetch peer address failed: {:?}", err);
                Err(())
            }
        }).and_then(move |(socket, peer)| {
                let buf = vec![0; 5];
                let svc_fut = tio::read_exact(socket, buf)
                    .and_then(|(socket, buf)| {
                        tio::write_all(socket, buf)
                    })
                    .then(|_| Ok(()));
    
                tokio::spawn(svc_fut);
                ok(())
            })
    }
    
    fn server_fut(listener: TcpListener) -> impl Future<Item = (), Error = ()> + 'static + Send {
        listener
            .incoming()
            .for_each(|socket| {
                tokio::spawn(client_fut(socket));
                Ok(())
            })
            .map_err(|err| {
                error!("Accept connection failed: {:?}", err);
            })
    }
    
    fn run() -> Result<(), io::Error> {
        let addr = "127.0.0.1:1234".parse().unwrap();
        info!("Listening on {:?}", addr);
    
        let listener = TcpListener::bind(&addr)?;
        let server_fut = server_fut(listener);
    
        tokio::run(server_fut);
        Ok(())
    }
    
    fn print<T: fmt::Debug, E: error::Error>(result: Result<T, E>) {
        match result {
            Ok(any) => info!("Result: {:?}", any),
            Err(err) => error!("Error: {:?}", err),
        }
    }
    
    fn init() {
        pretty_env_logger::init();
    }
    
    fn main() {
        init();
        print(run());
    }

## Timers ##

在编写基于网络的应用程序时，通常需要根据时间执行操作。

*  在一段时间后运行一些代码。
 *  取消运行时间过长的运行操作。
 *  以一定间隔重复执行操作。

这些用例通过使用 `timer` 模块中提供的各种计时器 API 来处理。

### 延迟运行代码 ###

在这个例子中，我们希望在一段时间后执行任务。为此，我们使用 `Delay` API。我们要做的只是将 `"Hello world!"` 写到终端。

use tokio::prelude::*;
    use tokio::timer::Delay;
    
    use std::time::{Duration, Instant};
    
    fn main() {
        let when = Instant::now() + Duration::from_millis(100);
        let task = Delay::new(when)
            .and_then(|_| {
                println!("Hello world!");
                Ok(())
            })
            .map_err(|e| panic!("delay errored; err={:?}", e));
    
        tokio::run(task);
    }

### 为长时间运行的操作设置 Timeout ###

在编写健壮的网络应用程序时，确保在合理的时间内完成操作至关重要。在等待来自外部的，不受信任的来源的数据时尤其如此。

该 `Deadline` 类型确保操作在固定的时间内完成。

use tokio::io;
    use tokio::net::TcpStream;
    use tokio::prelude::*;
    
    use std::time::{Duration, Instant};
    
    fn read_four_bytes(socket: TcpStream)
        -> Box<Future<Item = (TcpStream, Vec<u8>), Error = ()>>
    {
        // The instant at which the read will be aborted if
        // it has not yet completed.
        let when = Instant::now() + Duration::from_secs(5);
    
        let buf = vec![0; 4];
        let fut = io::read_exact(socket, buf)
            .deadline(when)
            .map_err(|_| println!("failed to read 4 bytes by deadline"));
    
        Box::new(fut)
    }

### 周期性运行代码 ###

在一个时间间隔内重复运行代码对于在套接字上发送 PING 消息，或经常检查配置文件等情况很有用。

`Interval` 类型实现了 `Stream`，并以指定的速率挂起。

use tokio::prelude::*;
    use tokio::timer::Interval;
    
    use std::time::{Duration, Instant};
    
    fn main() {
        let task = Interval::new(Instant::now(), Duration::from_millis(100))
            .take(10)
            .for_each(|instant| {
                println!("fire; instant={:?}", instant);
                Ok(())
            })
            .map_err(|e| panic!("interval errored; err={:?}", e));
    
        tokio::run(task);
    }

### 计时器的注意事项 ###

Tokio 计时器的粒度为 1 毫秒。任何更小的间隔都会向上舍入到最接近的毫秒。定时器在用户域中实现（即不使用操作系统定时器，像 linux 上的 timerfd）。它使用分层散列计时器轮实现，在创建，取消和触发超时时提供有效的恒定时间复杂度。

Tokio 运行时包括每个工作线程一个计时器实例。这意味着，如果运行时启动4个工作线程，则将有4个计时器实例。这在大多数情况下避免了同步，因为当使用计时器时，任务将在位于当前线程上的状态下操作。

也就是说，计时器实现是线程安全的，并支持从任何线程使用。

## 基本组合器 ##

下面是关于 Future 的图表，来自于 [Cheatsheet for Futures][] 。

// Constructing leaf futures
    fn empty ()             -> Future<T, E>
    fn ok    (T)            -> Future<T, E>
    fn err   (E)            -> Future<T, E>
    fn result(Result<T, E>) -> Future<T, E>
    
    // General future constructor
    fn poll_fn(FnMut(thread_local!(Task)) -> Poll<T, E>) -> Future<T, E>
    
    // Mapping futures
    fn Future::map     (Future<T, E>, FnOnce(T) -> U) -> Future<U, E>
    fn Future::map_err (Future<T, E>, FnOnce(E) -> F) -> Future<T, F>
    fn Future::from_err(Future<T, Into<E>>)           -> Future<T, E>
    
    // Chaining (sequencing) futures
    fn Future::then    (Future<T, E>, FnOnce(Result<T, E>) -> IntoFuture<U, F>) -> Future<U, F>
    fn Future::and_then(Future<T, E>, FnOnce(T)            -> IntoFuture<U, E>) -> Future<U, E>
    fn Future::or_else (Future<T, E>, FnOnce(E)            -> IntoFuture<T, F>) -> Future<T, F>
    fn Future::flatten (Future<Future<T, E>, Into<E>>)                          -> Future<T, E>
    
    // Joining (waiting) futures
    fn Future::join (Future<T, E>, IntoFuture<U, E>)                                                       -> Future<(T, U),          E>
    fn Future::join3(Future<T, E>, IntoFuture<U, E>, IntoFuture<V, E>)                                     -> Future<(T, U, V),       E>
    fn Future::join4(Future<T, E>, IntoFuture<U, E>, IntoFuture<V, E>, IntoFuture<W, E>)                   -> Future<(T, U, V, W),    E>
    fn Future::join5(Future<T, E>, IntoFuture<U, E>, IntoFuture<V, E>, IntoFuture<W, E>, IntoFuture<X, E>) -> Future<(T, U, V, W, X), E>
    fn join_all     (IntoIterator<IntoFuture<T, E>>)                                                       -> Future<Vec<T>,          E>
    
    // Selecting (racing) futures
    fn Future::select (Future<T, E>, IntoFuture<T, E>) -> Future<(T, Future<T, E>), (E, Future<T, E>)>
    fn Future::select2(Future<T, E>, IntoFuture<U, F>) -> Future<Either<(T, Future<U, F>), (U, Future<T, E>)>, Either<(E, Future<U, F>), (F, Future<T, E>)>>
    fn select_all     (IntoIterator<IntoFuture<T, E>>) -> Future<(T, usize, Vec<Future<T, E>>), (E, usize, Vec<Future<T, E>>)>
    fn select_ok      (IntoIterator<IntoFuture<T, E>>) -> Future<(T, Vec<Future<T, E>>), E>
    
    // Utility
    fn lazy         (FnOnce() -> IntoFuture<T, E>)             -> Future<T, E>
    fn loop_fn      (S, FnMut(S) -> IntoFuture<Loop<T, S>, E>) -> Future<T, E>
    
    // Miscellaneous
    fn Future::into_stream   (Future<T, E>)            -> Stream<T, E>
    fn Future::flatten_stream(Future<Stream<T, E>, E>) -> Stream<T, E>
    fn Future::fuse          (Future<T, E>)            -> Future<T, E>
    fn Future::catch_unwind  (Future<T, E>+UnwindSafe) -> Future<Result<T, E>, Any+Send>
    fn Future::shared        (Future<T, E>)            -> Future<SharedItem<T>, SharedError<E>>+Clone
    fn Future::wait          (Future<T, E>)            -> Result<T, E>

这部分的内容推荐参考这篇文章，[https://www.jianshu.com/p/5059c403a335。][https_www.jianshu.com_p_5059c403a335]

本文不再赘述。

## 返回 futures ##

在使用 futures 时，您可能需要做的第一件事就是返回一个 `Future`。这有几种选择，从最符合人体工程学到最不符合。

*  Trait 对象
 *  `impl Trait`

### Trait 对象 ###

首先，您始终可以选择返回一个 boxed `trait 对象`：

fn foo() -> Box<Future<Item = u32, Error = io::Error>> {
        // ...
    }

这个策略的好处是它很容易写出来并且易于创建。

这种方法的缺点是，在构建 future 时需要运行时分配，在使用该 future 时需要动态分派。`Box` 需要在堆上分配而 future 会被置入其中。

通常可以通过仅在您想要返回的 future 链的最后来 `Boxing` 来减少分配。

### `impl Trait` ###

在 Rust 1.26 版本之后（2018年5月7日发布），我们可以使用叫做 `impl Trait` 的新的语言特性。

fn add_10<F>(f: F) -> impl Future<Item = i32, Error = F::Error>
        where F: Future<Item = i32>,
    {
        f.map(|i| i + 10)
    }

这种方法的好处在于它是零开销的，不再需要 `Box`。

## 使用 framed streams ##

Tokio 有帮助函数将字节流转换为帧流。字节流的例子包括 TCP 连接，管道，文件对象以及标准输入和输出。在Rust中，streams 很容易识别，因为它们实现了 `Read` 和 `Write` trait。

最简单的帧化的消息形式之一是行分隔消息。每条消息都以一个 `\n` 字符结尾。让我们看一下如何使用 tokio 实现行分隔消息流。

### 编写编解码器 ###

编解码器实现 `tokio_codec::Decoder` 和 `tokio_codec::Encoder` trait。他的工作就是将字节转为帧以及相反。这些 `trait` 与 `tokio_codec::Framed` struct一起使用，以提供字节流的缓冲，解码和编码。

让我们看一下`LinesCodec` struct 的简化版本，它实现了行分隔消息的解码和编码。

pub struct LinesCodec {
        // Stored index of the next index to examine for a `\n` character.
        // This is used to optimize searching.
        // For example, if `decode` was called with `abc`, it would hold `3`,
        // because that is the next index to examine.
        // The next time `decode` is called with `abcde\n`, the method will
        // only look at `de\n` before returning.
        next_index: usize,
    }

这里的注释解释了，由于字节被缓存直到找到一行，因此每次接收数据时从缓冲区的开头搜索 `\n` 是浪费的。保存缓冲区的最后长度并在收到新数据时从那里开始搜索将更有效。

`Decoder::decode` 是在底层流上接收到数据时调用的方法。该方法可以生成帧或返回 `Ok(None)` 以表示它需要更多数据来生成帧。该 decode 方法负责通过使用 BytesMut 的方法将不再需要缓冲的数据删除。如果数据未删除，缓冲区将持续增长。

让我们来看看如何为 `LinesCodec` 实现 `Decoder::decode`。

fn decode(&mut self, buf: &mut BytesMut) -> Result<Option<String>, io::Error> {
        // Look for a byte with the value '\n' in buf. Start searching from the search start index.
        if let Some(newline_offset) = buf[self.next_index..].iter().position(|b| *b == b'\n')
        {
            // Found a '\n' in the string.
    
            // The index of the '\n' is at the sum of the start position + the offset found.
            let newline_index = newline_offset + self.next_index;
    
            // Split the buffer at the index of the '\n' + 1 to include the '\n'.
            // `split_to` returns a new buffer with the contents up to the index.
            // The buffer on which `split_to` is called will now start at this index.
            let line = buf.split_to(newline_index + 1);
    
            // Trim the `\n` from the buffer because it's part of the protocol,
            // not the data.
            let line = &line[..line.len() - 1];
    
            // Convert the bytes to a string and panic if the bytes are not valid utf-8.
            let line = str::from_utf8(&line).expect("invalid utf8 data");
    
            // Set the search start index back to 0.
            self.next_index = 0;
    
            // Return Ok(Some(...)) to signal that a full frame has been produced.
            Ok(Some(line.to_string()))
        } else {
            // '\n' not found in the string.
    
            // Tell the next call to start searching after the current length of the buffer
            // since all of it was scanned and no '\n' was found.
            self.next_index = buf.len();
    
            // Ok(None) signifies that more data is needed to produce a full frame.
            Ok(None)
        }
    }

当需要将帧写入下层流时，`Encoder::encode` 方法被调用。帧必须写入缓冲区并作为一个参数。写入缓冲区的数据将在准备好发送数据时写入流。

现在让我们来看看如何为 `LinesCodec` 实现 `Encoder::encode`。

fn encode(&mut self, line: String, buf: &mut BytesMut) -> Result<(), io::Error> {
        // It's important to reserve the amount of space needed. The `bytes` API
        // does not grow the buffers implicitly.
        // Reserve the length of the string + 1 for the '\n'.
        buf.reserve(line.len() + 1);
    
        // String implements IntoBuf, a trait used by the `bytes` API to work with
        // types that can be expressed as a sequence of bytes.
        buf.put(line);
    
        // Put the '\n' in the buffer.
        buf.put_u8(b'\n');
    
        // Return ok to signal that no error occured.
        Ok(())
    }

编码信息通常更简单。这里我们只需保留所需的空间并将数据写入缓冲区。

### 使用编解码器 ###

使用编解码器的最简单方法是使用 `Framed` 结构体。它是实现自动缓冲的编解码器的包装器。该 `Framed` 结构体既是 `Stream` 也是 `Sink`。因此，您可以从中接收帧并向其发送帧。

您可以使用任何实现了 `AsyncRead` 和 `AsyncWrite` trait 的类型，使用 `AsyncRead::framed` 方法创建一个 `Framed` 结构体。

TcpStream::connect(&addr).and_then(|sock| {
        let framed_sock = sock.framed(LinesCodec::new());
        framed_sock.for_each(|line| {
            println!("Received line {}", line);
            Ok(())
        })
    });

# 推荐阅读 #

*  [Rust futures][]
 *  [Rust futures 译文][Rust futures 1]
 *  [学习 Rust Futures - Future 和 Stream][Rust Futures - Future _ Stream]
 *  [学习 Rust Futures - Executor and Task][Rust Futures - Executor and Task]
 *  [学习 Rust Futures - Concrete and Combinators][Rust Futures - Concrete and Combinators]

[https_github.com_Hevienz_mio_test_blob_master_src_main.rs]: https://github.com/Hevienz/mio_test/blob/master/src/main.rs
[Cheatsheet for Futures]: https://tokio.rs/img/diagrams/cheatsheet-for-futures.html
[https_www.jianshu.com_p_5059c403a335]: https://www.jianshu.com/p/5059c403a335%E3%80%82
[Rust futures]: https://dev.to/mindflavor/rust-futures-an-uneducated-short-and-hopefully-not-boring-tutorial---part-1-3k3
[Rust futures 1]: https://log.lain.li/blog/rust-futures-for-the-uneducated/
[Rust Futures - Future _ Stream]: https://www.jianshu.com/p/6e3acd12b233
[Rust Futures - Executor and Task]: https://www.jianshu.com/p/feafe6346929
[Rust Futures - Concrete and Combinators]: https://www.jianshu.com/p/5059c403a335