Plinq gives different results from Linq - what am I doing wrong?

Question

Can anyone tell me what the correct Plinq code is for this? I'm adding up the square root of the absolute value of the sine of each element fo a double array, but the Plinq is giving me the wrong result.

Output from this program is:

Linq aggregate = 75.8310477905274 (correct) Plinq aggregate = 38.0263653589291 (about half what it should be)

I must be doing something wrong, but I can't work out what...

(I'm running this with Visual Studio 2008 on a Core 2 Duo Windows 7 x64 PC.)

Here's the code:

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Collections;

namespace ConsoleApplication1
{
    class Program
    {
        static void Main()
        {
            double[] array = new double[100];

            for (int i = 0; i < array.Length; ++i)
            {
                array[i] = i;
            }

            double sum1 = array.Aggregate((total, current) => total + Math.Sqrt(Math.Abs(Math.Sin(current))));
            Console.WriteLine("Linq aggregate = " + sum1);

            IParallelEnumerable<double> parray = array.AsParallel<double>();
            double sum2 = parray.Aggregate((total, current) => total + Math.Sqrt(Math.Abs(Math.Sin(current))));
            Console.WriteLine("Plinq aggregate = " + sum2);
        }
    }
}

Answer 1

Aggregate works slightly differently in PLINQ.

From MSDN Blogs:

Rather than expecting a value to initialize the accumulator to, the user gives us a factory function that generates the value:

public static double Average(this IEnumerable<int> source)
{
    return source.AsParallel().Aggregate(
        () => new double[2],
        (acc, elem) => { acc[0] += elem; acc[1]++; return acc; },
        (acc1, acc2) => { acc1[0] += acc2[0]; acc1[1] += acc2[1]; return acc1; },
        acc => acc[0] / acc[1]);
}

Now, PLINQ can initialize an independent accumulator for each thread. Now that each thread gets its own accumulator, both the folding function and the accumulator combining function are free to mutate the accumulators. PLINQ guarantees that accumulators will not be accessed concurrently from multiple threads .

So, in your case, you would also need to pass an accumulator function which sums the outputs of the paralleled aggregates (hence why you're seeing a result that is roughly half of what it should be).

Answer 2

Thank you MSDN Blogs. It now seems to be working correctly. I changed my code as follows:

using System;
using System.Linq;

namespace ConsoleApplication1
{
    class Program
    {
        static void Main()
        {
            Test();
        }

        static void Test()
        {
            double[] array = new double[100];

            for (int i = 0; i < array.Length; ++i)
            {
                array[i] = i;
            }

            double sum1 = array.Aggregate((total, current) => total + Math.Sqrt(Math.Abs(Math.Sin(current))));
            Console.WriteLine("Linq aggregate = " + sum1);

            IParallelEnumerable<double> parray = array.AsParallel();

            double sum2 = parray.Aggregate
            (
                0.0,
                (total1, current1) => total1 + Math.Sqrt(Math.Abs(Math.Sin(current1))),
                (total2, current2) => total2 + current2,
                acc => acc
            );

            Console.WriteLine("Plinq aggregate = " + sum2);
        }
    }
}