BufferBlock missing values

Question

My BufferBlock from Dataflow library seems to miss values when the period between sending the message is low. Here is the code:

        private async static Task Main(string[] args)
        {
            await foreach (var x in Exec())
            {
                Console.WriteLine(x);
            }
        }

        public static async IAsyncEnumerable<int> Exec()
        {
            BufferBlock<int> buffer = new BufferBlock<int>();
            const int commandTime = 10;
            var tasks = Enumerable
                .Range(0, 10)
                .Select(e =>
                    Task.Run(async () =>
                    {
                        var x = new Random().Next(0, commandTime);
                        await Task.Delay(x);
                        while (!await buffer.SendAsync(x)) ;
                    }));

            var t = Task.WhenAll(tasks);

            while (!t.IsCompleted)
            {
                yield return await buffer.ReceiveAsync();
            }
        }

The await Task.Delay(x) is representing a call to external service. When I set commandTime to 10 I will get only one result (sometimes more) but when I extend possible execution time of a command (e.g. 1000) then I get all 10. Can someone explain me why the BufferBlock is not consuming values?

Answer 1

SendAsync and ReceiveAsync themselves wait for results (or a result can be sent). No loops are necessary.

I have added some output to your code to make this behavior clearer. For a better demonstration the buffer block can hold 3 elements.

public static class Program
{
    public static async Task Main()
    {
        await foreach (var x in Exec())
        {
            Console.WriteLine(x);
        }
        Console.WriteLine("READY");
    }

    public static async IAsyncEnumerable<int> Exec()
    {
        // We define a buffer block that can hold 3 elements. The fourth element will have to wait.
        BufferBlock<int> buffer = new BufferBlock<int>(new DataflowBlockOptions { BoundedCapacity = 3 });
        const int taskCount = 10;

        var tasks = Enumerable
            .Range(0, 10)
            .Select((e, number) =>
                Task.Run(async () =>
                {
                    Console.WriteLine($"Sending {number}...");
                    // From documentation: If the target
                    // postpones the offered element, the element will be buffered until such time that
                    // the target consumes or releases it, at which point the task will complete, with
                    // its System.Threading.Tasks.Task`1.Result indicating whether the message was consumed.
                    // So we don't need a waiting loop.
                    var result = await buffer.SendAsync(number);
                    Console.WriteLine($"Result of SendAsync({number}) is {result}.");
                }))
            // Now our LINQ statement will be executed and 10 Tasks will write into the buffer.
            .ToList();

        // Our writing tasks are running and we receive the results as soon as they are available.
        Console.WriteLine("Waiting for receive.");
        for (int i = 0; i < taskCount; i++)
        {
            // To demonstrate that SendAsync will wait.
            await Task.Delay(2000);
            yield return await buffer.ReceiveAsync();
        }
    }
}

Signalling completion

The receive loop knows the number of messages. If the receiver does not know this, the sender can signal completion to the BufferBlock:

public static async IAsyncEnumerable<int> Exec()
{
    BufferBlock<int> buffer = new BufferBlock<int>(new DataflowBlockOptions { BoundedCapacity = 3 });
    const int taskCount = 10;
    int sentNumbers = 0;
    var tasks = Enumerable
        .Range(0, taskCount)
        .Select((e, number) =>
            Task.Run(async () =>
            {
                var result = await buffer.SendAsync(number);
                // We have sent the last number. Now we signal to the BufferBlock that it should
                // not accept nor produce any more messages nor consume any more postponed messages.
                if (Interlocked.Add(ref sentNumbers, 1) == taskCount)
                {
                    buffer.Complete();
                }
            }))
        .ToList();

    while (await buffer.OutputAvailableAsync())
    {
        yield return await buffer.ReceiveAsync();
    }
}

Answer 2

There are several problems with your code:

1. The while (!await buffer.SendAsync(x)) loop is asking for trouble. If the SendAsync fails once to send the message, it will fail forever. The SendAsync fails to send a message only if the target block responds to the OfferMessage with DecliningPermanently. This happens when the target block has completed, either manually or as a result of an exception, and it doesn't accept any more messages. In your case the SendAsync will always succeed, so it's not a problem, but in other cases it could throw your code into an infinite loop pretty easily.
2. Creating multiple Random instances in quick succession may result to all instances being initialized with the same seed, and so producing the same random sequence of numbers. AFAIK this is a problem for .NET Framework, not for the .NET Core / .NET 5. IMHO it is safer if you use a single Random instance, and synchronize the access to it (because the Random class is not thread-safe).
3. The while (!t.IsCompleted) loop introduces a race condition. The task may be completed at a moment that the buffer still contains messages. You could try fixing it like this: while (!t.IsCompleted && buffer.Count > 0), but this would just exchange one race condition for another. These properties are not intended for controlling the execution flow. The correct way is to use a signaling mechanism, and specifically the OutputAvailableAsync method, as shown here.
4. Creating 10 tasks manually and awaiting them with await Task.WhenAll defeats the purpose of using the TPL Dataflow in the first place. This library includes powerful components that can do the same thing easier, with more options, and with better behavior in case of exceptions. Like the TransformBlock for example. Below is how I would refactor your code, to take advantage of the power of this component:

public static async IAsyncEnumerable<int> Exec()
{
    const int commandTime = 10;
    var random = new Random();

    var block = new TransformBlock<object, int>(async _ =>
    {
        int x; lock (random) x = random.Next(0, commandTime);
        await Task.Delay(x);
        return x;
    }, new ExecutionDataflowBlockOptions()
    {
        EnsureOrdered = false,
        MaxDegreeOfParallelism = 5, // Optional
    });

    // Feeder
    _ = Task.Run(async () =>
    {
        try
        {
            foreach (var _ in Enumerable.Range(0, 10))
            {
                bool accepted = await block.SendAsync(null);
                if (!accepted) break; // The block has failed
            }
            block.Complete();
        }
        catch (Exception ex)
        {
            ((IDataflowBlock)block).Fault(ex);
        }
    });

    // Emit the results as they become available
    while (await block.OutputAvailableAsync())
    {
        while (block.TryReceive(out var item))
        {
            yield return item;
        }
    }
    await block.Completion; // Propagate possible exception
}

The input messages that are sent to the TransformBlock are irrelevant in this case, so I declared the TInput as object, and I passed nulls as messages.

The // Feeder task demonstrates how to feed the TransformBlock with messages in a separate asynchronous workflow, that doesn't interfere with the results-producing loop. It is not really nessesary for this specific example, where a simple foreach (var _ in Enumerable.Range(0, 10)) block.Post(null); would suffice.

Instead of a fire-and-forget task _ = Task.Run, you could also implement the feeder as an async void method. In practice it will make no difference, but in theory the async void is the more responsible option, because it will propagate any unhandled exception instead of swallowing it.