Message routing distribution among queues is not uniformly random

[mlmitch]

Overview

Hi RabbitMQ team. Before getting to the problem, I wanted to say that this exchange has worked well in our production environment. We use it to execute distributed caching in our applications, and it has been very reliable.

In our environment, we have 48 queues bound to a consistent hash exchange and each queue is bound with routing key "1". i.e. All have equal weighting. We are consistently pushing 1000 messages per second through this exchange. I noticed that the message distribution wasn't equal because some queues were receiving over 100 messages per second, and others were receiving close to 0 messages per second. These queues should all be receiving close to 20 messages per second. The 48 queues come from having 12 application instances create dedicated queues for themselves on every node of a 4-node cluster (4 * 12 = 48).

That is the problem "as seen in the wild". This negatively impacts us as the applications need to be provisioned for the worst case. i.e. Since one application instance has to handle hundreds of messages per second, they all have to. This is the reality of our deployment environment and we would rather provision our apps like this than put in the engineering to have per-instance app sizes based on queue publish rates.

Experiment

This problem description is a little vague though. I created an experiment to help diagnose the issue. This Gist has the final form of the experiment. The script has three modes: load, analyze and reset. The load mode creates the exchange, binds some number of queues, and publishes a large number of messages to the exchange. The analyze mode gets the number of messages in each queue and outputs some stats. The reset mode deletes the queues and exchanges. Here is example usage:

./consistent_hash_exchange_experiment.py user password hostname port vhost load
./consistent_hash_exchange_experiment.py user password hostname port vhost analyze
./consistent_hash_exchange_experiment.py user password hostname port vhost reset

Other parameters are hard coded inside the script.

Wrong Guesses

There were a few theories I had which turned out to be wrong. I'll explain them and indicate how this is handled in the experiment now.

1. Repeated routing key
   My first guess is that certain routing keys were occurring more than others. Since the messages have the same routing key, they would go to the same queue due to consistent hashing. I verified this is not the case in our production environment. The experiment script uses a unique routing key for each message to accommodate.

2. Bad hash function
   After some googling, I came across this Erlang mailing list thread. This made me think that the use of phash2 was the issue. I don't think this is the case anymore since I modified our production environment as well as the experiment to set the routing key to a cryptographic hash of the previous routing key. The issue still persisted after the change.

3. Maybe this is actually from a uniform distribution
   Just because message distribution is supposed to be uniform across queues, it doesn't mean it will happen exactly like this in practice. The experiment writes the messages to the exchange and then messages are routed to the queues and sit in the queues. The experiment then queries the number of messages in each queue. The Chi-Squared goodness of fit test gives an indication of how likely it is that a set of data came from a given distribution. scipy.stats.chisquare performs this analysis on a List of bucket counts for the uniform distribution over the number of buckets. It outputs a p-value which can be interpreted as "what is the probability that these counts came from a uniform distribution". The p-values generated during most (I'll get to this later) experiments are extremely small.

Current Working Theory

The above incorrect theories pointed to the issue being with how buckets are assigned. This was somewhat confirmed by the source code. What follows is my understanding of how buckets are assigned. When a queue is bound to the exchange, a number of points in the hash space are generated and assigned to that queue (I call them "queue points"). Then when a message comes in, the routing key is hashed and the message is routed to the queue corresponding to the first queue point that the message point is greater than. I think this routing mechanism is fine. The problem lies in how the queue points are generated.

The number of queue points is equal to the number put in the routing key during binding. The queue points are randomly chosen. They are chosen according to a uniform distribution over [0, HASH_MAX] using phash. Since the points are generated randomly, there is no guarantee that "bucket sizes" will be equal. I believe the discrepancy in bucket size using this method is the cause of the non-uniform distribution for message routing.

