Doug Barth

Approximating Priority With RabbitMQ

2011-07-01T00:00:00+00:00

As mentioned in a previous blog post, at Signal we use RabbitMQ for all our queuing needs. Though the AMQP protocol supports the concept of priority, RabbitMQ does not yet implement that feature. However, with a little bit of creative client side code, we’ve been able to approximate priority in our application in a way that will allow us to seamlessly remove that workaround once RabbitMQ implements priority internally.

AMQP’s priority support

The AMQP protocol supports up to 10 levels of priority, starting at zero. 0 has the lowest priority and 9 has the highest. The priority of a message is set by the publisher using the priority header. The consumer will then have messages pushed to it in priority order.

Brokers that say they fully conform to the spec must implement at least 2 levels of priority. In that case, the 10 levels of priority are broken up into 2 ranges: 0-4 and 5-9. Each range is then treated as a single priority level.

Dealing with it on the client

To workaround the lack of priority support in RabbitMQ, we need to split one logical queue into several physical queues based on priority. The consumer pulls messages from all queues, but gives preference to the higher priority queues if work is available.

At Signal, we made this nuance transparent to our application by adding a layer of indirection. Our application code publishes using the logical queue name and priority through a common function. That function then determines the physical queue to publish to based on the priority (eg. 0-4 goes to the slow queue, 5-9 goes to the fast queue). Consumers subscribe to both queues, but always work the higher priority queues first.

Our first implementation had two physical queues (fast & slow), and consumers popped messages off those queues in priority order. First, they would try to pop a message off the fast queue. If a message was returned, they processed it and then tried popping from the fast queue again. If no message was on the fast queue, tried popping from the slow queue. If a message was on the slow queue, they processed it and then tried popping from the fast queue immediately. If no message was on the slow queue, we would sleep for a short while to ensure that we don’t overwhelm the broker with polling.

This approach worked well enough, but when queues had a large backlog, we found RabbitMQ would eat up a bunch of CPU processing the polling logic. Clearly, we needed to switch to a subscription model, so RabbitMQ will push messages to us with less overhead.

Dealing with the subscription flood

Switching to an AMQP subscription model presents a new set of issues: namely, subscriptions in AMQP push messages to consumers as quickly the network interface is serviced. Since we process work on a different thread than EventMachine’s reactor thread (to leave it free to publish new messages), messages will be pulled off the network interface and into memory faster than we can process them. Without an approach to managing the flood of messages coming into the consumer, we would be unable to guarantee a priority order processing of messages nor keep our process from dying due to memory usage.

To manage the influx of messages for standard AMQP subscriptions, we lean on prefetch counts and explicit client acks. The prefetch count, set on a channel, determines how many unacked messages a channel may be sent before the broker will wait for acknowledgments.

To service multiple subscriptions in priority order, we subscribe to each queue and put incoming messages on an in memory priority queue. Each subscription is on its own channel, with an appropriate prefetch count: higher prefetch counts for higher priority queues. The sizing of the prefetch counts are crucial. If they are too low, low priority messages that are in the worker’s memory will be processed ahead of higher priority messages that need to be sent from the server. In our application, we found that a good rule of thumb is to double the prefetch count for each higher level of priority.

As messages are received from RabbitMQ, we put them on an in memory priority queue. A background worker thread pulls messages from that queue (blocking if empty) and processes them. After a message is processed, an ack is sent to RabbitMQ to signal that a new message of that same priority can be sent.

Add a third queue

With 2 levels of priority approximation, we were able to ensure high priority outbound SMS messages (eg. responding to a STOP text message) were processed before lower priority messages (eg. a scheduled blast to a subscription list). This approach worked great until we started having clients who’s lists were large enough to saturate our outbound connections for long periods of time. Our solution was to add a third priority queue and send really large blasts with the lowest priority and smaller blasts with the middle priority level. With all the plumbing in place, adding this extra queue was little more than adding the third queue and adjusting the priority ranges.

