标签: parallel.foreach

我正在使用Parallel.ForEach多个线程进行工作,每次迭代都使用一个新的EF5 DbContext,所有这些都包含在TransactionScope中,如下所示:

using (var transaction = new TransactionScope())
{
    int[] supplierIds;

    using (var appContext = new AppContext())
    {
        supplierIds = appContext.Suppliers.Select(s => s.Id).ToArray();
    }

    Parallel.ForEach(
        supplierIds,
        supplierId =>
    {
        using (var appContext = new AppContext())
        {
            Do some work...

            appContext.SaveChanges();                
        }
    });

    transaction.Complete();
}

运行几分钟后,它会抛出一个EntityException"底层提供程序在Open上失败",其中包含以下内部细节:

"此时SQL Server数据库引擎的实例无法获取LOCK资源.当活动用户较少时,请重新运行语句.请数据库管理员检查此实例的锁定和内存配置,或检查长时间运行交易."

有谁知道导致这种情况的原因或如何预防？谢谢.

今天我们将新创建的ASP.NET应用程序部署到服务器,很快我们意识到存在一个与安全相关的奇怪问题导致应用程序崩溃.这是一个内部应用程序,我们使用Impersonation来管理用户访问资源的方式.但是,当用户尝试访问他们完全控制的文件夹时,应用程序会抛出"拒绝访问"异常.

事实上是一个异常,AggregateException并且被抛入一个方法,该方法Parallel.ForEach用于枚举列表和正文内部,它尝试访问该文件夹,但此时模拟上下文被更改,工作线程作为应用程序池运行身份,无权访问该文件夹,因此异常.

为了证实这一点,我查看了身体之前和之内的过程身份Parallel.ForEach:

string before = WindowsIdentity.GetCurrent().Name;
Debug.WriteLine("Before Loop: {0}", before);

Parallel.ForEach(myList, currentItem =>
{
    string inside = WindowsIdentity.GetCurrent().Name;
    Debug.WriteLine("Inside Loop: {0} (Worker Thread {1})", inside, Thread.CurrentThread.ManagedThreadId);
});

当我运行应用程序时,这就是打印出来的内容:

Before Loop: MyDomain\ImpersonatedUser

Inside Loop: NT AUTHORITY\SYSTEM (Worker Thread 8)
Inside Loop: MyDomain\ImpersonatedUser (Worker Thread 6)
Inside Loop: MyDomain\ImpersonatedUser (Worker Thread 7)
Inside Loop: NT AUTHORITY\SYSTEM (Worker Thread 9)
Inside Loop: NT AUTHORITY\SYSTEM (Worker Thread 10)
Inside Loop: MyDomain\ImpersonatedUser (Worker Thread 7)
Inside Loop: MyDomain\ImpersonatedUser …

我已经阅读了SO中的所有相关问题,但对于我的场景中触发多个Web服务调用的最佳方法有点困惑.

我有一个聚合器服务,它接受输入,解析并将其转换为多个Web请求,进行Web请求调用(不相关,因此可以并行触发)并合并发送回调用者的响应.现在使用以下代码 -

list.ForEach((object obj) =>
{
     tasks.Add(Task.Factory.StartNew((object state) => 
     {
           this.ProcessRequest(obj);
     }, obj, CancellationToken.None, TaskCreationOptions.AttachedToParent, TaskScheduler.Default));
});
await Task.WhenAll(tasks);

在await Task.WhenAll(tasks)来自斯科特Hanselman的帖子在那里说,

斯蒂芬说:"从可伸缩性的角度来看,更好的解决方案是利用异步I/O.当你通过网络呼叫时,没有理由(除了方便之外)在等待响应来阻止线程时背部"

现有代码似乎消耗了太多线程,并且处理器时间在生产负载上高达100%,这让我思考.

另一个替代方法是使用Parallel.ForEach,它使用分区器,但也"阻塞"调用,这对我的场景来说很好.

考虑到这是所有"异步IO"工作而不是"CPU绑定"工作,并且Web请求不会长时间运行(最多返回3秒),我倾向于认为现有代码足够好.但是这会提供比Parallel.ForEach更好的吞吐量吗？Parallel.ForEach可能使用"最小"数量的任务,因为分区因此最佳使用线程(？).我用一些本地测试测试了Parallel.ForEach,它似乎没有更好.

