Inefficient memory pooling in .NET 6 and RabbitMQ.Client 6.x.x

[aleshinid]

Hello,

We observing high memory consumption for the code utilizing BasicPublish method, when that method has a higher rate of invocation.
The problem and suggestion are described here - https://groups.google.com/g/rabbitmq-users/c/2YuLVYzE7Vg?pli=1

Example:

using RabbitMQ.Client;
using System.Buffers;

namespace RabbitMqTest;

internal class Program
{
	public static IConnection CreateConnection()
	{
		var factory = new ConnectionFactory()
		{
			HostName = "localhost",
			Port = 5672,
			DispatchConsumersAsync = true,
		};
		return factory.CreateConnection();
	}

	static void Main(string[] args)
	{
		byte[] array = ArrayPool<byte>.Shared.Rent(1024 * 1024 * 7);

		IConnection connection = CreateConnection();
		IModel channel = connection.CreateModel();

		IBasicProperties properties = channel.CreateBasicProperties();
		properties.Expiration = "5";

		channel.QueueDeclare(queue: "test",
							 durable: false,
							 exclusive: false,
							 autoDelete: true,
							 arguments: null);

		try
		{
			for (int i = 0; i < 20; i++)
			{
				Thread.Sleep(5); 
				Console.WriteLine(array.Length);
				channel.BasicPublish(exchange: "",
							 routingKey: "test",
							 basicProperties: properties,
							 body: array);
			}
		}
		finally
		{
			channel.Dispose();
			connection.Dispose();
			ArrayPool<byte>.Shared.Return(array);
		}
	}
}

The higher the delay (line with Thread.Sleep) when calling the BasicPublish method, the less memory is allocated.

for example 5 milliseconds:

for example 500 milliseconds:

/cc @sakno.

[michaelklishin]

You are welcome to submit a PR that improves the pooling. This community does not guess (so profiler data is a good start) but we don't use issues for anything but specific, well understood, actionable items the community agrees to.

Any publishing in a tight loop will outpace the actual ability of the client to send data and release memory, so "the more frequently I publish the higher the memory usage" is expected, with pretty much any client.

[sakno]

Option 4: I found a way to utilize existing API without extending IModel interface:

public class ConnectionFactory
{
   public Predicate<MemoryManager<byte>> MemoryToDisposeFilter { get; set; }
}

Now we can reuse existing method void BasicPublish(string exchange, string routingKey, bool mandatory, IBasicProperties basicProperties, ReadOnlyMemory<byte> body). It can extract the memory managed from ReadOnlyMemory<byte> and check the manager with the predicate mentioned previously. If predicate returns true then WriteLoop can call Dispose on the manager without making the copy of the buffer.

[michaelklishin]

That sounds like a change we'd have to benchmark to really understand if it has any unexpected side effects. Which does not rule it out, of course,

[danielmarbach]

I'm in kind of weekend mode and haven't really thought this through so it might be nonsense 😉 could we instead do a MemoryMarshal.TryGetMemoryManager?

[sakno]

@danielmarbach , yes, option 4 proposes this approach instead of introducing a new overload. However, we should not dispose every memory manager due to backward compatibility.

[danielmarbach]

Yeah but option 4 proposes a predicate based approach which can also lead to all sorts of closure issues. Why not make it a global option that is by default false and otherwise we dispose all? In which scenario would you see your predicate not returning true for your cases?

I mean I get how the delegate based approach would be more elegant because the lib would not dispose memory it doesnt own and therefore that is definitely the cleaner design.

Another thought. Couldn't you use publisher confirms to get rid of the memory? If course using those confirms has other implications in terms of throughput. And the confirms are not raised on the same thread so that is also not entirely ideal but a delegate based approach has similar shortcomings

[michaelklishin]

By the way, once you have a clear understanding of what can be done, you are welcome to file issues for those specific changes.
Discussions like this are meant to "shape up" both the problem and a specific solution. Or you can go straight to pull requests and explain what's being addressed there 👍

[sakno]

@michaelklishin , should I fork main branch? Or 6.x?

[michaelklishin]

Please fork main, if we can't backport, we will ask for your help :)

[bollhals]

Thanks for bringing up this topic and yes, your analysis is absolutely right. Now to the but =)

1. Is there a real world use case behind this that is impacted by the current design? (It is very easy to showcase the problem in the example you gave, but I have yet to see a real world usecase where a similar situation happened (e.g. sending multiple MB of data every few ms over a significant amount of time)

2. I think your option 1 is doable and would help use different pooling mechanisms if desired. Question though is whether or not it is seen as a case the library wants to improve on. How many users of this library will benefit from this change vs how many are impacted negatively? (e.g. by additonal overhead performancewise)

If I'm not mistaken, the arrays in the shared arraypool are now collected if there is high memory pressure as well as they've changed the max size that is pooled, so lots of improvements done in newer .NET to lessen the impact here as well.

3. Option 2 and others: I'm not sure how you can easily solve it by the proposed API. There are for me two reason why we use the arraypool copy:

• The publish isn't fully synchronous, meaning after the publish is done, it's not yet sent. Meaning if a user would overwrite the passed memory immediately after the publish, we would publish the new value instead of the old if it wasn't copied. (Or we'd have to make it synchroneous, which has other impacts)
• There is protocol data send before and after the body payload. These have to be passed as well somehow to the one who write to the connection. I don't see how this is easily possible without copying the data.
• Extension of the above, the body payload is split into multiple packages, if it's above the max body size. What would you do in this case with the proposed API?
• Last but not least, the copying enables that the body is put together with the method & header framing, this makes sure that there is no threading issue if multiple threads publish at the same time. (They were serialized individually in < 6. something. which made it necessary to use lock() around the basic publish operation)

So without solving the above mentioned issues, I'm not sure there's much to go on. Or maybe I missed something, so please correct me if this is the case.

[danielmarbach]

It would only work if you make the producer aware to overrent and copying the framing into the producer content which is definitely the wrong approach from a design perspective for this libs abstraction level.

[danielmarbach]

It is true though there is a gap between the producer and the inner consumer side and because the lib is not async it is no possible to properly throttle the producers yet. But I think we should rather focus on improving that story instead of trying to optimize the scenario we are talking about here.

It would also be necessary to see things in the light of a more realistic benchmark with various different payloads and network latency involved to properly judge things

[stebet]

Pipelines solve this problem since they introduce the concept of back-pressure, but that requires us to make the publish methods truly async. This is done in one of my PRs which I need to update b.t.w. That'll make sure that if the pipeline is full, the writers will not be able to write synchronously to the output buffers until they go below a certain threshold.

[sakno]

@stebet , awaitable version of BasicPublish is also a good option that could solve the problem.

[stebet]

@stebet , awaitable version of BasicPublish is also a good option that could solve the problem.

That's exactly what the Pipelines make easy, and is the purpose of the Pipelines implementation :)

[lukebakken]

Hello everyone who has commented here -

If you have time to test version 7 alphas, it would be really appreciated

https://www.nuget.org/packages/RabbitMQ.Client/7.0.0-alpha.6

I'm hoping to get the final release done this month, but only if there is enough community testing (with reports back!)