TPL Dataflow's BufferBlock bug with OutputAvailableAsync after TryReceiveAll

[i3arnon]

If you call OutputAvailableAsync after TryReceiveAll more than once you get back a task that never completes. No matter how many items are added to the BufferBlock.

The following sample reproduces the issue:

var buffer = new BufferBlock<object>();
var producer = Task.Run(async () =>
{
    while (true)
    {
        await Task.Delay(TimeSpan.FromMilliseconds(100));
        buffer.Post(null);
        Console.WriteLine("Post " + buffer.Count);
    }
});
var consumer = Task.Run(async () =>
{
    while (await buffer.OutputAvailableAsync())
    {
        IList<object> items;
        buffer.TryReceiveAll(out items);
        Console.WriteLine("TryReceiveAll " + buffer.Count);
    }
});
await Task.WhenAll(consumer, producer);

I have previously documented this bug in my stackoverflow question (BufferBlock deadlock with TryReceiveAll) and on VS connect (TPL Dataflow's BufferBlock possible deadlock with TryReceiveAll)

[RaringCoder]

I just had this exact problem (I can translate from TryReceiveAll to while (TryReceive) and it works), yet this was apparently fixed years ago. Has it regressed?

[stephentoub]

Has it regressed?

It shouldn't have. What build are you using? Can you share a repro?

[RaringCoder]

Hi @stephentoub ,

I'm using v4.5.24 of the Microsoft.Tpl.Dataflow nuget package.

My code is functionality equivalent to the Stephen Cleary example: https://blog.stephencleary.com/2012/11/async-producerconsumer-queue-using.html

However, because I don't want my consumer to receive one message at a time, rather, to take all messages and process the batch. I am using the BufferBlock to basically let the requests fly in until the consumer is ready to process again as the consumer task is a database query and zip file upload to blob store. I therefore amended the consumer to use TryReceiveAll. However, on returning around the loop to OutputAvailableAsync it blocks forever. My consumer looks like this (it behaves as expected):

    private async Task ConsumeAsync()
    {
        while (await _generateFileRequests.OutputAvailableAsync().ConfigureAwait(false))
        {
            // Seems to be an issue with this (still): https://github.com/dotnet/corefx/issues/202
            //_generateFileRequests.TryReceiveAll(out var completionSources);

            var completionSources = new List<TaskCompletionSource<object>>();

            while (_generateFileRequests.TryReceive(out var tcs))
            {
                completionSources.Add(tcs);
            }

            await GenerateTillDataFileAsync().ConfigureAwait(false);

            foreach (var completionSource in completionSources)
            {
                // Notify completion.
                completionSource.SetResult(null);
            }
        }
    }

[RaringCoder]

I would actually much rather achieve this without the open-ended while loop and non-completing pipeline, but I guess that's a Stack Overflow question :-)

This way works because it is more of a "pull" action from the consumer, rather than a push action by the producer.

[svick]

@RaringCoder

I'm using v4.5.24 of the Microsoft.Tpl.Dataflow nuget package.

That package is ancient, it was last updated in 2014, before this issue was fixed. You should instead use System.Threading.Tasks.Dataflow. Latest version of that package is 4.8.0 from August of this year.

[RaringCoder]

@svick Ah well that answers that riddle. Hadn't noticed the dates if I'm being honest, searched "dataflow" and it was the top result lol, ta.

[RaringCoder]

@stephentoub @svick Yeah that was it, thanks. Made the change back to the TryReceiveAll method and tidied the code up a little. Not regressed 👍

[stephentoub]

Not regressed

Great. Thanks for confirming.