Stwórz niestatyczną przyszłość z Tokio 0.2

Dec 13 2020

Mam metodę asynchroniczną, która powinna wykonywać niektóre futures równolegle i powracać dopiero po zakończeniu wszystkich futures. Jednak przez odniesienie przekazywane są pewne dane, które nie istnieją tak długo, jak 'static(zostaną usunięte w pewnym momencie w głównej metodzie). Koncepcyjnie jest podobny do tego ( Plac zabaw ):

async fn do_sth(with: &u64) {
    delay_for(Duration::new(*with, 0)).await;
    println!("{}", with);
}

async fn parallel_stuff(array: &[u64]) {
    let mut tasks: Vec<JoinHandle<()>> = Vec::new();
    for i in array {
        let task = spawn(do_sth(i));
        tasks.push(task);
    }
    for task in tasks {
        task.await;
    }
}

#[tokio::main]
async fn main() {
    parallel_stuff(&[3, 1, 4, 2]);
}

Teraz tokio chce, aby futures, które są przekazywane, spawnbyły ważne na 'staticcałe życie, ponieważ mógłbym upuścić uchwyt bez zatrzymywania się w przyszłości. Oznacza to, że mój przykład powyżej generuje ten komunikat o błędzie:

error[E0759]: `array` has an anonymous lifetime `'_` but it needs to satisfy a `'static` lifetime requirement
  --> src/main.rs:12:25
   |
