标签: parallel.foreach

由于逐行反射相当昂贵,我一直在寻找构建和插入实体的更快的替代方案.我做了一些关于这个主题的研究,并且还发现了一些性能比较,这似乎表明表达树是要走的路.我将如何重写以下功能以利用此功能？

    public static void InsertTable(IEnumerable<DataTable> chunkedTable)
    {

        Parallel.ForEach(
            chunkedTable,
            new ParallelOptions
            {
                MaxDegreeOfParallelism = Convert.ToInt32(ConfigurationManager.AppSettings["MaxThreads"])
            },
            chunk =>
            {
                Realty_Records_ProdEntities entities = null;
                try
                {
                    entities = new Realty_Records_ProdEntities();
                    entities.Configuration.AutoDetectChangesEnabled = false;

                    foreach (DataRow dr in chunk.Rows)
                    {
                        var parcelToInsert = new Parcel();

                        foreach (DataColumn c in dr.Table.Columns)
                        {
                            var propertyInfo = parcelToInsert.GetType()
                                .GetProperty(
                                    c.ColumnName,
                                    BindingFlags.SetProperty | BindingFlags.IgnoreCase
                                    | BindingFlags.Public | BindingFlags.Instance);

                            propertyInfo?.SetValue(
                                parcelToInsert,
                                TaxDataFunction.ChangeType(
                                    dr[c.ColumnName],
                                    propertyInfo.PropertyType),
                                null);
                        }
                        entities.Parcels.Add(parcelToInsert);
                    }
                    entities.SaveChanges();
                }
                catch (Exception ex) …

我对async/await如何作为并行工作感到困惑所以我在这里制作了一个测试代码:我尝试发送6个用模拟模拟的任务.每个任务都将执行3个其他子任务.

你可以复制/粘贴测试

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

namespace ConsoleApplication1
{
    class Program
    {
         static void Main(string[] args)
        {
            //job simulation 
            Func<int, string, Tuple<int, string>> tc = Tuple.Create;
            var input = new List<Tuple<int, string>>{
                  tc( 6000, "task 1" ),
                  tc( 5000, "task 2" ),
                  tc( 1000, "task 3" ),
                  tc( 1000, "task 4" ),
                  tc( 1000, "task 5" ),
                  tc( 1000, "task 6" )
            };

            List<Tuple<int, string>> JobsList = new List<Tuple<int, string>>(input);

            //paralelism atempt …

我有一个Parallel.Foreach循环让我感到悲伤，想看看你们中的一个人是否可以解释这一点。不幸的是，在谷歌上几乎找不到这个。

这是结：

我的 foreach 循环：

string [] Ids = {........}; //a string array of ID's
using(IUnitOfWork uw = GetUnitOfWork())
     {
            Parallel.ForEach(Ids, currentRecord =>
            {
                var x = (from h in uw.GetRepository<EFEntity1>().AsQueryable()
                         join k in uw.GetRepository<EFEntity2>().AsQueryable()
                         on h.ID equals k.ID
                         join l in uw.GetRepository<EFEntity3>().AsQueryable() on 
                          h.FundAccount equals l.FundAccount
                         where h.ID == currentRecord
                         select new { h.x, h.y, h.z});
                foreach (var v in x)
                {
                    if (v.SomeMember == "foo")
                    {

                    }
                    Console.WriteLine("Output : {0} {1} {2} {3} {4} ", v.x, …

我们构建了这个应用程序，需要在远程计算机（实际上是 MatLab 服务器）上完成一些计算。我们使用 Web 服务连接到 MatLab 服务器并执行计算。

为了加快速度，我们使用了Parallel.ForEach()同时进行多个服务调用的方法。如果我们非常保守地将ParallelOptions.MaxDegreeOfParallelism(DOP) 设置为 4 或其他值，那么一切都会运行良好。然而，如果我们让框架决定 DOP，它将产生如此多的线程，从而迫使远程计算机屈服并开始发生超时（> 10 分钟）。

我们该如何解决这个问题呢？我希望能够做的是利用响应时间来限制调用。如果响应时间小于 30 秒，则继续添加线程，一旦超过 30 秒，就减少使用。有什么建议么？

注意与此问题中的响应相关：Parallel Foreach webservice call。

我正在寻找一种“整洁”且有效的方法来实现长步骤 1（可以并行化）和步骤 2 的组合，步骤 2 需要按原始顺序（如果可能的话，尽量减少来自第一步保存在 RAM 中）同时允许第二步在第一个对象的步骤 1 中的数据可用时立即开始，并与步骤 2 一起提供更多数据。

为了更详细地说明这一点，我需要压缩大量图像（慢速 - 第 1 步），然后通过网络连接按顺序发送每个图像（第 2 步）。在任何阶段限制 RAM 中准备好的压缩数据块的数量也很重要，例如，如果发送 1000 张图像，我想将“已完成”但未发送的图像数量限制为（例如）线程数/使用的处理器。

我已经完成了这个的“手写”版本，使用了一组 Task 对象，但它看起来很混乱，而且我相信其他人一定有类似的需求，所以有没有更“标准”的方法来做到这一点? 理想情况下，我希望有 2 个代表的 Parallel.ForEach 变体 - 一个用于第 1 步，一个用于第 2 步，我希望标准覆盖之一（例如包含“localFinal”参数的覆盖）可能有所帮助，但在原来这些最后阶段是“每个线程”，而不是“每个委托”。

任何人都可以指出我现有的巧妙方法来实现这一目标吗？

我一直在试验 Parallel.For。特别是支持线程本地数据的重载，例如

public static System.Threading.Tasks.ParallelLoopResult For (long fromInclusive, long toExclusive, System.Threading.Tasks.ParallelOptions parallelOptions, Func localInit, Func body, Action localFinally);

根据文档

为参与循环执行的每个线程调用一次 localInit 委托

但是我认为我下面的例子与它相矛盾

var threads = new ConcurrentBag<int>();
ValueTuple LocalInit()
{
    threads.Add(Thread.CurrentThread.ManagedThreadId);
    return new System.ValueTuple();
}
ValueTuple Body(long i, ParallelLoopState _, ValueTuple state) 
{
    Thread.Sleep(100);
    return state;
}
void LocalFinally(ValueTuple state) { };

Parallel.For(0L, 1000L, new ParallelOptions(), LocalInit, Body, LocalFinally);

Console.WriteLine($"{threads.Count} inits between {threads.Distinct().Count()} threads");

它打印一条消息，例如

13 个线程之间的 79 个初始化

这是怎么回事？

比方说，我编写了一个程序，该程序应该读取给定的所有文本文件并从中生成对象列表。

所以步骤是

1. 从磁盘读取给定文件夹中所有文件的文件内容。
2. 每个文件内容创建唯一的对象。

我想知道 .NET 中的 Parallel.ForEach （或任何其他并行结构）是否会提高步骤 1 的性能以及如何提高。磁盘的IO本质上不是同步的吗即磁盘的磁头不能同时在5个地方。事实上，这可能会让事情变得更慢？

您对此有何看法？

在.Net进程中，只有一个托管线程池。我们可以根据需要通过公共属性设置最小和最大线程数。

在.Net 中，我们还可以Parallel.ForEach从后台的托管线程池中获取线程。

我们Parallel.ForEach还可以设置MaxDegreeOfParallelism限制最大线程数。

我有两个Parallel.ForEach并行运行。一个MaxDegreeOfParallelism设置为 3，另一个设置为 7。

我的问题是：我的两个Parallel.ForEach循环是否在幕后使用相同的线程池。如果是，如何Parallel.ForEach限制线程MaxDegreeOfParallelism。多个 Parallel.ForEach循环和一个托管线程池如何协同工作？如果您能在我深入了解 .net core 源代码之前提供高级解释或一些指示，那将会非常有帮助。

在 microsoft 文档的这篇文章中，在方法中 Parallel.For 的示例中，有一个 Break 调用并按以下方式处理 ShouldExitCurrentIteration 和 LowestBreakIteration 等属性：

    if (state.ShouldExitCurrentIteration)
    {
       if (state.LowestBreakIteration < i)
       return;
    }

LowestBreakIteration 存储调用 Break 方法的最小迭代次数。此外，此属性可以存储内部生成的索引的值，例如在方法 Parallel.ForEach 的情况下（来自 microsoft文档）

问题。在 Parallel.ForEach 的情况下，我应该如何处理 LowestBreakIteration 属性，我应该如何使用它以及我应该与什么进行比较？

我轻松地使用 Break for Parallel.For 重复了该示例，但我不知道如何在使用 Break for Parallel.ForEach 的示例中使用属性 LowestBreakIteration。

using System;
using System.Collections.Generic;
using System.Threading.Tasks;

namespace ConsoleApp1
{
    class Program
    {
        static void Main(string[] args)
        {
            //Parallel.For
            void MyMethodForParallelForBreak(int i, ParallelLoopState MyParallelLoopState)
            {
                Console.WriteLine($"Start {i}");

                if (MyParallelLoopState.ShouldExitCurrentIteration)
                {
                    if (MyParallelLoopState.LowestBreakIteration < …

这是出于好奇而提出的一般性问题。我正在使用该doParallel包进行并行计算。我使用这些包来进行模拟。

我观察到，当我使用foreach循环进行模拟时，Rstudio 中的当前使用内存急剧上升 (4+GiB)，并且 Rstudio 有时崩溃。

现在我再次parallel::mclapply进行了相同的模拟，但令人惊讶的是没有问题，并且当前使用内存没有增加太多（10+MiB）。

我不明白代码内部发生了什么。我期待对上述过程的详细解释。

sessionInfo()因为我的 R 是

R version 4.2.1 (2022-06-23) -- "Funny-Looking Kid"
Copyright (C) 2022 The R Foundation for Statistical Computing
Platform: aarch64-apple-darwin20 (64-bit)

操作系统是MacOS。

doParallel软件包版本 1.0.17。

RStudio 版本 2023.03.01。

例子：

假设我们正在尝试计算 Erdos-Renyi 图的边数。我试图每次模拟图形并存储每次模拟的边计数值。

代码如下

#ER random graph generator
src1 <- {"#include <Rcpp.h>
using namespace Rcpp;
// [[Rcpp::export]]
NumericMatrix ER_AdjMatGEN_cpp(int N, double p){
  NumericMatrix temp(N,N);
  for(int i=0; i< N; i++){
    for(int j=0; j < i; j++){ …