Créez un avenir non statique avec Tokio 0.2

Dec 13 2020

J'ai une méthode asynchrone qui devrait exécuter certains contrats à terme en parallèle, et ne revenir qu'une fois tous les contrats terminés. Cependant, certaines données lui sont transmises par référence qui ne vivent pas aussi longtemps que 'static(elles seront supprimées à un moment donné dans la méthode principale). Conceptuellement, c'est similaire à ceci ( Playground ):

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

Maintenant, tokio veut que les contrats à terme passés spawnsoient valides pour la 'staticvie, car je pourrais laisser tomber la poignée sans que l'avenir ne s'arrête. Cela signifie que mon exemple ci-dessus produit ce message d'erreur:

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

Ma question est donc la suivante: comment créer des futurs qui ne sont valables que pour le contexte actuel et que je peux ensuite attendre qu'ils soient tous terminés?

(si cela est possible sur tokio 0.3 mais pas 0.2, je suis toujours intéressé, même si cela impliquerait beaucoup de dépendances git pour le moment)

Réponses

3 AliceRyhl Dec 14 2020 at 17:35

Il n'est pas possible de générer un non- 'staticfutur à partir de Rust asynchrone. En effet, toute fonction asynchrone peut être annulée à tout moment, il n'y a donc aucun moyen de garantir que l'appelant survit vraiment aux tâches générées.

Il est vrai qu'il existe différentes caisses qui permettent des apparitions étendues de tâches asynchrones, mais ces caisses ne peuvent pas être utilisées à partir du code asynchrone. Ce qu'ils ne permettent est de Spawn scope async tâches de non-async code. Cela ne viole pas le problème ci-dessus, car le code non asynchrone qui les a générés ne peut être annulé à aucun moment, car il n'est pas asynchrone.

En règle générale, il existe deux approches à cela:

  1. Créez une 'statictâche en utilisant Arcplutôt que des références ordinaires.
  2. Utilisez les primitives de concurrence de la caisse des contrats à terme au lieu de se reproduire.

Notez que cette réponse s'applique à la fois à Tokio 0.2.xet 0.3.x.


En règle générale, pour générer une tâche et une utilisation statiques Arc, vous devez être propriétaire des valeurs en question. Cela signifie que puisque votre fonction a pris l'argument par référence, vous ne pouvez pas utiliser cette technique sans cloner les données.

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

Notez que si vous avez une référence mutable aux données, et que les données ne sont Sized, c'est-à-dire pas une tranche, il est possible d'en prendre temporairement possession.

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

Une autre option consiste à utiliser les primitives de concurrence de la caisse à terme. Ceux-ci ont l'avantage de travailler avec des non- 'staticdonnées, mais l'inconvénient que les tâches ne pourront pas s'exécuter sur plusieurs threads en même temps.

Pour de nombreux flux de travail, cela convient parfaitement, car le code asynchrone devrait de toute façon passer la plupart de son temps à attendre les E / S.

Une approche consiste à utiliser FuturesUnordered. Il s'agit d'une collection spéciale qui peut stocker de nombreux futurs différents, et elle a une nextfonction qui les exécute tous simultanément et revient une fois le premier terminé. (La nextfonction n'est disponible que lorsqu'elle StreamExtest importée)

Vous pouvez l'utiliser comme ceci:

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

Remarque: Le FuturesUnordereddoit être défini après la valeur partagée. Sinon, vous obtiendrez une erreur d'emprunt causée par leur chute dans le mauvais ordre.


Une autre approche consiste à utiliser un fichier Stream. Avec les flux, vous pouvez utiliser buffer_unordered. Il s'agit d'un utilitaire qui utilise en FuturesUnorderedinterne.

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

Notez que dans les deux cas, l'importation StreamExtest importante car elle fournit diverses méthodes qui ne sont pas disponibles sur les flux sans importer le trait d'extension.