12 | async fn parallel_stuff(array: &[u64]) {
   |                         ^^^^^  ------ this data with an anonymous lifetime `'_`...
   |                         |
   |                         ...is captured here...
...
15 |         let task = spawn(do_sth(i));
   |                    ----- ...and is required to live as long as `'static` here

Więc moje pytanie brzmi: jak mogę stworzyć futures, które są ważne tylko dla obecnego kontekstu, a następnie mogę poczekać, aż wszystkie z nich zostaną ukończone?

(jeśli jest to możliwe w tokio 0.3, ale nie 0.2, nadal jestem zainteresowany, chociaż na razie wymagałoby to wielu zależności od git)

Odpowiedzi

3 AliceRyhl Dec 14 2020 at 17:35

Nie jest możliwe zrobienie czegoś innego niż 'staticprzyszłość z asynchronicznego Rust. Dzieje się tak, ponieważ każda funkcja asynchroniczna może zostać anulowana w dowolnym momencie, więc nie ma możliwości zagwarantowania, że ​​wywołujący naprawdę przeżyje zrodzone zadania.

Prawdą jest, że istnieją różne skrzynki, które pozwalają na tworzenie zakresu zadań asynchronicznych, ale tych skrzynek nie można używać z kodu asynchronicznego. Co oni nie pozwalają na tarło jest zawężona asynchroniczny zadań z non-asynchronicznym kodzie. Nie narusza to powyższego problemu, ponieważ kod inny niż asynchroniczny, który je wywołał, nie może zostać anulowany w dowolnym momencie, ponieważ nie jest on asynchroniczny.

Generalnie istnieją dwa podejścia do tego:

  1. Stwórz 'staticzadanie, używając Arczamiast zwykłych odniesień.
  2. Użyj prymitywów współbieżności ze skrzynki futures zamiast odradzania się.

Zwróć uwagę, że ta odpowiedź dotyczy zarówno Tokio, jak 0.2.xi 0.3.x.


Ogólnie rzecz biorąc, aby utworzyć zadanie statyczne i użyć go Arc, musisz mieć prawo własności do danych wartości. Oznacza to, że ponieważ twoja funkcja przyjęła argument przez odniesienie, nie możesz użyć tej techniki bez klonowania danych.

async fn do_sth(with: Arc<[u64]>, idx: usize) {
    delay_for(Duration::new(with[idx], 0)).await;
    println!("{}", with[idx]);
}

async fn parallel_stuff(array: &[u64]) {
    // Make a clone of the data so we can shared it across tasks.
    let shared: Arc<[u64]> = Arc::from(array);
    
    let mut tasks: Vec<JoinHandle<()>> = Vec::new();
    for i in 0..array.len() {
        // Cloning an Arc does not clone the data.
        let shared_clone = shared.clone();
        let task = spawn(do_sth(shared_clone, i));
        tasks.push(task);
    }
    for task in tasks {
        task.await;
    }
}

Zwróć uwagę, że jeśli masz zmienne odniesienie do danych, a dane nie są Sized, tj. Wycinkiem, możliwe jest tymczasowe przejęcie ich na własność.

async fn do_sth(with: Arc<Vec<u64>>, idx: usize) {
    delay_for(Duration::new(with[idx], 0)).await;
    println!("{}", with[idx]);
}

async fn parallel_stuff(array: &mut Vec<u64>) {
    // Swap the array with an empty one to temporarily take ownership.
    let vec = std::mem::take(array);
    let shared = Arc::new(vec);
    
    let mut tasks: Vec<JoinHandle<()>> = Vec::new();
    for i in 0..array.len() {
        // Cloning an Arc does not clone the data.
        let shared_clone = shared.clone();
        let task = spawn(do_sth(shared_clone, i));
        tasks.push(task);
    }
    for task in tasks {
        task.await;
    }
    
    // Put back the vector where we took it from.
    // This works because there is only one Arc left.
    *array = Arc::try_unwrap(shared).unwrap();
}

Inną opcją jest użycie prymitywów współbieżności ze skrzynki futures. Mają one tę zaletę, że pracują z 'staticdanymi niebędącymi danymi, ale mają tę wadę, że zadania nie będą mogły być wykonywane w wielu wątkach jednocześnie.

W przypadku wielu przepływów pracy jest to całkowicie w porządku, ponieważ kod asynchroniczny i tak powinien spędzać większość czasu na czekaniu na IO.

Jednym podejściem jest użycie FuturesUnordered. Jest to specjalna kolekcja, która może przechowywać wiele różnych przyszłości i ma nextfunkcję, która uruchamia je wszystkie jednocześnie i zwraca po zakończeniu pierwszej z nich. ( nextFunkcja jest dostępna tylko po StreamExtzaimportowaniu)

Możesz go używać w ten sposób:

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

async fn do_sth(with: &u64) {
    delay_for(Duration::new(*with, 0)).await;
    println!("{}", with);
}

async fn parallel_stuff(array: &[u64]) {
    let mut tasks = FuturesUnordered::new();
    for i in array {
        let task = do_sth(i);
        tasks.push(task);
    }
    // This loop runs everything concurrently, and waits until they have
    // all finished.
    while let Some(()) = tasks.next().await { }
}

Uwaga:FuturesUnordered muszą być zdefiniowane po udostępnionym wartości. W przeciwnym razie pojawi się błąd pożyczki spowodowany upuszczeniem ich w złej kolejności.


Innym podejściem jest użycie pliku Stream. W przypadku strumieni możesz użyć buffer_unordered. Jest to narzędzie używane FuturesUnorderedwewnętrznie.

use futures::stream::StreamExt;

async fn do_sth(with: &u64) {
    delay_for(Duration::new(*with, 0)).await;
    println!("{}", with);
}

async fn parallel_stuff(array: &[u64]) {
    // Create a stream going through the array.
    futures::stream::iter(array)
    // For each item in the stream, create a future.
        .map(|i| do_sth(i))
    // Run at most 10 of the futures concurrently.
        .buffer_unordered(10)
    // Since Streams are lazy, we must use for_each or collect to run them.
    // Here we use for_each and do nothing with the return value from do_sth.
        .for_each(|()| async {})
        .await;
}

Należy pamiętać, że w obu przypadkach importowanie StreamExtjest ważne, ponieważ zapewnia różne metody, które nie są dostępne w strumieniach bez importowania cechy rozszerzenia.