【问题标题】:Multiple Parallel.ForEach calls, MemoryBarrier?多个 Parallel.ForEach 调用,MemoryBarrier?
【发布时间】:2015-05-15 03:19:22
【问题描述】:

我有一堆数据行,我想使用 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);
        }

        // Ever Needed? (I)
        //Thread.MemoryBarrier();

        // Parallel For Each (II)
        Parallel.ForEach(df, dr =>
        {
            dr.B = 2.0*dr.A;
        });

        // Ever Needed? (III)
        //Thread.MemoryBarrier();

        // Parallel For Each 2 (IV)
        Parallel.ForEach(df, dr =>
        {
            dr.C = 2.0 * dr.B;
        });
    }
}

(在这个例子中,没有必要并行化,如果有的话,它可以全部放在一个 Parallel.ForEach 中。但这意味着是一些代码的简化版本,像这样设置它是有意义的)。

是否可以在此处对读取进行重新排序,以便最终得到 B != 2A 或 C != 2B 的数据行?

假设第一个 Parallel.ForEach (II) 分配工作线程 42 处理数据行 0。第二个 Parallel.ForEach (IV) 分配工作线程 43 处理数据行 0(与第一个 Parallel.ForEach 相同) ForEach 完成)。线程 43 上第 0 行的 dr.B 读取是否有可能返回 double.NaN,因为它还没有看到线程 42 的写入?

如果是这样,在 III 处插入内存屏障是否有帮助?这是否会强制第一个 Parallel.ForEach 的更新在第二个 Parallel.ForEach 启动之前对所有线程可见?

【问题讨论】:

  • 简而言之..我认为您不需要显式的内存屏障..有根据的猜测是 Parallel.ForEach 的实现具有某种同步来结束循环/在调用@之前987654322@返回
  • 如果您能更好地了解您的实际代码,我或许可以给您一个更好的答案,而不是“不,别担心”。 :)
  • 如果我说第二个并行循环(IV)的每一行中的计算取决于一些只有在第一个循环之后才能知道的值(II ) 完成。假设我们需要所有行的 dr.B 值的中位数,然后才能计算每行的 dr.C 值。

标签: c# multithreading task-parallel-library parallel.foreach


【解决方案1】:

Parallel.ForEach() 开始的工作将在它返回之前完成。在内部,ForEach() 为每次迭代生成一个 Task,并在每个迭代上调用 Wait()。因此,您无需在 ForEach() 调用之间同步访问。

确实需要记住,对于具有ForEach() 重载的单个任务,这些任务允许您访问循环状态、聚合来自任务的结果等。例如,在这个简单的例子中总结1 ≤ x ≤ 100,传递给localFinallyParallel.For()Action不得不关注同步问题,

var total = 0;

Parallel.For(0, 101, () => 0,  // <-- localInit
(i, state, localTotal) => { // <-- body
  localTotal += i;
  return localTotal;
}, localTotal => { <-- localFinally
  Interlocked.Add(ref total, localTotal); // Note the use of an `Interlocked` static method
});

// Work of previous `For()` call is guaranteed to be done here

Console.WriteLine(total);

在您的示例中,没有必要在 ForEach() 调用之间插入内存屏障。具体来说,循环IV 可以取决于II 正在完成的结果,并且Parallel.ForEach() 已经为您插入III

片段来自:Parallel Framework and avoiding false sharing

【讨论】:

  • 谢谢。当我向下查看 Parallel.ForEach 的代码时,看起来“私有静态 ParallelLoopResult ForWorker”正在完成大部分工作。对我来说有点难以理解,但看起来有人调用了“rootTask.Wait();”在继续之前等待所有工作线程完成。但是,即使我的主线程正在等待工作线程完成,但这并不能保证分散在所有其他处理器中的工作线程在读取值时一定会看到最近的写入,不是吗?
  • 那是正确的,我将编辑我的答案,也许会更清楚一点。 same ForEach() 调用产生的任务需要注意并发问题 - 通常需要关注的地方是您传递给 localFinallyAction。但是,ForEach() 的不同调用可以安全地依赖于之前ForEach() 调用的结果。
  • 我想我的问题与此有关...stackoverflow.com/questions/6581848/memory-barrier-generators。我只是想确保 Parallel.ForEach 的结尾落入这些桶之一。这样它就有自己的 MemoryBarrier(有效),并保证在下一个 Parallel.ForEach 开始之前所有内容都已完全写入。
  • 相关的是 - 这更像是“产生内存障碍的原因”,而您的则是“我是否需要在 Parallel.ForEach() 调用之间使用一个”。
  • 我猜这是否正确...albahari.com/threading/part4.aspx 然后“任何依赖于信号的东西,例如启动或等待任务”隐含地生成一个完整的围栏。在 Parallel.ForEach 中完成大部分工作的函数“private static ParallelLoopResult ForWorker”主体中的最后一件事是调用“rootTask.Wait();” (在 catch 和 finally 块之前)。所以看起来这个调用生成了与 MemoryBarrier 相同的完整栅栏。所以不需要。
【解决方案2】:

由于多个线程将访问同一个变量“dr.B”,因此您需要确保您的 C# 代码是线程安全的。

尝试在每个操作中使用“锁定” https://msdn.microsoft.com/en-us/library/c5kehkcz.aspx

例如

private Object thisLock1 = new Object();
...
lock(thisLock1)
{
    dr.C = 2.0 * dr.B;
}

...
lock(thisLock1)
{
    dr.B = 2.0*dr.A;
}

但是,这样做会破坏并行处理。因为每个线程都必须等到下一个线程完成。

请务必阅读并行处理的潜在缺陷: https://msdn.microsoft.com/en-us/library/dd997403%28v=vs.110%29.aspx

【讨论】:

  • 在使用 Parallel.ForEach() 的 OP 的具体示例中,每个 ForEach() 调用已经处理同步,特别是确保调用产生的任何并行操作在它返回之前完成。
  • @jdphenix - 你能提供一个参考(为了我的教育)吗?注意 Microsoft MSDN 显示:如何:编写具有线程局部变量 msdn.microsoft.com/en-us/library/dd460703%28v=vs.110%29.aspx 的 Parallel.ForEach 循环,该循环使用 (finalResult) => Interlocked.Add(ref total, finalResult)
  • 声明个人ForEach() 确实需要考虑线程安全是正确的,因此ForEach() 提供了允许指定线程本地和终结器的重载,因为您已链接。至于 Wait() 内部对 ForEach() 的调用,我必须查看参考源以确认这一点。
  • 我可以锁定每个数据行。我同意在 99% 的情况下这是正确的做法,可以避免推理低锁码。但在这里,我确实有 huge 数量的数据行,它们很适合并行处理。添加锁定会大大减慢我的真实代码。我真的不需要在这里保证互斥。根据设计,每个工作线程一次只能操作一个数据行。我真正需要做的是确保如果thead A 在第一个循环中的第 i 行上工作,并且在第二个循环中的线程 B 上工作,B 可以在继续之前看到 A 的所有写入。
猜你喜欢
  • 2014-05-04
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 2021-05-21
  • 1970-01-01
  • 1970-01-01
相关资源
最近更新 更多