ApplicationMessageReceivedAsync delegate executing syncronously

[Lacutas82]

I have a delegate setup for ApplicationMessageReceivedAsync, and this is correctly processing the messages. I'm now in the process of stress testing, and handling multiple messages at once and it appears that the delegate is executing synchronously, processing each message in the order it's received.

I have a database operation within my delegate, and have delayed the acknowledgement until the ApplicationMessage has been successfully saved to the database. If this takes a few seconds for whatever reason, I don't want the other messages to be delayed.

Is it correct that the ApplicationMessageReceived delegate is running synchronously, and if so, is there an option to have it run asynchronously. If not, am I right in understanding, I have to handle the multiple threads within my delegate?

[Lacutas82]

After some reading it appears event handlers execute on the initiating thread as standard in C#. I've handled this by spinning off the actual execution to a Task. One thing I did find was I had to set the AutoAcknowledge flag to false before passing in the parameter to the Task.

        public async Task MqttApplicationMessageReceived(MqttApplicationMessageReceivedEventArgs args)
        {
            args.AutoAcknowledge = false;
            Task.Run(() => ProcessApplicationMessage(args));
        }

        private void ProcessApplicationMessage(MqttApplicationMessageReceivedEventArgs args)
        {
            var threadSafeArgs = args;
            // Do some long running db Task
            threadSafeArgs.AcknowledgeAsync(_cancellationToken);
        }

If I didn't do this, by the time my ProcessApplicationMessage method had got around to acknowledge the message, it had already been auto acknowledged.

[gfoidl]

My response is just for this piece of code shown.

I wouldn't use Task.Run here as

• there's a closure for args, so allocations (also for the delegate) that can be avoided
• it puts stress to the thread-pool* -- it can handle it of course, and depends on the actual load, but that can be avoided too

I'd use Channels where MqttApplicationMessageReceived writes to the channel and ProcessApplicationMessage reads from the channel. Depending on the load, several consumers (readers) can be applied (even dynamically, but there's no out-of-the-box scaling solution).
It may sound a bit more complex on first glance, as Task.Run is really easy to type, but Channels aren't that complicated.

Task.Run uses the thread-pool as "implict producer/consumer", with Channels you have that explicit, but with less allocations and more controle (e.g. how many consumers, behavior if the channel is full).
As you said it's for DB-operations with Channels kind of retry is also simples, just write to the Channel again (if the consumer has a reference to the writer) and you're done. With Task.Run it's also possible, but with more cerermony.

Of course, one should benchmark the approaches to see numbers.
I'm only sure that Task.Run has more allocations than the explicit producer/consumer way.

---

* say you have 1.000 messages arriving, then you put 1.000 work-items with all the internal state needed to the thread-pool. That may or may not spin up additional threads to work on these items. When they run they perform a DB-call, which is IO, so they do actually not really a lot, just (a)waiting the async-IO then they're done.
In contrast assume for Channels you have 1 producer and $n$-readers ($n$ can be 1 or Environment.ProcessorCount or any numbers that suits your workload), then those n readers (a)wait the items from the channel, kick off the async-IO to the DB and can process the next item from the Channel. This uses way less ressources than queuing all these work-items to the thread-pool.

[Lacutas82]

Thanks Gunther, I'll take a look!

[hem-bhagat-tark]

@gfoidl I am also facing a similar issue. I am having some complex database operations in my ApplicationMessageReceivedAsync handler. I am facing a delay when the rate at which I receive messages is more than the rate at which I can process the message. In my scenario would you suggest use of Channels or is there a better way?

[gfoidl]

W/o knowing the details, I'd go with Channels. The reason for this is to off-load the work from ApplicationMessageReceivedAsync handler to a separate piece that processes at its own pace and makes the handler complete as soon as possible (once the message is in the Channel).
Then you can also play around with the setup of the Channels, e.g. use two or more consumers (Readers) to parallize the slow DB-operation.

If this doesn't help, you could also consider scaling out the handler, by running n instances. But that's probably more work and administration to do.

[hem-bhagat-tark]

Thanks for the response @gfoidl. I will take a look at Channels.