rust - 当前没有任务正在运行

Question

当我尝试在不使用 tokio 的情况下使用 UnboundedSender 和 UnboundedReceiver 时，出现错误:当前没有任务正在运行: :run().

这是错误:

Press Ctrl-C to abort
thread 'receiver thread' panicked at 'no Task is currently running', src/libcore/option.rs:1166:5
stack backtrace:
   0: backtrace::backtrace::libunwind::trace
             at /cargo/registry/src/github.com-1ecc6299db9ec823/backtrace-0.3.29/src/backtrace/libunwind.rs:88
   1: backtrace::backtrace::trace_unsynchronized
             at /cargo/registry/src/github.com-1ecc6299db9ec823/backtrace-0.3.29/src/backtrace/mod.rs:66
   2: std::sys_common::backtrace::_print
             at src/libstd/sys_common/backtrace.rs:47
   3: std::sys_common::backtrace::print
             at src/libstd/sys_common/backtrace.rs:36
   4: std::panicking::default_hook::{{closure}}
             at src/libstd/panicking.rs:200
   5: std::panicking::default_hook
             at src/libstd/panicking.rs:214
   6: std::panicking::rust_panic_with_hook
             at src/libstd/panicking.rs:477
   7: std::panicking::continue_panic_fmt
             at src/libstd/panicking.rs:384
   8: rust_begin_unwind
             at src/libstd/panicking.rs:311
   9: core::panicking::panic_fmt
             at src/libcore/panicking.rs:85
  10: core::option::expect_failed
             at src/libcore/option.rs:1166
  11: core::option::Option<T>::expect
             at /rustc/dddb7fca09dc817ba275602b950bb81a9032fb6d/src/libcore/option.rs:345
  12: futures::task_impl::with
             at /home/nickhash/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-0.1.29/src/task_impl/mod.rs:46
  13: futures::task_impl::current
             at /home/nickhash/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-0.1.29/src/task_impl/mod.rs:118
  14: futures::sync::mpsc::Receiver<T>::try_park
             at /home/nickhash/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-0.1.29/src/sync/mpsc/mod.rs:902
  15: <futures::sync::mpsc::Receiver<T> as futures::stream::Stream>::poll
             at /home/nickhash/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-0.1.29/src/sync/mpsc/mod.rs:936
  16: <futures::sync::mpsc::UnboundedReceiver<T> as futures::stream::Stream>::poll
             at /home/nickhash/.cargo/registry/src/github.com-1ecc6299db9ec823/futures-0.1.29/src/sync/mpsc/mod.rs:1004
  17: fut2::launch_receiver_thread::{{closure}}
             at src/main.rs:16
note: Some details are omitted, run with `RUST_BACKTRACE=full` for a verbose backtrace.
^C

我正在运行的代码是:

use std:: {thread,time::Duration};
use futures::stream::*;
use futures::prelude::*;


fn launch_sender_thread(tx: futures::sync::mpsc::UnboundedSender<()>) {
    let _ = thread::Builder::new().name("sender thread".to_string()).spawn(move || {
        loop {
            tx.clone().send(());
            thread::sleep(Duration::from_millis(3000));
        }   
    });
}
fn launch_receiver_thread(mut rx: futures::sync::mpsc::UnboundedReceiver<()>) {
    let _ = thread::Builder::new().name("receiver thread".to_string()).spawn(move || {
        match rx.poll() {
            Ok(Async::Ready(Some(_))) => {
                println!("thread2: received data");
            },  
            Ok(_) => (), 
            Err(err) => {
                println!("poll err {:?}",err);
            },
        } 
        thread::sleep(Duration::from_millis(1000));
    });
}

fn main() {

    let (sender, receiver) = futures::sync::mpsc::unbounded::<()>();
    launch_receiver_thread(receiver);
    launch_sender_thread(sender);
    println!("Press Ctrl-C to abort");
    thread::sleep(Duration::from_millis(10000000));        
}

所以，基本上我想从一个线程向另一个线程发送和接收数据，但不使用 Future 对象，这可能吗？如果不可能，那么我该如何通过最少使用 Futures crate 来修复这个错误？要求是使用 futures::sync::mpsc::UnboundedSender 和 futures::sync::mpsc::UnboundedSender 进行通信。

Cargo.toml:

[package]
name = "fut2"
version = "0.1.0"
edition = "2018"

[dependencies]
futures = { version = "0.1" }

Answer 1

but without using the Future object, is this possible?

可以使用 UnboundedSender 和 UnboundedSender 与 futures crate 进行最少的交互，但您必须将 futures 与任务一起使用。但这不完全是你的错； future 文档和 API 设计对此不是很清楚。在其 0.2 版本中，将有一个 task context通过各种 API 调用显式线程化。例如，Stream::poll在 0.1.x 中定义为:

fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error>

在0.2.x中，是Stream::poll_next :

fn poll_next(
    &mut self, 
    cx: &mut Context
) -> Result<Async<Option<Self::Item>>, Self::Error>

因此您将无法在任务之外犯同样的轮询错误。

在那之前这适用于 0.1.x futures crate:

use futures::sync::mpsc::{UnboundedReceiver, UnboundedSender};
use std::thread::{self, JoinHandle};

fn launch_sender_thread(tx: UnboundedSender<()>) -> JoinHandle<()> {
    thread::Builder::new()
        .name("sender thread".to_string())
        .spawn(move || loop {
            let _ = tx.unbounded_send(());
        })
        .unwrap()
}

fn launch_receiver_thread(rx: UnboundedReceiver<()>) -> JoinHandle<()> {
    thread::Builder::new()
        .name("receiver thread".to_string())
        .spawn(move || {
            let mut task = futures::executor::spawn(rx);
            match task.wait_stream() {
                Some(Ok(_)) => println!("thread2: received data"),
                Some(Err(e)) => println!("poll err {:?}", e),
                None => (),
            }
        })
        .unwrap()
}

fn main() {
    let (sender, receiver) = futures::sync::mpsc::unbounded::<()>();
    let recv = launch_receiver_thread(receiver);
    let send = launch_sender_thread(sender);

    println!("Press Ctrl-C to abort");
    recv.join().unwrap();
    send.join().unwrap();
}

请注意，您也可以使用任务驱动 tx。在这里调用 unbounded_send 更简单。