Parallel.Invoke的执行速度较慢。

Question

我编写了两种比较性能的代码，分别是Methode #1和Methode #2。Methode #1使用for循环，而Methode #2使用Parallel.Invoke。在第二种情况下运行非常缓慢。我不明白为什么会这样发生？

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Collections.Concurrent;
using Mpir.NET;
using System.Runtime.Serialization;
using System.Diagnostics;

namespace ConsoleApplication4
{
    class Program
    {
        public class numbers
        {
            public numbers(mpz_t p, mpz_t q)
            {
                this.q = q;
                this.p = p;
            }
            mpz_t q;
            mpz_t p;
        };

        static void Main(string[] args)
        {
            Int32 arraySize = 400;
            ConcurrentBag<numbers> pairsCB = new ConcurrentBag<numbers>();
            ConcurrentBag<Action> cbTasks = new ConcurrentBag<Action>();
            HashSet<numbers> uniqueCB = new HashSet<numbers>(pairsCB.ToArray());

            mpz_t[] numbersArray = new mpz_t[arraySize];

            for (Int32 i = 0; i < arraySize; i++)
            {
                numbersArray[i] = i*i+i+1; 
            }


            //Methode #1 
            Stopwatch stopwatch1 = new Stopwatch();
            stopwatch1.Start();

            for (Int32 j = 0; j < arraySize; j++)
            {
                checkDivisible(numbersArray[j], pairsCB);
            }

            uniqueCB = new HashSet<numbers>(pairsCB.ToArray());
            stopwatch1.Stop();
            Console.WriteLine("Methode Count Unique Pairs  Count:{0}\tTime elapsed:{1}", uniqueCB.Count(), stopwatch1.Elapsed);

            //Methode #2 
            Stopwatch stopwatch2 = new Stopwatch();
            stopwatch2.Start();
            pairsCB = new ConcurrentBag<numbers>();

            for(Int32 j = 0; j < arraySize; j++)
            {
                mpz_t value = numbersArray[j];
                cbTasks.Add(new Action(() => checkDivisible(value, pairsCB)));
            }

            Action[] tasks =  cbTasks.ToArray();

            Parallel.Invoke(tasks);

            stopwatch2.Stop();
            Console.WriteLine("Methode Count Unique Pairs  Count:{0}\tTime elapsed:{1}", uniqueCB.Count(), stopwatch2.Elapsed);


            Console.ReadKey();
        }

        private static void checkDivisible(mpz_t n, ConcurrentBag<numbers>  pq)
        {
            mpz_t p = 1; 
            mpz_t q = 1;    

            for (Int32 i = 2; i < n; i++)
            {
                if (n % i == 0)
                {
                    q = i;
                    p = n / i;
                    pq.Add(new numbers(p, q));
                }
            }
        }
    }
}

- Yuriy Tigiev

---

这是因为这种方法非常小，创建线程会增加开销。只有在处理长时间进程（文件访问、下载、大量计算等）时才使用并行处理。 - Florian Schmidinger

如果一个任务的持续时间不足0.1秒，将其并行化是没有意义的。调度它所花费的时间比运行它还要长。 - xanatos

ConcurrentBag不是通用的集合，它用于为每个写入其中的线程本地存储数据。无论如何，因为您正在使用Parallel.Invoke，对于任务来说根本不需要并发集合。请使用ConcurrentQueue存储数字。更好的方法是完全避免使用共同的集合，通过从每个调用返回一个numbers数组，并在最后合并它们 - 这本质上就是任何map/reduce算法中的“reduce”步骤。 - Panagiotis Kanavos

Panagiotis Kanavos - 你能提供优化的代码吗？我不知道如何在 Parallel.Invoke 中返回数组并合并数据。我想使用 Parallel.Invoke 来理解它的工作原理。 - Yuriy Tigiev

Answer 1

你也可以使用以下方法：

//Methode #2 
Stopwatch stopwatch2 = new Stopwatch();
stopwatch2.Start();
pairsCB = new ConcurrentBag<numbers>();

Parallel.For(0, arraySize, (index) => 
{
     checkDivisible(numbersArray[index], pairsCB);
});

stopwatch2.Stop();

当arraySize为1000时的输出结果

Methode Count Unique Pairs  Count:3878  Time elapsed:00:00:01.5671572
Methode Count Unique Pairs  Count:3878  Time elapsed:00:00:00.7917211

