Prefetch Peek

[mmcnabb-vms]

I'd like some feedback on a feature I implemented locally. I haven't just gone ahead with a PR yet as it potentially incurs a performance penalty and overwrites other metadata that I didn't need for my use case so would be interested in some discussion or other suggestions.

The feature is "Prefetch Peek", i.e. when a message is delivered via an AsyncConsumer, the metadata will also include a flag to report whether or not further messages are waiting or not.

My consumer application processes and forwards messages upstream in batches of up to 100, however if there is not a full batch of data available I don't want it to potentially sit there for hours with unprocessed messages waiting for a batch to fill. It makes the consumer code much simpler to just have this information available when handling a message; otherwise I have to have a second "idle sweep" thread that looks for partial batches and flushes them, and guard against race conditions and so on.

My code change is:

--- projects/RabbitMQ.Client/ConsumerDispatching/AsyncConsumerDispatcher.cs ---
index 6ec89ce0..854f216d 100644
@@ -34,11 +34,14 @@ protected override async Task ProcessChannelAsync()
                                 switch (work.WorkType)
                                 {
                                     case WorkType.Deliver:
+                                        // MM: let caller know if more stuff is prefetched
+                                        String customTag = _reader.TryPeek(out _) ? "MM1" : "MM0";
+
                                         using (Activity? activity = RabbitMQActivitySource.Deliver(work.RoutingKey!, work.Exchange!,
                                             work.DeliveryTag, work.BasicProperties!, work.Body.Size))
                                         {
                                             await work.Consumer.HandleBasicDeliverAsync(
-                                                work.ConsumerTag!, work.DeliveryTag, work.Redelivered,
+                                                customTag, work.DeliveryTag, work.Redelivered,

and in my application's handler I will dispatch a partial batch if and only if "MM1" was not received in the ConsumerTag.

It seems to work well in practice. Of course there are often "false negatives", i.e. it can report no messages waiting when a message was fetched but is microseconds away from being ready; but that's not an issue for my use case. It's no big deal if several small batches are sent, but it is a big deal if messages grow stale.

I am ACKing messages using MultiAck after upstream processing of the batch returns, so that if my application crashes or is killed, then messages will be redelivered later when it is restarted.

[michaelklishin]

@mmcnabb-vms no other RabbitMQ client uses magic values such as "MM1" or "MM0" to communicate something to the consumer implementation.

This should be a new consumer implementation that handles batching however you see fit. In the Java client of the past, and possibly this client, there was a QueueingBasicConsumer that used an internal queue data structure to accumulate a stream of deliveries.

You can implement something similar tailored for your needs, in your own codebase.

Note that QueueingBasicConsumer was removed after years of experience with massive heap size explosions that were not due to a bug but users always blamed this library for. Buffering is hard to get right in a generic library that knows nothing about the application.

[mmcnabb-vms]

Well, of course (regarding the magic values), it's just proof of concept. My thought is that if it's agreed to be a viable feature then we could find some better place in the metadata to put this information.

Agree that a batching consumer could be a good idea although my method does give maximum flexibility to the application to decide on a batching strategy ; as you suggest later it could be hard for a batching consumer to please everybody. (For example I need to be able to ack/nak/reject each item individually)

[michaelklishin]

if @lukebakken @Gsantomaggio @danielmarbach @bollhals @stebet agree that we need such a delivery dispatcher feature, I won't object but if you ask just me, I'd say that a system-specific consumer implementation would be sufficient.

[Gsantomaggio]

I agree with @michaelklishin; It is possible to reach the same result with a consumer implementation.
@mmcnabb-vms Thank you for the proposal, but I don't think we should add it.
Let's hear the others btw.

[danielmarbach]

That's a classical problem of how much convenience should be in the driver VS what shouldn't. I'm more in the camp of keep the "core" simple and move the rest into "extensions"

[lukebakken]

have to have a second "idle sweep" thread that looks for partial batches and flushes them

Yep, this is exactly what you should do.

As soon as we add TryPeek, we own it, and it isn't guaranteed to work as expected 100% of the time.