1

We are trying to rewrite a session management system as our legacy system can't scale up anymore and is quite buggy.

I'll briefly explain our system, we have huge number of machines on which end users create a session and our system is supposed to track what happens during a session. Any activity that an user does on the machine is posted as a RabbitMQ message. The system consumes the message, processes it and updates the database(Hekaton in our case).

The difficulties in our system is that, the message processing for each machine should be ordered and we can't miss any of the messages. To achieve this, we have used RabbitMQ's consistent hash exchange as well.

Our CTO has asked us to use TPL dataflow library in our product and I'm not sure if we need the library, basically the code right now is,

  1. Consume a message
  2. Get additional info needed for processing a message from other microservices and in-memory caches.
  3. Perform calculations.
  4. Commit the changes to an SQL Server memory optimized table.
  5. Publish integration events to other microservices.
  6. Acknowledge the message.

We have used async-await wherever possible and the issues we come up with is messages for a certain machine getting processed before the previous message has completed. This causes the processing to be out of order. If we remove all the async-await and make the code synchronous, it's going to cause performance issues. The application is expected to be highly performant and expects a huge load as well.

Does my business case need TPL dataflow? I'm just trying to learn about TPL dataflow and I'm having trouble understanding how splitting up the above steps in to "blocks" is going to solve my problems.

public async Task<bool> ConsumeMessage(IRabbitMQConsumeContext<EventDto> context)
    {
        var message = context.Message.Body;
        var command = new UpdateCommand()
        {
            //Build object
        };

        Result<bool> result;
        using (var scope = _serviceProvider.CreateScope())
        {
            var mediator = scope.ServiceProvider.GetRequiredService<IMediator>();
            result = await mediator.Send(command, default(CancellationToken));
        }
        if (result.IsFailure)
        {
            Nack(context.Message);
            return false;
        }
        Ack(context.Message);
        return true;
    }
2
  • 1
    please don't cross-post: stackoverflow.com/questions/78089524/… "Cross-posting is frowned upon as it leads to fragmented answers splattered all over the network..."
    – gnat
    Commented Mar 1 at 18:45
  • It's like load-balancing. All messages from a certain machine M = M0 should be routed to the same consumer C0 = C(hash(M0)).
    – rwong
    Commented Mar 1 at 19:23

2 Answers 2

3

TPL data flow is based around producers and consumers. This is a powerful pattern in horizontally scaling, decoupled applications. If you want to solve your scalability issues, sooner or later you will use TPL data flow or some similar library.

Does my business case need TPL dataflow? I'm just trying to learn about TPL dataflow and I'm having trouble understanding how splitting up the above steps in to "blocks" is going to solve my problems.

This will allow you to do an OOP approach to solving your data processing flows. The easiest solution to your problem is to have an object perform steps 1-6 in a background thread. You can look at that object as a wrapper over a thread that only knows to do 1-6.

If you want to process with multiple threads, basically achieving horizontal scaling inside your app, you can simply spawn more objects, each doing the 1-6 steps in parallel, each on a different message.

Sure, you can use plain old threads, Task.Run etc to achieve the same thing, but it will be the same as reinventing the TPL library, probably in a less efficient manner. And for more complicated scenarios, it will be very difficult to model the flow without OOP.

I have used data flow with simple stuff with 1 producer and 1 consumer to extremely complicated flows with multiple types of consumers based on message kind with retry and auto scaling number of consumers, the TPL library was extremely useful in simplifying the implementation. It is a hidden gem in the .net world.

5
  • Will TPL dataflow also help in the message ordering scenario? Im already using consitent hash exchange to order my messages per machine. What I'm told is that using async-await on each of these 6 steps would cause message to be processed in a different order, is this true? And would using TPL dataflow ensure ordered and highly performant, ordered processing of these messages? Commented Mar 1 at 19:14
  • Yes. Take a look at the BufferBlock class. It supports exactly what you need plus more. I suggest you start with a simple 1 producer 1 consumer scenario and work your way from there.
    – Ccm
    Commented Mar 1 at 20:28
  • Also the problem with messages being processed at unexpected times is probably caused by improper use of async await.
    – Ccm
    Commented Mar 1 at 20:30
  • can you clarify, does dataflow scale over multiple machines? It looks like its just parallel processing in a single process? I wouldn't call that "horizontal scaling"?
    – Ewan
    Commented Mar 2 at 0:06
  • @Ewan TPL is a general purpose library, you can definitely use it to achieve orchestration over several microservices on different machines. My example is trying to simplify the issue as much as possible.
    – Ccm
    Commented Mar 2 at 0:29
3

I don't think so.

It seems to me that "Dataflow (TPL)" is for running tasks in parallel with various ordering options on a single machine.

Your problem, as I understand your explanation, is that you have sets of messages which have to be run in order across multiple machines.

I'm not sure that consistent hashing solves this either.

You could use Dataflow within the processing of a single message, but I don't think you really need its advanced functions for this, you can use async tasks and await Task.WhenAll() if you want to run your microservice calls in parallel.

To get the ordering of your messages from Rabbit, you need to introduce a routing program in between the incoming messages and your processors.

This router pulls incoming messages and divides them up into ordered sets in memory. I guess a set in this case is a users session.

It also listens for ready messages from the consumer machines on a separate queue.

Your Consumer applications send a "ready for a set" message when they have finished the previous set. This message contains the queue id that they are listening to. (or you can have the router send an id)

When it gets a ready message the router puts an ordered set of messages on the consumers queue.

If the set is a continuous stream, it remembers which queue relates to that set and forwards those messages to that queue.

Because there is only one consumer per queue and the message are put on in order the order of processing is guaranteed*

The router holds the messages in memory for some short period of time before sending them on so that if out of order messages are received it has a chance to reorder them. Or, if you have some definite way of recognising gaps you can capture that and note the error.

You scale the consumers by adding more consumers, since the router knows the number of consumers and the rate of processing you can use it to spin up more boxes as required.

You scale the routers by sharding the incoming messages

*guaranteed barring various crashes, you will have to add handling for.

31
  • Sorry if i haven't explained my problem clearly. The business need is that my consumer service has to consume huge loads of messages and each of these messages has different identifers. Messages belonging to a particular identifier have to processed in order. The identifier in my case is a machine id. Consistent hashing helps with this out of order processing problem by routing messages belong to a particular identifier always to a consumer(we can achieve this by using the machine id as a routing key). Commented Mar 4 at 14:12
  • The confusion i have now is that, whether async-await used extensively results in a service processing messages quickly and becoming out of order again. I'm told that TPL dataflow solves this problem and I'm not clear on how it does or whether i have a problem in the first place. Commented Mar 4 at 14:14
  • yeah so the hashing will send the messages to the right consumer, but they may still be out of order. The router checks this and reorders.
    – Ewan
    Commented Mar 4 at 14:56
  • It doesn't really make sense that async calls when processing a message would change the order of processing. As long as you await the results before moving to the next.
    – Ewan
    Commented Mar 4 at 14:57
  • 1
    I checked our abstraction code as well, we don't increase the Concurrency setting anywhere. So it is processing sequentially one by one. I'm confused now as well. Thanks for taking your time to help me out. I'll discuss with my colleagues. Commented Mar 6 at 16:56

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.