代码之家 › 专栏 › 技术社区 › Juliet

在嵌套循环中并行化代码

async-workflow f# multithreading

Juliet · 技术社区 · 16 年前

你总是听说函数代码比非函数代码更容易并行化,所以我决定编写一个函数,它执行以下操作:

给定一个字符串输入,合计每个字符串的唯一字符数。所以,考虑到输入 [ "aaaaa"; "bbb"; "ccccccc"; "abbbc" ] ,我们的方法将返回 a: 6; b: 6; c: 8 .

以下是我写的:

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

此代码在理想情况下是可并行的,因为 input 可以在自己的线程上处理。它不像看起来那么简单,因为innerloop将项目添加到所有输入之间共享的映射中。

我希望内环分解成它自己的线程,我不想使用任何可变状态。 如何使用异步工作流重新编写此函数?

4 回复 | 直到 15 年前

Brian 16 年前

正如已经指出的,如果您尝试让不同的线程处理不同的输入字符串,则会有更新争用,因为每个线程都可以增加每个字母的计数。您可以让每个线程生成自己的映射,然后“添加所有映射”,但最后一步可能会很昂贵(而且由于共享数据,不太适合使用线程)。我认为使用下面的算法,大型输入可能运行得更快,其中每个线程处理不同的字母进行计数(对于输入中的所有字符串)。因此,每个线程都有自己的独立计数器,因此没有更新争用,也没有最后一步来组合结果。然而,我们需要预处理来发现“唯一字母集”,这一步确实有相同的争用问题。(实际上,您可能知道前面的字符世界,例如字母,然后就可以创建26个线程来处理a-z,并绕过这个问题。)在任何情况下,可能的问题主要是探索“如何编写f异步代码来划分线程间的工作”,因此下面的代码演示了这一点。

#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

我确实“完全改变了算法”,主要是因为我原来的算法不太适合由于更新争用而直接进行数据并行化。取决于你到底想学什么,这个答案可能会让你特别满意,也可能不会让你特别满意。

J D 15 年前

你可以这样写:

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

并用两个额外的字符将其并行以使用 PSeq 模块从 FSharp.PowerPack.Parallel.Seq 而不是普通的 Seq 模块:

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

例如,从5.5MB King James Bible计算频率所用的时间从4.75s下降到0.66s,这是8核机器的7.2倍加速。

Mauricio Scheffer 16 年前

Parallel与Async不同,因为 Don Syme explains .

所以我觉得你最好用plinq来并行化。

Charlie Martin 16 年前

我一点也不会说F,但我可以解决这个问题。考虑使用map/reduce:

让 n = 卡(七) 是字母表∑中的符号数σ。

地图阶段:

产卵 n 进程,其中我 -这个过程是计算符号出现的次数。 西格玛 _我 在整个输入向量中。

减少阶段 :

为每个 n 按顺序处理。这个向量就是你的结果。

现在,这个版本不会比一个串行版本有任何改进;我怀疑这里有一个隐藏的依赖性,这使得它本来就很难并行,但是我太累了,大脑已经死了,无法在今晚证明它。