Merge pull request #274 from rabbitmq/quorum-queue

Quorum queue

Author: kjnilsson

Makefile

@@ -47,6 +47,8 @@ DEP_PLUGINS = $(PROJECT)/mk/rabbitmq-build.mk \
 
 WITHOUT = plugins/proper
 
+PLT_APPS += mnesia crypto ssl
+
 include mk/rabbitmq-components.mk
 include erlang.mk
 

include/rabbit.hrl

@@ -143,7 +143,9 @@
           policy_version,
           slave_pids_pending_shutdown,
           vhost,                       %% secondary index
-          options = #{}}).
+          options = #{},
+          type = classic,
+          quorum_nodes }).
 
 -record(exchange_serial, {name, next}).
 
@@ -234,8 +236,8 @@
 
 -define(COPYRIGHT_MESSAGE, "Copyright (C) 2007-2018 Pivotal Software, Inc.").
 -define(INFORMATION_MESSAGE, "Licensed under the MPL.  See http://www.rabbitmq.com/").
--define(OTP_MINIMUM, "19.3").
--define(ERTS_MINIMUM, "8.3").
+-define(OTP_MINIMUM, "21.0").
+-define(ERTS_MINIMUM, "10.0").
 
 %% EMPTY_FRAME_SIZE, 8 = 1 + 2 + 4 + 1
 %%  - 1 byte of frame type

mk/rabbitmq-run.mk

@@ -64,6 +64,7 @@ node_pid_file = $(call node_tmpdir,$(1))/$(1).pid
 node_log_base = $(call node_tmpdir,$(1))/log
 node_mnesia_base = $(call node_tmpdir,$(1))/mnesia
 node_mnesia_dir = $(call node_mnesia_base,$(1))/$(1)
+node_quorum_dir = $(call node_mnesia_dir,$(1))/quorum
 node_schema_dir = $(call node_tmpdir,$(1))/schema
 node_plugins_expand_dir = $(call node_tmpdir,$(1))/plugins
 node_generated_config_dir = $(call node_tmpdir,$(1))/config
@@ -78,6 +79,7 @@ RABBITMQ_PID_FILE ?= $(call node_pid_file,$(RABBITMQ_NODENAME_FOR_PATHS))
 RABBITMQ_LOG_BASE ?= $(call node_log_base,$(RABBITMQ_NODENAME_FOR_PATHS))
 RABBITMQ_MNESIA_BASE ?= $(call node_mnesia_base,$(RABBITMQ_NODENAME_FOR_PATHS))
 RABBITMQ_MNESIA_DIR ?= $(call node_mnesia_dir,$(RABBITMQ_NODENAME_FOR_PATHS))
+RABBITMQ_QUORUM_DIR ?= $(call node_quorum_dir,$(RABBITMQ_NODENAME_FOR_PATHS))
 RABBITMQ_SCHEMA_DIR ?= $(call node_schema_dir,$(RABBITMQ_NODENAME_FOR_PATHS))
 RABBITMQ_PLUGINS_EXPAND_DIR ?= $(call node_plugins_expand_dir,$(RABBITMQ_NODENAME_FOR_PATHS))
 RABBITMQ_GENERATED_CONFIG_DIR ?= $(call node_generated_config_dir,$(RABBITMQ_NODENAME_FOR_PATHS))
@@ -105,6 +107,7 @@ RABBITMQ_PID_FILE="$(call node_pid_file,$(2))" \
 RABBITMQ_LOG_BASE="$(call node_log_base,$(2))" \
 RABBITMQ_MNESIA_BASE="$(call node_mnesia_base,$(2))" \
 RABBITMQ_MNESIA_DIR="$(call node_mnesia_dir,$(2))" \
+RABBITMQ_QUORUM_DIR="$(call node_quorum_dir,$(2))" \
 RABBITMQ_SCHEMA_DIR="$(call node_schema_dir,$(2))" \
 RABBITMQ_GENERATED_CONFIG_DIR="$(call node_generated_config_dir,$(2))" \
 RABBITMQ_PLUGINS_DIR="$(RMQ_PLUGINS_DIR)" \
@@ -162,6 +165,9 @@ define test_rabbitmq_config
 [
   {rabbit, [
       {loopback_users, []}
+    ]},
+  {ra, [
+      {data_dir, "$(RABBITMQ_QUORUM_DIR)"}
     ]}
 ].
 endef
@@ -193,7 +199,10 @@ define test_rabbitmq_config_with_tls
             {fail_if_no_peer_cert, false},
             {honor_cipher_order, true}]}
         ]}
-  ]}
+  ]},
+  {ra, [
+      {data_dir, "$(RABBITMQ_QUORUM_DIR)"}
+    ]}
 ].
 endef
 
