Comment puis-je attendre la fin d'un nombre indéterminé de threads Rust sans utiliser de poignées de suivi?

Question

Quelles sont les bonnes façons d'adapter cet exemple Barrier pour gérer deux différences:

1. le nombre d'éléments ne sont pas connus à l'avance (par exemple, dans le cas où la division d'un gros fichier en lignes)

2. sans suivre les poignées de fil (par exemple, sans utiliser le vecteur handles dans l'exemple ci-dessous). La motivation est que cela ajoute des frais généraux supplémentaires.

code Exemple:

use std::sync::{Arc, Barrier}; 
use std::thread; 

let mut handles = Vec::with_capacity(10); 
let barrier = Arc::new(Barrier::new(10)); 
for _ in 0..10 { 
    let c = barrier.clone(); 
    handles.push(thread::spawn(move|| { 
     // do some work 
     c.wait(); 
    })); 
} 
// Wait for other threads to finish. 
for handle in handles { 
    handle.join().unwrap(); 
} 

extrait de code

est adapté légèrement de la Barrier docs.

La première chose qui m'a traversé l'esprit serait (si possible) de muter la valeur interne du Barrier; cependant, l'API ne fournit pas d'accès modifiable à la propriété num_threads de la structure Barrier.

Une autre idée serait de ne pas utiliser le Barrier et d'écrire à la place une logique personnalisée avec AtomicUsize.

Je suis ouvert à l'apprentissage des façons les plus ergonomiques/idiomatiques de le faire dans Rust.

Answer 1

Vous pouvez utiliser spinlock sur atomic pour attendre que tous les threads se terminent. Bien sûr, au lieu d'utiliser l'atomique statique, vous pouvez passer Arc<AtomicUsize> dans chaque thread.

Ordering::SeqCst est probablement trop fort, mais la programmation concurrente est difficile, et je ne suis pas sûr de la façon dont cette commande peut être assouplie.

Alors que cela peut être fait de cette façon, le coût de création de threads va probablement éclipser la micro-optimisation comme celle-ci. En outre, cela vaut la peine de considérer qu'une attente chargée peut diminuer les performances d'un programme.

use std::panic; 
use std::sync::atomic::{AtomicUsize, Ordering, ATOMIC_USIZE_INIT}; 
use std::thread; 
use std::time::Duration; 

static GLOBAL_THREAD_COUNT: AtomicUsize = ATOMIC_USIZE_INIT; 

fn main() { 
    for i in 0..10 { 
     // mark that the thread is about to run 
     // we need to do it in the main thread to prevent spurious exits 
     GLOBAL_THREAD_COUNT.fetch_add(1, Ordering::SeqCst); 
     thread::spawn(move|| { 
      // We need to catch panics to reliably signal exit of a thread 
      let result = panic::catch_unwind(move || { 
       // do some work 
       println!("{}-th thread reporting", i+1); 
      }); 
      // process errors 
      match result { 
       _ => {} 
      } 
      // signal thread exit 
      GLOBAL_THREAD_COUNT.fetch_sub(1, Ordering::SeqCst); 
     }); 
    } 
    // Wait for other threads to finish. 
    while GLOBAL_THREAD_COUNT.load(Ordering::SeqCst) != 0 { 
     thread::sleep(Duration::from_millis(1)); 
    } 
} 

Playground link