Spawn nicht statische Zukunft mit Tokio 0.2

Dec 13 2020

Ich habe eine asynchrone Methode, die einige Futures parallel ausführen und erst zurückkehren soll, wenn alle Futures abgeschlossen sind. Es werden jedoch einige Daten als Referenz übergeben, die nicht so lange leben wie 'static(sie werden irgendwann in der Hauptmethode gelöscht). Konzeptionell ist es ähnlich ( Spielplatz ):

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]);
}

Jetzt möchte Tokio, dass Futures, die übergeben spawnwerden, 'staticlebenslang gültig sind , weil ich den Griff fallen lassen könnte, ohne dass die Zukunft stehen bleibt. Das bedeutet, dass mein Beispiel oben diese Fehlermeldung erzeugt:

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

Meine Frage lautet also: Wie kann ich Futures erzeugen, die nur für den aktuellen Kontext gültig sind und die ich dann abwarten kann, bis alle abgeschlossen sind?

(Wenn dies auf Tokio 0.3 möglich ist, aber nicht auf 0.2, bin ich immer noch interessiert, obwohl dies vorerst viele Git-Abhängigkeiten mit sich bringen würde.)

Antworten

3 AliceRyhl Dec 14 2020 at 17:35

Es ist nicht möglich, 'staticaus Async Rust eine Nicht- Zukunft zu erzeugen. Dies liegt daran, dass jede asynchrone Funktion jederzeit abgebrochen werden kann. Daher kann nicht garantiert werden, dass der Anrufer die erzeugten Aufgaben wirklich überlebt.

Es ist wahr, dass es verschiedene Kisten gibt, die das Auftreten von asynchronen Aufgaben ermöglichen, aber diese Kisten können nicht aus asynchronem Code verwendet werden. Was sie tun lassen ist async Aufgaben aus , um laichen scoped nicht-Asynchron - Code. Dies verstößt nicht gegen das oben genannte Problem, da der nicht asynchrone Code, der sie erzeugt hat, zu keinem Zeitpunkt abgebrochen werden kann, da er nicht asynchron ist.

Generell gibt es hierfür zwei Ansätze:

  1. Erstellen Sie eine 'staticAufgabe, indem Sie keine Arcgewöhnlichen Referenzen verwenden.
  2. Verwenden Sie die Parallelitätsprimitive aus der Futures-Kiste, anstatt zu laichen.

Beachten Sie, dass diese Antwort sowohl für Tokio 0.2.xals auch für Tokio gilt 0.3.x.


Um eine statische Aufgabe und Verwendung zu Arcerzeugen, müssen Sie im Allgemeinen Eigentümer der betreffenden Werte sein. Dies bedeutet, dass Sie diese Technik nicht verwenden können, ohne die Daten zu klonen, da Ihre Funktion das Argument als Referenz verwendet hat.

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;
    }
}

Beachten Sie Sized, dass es möglich ist, vorübergehend das Eigentum daran zu übernehmen , wenn Sie einen veränderlichen Verweis auf die Daten haben und die Daten , dh kein Slice, sind.

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();
}

Eine andere Möglichkeit besteht darin, die Parallelitätsprimitive aus der Futures-Kiste zu verwenden. Diese haben den Vorteil, mit Nicht- 'staticDaten zu arbeiten, aber den Nachteil, dass die Aufgaben nicht gleichzeitig auf mehreren Threads ausgeführt werden können.

Für viele Workflows ist dies vollkommen in Ordnung, da asynchroner Code ohnehin die meiste Zeit auf E / A warten sollte.

Ein Ansatz ist zu verwenden FuturesUnordered. Dies ist eine spezielle Sammlung, in der viele verschiedene Futures gespeichert werden können. Sie verfügt über eine nextFunktion, die alle gleichzeitig ausführt und nach Abschluss der ersten zurückgegeben wird. (Die nextFunktion ist nur verfügbar, wenn sie StreamExtimportiert wird.)

Sie können es so verwenden:

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 { }
}

Hinweis: Das FuturesUnorderedmuss nach dem gemeinsamen Wert definiert werden. Andernfalls erhalten Sie einen Ausleihfehler, der dadurch verursacht wird, dass sie in der falschen Reihenfolge abgelegt werden.


Ein anderer Ansatz ist die Verwendung von a Stream. Mit Streams können Sie verwenden buffer_unordered. Dies ist ein Dienstprogramm, das FuturesUnorderedintern verwendet wird.

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;
}

Beachten Sie, dass in beiden Fällen der Import StreamExtwichtig ist, da er verschiedene Methoden bietet, die in Streams ohne Import des Erweiterungsmerkmals nicht verfügbar sind.