Parallel.ForEach зависает после 3 возврата доходности?

Question

Я столкнулся с проблемой, когда действие в Parallel.ForEach иногда не вызывается.Я создал простую игрушечную программу, которая показывает проблему в однопоточном случае:

class Program
{
    static void Main(string[] args) 
    {
        // Any value > 3 here causes Parallel.ForEach to hang on the yield return
        int workCount = 4;
        bool inProcess = false;

        System.Collections.Generic.IEnumerable<int> getWorkItems()
        {
            while (workCount > 0)
            {
                if (!inProcess)
                {
                    inProcess = true;
                    System.Console.WriteLine($"    Returning work: {workCount}");
                    yield return workCount;
                }
            }
        }

        System.Threading.Tasks.Parallel.ForEach(getWorkItems(),
            new System.Threading.Tasks.ParallelOptions { MaxDegreeOfParallelism = 1 },
            (workItem) =>
            {
                System.Console.WriteLine($"      Parallel start:  {workItem}");
                workCount--;
                System.Console.WriteLine($"      Parallel finish: {workItem}");
                inProcess = false;
            });

        System.Console.WriteLine($"=================== Finished ===================\r\n");
    }
}

Вывод этой программы:

Returning work: 4
  Parallel start:  4
  Parallel finish: 4
Returning work: 3
  Parallel start:  3
  Parallel finish: 3
Returning work: 2
  Parallel start:  2
  Parallel finish: 2
Returning work: 1

... и он зависаетпрямо там.Действие никогда не называется 1. Что здесь происходит?

--------------------------- РЕДАКТИРОВАТЬ: Подробнееподробный пример -----------------------

Вот та же программа с более подробным выводом и некоторыми блокировками для защиты общих значений:

static object lockOnMe = new object();
static void Run()
{
    System.Console.WriteLine($"Starting ThreadId: {Thread.CurrentThread.ManagedThreadId}");

    // Any value > 3 here causes Parallel.ForEach to hang on the yield return
    int workCount = 40;
    bool inProcess = false;

    System.Collections.Generic.IEnumerable<int> getWorkItems()
    {
        while (workCount > 0)
        {
            lock(lockOnMe)
            {
                if (!inProcess)
                {
                    inProcess = true;
                    System.Console.WriteLine($"    Returning work: {workCount} ThreadId: {Thread.CurrentThread.ManagedThreadId}");
                    yield return workCount;
                }
            }

            Thread.Sleep(100);
            System.Console.Write($".");
        }
    }

    System.Threading.Tasks.Parallel.ForEach(getWorkItems(),
    new System.Threading.Tasks.ParallelOptions { MaxDegreeOfParallelism = 1 },
    (workItem) =>
    {
        lock(lockOnMe)
        {
            System.Console.WriteLine($"      Parallel start:  {workItem}  ThreadId: {Thread.CurrentThread.ManagedThreadId}");
            Interlocked.Decrement(ref workCount);
            System.Console.WriteLine($"      Parallel finish: {workItem}");
            inProcess = false;

        }
    });

    System.Console.WriteLine($"=================== Finished ===================\r\n");
}

выход:

Starting ThreadId: 1
Returning work: 40 ThreadId: 1
  Parallel start:  40  ThreadId: 1
  Parallel finish: 40
Returning work: 39 ThreadId: 1
  Parallel start:  39  ThreadId: 1
  Parallel finish: 39
Returning work: 38 ThreadId: 1
  Parallel start:  38  ThreadId: 1
  Parallel finish: 38
Returning work: 37 ThreadId: 1
......................

Answer 1

Не полный ответ, но вот что я узнал до сих пор:

Parallel.ForEach не ведет себя как традиционный многопоточный подход, который не будет иметь проблем с обменом данными, как это делает пример кода.Даже при защите общих данных для предотвращения одновременного доступа, по-видимому, происходит оптимизация в логике ForEach, которая не работает должным образом, если параллельные потоки должны быть заблокированы (например, обработка объектов в последовательном порядке).

Чтобы получить небольшую картину странности, которая происходит с Parallel.ForEach, попробуйте запустить этот фрагмент:

