标签: parallel.foreach

什么是最有效的暂停和停止(在它结束之前)parallel.foreach的方法？

Parallel.ForEach(list, (item) =>
{
    doStuff(item);
});

我的应用程序中有一个有趣的死锁问题.有一个内存数据存储,它使用ReaderWriterLockSlim来同步读写.其中一种读取方法使用Parallel.ForEach在给定一组过滤器的情况下搜索商店.其中一个过滤器可能需要对同一商店进行恒定时间读取.这是产生死锁的场景:

更新:下面的示例代码.实际方法调用更新步骤
鉴于单一实例store的ConcreteStoreThatExtendsGenericStore

1. Thread1获取商店的读锁定 - store.Search(someCriteria)
2. Thread2尝试使用写锁定来更新存储 - store.Update()- 阻塞Thread1
3. Thread1对商店执行Parallel.ForEach以运行一组过滤器
4. Thread3(由派生线程1的Parallel.ForEach)尝试恒定时读取的商店.它尝试获取读锁定但在Thread2的写锁定后被阻止.
5. Thread1无法完成,因为它无法加入Thread3. 线程2无法完成,因为它阻止后面线程1.

理想情况下,如果当前线程的祖先线程已经具有相同的锁,我想要做的就是不尝试获取读锁.有没有办法做到这一点？或者还有另一种/更好的方法吗？

public abstract class GenericStore<TKey, TValue>
{
    private ReaderWriterLockSlim _lock = new ReaderWriterLockSlim();
    private List<IFilter> _filters;  //contains instance of ExampleOffendingFilter

    protected Dictionary<TKey, TValue> Store { get; private set; }

    public void Update()
    {
        _lock.EnterWriterLock();
        //update the store
        _lock.ExitWriteLock();
    }