Playing around with the routing key used for queue binding indicates this is the case. For example, binding all queues with "10" for the routing key gives each queue 10 buckets. Since each queue now has more randomly sized buckets, their collective share of messages should be more equal. This has been confirmed in our production environment where using "10" for the routing key makes the situation a little better, but does not solve the problem. Additionally, running the routing experiment with higher and higher binding keys produces higher p-values. There is a limit to this trick since performance drastically degrades. e.g. Running the experiment with the binding key set to "10000" with 100 queues produced a p-value of about 0.6 which is a good indication that the message distribution close to uniformly distributed. However, only 5 messages per second could be published to the exchange (using a binding key of "10" permitted 2500 messages per second). I suspect this is due to a prohibitively large mnesia table that stores queue points.

Suggested Fix

For anyone reading this with the same problem, I suggest bumping up the binding keys you are using as a workaround. Like I said, this won't fix the situation, but it will make it a little better. Also, don't make the values too large or you'll incur a performance hit.

I suggest the examples for this exchange be changed to use slightly higher binding keys in the short term. Maybe change some of the documentation language too? I'm not sure about this one.

In order to fix the issue, I think a better allocation strategy is needed for queue points. The easiest example to think of is that binding N queues with a binding routing key of "1" should result in N equally sized buckets. However, the general case of unequal weightings needs to be accommodated, and I'm not sure how to accomplish this in a smart manner like I'm suggesting.

Thanks for reading this issue, which seems more like a blog post now that I've written it! I'd be thrilled to discuss potential solutions or work on a PR if their is appetite.

Relevant Version Information

RabbitMQ Version: 3.7.6
Erlang Version: 20.3
Consisten Hash Exchange Version: 3.7.6
Server OS: Ubuntu 16.04 Server with 4.4.0-97-generic Kernel
Python Version: 3.5.2
Pika Version: 0.10.0
Client OS: Ubuntu 16.04 Server with 4.4.0-130-generic Kernel

[michaelklishin]

Thank you for the details and spending time to investigate. We'll have to discuss what options we have with our team.

[michaelklishin]

@mlmitch our team concluded that we'd need to pick a distribution criteria for this plugin since anything we change would benefit some workloads but not others. Can you help us figure out what options are there? I haven't done much statistics in a while but willing to catch up :) Cheers.

[mlmitch]

I wonder if there are performance improvements that could be made in the mnesia access of queue points. Setting extremely high routing keys yield a distribution which is pretty close to uniform. As I mentioned, this doesn't work because publishing rates are negatively impacted.

Some brief searching has indicated that there are other schemes which will give the desired distribution. These schemes seem to based on other data structures (e.g. tree-based decision).

I'm curious what your thoughts are on some sort of in-memory caching. It could alleviate some of the mnesia stress when there are a large number of queue points. It might also be necessary to enable alternative schemes that give a better routing distribution.

Thanks for the time you've spent on this already! I'll comment again about various consistent hashing mechanisms and I'll try to weigh them against the RabbitMQ use case.

[michaelklishin]

We are definitely not married to Mnesia but it does have powerful querying capabilities. This plugin uses a single node-local, RAM-only replica, which is trivial to replace compatibility-wise. So we don't have any reasons not to replace it with an in-memory data structure that fits our needs better.

[michaelklishin]

@mlmitch @dcorbacho and I plan to apply a couple of changes:

• Enforce one binding per queue for this exchange, most likely with a "last binding wins" kind of semantics. It makes a lot of sense for this plugin anyway.
• Use binding weight (the value in the routing key) to compute bucket range/width in the most straightforward way (e.g. with 2 bindings with routing keys 10 and 20 the former would get 1/3rd of the range)

A little experiment suggests that the results of the Chi square test you provided are significantly better. We can also adopt this test to the test suite.

Any thoughts on the plan?

[michaelklishin]

One reason to stick to Mnesia here is that it's one of the mechanisms that allow all channels in a virtual host [on the local node] share the hash space/bucket list.

[mlmitch]

There is an issue with computing bucket size directly from the binding weight. This method will get the distribution that we want in a straightforward manner - I'm glad the Chi square test verifies it. However, it alters how keys get redistributed when a queue is added or removed.

