BufferBlock deadlock with OutputAvailableAsync aft

问题:

While working on an answer to this question, I wrote this snippet:

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);

The producer should post items to the buffer every 100 ms and the consumer should clear all items out of the buffer and asynchronously wait for more items to show up.

What actually happens is that the producer clears all items once, and then never again moves beyond OutputAvailableAsync. If I switch the consumer to remove items one by one it works as excepted:

while (await buffer.OutputAvailableAsync())
{
    object item;
    while (buffer.TryReceive(out item)) ;
}

Am I misunderstanding something? If not, what is the problem?

回答1:

This is a bug in SourceCore being used internally by BufferBlock. Its TryReceiveAll method doesn't turn on the _enableOffering boolean data member while TryReceive does. That results in the task returned from OutputAvailableAsync never completing.

Here's a minimal reproduce:

var buffer = new BufferBlock<object>();
buffer.Post(null);

IList<object> items;
buffer.TryReceiveAll(out items);

var outputAvailableAsync = buffer.OutputAvailableAsync();
buffer.Post(null);

await outputAvailableAsync; // Never completes

I've just fixed it in the .Net core repository with this pull request. Hopefully the fix finds itself in the nuget package soon.

回答2:

Alas, it's the end of September 2015, and although i3arnon fixed the error it is not solved in the version that was released two days after the error was fixed: Microsoft TPL Dataflow version 4.5.24.

However IReceivableSourceBlock.TryReceive(...) works correctly. An extension method will solve the problem. After a new release of TPL Dataflow it will be easy to change the extension method.

/// <summary>
/// This extension method returns all available items in the IReceivableSourceBlock
/// or an empty sequence if nothing is available. The functin does not wait.
/// </summary>
/// <typeparam name="T">The type of items stored in the IReceivableSourceBlock</typeparam>
/// <param name="buffer">the source where the items should be extracted from </param>
/// <returns>The IList with the received items. Empty if no items were available</returns>
public static IList<T> TryReceiveAllEx<T>(this IReceivableSourceBlock<T> buffer)
{
    /* Microsoft TPL Dataflow version 4.5.24 contains a bug in TryReceiveAll
     * Hence this function uses TryReceive until nothing is available anymore
     * */
    IList<T> receivedItems = new List<T>();
    T receivedItem = default(T);
    while (buffer.TryReceive<T>(out receivedItem))
    {
        receivedItems.Add(receivedItem);
    }
    return receivedItems;
}

usage:

while (await this.bufferBlock.OutputAvailableAsync())
{
    // some data available
    var receivedItems = this.bufferBlock.TryReceiveAllEx();
    if (receivedItems.Any())
    {
        ProcessReceivedItems(bufferBlock);
    }
}