如何在不等待的情况下轮询未来状态？

Question

我正在尝试使用 Future.rs 在单独的进程中管理一些任务。我看到了如何等待每个创建的未来以及如何一个接一个地处理它们，但我无法在其执行期间轮询未来以了解其状态。我总是有错误：

线程“主”因“当前没有任务正在运行”而恐慌

我想在以后的处理过程中做一些事情，直到它完成。也许我没有以正确的方式使用它？我设法通过使用通道使其工作，但我认为应该可以对未来进行轮询，并在它准备好时得到结果。 我用来测试的代码是：

fn main() {
    println!("test future");
    let thread_pool = CpuPool::new(4);
    let mut future_execution_list = vec![];
    let mutex = Arc::new(AtomicUsize::new(0));
    //create the future to process
    for _ in 0..10 {
        let send_mutex = mutex.clone();
        let future = thread_pool.spawn_fn(move || {
            //Simulate long processing
            thread::sleep(time::Duration::from_millis(10));
            let num = send_mutex.load(Ordering::Relaxed);
            send_mutex.store(num + 1, Ordering::Relaxed);
            let res: Result<usize, ()> = Ok(num);
            res

        });
        future_execution_list.push(future);
    }
    // do the job
    loop {
        for future in &mut future_execution_list {
            match future.poll() {
                Ok(Async::NotReady) => (), //do nothing
                Ok(Async::Ready(num)) => {
                    //update task status
                    println!(" future {:?}", num);
                }
                Err(_) => {
                    //log error and set task status to err
                    ()
                }
            };
        }
        //do something else
    }
}

所以我在 Shepmaster 回答后完成了我的问题。您的评论很有趣，但我仍然找不到解决问题的方法。我将添加一些有关我的问题的信息。我想在一次可以管理多个任务的自动化设备上安排任务。有一个循环，其中管理事件并计算任务调度。当一个任务被安排时，它就会产生。当一个任务结束时，一个新的调度就完成了。在任务执行期间，管理事件。伪代码可以是：

loop {
   event.try_recv() { ...} //manage user command for exemple
   if (schedule) {
      let tasks_to_spawn = schedule_task();
      let futures = tasks_to_spawn.map(|task| {
           thread_pool.spawn_fn( ....)}); 
      let mut one = future::select_all(futures);
      while let Ok((value, _idx, remaining)) = one.wait() {..} //wait here
   }
   //depend on task end state and event  set schedule to true or false.

}

我可以在未来联合调度和任务，例如：

let future = schedule.and_them(|task| execute_task);

但是我仍然需要等待第一个任务的执行结束。 我可以把一切都放在未来（事件管理、日程安排、任务），然后像你建议的那样等待第一个结束。我尝试过，但我没有看到如何使用不同的 Item 和 Error 类型来制作未来的 vec。有了这个概念，我必须在线程之间管理更多数据。事件管理和调度不必在不同的线程中执行。

我看到另一个问题，select_all 取得了 vec 的所有权。如果在执行另一个任务期间必须取消一个新任务，我该如何更改 vec 并添加新的未来？

不知道你是否有一个简单的解决方案。我在想，在执行过程中使用 isDone() 之类的方法获取未来状态很简单，无需等待。也许这是计划好的，我没有看到关于此的 PR。 如果您有一个简单的解决方案，那就太好了，否则我会重新考虑我的概念。

Answer 1

要轮询Future，您必须拥有Task。要获得Task，您可以在传递给futures::executor::spawn() 的Future 中进行投票。如果您像这样重写示例的loop：

futures::executor::spawn(futures::lazy(|| {
    // existing loop goes here
})).wait_future();

它运行。

至于为什么一个Future只能在任务中被轮询，我相信是这样轮询才能调用Task::Unpark。

Answer 2

我想在以后的处理过程中做点什么

据我了解，这就是未来 - 可以并行发生的事情。如果你想做别的事情，那就创造另一个未来并投入其中！

您基本上已经在这样做了 - 您的每个线程都在“做其他事情”。

轮询未来，当它准备好时得到结果

使用future::select_all，您可以组合多个期货并获得先完成的期货。然后由您决定等待下一个。

一种可能的实现方式：

extern crate rand;
extern crate futures;
extern crate futures_cpupool;

use rand::Rng;
use futures::{future, Future};
use futures_cpupool::CpuPool;

use std::{thread, time};

fn main() {
    let thread_pool = CpuPool::new(4);

    let futures = (0..10).map(|i| {
        thread_pool.spawn_fn(move || -> Result<usize, ()> {
            let mut rng = rand::thread_rng();
            // Simulate long processing
            let sleep_time = rng.gen_range(10, 100);
            let sleep_time = time::Duration::from_millis(sleep_time);
            for _ in 0..10 {
                println!("Thread {} sleeping", i);
                thread::sleep(sleep_time);
            }
            Ok(i)
        })
    });

    let mut one = future::select_all(futures);
    while let Ok((value, _idx, remaining)) = one.wait() {
        println!("Future #{} finished", value);
        if remaining.is_empty() {
            break;
        }
        one = future::select_all(remaining);
    }
}

在调用wait 期间，同时发生了多件事情！这可以从交错输出中看出：

Thread 2 sleeping
Thread 0 sleeping
Thread 3 sleeping
Thread 1 sleeping
Thread 3 sleeping
Thread 0 sleeping
Thread 1 sleeping
Thread 2 sleeping
Thread 3 sleeping

您可以通过将每个线程的睡眠时间设置为 1 秒并对整个程序进行计时来验证事情是否并行发生。由于有 10 个 future，耗时 1 秒，并行度为 4，因此整个程序运行需要 3 秒。

---

奖励代码审查：

1. 不要拆分加载和设置原子变量来实现递增 - 存储的值可能已被两个调用之间的另一个线程更改。使用fetch_add。
2. 您真的，真的在使用它们之前应该知道what the orderings mean。我不认识他们，所以我总是使用SeqCst。
3. 由于它对本示例并不重要，因此我完全删除了原子变量。
4. 总是更喜欢收集到Vec，而不是在循环中推入它。这样可以实现更优化的分配。
5. 在这种情况下，根本不需要Vec，因为select_all 接受任何可以变成迭代器的东西。