如何修复并行 for() 循环不返回与串行 for() 循环不同的值？答案

【问题标题】：How to repair a Parallel for()-loop not to return a different value than a serial for()-loop?如何修复并行 for() 循环不返回与串行 for() 循环不同的值？
【发布时间】：2017-11-12 20:33:11
【问题描述】：

我对以下代码有疑问。该代码可以正常工作，但是在使用并行 for 循环与常规 for 循环时，我收到了不同的输出值。我需要让并行 for 循环正常工作，因为我运行此代码数千次。有谁知道为什么我的并行 for 循环返回不同的输出？

private object _lock = new object();

public double CalculatePredictedRSquared()
    {
        double press = 0, tss = 0, press2 = 0, press1 = 0;
        Vector<double> output = CreateVector.Dense(Enumerable.Range(0, 400).Select(i => Convert.ToDouble(i)).ToArray());
        List<double> input1 = new List<double>(Enumerable.Range(0, 400).Select(i => Convert.ToDouble(i)));
        List<double> input2 = new List<double>(Enumerable.Range(200, 400).Select(i => Convert.ToDouble(i)));

            Parallel.For(0, output.Count, i =>
            {
                ConcurrentBag<MultipleRegressionInfo> listMRInfoBag = new ConcurrentBag<MultipleRegressionInfo>(listMRInfo);
                ConcurrentBag<double> vectorArrayBag = new ConcurrentBag<double>(output);
                ConcurrentBag<double[]> matrixList = new ConcurrentBag<double[]>();

                lock (_lock)
                {
                    matrixList.Add(input1.Where((v, k) => k != i).ToArray());
                    matrixList.Add(input2.Where((v, k) => k != i).ToArray());
                }

                var matrixArray2 = CreateMatrix.DenseOfColumnArrays(matrixList);
                var actualResult = vectorArrayBag.ElementAt(i);
                var newVectorArray = CreateVector.Dense(vectorArrayBag.Where((v, j) => j != i).ToArray());
                var items = FindBestMRSolution(matrixArray2, newVectorArray);
                double estimate1 = 0;

                if (items != null)
                {
                    lock (_lock)
                    {
                        var y = 0d;
                        var independentCount = matrixArray2.RowCount;
                        var dependentCount = newVectorArray.Count;

                        if (independentCount == dependentCount)
                        {
                            var populationCount = independentCount;
                            y = newVectorArray.Average();

                            for (int l = 0; l < matrixArray2.ColumnCount; l++)
                            {
                                var avg = matrixArray2.Column(l).Average();
                                y -= avg * items[l];
                            }
                        }

                        for (int m = 0; m < 2; m++)
                        {
                            var coefficient = items[m];

                            if (m == 0)
                            {
                                estimate1 += input1.ElementAt(i) * coefficient;
                            }
                            else
                            {
                                estimate1 += input2.ElementAt(i) * coefficient;
                            }
                        }

                        estimate1 += y;
                    }
                }
                else
                {
                    lock (_lock)
                    {
                        estimate1 = 0;
                    }
                }

                lock (_lock)
                {
                    press1 += Math.Pow(actualResult - estimate1, 2);
                }
            });

            for (int i = 0; i < output.Count; i++)
            {
                List<double[]> matrixList = new List<double[]>();
                matrixList.Add(input1.Where((v, k) => k != i).ToArray());
                matrixList.Add(input2.Where((v, k) => k != i).ToArray());
                var matrixArray = CreateMatrix.DenseOfColumnArrays(matrixList);
                var actualResult = output.ElementAt(i);
                var newVectorArray = CreateVector.Dense(output.Where((v, j) => j != i).ToArray());
                var items = FindBestMRSolution(matrixArray, newVectorArray);
                double estimate = 0;

                if (items != null)
                {
                    var y = CalculateYIntercept(matrixArray, newVectorArray, items);
                    for (int m = 0; m < 2; m++)
                    {
                        var coefficient = items[m];

                        if (m == 0)
                        {
                            estimate += input1.ElementAt(i) * coefficient;
                        }
                        else
                        {
                            estimate += input2.ElementAt(i) * coefficient;
                        }
                    }
                }
                else
                {
                    estimate = 0;
                }

                press2 += Math.Pow(actualResult - estimate, 2);
            }

            tss = CalculateTotalSumOfSquares(vectorArray.ToList());
            var test1 = 1 - (press1 / tss);
            var test2 = 1 - (press2 / tss);
}