    public TValue GetByKey(TKey …

我需要从美国中北部移动到美国西部的420万张图像附近,作为利用Azure VM支持的大型迁移的一部分(对于那些不知道,美国中北部不支持的人)他们).图像都在一个容器中,分成大约119,000个目录.

我正在使用Copy Blob API中的以下内容:

public static void CopyBlobDirectory(
        CloudBlobDirectory srcDirectory,
        CloudBlobContainer destContainer)
{
    // get the SAS token to use for all blobs
    string blobToken = srcDirectory.Container.GetSharedAccessSignature(
        new SharedAccessBlobPolicy
        {
            Permissions = SharedAccessBlobPermissions.Read |
                            SharedAccessBlobPermissions.Write,
            SharedAccessExpiryTime = DateTime.UtcNow + TimeSpan.FromDays(14)
        });

    var srcBlobList = srcDirectory.ListBlobs(
        useFlatBlobListing: true,
        blobListingDetails: BlobListingDetails.None).ToList();

    foreach (var src in srcBlobList)
    {
        var srcBlob = src as ICloudBlob;

        // Create appropriate destination blob type to match the source blob
        ICloudBlob destBlob;
        if (srcBlob.Properties.BlobType == BlobType.BlockBlob)
            destBlob …

ArgumentOutOfRangeException: Non-negative number required.调用类的ReadTo()方法时,我的代码不确定地抛出SerialPort:

public static void RetrieveCOMReadings(List<SuperSerialPort> ports)
{
    Parallel.ForEach(ports, 
        port => port.Write(port.ReadCommand));

    Parallel.ForEach(ports,
        port =>
        {
            try
            {
                // this is the offending line.
                string readto = port.ReadTo(port.TerminationCharacter);

                port.ResponseData = port.DataToMatch.Match(readto).Value;
            }
            catch (Exception ex)
            {
                Debug.WriteLine(ex.Message);
                port.ResponseData = null;
            }
        });
}

SuperSerialPort是SerialPort类的扩展,主要用于保存端口上每个设备特定的通信所需的信息.
一个端口总是有TerminationCharacter定义的;
大部分时间它都是换行符:

我不明白为什么会这样.
如果ReadTo无法找到输入缓冲区中指定的字符,那么它不应该只是超时并且不返回任何内容吗？

StackTrace指向mscorlib中的一个违规函数,在SerialPort类的定义中:

System.ArgumentOutOfRangeException occurred
  HResult=-2146233086
  Message=Non-negative number required.
Parameter name: byteCount
  Source=mscorlib
  ParamName=byteCount
  StackTrace:
       at System.Text.ASCIIEncoding.GetMaxCharCount(Int32 byteCount)
  InnerException:

我跟着它,这就是我发现的:

private int …

我一直在敲打这个问题好几个小时,我总是因为线程争用而吃掉了并行化循环的性能改进.

我正在尝试计算8位灰度千兆像素图像的直方图.读过"CUDA by example"一书的人可能会知道它的来源(第9章).

该方法非常简单(导致非常紧凑的循环).它基本上就是这样

    private static void CalculateHistogram(uint[] histo, byte[] buffer) 
    {
        foreach (byte thisByte in buffer) 
        {
            // increment the histogram at the position
            // of the current array value
            histo[thisByte]++;
        }
    }

其中buffer是1024 ^ 3个元素的数组.

在最新的Sandy Bridge-EX CPU构建中,10亿个元素的直方图在一个核心上运行1秒钟.

无论如何,我尝试通过在所有核心之间分配循环来加速计算,最终得到的解决方案慢了50倍.

    private static void CalculateHistrogramParallel(byte[] buffer, ref int[] histo) 
    {
        // create a variable holding a reference to the histogram array
        int[] histocopy = histo;

        var parallelOptions = new ParallelOptions { MaxDegreeOfParallelism = Environment.ProcessorCount };

        // loop through the …

我有一堆数据行,我想使用Parallel.ForEach来计算每一行的某些值,就像这样......

class DataRow
{
    public double A { get; internal set; }
    public double B { get; internal set; }
    public double C { get; internal set; }

    public DataRow()
    {
        A = double.NaN;
        B = double.NaN;
        C = double.NaN;
    }
}

class Program
{
    static void ParallelForEachToyExample()
    {
        var rnd = new Random();
        var df = new List<DataRow>();

        for (int i = 0; i < 10000000; i++)
        {
            var dr = new DataRow {A = rnd.NextDouble()};
            df.Add(dr);
        }

        // …

我在四核计算机上运行一个应用程序,它的CPU使用率最高可达14%,并且不断地停留在那里.

Parallel.ForEach(pages,
    new ParallelOptions { },
      page =>
      {
         // do some CPU intensive stuff
      });

只有当我启动应用程序的多个实例时,我才能获得超过14%.通过这样做,我总共可以接近100%.

为什么Parallel.ForEach没有并行完成任务以使用所有CPU？

我正在运行C#.NET 4.5.1.

我正在尝试使用来加快冗长的操作（几毫秒）* Parallel.For，但是在方法返回之前，我正在整个WinForms应用程序中获取Paint事件-建议以某种方式触发消息泵。但是，整体重绘会导致以不一致的状态访问数据，从而产生错误的错误和异常。我需要确保Parallel.For阻止时不会触发UI代码。

到目前为止，我对此的研究尚无定论，并大致向我指出了诸如同步上下文和TaskScheduler实现之类的内容，但是我还没有弄清这一切。

如果有人在整个过程中可以帮助我解决一些问题，将不胜感激。

1. 什么是导致Parallel.For触发WinForms消息泵的事件链？
2. 有什么办法可以完全防止这种情况发生？
3. 另外，是否有任何方法可以判断是否从常规消息泵中调用了UI事件处理程序，还是由“触发”消息触发了“忙”消息泵Parallel.For？

编辑： *一些上下文：以上几毫秒的操作是游戏引擎循环的一部分，其中16毫秒可用于完全更新-因此属性为“长”。此问题的上下文是在其编辑器（即WinForms应用程序）中执行游戏引擎核心。Parallel.For发生在内部引擎更新期间。

在 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++){ …