rust - 调用tokio::spawn时如何设置 `self`的适当生存期？

Question

我正在研究一种实现Stream特征的类型。它基本上需要另外两个流，从它们中获取next()值，然后根据某些条件返回一个next()值本身。我想到了这样的代码设置:

impl Stream for HeartbeatStream {
  type Item = char;

  fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
    // ...
    // Dealing with heartbeat and no heartbeat
    match &mut self.next_heartbeat {
      Some(hb) => {
        Pin::new(hb).poll(cx);
        tokio::spawn(async move {
          tokio::select! {
            Some(char) = stream1.next(), if self.cooldown.map_or(true, |cd| since_last_signal > cd) => {
              self.reset_last_signal_next_heartbeat();
              return Poll::Ready(Some(char))
            },
            Some(char) = stream2.next(), if self.cooldown.map_or(true, |cd| since_last_signal > cd) => {
              self.reset_last_signal_next_heartbeat();
              return Poll::Ready(Some(char))
            },
            _ = hb => {
              self.reset_last_signal_next_heartbeat();
              return Poll::Ready(Some(self.char))
            },
            else => Poll::Ready(None),
          }
        });
      },
      None => {
        ...
      }
    }

    Poll::Pending
  }
}

Full code is shown here。
当我编译它时，我得到以下错误信息:

error[E0495]: cannot infer an appropriate lifetime for lifetime parameter in function call due to conflicting requirements
  --> src/heartbeat.rs:59:14
   |
59 |         match &mut self.next_heartbeat {
   |                    ^^^^
   |
note: first, the lifetime cannot outlive the anonymous lifetime #1 defined on the method body at 52:2...
  --> src/heartbeat.rs:52:2
   |
52 |       fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
   |  _____^
53 | |         let mut stream1 = self.streams[0];
54 | |         let mut stream2 = self.streams[1];
55 | |         let since_last_signal = Instant::now().saturating_duration_since(
...  |
97 | |         Poll::Pending
98 | |     }
   | |_____^
note: ...so that the reference type `&mut std::pin::Pin<&mut heartbeat::HeartbeatStream>` does not outlive the data it points at
  --> src/heartbeat.rs:59:14
   |
59 |         match &mut self.next_heartbeat {
   |                    ^^^^
   = note: but, the lifetime must be valid for the static lifetime...
note: ...so that the type `impl std::future::Future` will meet its required lifetime bounds
  --> src/heartbeat.rs:62:5
   |
62 |                 tokio::spawn(async move {
   |                 ^^^^^^^^^^^^

error[E0495]: cannot infer an appropriate lifetime due to conflicting requirements
  --> src/heartbeat.rs:62:29
   |
62 |                   tokio::spawn(async move {
   |  _________________________________________^
63 | |                     tokio::select! {
64 | |                         Some(char) = stream1.next(), if self.cooldown.map_or(true, |cd| since_last_signal > cd) => {
65 | |                             self.reset_last_signal_next_heartbeat();
...  |
77 | |                     }
78 | |                 });
   | |_________________^
   |
note: first, the lifetime cannot outlive the anonymous lifetime #1 defined on the method body at 52:2...
  --> src/heartbeat.rs:52:2
   |
52 |       fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
   |  _____^
53 | |         let mut stream1 = self.streams[0];
54 | |         let mut stream2 = self.streams[1];
55 | |         let since_last_signal = Instant::now().saturating_duration_since(
...  |
97 | |         Poll::Pending
98 | |     }
   | |_____^
note: ...so that the types are compatible
  --> src/heartbeat.rs:62:29
   |
62 |                   tokio::spawn(async move {
   |  _________________________________________^
63 | |                     tokio::select! {
64 | |                         Some(char) = stream1.next(), if self.cooldown.map_or(true, |cd| since_last_signal > cd) => {
65 | |                             self.reset_last_signal_next_heartbeat();
...  |
77 | |                     }
78 | |                 });
   | |_________________^
   = note: expected `std::pin::Pin<&mut heartbeat::HeartbeatStream>`
              found `std::pin::Pin<&mut heartbeat::HeartbeatStream>`
   = note: but, the lifetime must be valid for the static lifetime...
note: ...so that the type `impl std::future::Future` will meet its required lifetime bounds
  --> src/heartbeat.rs:62:5
   |
62 |                 tokio::spawn(async move {
   |                 ^^^^^^^^^^^^

Full error code is shown here。
我从tokio文档中看到，通常这可以通过 异​​步移动并将变量移到 tokio::spawn()外部以归入异步代码内部来解决。但是在这种情况下，它提示 self对象。
我该怎么解决？
谢谢你的洒脱🙏

Answer 1

Rust编译器在删除tokio::spawn()之前不知道self是否将加入。即使tokio::spawn()返回了可以传递给JoinHandle且从未连接的std::mem::forget()，也使得闭包可以访问已删除的self引用(这是不安全的)。这就是为什么编译器强制执行'static生存期的原因。通常，“Rust的安全保证不包括析构函数将始终运行的保证”(source)。我建议您直接调用next_pool()而不是next()。