并行 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 与 float 比较，然后计算它的 cos。非常基本的东西可以用 LINQ 的 .Zip 操作符很好地完成......或者使用 for 循环的老式方法。

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

• Linq 0.95s
• Linq 并行 0,19s
• 优化 0.45 秒
• 优化并行 0,08s

更新1 结束

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

由于 IEnumerable 惰性和方法调用，时间差异几乎是 3 倍开销（我确实使用了发布模式 x32 Windows 7，.NET 4 双核）。我尝试在 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;
        }
    }
}