Report when input to first dataflow block finishes all linked blocks

user_5436583 0
Viewed 126

I am using TPL Dataflow to download data from a ticketing system.

The system takes the ticket number as the input, calls an API and receives a nested JSON response with various information. Once received, a set of blocks handles each level of the nested structure and writes it to a relational database. e.g. Conversation, Conversation Attachments, Users, User Photos, User Tags, etc

Json

{
    "conversations":[
        {
            "id":12345,
            "author_id":23456,
            "body":"First Conversation"
        },
        {
            "id":98765,
            "authorid":34567,
            "body":"Second Conversation",
            "attachments":[
            {
                "attachmentid":12345
                "attachment_name":"Test.jpg"
            }
        }
    ],
    "users":[
        {
            "userid":12345
            "user_name":"John Smith"
        },
        {
            "userid":34567
            "user_name":"Joe Bloggs"
            "user_photo":
            {
                "photoid":44556,
                "photo_name":"headshot.jpg"
            },
            tags:[
                "development",
                "vip"
            ]
        }
    ]

Code

Some blocks need to broadcast so that deeper nesting can still have access to the data. e.g. UserModelJson is broadcast so that 1 block can handle writing the user, 1 block can handle writing the User Tags and 1 block can handle writing the User Photos.

var loadTicketsBlock = new TransformBlock<int, ConversationsModelJson>(async ticketNumber => await p.GetConversationObjectFromTicket(ticketNumber));
var broadcastConversationsObjectBlock = new BroadcastBlock<ConversationsModelJson>(conversations => conversations);

//Conversation
var getConversationsFromConversationObjectBlock = new TransformManyBlock<ConversationsModelJson, ConversationModelJson>(conversation => ModelConverter.ConvertConversationsObjectJsonToConversationJson(conversation));
var convertConversationsBlock = new TransformBlock<ConversationModelJson, ConversationModel>(conversation => ModelConverter.ConvertConversationJsonToConversation(conversation));
var batchConversionBlock = new BatchBlock<ConversationModel>(batchBlockSize);
var convertConversationsToDTBlock = new TransformBlock<IEnumerable<ConversationModel>, DataTable>(conversations => ModelConverter.ConvertConversationModelToConversationDT(conversations));
var writeConversationsBlock = new ActionBlock<DataTable>(async conversations => await p.ProcessConversationsAsync(conversations));

var getUsersFromConversationsBlock = new TransformManyBlock<ConversationsModelJson, UserModelJson>(conversations => ModelConverter.ConvertConversationsJsonToUsersJson(conversations));
var broadcastUserBlock = new BroadcastBlock<UserModelJson>(userModelJson => userModelJson);

//User
var convertUsersBlock = new TransformBlock<UserModelJson, UserModel>(user => ModelConverter.ConvertUserJsonToUser(user));
var batchUsersBlock = new BatchBlock<UserModel>(batchBlockSize);
var convertUsersToDTBlock = new TransformBlock<IEnumerable<UserModel>, DataTable>(users => ModelConverter.ConvertUserModelToUserDT(users));
var writeUsersBlock = new ActionBlock<DataTable>(async users => await p.ProcessUsersAsync(users));

//UserTag
var getUserTagsFromUserBlock = new TransformBlock<UserModelJson, UserTagModel>(user => ModelConverter.ConvertUserJsonToUserTag(user));
var batchTagsBlock = new BatchBlock<UserTagModel>(batchBlockSize);
var convertTagsToDTBlock = new TransformBlock<IEnumerable<UserTagModel>, DataTable>(tags => ModelConverter.ConvertUserTagModelToUserTagDT(tags));
var writeTagsBlock = new ActionBlock<DataTable>(async tags => await p.ProcessUserTagsAsync(tags));


DataflowLinkOptions linkOptions = new DataflowLinkOptions()
{
    PropagateCompletion = true
};

loadTicketsBlock.LinkTo(broadcastConversationsObjectBlock, linkOptions);

//Conversation
broadcastConversationsObjectBlock.LinkTo(getConversationsFromConversationObjectBlock, linkOptions);
getConversationsFromConversationObjectBlock.LinkTo(convertConversationsBlock, linkOptions);
convertConversationsBlock.LinkTo(batchConversionBlock, linkOptions);
batchConversionBlock.LinkTo(convertConversationsToDTBlock, linkOptions);
convertConversationsToDTBlock.LinkTo(writeConversationsBlock, linkOptions);         

var tickets = await provider.GetAllTicketsAsync();

foreach (var ticket in tickets)
{
    cts.Token.ThrowIfCancellationRequested();
    await loadTicketsBlock.SendAsync(ticket.TicketID);
}

loadTicketsBlock.Complete();

The LinkTo blocks are repeated for each type of data to be written.

I know when the whole pipeline is complete by using

await Task.WhenAll(<Last block of each branch>.Completion);

but if I pass in ticket number 1 into the loadTicketsBlock block then how do I know when that specific ticket has been through all blocks in the pipeline and therefore is complete?

The reason that I want to know this is so that I can report to the UI that ticket 1 of 100 is complete.

tpl-dataflow c#
2 Answers

If I am tracking progress in Dataflow, usually I will set up the last block as a notify the UI of progress type block. To be able to track the progress of your inputs, you will need to keep the context of the original input in all the objects you are passing around, so in this case you need to be able to tell that you are working on ticket 1 all the way through your pipeline, and if one of your transforms removes the context that it is working on ticket 1, then you will need to rethink the object types that you are passing through your pipeline so you can keep that context.

A simple example of what I'm talking about is laid out below with a broadcast block going to three transform blocks, and then all three transform blocks going back to an action block that notifies about the progress of the pipelines.

When combining back into the single action block you need to make sure not to propagate completion at that point because as soon as one block propagates completion to the action block, that action block will stop accepting input, so you will still wait for the last block of each pipeline to complete, and then after that manually complete your final notify the UI action block.

using System;
using System.Threading.Tasks.Dataflow;
using System.Threading.Tasks;
using System.Collections.Generic;

public class Program
{
    public static void Main()
    {
        var broadcastBlock = new BroadcastBlock<string>(x => x);
        
        var transformBlockA = new TransformBlock<string, string>(x =>
        {
            return x + "A";
        });
        
        var transformBlockB = new TransformBlock<string, string>(x =>
        {
            return x + "B";
        });
        
        var transformBlockC = new TransformBlock<string, string>(x =>
        {
            return x + "C";
        });
        
        var ticketTracking = new Dictionary<int, List<string>>();
        var notifyUiBlock = new ActionBlock<string>(x =>
        {
            var ticketNumber = int.Parse(x.Substring(5,1));
            var taskLetter = x.Substring(7,1);
            var success = ticketTracking.TryGetValue(ticketNumber, out var tasksComplete);
            if (!success)
            {
                tasksComplete = new List<string>();
                ticketTracking[ticketNumber] = tasksComplete;
            }
            tasksComplete.Add(taskLetter);
            
            if (tasksComplete.Count == 3)
            {
                Console.WriteLine($"Ticket {ticketNumber} is complete");
            }
        });
        
        DataflowLinkOptions linkOptions = new DataflowLinkOptions() {PropagateCompletion = true};
        
        broadcastBlock.LinkTo(transformBlockA, linkOptions);
        broadcastBlock.LinkTo(transformBlockB, linkOptions);
        broadcastBlock.LinkTo(transformBlockC, linkOptions);
        transformBlockA.LinkTo(notifyUiBlock);
        transformBlockB.LinkTo(notifyUiBlock);
        transformBlockC.LinkTo(notifyUiBlock);
        
        for(var i = 0; i < 5; i++)
        {
            broadcastBlock.Post($"Task {i} ");
        }
        
        broadcastBlock.Complete();
        
        Task.WhenAll(transformBlockA.Completion, transformBlockB.Completion, transformBlockC.Completion).Wait();
        notifyUiBlock.Complete();
        notifyUiBlock.Completion.Wait();
        
        Console.WriteLine("Done");
    }
}

This will give an output similar to this

Ticket 0 is complete
Ticket 1 is complete
Ticket 2 is complete
Ticket 3 is complete
Ticket 4 is complete
Done

user_47081500

You could consider using the TaskCompletionSource as the base class for all your sub-entities. For example:

class Attachment : TaskCompletionSource
{
}

class Conversation : TaskCompletionSource
{
}

Then every time you insert a sub-entity in the database, you mark it as completed:

attachment.SetResult();

...or if the insert fails, mark it as faulted:

attachment.SetException(ex);

Finally you can combine all the asynchronous completions in one, with the method Task.WhenAll:

Task ticketCompletion = Task.WhenAll(Enumerable.Empty<Task>()
    .Append(ticket.Task)
    .Concat(attachments.Select(e => e.Task))
    .Concat(conversations.Select(e => e.Task)));
user_111785490
Related