使用tokio 0.1.x生成具有非静态生命周期的任务

Question

我有一个tokio核心，其主要任务是运行websocket（客户端）。当我从服务器接收到一些消息时，我想执行一个新的任务来更新一些数据。以下是一个最小化失败的例子：

use tokio_core::reactor::{Core, Handle};
use futures::future::Future;
use futures::future;

struct Client {
    handle: Handle,
    data: usize,
}

impl Client {
    fn update_data(&mut self) {
        // spawn a new task that updates the data
        self.handle.spawn(future::ok(()).and_then(|x| {
            self.data += 1; // error here
            future::ok(())
        }));
    }
}

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

    let mut client = Client {
        handle: runtime.handle(),
        data: 0,
    };

    let task = future::ok::<(), ()>(()).and_then(|_| {
        // under some conditions (omitted), we update the data
        client.update_data();
        future::ok::<(), ()>(())
    });
    runtime.run(task).unwrap();
}

这会产生以下错误：

error[E0477]: the type `futures::future::and_then::AndThen<futures::future::result_::FutureResult<(), ()>, futures::future::result_::FutureResult<(), ()>, [closure@src/main.rs:13:51: 16:10 self:&mut &mut Client]>` does not fulfill the required lifetime
  --> src/main.rs:13:21                                                                                                                                                                
   |                                                                                                                                                                                   
13 |         self.handle.spawn(future::ok(()).and_then(|x| {                                                                                                                           
   |                     ^^^^^                                                                                                                                                         
   |                                                                                                                                                                                   
   = note: type must satisfy the static lifetime      

问题在于通过句柄生成的新任务需要是静态的。同样的问题在这里有描述。不幸的是，我不清楚该如何解决这个问题。即使尝试使用Arc和Mutex（对于单线程应用程序来说确实不需要），也没有成功。

由于Tokio领域的发展相当迅速，我想知道当前最佳解决方案是什么。你有什么建议吗？

编辑

Peter Hall的解决方案适用于上面的示例。不幸的是，当我构建失败的示例时，我更改了Tokio反应器，认为它们会很相似。使用tokio::runtime::current_thread

use futures::future;
use futures::future::Future;
use futures::stream::Stream;
use std::cell::Cell;
use std::rc::Rc;
use tokio::runtime::current_thread::{Builder, Handle};

struct Client {
    handle: Handle,
    data: Rc<Cell<usize>>,
}

impl Client {
    fn update_data(&mut self) {
        // spawn a new task that updates the data
        let mut data = Rc::clone(&self.data);
        self.handle.spawn(future::ok(()).and_then(move |_x| {
            data.set(data.get() + 1);
            future::ok(())
        }));
    }
}

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

    let mut runtime = Builder::new().build().unwrap();

    let mut client = Client {
        handle: runtime.handle(),
        data: Rc::new(Cell::new(1)),
    };

    let task = future::ok::<(), ()>(()).and_then(|_| {
        // under some conditions (omitted), we update the data
        client.update_data();
        future::ok::<(), ()>(())
    });
    runtime.block_on(task).unwrap();
}

我得到：

error[E0277]: `std::rc::Rc<std::cell::Cell<usize>>` cannot be sent between threads safely
--> src/main.rs:17:21                                                         
|                                                                            
17 |         self.handle.spawn(future::ok(()).and_then(move |_x| {              
|                     ^^^^^ `std::rc::Rc<std::cell::Cell<usize>>` cannot be sent between threads safely
|                                                                            
= help: within `futures::future::and_then::AndThen<futures::future::result_::FutureResult<(), ()>, futures::future::result_::FutureResult<(), ()>, [closure@src/main.rs:17:51: 20:10 data:std::rc::Rc<std::cell::Cell<usize>>]>`, the trait `std::marker::Send` is not implemented for `std::rc::Rc<std::cell::Cell<usize>>`
= note: required because it appears within the type `[closure@src/main.rs:17:51: 20:10 data:std::rc::Rc<std::cell::Cell<usize>>]`
= note: required because it appears within the type `futures::future::chain::Chain<futures::future::result_::FutureResult<(), ()>, futures::future::result_::FutureResult<(), ()>, [closure@src/main.rs:17:51: 20:10 data:std::rc::Rc<std::cell::Cell<usize>>]>`
= note: required because it appears within the type `futures::future::and_then::AndThen<futures::future::result_::FutureResult<(), ()>, futures::future::result_::FutureResult<(), ()>, [closure@src/main.rs:17:51: 20:10 data:std::rc::Rc<std::cell::Cell<usize>>]>`

所以在这种情况下，即使整个代码是单线程的，似乎我仍需要一个Arc和一个Mutex？

- Ben Ruijl

---

请不要更新您已回答的问题以提出新问题。相反，请花时间创建一个改进的MCVE，提出新问题，解释两个问题的区别，并可能在它们之间链接。 - Shepmaster

Answer 1

在单线程程序中，您不需要使用 Arc；Rc 就足够了：

use std::{rc::Rc, cell::Cell};

struct Client {
    handle: Handle,
    data: Rc<Cell<usize>>,
}

impl Client {
    fn update_data(&mut self) {
        let data = Rc::clone(&self.data);
        self.handle.spawn(future::ok(()).and_then(move |_x| {
            data.set(data.get() + 1);
            future::ok(())
        }));
    }
}

重点是您不再需要担心生命周期，因为每个 Rc 的克隆都表现得像它拥有数据一样，而不是通过对 self 的引用来访问数据。内部的 Cell（或非 Copy 类型的 RefCell）是必需的，因为无法对 Rc 进行可变引用，因为它已被克隆。

---

tokio::runtime::current_thread::Handle 的 spawn 方法要求未来是 Send 的，这就是更新您的问题时出现问题的原因。在 Tokio Github 问题 中解释了这种情况的原因（某种程度上）。

您可以使用 tokio::runtime::current_thread::spawn 替代 Handle 的方法，在当前线程中始终运行未来，并且不需要未来是 Send。您可以在上面的代码中替换 self.handle.spawn，它将正常工作。

如果您需要使用 Handle 上的方法，则还需要使用 Arc 和 Mutex（或 RwLock）来满足 Send 要求：

use std::sync::{Mutex, Arc};

struct Client {
    handle: Handle,
    data: Arc<Mutex<usize>>,
}

impl Client {
    fn update_data(&mut self) {
        let data = Arc::clone(&self.data);
        self.handle.spawn(future::ok(()).and_then(move |_x| {
            *data.lock().unwrap() += 1;
            future::ok(())
        }));
    }
}

如果你的数据确实是一个 usize，你也可以使用 AtomicUsize 而不是 Mutex<usize>，但我个人认为这种方式同样难以操作。

- Peter Hall