并行 Linq 查询优化

Question

一段时间以来，我一直在围绕没有副作用的方法构建我的代码，以便使用并行 linq 来加快速度。在此过程中，我不止一次偶然发现惰性评估使事情变得更糟而不是更好，我想知道是否有任何工具可以帮助优化并行 linq 查询。

我之所以这么问是因为我最近通过修改一些方法并在某些关键位置添加AsParallel 来重构一些令人尴尬的并行代码。运行时间从 2 分钟下降到 45 秒，但从性能监视器中可以清楚地看出，有些地方 CPU 上的所有内核都没有得到充分利用。在几次错误启动后，我使用ToArray 强制执行一些查询，运行时间进一步下降到 16 秒。减少代码的运行时间感觉很好，但也有点令人不安，因为不清楚代码查询的哪个位置需要使用ToArray 强制执行。等到最后一分钟才执行查询并不是最佳策略，但根本不清楚代码中的哪些点需要强制执行某些子查询才能利用所有 CPU 内核。

因为我不知道如何正确地使用 ToArray 或其他强制执行 linq 计算的方法以获得最大的 CPU 利用率。那么是否有任何通用指南和工具来优化并行 linq 查询？

这是一个伪代码示例：

var firstQuery = someDictionary.SelectMany(FirstTransformation);
var secondQuery = firstQuery.Select(SecondTransformation);
var thirdQuery = secondQuery.Select(ThirdTransformation).Where(SomeConditionCheck);
var finalQuery = thirdQuery.Select(FinalTransformation).Where(x => x != null);

FirstTransformation、SecondTransformation、ThirdTransformation 都受 CPU 限制，就复杂性而言，它们是一些 3x3 矩阵乘法和一些 if 分支。 SomeConditionCheck 几乎是 null 支票。 FinalTransformation 是代码中 CPU 最密集的部分，因为它将执行一大堆线平面相交，并检查这些相交的多边形包含情况，然后提取最接近线上某个点的相交。

我不知道为什么我放置AsParallel 的地方会减少代码的运行时间。我现在在运行时间方面达到了局部最小值，但我不知道为什么。我偶然发现它只是运气不好。如果您想知道放置AsParallel 的位置是第一行和最后一行。将AsParallel 放在其他任何地方只会增加运行时间，有时最多会增加 20 秒。第一行还有一个隐藏的ToArray。

Answer 1

这里发生了几件事：

1. PLINQ 并行化集合的效率高于未计数的 IEnumerables。如果您有一个数组，它会将数组长度除以您的 CPU 内核数，然后将它们平均分配出去。但是，如果您有一个长度未知的 IEnumerable，它会执行一种愚蠢的指数加速类型的事情，其中​​任务将一次处理 1、2、4、8 等元素，直到它到达 IEnumerable 的末尾。
2. 通过并行化所有查询，您可以将工作分解为小块。如果你有跨 N 个元素的 M 个并行查询，你最终会得到 M*N 个任务。与仅并行化最后一个查询相比，这会产生更多的线程开销，在这种情况下，您最终会得到 N 个任务。
3. 当每项任务的处理时间大致相同时，PLINQ 的效果最好。这样它就可以在核心之间平均分配它们。通过并行化具有不同性能行为的每个查询，您有 M*N 个任务需要不同的时间，而 PLINQ 无法以最佳方式调度它们（因为它不提前知道每个可能需要多长时间)。

所以这里的总体指导方针是：确保在开始之前，如果可能的话，您有一个数组，并且只在评估之前将 AsParallel 放在最后一个查询上。所以像下面这样的东西应该工作得很好：

var firstQuery = someDictionary.SelectMany().ToArray().Select(FirstTransformation);
var secondQuery = firstQuery.Select(SecondTransformation);
var thirdQuery = secondQuery.Select(ThirdTransformation).AsParallel().Where(SomeConditionCheck).ToArray();
var finalQuery = thirdQuery.Select(FinalTransformation).AsParallel().Where(x => x != null);

Answer 2

