Lifetime error with async closure

[yaiba]

Hi, I had a lifetime error when referencing data inside a async closure like below.
Please help. Thanks a lot.

use futures::future;
use futures::stream::{self, StreamExt};
use std::error::Error;

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let stream = stream::iter(1..=3);
    let stream2 = stream::iter(1..=3);

    let data = Data::new();

    future::join(
        stream.for_each(|_| inc1(&mut data)),
        stream2.collect::<Vec<_>>(),
    );

    Ok(())
}

async fn inc1(data: &mut Data) -> () {
    data.n += 1;
    future::ready(()).await
}

struct Data {
    n: u32,
}

impl Data {
    fn new() -> Data {
        Data { n: 0 }
    }
}

(Playground)

Errors:

   Compiling playground v0.0.1 (/playground)
error[E0495]: cannot infer an appropriate lifetime for borrow expression due to conflicting requirements
  --> src/main.rs:13:34
   |
13 |         stream.for_each(|_| inc1(&mut data)),
   |                                  ^^^^^^^^^
   |
note: first, the lifetime cannot outlive the lifetime `'_` as defined on the body at 13:25...
  --> src/main.rs:13:25
   |
13 |         stream.for_each(|_| inc1(&mut data)),
   |                         ^^^^^^^^^^^^^^^^^^^
note: ...so that closure can access `data`
  --> src/main.rs:13:34
   |
13 |         stream.for_each(|_| inc1(&mut data)),
   |                                  ^^^^^^^^^
note: but, the lifetime must be valid for the call at 12:5...
  --> src/main.rs:12:5
   |
12 | /     future::join(
13 | |         stream.for_each(|_| inc1(&mut data)),
14 | |         stream2.collect::<Vec<_>>(),
15 | |     );
   | |_____^
note: ...so type `futures_util::future::join::Join<futures_util::stream::stream::for_each::ForEach<futures_util::stream::iter::Iter<std::ops::RangeInclusive<i32>>, impl core::future::future::Future, [closure@src/main.rs:13:25: 13:44 data:&mut Data]>, futures_util::stream::stream::collect::Collect<futures_util::stream::iter::Iter<std::ops::RangeInclusive<i32>>, std::vec::Vec<i32>>>` of expression is valid during the expression
  --> src/main.rs:12:5
   |
12 | /     future::join(
13 | |         stream.for_each(|_| inc1(&mut data)),
14 | |         stream2.collect::<Vec<_>>(),
15 | |     );
   | |_____^

error: aborting due to previous error

For more information about this error, try `rustc --explain E0495`.
error: could not compile `playground`.

To learn more, run the command again with --verbose.

[jonh]

Use Arc<Mutex<.

Your code is like this non async.

    let mut data = Data::new();
    fn xxx1<'a>(_data: &'a mut Data) -> &'a() { &() }
    let mut _c = |_:i32| xxx1(&mut data);

which gives
error: captured variable cannot escape FnMut closure body

[alice]

There's no reason to use Arc or Mutex here. The culprit is for_each.

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let mut stream = stream::iter(1..=3);
    let stream2 = stream::iter(1..=3);

    let mut data = Data::new();

    future::join(
        async {
            while let Some(_) = stream.next().await {
                inc1(&mut data).await;
            }
        },
        stream2.collect::<Vec<_>>(),
    ).await;

    Ok(())
}

playground

[yaiba]

Thanks a lot Alice, it works now. BTW can you elaborate a bit on why it was error?

[alice]

Yes, it's basically because the signature of for_each does not allow the returned future to contain borrows of the item, but you are giving it a mutable reference to the item. The while loop does allow this kind of reference.

[yaiba]

Sorry I'm still new to Rust so please bear with me , which part in the declaration of for_each hints that phrase does not allow the returned future to contain borrows of the item?

[yaiba]

Hi @alice , can you give a bit more details on why for_each does not allow the returned future to contain borrows of the item? In my code let mut data = Data::new(); should live as long as and for_each so I'm not sure why holding a reference to data could cause compile error.

Also I'm confused about error message, it said

the lifetime cannot outlive the lifetime `'_` ...

which reference has that lifetime '_ ?

[alice]

Maybe I described the issue poorly. The issue is that to give inc1 mutable access to data, you must first move the mutable reference into the closure, and then on each call to the closure, create the future object associated with inc1 and give that future mutable access to data, and then return the future.

So the issue is really that by calling the closure multiple times and keeping the future objects around, you can get multiple mutable references to data in independent future objects, but mutable references require exclusive access, thus your error.

In fact, to see that for_each doesn't promise not to create multiple future objects that exist at the same time, take a look at for_each_concurrent which actually does this, and notice that ignoring the limit argument, they have the same signature:

fn for_each<Fut, F>(self, f: F) -> ForEach<Self, Fut, F>
where
    F: FnMut(Self::Item) -> Fut,
    Fut: Future<Output = ()>, 

fn for_each_concurrent<Fut, F>(self, f: F) -> ForEachConcurrent<Self, Fut, F>
where
    F: FnMut(Self::Item) -> Fut,
    Fut: Future<Output = ()>, 

There is not currently any way to change for_each to accept your closure without changes to the language. Note that had you only given immutable access to data, there would be no issue, as immutable references are allowed to overlap with each other.

As for the '_ lifetime, it is the lifetime of on &mut self in the definition of an FnMut closure:

fn call_mut(&'_ mut self, args: Args) -> Self::Output
//           ^ this lifetime

[yaiba]

Oh I see now.
A huge thanks to you @alice , much appreciated.
Compiler message was really confused to me, when it said the lifetime cannot outlive the lifetime '_ I spoke to myself "why is that?? outlive here is good because data has longer lifetime than the closure so closure can safely access data".
It's all good now anyway.