F # PSeq.iter, похоже, не использует все ядра

В F # я делаю некоторую вычислительно-интенсивную работу. Такие функции, как Array.Parallel.map, которые используют параллельную библиотеку .Net Task, ускорили мой код экспоненциально для действительно весьма минимальных усилий.

Однако из-за проблем с памятью я переделываю раздел моего кода, чтобы его можно было лениво оценить внутри выражения последовательности (это означает, что я должен хранить и передавать меньше информации). Когда пришло время оценить, я использовал:

// processor and memory intensive task, results are not stored
let calculations : seq<Calculation> =  seq { ...yield one thing at a time... }

// extract results from calculations for summary data
PSeq.iter someFuncToExtractResults results

Вместо:

// processor and memory intensive task, storing these results is an unnecessary task
let calculations : Calculation[] = ...do all the things...

// extract results from calculations for summary data
Array.Parallel.map someFuncToExtractResults calculations

При использовании любой из функций Array.Parallel я могу отчетливо видеть, что все ядра на моем компьютере перешли на передачу (~ 100% использования ЦП). Однако требуемая дополнительная память означает, что программа никогда не заканчивается.

С версией PSeq.iter, когда я запускаю программу, используется только около 8% использования ЦП (и минимальное использование ОЗУ).

Итак: есть ли причина, по которой версия PSeq работает намного медленнее? Это из-за ленивой оценки? Есть ли какая-то магия "быть параллельной", которую мне не хватает?

Спасибо,

Другие ресурсы, реализация исходного кода обоих (они, похоже, используют разные параллельные библиотеки в .NET):

https://github.com/fsharp/fsharp/blob/master/src/fsharp/FSharp.Core/array.fs

https://github.com/fsharp/powerpack/blob/master/src/FSharp.PowerPack.Parallel.Seq/pseq.fs

EDIT: добавлено более подробное описание примеров и деталей кода

код:

Seq

// processor and memory intensive task, results are not stored
let calculations : seq<Calculation> =  
    seq { 
        for index in 0..data.length-1 do
            yield calculationFunc data.[index]
    }

// extract results from calculations for summary data (different module)
PSeq.iter someFuncToExtractResults results

Массив

// processor and memory intensive task, storing these results is an unnecessary task
let calculations : Calculation[] =
    Array.Parallel.map calculationFunc data

// extract results from calculations for summary data (different module)
Array.Parallel.map someFuncToExtractResults calculations

Подробнее:

Сохранение версии промежуточного массива выполняется быстро (насколько это возможно до краха) менее чем за 10 минут, но использует ~ 70 ГБ оперативной памяти до того, как он сработает (64 ГБ физического, остальные выгружены)
Версия seq занимает более 34 минут и использует долю оперативной памяти (всего около 30 ГБ).
Там вычисляется миллиард значений. Следовательно, миллиард удваивается (по 64 бита каждый) = 7,4505806 ГБ. Там более сложные формы данных... и несколько ненужных копий, которые я очищаю, следовательно, в настоящее время массовое использование ОЗУ.
Да, архитектура невелика, ленивая оценка - первая часть меня, пытающаяся оптимизировать программу и/или доставлять данные в более мелкие куски.
С меньшим набором данных обе части кода выводят те же результаты.
@pad, я попробовал то, что вы предложили, PSeq.iter, казалось, работал правильно (все активные ядра), когда он кормился Calculation [], но все еще остается вопрос о RAM (в конечном итоге он разбился).
как итоговая часть кода, так и часть вычисления интенсифицированы ЦП (в основном из-за больших наборов данных)
В версии Seq я просто планирую распараллеливать один раз

Ответ 1

Основываясь на вашей обновленной информации, я сокращаю свой ответ только на соответствующую часть. Вам просто нужно это вместо того, что вы в настоящее время имеете:

let result = data |> PSeq.map (calculationFunc >> someFuncToExtractResults)

И это будет работать одинаково, если вы используете PSeq.map или Array.Parallel.map.

Однако ваша реальная проблема не будет решена. Эта проблема может быть сформулирована так: когда достигается желаемая степень параллельной работы, чтобы получить 100% -ное использование ЦП, недостаточно памяти для поддержки процессов.

Вы видите, как это не будет разрешено? Вы можете обрабатывать вещи последовательно (меньше эффективности ЦП, но эффективно с памятью), или вы можете обрабатывать вещи параллельно (более эффективный процессор, но заканчивается из памяти).

Возможны следующие варианты:

Измените степень parallelism, которая будет использоваться этими функциями, чтобы что-то не взорвало вашу память:
```
let result = data 
             |> PSeq.withDegreeOfParallelism 2 
             |> PSeq.map (calculationFunc >> someFuncToExtractResults)
```
Измените базовую логику для calculationFunc >> someFuncToExtractResults, чтобы она была более эффективной и передавала данные по результатам. Не зная больше деталей, не просто понять, как это можно сделать. Но внутренне, конечно, может быть возможна ленивая загрузка.

Ответ 2

Array.Parallel.map использует Parallel.For под капотом, а PSeq - тонкая оболочка вокруг PLINQ. Но причина, по которой они ведут себя по-другому, заключается в том, что для PSeq.iter недостаточно рабочих нагрузок, когда seq<Calculation> является последовательным и слишком медленным с получением новых результатов.

Я не понимаю, как использовать промежуточный сегмент или массив. Предположим, что data является входным массивом, перемещение всех вычислений в одном месте - путь:

// Should use PSeq.map to match with Array.Parallel.map
PSeq.map (calculationFunc >> someFuncToExtractResults) data

и

Array.Parallel.map (calculationFunc >> someFuncToExtractResults) data

Вы избегаете использования слишком большого объема памяти и интенсивного вычисления в одном месте, что приводит к повышению эффективности параллельного выполнения.