select - 为什么 Future::select 首先选择 sleep 时间更长的 future ？

Question

我正在努力理解 Future::select : 在此示例中，首先返回具有较长时间延迟的 future 。

当我读到this article以它的例子，我得到了认知失调。作者写道:

The select function runs two (or more in case of select_all) futures and returns the first one coming to completion. This is useful for implementing timeouts.

看来我不明白select的意义。

extern crate futures; // v0.1 (old)
extern crate tokio_core;

use std::thread;
use std::time::Duration;
use futures::{Async, Future};
use tokio_core::reactor::Core;

struct Timeout {
    time: u32,
}

impl Timeout {
    fn new(period: u32) -> Timeout {
        Timeout { time: period }
    }
}

impl Future for Timeout {
    type Item = u32;
    type Error = String;

    fn poll(&mut self) -> Result<Async<u32>, Self::Error> {
        thread::sleep(Duration::from_secs(self.time as u64));
        println!("Timeout is done with time {}.", self.time);
        Ok(Async::Ready(self.time))
    }
}

fn main() {
    let mut reactor = Core::new().unwrap();

    let time_out1 = Timeout::new(5);
    let time_out2 = Timeout::new(1);

    let task = time_out1.select(time_out2);

    let mut reactor = Core::new().unwrap();
    reactor.run(task);
}

我需要处理时间延迟较小的早期 future ，然后处理延迟较长的 future 。我该怎么做？

Answer 1

TL;DR: 使用 tokio::time

如果有一点可以避免:永远不要在异步操作中执行阻塞或长时间运行的操作。

如果你想要超时，使用来自 tokio::time 的东西，例如 delay_for或 timeout :

use futures::future::{self, Either}; // 0.3.1
use std::time::Duration;
use tokio::time; // 0.2.9

#[tokio::main]
async fn main() {
    let time_out1 = time::delay_for(Duration::from_secs(5));
    let time_out2 = time::delay_for(Duration::from_secs(1));

    match future::select(time_out1, time_out2).await {
        Either::Left(_) => println!("Timer 1 finished"),
        Either::Right(_) => println!("Timer 2 finished"),
    }
}

有什么问题？

要理解为什么会出现这种行为，您必须从较高的层次理解 Futures 的实现。

当您调用 run 时，会有一个循环调用 poll 传入的 future。它循环直到 future 返回成功或失败，否则 future 尚未完成。

poll 的实现“锁定”了这个循环 5 秒，因为没有什么可以中断对 sleep 的调用。当 sleep 完成时， future 已经准备就绪，因此选择了 future 。

异步超时的实现在概念上是通过在每次轮询时检查时钟来判断是否已经过了足够的时间。

最大的区别在于，当一个 future 返回它还没有准备好时，可以检查另一个 future 。这就是 select 所做的!

戏剧性的重演:

基于 sleep 的定时器

core: Hey select, are you ready to go?

select: Hey future1, are you ready to go?

future1: Hold on a seconnnnnnnn [... 5 seconds pass ...] nnnnd. Yes!

基于异步的简单计时器

core: Hey select, are you ready to go?

select: Hey future1, are you ready to go?

future1: Checks watch No.

select: Hey future2, are you ready to go?

future2: Checks watch No.

core: Hey select, are you ready to go?

[... polling continues ...]

[... 1 second passes ...]

core: Hey select, are you ready to go?

select: Hey future1, are you ready to go?

future1: Checks watch No.

select: Hey future2, are you ready to go?

future2: Checks watch Yes!

这个简单的实现会一遍又一遍地轮询 future ，直到它们全部完成。这不是最有效的，也不是大多数执行者所做的。

参见 How do I execute an async/await function without using any external dependencies?用于实现这种执行器。

基于异步的智能计时器

core: Hey select, are you ready to go?

select: Hey future1, are you ready to go?

future1: Checks watch No, but I'll call you when something changes.

select: Hey future2, are you ready to go?

future2: Checks watch No, but I'll call you when something changes.

[... core stops polling ...]

[... 1 second passes ...]

future2: Hey core, something changed.

core: Hey select, are you ready to go?

select: Hey future1, are you ready to go?

future1: Checks watch No.

select: Hey future2, are you ready to go?

future2: Checks watch Yes!

这个更有效的实现在轮询时将唤醒器交给每个 future 。当 future 还没有准备好时，它会保存那个唤醒者以备后用。当事情发生变化时，唤醒者通知执行者的核心，现在是重新检查 future 的好时机。这允许执行者不执行有效的忙等待。

通用解决方案

当您有一个阻塞或长时间运行的操作时，适当的做法是将该工作移出异步循环。参见 What is the best approach to encapsulate blocking I/O in future-rs?有关详细信息和示例。