多对多 TPL 数据流不处理所有输入答案

【问题标题】：Many to Many TPL Dataflow does not process all inputs多对多 TPL 数据流不处理所有输入
【发布时间】：2020-09-04 22:46:28
【问题描述】：

我有一个 TPL Datalow 管道，其中包含以多对多方式链接的两个源和两个目标。目标块似乎成功完成，但是，它通常会丢弃一个或多个输入。我在下面附上了我能想到的最简单的完整复制品。有什么想法吗？

注意事项：

只有在生成输入时使用人为延迟时才会出现此问题。
已成功为两个源调用 Complete()，但其中一个源的 Completion 任务挂在 WaitingForActivation 状态，即使两个目标都成功完成。
我找不到任何说明不支持多对多数据流的文档，这个问题的答案暗示它是 - https://social.msdn.microsoft.com/Forums/en-US/19d831af-2d3f-4d95-9672-b28ae53e6fa0/completion-of-complex-graph-dataflowgraph-object?forum=tpldataflow

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

class Program
{
    private const int NumbersPerSource = 10;
    private const int MaxDelayMilliseconds = 10;

    static async Task Main(string[] args)
    {
        int numbersProcessed = 0;

        var source1 = new BufferBlock<int>();
        var source2 = new BufferBlock<int>();

        var target1 = new ActionBlock<int>(i => Interlocked.Increment(ref numbersProcessed));
        var target2 = new ActionBlock<int>(i => Interlocked.Increment(ref numbersProcessed));

        var linkOptions = new DataflowLinkOptions() { PropagateCompletion = true };
        source1.LinkTo(target1, linkOptions);
        source1.LinkTo(target2, linkOptions);
        source2.LinkTo(target1, linkOptions);
        source2.LinkTo(target2, linkOptions);

        var task1 = Task.Run(() => Post(source1));
        var task2 = Task.Run(() => Post(source2));

        // source1 or source2 Completion tasks may never complete even though Complete is always successfully called.
        //await Task.WhenAll(task1, task2, source1.Completion, source2.Completion, target1.Completion, target2.Completion);
        await Task.WhenAll(task1, task2, target1.Completion, target2.Completion);

        Console.WriteLine($"{numbersProcessed} of {NumbersPerSource * 2} numbers processed.");
    }

    private static async Task Post(BufferBlock<int> source)
    {
        foreach (var i in Enumerable.Range(0, NumbersPerSource)) {
            await Task.Delay(TimeSpan.FromMilliseconds(GetRandomMilliseconds()));
            Debug.Assert(source.Post(i));
        }
        source.Complete();
    }

    private static Random Random = new Random();

    private static int GetRandomMilliseconds()
    {
        lock (Random) {
            return Random.Next(0, MaxDelayMilliseconds);
        }
    }
}

【问题讨论】：

问题在于，因为您已链接到两个源，并且正在使用 PropagateCompletion，所以第一个完成的源会关闭两个目标。所以第二个来源最终发布到一个封闭的目标块，你错过了一些数字。
啊，听起来可能是对的。唯一让我摸不着头脑的是对 source.Post 的调用返回 true。如果没有目标接受它，我认为它会返回 false。我将不得不深入研究一下，看看预期的行为是什么。谢谢！
相关：TPL Dataflow, guarantee completion only when ALL source data blocks completed，还有TPL Dataflow: process messages from two incoming blocks sequentially。
更大的问题可能是你想通过发布多对多这样的方式来解决什么问题？
我有金融股票数据流，我将其拆分为多个数据流，每个数据流处理不同的指标（移动平均线等）。然后将这些整理回一个数据流，所有数据点都被发送到交易策略以生成交易信号。还有其他数据源，如新闻提要和数据库数据，也被整理/加入到最终输出中，但股票代码数据被多次计算使用，因此这是多对多数据流的主要原因。

标签： c# task-parallel-library tpl-dataflow

【解决方案1】：

正如@MikeJ 在comment 中指出的那样，在多对多数据流配置中使用PropagateCompletion 链接块可能会导致某些目标块过早完成。在这种情况下，target1 和 target2 在两个源块中的任何一个完成时都被标记为已完成，而另一个源无法完成，因为它的输出缓冲区中仍有消息。这些消息永远不会被消费，因为没有一个链接的目标块愿意接受它们。

要解决此问题，您可以使用下面的自定义 PropagateCompletion 方法：

public static async void PropagateCompletion(IDataflowBlock[] sources,
    IDataflowBlock[] targets)
{
    // Arguments validation omitted
    Task allSourcesCompletion = Task.WhenAll(sources.Select(s => s.Completion));

    try { await allSourcesCompletion.ConfigureAwait(false); } catch { }

    var exception = allSourcesCompletion.IsFaulted ?
        allSourcesCompletion.Exception : null;

    foreach (var target in targets)
    {
        if (exception != null) target.Fault(exception); else target.Complete();
    }
}

使用示例：

source1.LinkTo(target1);
source1.LinkTo(target2);
source2.LinkTo(target1);
source2.LinkTo(target2);
PropagateCompletion(new[] { source1, source2 }, new[] { target1, target2 });

请注意，在此示例中将源链接到目标时，没有传递 DataflowLinkOptions。

【讨论】：