DataflowLinkOptions.MaxMessages is ignored for a link to encapsulated block

[azyobuzin]

Area

System.Threading.Tasks.Dataflow

Version

.NET Core SDK 5.0.100-preview.3.20216.6

Summary

DataflowLinkOptions.MaxMessages is ignored when I make a link from a source block to a propagator block created by DataflowBlock.Encapsulate method. I could not find out this behavior in the documentation.

Repro

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

class Program
{
    static void Main(string[] args)
    {
        var (propagator, actualTargetBlock) = CreatePropagator();

        var outputBlock = new ActionBlock<int>(
            async i =>
            {
                Console.WriteLine(i);
                await Task.Delay(500); // Delay to postpone offer
            },
            new ExecutionDataflowBlockOptions() { BoundedCapacity = 1 });

        propagator.LinkTo(outputBlock);

        // sourceBlock emits 1, 2, 3, 4, 5
        var sourceBlock = new BufferBlock<int>();
        for (var i = 1; i <= 5; i++) sourceBlock.Post(i);

        sourceBlock.LinkTo(propagator, new DataflowLinkOptions() { MaxMessages = 2 });

        // Here works as expected:
        // sourceBlock.LinkTo(actualTargetBlock, new DataflowLinkOptions() { MaxMessages = 2 });

        Thread.Sleep(5000);
    }

    static (IPropagatorBlock<int, int>, ITargetBlock<int>) CreatePropagator()
    {
        // Set 1 to BoundedCapacity to postpone offers
        var bufferBlock = new BufferBlock<int>(new DataflowBlockOptions() { BoundedCapacity = 1 });
        var doNothingTransformBlock = new TransformBlock<int, int>(x => x, new ExecutionDataflowBlockOptions() { BoundedCapacity = 1 });
        bufferBlock.LinkTo(doNothingTransformBlock, new DataflowLinkOptions() { PropagateCompletion = true });
        var propagator = DataflowBlock.Encapsulate(bufferBlock, doNothingTransformBlock);
        return (propagator, bufferBlock);
    }
}

Expected output:

1
2

Actual output:

1
2
3
4
5

[ghost]

Tagging subscribers to this area: @tarekgh
Notify danmosemsft if you want to be subscribed.

[stephentoub]

Thanks for reporting.

I took a look at the repro, and the problem is specific to the case where the encapsulated target of the block postpones messages (this is happening in the repro because the targets are bounded, and there's enough back pressure to cause the target buffer block to postpone rather than immediately accept). When the target block then has space free up and it goes back to the original source to ask for the postponed message, the accounting logic sees it as coming from the encapsulatee rather than the encapsulator and doesn't apply the intended max messages limit.