How do I borrow from inside async task inside a loop?

[bobajeff]

I'm having trouble with this async task. The idea here is to run a bunch of async tasks that may or may not end. So I'm trying to join_all() in order to await on all of them to stop so I can borrow from my_vecs here.

use futures::future::{join_all};

#[async_std::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    
    let mut futures: Vec<_> = vec![];

    let my_vecs: Vec<i32> = vec![1,2,3,4];

    for a_vec in my_vecs.iter()
    {
        let handel = async_std::task::spawn(async move {
            let b = a_vec + 1;
        });
        futures.push(handel);
    }

    join_all(futures).await;
    Ok(())
}

But as you might guess that's a no go:

  --> src/main.rs:10:18
   |
10 |     for a_vec in my_vecs.iter()
   |                  ^^^^^^^-------
   |                  |
   |                  borrowed value does not live long enough
   |                  argument requires that `my_vecs` is borrowed for `'static`
...
20 | }
   | - `my_vecs` dropped here while still borrowed

error: aborting due to previous error

[alice]

I have answered a very similar question here:

https://stackoverflow.com/questions/65269738/spawn-non-static-future-with-tokio-0-2/65287449#65287449

[bobajeff]

Thanks for your help. I wonder if you could help me understand whats happening here? Out of all of those options. The only one that would compile for me was the second variation using Arc which I applied like so:

use std::sync::Arc;

#[async_std::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    
    let mut futures: Vec<_> = vec![];
    
    let mut my_vecs: Vec<i32> = vec![1,2,3,4];
    let vec = std::mem::take(&mut my_vecs);

    let shared: Arc<Vec<i32>> = Arc::new(vec);

    for i in 0..shared.len()
    {
        let another_vec = shared.clone();
        let handel = async_std::task::spawn(async move {
            let b = another_vec[i] + 1;
        });
        futures.push(handel);
    }

    for future in futures {
        future.await;
    }
    Arc::try_unwrap(shared).unwrap();
    Ok(())
}

The this will compile and run. However this:

#[async_std::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    
    let mut futures: Vec<_> = vec![];

    let my_vecs: Vec<i32> = vec![1,2,3,4];
    let shared: Arc<Vec<i32>> = Arc::from(my_vecs);

    for i in 0..my_vecs.len()
    {
        let another_vec = shared.clone();
        let handel = async_std::task::spawn(async move {
            let b = another_vec[i] + 1;
        });
        futures.push(handel);
    }

    for future in futures {
        future.await;
    }
    Ok(())
}

would again complain of borrowing:

error[E0382]: borrow of moved value: `my_vecs`
  --> src/arc_method.rs:12:17
   |
9  |     let my_vecs: Vec<i32> = vec![1,2,3,4];
   |         ------- move occurs because `my_vecs` has type `Vec<i32>`, which does not implement the `Copy` trait
10 |     let shared: Arc<Vec<i32>> = Arc::from(my_vecs);
   |                                           ------- value moved here
11 | 
12 |     for i in 0..my_vecs.len()
   |                 ^^^^^^^ value borrowed here after move

and neither of the ones using concurrency primitives would work at all. These are the ones I've tried:

use futures::stream::{futures_unordered::FuturesUnordered, StreamExt};

async fn handlethis(a_vec: &i32)
{
    let b = a_vec + 1;
}

#[async_std::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    
    let mut futures = FuturesUnordered::new();

    let mut my_vecs: Vec<i32> = vec![1,2,3,4];
    let et_vecs = &mut my_vecs;

    for a_vec in et_vecs
    {
        let future = handlethis(a_vec);
        futures.push(future);
    }

    while let Some(()) = futures.next().await {}
    Ok(())
}

gives:

error[E0597]: `my_vecs` does not live long enough
  --> src/futures_method.rs:14:19
   |
14 |     let et_vecs = &mut my_vecs;
   |                   ^^^^^^^^^^^^ borrowed value does not live long enough
...
24 | }
   | -
   | |
   | `my_vecs` dropped here while still borrowed
   | borrow might be used here, when `futures` is dropped and runs the `Drop` code for type `FuturesUnordered`
   |
   = note: values in a scope are dropped in the opposite order they are defined

and

use futures::stream::StreamExt;

async fn handlethis(a_vec: &i32)
{
    let b = a_vec + 1;
}

#[async_std::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
   
    let mut my_vecs: Vec<i32> = vec![1,2,3,4];
    let et_vecs = &mut my_vecs;
    futures::stream::iter(et_vecs)
    .map(|a_vec| handlethis(a_vec))
    .buffer_unordered(10)
    .for_each(|()| {})
    .await;
    Ok(())
}

gives:

error[E0277]: `()` is not a future
  --> src/futures_method_two.rs:16:6
   |
16 |     .for_each(|()| {})
   |      ^^^^^^^^ `()` is not a future
   |
   = help: the trait `futures::Future` is not implemented for `()`

error[E0277]: `()` is not a future
  --> src/futures_method_two.rs:13:5
   |
13 | /     futures::stream::iter(et_vecs)
14 | |     .map(|a_vec| handlethis(a_vec))
15 | |     .buffer_unordered(10)
16 | |     .for_each(|()| {})
17 | |     .await;
   | |__________^ `()` is not a future
   |
   = help: the trait `futures::Future` is not implemented for `()`
   = note: required because of the requirements on the impl of `futures::Future` for `ForEach<BufferUnordered<futures::stream::Map<futures::stream::Iter<std::slice::IterMut<'_, i32>>, [closure@src/futures_method_two.rs:14:10: 14:35]>>, (), [closure@src/futures_method_two.rs:16:15: 16:22]>`
   = note: required by `futures::Future::poll`

[alice]

When you do Arc::from(my_vecs), this transfers ownership of my_vecs into the Arc, so you wont be able to use my_vecs anymore. You would need to do shared.len().

The second error has to do with order of definition. If you define the FuturesUnordered after the vector, it will work.

In the third error, you need to put an async block inside the for_each, instead of just {}. So it would be .for_each(|()| async {}). I see that the SO answer is wrong on this, I'll correct it.

[bobajeff]

Once again. Thanks for your help.

Could you explain what was happening with the order of operations? Why did I need to define futures after my_vecs?

[alice]

It is because variables are always destroyed in opposite order as creation, but futures contains references into my_vecs, so if my_vecs is dropped before futures, then futures will contain dangling references after my_vecs is dropped.

[bobajeff]

Yeah, that makes sense. Thanks again.