c# – 任务并行不稳定,有时使用100％CPU

// Using normal foreach
            ConcurrentBag<int> resultData = new ConcurrentBag<int>();
            Stopwatch sw = new Stopwatch();
            sw.Start();
            foreach (var item in testData)
            {
                if (item.Equals(1))
                {
                    resultData.Add(item);
                }
            }
            Console.WriteLine("Normal ForEach " + sw.ElapsedMilliseconds);

            // Using list parallel for
            resultData = new ConcurrentBag<int>();
            sw.Restart();
            System.Threading.Tasks.Parallel.For(0,testData.Count() - 1,(i,loopState) =>
            {
                int data = testData[i];
                if (data.Equals(1))
                {
                    resultData.Add(data);
                }
            });
            Console.WriteLine("List Parallel For " + sw.ElapsedMilliseconds);

            // Using list parallel foreach
            //resultData.Clear();
            resultData = new ConcurrentBag<int>();
            sw.Restart();
            System.Threading.Tasks.Parallel.ForEach(testData,(item,loopState) =>
            {
                if (item.Equals(1))
                {
                    resultData.Add(item);
                }
            });
            Console.WriteLine("List Parallel ForEach " + sw.ElapsedMilliseconds);

            // Using concurrent parallel for 
            ConcurrentStack<int> resultData2 = new ConcurrentStack<int>();
            sw.Restart();
            System.Threading.Tasks.Parallel.For(0,loopState) =>
            {
                int data = testData[i];
                if (data.Equals(1))
                {
                    resultData2.Push(data);
                }
            });
            Console.WriteLine("Concurrent Parallel For " + sw.ElapsedMilliseconds);

            // Using concurrent parallel foreach
            resultData2.Clear();
            sw.Restart();
            System.Threading.Tasks.Parallel.ForEach(testData,loopState) =>
            {
                if (item.Equals(1))
                {
                    resultData2.Push(item);
                }
            });
            Console.WriteLine("Concurrent Parallel ForEach " + sw.ElapsedMilliseconds);

var resultData3 = testData.AsParallel().Where(x => x == 1).ToList();

首先,小心并行 – 它不会保护您免受线程安全问题的影响.在原始代码中,在填充结果列表时使用了非线程安全代码.通常,您希望避免共享任何状态(尽管在这种情况下对列表的只读访问权限很好).如果你真的想使用Parallel.For或Parallel.ForEach进行过滤和聚合(实际上,AsParallel就是你想要的那些情况),你应该使用带有线程局部状态的重载 – 你要做最后的结果聚合localFinally委托(请注意,它仍然在不同的线程上运行,因此您需要确保线程安全;但是,在这种情况下,锁定很好,因为每个线程只执行一次,而不是每次迭代).

现在,在这样的问题中尝试的第一件事就是使用分析器.所以我做到了.结果如下：

>这些解决方案中几乎没有任何内存分配.它们与初始测试数据分配完全相形见绌,即使对于相对较小的测试数据(我在测试时使用了1M,10M和100M的整数).
>正在进行的工作是在Parallel.For或Parallel.ForEach实体本身,不在你的代码中(简单的if(data [i] == 1)results.Add(data [i])).

第一种方式我们可以说GC可能不是罪魁祸首.实际上,它没有任何机会运行.第二个更好奇 – 这意味着在某些情况下,Parallel的开销是不合时宜的 – 但它看起来是随机的,有时它可以毫无障碍地工作,有时需要半秒钟.这通常指向GC,但我们已经排除了这一点.

我已经尝试使用没有循环状态的重载,但这没有帮助.我试过限制MaxDegreeOfParallelism,但它只会伤害事物.现在,很明显,这个代码绝对由高速缓存访??问控制 – 几乎没有CPU工作和没有I / O – 这总是有利于单线程解决方案;但即使使用1的MaxDegreeOfParallelism也无济于事 – 事实上,2似乎是我系统中最快的.更多是无用的 – 再次,缓存访问占主导地位.它仍然很奇怪 – 我正在使用服务器CPU进行测试,它同时为所有数据提供了大量缓存,而我们没有进行100％顺序访问(这几乎完全消除了延迟) ),应该足够顺序.无论如何,我们在单线程解决方案中拥有内存吞吐量的基线,并且当它运行良好时非常接近并行化案例的速度(并行化,我读取的运行时间比单线程少40％)四核服务器CPU用于一个令人尴尬的并行问题 – 显然,内存访问是极限.

