Rabbitmq-server: In manual ack mode, tag collection assumes FIFO order, can be O(n^2)

[benmmurphy]

In the degenerate case where a consumer acks messages individually in the reverse order that they receive them the rabbitmq server will traverse the full length of the ack queue in the subtract_acks method each time.

https://github.com/rabbitmq/rabbitmq-server/blob/master/src/rabbit_queue_consumers.erl#L277

    subtract_acks([], [], CTagCounts, AckQ) ->
        {CTagCounts, AckQ};
    subtract_acks([], Prefix, CTagCounts, AckQ) ->
        {CTagCounts, queue:join(queue:from_list(lists:reverse(Prefix)), AckQ)};
    subtract_acks([T | TL] = AckTags, Prefix, CTagCounts, AckQ) ->
        case queue:out(AckQ) of
            {{value, {T, CTag}}, QTail} ->
                subtract_acks(TL, Prefix,
                              maps:update_with(CTag, fun (Old) -> Old + 1 end, 1, CTagCounts), QTail);
            {{value, V}, QTail} ->
                subtract_acks(AckTags, [V | Prefix], CTagCounts, QTail);
            {empty, _} ->
                subtract_acks([], Prefix, CTagCounts, AckQ)
        end.

queue:out is called until the ack is found. if the ack is at the tail of the queue then the whole of the ack queue is processed.

The same issue also seems to be present in rabbit_channel#collect_acks:

https://github.com/rabbitmq/rabbitmq-server/blob/master/src/rabbit_channel.erl#L1732

collect_acks(Q, 0, true) ->
    {lists:reverse(queue:to_list(Q)), queue:new()};
collect_acks(Q, DeliveryTag, Multiple) ->
    collect_acks([], [], Q, DeliveryTag, Multiple).

collect_acks(ToAcc, PrefixAcc, Q, DeliveryTag, Multiple) ->
    case queue:out(Q) of
        {{value, UnackedMsg = {CurrentDeliveryTag, _ConsumerTag, _Msg}},
         QTail} ->
            if CurrentDeliveryTag == DeliveryTag ->
                    {[UnackedMsg | ToAcc],
                     case PrefixAcc of
                         [] -> QTail;
                         _  -> queue:join(
                                 queue:from_list(lists:reverse(PrefixAcc)),
                                 QTail)
                     end};
               Multiple ->
                    collect_acks([UnackedMsg | ToAcc], PrefixAcc,
                                 QTail, DeliveryTag, Multiple);
               true ->
                    collect_acks(ToAcc, [UnackedMsg | PrefixAcc],
                                 QTail, DeliveryTag, Multiple)
            end;
        {empty, _} ->
            precondition_failed("unknown delivery tag ~w", [DeliveryTag])
    end.

This can lead to big delays to anyone trying to interact with the queue while the consumer is acknowledging messages. I have included a bunny ruby script that reproduces the problem. You can it via: ./bunny_performance lifo_strategy

On my machine it shows the queue blocked for 15 seconds while it is processing the acknowledgments.

acknowledging_messages
finished_acknowledging_messages
publish_time: 15.240954

The performance is much better if the messages are acknowledged in fifo order or if they are acknowledged all at once with multiple=true.

Looking at the code for subtract_acks I don't think queue() is the appropriate data structure. It might be faster when there are a small number of outstanding acks or if acks are heavily biased to be processed in order (which is kind of a reasonable assumption) but it would seem dict or map would handle the degenerate case better.

However, in the case of the collect_acks it needs to handle multiple: true. Dict or map wouldn't let you handle this efficiently. The only reasonable erlang option from the standard library would be gb_sets/gb_trees. In the erlang client libraries gb_sets is used to handle multiple: true from the server. (https://github.com/rabbitmq/rabbitmq-erlang-client/blob/95af0b85337cdb1420daacdb3fe8876ccfff72e5/src/amqp_channel.erl#L99)

#!/usr/bin/env ruby
require 'bunny'

QUEUE_NAME = "test_lots_of_acks"
QUEUE_ARGS = {durable: true}

ACK_SIZE = 8000

PUBLISH_INTERVAL = 0.1

def seed_queue
  conn = Bunny.new
  conn.start
  begin

    ch = conn.create_channel
    ch.confirm_select

    q = ch.queue(QUEUE_NAME, QUEUE_ARGS)
    q.purge

    ACK_SIZE.times do |i|
      q.publish("test_message #{i}", persistent: true)
    end
    ch.wait_for_confirms
  ensure
    conn.stop
  end
end

def fifo_strategy(ch, tags)
  tags.each do |tag|
    ch.acknowledge(tag, false)
  end
