TPL Dataflow, can I query whether a data block is marked complete but has not yet completed?

Question

Given the following:

BufferBlock<int> sourceBlock = new BufferBlock<int>();
TransformBlock<int, int> targetBlock = new TransformBlock<int, int>(element =>
{
    return element * 2;
});

sourceBlock.LinkTo(targetBlock, new DataflowLinkOptions { PropagateCompletion = true });

//feed some elements into the buffer block
for(int i = 1; i <= 1000000; i++)
{
    sourceBlock.SendAsync(i);
}

sourceBlock.Complete();

targetBlock.Completion.ContinueWith(_ =>
{
    //notify completion of the target block
});

The targetBlock never seems to complete and I think the reason is that all the items in the TransformBlock targetBlock are waiting in the output queue as I have not linked the targetBlock to any other Dataflow block. However, what I actually want to achieve is a notification when (A) the targetBlock is notified of completion AND (B) the input queue is empty. I do not want to care whether items still sit in the output queue of the TransformBlock. How can I go about that? Is the only way to get what I want to query the completion status of the sourceBlock AND to make sure the InputCount of the targetBlock is zero? I am not sure this is very stable (is the sourceBlock truly only marked completed if the last item in the sourceBlock has been passed to the targetBlock?). Is there a more elegant and more efficient way to get to the same goal?

Edit: I just noticed even the "dirty" way to check on completion of the sourceBlock AND InputCount of the targetBlock being zero is not trivial to implement. Where would that block sit? It cannot be within the targetBlock because once above two conditions are met obviously no message is processed within targetBlock anymore. Also checking on the completion status of the sourceBlock introduces a lot of inefficiency.

Answer 1

I believe you can't directly do this. It's possible you could get this information from some private fields using reflection, but I wouldn't recommend doing that.

But you can do this by creating custom blocks. In the case of Complete() it's simple: just create a block that forwards each method to the original block. Except Complete(), where it will also log it.

In the case of figuring out when processing of all items is complete, you could link your block to an intermediate BufferBlock. This way, the output queue will be emptied quickly and so checking Completed of the internal block would give you fairly accurate measurement of when the processing is complete. This would affect your measurements, but hopefully not significantly.

Another option would be to add some logging at the end of the block's delegate. This way, you could see when processing of the last item was finished.

Answer 2

It would be nice if the TransformBlock had a ProcessingCompleted event that would fire when the block has completed the processing of all messages in its queue, but there is no such event. Below is an attempt to rectify this omission. The CreateTransformBlockEx method accepts an Action<Exception> handler, that is invoked when this "event" occurs.

The intention was to always invoke the handler before the final completion of the block. Unfortunately in the case that the supplied CancellationToken is canceled, the completion (cancellation) happens first, and the handler is invoked some milliseconds later. To fix this inconsistency would require some tricky workarounds, and may had other unwanted side-effects, so I am leaving it as is.

public static IPropagatorBlock<TInput, TOutput>
    CreateTransformBlockEx<TInput, TOutput>(Func<TInput, Task<TOutput>> transform,
    Action<Exception> onProcessingCompleted,
    ExecutionDataflowBlockOptions dataflowBlockOptions = null)
{
    if (onProcessingCompleted == null)
        throw new ArgumentNullException(nameof(onProcessingCompleted));
    dataflowBlockOptions = dataflowBlockOptions ?? new ExecutionDataflowBlockOptions();

    var transformBlock = new TransformBlock<TInput, TOutput>(transform,
        dataflowBlockOptions);
    var bufferBlock = new BufferBlock<TOutput>(dataflowBlockOptions);

    transformBlock.LinkTo(bufferBlock);
    PropagateCompletion(transformBlock, bufferBlock, onProcessingCompleted);
    return DataflowBlock.Encapsulate(transformBlock, bufferBlock);

    async void PropagateCompletion(IDataflowBlock block1, IDataflowBlock block2,
        Action<Exception> completionHandler)
    {
        try
        {
            await block1.Completion.ConfigureAwait(false);
        }
        catch { }
        var exception = 
            block1.Completion.IsFaulted ? block1.Completion.Exception : null;
        try
        {
            // Invoke the handler before completing the second block
            completionHandler(exception);
        }
        finally
        {
            if (exception != null) block2.Fault(exception); else block2.Complete();
        }
    }
}

// Overload with synchronous lambda
public static IPropagatorBlock<TInput, TOutput>
    CreateTransformBlockEx<TInput, TOutput>(Func<TInput, TOutput> transform,
    Action<Exception> onProcessingCompleted,
    ExecutionDataflowBlockOptions dataflowBlockOptions = null)
{
    return CreateTransformBlockEx<TInput, TOutput>(
        x => Task.FromResult(transform(x)), onProcessingCompleted,
        dataflowBlockOptions);
}

The code of the local function PropagateCompletion mimics the source code of the LinkTo built-in method, when invoked with the PropagateCompletion = true option.

Usage example:

var httpClient = new HttpClient();
var downloader = CreateTransformBlockEx<string, string>(async url =>
{
    return await httpClient.GetStringAsync(url);
}, onProcessingCompleted: ex =>
{
    Console.WriteLine($"Download completed {(ex == null ? "OK" : "Error")}");
}, new ExecutionDataflowBlockOptions()
{
    MaxDegreeOfParallelism = 10
});

Answer 3

First thing it is not right to use a IPropagator Block as a leaf terminal. But still your requirement can be fulfilled by asynchronously checking the output buffer of the TargetBlock for output messages and then consuming then so that the buffer could be emptied.

    `  BufferBlock<int> sourceBlock = new BufferBlock<int>();
       TransformBlock<int, int> targetBlock = new TransformBlock<int, int> 
       (element =>
       {

        return element * 2;
        });
        sourceBlock.LinkTo(targetBlock, new DataflowLinkOptions { 
        PropagateCompletion = true });

        //feed some elements into the buffer block
        for (int i = 1; i <= 100; i++)
        {
             sourceBlock.SendAsync(i);
        }

        sourceBlock.Complete();

        bool isOutputAvailable = await targetBlock.OutputAvailableAsync();
        while(isOutputAvailable)
        {
            int value = await targetBlock.ReceiveAsync();

            isOutputAvailable = await targetBlock.OutputAvailableAsync();
        }


        await targetBlock.Completion.ContinueWith(_ =>
        {
            Console.WriteLine("Target Block Completed");//notify completion of the target block
        });

`