Come posso inviare una funzione ad un altro thread?

Question

Sto tentando di scrivere un semplice test runner dell'unità per il mio progetto Rust. Ho creato un tratto TestFixture che implementerà le mie strutture di prova, simili all'ereditarietà della classe di base del test unitario in altri framework di test. Il tratto è abbastanza semplice. Questo è il mio dispositivo di prova

pub trait TestFixture { 
    fn setup(&mut self) ->() {} 
    fn teardown(&mut self) ->() {} 
    fn before_each(&mut self) ->() {} 
    fn after_each(&mut self) ->() {} 
    fn tests(&mut self) -> Vec<Box<Fn(&mut Self)>> 
     where Self: Sized { 
     Vec::new() 
    } 
} 

La mia funzione di test di funzionamento è il seguente

pub fn test_fixture_runner<T: TestFixture>(fixture: &mut T) { 
    fixture.setup(); 

    let _r = fixture.tests().iter().map(|t| { 
     let handle = thread::spawn(move || { 
      fixture.before_each(); 
      t(fixture); 
      fixture.after_each(); 
     }); 

     if let Err(_) = handle.join() { 
      println!("Test failed!") 
     } 
    }); 

    fixture.teardown(); 
} 

ottengo l'errore

src/tests.rs:73:22: 73:35 error: the trait `core::marker::Send` is not implemented for the type `T` [E0277] 
src/tests.rs:73   let handle = thread::spawn(move || { 
            ^~~~~~~~~~~~~ 
note: in expansion of closure expansion 
src/tests.rs:69:41: 84:6 note: expansion site 
src/tests.rs:73:22: 73:35 note: `T` cannot be sent between threads safely 
src/tests.rs:73   let handle = thread::spawn(move || { 
            ^~~~~~~~~~~~~ 
note: in expansion of closure expansion 
src/tests.rs:69:41: 84:6 note: expansion site 
src/tests.rs:73:22: 73:35 error: the trait `core::marker::Sync` is not implemented for the type `for<'r> core::ops::Fn(&'r mut T)` [E0277] 
src/tests.rs:73   let handle = thread::spawn(move || { 
            ^~~~~~~~~~~~~ 
note: in expansion of closure expansion 
src/tests.rs:69:41: 84:6 note: expansion site 
src/tests.rs:73:22: 73:35 note: `for<'r> core::ops::Fn(&'r mut T)` cannot be shared between threads safely 
src/tests.rs:73   let handle = thread::spawn(move || { 
            ^~~~~~~~~~~~~ 
note: in expansion of closure expansion 

Ho provato ad aggiungere Arcs intorno ai tipi di essere inviato al filo , niente dado, stesso errore. dispositivo di prova

pub fn test_fixture_runner<T: TestFixture>(fixture: &mut T) { 
    fixture.setup(); 

    let fix_arc = Arc::new(Mutex::new(fixture)); 
    let _r = fixture.tests().iter().map(|t| { 
     let test_arc = Arc::new(Mutex::new(t)); 
     let fix_arc_clone = fix_arc.clone(); 
     let test_arc_clone = test_arc.clone(); 
     let handle = thread::spawn(move || { 
      let thread_test = test_arc_clone.lock().unwrap(); 
      let thread_fix = fix_arc_clone.lock().unwrap(); 
      (*thread_fix).before_each(); 
      (*thread_test)(*thread_fix); 
      (*thread_fix).after_each(); 
     }); 

     if let Err(_) = handle.join() { 
      println!("Test failed!") 
     } 
    }); 

    fixture.teardown(); 
} 

Un campione sarebbe qualcosa di simile

struct BuiltinTests { 
    pwd: PathBuf 
} 

impl TestFixture for BuiltinTests { 
    fn setup(&mut self) { 
     let mut pwd = env::temp_dir(); 
     pwd.push("pwd"); 

     fs::create_dir(&pwd); 
     self.pwd = pwd; 
    } 

    fn teardown(&mut self) { 
     fs::remove_dir(&self.pwd); 
    } 

    fn tests(&mut self) -> Vec<Box<Fn(&mut BuiltinTests)>> { 
     vec![Box::new(BuiltinTests::cd_with_no_args)] 
    } 
} 

impl BuiltinTests { 
    fn new() -> BuiltinTests { 
     BuiltinTests { 
      pwd: PathBuf::new() 
     } 
    } 
} 

fn cd_with_no_args(&mut self) { 
    let home = String::from("/"); 
    env::set_var("HOME", &home); 

    let mut cd = Cd::new(); 
    cd.run(&[]); 

    assert_eq!(env::var("PWD"), Ok(home)); 
} 

#[test] 
fn cd_tests() { 
    let mut builtin_tests = BuiltinTests::new(); 
    test_fixture_runner(&mut builtin_tests); 
} 

Tutta la mia intenzione di usare le discussioni è isolamento dal test runner. Se un test fallisce un'affermazione provoca un panico che uccide il corridore. Grazie per qualsiasi intuizione, sono disposto a cambiare il mio design se questo risolverà il problema di panico.

Answer 1

Ci sono diversi problemi con il tuo codice, ti mostrerò come correggerli uno per uno.

Il primo problema è che si sta utilizzando map() per iterare su un iteratore. Non funzionerà correttamente perché map() è pigro - a meno che non si consumi l'iteratore, la chiusura passata non verrà eseguita. Il modo corretto è quello di utilizzare for ciclo:

for t in fixture().tests().iter() { 

In secondo luogo, si sta iterando il vettore delle chiusure per riferimento:

fixture.tests().iter().map(|t| { 

iter() su un Vec<T> restituisce un iteratore producendo oggetti di tipo &T, così il tuo t sarà di tipo &Box<Fn(&mut Self)>. Tuttavia, Box<Fn(&mut T)> non implementa Sync per impostazione predefinita (si tratta di un oggetto tratto che non dispone di informazioni sul tipo sottostante tranne che specificato esplicitamente), pertanto &Box<Fn(&mut T)> non può essere utilizzato su più thread. Questo è il secondo errore che vedi.

Molto probabilmente non si desidera utilizzare queste chiusure per riferimento; probabilmente vorrai spostarli interamente nel thread generato. Per questo è necessario utilizzare into_iter() invece di iter():

for t in fixture.tests().into_iter() { 

Ora t sarà di tipo Box<Fn(&mut T)>.Tuttavia, non può ancora essere inviato attraverso i thread. Ancora una volta, è un oggetto tratto e il compilatore non sa se il tipo contenuto all'interno è Send. Per questo è necessario aggiungere Send legato al tipo di chiusura:

fn tests(&mut self) -> Vec<Box<Fn(&mut Self)+Send>> 

Ora l'errore circa Fn è andato.

L'ultimo errore è circa Send non implementato per T. Abbiamo bisogno di aggiungere un Send rilegato su T:

pub fn test_fixture_runner<T: TestFixture+Send>(fixture: &mut T) { 

E ora l'errore diventa più comprensibile:

test.rs:18:22: 18:35 error: captured variable `fixture` does not outlive the enclosing closure 
test.rs:18   let handle = thread::spawn(move || { 
           ^~~~~~~~~~~~~ 
note: in expansion of closure expansion 
test.rs:18:5: 28:6 note: expansion site 
test.rs:15:66: 31:2 note: captured variable is valid for the anonymous lifetime #1 defined on the block at 15:65 
test.rs:15 pub fn test_fixture_runner<T: TestFixture+Send>(fixture: &mut T) { 
test.rs:16  fixture.setup(); 
test.rs:17 
test.rs:18  for t in fixture.tests().into_iter() { 
test.rs:19   let handle = thread::spawn(move || { 
test.rs:20    fixture.before_each(); 
      ... 
note: closure is valid for the static lifetime 

Questo errore si verifica perché si sta cercando di utilizzare un riferimento in un thread spawn() ed. spawn() richiede che l'argomento di chiusura abbia il limite 'static, ovvero che l'ambiente catturato non contenga riferimenti con durate diverse da 'static. Ma questo è esattamente ciò che accade qui: &mut T non è 'static. Il design spawn() non vieta di evitare l'unione, pertanto è stato scritto esplicitamente per non consentire il passaggio di riferimenti non-'static al thread generato.

notare che, mentre si sta utilizzando &mut T, questo errore è inevitabile, anche se si mette in &mut TArc, perché allora la vita di &mut T sarebbe "immagazzinata" in Arc e così Arc<Mutex<&mut T>>, inoltre, non sarà 'static.

Ci sono due modi per fare quello che vuoi.

Innanzitutto, è possibile utilizzare l'API instabile thread::scoped(). È instabile perché it is shown consente di non proteggere la memoria in un codice sicuro e il piano è di fornire un qualche tipo di sostituzione per il futuro. Tuttavia, è possibile utilizzarlo a Rust notte (non causerà memoria insicurezza per sé, solo in situazioni appositamente predisposte):

pub fn test_fixture_runner<T: TestFixture+Send>(fixture: &mut T) { 
    fixture.setup(); 

    let tests = fixture.lock().unwrap().tests(); 
    for t in tests.into_iter() { 
     let f = &mut *fixture; 

     let handle = thread::scoped(move || { 
      f.before_each(); 
      t(f); 
      f.after_each(); 
     }); 

     handle.join(); 
    } 

    fixture.teardown(); 
} 

Questo codice viene compilato perché scoped() è scritto in modo tale da garantire (in la maggior parte dei casi) che il thread non sopravvivrà a tutti i riferimenti catturati. Ho dovuto rinviare fixture perché altrimenti (poiché i riferimenti &mut non sono copiabili) verrebbero spostati nella discussione e fixture.teardown() sarebbe vietato. Inoltre ho dovuto estrarre la variabile tests perché altrimenti il ​​mutex sarà bloccato dal thread principale per la durata del ciclo for che naturalmente non consentirebbe di bloccarlo nei thread figli.

Tuttavia, con scoped() non è possibile isolare il panico nel thread secondario. Se il thread secondario si interrompe, il panico verrà riscritto dalla chiamata join(). Questo può o non può essere un problema in generale, ma penso che lo sia un problema per il tuo codice.

Un altro modo è quello di refactoring il codice per tenere l'apparecchio in Arc<Mutex<..>> fin dall'inizio:

pub fn test_fixture_runner<T: TestFixture + Send + 'static>(fixture: Arc<Mutex<T>>) { 
    fixture.lock().unwrap().setup(); 

    for t in fixture.lock().unwrap().tests().into_iter() { 
     let fixture = fixture.clone(); 

     let handle = thread::spawn(move || { 
      let mut fixture = fixture.lock().unwrap(); 

      fixture.before_each(); 
      t(&mut *fixture); 
      fixture.after_each(); 
     }); 

     if let Err(_) = handle.join() { 
      println!("Test failed!") 
     } 
    } 

    fixture.lock().unwrap().teardown(); 
} 

Si noti che ora T deve anche essere 'static, ancora una volta, perché altrimenti non poteva essere utilizzato con thread::spawn() come richiede 'static. fixture all'interno della chiusura interna non è &mut T ma a MutexGuard<T>, e quindi deve essere convertito in modo esplicito in &mut T per passarlo a t.

Questo può sembrare eccessivamente e inutilmente complesso, tuttavia, tale progettazione di un linguaggio di programmazione impedisce di effettuare molti errori nella programmazione multithread. Ognuno degli errori sopra riportati è valido - ognuno di questi sarebbe una potenziale causa di insicurezza della memoria o di dati razziali se fosse stato ignorato.

Answer 2

Come indicato nel Rust HandBook's Concurrency section:

quando un tipo T implementa Invia, si indica al compilatore che qualcosa di questo tipo è in grado di avere la proprietà trasferiti in modo sicuro tra i thread.

Se non si implementa Invia, la proprietà non può essere trasferita tra i thread.