The RabbitMQ .NET Client with more than 31 million downloads has a long and evolutionary history. In this webinar, Daniel Marbach dives into the inception of the first ideas on how to gradually move the client from its synchronous resource-hogging-nature to a more modern, asynchronous, and joyful client truly embracing concurrency.

In this webinar, you’ll learn about:

• The RabbitMQ .NET Client whereabouts
• The inception of an asynchronous code path into a purely synchronous code base
• The improvements of the message consumption loops through several iterations up to using channels
• The impact of async and concurrency on the client

It is very important to me to not give the wrong impression. I was heavily involved in the worker service changes from the client version 4.1 until the recent changes at the time of writing introducing the redundant asynchronous code path I'm going to talk about here. But of course I would never claim to be the single owner of these changes. The VMWare team namely

• Luke Bakken
• Michael Klishin

as well as the community such as

• Brandon Ording
• Szymon Kulec
• Stefán Jökull Sigurðarson
• Sandro Bollhalder

RabbitMQ for historic reasons implements AMQP 0-9-1 protocol. AMQP 1.0 is supported via a plugin on the broker only. AMQP 0-9-1 has become the RabbitMQ protocol because there is as far as I know nobody else that supports AMQP 0-9-1. These are the basic primitives we need to know about

• ConnectionFactory (IConnectionFactory) constructs connections
• Connections (IConnection) represent a long-lived AMQP 0-9-1 connection that owns connection recovery, mapping of frames to headers, commands and more
• Model (IModel) represents an AMQP 0-9-1 channel that is meant to be long-lived providing the operations to interact with the broker (protocol methods). For applications that use multiple threads/processes for processing, it is very common to open a new channel per thread/process and not share channels between them. As a rule of thumb, IModel instance usage by more than one thread simultaneously should be avoided. Application code should maintain a clear notion of thread ownership for IModel instances. This is a hard requirement for publishers: sharing a channel (an IModel instance) for concurrent publishing will lead to incorrect frame interleaving at the protocol level. Channel instances must not be shared by threads that publish on them. A single connection can have at max 100 channels (setting on the server) but managing many channels at the same time can decrease performance/throughput.
• Consumer (IBasicConsumer and other flavours) consumes protocol interactions from the broker (message received...)

The demo shows that we are handling one message at a time and if the sender increases the speed of sending messages they start piling up on the broker.

• The basic consumer makes it very hard to achieve concurrency and asynchronicity. Ideally we'd want to use task based async for high throughput by void returning method make that painful
• Either custom offloading to the worker thread pool can be used or the evil async void
• Unfortunately any sort of concurrency within the consumer invalidates the client assumptions being able to return internally managed buffers once the consumer returned which then requires copying the body before yielding (will be solved in 6.2.0)
• Concurrency has to be limited by using custom techniques like SemaphoreSlim
• Because the semaphore WaitAsync might complete synchronously if there are still slots available Yield needs to be used to offload the current thread back to the worker thread pool to avoid blocking the clients reader loop thread.
• Because interleaves on the same model are not allowed all model/channel operations need to be executed on a dedicated custom scheduler that prevents critical operations to interleave, see BasicAckSingle

• Even though IO-bound the RabbitMQ client is mostly purely sync
• Trying to bring asynchronicity even if only from a consumer perspective is a bit of a hassle and requires quite advanced trickery to get things working

How can we achieve better concurrency and asynchronicity support as a first class citizen in the client?

Touching the model/channel based methods was not yet an option at the time so we started from the receiver perspective.

The worker service or consumer work service is the core worker service associated with a connection that "pumps" work to the consumers.

To get a better grasp of what is going on we first inspected the sync loop.

The lock contention was so high that with more modern CPUs (at that time) when introducing concurrency only the throughput dropped in orders of magnitude.

The reason was the BatchingWorkPool and it's usage of locks as well as blocking collection.

Takeaways:

• Lock contention, bad!
• Locks bad in general because you cannot await inside lock statements (thread that enters lock needs to release it)
• If you don't work on your lock problems not even async is going to help you much

See RabbitMQ NServiceBus 6 updates