public Vector<double> CalculateWithQR(Matrix<double> x, Vector<double> y)
    {
        Vector<double> result = null;

            result = MultipleRegression.QR(x, y);

            for (int i = 0; i < result.Count; i++)
            {
                var value = result.ElementAt(i);

                if (Double.IsNaN(value) || Double.IsInfinity(value))
                {
                    return null;
                }
            }

        return result;
    }

    public Vector<double> CalculateWithNormal(Matrix<double> x, Vector<double> y)
    {
        Vector<double> result = null;

            result = MultipleRegression.NormalEquations(x, y);

            for (int i = 0; i < result.Count; i++)
            {
                var value = result.ElementAt(i);

                if (Double.IsNaN(value) || Double.IsInfinity(value))
                {
                    return null;
                }
            }

        return result;
    }

    public Vector<double> CalculateWithSVD(Matrix<double> x, Vector<double> y)
    {
        Vector<double> result = null;

            result = MultipleRegression.Svd(x, y);

            for (int i = 0; i < result.Count; i++)
            {
                var value = result.ElementAt(i);

                if (Double.IsNaN(value) || Double.IsInfinity(value))
                {
                    return null;
                }
            }

        return result;
    }

    public Vector<double> FindBestMRSolution(Matrix<double> x, Vector<double> y)
    {
        Vector<double> result = null;

            result = CalculateWithNormal(x, y);

            if (result != null)
            {
                return result;
            }
            else
            {
                result = CalculateWithSVD(x, y);

                if (result != null)
                {
                    return result;
                }
                else
                {
                    result = CalculateWithQR(x, y);

                    if (result != null)
                    {
                        return result;
                    }
                }
            }

        return result;
    }

public double CalculateTotalSumOfSquares(List<double> dependentVariables)
    {
        double tts = 0;

            for (int i = 0; i < dependentVariables.Count; i++)
            {
                tts += Math.Pow(dependentVariables.ElementAt(i) - dependentVariables.Average(), 2);
            }

        return tts;
    }

实际输出（更新结果）：

test1 = 137431.12889999992 (parallel for loop)

test2 = 7.3770258447689254E- (regular for loop)

结语：如何设置符合 MCVE 的测试

这可能是准备一个真正完全可重现设置的 MCVE 代码 + A/B/C/... DataSET-s ，放入一个可立即运行的 [IDE 和测试沙箱， 超链接此处][1]，以便社区成员可以单击重新运行按钮并专注于根本原因分析，而不是解码和重新设计大量不完整的 SLOC。

如果这适用于 O/P，它将适用于 O/P 要求答案或帮助的其他社区成员。

Try it online!

我的新版代码：

public double CalculatePredictedRSquared()
    {
        Vector<double> output = CreateVector.Dense(Enumerable.Range(0, 400).Select(i => Convert.ToDouble(i)).ToArray());
        List<double> input1 = new List<double>(Enumerable.Range(0, 400).Select(i => Convert.ToDouble(i)));
        List<double> input2 = new List<double>(Enumerable.Range(200, 400).Select(i => Convert.ToDouble(i)));
        double tss = CalculateTotalSumOfSquares(output.ToList());

        IEnumerable<int> range = Enumerable.Range(0, output.Count);
        var query = range.Select(i => DoIt(i, output, input1, input2));
        var result = 1 - (query.Sum() / tss);
        return result;
    }

    public double DoIt(int i, Vector<double> output, List<double> input1, List<double> input2)
    {
        List<double[]> matrixList = new List<double[]>
                {
                    input1.Where((v, k) => k != i).ToArray(),
                    input2.Where((v, k) => k != i).ToArray()
                };
        var matrixArray = CreateMatrix.DenseOfColumnArrays(matrixList);
        var actualResult = output.ElementAt(i);
        var newVectorArray = CreateVector.Dense(output.Where((v, j) => j != i).ToArray());
        var items = FindBestMRSolution(matrixArray, newVectorArray);
        double estimate = 0;

        if (items != null)
        {
            var y = CalculateYIntercept(matrixArray, newVectorArray, items);
            for (int m = 0; m < 2; m++)
            {
                var coefficient = items[m];

                if (m == 0)
                {
                    estimate += input1.ElementAt(i) * coefficient;
                }
                else
                {
                    estimate += input2.ElementAt(i) * coefficient;
                }
            }
        }
        else
        {
            estimate = 0;
        }

       return Math.Pow(actualResult - estimate, 2);
    }

【问题讨论】：

我在并行代码中看到了外部调用，比如FindBestMRSolution，你确定这些东西是线程安全的，因为它们可以以这种方式/上下文安全地使用吗？
无论如何，请发minimal reproducible example。
“发布完整示例的唯一方法是......” - 不，学习将问题减少到不到 30 行。以File|New Project开头。
如果你不能缩小范围，这里没有人可以帮助你。并行代码产生与顺序版本不同的结果的典型原因是您弄乱了共享状态，因此多个并行线程会更改并使用相同的值。另外，您是否验证了这两个值中的哪一个是正确的？
说真的：从一个更简单的问题开始。您试图通过随机使东西线程安全并随机引入锁来解决问题；如果您继续这样做，您将失败。编写正确的多线程程序需要详细了解内存和控制流在 C# 中的工作原理。

