Microsoft TPL Dataflow - processing correlating requests synchronously

Question

I apologize in advance for the title but it's the best I could think of to describe the action.

The requirement is processing requests of a message bus. Requests coming in maybe related to an id which correlates or groups these requests. The behavior I want is that for a stream of requests to process correlating ids synchronously. However different ids can be processed asynchronously.

I am using a concurrentdictionary to track the request being processed and the predicate in the linkto.

This is suppose to provide the synchronous processing of related requests.

However the behavior I get is that the first request gets processed and the second request gets dropped.

I have attached the sample code from a console application to simulate the issue.

Any direction or feedback will be appreciated.

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;

namespace ConsoleApplication2
{
    class Program
    {
        static void Main(string[] args)
        {
            var requestTracker = new ConcurrentDictionary<string, string>();

            var bufferBlock = new BufferBlock<Request>();

            var actionBlock = new ActionBlock<Request>(x => 
            {
                Console.WriteLine("processing item {0}",x.Name);
                Thread.Sleep(5000);
                string itemOut = null;
                requestTracker.TryRemove(x.Id, out itemOut);
            });

            bufferBlock.LinkTo(actionBlock, x => requestTracker.TryAdd(x.Id,x.Name));


            var publisher = Task.Run(() =>
            {
                var request = new Request("item_1", "first item");
                bufferBlock.SendAsync(request);

                var request_1 = new Request("item_1", "second item");
                bufferBlock.SendAsync(request_1);

            });

            publisher.Wait();
            Console.ReadLine();
        }
    }

    public class Request
    {
        public Request(string id, string name)
        {
            this.Id = id;
            this.Name = name;
        }
        public string Id { get; set; }
        public string Name { get; set; }
    }
}

Answer 1

1. You say you want to process some requests in parallel (at least I assume that's what you meant by “asynchronously”), but ActionBlock is not parallel by default. To change that, set MaxDegreeOfParallelism.

2. You're trying to use TryAdd() as the filter, but that won't work for two reasons:

   1. The filter is invoked only once, it's not automatically retried or anything like that. That means that if an item doesn't go through, it would never go through, even after the item that was blocking it was completed.
   2. If an item is stuck in the output queue of a block, no other items are going to get out of that block. This could significantly reduce the level of parallelism, even if you somehow worked around the previous issue.

3. I think the simplest solution here would be to have a block for each group, that way, items from each group will be processed sequentially, but items from different groups will be processed in parallel. In code, it could look something like:

   var processingBlocks = new Dictionary<string, ActionBlock<Request>>();
   
   var splitterBlock = new ActionBlock<Request>(request =>
   {
       ActionBlock<Request> processingBlock;
   
       if (!processingBlocks.TryGetValue(request.Id, out processingBlock))
       {
           processingBlock = processingBlocks[request.Id] =
               new ActionBlock<Request>(r => /* process the request here */);
       }
   
       processingBlock.Post(request);
   });
   

   The issue with this approach is that the processing blocks for groups never go away. If you can't afford that (it's a memory leak), because you're going to have a large amount of groups, then the hashing approach suggested by I3arnon is the way to go.

Answer 2

I believe this is because your LinkTo() is not set up properly. By having a LinkTo() and passing a function as an argument you are adding in a condition. So this line:

bufferBlock.LinkTo(actionBlock, x => requestTracker.TryAdd(x.Id, x.Name));

Is essentially saying, pass data from the bufferBlock to the actionBlock IF you are able to add to your concurrent dictionary, which doesn't necessarily make sense (at least in your sample code)

Instead, you should link your bufferBlock to the actionblock without the lambda, since you don't need conditional linking in this situation (at least I don't think so based on your sample code).

Also, take a look at this SO question to see if you should be using SendAsync() or Post() since Post() can be easier to handle for simply adding data into the pipeline: TPL Dataflow, whats the functional difference between Post() and SendAsync()? . SendAsync will return a task, while Post will return true/false based on success entering the pipeline.

So to essentially find out what's going wrong you need to handle the continuations of your blocks. There is a nice tutorial over at MSDN in their TPL Dataflow introduction over here: Create a DataFlow Pipeline It would essentially look like this:

//link to section
bufferBlock.LinkTo(actionBlock);
//continuations
bufferBlock.Completion.ContinueWith(t =>
{
     if(t.IsFaulted)  ((IDataFlowBlock).actionBlock).Fault(t.Exception); //send the exception down the pipeline
     else actionBlock.Complete(); //tell the next block that we're done with the bufferblock
 });

You can then catch the exception (AggregateException) when waiting for the pipeline. Do you really need to use the concurrentdictionary in your actual code for the tracking, because that could be causing the issue when it fails to add, since when the linkto predicate returns false, it does not pass the data into the next block of the pipeline.