并行流水线

Question

 (fileNameToCharStream "bigfile"
 |>> fuse [length;
           splitBy (fun x -> x = ' ' || x = '\n') removeEmpty |>> length;
           splitBy (fun x -> x = '\n') keepEmpty |>> length;
         ])
  (*fuse "fuses" the three functions to run concurrently*)
 |> run 2  (*forces to run in parallel on two threads*)
 |> (fun [num_chars; num_words; num_lines] -> 
       printfn "%d %d %d"
           num_chars num_words, num_lines))

我想让这段代码按以下方式工作： 将原始流正好在中间分成两部分；然后 对于每一半运行一个单独的计算 计算三件事：长度（即字符数）， 字数、行数。 但是，我不想出现问题，如果 我错误地分割了一个单词。这必须是 得到照顾。该文件只能读取一次。

我应该如何编写指定的函数和运算符|>>？ 是否可以？

Answer 1

看来你的要求有点高了。我将让您自行弄清楚字符串操作，但我将向您展示如何定义一个并行执行一系列操作的运算符。

第 1 步：编写 fuse 函数

您的 fusion 函数似乎使用多个函数映射单个输入，这很容易编写如下：

//val fuse : seq<('a -> 'b)> -> 'a -> 'b list
let fuse functionList input = [ for f in functionList -> f input]

请注意，所有映射函数需要有相同的类型。

第 2 步：定义并行执行函数的运算符

标准并行映射函数可以编写如下：

//val pmap : ('a -> 'b) -> seq<'a> -> 'b array
let pmap f l =
    seq [for a in l -> async { return f a } ]
    |> Async.Parallel
    |> Async.RunSynchronously

据我所知，Async.Parallel 将并行执行异步操作，其中在任何给定时间执行的并行任务数量等于机器上的核心数量（如果我错了，有人可以纠正我）。因此，在双核机器上，当调用此函数时，我们的机器上最多应该有 2 个线程运行。这是一件好事，因为我们不希望通过每个核心运行多个线程来提高速度（事实上，额外的上下文切换可能会减慢速度）。

我们可以根据pmap和fuse定义一个运算符|>>：

//val ( |>> ) : seq<'a> -> seq<('a -> 'b)> -> 'b list array
let (|>>) input functionList = pmap (fuse functionList) input

所以|>> code> 运算符接受一堆输入并使用许多不同的输出来映射它们。到目前为止，如果我们将所有这些放在一起，我们会得到以下结果（在 fsi 中）：

> let countOccurrences compareChar source =
    source |> Seq.sumBy(fun c -> if c = compareChar then 1 else 0)

let length (s : string) = s.Length

let testData = "Juliet is awesome|Someone should give her a medal".Split('|')
let testOutput =
    testData
    |>> [length; countOccurrences 'J'; countOccurrences 'o'];;

val countOccurrences : 'a -> seq<'a> -> int
val length : string -> int
val testData : string [] =
  [|"Juliet is awesome"; "Someone should give her a medal"|]
val testOutput : int list array = [|[17; 1; 1]; [31; 0; 3]|]

testOutput 包含两个元素，这两个元素都是并行计算的。

第 3 步：将元素聚合到单个输出

好吧，现在我们有了由数组中的每个元素表示的部分结果，并且我们希望将部分结果合并到单个聚合中。我假设数组中的每个元素应该合并相同的函数，因为输入中的每个元素都具有相同的数据类型。

这是我为这项工作编写的一个非常丑陋的函数：

> let reduceMany f input =
    input
    |> Seq.reduce (fun acc x -> [for (a, b) in Seq.zip acc x -> f a b ]);;

val reduceMany : ('a -> 'a -> 'a) -> seq<'a list> -> 'a list

> reduceMany (+) testOutput;;
val it : int list = [48; 1; 4]

reduceMany 接受 n 长度序列的序列，并返回一个 n 长度数组作为输出。如果您能想出更好的方法来编写此函数，请成为我的客人:)

解码上面的输出：

• 48 = 我的两个输入字符串的长度之和。请注意，原始字符串有 49 个字符，但将其拆分为“|”每个“|”消耗一个字符。
• 1 = 我的输入中所有“J”实例的总和
• 4 = 所有“O”实例的总和。

第 4 步：将所有内容放在一起

let pmap f l =
    seq [for a in l -> async { return f a } ]
    |> Async.Parallel
    |> Async.RunSynchronously

let fuse functionList input = [ for f in functionList -> f input]

let (|>>) input functionList = pmap (fuse functionList) input

let reduceMany f input =
    input
    |> Seq.reduce (fun acc x -> [for (a, b) in Seq.zip acc x -> f a b ])

let countOccurrences compareChar source =
    source |> Seq.sumBy(fun c -> if c = compareChar then 1 else 0)

let length (s : string) = s.Length

let testData = "Juliet is awesome|Someone should give her a medal".Split('|')
let testOutput =
    testData
    |>> [length; countOccurrences 'J'; countOccurrences 'o']
    |> reduceMany (+)