如果不查看实际代码，几乎不可能说出来。但作为一般准则，您应该考虑在复数运算期间避免使用 P/LINQ，因为委托和 IEnumerable 开销太高了。使用线程获得的速度很可能被 LINQ 提供的方便抽象所消耗。

这里有一些代码确实计算了 2 个整数列表的总和，它对浮点数进行了一些 int 比较，然后计算它的 cos。非常基本的东西，可以用 LINQ 的 .Zip 运算符很好地完成……或者用 for 循环的老式方法。

在我的 Haswell 8 核机器上使用更新的 ParallelLinq 更新 1

• Linq 0.95s
• Linq 并行 0.19 秒
• 优化 0.45 秒
• 优化并行 0.08s

更新 1 结束

• LINQ 1.65s
• 优化 0.64 秒
• 优化并行 0.40 秒

由于 IEnumerable 惰性和方法调用开销（我确实使用了发布模式 x32 Windows 7、.NET 4 双核），时间差几乎是 3 倍。我曾尝试在 LINQ 版本中使用 AsParallel，但它确实变得更慢（2,3 秒）。如果您是数据驱动的，您应该使用 Parallel.For 构造以获得良好的可扩展性。 IEnumerable 本身不适合并行化，因为

• 直到最后，你都不知道自己做了多少工作。
• 您不能进行即时分块，因为您不知道 IEnumerable 将多快返回下一个元素（可能是 Web 服务调用或数组索引访问）。

下面是一个代码示例来说明这一点。如果您想针对裸机进行更多优化，您首先需要摆脱每个项目成本过高的抽象。与非内联 MoveNext() 和 Current 方法调用相比，数组访问要便宜得多。

    class Program
    {
        static void Main(string[] args)
        {
            var A = new List<int>(Enumerable.Range(0, 10*1000*1000));
            var B = new List<int>(Enumerable.Range(0, 10*1000*1000));

            double[] Actual = UseLinq(A, B);
            double[] pActual = UseLinqParallel(A, B);

            var other = Optimized(A, B);
            var pother = OptimizedParallel(A, B);
        }

        private static double[] UseLinq(List<int> A, List<int> B)
        {
            var sw = Stopwatch.StartNew();
            var Merged = A.Zip(B, (a, b) => a + b);
            var Converted = A.Select(a => (float)a);

            var Result = Merged.Zip(Converted, (m, c) => Math.Cos((double)m / ((double)c + 1)));

            double[] Actual = Result.ToArray();
            sw.Stop();

            Console.WriteLine("Linq {0:F2}s", sw.Elapsed.TotalSeconds);
            return Actual;
        }

    private static double[] UseLinqParallel(List<int> A, List<int> B)
    {
        var sw = Stopwatch.StartNew();
        var x = A.AsParallel();
        var y = B.AsParallel();

        var Merged = x.Zip(y, (a, b) => a + b);
        var Converted = x.Select(a => (float)a);

        var Result = Merged.Zip(Converted, (m, c) => Math.Cos((double)m / ((double)c + 1)));

        double[] Actual = Result.ToArray();
        sw.Stop();

        Console.WriteLine("Linq Parallel {0:F2}s", sw.Elapsed.TotalSeconds);
        return Actual;
    }        

        private static double[] OptimizedParallel(List<int> A, List<int> B)
        {
            double[] result = new double[A.Count];
            var sw = Stopwatch.StartNew();
            Parallel.For(0, A.Count, (i) =>
            {
                var sum = A[i] + B[i];
                result[i] = Math.Cos((double)sum / ((double)((float)A[i]) + 1));
            });
            sw.Stop();

            Console.WriteLine("Optimized Parallel {0:F2}s", sw.Elapsed.TotalSeconds);
            return result;
        }

        private static double[] Optimized(List<int> A, List<int> B)
        {
            double[] result = new double[A.Count];
            var sw = Stopwatch.StartNew();
            for(int i=0;i<A.Count;i++)
            {
                var sum = A[i] + B[i];
                result[i] = Math.Cos((double)sum / ((double)((float)A[i]) + 1));
            }
            sw.Stop();

            Console.WriteLine("Optimized {0:F2}s", sw.Elapsed.TotalSeconds);
            return result;
        }
    }
}