键盘快捷键

按 ← 或 → 在章节之间导航

按 S 或 / 在本书中搜索

按 ? 显示此帮助

按 Esc 隐藏此帮助

进一步了解 Async 相关的 trait

在本章中，我们以多种方式使用了 Future、Stream 和 StreamExt trait。不过到目前为止，我们一直避免过多深入它们如何工作以及如何组合在一起 的细节——对于日常 Rust 工作来说，这在大多数情况下都没问题。但有时你会遇到 一些场景，需要进一步了解这些 trait 的细节， 以及 Pin 类型和 Unpin trait。在本节中，我们将深入探讨到足以 应对这些场景的程度，真正更深入的内容则留给其他文档。

Future trait

让我们先仔细看看 Future trait 的工作原理。Rust 对它的 定义如下：

#![allow(unused)]
fn main() {
use std::pin::Pin;
use std::task::{Context, Poll};

pub trait Future {
    type Output;

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

这个 trait 定义包含了一些新类型，还有一些我们 之前没见过的语法，让我们逐部分讲解这个定义。

首先，Future 的关联类型 Output 表明 future 最终会解析出什么。 这类似于 Iterator trait 的 Item 关联类型。 其次，Future 有一个 poll 方法，它的 self 参数接受一个特殊的 Pin 引用， 并接受一个 Context 类型的可变引用， 返回 Poll<Self::Output>。我们稍后会进一步讨论 Pin 和 Context。 现在，先专注于这个方法的返回值——Poll 类型：

#![allow(unused)]
fn main() {
pub enum Poll<T> {
    Ready(T),
    Pending,
}
}

这个 Poll 类型与 Option 类似。它有一个带值的变体 Ready(T)，以及一个不带值的变体 Pending。不过，Poll 的含义与 Option 大不相同！Pending 变体表明 future 仍有工作要做，因此调用者稍后需要再次检查。Ready 变体表明 Future 已完成其工作，T 值 可用了。

注意：需要直接调用 poll 的情况很罕见，但如果确实需要，请记住： 对于大多数 future，在 future 返回 Ready 之后， 调用者不应再次调用 poll。许多 future 在就绪后被再次 poll 时会 panic。可以安全地再次 poll 的 future 会在 其文档中明确说明。这与 Iterator::next 的行为类似。

当你看到使用 await 的代码时，Rust 会在底层将其编译为调用 poll 的代码。回顾一下示例 17-4，我们在其中打印了 单个 URL 解析完成后的页面标题，Rust 会将其编译成 类似（但不完全等同）这样的代码：

match page_title(url).poll() {
    Ready(page_title) => match page_title {
        Some(title) => println!("The title for {url} was {title}"),
        None => println!("{url} had no title"),
    }
    Pending => {
        // But what goes here?
    }
}

当 future 仍处于 Pending 时，我们该怎么办？我们需要某种方式 不断重试，一次又一次，直到 future 最终就绪。换句话说， 我们需要一个循环：

let mut page_title_fut = page_title(url);
loop {
    match page_title_fut.poll() {
        Ready(value) => match page_title {
            Some(title) => println!("The title for {url} was {title}"),
            None => println!("{url} had no title"),
        }
        Pending => {
            // continue
        }
    }
}

然而，如果 Rust 真的把它编译成那样的代码，每个 await 都会是 阻塞的——这与我们的目标恰恰相反！相反，Rust 会确保 这个循环能把控制权交给某个可以暂停这个 future 的工作、 转而去处理其他 future、之后再回来检查这个 future 的东西。正如我们 所见，那个东西就是异步运行时，而这种调度与协调 工作是它的主要职责之一。

在「使用消息传递在两个任务之间发送 数据」一节中，我们介绍了等待 rx.recv。recv 调用返回一个 future，而 await 这个 future 就会对它进行 poll。 我们提到过，运行时会暂停该 future，直到它就绪——要么返回 Some(message)，要么在信道关闭时返回 None。有了对 Future trait、特别是 Future::poll 的更深入理解，我们就能 明白这是如何工作的。当它返回 Poll::Pending 时，运行时就知道 future 尚未就绪。相反，当 poll 返回 Poll::Ready(Some(message)) 或 Poll::Ready(None) 时，运行时就知道 future 已 就绪并推进它。

运行时具体如何做到这一点的细节超出了本书的范围， 但关键在于理解 future 的基本机制：运行时会轮询（poll）它负责的每一个 future，并在其尚未就绪时将其放回休眠状态。

Pin 类型与 Unpin trait

回顾示例 17-13，我们使用 trpl::join! 宏来 await 三个 future。然而，常见的情况是有一个集合（比如 vector），其中包含的 future 数量要到运行时才能确定。让我们把示例 17-13 改成示例 17-23 中的代码：将三个 future 放入一个 vector 并改为调用 trpl::join_all 函数——这段代码还无法编译。

Filename: src/main.rs 
extern crate trpl; // required for mdbook test

use std::time::Duration;

fn main() {
    trpl::block_on(async {
        let (tx, mut rx) = trpl::channel();

        let tx1 = tx.clone();
        let tx1_fut = async move {
            let vals = vec![
                String::from("hi"),
                String::from("from"),
                String::from("the"),
                String::from("future"),
            ];

            for val in vals {
                tx1.send(val).unwrap();
                trpl::sleep(Duration::from_secs(1)).await;
            }
        };

        let rx_fut = async {
            while let Some(value) = rx.recv().await {
                println!("received '{value}'");
            }
        };

        let tx_fut = async move {
            // --snip--
            let vals = vec![
                String::from("more"),
                String::from("messages"),
                String::from("for"),
                String::from("you"),
            ];

            for val in vals {
                tx.send(val).unwrap();
                trpl::sleep(Duration::from_secs(1)).await;
            }
        };

        let futures: Vec<Box<dyn Future<Output = ()>>> =
            vec![Box::new(tx1_fut), Box::new(rx_fut), Box::new(tx_fut)];

        trpl::join_all(futures).await;
    });
}
示例 17-23：await 集合中的多个 future

我们将每个 future 放入 Box 中，使它们成为trait 对象（trait objects）， 就像我们在第 12 章「从 run 返回错误」一节中所做的那样。（我们将在 第 18 章详细介绍 trait 对象。）使用 trait 对象让我们可以把 这些类型产生的匿名 future 当作同一种类型来处理， 因为它们都实现了 Future trait。

这可能令人惊讶。毕竟，这些异步代码块都不返回任何东西， 所以每个都产生 Future<Output = ()>。但请记住，Future 是一个 trait，编译器会为每个异步代码块创建一个独特的枚举， 即使它们具有相同的输出类型。正如你不能把 两个不同的手写结构体放进同一个 Vec 一样，你也不能混合使用编译器生成的 枚举。

然后我们将这个 future 集合传给 trpl::join_all 函数并 await 其结果。然而，这无法编译；下面是错误信息中的相关部分。

error[E0277]: `dyn Future<Output = ()>` cannot be unpinned
  --> src/main.rs:48:33
   |
48 |         trpl::join_all(futures).await;
   |                                 ^^^^^ the trait `Unpin` is not implemented for `dyn Future<Output = ()>`
   |
   = note: consider using the `pin!` macro
           consider using `Box::pin` if you need to access the pinned value outside of the current scope
   = note: required for `Box<dyn Future<Output = ()>>` to implement `Future`
note: required by a bound in `futures_util::future::join_all::JoinAll`
  --> file:///home/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/futures-util-0.3.30/src/future/join_all.rs:29:8
   |
27 | pub struct JoinAll<F>
   |            ------- required by a bound in this struct
28 | where
29 |     F: Future,
   |        ^^^^^^ required by this bound in `JoinAll`

错误信息中的提示告诉我们，应该使用 pin! 宏来固定（pin） 这些值，也就是把它们放进 Pin 类型中， 以保证这些值不会在内存中移动。错误信息说需要固定， 是因为 dyn Future<Output = ()> 需要实现 Unpin trait，而它目前并没有实现。

trpl::join_all 函数返回一个名为 JoinAll 的结构体。该结构体 泛型于类型 F，F 被约束为实现 Future trait。 直接用 await 等待一个 future 会隐式地固定该 future。这就是 我们不需要在每个想 await future 的地方都使用 pin! 的原因。

然而，这里我们并没有直接 await 一个 future。相反，我们通过将 future 集合传给 join_all 函数来构造一个新的 future——JoinAll。 join_all 的签名要求集合中各项的类型 都实现 Future trait，而 Box<T> 仅当其包裹的 T 是 实现了 Unpin trait 的 future 时才实现 Future。

要消化的内容真不少！为了真正理解它，让我们进一步 深入 Future trait 的实际工作原理，特别是关于固定的部分。 再看看 Future trait 的定义：

#![allow(unused)]
fn main() {
use std::pin::Pin;
use std::task::{Context, Poll};

pub trait Future {
    type Output;

    // Required method
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>;
}
}

cx 参数及其 Context 类型是运行时在保持惰性的同时、知道何时检查 某个 future 的关键。同样，其工作原理的细节 超出了本章的范围，通常只有在编写自定义 Future 实现时才需要 考虑这一点。我们将重点放在 self 的类型上，因为这是我们第一次见到 self 带有类型标注的方法。self 的类型标注 与其他函数参数的类型标注类似，但有两个关键 区别：

  • 它告诉 Rust 调用该方法时 self 必须是什么类型。
  • 它不能是任意类型。它被限制为：实现该方法的类型、 该类型的引用或智能指针，或者包裹该类型引用的 Pin。

我们将在第 18 章看到更多关于这种语法的内容。 目前只需知道：如果我们想 poll 一个 future 来检查它是 Pending 还是 Ready(Output)，就需要一个由 Pin 包裹的指向该 类型的可变引用。

Pin 是类指针类型的包装器，如 &、&mut、Box 和 Rc。 （严格来说，Pin 适用于实现了 Deref 或 DerefMut trait 的类型，但这实际上等效于只处理引用和 智能指针。）Pin 本身不是指针，也不像 Rc 和 Arc 那样具有 引用计数之类的自身行为；它纯粹是编译器 用来强制执行指针使用约束的工具。

回想一下 await 是通过调用 poll 实现的，这开始解释我们之前 看到的错误信息了，但错误信息涉及的是 Unpin 而不是 Pin。那么 Pin 与 Unpin 究竟是什么关系，为什么 Future 需要 self 位于 Pin 类型中才能调用 poll 呢？

还记得本章前面提到的，future 中的一系列 await 点 会被编译成一个状态机，编译器确保该状态 机遵循 Rust 所有关于安全性的常规规则，包括借用 和所有权。为了实现这一点，Rust 会查看一个 await 点与下一个 await 点或异步代码块结束之间需要哪些数据， 然后在编译后的状态机中创建相应的变体。每个 变体都获得对源代码相应部分所用数据的访问权， 无论是取得该数据的所有权，还是获得其可变或不可变引用。

到目前为止一切顺利：如果我们在某个异步代码块中的所有权或引用 出了任何问题，借用检查器会告诉我们。但当我们想要移动 与该代码块对应的 future 时——比如把它移入 Vec 传给 join_all——事情就变得棘手了。

当我们移动一个 future 时——无论是将其推入数据结构以便用 join_all 迭代，还是从函数返回它——实际上移动的是 Rust 为我们创建的状态机。而且与 Rust 中大多数其他类型不同， Rust 为异步代码块创建的 future，其某个变体的字段中 可能包含对自身的引用，如图 17-4 的简化示意图所示。

A single-column, three-row table representing a future, fut1, which has data values 0 and 1 in the first two rows and an arrow pointing from the third row back to the second row, representing an internal reference within the future.
图 17-4：自引用数据类型

然而，默认情况下，任何包含对自身引用的对象移动起来都是不安全的， 因为引用总是指向其所引用内容的实际内存地址 （见图 17-5）。如果移动了数据结构本身，那些 内部引用仍会指向旧位置。然而，那个 内存位置现在已经失效了。一方面，当你修改 数据结构时，它的值不会被更新。另一方面——更重要的 是——计算机现在可以随意将那块内存挪作他用！你之后 可能会读到完全无关的数据。

Two tables, depicting two futures, fut1 and fut2, each of which has one column and three rows, representing the result of having moved a future out of fut1 into fut2. The first, fut1, is grayed out, with a question mark in each index, representing unknown memory. The second, fut2, has 0 and 1 in the first and second rows and an arrow pointing from its third row back to the second row of fut1, representing a pointer that is referencing the old location in memory of the future before it was moved.
图 17-5：移动自引用数据类型导致的不安全结果

从理论上讲，Rust 编译器可以尝试在对象每次被移动时更新指向它的 所有引用，但这会增加大量性能开销， 尤其是在需要更新整张引用网络的情况下。相反，如果我们能确保 所涉及的数据结构不在内存中移动，就无需 更新任何引用。这正是 Rust 借用检查器的用途： 在安全代码中，它阻止你移动任何存在活动引用的项。

Pin 在此基础上提供了我们恰好需要的保证。当我们通过将指向 某个值的指针包裹在 Pin 中来固定该值时，它就不能再移动了。因此， 如果你有 Pin<Box<SomeType>>，实际上固定的是 SomeType 值，而不是 Box 指针。图 17-6 说明了这一过程。

Three boxes laid out side by side. The first is labeled “Pin”, the second “b1”, and the third “pinned”. Within “pinned” is a table labeled “fut”, with a single column; it represents a future with cells for each part of the data structure. Its first cell has the value “0”, its second cell has an arrow coming out of it and pointing to the fourth and final cell, which has the value “1” in it, and the third cell has dashed lines and an ellipsis to indicate there may be other parts to the data structure. All together, the “fut” table represents a future which is self-referential. An arrow leaves the box labeled “Pin”, goes through the box labeled “b1” and terminates inside the “pinned” box at the “fut” table.
图 17-6：固定指向自引用 future 类型的 `Box`

事实上，Box 指针本身仍然可以自由移动。请记住：我们关心 的是确保最终被引用的数据保持不动。如果指针移动了， 但它所指向的数据仍在原处，如图 17-7 所示，就不存在潜在问题。（作为独立练习，请查阅 这些类型以及 std::pin 模块的文档，尝试弄清楚如何用包裹 Box 的 Pin 来做到这一点。）关键在于自引用类型 本身不能移动，因为它仍被固定着。

Four boxes laid out in three rough columns, identical to the previous diagram with a change to the second column. Now there are two boxes in the second column, labeled “b1” and “b2”, “b1” is grayed out, and the arrow from “Pin” goes through “b2” instead of “b1”, indicating that the pointer has moved from “b1” to “b2”, but the data in “pinned” has not moved.
图 17-7：移动指向自引用 future 类型的 `Box`

然而，大多数类型即使恰好位于 Pin 指针之后，移动起来也是 完全安全的。只有当项具有内部引用时，我们才需要 考虑固定的问题。数字和布尔值等原始类型是安全的， 因为它们显然没有任何内部引用。 你在 Rust 中日常使用的大多数类型也是如此。例如，你可以随意移动 一个 Vec 而无需担心。根据我们目前所见，如果 你有一个 Pin<Vec<String>>，你就必须通过 Pin 提供的安全但 受限的 API 来做所有事情——尽管 Vec<String> 在没有其他引用的情况下 移动总是安全的。我们需要一种方法来告诉 编译器，在这类情况下移动项是没问题的——这就是 Unpin 的用武之地。

Unpin 是一个标记 trait，类似于我们在第 16 章见过的 Send 和 Sync trait， 因此它自身没有任何功能。标记 trait 的存在只是为了 告诉编译器：在特定上下文中使用实现了给定 trait 的类型是 安全的。Unpin 告知编译器：给定类型不 需要就相关值能否安全移动而遵守任何保证。

与 Send 和 Sync 一样，编译器会为所有能证明安全的类型自动 实现 Unpin。同样与 Send 和 Sync 类似，有一个特殊情况： 某个类型不实现 Unpin。其表示法为 impl !Unpin for SomeType，其中 SomeType 是这样一个类型的名称：每当指向该类型的指针被用于 Pin 中时，它确实需要遵守那些保证才是安全的。

换句话说，关于 Pin 和 Unpin 的关系，有两点需要 记住。第一，Unpin 是「正常」情况，!Unpin 是 特殊情况。第二，一个类型实现的是 Unpin 还是 !Unpin，只有 在你使用指向该类型的固定指针（如 Pin<&mut SomeType>）时才重要。

具体来说，想想 String：它有一个长度和组成它的 Unicode 字符。我们可以把 String 包裹在 Pin 中，如图 17-8 所示。然而，String 自动实现了 Unpin，Rust 中的大多数其他类型 也是如此。

A box labeled “Pin” on the left with an arrow going from it to a box labeled “String” on the right. The “String” box contains the data 5usize, representing the length of the string, and the letters “h”, “e”, “l”, “l”, and “o” representing the characters of the string “hello” stored in this String instance. A dotted rectangle surrounds the “String” box and its label, but not the “Pin” box.
图 17-8：固定一个 `String`；虚线表示该 `String` 实现了 `Unpin` trait，因此并未被固定

因此，我们可以做一些在 String 实现 !Unpin 时属于非法的操作，比如像图 17-9 那样在内存中完全相同的位置 用另一个字符串替换原来的字符串。这并不违反 Pin 契约， 因为 String 没有会让移动变得不安全的内部引用。 这正是它实现 Unpin 而非 !Unpin 的原因。

The same “hello” string data from the previous example, now labeled “s1” and grayed out. The “Pin” box from the previous example now points to a different String instance, one that is labeled “s2”, is valid, has a length of 7usize, and contains the characters of the string “goodbye”. s2 is surrounded by a dotted rectangle because it, too, implements the Unpin trait.
图 17-9：在内存中用完全不同的 `String` 替换原来的 `String`

现在我们已经掌握了足够的知识，可以理解示例 17-23 中那个 join_all 调用 所报告的错误了。我们最初尝试将异步代码块产生的 future 移入 Vec<Box<dyn Future<Output = ()>>>，但正如我们所见， 这些 future 可能包含内部引用，因此它们不会自动 实现 Unpin。一旦我们将它们固定，就可以将得到的 Pin 类型传入 Vec，并确信 future 中的底层数据不会被 移动。示例 17-24 展示了如何通过在三个 future 各自的定义处调用 pin! 宏 并调整 trait 对象类型来修复代码。

extern crate trpl; // required for mdbook test

use std::pin::{Pin, pin};

// --snip--

use std::time::Duration;

fn main() {
    trpl::block_on(async {
        let (tx, mut rx) = trpl::channel();

        let tx1 = tx.clone();
        let tx1_fut = pin!(async move {
            // --snip--
            let vals = vec![
                String::from("hi"),
                String::from("from"),
                String::from("the"),
                String::from("future"),
            ];

            for val in vals {
                tx1.send(val).unwrap();
                trpl::sleep(Duration::from_secs(1)).await;
            }
        });

        let rx_fut = pin!(async {
            // --snip--
            while let Some(value) = rx.recv().await {
                println!("received '{value}'");
            }
        });

        let tx_fut = pin!(async move {
            // --snip--
            let vals = vec![
                String::from("more"),
                String::from("messages"),
                String::from("for"),
                String::from("you"),
            ];

            for val in vals {
                tx.send(val).unwrap();
                trpl::sleep(Duration::from_secs(1)).await;
            }
        });

        let futures: Vec<Pin<&mut dyn Future<Output = ()>>> =
            vec![tx1_fut, rx_fut, tx_fut];

        trpl::join_all(futures).await;
    });
}
示例 17-24：固定这些 future，以便将它们移入 vector

这个示例现在可以编译并运行了，而且我们可以在运行时向 vector 中添加或移除 future，并将它们全部 join 起来。

Pin 和 Unpin 主要对构建底层库或构建运行时本身很重要， 而不是日常 Rust 代码。不过，当你在错误信息中看到 这些 trait 时，现在你会更清楚如何修复代码了！

注意：Pin 和 Unpin 的组合使得在 Rust 中安全地实现一整类 复杂类型成为可能，否则这些类型会因自引用而 难以实现。需要 Pin 的类型如今最常出现在异步 Rust 中， 但偶尔你也可能在其他场景中见到它们。

Pin 和 Unpin 的工作原理细节以及它们需要遵守 的规则，在 std::pin 的 API 文档中有详细介绍，所以 如果你有兴趣深入了解，那是一个很好的起点。

如果你想更详细地了解底层工作原理， 请参阅《Asynchronous Programming in Rust》的 第 2 章和第 4 章。

Stream trait

现在你已经对 Future、Pin 和 Unpin trait 有了更深入的理解， 我们可以把注意力转向 Stream trait。正如你在本章前面所学到的， 流类似于异步迭代器。然而，与 Iterator 和 Future 不同，截至撰写本文时，Stream 在标准库中尚 无定义，但有一个来自 futures crate 的、 在整个生态系统中广泛使用的常见定义。

在看看 Stream trait 如何将它们融合在一起之前，让我们回顾一下 Iterator 和 Future trait 的定义。从 Iterator 那里， 我们得到了序列的概念：它的 next 方法提供 Option<Self::Item>。从 Future 那里，我们得到了随时间推移而就绪的概念： 它的 poll 方法提供 Poll<Self::Output>。为了表示随时间推移 逐渐就绪的项序列，我们定义了一个将这些 特性组合在一起的 Stream trait：

#![allow(unused)]
fn main() {
use std::pin::Pin;
use std::task::{Context, Poll};

trait Stream {
    type Item;

    fn poll_next(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>
    ) -> Poll<Option<Self::Item>>;
}
}

Stream trait 定义了一个名为 Item 的关联类型，表示 流所产生的项的类型。这类似于 Iterator——可能有 零个到多个项；又不同于 Future——后者总是只有一个 Output，哪怕它是单元类型 ()。

Stream 还定义了一个获取这些项的方法。我们称之为 poll_next， 以明确它的轮询方式与 Future::poll 相同，并且像 Iterator::next 那样产生一系列项。它的返回类型 将 Poll 与 Option 组合在一起。外层类型是 Poll， 因为必须像 future 一样检查它就绪与否。内层类型是 Option， 因为需要像迭代器一样表明是否还有更多消息。

与此非常相似的定义很可能最终会成为 Rust 标准库的一部分。与此同时，它是大多数运行时 工具包的一部分，所以你可以放心依赖它，我们接下来介绍的内容 通常也都适用！

不过，在「Stream：按顺序到达的 Future」一节的示例中，我们并没有使用 poll_next 或 Stream， 而是使用了 next 和 StreamExt。当然，我们可以 通过手写自己的 Stream 状态机来直接使用 poll_next API，就像我们可以通过 poll 方法直接使用 future 一样。不过，使用 await 要方便得多，而 StreamExt trait 提供了 next 方法，让我们可以这样做：

#![allow(unused)]
fn main() {
use std::pin::Pin;
use std::task::{Context, Poll};

trait Stream {
    type Item;
    fn poll_next(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Self::Item>>;
}

trait StreamExt: Stream {
    async fn next(&mut self) -> Option<Self::Item>
    where
        Self: Unpin;

    // other methods...
}
}

注意：我们在本章前面使用的实际定义与此略有 不同，因为它要兼容尚不支持在 trait 中使用异步函数的 Rust 版本。因此，它看起来是这样的：

fn next(&mut self) -> Next<'_, Self> where Self: Unpin;

那个 Next 类型是一个实现了 Future 的 struct，它允许我们 用 Next<'_, Self> 来命名 self 引用的生命周期， 从而使 await 可以与这个方法配合使用。

StreamExt trait 也是所有可用于流的实用方法的归宿。 StreamExt 会为每个实现了 Stream 的类型自动实现， 但这两个 trait 是分开定义的，以便 社区能够在不影响基础 trait 的情况下迭代便捷 API。

在 trpl crate 使用的 StreamExt 版本中，该 trait 不仅 定义了 next 方法，还提供了 next 的默认实现， 正确处理了调用 Stream::poll_next 的细节。这意味着， 即使你需要编写自己的流数据类型，也只需 实现 Stream，之后任何使用你数据类型的人都可以自动 使用 StreamExt 及其方法。

关于这些 trait 的底层细节，我们就介绍到这里。 最后，让我们思考一下 future（包括流）、任务和线程 是如何协同配合的！