目标是减少CPU时间并提高吞吐量,从而提高可扩展性.是否有更好的方法来并行处理Web请求？

感谢任何投入.

编辑: 代码示例中显示的ProcessRequest方法确实使用HttpClient及其异步方法来触发请求(PostAsync,GetAsync,PutAsync).

有一个C#函数A(arg1, arg2)需要多次调用.为了做到这一点,我正在使用并行编程.

以下面的代码为例:

long totalCalls = 2000000;
int threads = Environment.ProcessorCount;

ParallelOptions options = new ParallelOptions(); 
options.MaxDegreeOfParallelism = threads;

Parallel.ForEach(Enumerable.Range(1, threads), options, range =>
{
    for (int i = 0; i < total / threads; i++)
    {
        // init arg1 and arg2
        var value = A(arg1, agr2);
        // do something with value
    }
});

现在的问题是,随着核心数量的增加,这不会扩大规模; 例如,在8个内核上,它使用80%的CPU,在16个内核上,它使用40-50%的CPU.我想最大限度地使用CPU.

您可能认为A(arg1, arg2)内部包含复杂的计算,但它没有任何IO或网络绑定操作,也没有线程锁定.有哪些其他可能性可以找出代码的哪一部分使其不能以100%并行的方式执行？

我也试过提高并行度,例如

int threads = Environment.ProcessorCount * 2;
// AND
int threads = Environment.ProcessorCount * 4;
// etc.

但它没有任何帮助. …

好的,这是基本背景.该程序连接到outlook/exchange并解析所有邮件消息以查看哪些是加密的.我想做的一件事是使用多线程来减少扫描消息所需的时间.

目前代码如下所示:

foreach (Object item in folder.Items) {
//Checks for encryption and gets needed info and updates count
}

我想改用Parallel.ForEach函数.我想知道如何设置它.我尝试将表达式设置为现在的样式,但是我得到一个错误,指出Object类型被用作变量.任何有关这方面的帮助将不胜感激.

好的,我给出的布局似乎是正确的.代码现在看起来像这样:

Parallel.ForEach(folder.Items, item =>
{
//does stuff
});

我现在收到以下错误:

错误15无法从用法中推断出方法System.Threading.Tasks.Parallel.ForEach(System.Collections.Concurrent.OrderablePartitioner,System.Action)的类型参数.尝试显式指定类型参数.

有任何想法吗？感谢您的帮助,非常感谢.

好的,我找到了这个网站:http://blogs.msdn.com/b/pfxteam/archive/2010/03/02/9971804.aspx,它给了我错误所需的答案.我只需要通过制作一个转换函数将集合更改为通用集合.

static IEnumerable<object> Cast(IEnumerable source)
{
    foreach (object o in source)
        yield return o;
}

然后调整原件

Parallel.ForEach(Cast(folder.Items), item =>
{
//does stuff
});

现在它运行没有错误.欢呼.

我试图允许取消Parallel.ForEach循环.根据这篇MSDN文章,这是可能的,我正在关注他们的编码.

// Tokens for cancellation 
ParallelOptions po = new ParallelOptions();
po.CancellationToken = cts.Token;

try
{
    Parallel.ForEach(queries, po, (currentQuery) =>
    {
        // Execute query
        ExecuteQuery(currentQuery);

        // Throw exception if cancelled 
        po.CancellationToken.ThrowIfCancellationRequested(); // ***
    }); 
}
catch (OperationCanceledException cancelException)
{
    Console.WriteLine(cancelException.Message);
}

但是,当我cts.Cancel();从用户可访问的函数调用时,应用程序在标有上述星号的行上崩溃并出现错误:

System.OperationCanceledException was unhandled by user code
  Message=The operation was canceled.
  Source=mscorlib
  StackTrace:
   at System.Threading.CancellationToken.ThrowIfCancellationRequested()
   at CraigslistReader.SearchObject.<>c__DisplayClass7.<bw_DoWork>b__5(Query currentQuery) in {PATH}:line 286
   at System.Threading.Tasks.Parallel.<>c__DisplayClass2d`2.<ForEachWorker>b__23(Int32 i)
   at System.Threading.Tasks.Parallel.<>c__DisplayClassf`1.<ForWorker>b__c()