标签： c# multithreading parallel-processing

【解决方案1】：

这整件事是狗的早餐；你应该完全放弃并行的尝试。

重新开始。这就是我要你做的。我希望您编写一个方法 DoIt，它返回 double 并采用 int i 和执行循环的单次迭代所需的任何其他状态。

然后你将重写你的方法如下：

public double CalculatePredictedRSquared()
{
    Vector<double> output = whatever;
    // Whatever other state you need here
    IEnumerable<int> range = Enumerable.Range(0, output.Count);
    var query = range.Select(i => DoIt(i, whatever_other_state));
    return query.Sum();
}

明白了吗？ DoIt 是你现在循环中的东西。它必须包含i 和output 以及您需要传递给它的任何其他向量。它必须仅计算一个双精度数——在这种情况下，是估计误差的平方——并返回该双精度数。

它必须是纯：它不能读取或写入任何非局部变量，不能调用任何非纯方法，并且在给定相同输入时必须给出完全相同的结果，每次。纯方法是最容易编写、阅读、理解、测试和并行化的方法； 在进行数学计算时总是尝试编写纯方法。

为DoIt 编写测试用例，并对其进行测试。这是一种纯粹的方法；你应该能够编写很多测试用例。同样测试DoIt调用的任何纯方法。

一旦您对DoIt 既正确又纯粹感到满意，那么奇迹就会发生。只需将其更改为：

range.AsParallel().Select...

然后比较并行和非并行版本。它们应该产生相同的结果；如果不是，那么有些东西是不纯的。弄清楚它是什么。

然后，验证并行版本是否更快。如果没有，那么您在DoIt 中没有做足够的工作来证明并行性是合理的；详情请见https://en.wikipedia.org/wiki/Amdahl%27s_law。

【讨论】：

我刚刚更新了我的问题，以显示我按照您的建议写的内容。它看起来正确吗？
@user3610374：你告诉我！你写了一些测试吗？它会产生正确的结果吗？
是的，我写了一些测试，我似乎得到了正确的结果
@user3610374 虽然 Eric 提到了维基百科上的原始（overhead-naive, atomic-process-agnostic）阿姆达尔定律，但让我指导您宁愿阅读这两个最初的阿姆达尔定律公式和它的批评是由于附加开销和进程原子性甚至来自无限数量的处理器的进一步不可分割的事务...... 关于 >>> stackoverflow.com/tags/parallelism-amdahl/info b> 如果盲目相信间接费用和不可分割的交易都不存在，那么原始公式通常会被误导，但它们确实很重要。无论如何，请继续关注，男人们

【解决方案2】：

一些事情：

lock (_lock)
{
   matrixList.Add(input1.Where((v, k) => k != i).ToArray());
   matrixList.Add(input2.Where((v, k) => k != i).ToArray());
}

您将项目添加到设计上已经是线程安全的集合中，因此无需锁定。虽然List 不是线程安全的，但同时读取它应该没问题。来自documentation：

对 List 执行多次读取操作是安全的，但如果在读取时修改了集合，则可能会出现问题。为确保线程安全，请在读取或写入操作期间锁定集合。要使一个集合能够被多个线程访问以进行读写，您必须实现自己的同步。

还要注意matrixList存储在一个局部变量中；在这种情况下，集合不能从多个线程调用，因为委托的整个主体都保证在同一个线程上运行 - 不会出现一半的主体例如，Parallel.For 循环将在线程 A 上运行，而另一半将在线程 B 上运行。

同样，在对estimate1 进行更改时没有理由锁定，因为它不可能从其他线程中修改。

免责声明：不能保证Parallel.For 循环的整体并行度。甚至无法保证它会完全并行运行。

但是，

press1 和 press2不是局部变量，因此您确实需要以某种方式同步这些变量。（不过，如果你能找到某种方法来避免每次都锁定会更好，因为这至少会部分扼杀多线程的意义）。

也许最关键的是，ConcurrentBag 是一个无序集合。您没有显示您在矩阵上执行的所有操作，但是如果您在任何地方进行矩阵乘法，这很容易导致错误的结果。 不能保证矩阵乘法会通勤。虽然A * B = B * A 用于整数，但not 通常对于矩阵来说是正确的。您的逻辑很可能巧妙地依赖于以特定顺序发生的操作（它们不会因为ConcurrentBag 是无序的）。

【讨论】：