c# - 链接数据流 block 如何取消目标 block

Question

在 TPL 数据流中，当一个 block 通过传播链接到另一个 block 时，它会转发异常和取消。我可以想象转发异常只需使用 dataFlowBlock.Fault(exception) 即可完成，但我很好奇取消是如何转发的，因为没有 dataFlowBlock 这样的东西。取消()。它是通过相同的 Fault() 方法使用传递 TaskCancelledException 作为参数完成的吗？

更新:

为了澄清，请考虑以下示例，其中只有 block1 是通过选项使用 CancelationToken 创建的，而 block2 不是。 Block1 通过传播链接到 block2:

block1 { CancellationToken = ct } -> block2 { }

当 ct 收到取消请求时，block1 完成转换为取消。我的问题是此时 block2 会发生什么？ block1 是否主动取消 block2，如果是，它是否使用 block.Fault(TaskCanceledException) 来取消？或者它是否使用了一些内部的 ocus-pocus 来神奇地取消 block2，即使它是在没有取消 token 的情况下创建的？

Answer 1

好的，我认为我们与更新后的帖子在同一页面上。简而言之，如果传播不正确，取消不会流向链接 block 。

[TestFixture]
public class DataFlowTests
{

    [Test]
    public async Task DataflowTest()
    {
        var cts = new CancellationTokenSource();
        var buffer = new BufferBlock<int>(new DataflowBlockOptions() { BoundedCapacity = 200, CancellationToken = cts.Token });
        var action = new ActionBlock<int>(x => Task.Delay(100), new ExecutionDataflowBlockOptions() { BoundedCapacity = 5 });
        buffer.LinkTo(action, new DataflowLinkOptions() { PropagateCompletion = false});

        foreach (var data in Enumerable.Range(0, 20))
        {
            if (data > 10) break;
            await buffer.SendAsync(data);
        }
        cts.Cancel();
        //action.Complete();
        await action.Completion;
        Console.WriteLine(buffer.Completion.Status);
        Console.WriteLine(action.Completion.Status);
    }
}

该示例将永远挂起等待 action去完成。现在打电话Complete() ActionBlock<> 上的明确性产生这些结果状态:

Cancelled - buffer
RanToCompletion - action

最终传播完成会产生相同的结果，而无需在下游 block 上手动调用完成:

[TestFixture]
public class DataFlowTests
{

    [Test]
    public async Task DataflowTest()
    {
        var cts = new CancellationTokenSource();
        var buffer = new BufferBlock<int>(new DataflowBlockOptions() { BoundedCapacity = 200, CancellationToken = cts.Token });
        var action = new ActionBlock<int>(x => Task.Delay(100), new ExecutionDataflowBlockOptions() { BoundedCapacity = 5 });
        buffer.LinkTo(action, new DataflowLinkOptions() { PropagateCompletion = true});

        foreach (var data in Enumerable.Range(0, 20))
        {
            if (data > 10) break;
            await buffer.SendAsync(data);
        }
        cts.Cancel();
        await action.Completion;
        Console.WriteLine(buffer.Completion.Status);
        Console.WriteLine(action.Completion.Status);
    }
}

yield 状态:

Canceled - buffer
RanToCompletion - action

请注意，取消不会导致整个管道被取消。其他 block 简单地完成。现在，如果通过取消通知以外的异常，则管道将通过 ...Fault(..) 出现故障。 .否则它会跳出标准完成传播。