By using a ConcurrentDictionary to map the models to the worker pools and using a ConcurrentQueue with a dedicated AutoResetEvent to unleash the consumption loop we did not only better understand the loop but also tremendously improve the receive performance.

Takeaways:

• Removing locks can be liberating
• Orders of magnitude of throughput improvements just by doing that

Because of the large userbase of the client we wanted to introduce gradual changes. A good way to do so is to start introducing an alternative async I/O path that can be opted-in. So we created an IAsyncConnectionFactory and a DispatchConsumersAsync that then opts-in a new consumer work service with a dedicated AsyncDefaultBasicConsumer as well as AsyncEventingBasicConsumer that deals with asynchronicity. Still targeting .NET 4.5.x required us to make a few workarounds.

At this stage it was possible to declare receive methods that return a task. Achieving concurrency was still a bit of a nightmare because the client wouldn't call the received handler concurrently you still needed to do all concurrency handling yourself including preventing interleaves.

Takeaways:

• Freeing up the threadpool threads is crucial to achieve high IO saturation
• Task based APIs are the key to achieve that and there is no way you can avoid supporting them
• Sometimes but not always it is possible to focus on a single code path to plug in async and Tasks/ValueTask without being affected by the viral nature of async/await. Async all the way is the key. Unfortunately we are still not there yet because many of the IO-bound channel methods are still sync and require dedicated offloading.

Once we targeted .NET 4.6 and higher we were able to get rid of some of the workarounds we had in place.

At some point we tried to replace the TaskCompletionSource with a SemaphoreSlim (below benchmark explores ingestion/enqueue speed)

BenchmarkDotNet=v0.12.1, OS=Windows 10.0.19041.450 (2004/?/20H1)
AMD Ryzen Threadripper 1920X, 1 CPU, 24 logical and 12 physical cores
.NET Core SDK=3.1.302
  [Host]   : .NET Core 3.1.6 (CoreCLR 4.700.20.26901, CoreFX 4.700.20.31603), X64 RyuJIT
  ShortRun : .NET Core 3.1.6 (CoreCLR 4.700.20.26901, CoreFX 4.700.20.31603), X64 RyuJIT

Job=ShortRun  IterationCount=3  LaunchCount=1  
WarmupCount=3  

Takeway:

• Measure, measure, measure

With the new 6.x major version we finally had the opportunity to start using System.Threading.Channels to overall improve the allocations of the client and hopefully the overall throughput.

• A single unbounded channel is used to write work into it
• A dedicated async work is scheduled to the worker thread pool that consumes messages from the channel and invokes the consumers

Here is the comparison benchmark that compares Channels vs TaskCompletionSource on the ingestion path. As you can see System.Threading.Channel is slightly slower but the overhead is neglectible given that the consumption is less allocation heavy and in general much more optimized with channels. Furthermore having the possibility to offload some of the complexity to a solution provided by Microsoft is very useful because over time when the channels get improved all the users of the RabbitmQ client can automatically benefit from those improvements as well. From a maintance perspective we can also say that having less code to own is generally a good thing.

Takeway:

• Channels are nice concurrency datastructures for producer/consumer patterns
• Not always the fastest solution is the best. All factors need to be taken into account

BenchmarkDotNet=v0.12.1, OS=Windows 10.0.19041.450 (2004/?/20H1)
AMD Ryzen Threadripper 1920X, 1 CPU, 24 logical and 12 physical cores
.NET Core SDK=3.1.302
  [Host]   : .NET Core 3.1.6 (CoreCLR 4.700.20.26901, CoreFX 4.700.20.31603), X64 RyuJIT
  ShortRun : .NET Core 3.1.6 (CoreCLR 4.700.20.26901, CoreFX 4.700.20.31603), X64 RyuJIT

Job=ShortRun  IterationCount=3  LaunchCount=1  
WarmupCount=3  

With the 6.2 release there will be first class support for concurrency which implements two redundant code paths. One for the concurrency = 1 case and one for the concurrency > 1.

Takeway:

• Having first class concurrency support makes the client so much more usable and therefore user friendly
• Because the client now also uses internally System.Threading.Channels are interleaving problems are gone

• Community effort to introduce a full async enabled channel API to truly unblock threads
• Even more allocation reductions