System.Threading.Tasks.Dataflow - TPL - Assist Request - Cannot Find Cause Of Deadlock

[Sn3akyP3t3]

I'm trying to explore use of the System.Threading.Tasks.Dataflow library with async calls. I found this example helpful when trying to integrate BatchBlock and TransformBlock together
https://stackoverflow.com/questions/31804794/tpl-dataflow-transform-block-post-to-batch-block-followed-by-actionblock
However, it seems that I have caused a deadlock and I can't seem to identify the cause. I've put together a fully functional repo to demonstrate my issue. Interestingly the pipeline seems to complete when observing through debugging, but for some reason doesn't make it past this line:
await _getLastDataBlock.Completion; // This appears to encounter a deadlock

Appreciate any help or tips given!

[elgonzo]

Unless you complete the BatchBlock, _ => _getLastDataBlock.Complete() won't be executed...
(Note how the code example in the SO post you linked to calls batchBlock.Complete() which then leads to obj => uploadFilesToAzureBlock.Complete() being called.)

[Sn3akyP3t3]

Apologies, I had some build errors that I corrected, added logging, and set it back to Greedy which was unintentionally flipped to false when I was previously testing, but forgot to revert. The main branch behaves as I advertised with my initial post.

I made a new branch to explore your suggestion and encountered an exception after the 4th calling of CallFirstEndpointAsync which matches the Batch Block size.

[elgonzo]

What is the exception. What's its stack trace (where does it originate from)?

[elgonzo]

Ah, sorry, didn't pay close attention to your source code. I got distracted by your code featuring something like _batchBlock.Completion.ContinueWith(_ => _getLastDataBlock.Complete()); and then made assumptions based on that without looking at the rest of your code.

The issue with your original code is not batchBlock.Complete() not being called. You linked each of the blocks with PropagateCompletion = true, so it would not be necessary to call Complete() on any block other than _getFirstDataBlock. (This also means the line _batchBlock.Completion.ContinueWith(_ => _getLastDataBlock.Complete()); is not necessary either).

The problem is with your last block in your pipeline being a TransformBlock, which not only is a target but also functions as a source (that's being left unconnected in this case). I'd suggest to either link an ActionBlock after _getLastDataBlock and await the completion of the ActionBlock, or turn _getLastDataBlock itself into an ActionBlock.

For example something like this:

    ActionBlock<List<Data>> _outputBlock = new(Output);

    ...

    _getFirstDataBlock.LinkTo(_batchBlock, linkOptions);
    _batchBlock.LinkTo(_getLastDataBlock, linkOptions);
    _getLastDataBlock.LinkTo(_outputBlock, linkOptions);

    ...

public async Task DoProcess()
{
    var originalData = GenerateData();
    
    foreach (var dataSet in originalData)
    {
        await _getFirstDataBlock.SendAsync(dataSet);
    }

    _getFirstDataBlock.Complete();
    
    await _outputBlock.Completion;    

    string validate = string.Empty;
}

    ...

private void Output(List<Data> data)
{
    foreach (var dataSet in data)
    {
        Console.WriteLine($"FakedId: {dataSet.FakedId}; MasterFakedId: {dataSet.MasterFakedId}; Days: {dataSet.Days}; Payload: {dataSet.Payload};");
    }
}

If for whatever reason you can't use an ActionBlock but need to stick with a TransformBlock as the final block in your pipeline, you'll need to drain it before it completes, like for example something like this:

    ...
    _getFirstDataBlock.Complete();

    IAsyncEnumerable<List<Data>> dataSetsBundle = _getLastDataBlock.ReceiveAllAsync();
    await foreach (var dataSets in dataSetsBundle)
    {
        foreach (var dataSet in dataSets)
        {
            Console.WriteLine($"FakedId: {dataSet.FakedId}; MasterFakedId: {dataSet.MasterFakedId}; Days: {dataSet.Days}; Payload: {dataSet.Payload};");
        }
    }
    
    //
    // Although _getLastDataBlock is completed when the async foreach loop above concludes, it
    // might have been completed in a cancelled or faulty state. To ensure exceptions and such
    // related to cancellation or faulty completion are propagated by the async state machine of the
    // async DoProcess method, _getLastDataBlock.Completion is still being awaited after the loop.
    //
    await _getLastDataBlock.Completion;

[Sn3akyP3t3]

You've managed to find where my lack of knowledge and experience was hiding. Thank you! I lacked the concept of the drain or sink which was key in resolving this deadlock. I can't seem to find this concept or practice discussed in documentation, but I've seen plenty of examples suggesting the last block in the workflow be an ActionBlock. This might be a reason why the struggle I encountered is not common. The working version is in this branch.

Again, thank you for helping me figure this out. I would have struggled for awhile to understand what was going on. I do believe that this library doesn't have nearly enough practical examples of how one might put it to use. That goes for both Microsoft and private hosted tutorials (some just copy Microsoft) and they almost always use simple data types like text or number processing and almost always terminate with an ActionBlock using simple Console.Write or write to file operations in there. I'll try to contribute to at least one more demonstration of how to use this library other than this sloppy one that I've posted to work through this issue.

On to the next hurdle I guess. Looks like Dataflow blocks are not re-usable. I'll change that to encapsulate it in a class and see if that will work for the intended use.