The current hash ring setup hash a very nice property. Going from n queues to n+1 queues (with equal weights) will result in about (1/n) * keySpaceSize of the keys being reassigned to the new queue, and they will be drawn in roughly equal proportions from all the existing queues. This depends on the binding weights though, and as binding weights go to infinity the key transfer is exactly (1/n) * keySpaceSize keys drawn equally from each of the currently bound queues.

If I interpret your scheme correctly, a large amount (I haven't work out how much) will be moved around. In our application, this would result in a significant number of cache misses when we scale up or scale down the number of application instances.

I'll write another comment with what I think is a better approach.

[mlmitch]

First, I think this page provides a good overview of the characteristics we want. I especially wanted to mention this page since it has pointers to other techniques.

Currently, I think there are two good ways forward

Path 1
Possibly the most straightforward option is to keep using the same scheme, but make it more performant. Enabling more queue points in the current scheme would produce better routing distributions and maintain the reassignment property I mentioned previously. However, a performance boost big enough to give uniform distribution might not be possible.

Path 2
This one is a bigger change, but I think it warrants some consideration. Google's jump-consistent hash algorithm provides a very attractive primitive to build some sort of scheme out of.

The algorithm is comprised of a stateless function that takes in a (key, number of buckets) and outputs a bucket number. They show that it has very even distribution across the buckets. Additionally, they claim it draws evenly from all buckets when increasing from n to n+1 buckets. Consequently, it distributes evenly when decreasing from n+1 to n buckets.

The challenge with this path is assigning buckets to queues. We want to support non-equal weightings across queues and accommodate graceful upsizing and downsizing when any queue is unbound from the exchange. I'll elaborate how I think this should be done.

1. Maintain a map from bucket number to queue. An ArrayList would give extremely efficient routing. Though some sort of BiMap might be better for the queue unbinding step explained below.
2. The number of buckets is equal to the sum of binding weights.
3. Each queue receives a number of buckets equal to their binding weight.

These three rules guarantee an ideal key distribution if John Lamping and Eric Veach's assertions hold true - even with minimal queue bindings.

4. When adding a queue, add its binding weight to the number of buckets and that queue receives the new buckets at the end of the map.

If Lamping and Veach are right, this guarantees optimal redistribution when adding a queue. Deleting/unbinding a queue is the difficult case though.

5. When unbinding a queue, first find the buckets that are assigned to it and its bindingWeight. This is where a BiMap would help.
6. Assign the buckets that belong to this queue to the bindingWeight queue values at the end of of the map.
7. Remove the last bindingWeight buckets from the map.

This does not provide optimal redistribution. However, we are guaranteed that only the queues which are reassigned to new buckets are adversely affected.

With regard to backward compatibility, its only the queue unbinding behaviour that could be worse in certain scenarios. However, queue unbinding behaviour isn't great currently given that a deleted queue might have been getting a disproportionately high number of messages.

[michaelklishin]

@mlmitch we understand what consistent hashing is and its benefits over, say, modulo-based hashing. However, we also know how this plugin is used in practice or at least recommended and it has
important differences from the canonical case (content delivery from N endpoints).

In this plugin's case, it is very common that

• The topology is mostly static
• Messages are consumed most of the time

which means once a binding is added or deleted, the temporary uneven distribution disappears over time assuming that the distribution function is reasonably uniform. I don't know if this can be called "eventual uniform distribution" 😃 but it works that way.

So a radically simpler solution that does not migrate data around affected queues (as Riak or Cassandra would when their ring members changes) might be perfectly acceptable in practice. The users of this plugin typically want uniform distribution most of the time so that they can parallelise competing consumers. This is a much simpler scenario than, say, even a distributed caching system in certain ways.

For example, to consume the data consumers need to know queue name. They would not be performing a hash table lookup the same way Riak or Cassandra clients do when they perform an operation.

We will take a look at your suggestion as the jump-consistent hashing almost sounds too good to be true and the paper is short.

@dcorbacho FYI.

[michaelklishin]

@mlmitch we discussed the options with @dcorbacho and here's our thinking. It will overlap with what I stated above.

This plugin has a couple of assumptions:

• There is only one binding between this exchange and a queue
• This is an exchange: it is meant to route messages but isn't concerned with message storage, rebalancing and so on

Its users want it to (from our experience over the course of a few years):

• Distribute messages between the currently bound queues as uniformly as possible
• Make it easier to have parallel competing consumers (or at least consumers of the same "type", even if they don't actually compete)
• Introduce as little overhead as possible

The consumers consume messages the way they do with any other exchange, namely: they have to
know queue names (and often they declare the queues). In other words, consumers do not rely
on the consistent hashing ring for data location for "reads" (which are actually writes since they modify queue state).

This is pretty different from the original use case for consistent hashing (distributed caches) and what data stores that use consistent hashing for data distribution and locality (Riak and Cassandra are two well known examples).

Therefore we think that it is appropriate for this plugin which is an exchange to focus on uniformity
and efficiency. Our team also would like it to focus on implementation simplicity since the current algorithm is not obvious to the reader. Note that data redistribution after ring topology changes are not in scope (in fact, they never are for any exchange type in RabbitMQ).

To be completely honest, a basic module-based hashing plugin would largely accomplish the same thing. We even have one in the rabbitmq-sharding plugin. FWIW historically this plugin has been
named after (and used) consistent hashing, so that's not going to change.

We will investigate the jump-consistent hashing function because it claims to provide nearly optimal uniformity and from the looks of it, should be pretty efficient. The scope of implementation changes is fine with us. This may or may not satisfy some of the other topics you are concerned with, e.g. "cache misses" (as explained above, consumers do not use the ring for data locality, so we are not sure what exactly that means in the context of this plugin).

Let us know what you think.

[michaelklishin]

@dcorbacho and I did an experiment with the jump-consistent hash function just in case. Its computed Chi^2 value for a certain number of buckets as well as p-value of 0.05 and p-value of 0.01 are presented below.

We had to work with the Erlang PRNG and seeding APIs available, which are different from the PRNG
algorithm used in the paper. Here's our code:

-module(experiment).

-export([jump_consistent_hash/2, compute/1]).

-define(MSGPQ, 10000).

jump_consistent_hash(_Key, 0) ->
    0;
jump_consistent_hash(Key, NumberOfBuckets) ->
    SeedState = rand:seed_s(exs1024s, {Key, Key, Key}),
    jump_consistent_hash_value(-1, 0, NumberOfBuckets, SeedState).

jump_consistent_hash_value(B, J, NumberOfBuckets, _SeedState) when J >= NumberOfBuckets ->
    B;

jump_consistent_hash_value(_B0, J0, NumberOfBuckets, SeedState0) ->
    B = J0,
    {R, SeedState} = rand:uniform_s(SeedState0),
    J = math:floor((B + 1) / R),
    jump_consistent_hash_value(B, J, NumberOfBuckets, SeedState).



run(Buckets) ->
    run(Buckets, ?MSGPQ * Buckets,
        maps:from_list([{I, 0} || I <- lists:seq(0, Buckets - 1)])).

run(_, 0, Map) ->
    [V || {_, V} <- maps:to_list(Map)];
run(B, N, Map) ->
    Bucket = erlang:trunc(jump_consistent_hash(N, B)),
    run(B, N - 1, maps:update_with(Bucket, fun(V) -> V + 1 end, Map)).

compute(Buckets) ->
    Obs = run(Buckets),
    Chi = lists:sum([((O - ?MSGPQ) * (O - ?MSGPQ)) / ?MSGPQ || O <- Obs]),
    %% io:format("Result: ~p Chi-squared: ~p~n", [Obs, Chi]),
    Chi.

As far as we can tell the Ch^2 value is lower than even the p-value of 0.01 by a very comfortable margin. So even with a slight deviation from the function as presented in the paper we still get a uniform distribution:

Number of buckets	Chi-squared	Degrees of freedom	p = 0.05	p = 0.01
2	0.5	1	3.84	6.64
3	0.946	2	5.99	9.21
4	2.939	3	7.81	11.35
5	2.163	4	3.49	13.28
6	2.592	5	11.07	15.09
7	4.654	6	12.59	16.81
8	7.566	7	14.07	18.48
9	5.847	8	15.51	20.09
10	9.790	9	16.92	21.67
11	13.448	10	18.31	23.21
12	12.432	11	19.68	24.73
13	12.338	12	21.02	26.22
14	9.898	13	22.36	27.69
15	8.513	14	23.69	29.14
16	6.997	15	24.99	30.58
17	6.279	16	26.30	32.00
18	10.373	17	28.87	34.81
19	12.935	18	30.14	36.19
20	11.895	19	31.41	37.57

So far, so good.

[michaelklishin]

Switching to Jump-consistent hashing will also require making the ring state this plugin uses to be distributed across all cluster members (currently it isn't). It makes little sense with the current implementation but even less so with Jump-consistent hashing.

[mlmitch]

@mlmitch we understand what consistent hashing is and its benefits over, say, module-based hashing

I thought as much given that you wrote the plugin! Sorry if the comment came off a little 'tutorial-like'.

Note that data redistribution after ring topology changes are not in scope (in fact, they never are for any exchange type in RabbitMQ)

Fair enough. Data redistribution isn't a huge concern for us. It's just a 'nice to have'. All that will happen is a bit of cache churn if the redistribution isn't great.

---

@michaelklishin it seems like you have a good handle on this. Whatever you end up choosing will be an improvement. I've been very impressed with the level of effort you've put into the ticket.

[michaelklishin]

@mlmitch below is a new version of this plugin. Please unzip it and replace the one that ships with 3.7.7 in a separate installation (e.g. a generic UNIX build) and give it a shot. Note that we explicitly recommend lower binding weights in the docs; hopefully they should work better with Jump Consistent Hash.
We also update the docs in master to provide more code examples and have better structure.
Let us know how it goes!

rabbitmq_consistent_hash_exchange-git.af9beb0.ez.zip

[mlmitch]

Awesome.

Small bug I suspect:
I ran my original script in our development environment. No messages are delivered to queues. Looks like string routing keys aren't being handled properly.

{badarg,
  [{erlang, binary_to_integer,
    [<<"string routeing key">>],
    []},
  {rabbitmq_data_coercion,to_integer,1,
    [{file, "src/rabbitmq_data_coercion.erl"},{line,48}]},
  {rabbitmq_exchange_type_consistent_hash,jump_consistent_hash,2,
        [{file, "src/rabbitmq_exchange_type_consistent_hash.erl"},{line,260}]},
  {rabbitmq_exchange_type_consistent_hash,route,2,
        [{file, "src/rabbitmq_exchange_type_consistent_hash.erl"},{line,93}]},jump_consistent_hash
...

Let me know if you need more logs.

[michaelklishin]

It's not string keys, it's string keys that cannot be parsed to integers.

[michaelklishin]

@mlmitch ready for another round rabbitmq_consistent_hash_exchange-git.82fccac.dirty.ez.zip

[mlmitch]

We're going to wait until the 3.7.8 release to put this into production. However, everything looks good in our development environment currently.

Thanks for all the hard work @michaelklishin
You've been a huge help!

[michaelklishin]

Sure, we just try to make the reporter verify development versions if possible. Thank you for confirming!

[michaelklishin]

@mlmitch we have found a few bugs around hash ring state and binding management, some were addressed, some are in progress. A lot more tests are needed.

However, we have some good news, too: according to a different distribution variability test this version has a drastically lower rate of variability between bucket sizes (stddev goes from about 2400 to about 120).

[mlmitch]

Awesome!

And no problem. Take your time with regression testing. We really appreciate it.

[michaelklishin]

This should be addressed by #39 (although we still plan on adding a few more test cases). In the process we have identified some specific issues in rabbitmq/rabbitmq-server#1589 and filed rabbitmq/rabbitmq-server#1690, so it's been a long running issue but productive in more ways than addressing the uniformity of routing.

[michaelklishin]

Leaving this gist here for my own future reference: it helped us uncover a core server issue and come up with the final implementation in #39.