[AMQP] Limit the items in the priority queue - horizontal scalability question

[SMFloris]

Hello everyone,

I've been testing out and migrating slowly but surely to roadrunner to handle our queue processing via the JOBS plugin.

I tested and autoscaling does not work for JOBS via amqp - not a huge deal for us, we have autoscaling of the deployment based on the rabbitmq queues via KEDA. But this raises some questions about the horizontal scalability of a roadrunner server running jobs.

Here is what happens, generally:

1. The amqp driver starts consuming the rabbitmq queue (so far so good)
2. In the consume callback, when it receives a new message from rabbitmq, it pushes it to the priority queue for it to get processed by the workers.
3. The workers concurently get the items from the PQ and process the messages.

The big problem that affects horizontal scalability from my current understanding is that the amqp driver keeps pushing the messages to the PQ infinitely or at least as big as max_queue_size (I am not sure this actually works, tried it but did not seem to do anything).

What ends up happening is that if we have two roadrunner instances that start listening to the same queue, and one consumer was first, it will fill it's PQ up and then when the other consumer comes online, it might not receive any new messages since the first consumer is still online and did not ack/nack the messages.

Is my understanding correct? Is there anything I can do to mitigate this?

EDIT: Changed maximum_queue_size -> max_queue_size for posterity

[rustatian]

Hey @SMFloris 👋🏻
Autoscaling is currently in beta, It would be nice if you could add your case here: #2086 with an example on how to reproduce it.

When you have 2 consumers, for the RMQ queue, when the first consumer read a message from the queue, until ACK or timeout (which can be configured via RMQ options), the second consumer won't see that message twice.

I don't know where you get this maximum_queue_size option (AI?), but the real option looks like this: https://github.com/roadrunner-server/roadrunner/blob/master/.rr.yaml#L1272, and with it you can limit how many messages would be pushed to that intermediate priority queue. Keep in mind, that the pipeline_size option affects all underlying pipelines.

[SMFloris]

Sorry, the option I was looking at is: max_queue_size (just wrote it from memory in the first post, I now have edited it).

What does max_queue_size do?

[rustatian]

max_queue_size should not be used in the case you describe, only pipeline_size. max_queue_size is the workers pool property, not a jobs pipeline property, and is used to limit the number of requests (of any type, like HTTP, jobs, grpc, etc) which is allowed to wait for the worker. Usually, allocate_timeout is enough to control how much time any request can wait for the worker.

[SMFloris]

Thanks for the response! Now that I have a little bit more understanding, could you please correct me if I am wrong in my assumptions:

1. Let's say I limit the pipeline_size to 50
2. Once the pipeline size reaches max, pq.Insert will block until the pq is empty
3. Because pq.Insert blocks, the consume itself would also block - hence no new messages will be prefetched.

Is this correct?

[rustatian]

Yep, you're correct 👍🏻

[SMFloris]

Thank you so much! I'm happy we'll be able to use RR in production.

[rustatian]

My pleasure. Glad to hear that ❤️