Parallelizza il codice in loop nidificati

Question

Si sente sempre che il codice funzionale è intrinsecamente più semplice da parallelizzare rispetto al codice non funzionale, quindi ho deciso di scrivere una funzione che procede come segue:

Dato un input di stringhe, sommare il numero di caratteri univoci per ogni stringa. Quindi, dato l'input [" aaaaa " ;; & Quot; BBB " ;; & Quot; ccccccc " ;; & Quot; abbbc " ] , il nostro metodo restituirà a: 6; b: 6; c: 8 .

Ecco cosa ho scritto:

(* 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

Questo codice è idealmente parallelizzabile, poiché ogni stringa in input può essere elaborata sul proprio thread. Non è così semplice come sembra poiché innerloop aggiunge elementi a una mappa condivisa tra tutti gli input.

Vorrei che il ciclo interno fosse inserito nel proprio thread e non voglio usare alcuno stato mutabile. Come riscriverei questa funzione usando un flusso di lavoro asincrono?

Answer 1

Come già sottolineato, c'è una contesa di aggiornamento se si tenta di fare in modo che thread diversi elaborino stringhe di input diverse, poiché ogni thread può incrementare il conteggio di ogni lettera. Puoi fare in modo che ogni thread produca la propria mappa e quindi "sommare tutte le mappe", ma quel passaggio finale può essere costoso (e non è adatto all'utilizzo dei thread a causa dei dati condivisi). Penso che gli input di grandi dimensioni possano funzionare più velocemente usando un algoritmo come quello qui sotto, in cui ogni thread elabora una lettera da contare diversa (per tutte le stringhe nell'input). Di conseguenza, ogni thread ha il proprio contatore indipendente, quindi nessuna contesa di aggiornamento e nessun passaggio finale per combinare i risultati. Tuttavia abbiamo bisogno della preelaborazione per scoprire il "set di lettere univoche" e questo passaggio presenta lo stesso problema di contesa. (In pratica, probabilmente conosci l'universo dei personaggi in primo piano, ad esempio alfabetici, e quindi puoi semplicemente creare 26 thread per elaborare az e aggirare questo problema.) In ogni caso, presumibilmente la domanda è principalmente di esplorare "come scrivere F # codice asincrono per dividere il lavoro tra thread ", quindi il codice seguente lo dimostra.

#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

Ho davvero 'completamente cambiato l'algoritmo', soprattutto perché l'algoritmo originale che hai avuto non è particolarmente adatto per indirizzare la parallelizzazione dei dati a causa della contesa di aggiornamento. A seconda di cosa hai intenzione di imparare, questa risposta può o meno essere particolarmente soddisfacente per te.

Answer 2

Puoi scriverlo così:

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

e parallelizzalo con solo due caratteri extra per utilizzare il modulo PSeq dalla FSharp.PowerPack.Parallel.Seq anziché dalla normale Seq modulo:

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

Ad esempio, il tempo impiegato per calcolare le frequenze dalla bibbia King James da 5,5 Mb scende da 4,75 a 0,66 secondi. Questo è un aumento di velocità 7,2 × su questa macchina a 8 core.

Answer 3

Parallel non è lo stesso di asincrono, come spiega Don Syme .

Quindi IMO sarebbe meglio usare PLINQ per parallelizzare.

Answer 4

Non parlo affatto bene F #, ma posso affrontarlo. Pensa all'utilizzo della mappa / riduci:

lascia n = carta (S) essere il numero di simboli s nell'alfabeto S.

Stage mappa:

Genera processi n , in cui l'assegnazione del i -th processo consiste nel calcolare il numero di occorrenze del simbolo s _i nell'intero vettore di input.

Riduci livello :

Raccogli il totale per ciascuno dei processi n in ordine. Quel vettore è il tuo risultato.

Ora, questa versione non comporta alcun miglioramento rispetto a una versione seriale; Ho il sospetto che qui ci sia una dipendenza nascosta che rende intrinsecamente difficile parallelizzare questo, ma sono troppo stanco e cervello morto per provarlo stasera.