Parallel.For() с Interlocked.CompareExchange(): более низкая производительность и немного отличающиеся результаты от серийной версии

Question

Я экспериментировал с вычислением среднего значения списка, используя Parallel.For(), Я отказался от этого, так как он примерно в четыре раза медленнее, чем простая серийная версия. И все же я заинтригован тем фактом, что он не дает того же результата, что и серийный, и я подумал, что было бы полезно узнать почему.

Мой код:

public static double Mean(this IList<double> list)
{
        double sum = 0.0;


        Parallel.For(0, list.Count, i => {
                        double initialSum;
                        double incrementedSum;
                        SpinWait spinWait = new SpinWait();

                        // Try incrementing the sum until the loop finds the initial sum unchanged so that it can safely replace it with the incremented one.
                        while (true) {
                            initialSum = sum;
                            incrementedSum = initialSum + list[i];
                            if (initialSum == Interlocked.CompareExchange(ref sum, incrementedSum, initialSum)) break;
                            spinWait.SpinOnce();
                        }
                     });

        return sum / list.Count;
    }

Когда я запускаю код в случайной последовательности 2000000 точек, я получаю результаты, которые в последних 2 цифрах отличаются от серийного среднего значения.

Я искал stackru и обнаружил следующее: текущая сумма VB.NET во вложенном цикле внутри Parallel.for Synclock теряет информацию. Мой случай, однако, отличается от описанного там. Там локальная переменная потока temp является причиной неточности, но я использую одну сумму, которая обновляется (я надеюсь) в соответствии с учебником Interlocked.CompareExchange() шаблон. Вопрос, конечно, спорный из-за плохой работы (что меня удивляет, но я знаю о накладных расходах), но мне любопытно, есть ли что-то, что можно извлечь из этого случая.

Ваши мысли ценятся.

Answer 1

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

Вы обнаружили, что математика с плавающей точкой является коммуникативной, но не ассоциативной. Или, другими словами, a + b == b + a но a + b + c != a + c + b, В вашем коде подразумевается, что порядок добавления чисел довольно случайный.

Этот вопрос C++ говорит и об этом.

Answer 2

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

Поскольку вы не показали свой последовательный код, я приму абсолютно простой случай:

double sum = list.Sum();

Это очень простая операция, которая должна работать примерно так же быстро, как это возможно для одного ядра процессора. С очень большим списком кажется, что можно использовать несколько ядер для суммирования списка. И, как оказалось, вы можете:

double sum = list.AsParallel().Sum();

Несколько прогонов этого на моем ноутбуке (i3 с 2 ядрами /4 логическими процессорами) дают ускорение примерно в 2,6 раза за несколько прогонов против 2 миллионов случайных чисел (тот же список, несколько прогонов).

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

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

Предполагая, что операция, которую вы на самом деле выполняете, является более сложной, чем простая Sum() может справиться, вы можете обнаружить, что Aggregate() метод более полезен для вас, чем Parallel.For,

Есть несколько перегрузок Aggregate расширение, в том числе одно, которое фактически является реализацией Map Pattern, схожим с тем, как работают системы с большими данными, такие как MapReduce. Документация здесь.

Эта версия Aggregate использует начальное значение аккумулятора (начальное значение для каждого потока) и три функции:

1. updateAccumulatorFunc вызывается для каждого элемента в последовательности и возвращает обновленное значение аккумулятора

2. combineAccumulatorsFunc используется для объединения аккумуляторов из каждого раздела (потока) в ваш параллельный перечислимый

3. resultSelector выбирает окончательное выходное значение из накопленного результата.

Параллельная сумма с использованием этого метода выглядит примерно так:

double sum = list.AsParallel().Aggregate(
    // seed value for accumulators
    (double)0, 
    // add val to accumulator
    (acc, val) => acc + val,
    // add accumulators
    (acc1, acc2) => acc1 + acc2,
    // just return the final accumulator
    acc => acc
);

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

public class avg_acc
{
    public int count;
    public double sum;
}

public double ParallelAverage(IEnumerable<double> list)
{
    double avg = list.AsParallel().Aggregate(
        // accumulator factory method, called once per thread:
        () => new avg_acc { count = 0, sum = 0 },
        // update count and sum
        (acc, val) => { acc.count++; acc.sum += val; return acc; },
        // combine accumulators
        (ac1, ac2) => new avg_acc { count = ac1.count + ac2.count, sum = ac1.sum + ac2.sum },
        // calculate average
        acc => acc.sum / acc.count
    );
    return avg;
}

Хотя не так быстро, как стандарт Average расширение (~ в 1,5 раза быстрее, чем последовательное, в 1,6 раза медленнее, чем параллельное), это показывает, как вы можете выполнять довольно сложные операции параллельно, не блокируя выходы или не ожидая других потоков, чтобы прекратить связываться с ними, и как использовать сложный аккумулятор для держать промежуточные результаты.