end

def lifo_strategy(ch, tags)
  fifo_strategy(ch, tags.reverse)
end

def multiple_strategy(ch, tags)
  last = tags[-1]
  ch.acknowledge(last, true)
end

def run_subscriber(strategy)
  conn = Bunny.new
  conn.start
  ch = conn.create_channel
  ch.prefetch(ACK_SIZE)
  q = ch.queue(QUEUE_NAME, QUEUE_ARGS)

  delivery_tags = []

  q.subscribe(consumer_tag: "test", manual_ack: true) do |delivery_info, properties, payload|
    if delivery_tags.nil?
      next
    end

    delivery_tags << delivery_info.delivery_tag
    if delivery_tags.length % 100 == 0
      puts "received_message: #{delivery_tags.length}"
    end

    if delivery_tags.length == ACK_SIZE
      tags = delivery_tags
      delivery_tags = nil
      puts "acknowledging_messages"
      strategy.call(ch, tags)
      puts "finished_acknowledging_messages"
    end
  end

end

def run_publisher

  conn = Bunny.new(continuation_timeout: 30_000)
  conn.start
  begin

    ch = conn.create_channel
    ch.confirm_select
    q = ch.queue(QUEUE_NAME, QUEUE_ARGS)

    loop do
      start = Time.now
      q.publish("timing message", durable: true)
      ch.wait_for_confirms
      finish = Time.now
      elapsed = finish - start
      puts "publish_time: #{elapsed}"
      sleep_time = [0, PUBLISH_INTERVAL - elapsed].max
      Kernel.sleep(sleep_time)
    end
  ensure
    conn.stop
  end

end

if ARGV.length != 1
  puts "usage: #{$PROGRAM_NAME} lifo_strategy|fifo_strategy|multiple_strategy"
  exit(1)
end

seed_queue
run_subscriber(method(ARGV[0].to_sym))
run_publisher

[michaelklishin]

Thanks for the details.

I corrected the persistent option of Bunny::Channel#publish.

[michaelklishin]

The FIFO order assumption is fairly natural given the FIFO-based nature of RabbitMQ queues but I see your point. Using a different data structure should be a possibility. If you'd like to dig deeper we'd appreciate your suggestions.

By the way, this is something we'd be able to fit into 3.8.0 but likely not 3.7.x.

[michaelklishin]

A very quick investigation suggests that there's a natural order dependency: clients can acknowledge N messages at once in increasing delivery tag order. Therefore a map or dict isn't necessarily going to be a good fit.

[benmmurphy]

I updated the original issue but the erlang client libraries use gb_sets to handle a similar problem where the server can acknowledge multiple messages.

[michaelklishin]

@benmmurphy thanks, that's a useful data point I should have thought of. We do want to improve this, I just wanted to explain that the tendency towards FIFO has a reason behind it (besides also being a common workload).

[benmmurphy]

makes sense. kind of scary to change this because something that works better for the degenerate case might give worse performance for how most people are using it.

[acogoluegnes]

Looks like there's no one-size-fits-all solution, so maybe a new queue argument can provide a hint so the queue uses the appropriate strategy. It means we need to know the load beforehand, but it sounds reasonable for a "degenerate" case. WDYT @benmmurphy @michaelklishin?

[michaelklishin]

I'd like to investigate options that would not require new queue arguments. We already have a lot of them and they don't always work well when combined. Since @benmmurphy understands the problem, wrokarounds and has an example that uses different acknowledgement strategies, we can take some time to think this through.

[acogoluegnes]

I created a few tests to experiment and avoid regressions as well. I need to cover other scenarios like re-enqueueing to see if there are assumptions we can rely on to optimize the current algorithm or pick another one.

[acogoluegnes]

I changed the internal FIFO queue to a list and used sorting + binary search for single ack. The worse-case scenario is a bit faster, but the common scenario is much slower than before now. This is WIP though, I need to profile more thoroughly.

[benmmurphy]

i had a look at collect_acks and the raw performance of the function implemented in gb_trees vs queue is about 5x as slow in the fifo use case (under R20 with take_any its about 3x) and about 3x as slow in the multiple use case. would probably have to see how long this function takes relative to the whole of processing an ack to see if it would be a problem or not.

