How to set the proper lifetime of `self` when calling tokio::spawn?

Viewed 571

I am working on a type implementing Stream trait. It basically takes two other streams, get the next() values from them, and then return a next() value itself based on some conditions. I get to a point of having the code setup like this:

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.

When I compile this, I get error message of:

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.

I see from tokio doc that usually this can be solved by async move and move the variable outside of tokio::spawn() to be owned inside the async code. But in this case, it is complaining about the self object.

What can I do to solve it?

Thank you for shedding any light

1 Answers

The Rust compiler does not know whether tokio::spawn() will join before self is dropped. Even if tokio::spawn() returns a JoinHandle that can be passed to a std::mem::forget() and never be joined making the closure have access to a dropped reference of self (which is unsafe). That is why the compiler enforces a 'static lifetime. In general "Rust's safety guarantees do not include a guarantee that destructors will always run" (source). I would suggest you to call directly poll_next() instead of next().

Related