static void Run()
{
    System.Collections.Generic.IEnumerable<int> getWorkItems()
    {
        int workCount = 9999;
        while (workCount > 0)
        {
            System.Console.Write($"R");
            yield return workCount--;
            Thread.Sleep(10);
        }
    }

    System.Threading.Tasks.Parallel.ForEach(
        getWorkItems(),
        (workItem) => System.Console.Write($"."));
}

output:

R.R.R.R.R.R.R.R.R.R.R.R.R.R.R.R.R.R.R.R.R.R.R.R.R.R.R.R.R.R.R.R.R.R.R.R.R.R.R.
RR..RR..RR..RR..RR..RR..RR..RR..RR..RR..RR..RR..RR..RR..RR..RR..RR..RR..RR..RR
..RR..RR..RR..RR..RR..RR..RR..RR..RR..RR..RR..RR..RR..RR..RR..RR..RR..RR..RR..
RRRR....RRRR....RRRR....RRRR....RRRR....RRRR....RRRR....RRRR....RRRR....RRRR..
..RRRR....RRRR....RRRR....RRRR....RRRR....RRRR....RRRR....RRRR....RRRR....RRRR
....RRRR....RRRR....RRRR....RRRR....RRRR....RRRR....RRRR....RRRR....RRRR....RR
RR....RRRR....RRRR....RRRR....RRRR....RRRR....RRRR....RRRR....RRRR....RRRR....
RRRRRRRR........RRRRRRRR........RRRRRRRR........RRRRRRRR........RRRRRRRR......
..RRRRRRRR........RRRRRRRR........RRRRRRRR........RRRRRRRR........RRRRRRRR....
....RRRRRRRR........RRRRRRRR........R.RRRRRRRR........RRRRRRRR........RRRRRRRR
........RRRRRRRR........RRRRRRRR........RRRRRRRR........RRRRRRRR........RRRRRR
RR........RRRRRRRR........RRRRRRRR........RRRRRRRR........RRRRRRRR........RRRR
RRRR........R.RRRRRRRR........RRRRRRRR........RRRRRRRR........RRRRRRRR........
RRRRRRRR........RRRRRRRR........RRRRRRRR........RRRRRRRR........RRRRRRRR......
..RRRRRRRR........RRRRRRRR........RRRRRRRR........RRRRRRRR........R.RRRRRRRR..
......RRRRRRRRRRRR...

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

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

Answer 2

Вы получаете состояние, при котором вы возвращаете счетчик работ в одном потоке, после того как для inProcess установлено значение false в другом, что приводит вас в бесконечный цикл.Параллелизм охраняется на предметах, возвращаемых из перечисленного, за исключением того, что производитель и потребители будут изолированы.Если вы хотите, чтобы это работало, вам нужно будет установить блокировку везде, где вы получаете или устанавливаете процесс или рабочую станцию;

Если вы повысили уровень параллелизма, даже подсчет работы может быть неправильным.

Редактировать

Причина, по которой это не работает, является параметром по умолчанию, который Parallel.foreach использует для создания Partitioner для буферизации.Если вы создаете Partitioner самостоятельно и не разрешаете буферизацию, это работает как положено.По сути, в Partitioner есть эвристика, позволяющая забегать вперед и кэшировать результаты Ienumberable, что нарушает логику.

Если вы хотите, чтобы он работал должным образом, сделайте следующее.

    private static void Main(string[] args)
    {
        // Any value > 3 here causes Parallel.ForEach to hang on the yield return


        var partitioner = Partitioner.Create(getWorkItems(), EnumerablePartitionerOptions.NoBuffering);
        System.Threading.Tasks.Parallel.ForEach(partitioner,
            new System.Threading.Tasks.ParallelOptions { MaxDegreeOfParallelism = 1  },
            (workItem) =>
            {
                System.Console.WriteLine($"      Parallel start:  {workItem}");
                workCount--;
                System.Console.WriteLine($"      Parallel finish: {workItem}");
                inProcess = false;
            });

        System.Console.WriteLine($"=================== Finished ===================\r\n");
        var s = System.Console.ReadLine();
    }