Parallelisieren von code in verschachtelten Schleifen

Question

Sie hören immer, dass die funktionalen code ist von Natur aus einfacher zu parallelisieren, als nicht-funktionalen code, so dass ich beschlossen, eine Funktion schreiben, die das folgende tut:

Gegeben ist eine Eingabe von Zeichenfolgen, insgesamt die Anzahl der eindeutigen Zeichen, die für jede saite.So, angesichts der input - [ "aaaaa"; "bbb"; "ccccccc"; "abbbc" ], unsere Methode gibt a: 6; b: 6; c: 8.

Hier ist, was ich geschrieben habe:

(* seq<#seq<char>> -> Map<char,int> *)
let wordFrequency input =
    input
    |> Seq.fold (fun acc text ->
        (* This inner loop can be processed on its own thread *)
        text
        |> Seq.choose (fun char -> if Char.IsLetter char then Some(char) else None)
        |> Seq.fold (fun (acc : Map<_,_>) item ->
            match acc.TryFind(item) with
            | Some(count) -> acc.Add(item, count + 1)
            | None -> acc.Add(item, 1))
            acc
        ) Map.empty

Dieser code ist optimal parallelisierbare, da jede saite in input können verarbeitet werden auf seinen eigenen thread.Es ist nicht so einfach, wie es aussieht, da die innerloop fügt Elemente zu einer Karte in gemeinsamer für die Eingänge.

Ich möchte die innere Schleife berücksichtigt in seinen eigenen thread, und ich nicht wollen jede mutable state. Wie würde ich re-schreiben Sie diese Funktion mit einem Asynchronen workflow?

Answer 1

Wie bereits erwähnt, gibt es update-Konflikte, wenn Sie versuchen, verschiedene threads verarbeiten unterschiedliche input-strings, da kann jeder thread erhöht die Anzahl jedes Buchstabens.Sie können jeden thread produzieren, seine eigene Karte, und dann "hinzufügen, bis alle Karten", aber die letzten Schritt kann teuer sein (und ist nicht so gut geeignet, um die Verwendung von threads durch die freigegebenen Daten).Ich denke, die großen Eingänge sind wahrscheinlich schneller ausgeführt werden können mit einem Algorithmus wie dem folgenden, wo jeder thread verarbeitet einen anderen Buchstaben zu zählen (für alle Zeichenfolgen bei der Eingabe).Als Ergebnis, jeder thread hat seine eigene, unabhängige Schalter, so kein update Streit und keine Letzte Schritt, um die Ergebnisse zu kombinieren.Aber wir müssen Vorverarbeitung zu entdecken, die "Reihe von einzigartigen Buchstaben", und dieser Schritt hat die gleiche Behauptung problem.(In der Praxis werden Sie wahrscheinlich wissen, das Universum der Zeichen vorne, z.B.alphabetics, und dann kann nur erstellt 26-threads zu verarbeiten, a-z, und umgehen Sie dieses Problem.) In jedem Fall, vermutlich die Frage geht hauptsächlich um die Erkundung ', wie die zum schreiben von F# async-code aufteilen der Arbeit auf mehrere threads', so dass der code unten zeigt es.

#light

let input = [| "aaaaa"; "bbb"; "ccccccc"; "abbbc" |]

// first discover all unique letters used
let Letters str = 
    str |> Seq.fold (fun set c -> Set.add c set) Set.empty 
let allLetters = 
    input |> Array.map (fun str -> 
        async { return Letters str })
    |> Async.Parallel 
    |> Async.Run     
    |> Set.union_all // note, this step is single-threaded, 
        // if input has many strings, can improve this

// Now count each letter on a separate thread
let CountLetter letter =
    let mutable count = 0
    for str in input do
        for c in str do
            if letter = c then
                count <- count + 1
    letter, count
let result = 
    allLetters |> Seq.map (fun c ->
        async { return CountLetter c })
    |> Async.Parallel 
    |> Async.Run

// print results
for letter,count in result do
    printfn "%c : %d" letter count

Ich habe in der Tat "völlig geändert" - Algorithmus, vor allem, weil ich die original-Algorithmus, den Sie hatte, ist nicht besonders geeignet, um direkt auf die Daten der Parallelisierung durch den update-Konflikte.Je nach genau das, was Sie lernen, diese Antwort kann oder kann nicht besonders befriedigend zu Sie.

Answer 2

Sie können schreiben, die wie diese:

let wordFrequency =
  Seq.concat >> Seq.filter System.Char.IsLetter >> Seq.countBy id >> Map.ofSeq

und parallelisieren Sie es mit nur zwei zusätzliche Zeichen zu verwenden, die PSeq Modul aus dem FSharp.PowerPack.Parallel.Seq DLL anstelle der gewöhnlichen Seq Modul:

let wordFrequency =
  Seq.concat >> PSeq.filter System.Char.IsLetter >> PSeq.countBy id >> Map.ofSeq

Für Beispiel, die Zeit zu berechnen Frequenzen von 5.5 Mb King-James-Bibel fällt von 4.75 s auf 0,66 s.Das ist ein 7.2× speedup auf dieser 8-core-Maschine.

Answer 3

Parallel ist nicht das gleiche wie async, als Don Syme erläutert.

Also IMO, Sie wäre besser dran mit PLINQ zu parallelisieren.

Answer 4

I don ' T speak F# an alle gut, aber ich kann dieses Problem lösen.Denken Sie über die Verwendung von map/reduce:

lassen Sie n = Karte(Σ) die Anzahl der Symbole in σ, der im alphabet Σ.

Map-Phase:

Spawn n Prozesse, bei denen die Zuordnung der ich-th Prozess zur Zählung der Anzahl der vorkommen des symbols σ_ich in der gesamten Eingabevektor.

Reduzieren Bühne:

Sammeln Sie die Summe für jede der n Prozesse um.Vector ist Ihr Ergebnis.

Nun, diese version nicht zu irgendwelchen Verbesserungen, die über eine serielle version;Ich vermute, es ist ein Versteck Abhängigkeit hier, dass die macht dieser Natur aus schwer zu parallelisieren, aber ich bin zu müde und Gehirn-tot-zu beweisen, dass es heute Abend.