c# - TPL 数据流 block 消耗所有可用内存

Question

我有一个具有以下设计的 TransformManyBlock:

• 输入:文件路径
• 输出:IEnumerable 文件内容，一次一行

我在一个巨大的文件 (61GB) 上运行此 block ，该文件太大而无法放入 RAM。为了避免无限内存增长，我已将此 block 和所有下游 block 的 BoundedCapacity 设置为一个非常低的值(例如 1)。尽管如此，该 block 显然贪婪地迭代了 IEnumerable，这消耗了计算机上的所有可用内存，使每个进程都停止了。 block 的 OutputCount 继续无限上升，直到我终止进程。

如何防止 block 以这种方式消耗 IEnumerable？

编辑:这是一个说明问题的示例程序:

using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;

class Program
{
    static IEnumerable<string> GetSequence(char c)
    {
        for (var i = 0; i < 1024 * 1024; ++i)
            yield return new string(c, 1024 * 1024);
    }

    static void Main(string[] args)
    {
        var options = new ExecutionDataflowBlockOptions() { BoundedCapacity = 1 };
        var firstBlock = new TransformManyBlock<char, string>(c => GetSequence(c), options);
        var secondBlock = new ActionBlock<string>(str =>
            {
                Console.WriteLine(str.Substring(0, 10));
                Thread.Sleep(1000);
            }, options);

        firstBlock.LinkTo(secondBlock);
        firstBlock.Completion.ContinueWith(task =>
            {
                if (task.IsFaulted) ((IDataflowBlock) secondBlock).Fault(task.Exception);
                else secondBlock.Complete();
            });

        firstBlock.Post('A');
        firstBlock.Complete();
        for (; ; )
        {
            Console.WriteLine("OutputCount: {0}", firstBlock.OutputCount);
            Thread.Sleep(3000);
        }
    }
}

如果您使用的是 64 位机器，请确保清除 Visual Studio 中的“首选 32 位”选项。我的计算机上有 16GB 的 RAM，这个程序会立即消耗所有可用字节。

Answer 1

您似乎误解了 TPL 数据流的工作原理。

BoundedCapacity 限制您可以发布到 block 中的项目数量。在您的情况下，这意味着将单个 char 放入 TransformManyBlock 并将单个 string 放入 ActionBlock。

因此，您将单个项目发布到 TransformManyBlock，然后返回 1024*1024 字符串并尝试将它们传递给 ActionBlock一次只接受一个。其余的字符串将位于 TransformManyBlock 的输出队列中。

您可能想要做的是创建一个单独的 block ，并在达到容量时等待(同步或其他方式)以流方式将项目发布到其中:

private static void Main()
{
    MainAsync().Wait();
}

private static async Task MainAsync()
{
    var block = new ActionBlock<string>(async item =>
    {
        Console.WriteLine(item.Substring(0, 10));
        await Task.Delay(1000);
    }, new ExecutionDataflowBlockOptions { BoundedCapacity = 1 });

    foreach (var item in GetSequence('A'))
    {
        await block.SendAsync(item);
    }

    block.Complete();
    await block.Completion;
}