停止线程直到有足够的内存可用

Question

环境：.net 4.0

我有一个使用 XSLT 样式表转换 XML 文件的任务，这是我的代码

public string TransformFileIntoTempFile(string xsltPath, 
    string xmlPath)
{
    var transform = new MvpXslTransform();
    transform.Load(xsltPath, new XsltSettings(true, false), 
        new XmlUrlResolver());

    string tempPath = Path.GetTempFileName();

    using (var writer = new StreamWriter(tempPath))
    {
        using (XmlReader reader = XmlReader.Create(xmlPath))
        {
            transform.Transform(new XmlInput(reader), null, 
                new XmlOutput(writer));
        }       
    }

    return tempPath;
}

我有 X 个线程可以并行启动此任务。 有时我的输入文件大约 300 MB，有时只有几 MB。

我的问题：当我的程序尝试同时转换一些大的 XML 文件时，我得到了 OutOfMemoryException。

如何避免这些 OutOfMemoryEception ？我的想法是在执行任务之前停止一个线程，直到有足够的可用内存，但我不知道该怎么做。或者还有其他一些解决方案（比如将我的任务放在一个不同的应用程序中）。

谢谢

Answer 1

我不建议阻塞线程。在最坏的情况下，您最终会饿死可能释放您需要的内存的任务，从而导致死锁或总体上非常糟糕的性能。

相反，我建议您保留一个有优先级的工作队列。从队列中获取跨线程池公平调度的任务。 确保没有线程在等待操作中阻塞，而是将任务重新发布到队列（优先级较低）。

因此，您要做的（例如，在收到 OutOfMemory 异常时）将相同的作业/任务发布到队列中并终止当前任务，从而为另一个任务释放线程。

一种简单的方法是使用 LIFO，它确保发布到队列中的任务将比该队列中已经存在的任何其他作业具有“低优先级”。

Answer 2

从 .NET Framework 4 开始，我们的 API 可以使用多年以来在 Win32API 中可用的旧 Memory-Mapped Files 功能，因此现在您可以从 .NET 托管代码中使用它。

为了您的任务更适合“持久内存映射文件”选项， MSDN:

持久化文件是与 磁盘上的源文件。当最后一个进程完成工作时 文件，数据保存到磁盘上的源文件中。这些 内存映射文件适用于处理非常大的文件 源文件。

在MemoryMappedFile.CreateFromFile() 方法描述页面上，您可以找到一个很好的示例，描述如何为超大文件创建内存映射视图。

编辑：有关 cmets 中大量注释的更新

刚刚找到的方法MemoryMappedFile.CreateViewStream() 创建了一个MemoryMappedViewStream 类型的流，该流继承自System.IO.Stream。 我相信您可以从此流创建 XmlReader 的实例，然后使用此阅读器/流实例化您的 XslTransform 自定义实现。

EDIT2： remi bourgarel (OP) 已经测试过这种方法，看起来这个特殊的 XslTransform 实现（我想知道是否有任何方法）不会以预期的方式与 MM-View 流一起工作

Answer 3

主要问题是您正在加载整个 Xml 文件。如果您只是在读取时进行转换，则通常不会出现内存不足问题。 话虽如此，我找到了一篇 MS 支持文章，其中建议了如何完成： http://support.microsoft.com/kb/300934

免责声明：我没有对此进行测试，所以如果您使用它并且它有效，请告诉我们。

Answer 4

您可以考虑使用队列来限制基于某种人工内存边界（例如文件大小。可以使用类似下面的东西。

这种节流策略可以与正在处理的最大并发文件数相结合，以确保您的磁盘不会被过多地颠簸。

NB 我没有在执行过程中包含必要的 try\catch\finally，以确保异常传播到调用线程并始终释放 Waithandles。我可以在这里更详细地介绍。

public static class QueuedXmlTransform
{
    private const int MaxBatchSizeMB = 300;
    private const double MB = (1024 * 1024);
    private static readonly object SyncObj = new object();
    private static readonly TaskQueue Tasks = new TaskQueue();
    private static readonly Action Join = () => { };
    private static double _CurrentBatchSizeMb;

    public static string Transform(string xsltPath, string xmlPath)
    {
        string tempPath = Path.GetTempFileName();

        using (AutoResetEvent transformedEvent = new AutoResetEvent(false))
        {
            Action transformTask = () =>
            {
                MvpXslTransform transform = new MvpXslTransform();

                transform.Load(xsltPath, new XsltSettings(true, false),
                    new XmlUrlResolver());

                using (StreamWriter writer = new StreamWriter(tempPath))
                using (XmlReader reader = XmlReader.Create(xmlPath))
                {
                    transform.Transform(new XmlInput(reader), null,
                        new XmlOutput(writer));
                }

                transformedEvent.Set();
            };

            double fileSizeMb = new FileInfo(xmlPath).Length / MB;

            lock (SyncObj)
            {
                if ((_CurrentBatchSizeMb += fileSizeMb) > MaxBatchSizeMB)
                {
                    _CurrentBatchSizeMb = fileSizeMb;

                    Tasks.Queue(isParallel: false, task: Join);
                }

                Tasks.Queue(isParallel: true, task: transformTask);
            }

            transformedEvent.WaitOne();
        }

        return tempPath;
    }

    private class TaskQueue
    {
        private readonly object _syncObj = new object();
        private readonly Queue<QTask> _tasks = new Queue<QTask>();
        private int _runningTaskCount;

        public void Queue(bool isParallel, Action task)
        {
            lock (_syncObj)
            {
                _tasks.Enqueue(new QTask { IsParallel = isParallel, Task = task });
            }

            ProcessTaskQueue();
        }

        private void ProcessTaskQueue()
        {
            lock (_syncObj)
            {
                if (_runningTaskCount != 0) return;

                while (_tasks.Count > 0 && _tasks.Peek().IsParallel)
                {
                    QTask parallelTask = _tasks.Dequeue();

                    QueueUserWorkItem(parallelTask);
                }

                if (_tasks.Count > 0 && _runningTaskCount == 0)
                {
                    QTask serialTask = _tasks.Dequeue();

                    QueueUserWorkItem(serialTask);
                }
            }
        }

        private void QueueUserWorkItem(QTask qTask)
        {
            Action completionTask = () =>
            {
                qTask.Task();

                OnTaskCompleted();
            };

            _runningTaskCount++;

            ThreadPool.QueueUserWorkItem(_ => completionTask());
        }

        private void OnTaskCompleted()
        {
            lock (_syncObj)
            {
                if (--_runningTaskCount == 0)
                {
                    ProcessTaskQueue();
                }
            }
        }

        private class QTask
        {
            public Action Task { get; set; }
            public bool IsParallel { get; set; }
        }
    }
}

更新

修复了滚动到下一个批处理窗口时保持批处理大小的错误：

_CurrentBatchSizeMb = fileSizeMb;