标签: parallel.foreach

我何时应该使用async/await,何时应该在C#中使用parallel.foreach？并行和异步/等待是否有同样的目的？它们有什么不同？

我有一些代码可以在我自己的R-like C#DataFrame类中处理数百万个数据行.有许多Parallel.ForEach调用并行迭代数据行.这段代码使用VS2013和.NET 4.5运行了一年多没有问题.

我有两台开发机器(A和B),最近升级了机器A到VS2015.大约一半时间我开始注意到我的代码中出现了一个奇怪的间歇性冻结.让它运行很长一段时间,事实证明代码最终会完成.它只需要15-120分钟而不是1-2分钟.

使用VS2015调试器尝试全部中断由于某种原因而失败.所以我插入了一堆日志语句.事实证明,在Parallel.ForEach循环期间存在Gen2集合时会发生此冻结(比较每个Parallel.ForEach循环之前和之后的集合计数).整个额外的13-118分钟用于任何Parallel.ForEach循环调用恰好与Gen2集合(如果有的话)重叠.如果在任何Parallel.ForEach循环期间没有Gen2集合(大约50%的时间运行它),那么一切都会在1-2分钟内完成.

当我在机器A上的VS2013中运行相同的代码时,我得到了相同的冻结.当我在机器B(从未升级过)的VS2013中运行代码时,它运行得很好.它一夜之间没有结冰了几十次.

我注意到/试过的一些事情:

• 无论是否在机器A上附加了调试器,都会发生冻结(我认为它首先是VS2015调试器的东西)
• 无论我是在构建调试还是在发布模式下,都会发生冻结
• 如果我的目标是.NET 4.5或.NET 4.6,就会发生冻结
• 我试过禁用RyuJIT,但这并没有影响冻结

我根本没有改变默认的GC设置.根据GCSettings,所有运行都发生在LatencyMode Interactive和IsServerGC为假.

我可以在每次调用Parallel.ForEach之前切换到LowLatency,但我真的更愿意了解正在发生的事情.

在VS2015升级之后,有没有其他人在Parallel.ForEach中看到过奇怪的冻结？关于下一步会有什么好处的任何想法？

更新1:在上面的模糊解释中添加一些示例代码...

以下是一些示例代码,我希望能够证明这个问题.此代码在B机器上运行10-12秒,始终如一.它遇到了许多Gen2系列,但它们几乎没有时间.如果我取消注释两个GC设置行,我可以强制它没有Gen2集合.它比30-50秒慢一些.

现在在我的A机器上,代码需要一段随机的时间.似乎是5到30分钟.它似乎变得更糟,它遇到的Gen2系列越多.如果我取消注释两个GC设置行,则机器A也需要30-50秒(与机器B相同).

可能需要对行数和数组大小进行一些调整才能显示在另一台机器上.

using System;
using System.Collections;
using System.Collections.Generic;
using System.IO;
using System.Diagnostics;
using System.Threading;
using System.Threading.Tasks;
using System.Linq;
using System.Runtime;    

public class MyDataRow
{
    public int Id { get; set; }
    public double Value { get; set; }
    public double DerivedValuesSum { get; set; }
    public double[] DerivedValues { get; set; }
}

