Параллельная конвейерная обработка

Похоже, вы просите совсем немного. Я оставлю это на ваше усмотрение, чтобы разобраться со строкой, но я покажу вам, как определить оператор, который выполняет серию операций параллельно.

Шаг 1: Напишите предохранитель function

Ваша функция fuse, похоже, отображает один вход с использованием нескольких функций, что достаточно легко записать следующим образом:

//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

Таким образом, оператор | >> принимает набор входных данных и отображает их, используя множество различных выходных данных. Пока, если мы сложим все это вместе, мы получим следующее (в 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 (+)