InnerException: 

我有异常处理程序,所以我不明白崩溃.有任何想法吗？

我在运行64位Windows Server 2008 R2 Enterprise的服务器上测试一个程序,它有4个Intel E7-4870处理器,总共40个内核和80个可用线程(我可以在Windows任务管理器中看到80个CPU使用率图表) ).

程序代码如下:

numlist是List包含数百个数字,每个数字都是在某些计算中使用的参数

Parallel.ForEach(numlist, num =>
                 {
                    // do some calculation using parameter = num             
                 });

问题是当我在服务器上运行这个程序时,只有一半的可用线程显示在Windows任务管理器中使用(当然CPU使用率显示为50%),剩下的40个都是完全未使用和空闲.

我还在另一台服务器上测试了相同的程序,该服务器只有2个处理器和24个可用线程,所有24个线程将被完全使用,CPU使用率显示为100%.

有没有什么办法可以让40核CPU服务器运行这个程序并充分利用它所有的80个线程(或接近80个线程)？当仅使用50%的CPU资源时,性能不够好.

这是我正在测试的完整程序代码:

namespace Test
{
    internal class Program
    {
        private static void Main(string[] args)
        {
            Console.WriteLine("Press any key to start");
            Console.ReadLine();
            List<int> numlist = new List<int>();
            for (int i = 0; i < 100; i++)
            {
                numlist.Add(i);
            }

            Parallel.ForEach(numlist, num =>
                                 {
                                 while (true)
                                 {
                                     num++;
                                 }
                             });

        }
    }
}

当它在具有2个Intel X5690处理器的服务器上运行时(共有24个线程可用),所有24个线程都被使用,CPU使用率显示为100%;

但是当我在具有80个可用线程的4处理器服务器上运行它时,只使用了40个线程,CPU使用率仅为50%.是否有与此相关的编译器设置？

我正在使用Parallel.ForEach它，它极大地提高了我的代码的性能，但我对DbContext多线程感到好奇。我知道它不是线程安全的，所以我在需要的地方使用锁。

循环遍历字典并计算统计信息：

Dictionary<string, List<decimal>> decimalStats = new Dictionary<string, List<decimal>>(); // this gets populated in another irrelevant loop

List<ComparativeStatistic> comparativeStats = db.ComparativeStatistics.ToList();
var statLock = new object();

Parallel.ForEach(decimalStats, entry =>
{
    List<decimal> vals = ((List<decimal>)entry.Value).ToList();

    if (vals.Count > 0)
    {
        string[] ids = entry.Key.Split('#');
        int questionId = int.Parse(ids[0]);
        int yearId = int.Parse(ids[1]);
        int adjacentYearId = int.Parse(ids[2]);

        var stat = comparativeStats.Where(l => l.QuestionID == questionId && l.YearID == yearId && l.AdjacentYearID == adjacentYearId).FirstOrDefault();

        if (stat == null)
        { …

考虑下面的代码,是不是OK做async/await里面Parallel.ForEach？

例如.

Parallel.ForEach(names, name =>
{
    // Do some stuff...

    var foo = await GetStuffFrom3rdPartyAsync(name);

    // Do some more stuff, with the foo.
});

还是有一些我需要知道的问题？

编辑:不知道这是否编译,顺便说一句.只是Pseduo代码..大声思考.

在代码的并行部分中，我将每个线程的结果保存到ConcurrentBag中。但是，完成此操作后，我需要遍历每个结果并通过我的评估算法运行它们。普通的foreach实际上会遍历所有成员，还是我需要特殊的代码？我还考虑过使用队列之类的东西来代替行李，但我不知道哪种方法最好。在并行代码的末尾，袋子通常只包含20个左右的项目。

即，实际上将为ConcurrentBag的所有成员访问并运行foreach吗？

ConcurrentBag futures = new ConcurrentBag();
foreach(move in futures)
{
 // stuff
}