multithreading - 为什么在返回 `Future::poll` 后没有重复调用 `NotReady` ？

Question

考虑以下代码

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

use std::sync::{atomic, Arc};
use futures::*;

struct F(Arc<atomic::AtomicBool>);

impl Future for F {
    type Item = ();
    type Error = ();

    fn poll(&mut self) -> Result<Async<Self::Item>, Self::Error> {
        println!("Check if flag is set");
        if self.0.load(atomic::Ordering::Relaxed) {
            Ok(Async::Ready(()))
        } else {
            Ok(Async::NotReady)
        }
    }
}

fn main() {
    let flag = Arc::new(atomic::AtomicBool::new(false));
    let future = F(flag.clone());
    ::std::thread::spawn(move || {
        ::std::thread::sleep_ms(10);
        println!("set flag");
        flag.store(true, atomic::Ordering::Relaxed);
    });
    // ::std::thread::sleep_ms(20);
    let result = future.wait();
    println!("result: {:?}", result);
}

生成的线程设置一个标志， future 等待。我们还休眠生成的线程，因此来自 .wait() 的初始 .poll() 调用是在设置标志之前。这会导致 .wait() 无限期地(看似)阻塞。如果我们取消注释另一个 thread::sleep_ms，.wait() 返回，并打印出结果 (())。

我希望当前线程尝试通过多次调用 poll 来解决 future 问题，因为我们正在阻塞当前线程。然而，这并没有发生。

我试过阅读 some docs ，似乎问题在于线程在第一次从 poll 获得 NotReady 后被 parked。但是，我不清楚为什么会这样，或者如何解决这个问题。

我错过了什么？

Answer 1

为什么您需要停放等待的 future 而不是反复轮询？答案很明显，恕我直言。因为归根结底，它会更快、更高效!

要反复轮询 future (可能被称为“忙等待”)，图书馆必须决定是经常还是很少这样做，而且两个答案都不令人满意。经常这样做，你会浪费 CPU 周期，很少这样做，代码 react 很慢。

是的，您需要在等待某事时停放任务，然后在等待完成后将其取消停放。像这样:

#![allow(deprecated)]

extern crate futures;

use std::sync::{Arc, Mutex};
use futures::*;
use futures::task::{park, Task};

struct Status {
    ready: bool,
    task: Option<Task>,
}

#[allow(dead_code)]
struct F(Arc<Mutex<Status>>);

impl Future for F {
    type Item = ();
    type Error = ();

    fn poll(&mut self) -> Result<Async<Self::Item>, Self::Error> {
        println!("Check if flag is set");
        let mut status = self.0.lock().expect("!lock");
        if status.ready {
            Ok(Async::Ready(()))
        } else {
            status.task = Some(park());
            Ok(Async::NotReady)
        }
    }
}

#[test]
fn test() {
    let flag = Arc::new(Mutex::new(Status {
                                       ready: false,
                                       task: None,
                                   }));
    let future = F(flag.clone());
    ::std::thread::spawn(move || {
        ::std::thread::sleep_ms(10);
        println!("set flag");
        let mut status = flag.lock().expect("!lock");
        status.ready = true;
        if let Some(ref task) = status.task {
            task.unpark()
        }
    });
    let result = future.wait();
    println!("result: {:?}", result);
}

请注意 Future::poll 在这里做了几件事:它正在检查外部条件并停止任务，因此可以进行比赛，例如时间:

1. poll 检查变量并发现它是false；
2. 外部代码将变量设置为true；
3. 外部代码检查任务是否被停放，发现没有；
4. poll 停放了任务，但是砰!为时已晚，没有人会再将其打开。

为了避免任何竞争，我使用了 Mutex 来同步这些交互。

附言如果您只需要将线程结果包装到 Future 中，那么请考虑使用 oneshot channel :它有Receiver已经实现了 Future 接口(interface)。