Parallel.ForEach 使用自定义 TaskScheduler 来防止 OutOfMemoryException答案

【问题标题】：Parallel.ForEach with a custom TaskScheduler to prevent OutOfMemoryExceptionParallel.ForEach 使用自定义 TaskScheduler 来防止 OutOfMemoryException
【发布时间】：2018-03-27 10:52:10
【问题描述】：

我正在通过 Parallel.ForEach 处理大小差异很大的 PDF（从简单的 2MB 到几百 MB 的高 DPI 扫描），并且偶尔会遇到 OutOfMemoryException - 这是可以理解的，因为该进程是 32 位的，并且线程由Parallel.ForEach 占用了未知数量的内存消耗工作。

限制MaxDegreeOfParallelism 确实有效，但在处理大量（10k+）批小 PDF 时的吞吐量不足，因为由于所述线程的内存占用小，可能有更多线程在工作.这是一个使用 Parallel.ForEach 很容易达到 100% CPU 的 CPU 密集型进程，然后偶尔会遇到大型 PDF 组并获得 OutOfMemoryException。运行 Performance Profiler 可以支持这一点。

据我了解，为我的 Parallel.ForEach 设置分区器不会提高我的性能。

这导致我使用通过MemoryFailPoint 检查传递给我的Parallel.ForEach 的自定义TaskScheduler。搜索它似乎没有关于创建自定义 TaskScheduler 对象的信息。

查看Specialized Task Schedulers in .NET 4 Parallel Extensions Extras、A custom TaskScheduler in C# 和 Stackoverflow 上的各种答案，我创建了自己的 TaskScheduler 并拥有我的 QueueTask 方法：

protected override void QueueTask(Task task)
{
    lock (tasks) tasks.AddLast(task);
    try
    {
        using (MemoryFailPoint memFailPoint = new MemoryFailPoint(600))
        {
            if (runningOrQueuedCount < maxDegreeOfParallelism)
            {
                runningOrQueuedCount++;
                RunTasks();
            }
        }
    }
    catch (InsufficientMemoryException e)
    {     
        // somehow return thread to pool?           
        Console.WriteLine("InsufficientMemoryException");
    }
}

虽然 try/catch 有点贵，但我的目标是捕捉 600MB 的可能最大 PDF（+ 一点额外的内存开销）何时会引发 OutOfMemoryException。当我捕捉到 InsufficientMemoryException 时，这个解决方案似乎杀死了试图完成工作的线程。有了足够大的 PDF，我的代码最终成为单线程 Parallel.ForEach。

在 Parallel.ForEach 和 OutOfMemoryExceptions 上的 Stackoverflow 上发现的其他问题似乎不适合我在线程上使用动态内存使用最大吞吐量的用例，并且通常只是利用 MaxDegreeOfParallelism 作为静态解决方案，例如：

因此，要获得可变工作内存大小的最大吞吐量，要么：

当线程通过MemoryFailPoint 检查被拒绝工作时，如何将线程返回到线程池中？
当有可用内存时，如何/在哪里安全地生成新线程以重新开始工作？

编辑：由于依赖于 PDF 内容的光栅化和光栅化图像处理组件，磁盘上的 PDF 大小可能无法线性表示内存中的大小。

【问题讨论】：

标签： c# multithreading parallel-processing

【解决方案1】：

使用来自Samples for Parallel Programming with the .NET Framework 的LimitedConcurrencyLevelTaskScheduler，我能够进行细微的调整，以获得看起来像我想要的东西。以下是修改后LimitedConcurrencyLevelTaskScheduler类的NotifyThreadPoolOfPendingWork方法：

private void NotifyThreadPoolOfPendingWork()
{
    ThreadPool.UnsafeQueueUserWorkItem(_ =>
    {
        // Note that the current thread is now processing work items.
        // This is necessary to enable inlining of tasks into this thread.
        _currentThreadIsProcessingItems = true;
        try
        {
            // Process all available items in the queue.
            while (true)
            {
                Task item;
                lock (_tasks)
                {
                    // When there are no more items to be processed,
                    // note that we're done processing, and get out.
                    if (_tasks.Count == 0)
                    {
                        --_delegatesQueuedOrRunning;
                        break;
                    }

                    // Get the next item from the queue
                    item = _tasks.First.Value;
                    _tasks.RemoveFirst();
                }

                // Execute the task we pulled out of the queue
                //base.TryExecuteTask(item);

                try
                {
                    using (MemoryFailPoint memFailPoint = new MemoryFailPoint(650))
                    {
                        base.TryExecuteTask(item);
                    }
                }
                catch (InsufficientMemoryException e)
                {
                    Thread.Sleep(500);

                    lock (_tasks)
                    {
                        _tasks.AddLast(item);
                    }
                }

            }
        }
        // We're done processing items on the current thread
        finally { _currentThreadIsProcessingItems = false; }
    }, null);
}

我们将看看捕获，但反过来。我们将要处理的任务添加回任务列表 (_tasks)，这会触发一个事件以获取可用线程来完成该工作。但是我们首先休眠当前线程，以便它不会直接开始工作并返回失败的MemoryFailPoint 检查。

【讨论】：