@@ -310,6 +319,7 @@ start-brokers start-cluster:
 		  RABBITMQ_NODE_PORT="$$((5672 + $$n - 1))" \
 		  RABBITMQ_SERVER_START_ARGS=" \
 		  -rabbit loopback_users [] \
+		  -ra data_dir "$(RABBITMQ_QUORUM_DIR)"
 		  -rabbitmq_management listener [{port,$$((15672 + $$n - 1))}] \
 		  "; \
 		if test '$@' = 'start-cluster' && test "$$nodename1"; then \

src/credit_flow.erl

@@ -62,6 +62,7 @@
 
 -export([send/1, send/2, ack/1, ack/2, handle_bump_msg/1, blocked/0, state/0]).
 -export([peer_down/1]).
+-export([block/1, unblock/1]).
 
 %%----------------------------------------------------------------------------
 

src/pmon.erl

@@ -34,62 +34,62 @@
 
 -compile({no_auto_import, [monitor/2]}).
 
--record(state, {dict, module}).
+-record(state, {monitors = #{} :: #{item() => reference()},
+                module = erlang :: module()}).
 
 %%----------------------------------------------------------------------------
 
 -export_type([?MODULE/0]).
 
--opaque(?MODULE() :: #state{dict   :: dict:dict(),
-                            module :: atom()}).
+-opaque(?MODULE() :: #state{}).
 
--type(item()         :: pid() | {atom(), node()}).
+-type(item() :: pid() | {atom(), node()}).
 
--spec new() -> ?MODULE().
--spec new('erlang' | 'delegate') -> ?MODULE().
--spec monitor(item(), ?MODULE()) -> ?MODULE().
--spec monitor_all([item()], ?MODULE()) -> ?MODULE().
--spec demonitor(item(), ?MODULE()) -> ?MODULE().
--spec is_monitored(item(), ?MODULE()) -> boolean().
--spec erase(item(), ?MODULE()) -> ?MODULE().
--spec monitored(?MODULE()) -> [item()].
--spec is_empty(?MODULE()) -> boolean().
 
+-spec new() -> ?MODULE().
 new() -> new(erlang).
 
-new(Module) -> #state{dict   = dict:new(),
-                      module = Module}.
+-spec new('erlang' | 'delegate') -> ?MODULE().
+new(Module) -> #state{module = Module}.
 
-monitor(Item, S = #state{dict = M, module = Module}) ->
-    case dict:is_key(Item, M) of
+-spec monitor(item(), ?MODULE()) -> ?MODULE().
+monitor(Item, S = #state{monitors = M, module = Module}) ->
+    case maps:is_key(Item, M) of
         true  -> S;
         false -> case node_alive_shortcut(Item) of
                      true  -> Ref = Module:monitor(process, Item),
-                              S#state{dict = dict:store(Item, Ref, M)};
+                              S#state{monitors = maps:put(Item, Ref, M)};
                      false -> self() ! {'DOWN', fake_ref, process, Item,
                                         nodedown},
                               S
                  end
     end.
 
+-spec monitor_all([item()], ?MODULE()) -> ?MODULE().
 monitor_all([],     S) -> S;                %% optimisation
 monitor_all([Item], S) -> monitor(Item, S); %% optimisation
 monitor_all(Items,  S) -> lists:foldl(fun monitor/2, S, Items).
 
-demonitor(Item, S = #state{dict = M, module = Module}) ->
-    case dict:find(Item, M) of
-        {ok, MRef} -> Module:demonitor(MRef),
-                      S#state{dict = dict:erase(Item, M)};
+-spec demonitor(item(), ?MODULE()) -> ?MODULE().
+demonitor(Item, S = #state{monitors = M0, module = Module}) ->
+    case maps:take(Item, M0) of
+        {MRef, M} -> Module:demonitor(MRef),
+                     S#state{monitors = M};
         error      -> S
     end.
 
-is_monitored(Item, #state{dict = M}) -> dict:is_key(Item, M).
+-spec is_monitored(item(), ?MODULE()) -> boolean().
+is_monitored(Item, #state{monitors = M}) -> maps:is_key(Item, M).
 
-erase(Item, S = #state{dict = M}) -> S#state{dict = dict:erase(Item, M)}.
+-spec erase(item(), ?MODULE()) -> ?MODULE().
+erase(Item, S = #state{monitors = M}) ->
+    S#state{monitors = maps:remove(Item, M)}.
 
-monitored(#state{dict = M}) -> dict:fetch_keys(M).
+-spec monitored(?MODULE()) -> [item()].
+monitored(#state{monitors = M}) -> maps:keys(M).
 
-is_empty(#state{dict = M}) -> dict:size(M) == 0.
+-spec is_empty(?MODULE()) -> boolean().
+is_empty(#state{monitors = M}) -> maps:size(M) == 0.
 
 %%----------------------------------------------------------------------------