Почему Parallel.Foreach создает бесконечные потоки?

Я думаю, что Parallel.ForEach() был создан в первую очередь для обработки ограниченных коллекций. И он не ожидает коллекций, подобных той, что возвращает GetConsumingPartitioner(), где MoveNext() блокируется на долгое время.

Проблема в том, что Parallel.ForEach() пытается найти наилучшую степень параллелизма, поэтому запускает столько Task, сколько позволяет TaskScheduler. Но TaskScheduler видит, что есть много Task, выполнение которых занимает очень много времени, и что они ничего не делают (они блокируются), поэтому он продолжает начинать новые.

Я думаю, что лучшее решение - установить MaxDegreeOfParallelism.

В качестве альтернативы вы можете использовать ActionBlock потока данных TPL. Основное отличие в этом случае состоит в том, что ActionBlock не блокирует потоки, когда нет элементов для обработки, поэтому количество потоков не приближается к пределу.

Шаблон производитель/потребитель в основном используется, когда есть только один производитель и один потребитель.

Однако то, чего вы пытаетесь достичь (несколько потребителей), более точно соответствует шаблону рабочего списка. Следующий код был взят со слайда для слайда модуля 2 «2c — Shared Memory Patterns» из курса параллельного программирования, преподаваемого в Университете штата Юта, который доступен для загрузки по адресу http://ppcp.codeplex.com/

BlockingCollection<Item> workList;
CancellationTokenSource cts;
int itemcount

public void Run()
{
  int num_workers = 4;

  //create worklist, filled with initial work
  worklist = new BlockingCollection<Item>(
    new ConcurrentQueue<Item>(GetInitialWork()));

  cts = new CancellationTokenSource();
  itemcount = worklist.Count();

  for( int i = 0; i < num_workers; i++)
    Task.Factory.StartNew( RunWorker );
}

IEnumberable<Item> GetInitialWork() { ... }

public void RunWorker() {
  try  {
    do {
      Item i = worklist.Take( cts.Token );
      //blocks until item available or cancelled
          Process(i);
      //exit loop if no more items left
    } while (Interlocked.Decrement( ref itemcount) > 0);
  } finally {
      if( ! cts.IsCancellationRequested )
        cts.Cancel();
    }
  }
}

public void AddWork( Item item) {
  Interlocked.Increment( ref itemcount );
  worklist.Add(item);
}

public void Process( Item i ) 
{
  //Do what you want to the work item here.
}

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

Другим замечательным ресурсом для Parallelism on .Net 4.0 является книга «Параллельное программирование с Microsoft .Net», которая находится в свободном доступе по адресу: http://msdn.microsoft.com/en-us/library/ff963553

Внутри библиотеки параллельных задач Parallel.For и Parallel.Foreach следуют алгоритму восхождения на вершину, чтобы определить, сколько параллелизма следует использовать для операции.

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

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

Я узнал об этом во время лекции, прочитанной одним из разработчиков Task Parallel Library.

Указание MaxDegreeOfParallelism, вероятно, самый простой способ.