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

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

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

Этот вопрос C++ также говорит об этом.

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

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

double sum = list.Sum();

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

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

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

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

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

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

Существует несколько перегрузок расширения Aggregate, в том числе одна, которая фактически является шаблоном карты. реализация, схожая с тем, как работают системы больших данных, такие как 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
);

Для простых агрегаций это работает нормально. Для более сложного агрегата, использующего нетривиальный аккумулятор, существует variant, который принимает функцию, создающую аккумуляторы для начального состояния. Это полезно, например, в реализации 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 раза медленнее, чем параллельное), оно показывает, как вы можете выполнять довольно сложные операции параллельно, не блокируя выходные данные или не ожидая, пока другие потоки перестанут возиться с ними. и как использовать сложный аккумулятор для хранения промежуточных результатов.