diff options
| author | Matthias Radestock <matthias@lshift.net> | 2010-04-26 17:52:22 +0100 |
|---|---|---|
| committer | Matthias Radestock <matthias@lshift.net> | 2010-04-26 17:52:22 +0100 |
| commit | 7c667f7005d66983188ae21c4e498ba219830f93 (patch) | |
| tree | cd635ecb5d36cfa72fcca97523381af99060cd51 /src | |
| parent | 765cb302de7b1e01a84490d56f4b8ed8e0f32801 (diff) | |
| download | rabbitmq-server-git-7c667f7005d66983188ae21c4e498ba219830f93.tar.gz | |
ensure queues are fully initialsed before we advertise their presence
Previously in a clustered setup a channel on another node could have
found a recently recovered queue before that queue had been fully
initialised and recovered its content.
Fixing that turns out to be simple since the main obstacle - the
re-declaration check for recovered durable queues and the associated
atomic update of the record with the new queue pid - can be removed
since it became superfluous with the changes in bug 20916.
Diffstat (limited to 'src')
| -rw-r--r-- | src/rabbit_amqqueue.erl | 24 |
1 files changed, 3 insertions, 21 deletions
diff --git a/src/rabbit_amqqueue.erl b/src/rabbit_amqqueue.erl index d705909a48..4e12bb7d47 100644 --- a/src/rabbit_amqqueue.erl +++ b/src/rabbit_amqqueue.erl @@ -144,30 +144,12 @@ find_durable_queues() -> end). recover_durable_queues(DurableQueues) -> - Qs = lists:foldl( - fun (RecoveredQ, Acc) -> - Q = start_queue_process(RecoveredQ, false), - %% We need to catch the case where a client - %% connected to another node has deleted the queue - %% (and possibly re-created it). - case rabbit_misc:execute_mnesia_transaction( - fun () -> - case mnesia:match_object( - rabbit_durable_queue, RecoveredQ, - read) of - [_] -> ok = store_queue(Q), - true; - [] -> false - end - end) of - true -> [Q | Acc]; - false -> exit(Q#amqqueue.pid, shutdown), - Acc - end - end, [], DurableQueues), + Qs = [start_queue_process(Q, false) || Q <- DurableQueues], %% Issue inits to *all* the queues so that they all init at the same time [ok = gen_server2:cast(Q#amqqueue.pid, init_backing_queue) || Q <- Qs], [ok = gen_server2:call(Q#amqqueue.pid, sync, infinity) || Q <- Qs], + rabbit_misc:execute_mnesia_transaction( + fun () -> [ok = store_queue(Q) || Q <- Qs] end), Qs. declare(QueueName, Durable, AutoDelete, Args) -> |