asynchronous - 如何实现从 Futures 0.2 中的另一个线程唤醒的 Future？

Question

我正试图找到一种简单的方法来从 future crate 0.2.1 版中实现 Future 特征:

extern crate futures;

use futures::executor::ThreadPool;
use futures::prelude::*;
use futures::task::Context;
use std::{thread, time::Duration};

struct SendThree {
    firstTime: bool,
}

impl Future for SendThree {
    type Item = u32;
    type Error = Never;

    fn poll(&mut self, ctx: &mut Context) -> Result<Async<Self::Item>, Never> {
        if self.firstTime {
            self.firstTime = false;
            thread::spawn(move || {
                thread::sleep(Duration::from_millis(10));
                ctx.waker().wake();
            });
            Ok(Async::Pending)
        } else {
            Ok(Async::Ready(3))
        }
    }
}

fn main() {
    let mut fut = SendThree { firstTime: true };
    let mut executor: ThreadPool = ThreadPool::new().unwrap();
    let result = executor.run(fut).unwrap();
    println!("{}", result);
}

playground

我的问题是 Context 变量不是 Send 所以我不能从另一个线程调用 wake:

error[E0277]: the trait bound `futures::executor::Executor: std::marker::Send` is not satisfied
  --> src/main.rs:19:13
   |
19 |             thread::spawn(move || {
   |             ^^^^^^^^^^^^^ `futures::executor::Executor` cannot be sent between threads safely
   |
   = help: the trait `std::marker::Send` is not implemented for `futures::executor::Executor`
   = note: required because of the requirements on the impl of `std::marker::Send` for `&mut futures::executor::Executor`
   = note: required because it appears within the type `std::option::Option<&mut futures::executor::Executor>`
   = note: required because it appears within the type `futures::task::Context<'_>`
   = note: required because of the requirements on the impl of `std::marker::Send` for `&mut futures::task::Context<'_>`
   = note: required because it appears within the type `[closure@src/main.rs:19:27: 22:14 ctx:&mut futures::task::Context<'_>]`
   = note: required by `std::thread::spawn`

如果我将代码更改为此它可以工作，但我无法在不阻塞的情况下执行 thread::sleep:

if self.firstTime {
    self.firstTime = false;
    ctx.waker().wake();
    Ok(Async::Pending)
}

有没有一种惯用的方法来实现这个？

Answer 1

虽然 Context 不能跨线程发送，但是 Waker 是:

fn poll(&mut self, ctx: &mut Context) -> Result<Async<Self::Item>, Never> {
    if self.first_time {
        self.first_time = false;
        let waker = ctx.waker().clone();
        thread::spawn(move || {
            thread::sleep(Duration::from_millis(10));
            waker.wake();
        });
        Ok(Async::Pending)
    } else {
        Ok(Async::Ready(3))
    }
}

请注意，这是实现超时的一种非常低效的方法。应该改用 Futures 原生方法。