- Florian Schmidinger

---

@YuriyTigiev 从评论和其他回答中可以看出，在这种情况下，这样做不会带来任何好处。如果你有更大的计算量需要运行，那么可能会有帮助。 - Florian Schmidinger

@YuriyTigiev 在这种情况下调用的方法是错误的（并行处理大量数据），而且在基准测试中也表现出来了。划分源数据非常重要，Parallel.Invoke 并不会做到这一点。 - Panagiotis Kanavos

在实际编程中，我会在checkDivisible函数中使用大量的计算。 - Yuriy Tigiev

@YuriyTigiev Parallel.For可能会更快...我会进行测试和基准测试以确保... - Florian Schmidinger

@YuriyTigiev 正如我们之前提到的，将并行性添加到其中并不是一个真正好的想法... 在我看到性能提升之前，我将数组大小更改为1000。 - Florian Schmidinger

Answer 2

第二种方法速度较慢的原因有很多。

1. checkDivisible 的工作量不足以证明并行化。并行化和同步的开销可能大于任何好处。
2. ConcurrentBag 是一种专门用于将数据存储在线程本地存储中的特殊集合，确保线程可以快速访问其写入的项目。但在其他情况下，它实际上比其他并发集合更慢。
3. 所有对 checkDivisible 的调用都写入同一个集合，这成为热点。最好从每个调用返回一个简单的数组，并在最后一步合并所有数组。
4. 并发调用过多。Parallel.Invoke 必须分别调度每个 Action。Parallel.For 或 Parallel.ForEach 则知道所有调用都相同，因此可以根据处理器数量分割数据，确保并行化开销最小。

第一步是修改 checkDivisible：

    private static List<number> checkDivisible(mpz_t n)
    {
        mpz_t p = 1; 
        mpz_t q = 1;    

        List<number> nums=new List<numbers>();

        for (Int32 i = 2; i < n; i++)
        {
            if (n % i == 0)
            {
                q = i;
                p = n / i;
                nums.Add(new numbers(p, q));
            }
        }
        return numbers;
    }

我建议使用迭代器方法，因为它避免了创建列表来收集结果的过程。

然后你可以使用Parallel.For：

var results=new ConcurrentQueue<IList<numbers>>();

Parallel.For(0, arraySize, (index) => 
{
     var local=checkDivisible(numbersArray[index]);
     results.Add(local);
});

var final=results.SelectMany(r=>r).ToList();

最后一步可以根据你想要的形式进行操作，例如使用ToDictionary或ToLookup按键来合并结果。另一种选择是使用PLINQ以更简洁的方式完成相同的操作。将checkDivision更改为迭代器：

    private static IEnumerable<number> checkDivisible(mpz_t n)
    {
        mpz_t p = 1; 
        mpz_t q = 1;    

        for (Int32 i = 2; i < n; i++)
        {
            if (n % i == 0)
            {
                q = i;
                p = n / i;
                yield return new numbers(p, q);
            }
        }
    }

您可以写成：

您可以编写：

var results= (from n in numbersArray.AsParallel()
             from number in checkDivisible(n)
             select n).ToList();

和 Parallel.For 一样，PLINQ 将根据机器上的核心数将数据在 numbersArray 中进行分区，以并行方式处理分区，最后将它们合并为一个列表。

- Panagiotis Kanavos

---

请问您能提供Parallel.Invoke的正确代码吗？我知道Parallel.For更好，但我想使用Parallel.Invoke。 - Yuriy Tigiev

@YuriyTigiev 为什么？你想解决什么问题？为每个数字运行单独的任务是错误的，但这就是 Invoke 将要做的。你需要重写 Parallel.For 已经完成的工作 - 分割数据，使用 Parallel.Invoke 处理每个分区，最后收集结果。你可以使用 Partitioner 类来实现，或者简单地将数组索引拆分成范围，并将范围（从，到）用作函数的输入。 - Panagiotis Kanavos

我正在测试分解数字的新方法。我希望有机制保存每个任务的状态，并且有载入状态的机制，以便从中断的步骤继续任务。 - Yuriy Tigiev

我需要调用一个特殊函数16384^2次来分解数字。对于这种情况，什么是最优方案？使用Parallel.For、PLINQ还是Tasks？ - Yuriy Tigiev