Kann ein Rostverschluss von mehreren Fäden verwendet werden?

Question

Ich möchte die Möglichkeit, mehrere Threads zu bewerten, die gleiche Schließung. Die Anwendung, die ich im Auge habe, ist die parallelisierte numerische Integration, also eine Situation, in der die Funktionsdomäne leicht in N Blöcke aufgeteilt und an Threads übergeben werden kann.

Außer es funktioniert nicht. Betrachten Sie diesen Code:

use std::thread; 
use std::sync::mpsc; 
const THREAD_COUNT: u64 = 4; 

fn average<F:Fn(f64)->f64>(f: F) -> f64 { 
    let (tx, rx) = mpsc::channel(); 
    for id in 0..THREAD_COUNT { 
     let thread_tx = tx.clone(); 
     thread::spawn(move || { 
      thread_tx.send(f(id as f64)); 
     }); 
    } 

    let mut total = 0.0; 
    for id in 0..THREAD_COUNT { 
     total += rx.recv().unwrap(); 
    } 
    total/THREAD_COUNT as f64 
} 

fn main() { 
    average(|x:f64|->f64{x}); 
} 

Dies ist eine einfache Funktion, die die angegebene Schließung mehrmals auswertet und das Ergebnis mittelt. Aber wenn ich kompilieren ich diesen Fehler:

error: the trait `core::marker::Send` is not implemented for the type `F` [E0277] 

So füge ich +Send auf die Grenzen auf F und einen neuen Fehler:

error: the parameter type `F` may not live long enough [E0310] 
consider adding an explicit lifetime bound `F: 'static`... 

Also ich +'static-F hinzufügen und diese:

error: capture of moved value: `f` [E0382] 
note: `f` moved into closure environment here because it has type `F`, which is non-copyable 

So füge ich +Copy-F und erhalten:

error: the trait `core::marker::Copy` is not implemented for the type `[closure@src/test.rs:115:11: 115:26] 

So scheint es, jeder Thread seine eigene Kopie des Verschlusses will (wegen move) aber Verschlüsse nicht umsetzen Copy so kein Glück. Es erscheint mir komisch, denn wenn die Closures niemals mutieren, was ist dann das Sicherheitsproblem, wenn mehrere Threads auf sie zugreifen?

Ich kann den Code zu arbeiten, indem eine reguläre Funktion anstelle einer Schließung, aber dies macht meinen Code nicht-generisch, d. H. Es funktioniert nur für eine bestimmte Funktion anstelle von allem, was Fn(f64)->f64 ist. Und für die Art der Integration, die ich mache, haben die integrierten Funktionen oft bestimmte feste Variablen, die mit der Variablen der Integration gemischt sind, so dass es naheliegend erscheint, die festen Variablen mit einer Schließung zu erfassen.

Gibt es einen Weg, um diese Art der Multithread-Funktion Bewertung in einer generischen Weise zu arbeiten? Denk ich nur über Dinge falsch?

Answer 1

Das ultimative Problem dreht sich um wer besitzt die Schließung. Der geschriebene Code besagt, dass das Eigentum an der Schließung an übertragen wird. Diese Funktion versucht dann, die Schließung mehreren Threads zuzuweisen, was fehlschlägt, wie Sie gesehen haben, da Sie einem Objekt nicht mehrere untergeordnete Elemente zuweisen können.

Ebenso kann man nicht einen Verweis auf die Schließung von average Besitz geben, weil der Faden mit thread::spawn erzeugt wird, kann den Anruf zu average überleben. Wenn beendet wird, werden alle im Stack zugewiesenen Variablen zerstört. Jede Verwendung von ihnen würde Speicherunsicherheit verursachen, die Rust verhindern soll.

Eine Lösung ist eine Arc zu verwenden. Dies ermöglicht mehrere gemeinsame Besitzer einer einzelnen Ressource in einem Multithread-Kontext. Wenn der umschlossene Abschluss geklont wird, wird nur ein neuer Verweis erstellt. Wenn alle Referenzen verschwinden, wird das Objekt freigegeben.

use std::thread; 
use std::sync::mpsc; 
use std::sync::Arc; 

const THREAD_COUNT: u64 = 4; 

fn average<F>(f: F) -> f64 
    where F: Fn(f64) -> f64 + Send + Sync + 'static 
{ 
    let (tx, rx) = mpsc::channel(); 
    let f = Arc::new(f); 

    for id in 0..THREAD_COUNT { 
     let thread_tx = tx.clone(); 
     let f = f.clone(); 
     thread::spawn(move || { 
      thread_tx.send(f(id as f64)).unwrap(); 
     }); 
    } 

    let mut total = 0.0; 
    for _ in 0..THREAD_COUNT { 
     total += rx.recv().unwrap(); 
    } 

    total/THREAD_COUNT as f64 
} 

fn main() { 
    average(|x| x); 
} 

Eine Standardlösung ist Threads verwenden scoped. Diese Threads werden garantiert zu einem bestimmten Zeitpunkt beendet, wodurch Sie Referenzen übergeben können, die die Threads über die Threads hinausreichen.Es gibt andere questions, dass address wie man diese benutzt.