After living with this priority approximation for over 2 years, we are very pleased. Other than the few incremental changes, we’ve had no issues supporting this solution in production. If you’re evaluating RabbitMQ and lack of proper priority has you down, I’d suggest implementing this solution.

Rabbit on a Leash — Rate Limited AMQP subscriptions

2011-06-10T00:00:00+00:00

RabbitMQ is fast: really fast. Consuming messages from a queue is extremely efficient. Consumers declare the queues they are subscribing to and the broker pushes messages to the consumer for processing as soon as they are ready. The AMQP protocol, which RabbitMQ implements, supports the concept of limiting how many outstanding messages a consumer can be tasked with processing via the prefetch_count and no_ack headers, but it does not have a way to control the rate of delivery of messages to consumers.

At Signal, we use RabbitMQ for all our queueing infrastructure needs. Outgoing SMS messages (MTs) are queued for a pool of workers to send to our aggregator. Our connection to our SMS aggregator requires us to limit the rate that we send messages to their system. It would seem that RabbitMQ is a poor fit for that use case, but we’ve been able to fulfill it using a bit of client side code and the existing AMQP protocol.

Using AMQP’s prefetch_count, client acks and a blocking token bucket, it’s possible to implement rate controlled processing of queued messages.

Token Buckets

A token bucket is an algorithm that is used to control the rate of data that flows through a system¹. Token buckets can be configured to allow traffic to burst to full speed, but they ensure that the average traffic processed is held at a configurable rate.

The concept of a token bucket is rather simple. Imagine the bucket in your freezer’s ice maker. Cubes of ice are added to the bucket at a certain rate (say 1 a second). The size of the bucket controls how many ice cubes (tokens) we can have waiting in the bucket before we will stop making more.

In order for traffic to be processed, we need to take a token (or more) from that bucket. If the bucket is empty, that work cannot be processed. The rate that tokens are added to the bucket controls the average speed that work is processed. If we started with an empty bucket we could process work at a rate equal to the rate that we added cubes of ice. The size of the bucket controls how much work we can burst. If the bucket held 10 tokens, we could process 10 units of work at full speed before we would be rate limited.

Putting it all together

With a correctly working token bucket, implementing fixed rate processing is fairly straightforward. First, when subscribing to a queue, we set an explicit prefetch_count on the channel and we set no_ack to false when subscribing. The prefetch_count limits how many unacked messages RabbitMQ will deliver and no_ack allows us to acknowledge the message once we’ve finished processing it. In our application, we size the prefetch_count so there are a few seconds worth of messages waiting in the worker’s memory to be sent.

We use the token bucket to control our rate of processing these messages from RabbitMQ. We need to take a token from the bucket before processing a message. If the bucket is empty, we block until a new token is added.

EM.run do
  channel = AQMP::Channel.new

  # Allow 10 unacked messages to be delivered to this worker.
  channel.prefetch(10)

  # Configure this worker to send at 1 msg/s on average with occassional bursts
  # up to 5 messages.
  token_bucket = TokenBucket.new(1, 5)

  # Refresh the token bucket every second. The bucket is also refreshed
  # when the take method is called.
  EM.add_periodic_timer(1) { token_bucket.refresh }
 
  # We subscribe with explicit acknowledgments so we can signal to RabbitMQ
  # that more work should be delivered. Without this setting, RabbitMQ would
  # send work over to us as fast as possible.
  channel.queue('send_mt').subscribe(:ack => true) do |header, message|
    # Defer the processing to a background thread since taking a token from
    # the bucket could potentially be a blocking operation and we don't want to
    # block the reactor.
    EM.defer(
      lambda {
        # Takes 1 token from the bucket. If the bucket is empty, this
        # method will block.
        token_bucket.take(1)

        process_message(message)

        # Acknowledge this message, allowing RabbitMQ to send more work.
        header.ack
      }
    )
  end
end

¹ Token Buckets