c# - 日志文件的并行 GZip 解压缩 - 调整 MaxDegreeOfParallelism 以获得最高吞吐量

Question

我们每天有多达 30 GB 的 GZip 日志文件。每个文件包含 100.000 行，压缩后大小在 6 到 8 MB 之间。已剥离解析逻辑的简化代码利用了 Parallel.ForEach 循环。

在双 NUMA 节点、32 个逻辑 CPU 盒(Intel Xeon E7-2820 @ 2 GHz)上，MaxDegreeOfParallelism 处理峰值的最大行数:

using System;

using System.Collections.Concurrent;

using System.Linq;
using System.IO;
using System.IO.Compression;

using System.Threading.Tasks;

namespace ParallelLineCount
{
    public class ScriptMain
    {
        static void Main(String[] args)
        {
            int    maxMaxDOP      = (args.Length > 0) ? Convert.ToInt16(args[0]) : 2;
            string fileLocation   = (args.Length > 1) ? args[1] : "C:\\Temp\\SomeFiles" ;
            string filePattern    = (args.Length > 1) ? args[2] : "*2012-10-30.*.gz";
            string fileNamePrefix = (args.Length > 1) ? args[3] : "LineCounts";

            Console.WriteLine("Start:                 {0}", DateTime.UtcNow.ToString("yyyy-MM-ddTHH:mm:ss.fffffffZ"));
            Console.WriteLine("Processing file(s):    {0}", filePattern);
            Console.WriteLine("Max MaxDOP to be used: {0}", maxMaxDOP.ToString());
            Console.WriteLine("");

            Console.WriteLine("MaxDOP,FilesProcessed,ProcessingTime[ms],BytesProcessed,LinesRead,SomeBookLines,LinesPer[ms],BytesPer[ms]");

            for (int maxDOP = 1; maxDOP <= maxMaxDOP; maxDOP++)
            {

                // Construct ConcurrentStacks for resulting strings and counters
                ConcurrentStack<Int64> TotalLines = new ConcurrentStack<Int64>();
                ConcurrentStack<Int64> TotalSomeBookLines = new ConcurrentStack<Int64>();
                ConcurrentStack<Int64> TotalLength = new ConcurrentStack<Int64>();
                ConcurrentStack<int>   TotalFiles = new ConcurrentStack<int>();

                DateTime FullStartTime = DateTime.Now;

                string[] files = System.IO.Directory.GetFiles(fileLocation, filePattern);

                var options = new ParallelOptions() { MaxDegreeOfParallelism = maxDOP };

                //  Method signature: Parallel.ForEach(IEnumerable<TSource> source, Action<TSource> body)
                Parallel.ForEach(files, options, currentFile =>
                    {
                        string filename = System.IO.Path.GetFileName(currentFile);
                        DateTime fileStartTime = DateTime.Now;

                        using (FileStream inFile = File.Open(fileLocation + "\\" + filename, FileMode.Open))
                        {
                            Int64 lines = 0, someBookLines = 0, length = 0;
                            String line = "";

                            using (var reader = new StreamReader(new GZipStream(inFile, CompressionMode.Decompress)))
                            {
                                while (!reader.EndOfStream)
                                {
                                    line = reader.ReadLine();
                                    lines++; // total lines
                                    length += line.Length;  // total line length

                                    if (line.Contains("book")) someBookLines++; // some special lines that need to be parsed later
                                }

                                TotalLines.Push(lines); TotalSomeBookLines.Push(someBookLines); TotalLength.Push(length);
                                TotalFiles.Push(1); // silly way to count processed files :)
                            }
                        }
                    }
                );

                TimeSpan runningTime = DateTime.Now - FullStartTime;

                // Console.WriteLine("MaxDOP,FilesProcessed,ProcessingTime[ms],BytesProcessed,LinesRead,SomeBookLines,LinesPer[ms],BytesPer[ms]");
                Console.WriteLine("{0},{1},{2},{3},{4},{5},{6},{7}",
                    maxDOP.ToString(),
                    TotalFiles.Sum().ToString(),
                    Convert.ToInt32(runningTime.TotalMilliseconds).ToString(),
                    TotalLength.Sum().ToString(),
                    TotalLines.Sum(),
                    TotalSomeBookLines.Sum().ToString(),
                    Convert.ToInt64(TotalLines.Sum() / runningTime.TotalMilliseconds).ToString(),
                    Convert.ToInt64(TotalLength.Sum() / runningTime.TotalMilliseconds).ToString());

            }
            Console.WriteLine();
            Console.WriteLine("Finish:                " + DateTime.UtcNow.ToString("yyyy-MM-ddTHH:mm:ss.fffffffZ"));
        }
    }
}

这是结果的总结，在 MaxDegreeOfParallelism = 8 处有一个明显的峰值:

CPU 负载(此处汇总显示，大部分负载都在单个 NUMA 节点上，即使 DOP 在 20 到 30 范围内):

我发现使 CPU 负载超过 95% 标记的唯一方法是将文件拆分到 4 个不同的文件夹并执行相同的命令 4 次，每次针对所有文件的一个子集。

有人能找到瓶颈吗？

Answer 1

一个问题很可能是默认的 FileStream 构造函数使用的缓冲区大小较小。我建议您使用更大的输入缓冲区。如:

using (FileStream infile = new FileStream(
    name, FileMode.Open, FileAccess.Read, FileShare.None, 65536))

默认缓冲区大小为 4 KB，线程会多次调用 I/O 子系统来填充其缓冲区。 64K 的缓冲区意味着您进行这些调用的频率会大大降低。

我发现缓冲区大小介于 32K 和 256K 之间可提供最佳性能，而 64K 是我不久前进行一些详细测试时的“最佳点”。大于 256K 的缓冲区大小实际上开始降低性能。

此外，虽然这不太可能对性能产生重大影响，但您可能应该将那些 ConcurrentStack 实例替换为 64 位整数并使用 Interlocked.Add 或 Interlocked.Increment 更新它们。它简化了您的代码并消除了管理集合的需要。

更新:

重新阅读你的问题描述，我对这个陈述感到震惊:

The only way I've found to make CPU load cross 95% mark was to split the files across 4 different folders and execute the same command 4 times, each one targeting a subset of all files.

对我来说，这表明打开文件时存在瓶颈。就好像操作系统在目录上使用互斥锁一样。即使所有数据都在缓存中并且不需要物理 I/O，进程仍然需要等待这个锁。文件系统也可能正在写入磁盘。请记住，无论何时打开文件，它都必须更新文件的上次访问时间。

如果 I/O 确实是瓶颈，那么您可以考虑让一个线程只加载文件并将它们填充到 BlockingCollection 或类似的数据结构中，这样处理线程就不会不必为了锁定目录而相互竞争。您的应用程序成为具有一个生产者和 N 个消费者的生产者/消费者应用程序。