Bagaimana saya bisa menelurkan metode asynchronous dalam satu lingkaran?

Aug 16 2020

Saya memiliki vektor objek yang memiliki resolve()metode yang digunakan reqwestuntuk meminta API web eksternal. Setelah saya memanggil resolve()metode pada setiap objek, saya ingin mencetak hasil dari setiap permintaan.

Inilah kode setengah asinkron saya yang mengkompilasi dan berfungsi (tetapi tidak benar-benar asinkron):

for mut item in items {
    item.resolve().await;

    item.print_result();
}

Saya telah mencoba menggunakan tokio::join!untuk menelurkan semua panggilan asinkron dan menunggu hingga selesai, tetapi saya mungkin melakukan sesuatu yang salah:

tokio::join!(items.iter_mut().for_each(|item| item.resolve()));

Inilah kesalahan yang saya dapatkan:

error[E0308]: mismatched types
  --> src\main.rs:25:51
   |
25 |     tokio::join!(items.iter_mut().for_each(|item| item.resolve()));
   |                                                   ^^^^^^^^^^^^^^ expected `()`, found opaque type
   | 
  ::: src\redirect_definition.rs:32:37
   |
32 |     pub async fn resolve(&mut self) {
   |                                     - the `Output` of this `async fn`'s found opaque type
   |
   = note: expected unit type `()`
            found opaque type `impl std::future::Future`

Bagaimana cara memanggil resolve()metode untuk semua contoh sekaligus?


Kode ini mencerminkan jawabannya - sekarang saya berurusan dengan kesalahan pemeriksa peminjam yang tidak terlalu saya mengerti - haruskah saya memberi keterangan pada beberapa variabel saya 'static?

let mut items = get_from_csv(path);

let tasks: Vec<_> = items
    .iter_mut()
    .map(|item| tokio::spawn(item.resolve()))
    .collect();

for task in tasks {
    task.await;
}

for item in items {
    item.print_result();
}
error[E0597]: `items` does not live long enough
  --> src\main.rs:18:25
   |
18 |       let tasks: Vec<_> = items
   |                           -^^^^
   |                           |
   |  _________________________borrowed value does not live long enough
   | |
19 | |         .iter_mut()
   | |___________________- argument requires that `items` is borrowed for `'static`
...
31 |   }
   |   - `items` dropped here while still borrowed

error[E0505]: cannot move out of `items` because it is borrowed
  --> src\main.rs:27:17
   |
18 |       let tasks: Vec<_> = items
   |                           -----
   |                           |
   |  _________________________borrow of `items` occurs here
   | |
19 | |         .iter_mut()
   | |___________________- argument requires that `items` is borrowed for `'static`
...
27 |       for item in items {
   |                   ^^^^^ move out of `items` occurs here

Jawaban

2 user4815162342 Aug 16 2020 at 20:29

Karena Anda ingin menunggu di futures secara paralel, Anda dapat menelurkan mereka ke tugas individu yang berjalan secara paralel. Karena mereka berjalan secara independen satu sama lain dan dari utas yang melahirkannya, Anda dapat menunggu pegangannya dalam urutan apa pun.

Idealnya Anda akan menulis sesuatu seperti ini:

// spawn tasks that run in parallel
let tasks: Vec<_> = items
    .iter_mut()
    .map(|item| tokio::spawn(item.resolve()))
    .collect();
// now await them to get the resolve's to complete
for task in tasks {
    task.await.unwrap();
}
// and we're done
for item in &items {
    item.print_result();
}

Tapi ini akan ditolak oleh pemeriksa peminjam karena masa depan dikembalikan dengan item.resolve()memegang referensi pinjaman item. Referensi diteruskan tokio::spawn()yang menyerahkannya ke utas lain, dan kompilator tidak dapat membuktikan bahwa itu itemakan hidup lebih lama dari utas itu. (Masalah yang sama terjadi saat Anda ingin mengirim referensi ke data lokal ke utas .)

Ada beberapa solusi yang mungkin untuk ini; yang menurut saya paling elegan adalah memindahkan item ke penutupan asinkron yang diteruskan tokio::spawn(), dan minta tugas mengembalikannya kepada Anda setelah selesai. Pada dasarnya Anda menggunakan itemsvektor untuk membuat tugas dan segera menyusunnya kembali dari hasil yang ditunggu:

// note the use of `into_iter()` to consume `items`
let tasks: Vec<_> = items
    .into_iter()
    .map(|mut item| {
        tokio::spawn(async {
            item.resolve().await;
            item
        })
    })
    .collect();
// await the tasks for resolve's to complete and give back our items
let mut items = vec![];
for task in tasks {
    items.push(task.await.unwrap());
}
// verify that we've got the results
for item in &items {
    item.print_result();
}

Kode yang dapat dijalankan di taman bermain .

Perhatikan bahwa futurespeti berisi join_allfungsi yang mirip dengan yang Anda butuhkan, kecuali itu mengumpulkan setiap masa depan tanpa memastikan bahwa mereka berjalan secara paralel. Kita bisa menulis generik join_parallelyang menggunakan join_all, tetapi juga menggunakan tokio::spawnuntuk mendapatkan eksekusi paralel:

async fn join_parallel<T: Send + 'static>(
    futs: impl IntoIterator<Item = impl Future<Output = T> + Send + 'static>,
) -> Vec<T> {
    let tasks: Vec<_> = futs.into_iter().map(tokio::spawn).collect();
    // unwrap the Result because it is introduced by tokio::spawn()
    // and isn't something our caller can handle
    futures::future::join_all(tasks)
        .await
        .into_iter()
        .map(Result::unwrap)
        .collect()
}

Dengan menggunakan fungsi ini, kode yang diperlukan untuk menjawab pertanyaan tersebut bermuara pada:

let items = join_parallel(items.into_iter().map(|mut item| async {
    item.resolve().await;
    item
})).await;
for item in &items {
    item.print_result();
}

Sekali lagi, kode yang dapat dijalankan di taman bermain .