class Program
{
    static void Example()
    {
        const …

这是代码:

using (var context = new AventureWorksDataContext())
{
    IEnumerable<Customer> _customerQuery = from c in context.Customers
                                           where c.FirstName.StartsWith("A")
                                           select c;

    var watch = new Stopwatch();
    watch.Start();

    var result = Parallel.ForEach(_customerQuery, c => Console.WriteLine(c.FirstName));

    watch.Stop();
    Debug.WriteLine(watch.ElapsedMilliseconds);

    watch = new Stopwatch();
    watch.Start();

    foreach (var customer in _customerQuery)
    {
        Console.WriteLine(customer.FirstName);
    }

    watch.Stop();
    Debug.WriteLine(watch.ElapsedMilliseconds);
}

问题是,Parallel.ForEach与常规相比需要大约400ms ,大约foreach需要40ms.我究竟做错了什么,为什么这不能像我期望的那样工作？

我想用这样的并行循环处理一些东西:

public void FillLogs(IEnumerable<IComputer> computers)
{
    Parallel.ForEach(computers, cpt=>
    {
        cpt.Logs = cpt.GetRawLogs().ToList();
    });

}

好的,它工作正常.但是如果我想让FillLogs方法返回一个IEnumerable怎么办？

public IEnumerable<IComputer> FillLogs(IEnumerable<IComputer> computers)
{
    Parallel.ForEach(computers, cpt=>
    {
        cpt.Logs = cpt.GetRawLogs().ToList();
        yield return cpt // KO, don't work
    });

}

编辑

似乎不可能......但我使用这样的东西:

public IEnumerable<IComputer> FillLogs(IEnumerable<IComputer> computers)
{
    return computers.AsParallel().Select(cpt => cpt);
}

但是我把cpt.Logs = cpt.GetRawLogs().ToList();指示放在哪里

在使用时,Parallel.ForEach我们可以选择定义Parallel选项并设置Max Degree of Parallelism,如:

Parallel.ForEach(values, new ParallelOptions {MaxDegreeOfParallelism = number}, value = > {
    // Do Work
})

但在做PLINQ时:

Tabel.AsEnumberable()
     .AsParallel()
     .Where(//Logic)

我无法找到一种方法来设置MaxDegreeOfParallelism.我也在网上抬头,但没有找到任何东西.有没有人找到解决方法呢？任何帮助表示赞赏.

以下两种方法之间的主要区别是什么:

ThreadPool.QueueUserWorkItem

    Clients objClient = new Clients();
    List<Clients> objClientList = Clients.GetClientList();

    foreach (var list in objClientList)
    {
        ThreadPool.QueueUserWorkItem(new WaitCallback(SendFilesToClient), list);
    } 

System.Threading.Tasks.Parallel ForEach

    Clients objClient = new Clients();
    List<Clients> objClientList = Clients.GetClientList();

    Parallel.ForEach<Clients>(objClientList, list =>
    {
        SendFilesToClient(list);
    });

我是多线程新手,想知道在每种情况下会发生什么(在执行过程方面)每种方法的多线程水平是多少？帮助我想象这两个过程.

SendFilesToClient:从数据库获取数据,转换为Excel并将Excel文件发送到相应的客户端.

谢谢!

我将在我的应用程序上运行许多任务.由于某种原因,每组任务都在运行.我想命名这些任务,所以当我观看Parallel Tasks窗口时,我可以轻松识别它们.

从另一个角度来看,考虑我在框架级别使用任务来填充列表.使用我的框架的开发人员也在为她的工作使用任务.如果她查看并行任务窗口,她会发现一些不知道的任务.我想命名任务,以便她可以将框架任务与她的任务区分开来.

如果有这样的API会很方便:

var task = new Task(action, "Growth calculation task")

或者可能:

var task = Task.Factory.StartNew(action, "Populating the datagrid")

甚至在工作时 Parallel.ForEach

Parallel.ForEach(list, action, "Salary Calculation Task"

可以命名任务吗？

是否有可能得到Parallel.ForEach一个命名结构(也许使用Lambda),所以它创造了这样的命名任务？

我错过了某个地方的API吗？

我还尝试使用继承的任务来覆盖它的ToString().但不幸的是,Parallel Tasks窗口不使用ToString()!

class NamedTask : Task
{
    private string TaskName { get; set; }
    public NamedTask(Action action, string taskName):base(action)
    {
        TaskName = taskName;
    }

    public override string ToString()
    {
        return TaskName;
    }
}

有没有一种简单的方法可以逐步完成parallel.foreach？使用断点调试此方法的最佳方法是什么？

在任何情况下是否有理由选择 Parallel.ForEach 而不是 wait ForEachAsync（反之亦然）？或者它们实际上是相同的？

await collection.ForEachAsync( m => { m.DoSomething(); } );

VS

Parallel.ForEach( collection, m => { m.DoSomething(); } );

在 C# 中，我对停止循环感兴趣Parallel.ForEachAsync（考虑和之间的差异StopBreak）；因为Parallel.ForEach我可以执行以下操作：

Parallel.ForEach(items, (item, state) =>
{
    if (cancellationToken.IsCancellationRequested)
    {
        state.Stop();
        return;
    }

    // some process on the item
    Process(item);
});

但是，由于我有一个需要异步执行的进程，所以我切换到了Parallel.ForEachAsync. ForEachAsync没有该方法Stop()，我可以按break如下方式循环，但我想知道这是否是打破循环的最有效方法（换句话说，循环在收到取消时需要尽快停止要求）。

await Parallel.ForEachAsync(items, async (item, state) =>
{
    if (cancellationToken.IsCancellationRequested)
    {
        return;
    }

    // some async process on the item
    await ProcessAsync(item);
});