如何在TPL Dataflows中安排流控制？

Question

我试图理解如何控制TPL Dataflow中的数据流。我有一个非常快的生产者和一个非常慢的消费者。(我的实际代码更复杂，但无论如何，这是一个相当好的模型，并且可以重现问题。)运行时，代码开始像饮料一样地喝内存--而生产者的输出队列以最快的速度填满。我真正希望看到的是，生产者停止运行一段时间，直到消费者有机会请求它。根据我对文档的阅读，这就是应该发生的：也就是说，我认为生产者要等待消费者有空间。显然并不是这样。我该怎么修复它，使得队列不会失控？

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

namespace MemoryLeakTestCase
{
    class Program
    {

        static void Main(string[] args)
        {
            var CreateData = new TransformManyBlock<int, string>(ignore =>
            {
                return Enumerable.Range(0, 1000 * 1000 * 1000).Select((s,i) => "Hello, World " + i);
            });

            var ParseFile = new TransformManyBlock<string, string>(fileContent =>
            {
                Thread.Sleep(1000);
                return Enumerable.Range(0, 100).Select((sst, iii) => "Hello, " + iii);
            }, new ExecutionDataflowBlockOptions() { BoundedCapacity = 1000 }
            );

            var EndOfTheLine = new ActionBlock<object>(f =>
                {
                });


            var linkOptions = new DataflowLinkOptions { PropagateCompletion = true, };
            CreateData.LinkTo(ParseFile, linkOptions);
            ParseFile.LinkTo(EndOfTheLine, linkOptions);

            Task t = new Task(() =>
            {
                while (true)
                {
                    Console.WriteLine("CreateData: " + Report(CreateData));
                    Console.WriteLine("ParseData:  " + Report(ParseFile));
                    Console.WriteLine("NullTarget: " +  EndOfTheLine.InputCount );
                    Thread.Sleep(1000);
                }

            });
            t.Start();

            CreateData.SendAsync(0);
            CreateData.Complete();

            EndOfTheLine.Completion.Wait();
        }

        public static string Report<T, U>(TransformManyBlock<T, U> block)
        {
            return String.Format("INPUT: {0}   OUTPUT: {1} ", block.InputCount.ToString().PadLeft(10, ' '), block.OutputCount.ToString().PadLeft(10, ' '));
        }


    }
}

- Danyel Fisher

---

你尝试将 ParseFile 的 BoundedCapacity 降低到一些较小的值，比如说 10，以确保它可以工作吗？ - Los Frijoles

Lois，好想法。我已经尝试过了；它使CreateData队列变得更大。我正在努力解决的主要挑战是如何确保当ParseFile队列太大（或其自身输出队列太大）时，CreateData暂停。 - Danyel Fisher

如果您需要一种输出受限且支持所有ExecutionDataflowBlockOptions的替代TransformManyBlock实现，请查看此处。链接 - Theodor Zoulias

Answer 1

通常在这种情况下，您需要设置CreateData块的BoundedCapacity。但是，在此处不起作用，因为TransformManyBlock似乎不考虑从单个IEnumerable填充输出队列时的BoundedCapacity。

相反，您可以创建一个函数来迭代集合，并使用SendAsync()仅在目标可以接受它们时发送更多数据：

/// <remarks>
/// If iterating data throws an exception, the target block is faulted
/// and the returned Task completes successfully.
/// 
/// Depending on the usage, this might or might not be what you want.
/// </remarks>
public static async Task SendAllAsync<T>(
    this ITargetBlock<T> target, IEnumerable<T> data)
{
    try
    {
        foreach (var item in data)
        {
            await target.SendAsync(item);
        }
    }
    catch (Exception e)
    {
        target.Fault(e);
    }
}

使用方法：

var data = Enumerable.Range(0, 1000 * 1000 * 1000).Select((s,i) => "Hello, World " + i);
await ParseFile.SendAllAsync(data);
ParseFile.Complete();

如果你仍然想要一个类似于原始代码的CreateData块，你可以有两个有界的BufferBlock，在它们之间使用SendAllAsync()，然后使用Encapsulate()将它们看作一个块。

/// <remarks>
/// boundedCapacity represents the capacity of the input queue
/// and the output queue separately, not their total.
/// </remarks>
public static IPropagatorBlock<TInput, TOutput>
    CreateBoundedTransformManyBlock<TInput, TOutput>(
    Func<TInput, IEnumerable<TOutput>> transform, int boundedCapacity)
{
    var input = new BufferBlock<TInput>(
        new DataflowBlockOptions { BoundedCapacity = boundedCapacity });
    var output = new BufferBlock<TOutput>(
        new DataflowBlockOptions { BoundedCapacity = boundedCapacity });

    Task.Run(
        async () =>
        {
            try
            {
                while (await input.OutputAvailableAsync())
                {
                    var data = transform(await input.ReceiveAsync());

                    await output.SendAllAsync(data);
                }

                output.Complete();
            }
            catch (Exception e)
            {
                ((IDataflowBlock)input).Fault(e);
                ((IDataflowBlock)output).Fault(e);
            }
        });

    return DataflowBlock.Encapsulate(input, output);
}

- svick

---

谢谢您的回复！那似乎解决了眼前的问题，虽然我仍在琢磨您的代码。特别是，这个模型以批处理模式运行--我一次获取一个“有界容量”的数据，然后它就会流出来。然后是一个新的“有界容量”。如果可能的话，我更愿意看到每个单元都被单独发送。 - Danyel Fisher

并且，稍微阐述一下@svick的评论：事实证明TransformMany仅检查输入条目之间的队列；为了使这种情况起作用，我需要在一个条目内进行检查。 - Danyel Fisher