Как ограничить скорость, с которой .net Parallel.ForEach потребляет вход?

Question

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

Ввод поиска создается на лету, используя метод, который возвращает IEnumerable с возвратом доходности, поэтому он не накапливает весь набор данных сразу, а скорее генерирует его как потребляемый.

Это прекрасно подходит для небольших поисковых пространств, но когда я попадаю в сотни миллиардов записей, чтобы проверить процесс генерации тестовых примеров, почти сразу же исчерпывает 16 ГБ ОЗУ на моей машине и просто начинает измельчать компьютер со страницей до того, как начнется любая обработка.

Похоже, что Parallel.ForEach жадно пытается перечислить весь набор входных данных, прежде чем начать что-либо обрабатывать.

Итак, мой вопрос: как ограничить скорость, с которой Parallel.ForEach запрашивает входные данные из моего генератора тестового случая?

Я попытался использовать ChunkPartitioner в Samples for Parallel Programming with the .NET Framework с мыслью, что, возможно, ForEach перестанет запрашивать больше разделов, когда он полностью насытил пул потоков работой, но это, похоже, не так.

Я попробовал поиск по всему Интернету для объяснения того, как Parallel.ForEach потребляет входные данные, и способы, которыми я могу влиять на этот процесс, но я не могу найти ничего, кроме способов разбиения набора входных данных.

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

Answer 1

Здесь я вижу две (потенциальные) проблемы. Один из них заключается в том, что перечислимый разделитель по умолчанию, действительно, жадный (что означает, что он пытается материализовать элементы, которые впереди в очереди). Это легко исправить с помощью EnumerablePartitionerOptions в .NET 4.5:

var partitioner = Partitioner.Create(
    EnumerateItems(), // IEnumerable<T> 
    EnumerablePartitionerOptions.NoBuffering 
); 

Parallel.ForEach(partioner, new ParallelOptions { MaxDegreeOfParallelism = 4 }, i => 
{ 
    // Process item. 
}); 

Вторая проблема может проявить себя, если ваша работа не CPU переплете в конечном счете, в результате чего Parallel.ForEach нарастить число рабочих потоков, который адресован указав MaxDegreeOfParallelism, чтобы предотвратить насыщение пула потоков.