[https://gist.github.com/benmmurphy/66e2498ba27ca9a46fd29e89f43402b6] just beware that collect_acks gb_trees implementation is slightly broken as well.

[benmmurphy]

i'm surprised using sorting + binary_search for a single ack would give better performance for the worst case scenario. so before for a single ack it had to traverse the whole list in order to remove the ack. so it had to do 'N' work. now, it has to sort the whole list (N log N work) and also remove an item from the list using list:delete (worst case N work). in theory this should run slower not faster.

[michaelklishin]

lists:sort/1 is a BIF and queue:out/1 is not. The Big O characteristics matter only when the input is sufficiently large. 8000 items is not a small list but it's nowhere near the "approaches infinity" case.

[benmmurphy]

i think you might be right in this case the lifo performance of the original implementation of subtract_acks and the sort implementation of subtract_acks when benchmarked is very close. in my benchmark the lists:sort() version is slightly slower but i don't think it is distinguishable from noise on my machine and it might run faster on different versions of erlang/etc.

1> c(subtract_acks_bench_original), subtract_acks_bench_original:bench_lifo().
{4998943,{[],[]}}
2> c(subtract_acks_bench_rmq), subtract_acks_bench_rmq:bench_lifo().
{5047869,[]}

the problem is the lifo performance is still going to be so bad that it will be unusable. i run the fifo benchmarks 1000x in order to distinguish between the different results but these lifo benchmarks don't terminate in a reasonable amount of time if i run them 1000x times. to make the lifo performance usable you need about 100x performance increase. like a 2x or 3x performance increase is not going to be good enough and the only way you can get that is to use map/dict/gb* structure.

i have some proof of concept code that uses map for subtract_acks and gb_tree for collect acks here: https://github.com/rabbitmq/rabbitmq-server/compare/master...benmmurphy:handle_degenerate_ack_ordering?expand=1

this was mostly to test if it would fully fix the lifo performance issue and if there wasn't any other hidden places that were slowing acks down. i can confirm it fixes the performance of the ruby bunny check. however, i've kind of approached fixing the code in the wrong way so there is probably lots of subtle breakages (or even completely broken implementation). writing these tests for the original implementation first was a much better way of approaching the problem. basically, i have little confidence in my code but it should give you an idea of the performance costs in the best fifo case and how much faster it can be in the worst lifo case.

[acogoluegnes]

@benmmurphy I applied only the map for substract_acks part of your patch and it does fix the Bunny check. I haven't looked into the acks collection part yet. I'll do more tests and benchmarks with the map solution to see if we can include it.

[benmmurphy]

that's interesting. i didn't test them separately and assumed both parts of the code would slow things down.

my new theory is the subtract_acks happens in the queue process so it blocks the queue which then becomes a problem for other people that are using the queue (ie: publishers) which is why it is showing up in the original bunny test. however, the collect_acks happens in the channel process so it blocks the channel and doesn't effect other consumers so doesn't show up in the original bunny test. i've modified the original bunny test so the consumer tries to publish after acking. without the collect_acks patch the delay is now visible in that part of the test.

without the collect_acks patch:

received_message: 100
..
received_message: 8000
acknowledging_messages
publish_time: 0.204051
finished_acknowledging_messages
trying to publish
..
consumer_publish_time: 5.214303
publish_time: 0.012243
..

with the collect_acks patch:

received_message: 100
..
received_message: 8000
acknowledging_messages
finished_acknowledging_messages
trying to publish
publish_time: 0.448687
consumer_publish_time: 0.073258

#!/usr/bin/env ruby
require 'bunny'

QUEUE_NAME = "test_lots_of_acks"
QUEUE_ARGS = {durable: true}

ACK_SIZE = 8000

PUBLISH_INTERVAL = 0.1

def seed_queue
  conn = Bunny.new
  conn.start
  begin

    ch = conn.create_channel
    ch.confirm_select

    q = ch.queue(QUEUE_NAME, QUEUE_ARGS)
    q.purge

    ACK_SIZE.times do |i|
      q.publish("test_message #{i}", persistent: true)
    end
    ch.wait_for_confirms
  ensure
    conn.stop
  end
end

def fifo_strategy(ch, tags)
  tags.each do |tag|
    ch.acknowledge(tag, false)
  end
end

def lifo_strategy(ch, tags)
  fifo_strategy(ch, tags.reverse)
end

def multiple_strategy(ch, tags)
  last = tags[-1]
  ch.acknowledge(last, true)
end

def run_subscriber(strategy)
  conn = Bunny.new
  conn.start
  ch = conn.create_channel
  ch.confirm_select
  ch.prefetch(ACK_SIZE)
  q = ch.queue(QUEUE_NAME, QUEUE_ARGS)

  delivery_tags = []

  q.subscribe(consumer_tag: "test", manual_ack: true) do |delivery_info, properties, payload|
    if delivery_tags.nil?
      next
    end

    delivery_tags << delivery_info.delivery_tag
    if delivery_tags.length % 100 == 0
      puts "received_message: #{delivery_tags.length}"
    end

    if delivery_tags.length == ACK_SIZE
      tags = delivery_tags
      delivery_tags = nil
      puts "acknowledging_messages"
      strategy.call(ch, tags)
      puts "finished_acknowledging_messages"

      puts "trying to publish"
      elapsed = time do
        q.publish("timing message", persistent: true)
        ch.wait_for_confirms
      end
      puts "consumer_publish_time: #{elapsed}"
    end
  end

end

def time
  start = Time.now
  yield
  finish = Time.now
  finish - start
end

def run_publisher

  conn = Bunny.new(continuation_timeout: 30_000)
  conn.start
  begin

    ch = conn.create_channel
    ch.confirm_select
    q = ch.queue(QUEUE_NAME, QUEUE_ARGS)

    loop do
      elapsed = time do
        q.publish("timing message", persistent: true)
        ch.wait_for_confirms
      end
      puts "publish_time: #{elapsed}"
      sleep_time = [0, PUBLISH_INTERVAL - elapsed].max
      Kernel.sleep(sleep_time)
    end
  ensure
    conn.stop
  end

end

if ARGV.length != 1
  puts "usage: #{$PROGRAM_NAME} lifo_strategy|fifo_strategy|multiple_strategy"
  exit(1)
end

seed_queue
run_subscriber(method(ARGV[0].to_sym))
run_publisher

[michaelklishin]

Publishers go through the channel and won't be blocked immediately, though (only via credit flow).
I'd recommend using Looking Glass, a profiler we've developed for longer running tests. @acogoluegnes was going to do just that, in fact.

[acogoluegnes]

I included the map-based ack-ing and some tests (2000 messages to ack, 1000 iterations). Tests can be run with make ct-unit_rabbit_queue_consumers t=performance_tests. Reverse-order ack is much faster, but other cases are slower:

*** User 2018-02-20 15:45:53.881 ***
Queue single ack fifo 782 ms

*** User 2018-02-20 15:45:55.282 ***
Map single ack fifo 1401 ms

*** User 2018-02-20 15:49:51.164 ***
Queue single ack lifo 235882 ms

*** User 2018-02-20 15:49:52.597 ***
Map single ack lifo 1433 ms

*** User 2018-02-20 15:49:53.133 ***
Queue multiple ack lifo 536 ms

*** User 2018-02-20 15:49:54.412 ***
Map multiple ack lifo 1279 ms

[michaelklishin]

With the multi-ack case, does LIFO vs. FIFO really matter? LIFO with multi-ack should be no different from single ack if we start at the end. Should we use a multi-ack FIFO test instead?

[acogoluegnes]

I tried a multiple ack with "ack in the middle", I got:

*** User 2018-02-20 16:27:08.161 ***
Queue multiple ack in the middle 523 ms

*** User 2018-02-20 16:27:09.224 ***
Map multiple ack in the middle 1063 ms

So not much different than multiple acking at the very end.

[acogoluegnes]

Not sure we'll come up with a data structure that will fix the worse case scenario without affecting the most common ones. We may try to detect we're hitting the degenerate case by peeking the rear of the queue after a few failed attempts to find the passed-in tag?

[jothan]

Does Erlang have a fast btree structure that would be appropriate for this ?

[michaelklishin]

We've found no solution that doesn't affect the much more common case. Closing this for now. If this becomes a common enough complaints we might revisit this again after a period of time. Note that there were changes to confirm tracking recently for unrelated reasons.

[avtc]

Encountered the issue, when having 10000 or more unacknowledged messages in queue, having single consumer, and making acknowledgements to random ones.
As a result publishers are not able to publish messages anymore, until all acknowledgements are processed by RabbitMQ.
RabbitMQ processing them in about 200 ack per 10 seconds.
And when all acknowledgements are processed, queue is quickly filled with new 20000 messages from publisher (that was generated during ack lockdown), and publish stuck again.

[michaelklishin]

@avtc we don't have much to add to the comment above. We have tried a couple of approaches, they all hurt the common case (and thousands of unacknowledged messages delivered to a single consumer is not the common case). If you have an idea of a data structure that would not introduce any regressions, we'd welcome a pull request. This is open source software, after all.

[michaelklishin]

This reminds me: this is only an issue when there's a large number of outstanding deliveries on a single channel. A large number of outstanding deliveries on many channels won't suffer from the same negative effects. It also won't apply to workloads where a large number of outstanding deliveries are acknowledged mostly in the FIFO order. So FWIW our team's empirical experience suggests that this is not something that affects a lot of workloads.