因此,是时候检查Parallel.For的参考源了.在这种情况下,它只是根据工人数量创建范围 – 每个范围一个范围.所以这不是范围 – 没有开销.
核心只是运行一个迭代给定范围的任务.有一些有趣的东西 – 例如,如果花费太长时间,任务将被“暂停”.但是,它似乎不太适合数据 – 为什么这样的事情会导致与数据大小无关的随机延迟？无论工作量多么小,无论MaxDegreeOfParallelism有多低,我们都会“随机”减速.这可能是一个问题,但我不知道如何检查它.

最有趣的是扩展测试数据与异常无关 – 虽然它使“好”并行运行得更快(甚至在我的测试中接近完美效率,奇怪的是),“坏”仍然只是一样糟糕.事实上,在我的一些测试中,它们非常糟糕(高达“正常”循环的十倍).

那么,让我们来看看线程.我巧妙地增加了ThreadPool中的线程数量,以确保扩展线程池不是瓶颈(如果一切运行良好,它不应该……).这是第一个惊喜 – 虽然“好”运行只使用4-8个有意义的线程,但“坏”运行会扩展到池中所有可用线程,即使它们有一百个.哎呀？

让我们再次深入研究源代码. Parallel内部使用Task.RunSynchronously运行根分区工作作业,并等待结果.当我查看并行堆栈时,有97个线程执行循环体,并且只有一个实际上在堆栈上具有RunSynchronously(正如预期的那样 – 这是主线程).其他是普通的线程池线程.任务ID也讲述了一个故事 – 在进行迭代时会创建数千个单独的任务.显然,这里有些不对劲.即使我移除整个循环体,这仍然会发生,所以它也不是一些封闭的怪异.

显式设置MaxDegreeOfParallelism有点抵消 – 使用的线程数量不会再爆炸 – 但是,任务量仍然有效.但是我们已经看到范围只是运行并行任务的数量 – 那么为什么要继续创建越来越多的任务呢？使用调试器确认这一点 – MaxDOP为4,只有五个范围(有一些对齐导致第五个范围).有趣的是,其中一个已完成的范围(第一个如何在其余范围之前完成？)的索引高于它迭代的范围 – 这是因为“调度程序”在最多16个切片中分配范围分区.

根任务是自我复制的,因此不是明确地启动,例如处理数据的四个任务,它等待调度程序复制任务以处理更多数据.这有点难以阅读 – 我们谈论的是复杂的多线程无锁代码,但它似乎总是将分配范围内的工作分配得比分区范围小得多.在我的测试中,切片的最大尺寸为16 – 与我正在运行的数百万个数据相差甚远.像这样的16次迭代根本没有时间,这可能会导致算法出现许多问题(最大的问题是基础设施占用的CPU工作量比实际的迭代器主体多).在某些情况下,缓存垃圾可能会进一步影响性能(可能在正文运行时存在很多变化时),但大多数情况下,访问是连续的.

TL; DR

如果您的每次迭代工作非常短(大约几毫秒),请不要使用Parallel.For和Parallel.ForEach. AsParallel或只运行迭代单线程将更快.

稍微长一点的解释：

似乎Parallel.For和Paraller.ForEach是针对你正在迭代的各个项目需要花费大量时间来执行的场景(即每个项目的大量工作,而不是很多项目的大量工作) .当迭代器体太短时,它们似乎表现不佳.如果您没有在迭代器主体中进行大量工作,请使用AsParallel而不是Parallel.*.甜点似乎在每片150ms以下(每次迭代大约10ms).否则,Parallel.*将花费大量时间在自己的代码中,并且几乎没有时间进行迭代(在我的情况下,通常的数字在体内约为5-10％ – 非常糟糕).

可悲的是,我没有在MSDN上发现任何有关此问题的警告 – 甚至有大量数据的样本,但没有暗示这样做的可怕性能损失.在我的计算机上测试相同的示例代码,我发现它确实经常比单线程迭代慢,并且在最好的时候,几乎更快(在四个CPU内核上运行时节省大约30-40％的时间) – 不是很有效率).

编辑：

Willaien在MSDN上发现了关于这个问题的提及,以及如何解决它 – https://msdn.microsoft.com/en-us/library/dd560853(v=vs.110).aspx.想法是使用自定义分区器并在Parallel.For主体中迭代它(例如,在Parallel.For循环中循环).但是,对于大多数情况,使用AsParallel可能仍然是一个更好的选择 – 简单的循环体通常意味着某种map / reduce操作,而AsParallel和LINQ通常都很棒.例如,您的示例代码可以简单地重写为：

var result = testData.AsParallel().Where(i => i == 1).ToList();

使用AsParallel的唯一情况是一个坏主意与所有其他LINQ相同 – 当你的循环体有副作用时.有些可能是可以忍受的,但完全避免它们会更安全.