如何进行并行过滤答案

【问题标题】：How to do Parallel filtering如何进行并行过滤
【发布时间】：2013-04-19 19:39:56
【问题描述】：

如何在 F# 中实现并行过滤？基本上我想做Array.Parallel.Choose，除了使用序列表达式来创建一个数组。

我试过了：

Async.Parallel [for x in 1..40 do yield async { if x % 5 = 0 then return x }]

但这是类型不匹配，因为 if 并不总是有值（即 Async<unit> 不是 Async<'a>）。

我正在迭代大量数字 (1,000,000,000 +)，所以我不想预先生成序列。

if语句中的实际检查是：

let isPalindrome (x : int) = let numberArray = x.ToString().ToCharArray()
                             numberArray = Array.rev numberArray

尝试使用 PSeq：

[for x in 990000..999999 do for y in 990000..999999 do yield (x, y, x*y)]
|> PSeq.filter(fun (x, y, z) -> isPalindrome z)
|> Seq.max

这会产生OutOfMemoryException。

丑陋的解决方法：

let numberArray = [|990000..999999|]
let result = numberArray |> Array.Parallel.collect(fun x -> [| for y in numberArray do if isPalindrome (x*y) then yield (x, y, x*y)|])
             |> Array.maxBy(fun (x, y, z) -> z)

【问题讨论】：

F# PowerPack 中有一个PSeq (Parallel Sequence) 模块，你试过了吗？另外，如果您只是在过滤器中执行简单的mod，我怀疑您是否会通过使用并行序列来实现很多（如果有的话）性能改进......
@ildjarn mod 只是一个占位符。我实际上是在玩我对 Project Euler 问题 4 projecteuler.net/problem=4 的解决方案，看看我能把它扩大到多少。我希望避免使用外部库。
@ildjarn 我查看了PSeq，但它似乎不能满足我的需求：即它适用于现有序列而不是序列生成
你明白序列是惰性的，对吧？有一个seq 生成器，并使用PSeq 进行过滤。

标签： f# parallel-processing task-parallel-library plinq

【解决方案1】：

let isPalindrome (x : int) = let numberArray = x.ToString().ToCharArray()
                             numberArray = Array.rev numberArray

我想出了如何使用PSeq 库对序列进行惰性过滤：

let generator =
    seq {
        for x in 990000..999999 do 
            for y in 990000..999999 do 
                yield (x, y, x*y) 
    }

generator
|> PSeq.filter(fun (x, y, z) -> isPalindrome z)
|> Seq.max

真实：00:00:28.938，CPU：00:01:49.078，GC gen0：3448，gen1：2，gen2：0

遗憾的是，这比我丑陋的 hack 慢：

let numberArray = [|990000..999999|]
let result = numberArray |> Array.Parallel.collect(fun x -> [| for y in numberArray do if isPalindrome (x*y) then yield (x, y, x*y)|])
             |> Array.maxBy(fun (x, y, z) -> z)

真实：00:00:24.779，CPU：00:01:32.109，GC gen0：2680，gen1：18，gen2：1

【讨论】：

@ildjarn 哎哟......我需要一些时间才能理解这一点！

【解决方案2】：

基于 Array.Choose 实现的更高效的解决方案：

 open System.Collections.Concurrent
 open System.Threading.Tasks
 open System


 let parallelFilter f (array: 'T[]) = 

            let inputLength = array.Length                      

            let isChosen : bool [] = Array.zeroCreate inputLength                
            let mutable outputLength = 0        
            let range = Partitioner.Create(0,inputLength)
            Parallel.ForEach(
                         range,
                         (fun () ->0),
                         (fun (start,finish) _ count -> 
                            let mutable localCount = 0
                            for i in start .. finish-1 do
                                match f array.[i] with 
                                | true -> () 
                                | false -> 
                                    isChosen.[i] <- true                                      
                                    localCount <- localCount+1
                            localCount),
                          Action<int> (fun x -> System.Threading.Interlocked.Add(&outputLength,x) |> ignore )
                          ) |> ignore         

            let output = Array.zeroCreate outputLength
            let mutable curr = 0
            for i = 0 to isChosen.Length-1 do 
                if isChosen.[i] then 
                    output.[curr] <- output.[i]
                    curr <- curr + 1
            output

【讨论】：