summaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/gm.erl2
-rw-r--r--src/rabbit_amqqueue_process.erl2
-rw-r--r--src/rabbit_backing_queue.erl2
-rw-r--r--src/rabbit_channel.erl2
-rw-r--r--src/rabbit_feature_flags.erl2
-rw-r--r--src/rabbit_ff_registry.erl2
-rw-r--r--src/rabbit_fifo.erl265
-rw-r--r--src/rabbit_fifo_client.erl8
-rw-r--r--src/rabbit_fifo_index.erl2
-rw-r--r--src/rabbit_guid.erl2
-rw-r--r--src/rabbit_lager.erl4
-rw-r--r--src/rabbit_mirror_queue_coordinator.erl2
-rw-r--r--src/rabbit_mirror_queue_misc.erl2
-rw-r--r--src/rabbit_msg_store.erl4
-rw-r--r--src/rabbit_node_monitor.erl2
-rw-r--r--src/rabbit_priority_queue.erl4
-rw-r--r--src/rabbit_queue_index.erl2
-rw-r--r--src/rabbit_reader.erl4
-rw-r--r--src/rabbit_upgrade.erl4
-rw-r--r--src/rabbit_variable_queue.erl4
20 files changed, 174 insertions, 147 deletions
diff --git a/src/gm.erl b/src/gm.erl
index 02ee76cd60..7b9bbc8c8a 100644
--- a/src/gm.erl
+++ b/src/gm.erl
@@ -1271,7 +1271,7 @@ neighbour_cast(N, Msg) -> ?INSTR_MOD:cast(get_pid(N), Msg).
neighbour_call(N, Msg) -> ?INSTR_MOD:call(get_pid(N), Msg, infinity).
%% ---------------------------------------------------------------------------
-%% View monitoring and maintanence
+%% View monitoring and maintenance
%% ---------------------------------------------------------------------------
ensure_neighbour(_Ver, Self, {Self, undefined}, Self) ->
diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl
index b467e6c02e..e1978a9173 100644
--- a/src/rabbit_amqqueue_process.erl
+++ b/src/rabbit_amqqueue_process.erl
@@ -1195,7 +1195,7 @@ prioritise_cast(Msg, _Len, State) ->
%% will be rate limited by how fast consumers receive messages -
%% i.e. by notify_sent. We prioritise ack and resume to discourage
%% starvation caused by prioritising notify_sent. We don't vary their
-%% prioritiy since acks should stay in order (some parts of the queue
+%% priority since acks should stay in order (some parts of the queue
%% stack are optimised for that) and to make things easier to reason
%% about. Finally, we prioritise ack over resume since it should
%% always reduce memory use.
diff --git a/src/rabbit_backing_queue.erl b/src/rabbit_backing_queue.erl
index ef7a6a2337..187f55288d 100644
--- a/src/rabbit_backing_queue.erl
+++ b/src/rabbit_backing_queue.erl
@@ -158,7 +158,7 @@
%% Drop messages from the head of the queue while the supplied
%% predicate on message properties returns true. Returns the first
-%% message properties for which the predictate returned false, or
+%% message properties for which the predicate returned false, or
%% 'undefined' if the whole backing queue was traversed w/o the
%% predicate ever returning false.
-callback dropwhile(msg_pred(), state())
diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl
index d5dd236937..036aa9a60c 100644
--- a/src/rabbit_channel.erl
+++ b/src/rabbit_channel.erl
@@ -1453,7 +1453,7 @@ handle_method(#'basic.qos'{prefetch_size = Size}, _, _State) when Size /= 0 ->
handle_method(#'basic.qos'{global = false,
prefetch_count = PrefetchCount},
_, State = #ch{limiter = Limiter}) ->
- %% Ensures that if default was set, it's overriden
+ %% Ensures that if default was set, it's overridden
Limiter1 = rabbit_limiter:unlimit_prefetch(Limiter),
{reply, #'basic.qos_ok'{}, State#ch{consumer_prefetch = PrefetchCount,
limiter = Limiter1}};
diff --git a/src/rabbit_feature_flags.erl b/src/rabbit_feature_flags.erl
index 8ba38179a5..74f5a34db0 100644
--- a/src/rabbit_feature_flags.erl
+++ b/src/rabbit_feature_flags.erl
@@ -1411,7 +1411,7 @@ does_node_support(Node, FeatureNames, Timeout) ->
%% If rabbit_feature_flags:is_supported_locally/1 is undefined
%% on the remote node, we consider it to be a 3.7.x node.
%%
- %% Theoritically, it could be an older version (3.6.x and
+ %% Theoretically, it could be an older version (3.6.x and
%% older). But the RabbitMQ version consistency check
%% (rabbit_misc:version_minor_equivalent/2) called from
%% rabbit_mnesia:check_rabbit_consistency/2 already blocked
diff --git a/src/rabbit_ff_registry.erl b/src/rabbit_ff_registry.erl
index 2b2614bf39..46d439001f 100644
--- a/src/rabbit_ff_registry.erl
+++ b/src/rabbit_ff_registry.erl
@@ -148,7 +148,7 @@ is_registry_written_to_disk() ->
always_return_true() ->
%% This function is here to trick Dialyzer. We want some functions
%% in this initial on-disk registry to always return `true` or
- %% `false`. However the generated regsitry will return actual
+ %% `false`. However the generated registry will return actual
%% booleans. The `-spec()` correctly advertises a return type of
%% `boolean()`. But in the meantime, Dialyzer only knows about this
%% copy which, without the trick below, would always return either
diff --git a/src/rabbit_fifo.erl b/src/rabbit_fifo.erl
index 1fe0b38cd9..ee0f65950c 100644
--- a/src/rabbit_fifo.erl
+++ b/src/rabbit_fifo.erl
@@ -156,7 +156,7 @@
#update_config{}.
-type command() :: protocol() | ra_machine:builtin_command().
-%% all the command types suppored by ra fifo
+%% all the command types supported by ra fifo
-type client_msg() :: delivery().
%% the messages `rabbit_fifo' can send to consumers.
@@ -184,7 +184,7 @@
%% command: `{consumer_credit, ReceiverDeliveryCount, Credit}'
credit_mode = simple_prefetch :: credit_mode(), % part of snapshot data
lifetime = once :: once | auto,
- suspected_down = false :: boolean()
+ status = up :: up | suspected_down | cancelled
}).
-type consumer() :: #consumer{}.
@@ -193,7 +193,7 @@
{next_seqno = 1 :: msg_seqno(),
% out of order enqueues - sorted list
pending = [] :: [{msg_seqno(), ra_index(), raw_msg()}],
- suspected_down = false :: boolean()
+ status = up :: up | suspected_down
}).
-record(state,
@@ -323,7 +323,7 @@ apply(Meta,
case Cons0 of
#{ConsumerId := Con0} ->
% need to increment metrics before completing as any snapshot
- % states taken need to includ them
+ % states taken need to include them
complete_and_checkout(Meta, MsgIds, ConsumerId,
Con0, [], State);
_ ->
@@ -426,10 +426,7 @@ apply(Meta, #checkout{spec = {dequeue, Settlement},
end
end;
apply(Meta, #checkout{spec = cancel, consumer_id = ConsumerId}, State0) ->
- {State, Effects} = cancel_consumer(ConsumerId, State0, []),
- % TODO: here we should really demonitor the pid but _only_ if it has no
- % other consumers or enqueuers. leaving a monitor in place isn't harmful
- % however
+ {State, Effects} = cancel_consumer(ConsumerId, State0, [], consumer_cancel),
checkout(Meta, State, Effects);
apply(Meta, #checkout{spec = Spec, meta = ConsumerMeta,
consumer_id = {_, Pid} = ConsumerId},
@@ -466,26 +463,30 @@ apply(_, {down, ConsumerPid, noconnection},
% mark all consumers and enqueuers as suspected down
% and monitor the node so that we can find out the final state of the
% process at some later point
- {Cons, State, Effects1} = maps:fold(
- fun({_, P} = K,
- #consumer{checked_out = Checked0} = C,
- {Co, St0, Eff}) when node(P) =:= Node ->
- St = return_all(St0, Checked0),
- Credit = increase_credit(C, maps:size(Checked0)),
- Eff1 = ConsumerUpdateActiveFun(St, K, C, false, suspected_down, Eff),
- {maps:put(K, C#consumer{suspected_down = true,
- credit = Credit,
- checked_out = #{}}, Co),
- St, Eff1};
- (K, C, {Co, St, Eff}) ->
- {maps:put(K, C, Co), St, Eff}
- end, {#{}, State0, []}, Cons0),
+ {Cons, State, Effects1} =
+ maps:fold(fun({_, P} = K,
+ #consumer{checked_out = Checked0} = C,
+ {Co, St0, Eff}) when (node(P) =:= Node) and
+ (C#consumer.status =/= cancelled)->
+ St = return_all(St0, Checked0),
+ Credit = increase_credit(C, maps:size(Checked0)),
+ Eff1 = ConsumerUpdateActiveFun(St, K, C, false,
+ suspected_down, Eff),
+ {maps:put(K,
+ C#consumer{status = suspected_down,
+ credit = Credit,
+ checked_out = #{}}, Co),
+ St, Eff1};
+ (K, C, {Co, St, Eff}) ->
+ {maps:put(K, C, Co), St, Eff}
+ end, {#{}, State0, []}, Cons0),
Enqs = maps:map(fun(P, E) when node(P) =:= Node ->
- E#enqueuer{suspected_down = true};
+ E#enqueuer{status = suspected_down};
(_, E) -> E
end, Enqs0),
% mark waiting consumers as suspected if necessary
- WaitingConsumers = maybe_mark_suspect_waiting_consumers(Node, State0, true),
+ WaitingConsumers = update_waiting_consumer_status(Node, State0,
+ suspected_down),
Effects2 = case maps:size(Cons) of
0 ->
@@ -514,12 +515,12 @@ apply(Meta, {down, Pid, _Info}, #state{consumers = Cons0,
DownConsumers = maps:keys(
maps:filter(fun({_, P}, _) -> P =:= Pid end, Cons0)),
{State, Effects} = lists:foldl(fun(ConsumerId, {S, E}) ->
- cancel_consumer(ConsumerId, S, E)
+ cancel_consumer(ConsumerId, S, E, down)
end, {State2, Effects1}, DownConsumers),
checkout(Meta, State, Effects);
apply(Meta, {nodeup, Node}, #state{consumers = Cons0,
- enqueuers = Enqs0,
- service_queue = SQ0} = State0) ->
+ enqueuers = Enqs0,
+ service_queue = SQ0} = State0) ->
%% A node we are monitoring has come back.
%% If we have suspected any processes of being
%% down we should now re-issue the monitors for them to detect if they're
@@ -528,20 +529,20 @@ apply(Meta, {nodeup, Node}, #state{consumers = Cons0,
|| P <- suspected_pids_for(Node, State0)],
% un-suspect waiting consumers when necessary
- WaitingConsumers = maybe_mark_suspect_waiting_consumers(Node, State0,
- false),
+ WaitingConsumers = update_waiting_consumer_status(Node, State0, up),
Enqs1 = maps:map(fun(P, E) when node(P) =:= Node ->
- E#enqueuer{suspected_down = false};
+ E#enqueuer{status = up};
(_, E) -> E
end, Enqs0),
ConsumerUpdateActiveFun = consumer_active_flag_update_function(State0),
{Cons1, SQ, Effects} =
maps:fold(fun({_, P} = ConsumerId, C, {CAcc, SQAcc, EAcc})
- when node(P) =:= Node ->
+ when (node(P) =:= Node) and
+ (C#consumer.status =/= cancelled) ->
EAcc1 = ConsumerUpdateActiveFun(State0, ConsumerId, C, true, up, EAcc),
update_or_remove_sub(
- ConsumerId, C#consumer{suspected_down = false},
+ ConsumerId, C#consumer{status = up},
CAcc, SQAcc, EAcc1);
(_, _, Acc) ->
Acc
@@ -587,26 +588,27 @@ handle_waiting_consumer_down(Pid,
State = State0#state{waiting_consumers = StillUp},
{Effects, State}.
-maybe_mark_suspect_waiting_consumers(_Node, #state{consumer_strategy = default},
- _Suspected) ->
+update_waiting_consumer_status(_Node, #state{consumer_strategy = default},
+ _Status) ->
[];
-maybe_mark_suspect_waiting_consumers(_Node,
- #state{consumer_strategy = single_active,
- waiting_consumers = []},
- _Suspected) ->
+update_waiting_consumer_status(_Node,
+ #state{consumer_strategy = single_active,
+ waiting_consumers = []},
+ _Status) ->
[];
-maybe_mark_suspect_waiting_consumers(Node,
- #state{consumer_strategy = single_active,
- waiting_consumers = WaitingConsumers},
- Suspected) ->
+update_waiting_consumer_status(Node,
+ #state{consumer_strategy = single_active,
+ waiting_consumers = WaitingConsumers},
+ Status) ->
[begin
case node(P) of
Node ->
- {ConsumerId, Consumer#consumer{suspected_down = Suspected}};
+ {ConsumerId, Consumer#consumer{status = Status}};
_ ->
{ConsumerId, Consumer}
end
- end || {{_, P} = ConsumerId, Consumer} <- WaitingConsumers].
+ end || {{_, P} = ConsumerId, Consumer} <- WaitingConsumers,
+ Consumer#consumer.status =/= cancelled].
-spec state_enter(ra_server:ra_state(), state()) -> ra_machine:effects().
state_enter(leader, #state{consumers = Cons,
@@ -691,7 +693,7 @@ get_checked_out(Cid, From, To, #state{consumers = Consumers}) ->
end.
init_aux(Name) when is_atom(Name) ->
- %% TODO: catch specific exeption throw if table already exists
+ %% TODO: catch specific exception throw if table already exists
ok = ra_machine_ets:create_table(rabbit_fifo_usage,
[named_table, set, public,
{write_concurrency, true}]),
@@ -739,12 +741,13 @@ query_consumers(#state{consumers = Consumers,
consumer_strategy = ConsumerStrategy } = State) ->
ActiveActivityStatusFun = case ConsumerStrategy of
default ->
- fun(_ConsumerId, #consumer{suspected_down = SuspectedDown}) ->
- case SuspectedDown of
- true ->
- {false, suspected_down};
- false ->
- {true, up}
+ fun(_ConsumerId,
+ #consumer{status = Status}) ->
+ case Status of
+ suspected_down ->
+ {false, Status};
+ _ ->
+ {true, Status}
end
end;
single_active ->
@@ -758,18 +761,24 @@ query_consumers(#state{consumers = Consumers,
end
end
end,
- FromConsumers = maps:map(fun ({Tag, Pid}, #consumer{meta = Meta} = Consumer) ->
- {Active, ActivityStatus} = ActiveActivityStatusFun({Tag, Pid}, Consumer),
- {Pid, Tag,
- maps:get(ack, Meta, undefined),
- maps:get(prefetch, Meta, undefined),
- Active,
- ActivityStatus,
- maps:get(args, Meta, []),
- maps:get(username, Meta, undefined)}
- end, Consumers),
+ FromConsumers = maps:fold(fun (_, #consumer{status = cancelled}, Acc) ->
+ Acc;
+ ({Tag, Pid}, #consumer{meta = Meta} = Consumer, Acc) ->
+ {Active, ActivityStatus} = ActiveActivityStatusFun({Tag, Pid}, Consumer),
+ maps:put({Tag, Pid},
+ {Pid, Tag,
+ maps:get(ack, Meta, undefined),
+ maps:get(prefetch, Meta, undefined),
+ Active,
+ ActivityStatus,
+ maps:get(args, Meta, []),
+ maps:get(username, Meta, undefined)},
+ Acc)
+ end, #{}, Consumers),
FromWaitingConsumers =
- lists:foldl(fun({{Tag, Pid}, #consumer{meta = Meta} = Consumer}, Acc) ->
+ lists:foldl(fun ({_, #consumer{status = cancelled}}, Acc) ->
+ Acc;
+ ({{Tag, Pid}, #consumer{meta = Meta} = Consumer}, Acc) ->
{Active, ActivityStatus} = ActiveActivityStatusFun({Tag, Pid}, Consumer),
maps:put({Tag, Pid},
{Pid, Tag,
@@ -854,42 +863,42 @@ num_checked_out(#state{consumers = Cons}) ->
end, 0, maps:values(Cons)).
cancel_consumer(ConsumerId,
- #state{consumer_strategy = default} = State, Effects) ->
+ #state{consumer_strategy = default} = State, Effects, Reason) ->
%% general case, single active consumer off
- cancel_consumer0(ConsumerId, State, Effects);
+ cancel_consumer0(ConsumerId, State, Effects, Reason);
cancel_consumer(ConsumerId,
#state{consumer_strategy = single_active,
waiting_consumers = []} = State,
- Effects) ->
+ Effects, Reason) ->
%% single active consumer on, no consumers are waiting
- cancel_consumer0(ConsumerId, State, Effects);
+ cancel_consumer0(ConsumerId, State, Effects, Reason);
cancel_consumer(ConsumerId,
#state{consumers = Cons0,
consumer_strategy = single_active,
waiting_consumers = WaitingConsumers0} = State0,
- Effects0) ->
+ Effects0, Reason) ->
%% single active consumer on, consumers are waiting
case maps:take(ConsumerId, Cons0) of
- {#consumer{checked_out = Checked0}, _} ->
+ {Consumer, Cons1} ->
% The active consumer is to be removed
% Cancel it
- State1 = return_all(State0, Checked0),
- Effects1 = cancel_consumer_effects(ConsumerId, State1, Effects0),
+ {State1, Effects1} = maybe_return_all(ConsumerId, Consumer, Cons1, State0, Effects0, Reason),
+ Effects2 = cancel_consumer_effects(ConsumerId, State1, Effects1),
% Take another one from the waiting consumers and put it in consumers
[{NewActiveConsumerId, NewActiveConsumer}
| RemainingWaitingConsumers] = WaitingConsumers0,
- #state{service_queue = ServiceQueue} = State0,
+ #state{service_queue = ServiceQueue} = State1,
ServiceQueue1 = maybe_queue_consumer(NewActiveConsumerId,
NewActiveConsumer,
ServiceQueue),
- State = State1#state{consumers = #{NewActiveConsumerId =>
- NewActiveConsumer},
+ State = State1#state{consumers = maps:put(NewActiveConsumerId,
+ NewActiveConsumer, State1#state.consumers),
service_queue = ServiceQueue1,
waiting_consumers = RemainingWaitingConsumers},
- Effects2 = consumer_update_active_effects(State, NewActiveConsumerId,
+ Effects = consumer_update_active_effects(State, NewActiveConsumerId,
NewActiveConsumer, true,
- single_active, Effects1),
- {State, Effects2};
+ single_active, Effects2),
+ {State, Effects};
error ->
% The cancelled consumer is not the active one
% Just remove it from idle_consumers
@@ -914,23 +923,39 @@ consumer_update_active_effects(#state{queue_resource = QName },
[QName, ConsumerId, false, Ack, Prefetch, Active, ActivityStatus, Args]}
| Effects].
-cancel_consumer0(ConsumerId,
- #state{consumers = C0} = S0, Effects0) ->
+cancel_consumer0(ConsumerId, #state{consumers = C0} = S0, Effects0, Reason) ->
case maps:take(ConsumerId, C0) of
- {#consumer{checked_out = Checked0}, Cons} ->
- S = return_all(S0, Checked0),
- Effects = cancel_consumer_effects(ConsumerId, S, Effects0),
- case maps:size(Cons) of
+ {Consumer, Cons1} ->
+ {S, Effects2} = maybe_return_all(ConsumerId, Consumer, Cons1, S0, Effects0, Reason),
+ Effects = cancel_consumer_effects(ConsumerId, S, Effects2),
+ case maps:size(S#state.consumers) of
0 ->
- {S#state{consumers = Cons}, [{aux, inactive} | Effects]};
+ {S, [{aux, inactive} | Effects]};
_ ->
- {S#state{consumers = Cons}, Effects}
+ {S, Effects}
end;
error ->
%% already removed: do nothing
{S0, Effects0}
end.
+maybe_return_all(ConsumerId, #consumer{checked_out = Checked0} = Consumer, Cons1,
+ #state{consumers = C0,
+ service_queue = SQ0} = S0, Effects0, Reason) ->
+ case Reason of
+ consumer_cancel ->
+ {Cons, SQ, Effects1} =
+ update_or_remove_sub(ConsumerId,
+ Consumer#consumer{lifetime = once,
+ credit = 0,
+ status = cancelled},
+ C0, SQ0, Effects0),
+ {S0#state{consumers = Cons, service_queue = SQ}, Effects1};
+ down ->
+ S1 = return_all(S0, Checked0),
+ {S1#state{consumers = Cons1}, Effects0}
+ end.
+
apply_enqueue(#{index := RaftIdx} = Meta, From, Seq, RawMsg, State0) ->
Bytes = message_size(RawMsg),
case maybe_enqueue(RaftIdx, From, Seq, RawMsg, [], State0) of
@@ -1083,7 +1108,7 @@ increase_credit(#consumer{lifetime = once,
increase_credit(#consumer{lifetime = auto,
credit_mode = credited,
credit = Credit}, _) ->
- %% credit_mode: credit also doens't automatically increment credit
+ %% credit_mode: credit also doesn't automatically increment credit
Credit;
increase_credit(#consumer{credit = Current}, Credit) ->
Current + Credit.
@@ -1303,7 +1328,9 @@ checkout_one(#state{service_queue = SQ0,
%% can happen when draining
%% recurse without consumer on queue
checkout_one(InitState#state{service_queue = SQ1});
- {ok, #consumer{suspected_down = true}} ->
+ {ok, #consumer{status = cancelled}} ->
+ checkout_one(InitState#state{service_queue = SQ1});
+ {ok, #consumer{status = suspected_down}} ->
checkout_one(InitState#state{service_queue = SQ1});
{ok, #consumer{checked_out = Checked0,
next_msg_id = Next,
@@ -1358,8 +1385,9 @@ update_or_remove_sub(ConsumerId, #consumer{lifetime = once,
case maps:size(Checked) of
0 ->
% we're done with this consumer
- {maps:remove(ConsumerId, Cons), ServiceQueue,
- [{demonitor, process, ConsumerId} | Effects]};
+ % TODO: demonitor consumer pid but _only_ if there are no other
+ % monitors for this pid
+ {maps:remove(ConsumerId, Cons), ServiceQueue, Effects};
_ ->
% there are unsettled items so need to keep around
{maps:put(ConsumerId, Con, Cons), ServiceQueue, Effects}
@@ -1379,7 +1407,6 @@ uniq_queue_in(Key, Queue) ->
queue:in(Key, Queue)
end.
-
update_consumer(ConsumerId, Meta, Spec,
#state{consumer_strategy = default} = State0) ->
%% general case, single active consumer off
@@ -1553,18 +1580,18 @@ message_size(Msg) ->
suspected_pids_for(Node, #state{consumers = Cons0,
enqueuers = Enqs0,
waiting_consumers = WaitingConsumers0}) ->
- Cons = maps:fold(fun({_, P}, #consumer{suspected_down = true}, Acc)
+ Cons = maps:fold(fun({_, P}, #consumer{status = suspected_down}, Acc)
when node(P) =:= Node ->
[P | Acc];
(_, _, Acc) -> Acc
end, [], Cons0),
- Enqs = maps:fold(fun(P, #enqueuer{suspected_down = true}, Acc)
+ Enqs = maps:fold(fun(P, #enqueuer{status = suspected_down}, Acc)
when node(P) =:= Node ->
[P | Acc];
(_, _, Acc) -> Acc
end, Cons, Enqs0),
lists:foldl(fun({{_, P},
- #consumer{suspected_down = true}}, Acc)
+ #consumer{status = suspected_down}}, Acc)
when node(P) =:= Node ->
[P | Acc];
(_, Acc) -> Acc
@@ -1814,10 +1841,11 @@ return_checked_out_test() ->
{State0, [_, _]} = enq(1, 1, first, test_init(test)),
{State1, [_Monitor,
{send_msg, _, {delivery, _, [{MsgId, _}]}, ra_event},
- {aux, active} | _
- ]} = check(Cid, 2, State0),
- % return
- {_State2, _, [_]} = apply(meta(3), make_return(Cid, [MsgId]), State1),
+ {aux, active} | _ ]} = check_auto(Cid, 2, State0),
+ % returning immediately checks out the same message again
+ {_, ok, [{send_msg, _, {delivery, _, [{_, _}]}, ra_event},
+ {aux, active}]} =
+ apply(meta(3), make_return(Cid, [MsgId]), State1),
ok.
return_auto_checked_out_test() ->
@@ -1844,15 +1872,19 @@ cancelled_checkout_out_test() ->
{State00, [_, _]} = enq(1, 1, first, test_init(test)),
{State0, [_]} = enq(2, 2, second, State00),
{State1, _} = check_auto(Cid, 2, State0),
- % cancelled checkout should return all pending messages to queue
+ % cancelled checkout should not return pending messages to queue
{State2, _, _} = apply(meta(3), make_checkout(Cid, cancel, #{}), State1),
?assertEqual(1, maps:size(State2#state.messages)),
- ?assertEqual(1, lqueue:len(State2#state.returns)),
+ ?assertEqual(0, lqueue:len(State2#state.returns)),
- {State3, {dequeue, {0, {_, first}}, _}, _} =
+ {State3, {dequeue, empty}} =
apply(meta(3), make_checkout(Cid, {dequeue, settled}, #{}), State2),
+ %% settle
+ {State4, ok, _} =
+ apply(meta(4), make_settle(Cid, [0]), State3),
+
{_State, {dequeue, {_, {_, second}}, _}, _} =
- apply(meta(4), make_checkout(Cid, {dequeue, settled}, #{}), State3),
+ apply(meta(5), make_checkout(Cid, {dequeue, settled}, #{}), State4),
ok.
down_with_noproc_consumer_returns_unsettled_test() ->
@@ -1914,16 +1946,6 @@ down_with_noproc_enqueuer_is_cleaned_up_test() ->
?assert(0 =:= maps:size(State1#state.enqueuers)),
ok.
-completed_consumer_yields_demonitor_effect_test() ->
- Cid = {<<"completed_consumer_yields_demonitor_effect_test">>, self()},
- {State0, [_, _]} = enq(1, 1, second, test_init(test)),
- {State1, [{monitor, process, _} | _]} = check(Cid, 2, State0),
- {_, Effects} = settle(Cid, 3, 0, State1),
- ?ASSERT_EFF({demonitor, _, _}, Effects),
- % release cursor for empty queue
- ?ASSERT_EFF({release_cursor, 3, _}, Effects),
- ok.
-
discarded_message_without_dead_letter_handler_is_removed_test() ->
Cid = {<<"completed_consumer_yields_demonitor_effect_test">>, self()},
{State0, [_, _]} = enq(1, 1, first, test_init(test)),
@@ -2117,7 +2139,7 @@ enq_check_settle_duplicate_test() ->
run_snapshot_test(?FUNCTION_NAME, Commands).
run_snapshot_test(Name, Commands) ->
- %% create every incremental permuation of the commands lists
+ %% create every incremental permutation of the commands lists
%% and run the snapshot tests against that
[begin
run_snapshot_test0(Name, C)
@@ -2186,7 +2208,7 @@ state_enter_test() ->
[{mod_call, m, f, [a, the_name]}] = state_enter(leader, S0),
ok.
-state_enter_montors_and_notifications_test() ->
+state_enter_monitors_and_notifications_test() ->
Oth = spawn(fun () -> ok end),
{State0, _} = enq(1, 1, first, test_init(test)),
Cid = {<<"adf">>, self()},
@@ -2299,7 +2321,10 @@ single_active_consumer_test() ->
?assertEqual(1, length(Effects1)),
% cancelling the active consumer
- {State3, _, Effects2} = apply(meta(3), #checkout{spec = cancel, consumer_id = {<<"ctag1">>, self()}}, State2),
+ {State3, _, Effects2} = apply(meta(3),
+ #checkout{spec = cancel,
+ consumer_id = {<<"ctag1">>, self()}},
+ State2),
% the second registered consumer is now the active one
?assertEqual(1, map_size(State3#state.consumers)),
?assert(maps:is_key({<<"ctag2">>, self()}, State3#state.consumers)),
@@ -2402,15 +2427,16 @@ single_active_consumer_mark_waiting_consumers_as_suspected_when_down_noconnnecti
State),
NewState
end,
- State1 = lists:foldl(AddConsumer, State0, [<<"ctag1">>, <<"ctag2">>, <<"ctag3">>, <<"ctag4">>]),
+ State1 = lists:foldl(AddConsumer, State0,
+ [<<"ctag1">>, <<"ctag2">>, <<"ctag3">>, <<"ctag4">>]),
% simulate node goes down
{State2, _, _} = apply(#{}, {down, self(), noconnection}, State1),
% all the waiting consumers should be suspected down
?assertEqual(3, length(State2#state.waiting_consumers)),
- lists:foreach(fun({_, #consumer{suspected_down = SuspectedDown}}) ->
- ?assert(SuspectedDown)
+ lists:foreach(fun({_, #consumer{status = Status}}) ->
+ ?assert(Status == suspected_down)
end, State2#state.waiting_consumers),
% simulate node goes back up
@@ -2418,8 +2444,8 @@ single_active_consumer_mark_waiting_consumers_as_suspected_when_down_noconnnecti
% all the waiting consumers should be un-suspected
?assertEqual(3, length(State3#state.waiting_consumers)),
- lists:foreach(fun({_, #consumer{suspected_down = SuspectedDown}}) ->
- ?assertNot(SuspectedDown)
+ lists:foreach(fun({_, #consumer{status = Status}}) ->
+ ?assert(Status /= suspected_down)
end, State3#state.waiting_consumers),
ok.
@@ -2504,7 +2530,8 @@ query_consumers_test() ->
State1 = lists:foldl(AddConsumer, State0, [<<"ctag1">>, <<"ctag2">>, <<"ctag3">>, <<"ctag4">>]),
Consumers0 = State1#state.consumers,
Consumer = maps:get({<<"ctag2">>, self()}, Consumers0),
- Consumers1 = maps:put({<<"ctag2">>, self()}, Consumer#consumer{suspected_down = true}, Consumers0),
+ Consumers1 = maps:put({<<"ctag2">>, self()},
+ Consumer#consumer{status = suspected_down}, Consumers0),
State2 = State1#state{consumers = Consumers1},
?assertEqual(4, query_consumer_count(State2)),
diff --git a/src/rabbit_fifo_client.erl b/src/rabbit_fifo_client.erl
index 5d9410918d..04918c3eb9 100644
--- a/src/rabbit_fifo_client.erl
+++ b/src/rabbit_fifo_client.erl
@@ -165,7 +165,7 @@ enqueue(Msg, State) ->
%% @doc Dequeue a message from the queue.
%%
-%% This is a syncronous call. I.e. the call will block until the command
+%% This is a synchronous call. I.e. the call will block until the command
%% has been accepted by the ra process or it times out.
%%
%% @param ConsumerTag a unique tag to identify this particular consumer.
@@ -297,7 +297,7 @@ discard(ConsumerTag, [_|_] = MsgIds,
%% @doc Register with the rabbit_fifo queue to "checkout" messages as they
%% become available.
%%
-%% This is a syncronous call. I.e. the call will block until the command
+%% This is a synchronous call. I.e. the call will block until the command
%% has been accepted by the ra process or it times out.
%%
%% @param ConsumerTag a unique tag to identify this particular consumer.
@@ -316,7 +316,7 @@ checkout(ConsumerTag, NumUnsettled, ConsumerInfo, State0) ->
%% @doc Register with the rabbit_fifo queue to "checkout" messages as they
%% become available.
%%
-%% This is a syncronous call. I.e. the call will block until the command
+%% This is a synchronous call. I.e. the call will block until the command
%% has been accepted by the ra process or it times out.
%%
%% @param ConsumerTag a unique tag to identify this particular consumer.
@@ -374,7 +374,7 @@ credit(ConsumerTag, Credit, Drain,
%% @doc Cancels a checkout with the rabbit_fifo queue for the consumer tag
%%
-%% This is a syncronous call. I.e. the call will block until the command
+%% This is a synchronous call. I.e. the call will block until the command
%% has been accepted by the ra process or it times out.
%%
%% @param ConsumerTag a unique tag to identify this particular consumer.
diff --git a/src/rabbit_fifo_index.erl b/src/rabbit_fifo_index.erl
index 82a75b4adc..3bda9bab26 100644
--- a/src/rabbit_fifo_index.erl
+++ b/src/rabbit_fifo_index.erl
@@ -87,7 +87,7 @@ find_next(Next, Last, Map) ->
_ ->
% in degenerate cases the range here could be very large
% and hence this could be very slow
- % the typical case should idealy be better
+ % the typical case should ideally be better
% assuming fifo-ish deletion of entries
find_next(Next+1, Last, Map)
end.
diff --git a/src/rabbit_guid.erl b/src/rabbit_guid.erl
index 6f03a1a04f..5728ac88d7 100644
--- a/src/rabbit_guid.erl
+++ b/src/rabbit_guid.erl
@@ -92,7 +92,7 @@ advance_blocks({B1, B2, B3, B4}, I) ->
%% hashing {B5, I}. The new hash is used as last block, and the
%% other three blocks are XORed with it.
%%
- %% Doing this is convenient because it avoids cascading conflits,
+ %% Doing this is convenient because it avoids cascading conflicts,
%% while being very fast. The conflicts are avoided by propagating
%% the changes through all the blocks at each round by XORing, so
%% the only occasion in which a collision will take place is when
diff --git a/src/rabbit_lager.erl b/src/rabbit_lager.erl
index 1a72d15009..bfdd26fee3 100644
--- a/src/rabbit_lager.erl
+++ b/src/rabbit_lager.erl
@@ -396,7 +396,7 @@ prepare_rabbit_log_config() ->
set_env_default_log_console();
FileName when is_list(FileName) ->
case os:getenv("RABBITMQ_LOGS_source") of
- %% The user explicitely sets $RABBITMQ_LOGS;
+ %% The user explicitly sets $RABBITMQ_LOGS;
%% we should override a file location even
%% if it's set in rabbitmq.config
"environment" -> set_env_default_log_file(FileName, override);
@@ -406,7 +406,7 @@ prepare_rabbit_log_config() ->
%% Upgrade log file never overrides the value set in rabbitmq.config
case UpgradeFile of
- %% No special env for upgrade logs - rederect to the default sink
+ %% No special env for upgrade logs - redirect to the default sink
undefined -> ok;
%% Redirect logs to default output.
DefaultFile -> ok;
diff --git a/src/rabbit_mirror_queue_coordinator.erl b/src/rabbit_mirror_queue_coordinator.erl
index 96474b0d4e..b502da201a 100644
--- a/src/rabbit_mirror_queue_coordinator.erl
+++ b/src/rabbit_mirror_queue_coordinator.erl
@@ -364,7 +364,7 @@ handle_cast({gm_deaths, DeadGMPids}, State = #state{q = Q}) when ?amqqueue_pid_r
%% Different slave is now master, stop current coordinator normally.
%% Initiating queue is now slave and the least we could do is report
%% deaths which we 'think' we saw.
- %% NOTE: Reported deaths here, could be inconsistant.
+ %% NOTE: Reported deaths here, could be inconsistent.
rabbit_mirror_queue_misc:report_deaths(MPid, false, QueueName,
DeadPids),
{stop, shutdown, State};
diff --git a/src/rabbit_mirror_queue_misc.erl b/src/rabbit_mirror_queue_misc.erl
index 04ed4e2887..7383152898 100644
--- a/src/rabbit_mirror_queue_misc.erl
+++ b/src/rabbit_mirror_queue_misc.erl
@@ -615,7 +615,7 @@ validate_sync_batch_size(none) ->
validate_sync_batch_size(N) when is_integer(N) andalso N > 0 ->
ok;
validate_sync_batch_size(N) ->
- {error, "ha-sync-batch-size takes an integer greather than 0, "
+ {error, "ha-sync-batch-size takes an integer greater than 0, "
"~p given", [N]}.
validate_pos(PromoteOnShutdown) ->
diff --git a/src/rabbit_msg_store.erl b/src/rabbit_msg_store.erl
index 337064ad39..a4dfba47e7 100644
--- a/src/rabbit_msg_store.erl
+++ b/src/rabbit_msg_store.erl
@@ -236,7 +236,7 @@
%% updated.
%%
%% On non-clean startup, we scan the files we discover, dealing with
-%% the possibilites of a crash having occurred during a compaction
+%% the possibilities of a crash having occurred during a compaction
%% (this consists of tidyup - the compaction is deliberately designed
%% such that data is duplicated on disk rather than risking it being
%% lost), and rebuild the file summary and index ETS table.
@@ -596,7 +596,7 @@ client_read2(false, undefined, _MsgLocation, Defer, _CState) ->
Defer();
client_read2(true, _Right, _MsgLocation, Defer, _CState) ->
%% Of course, in the mean time, the GC could have run and our msg
- %% is actually in a different file, unlocked. However, defering is
+ %% is actually in a different file, unlocked. However, deferring is
%% the safest and simplest thing to do.
Defer();
client_read2(false, _Right,
diff --git a/src/rabbit_node_monitor.erl b/src/rabbit_node_monitor.erl
index cd46ade0e2..9fa1dfa462 100644
--- a/src/rabbit_node_monitor.erl
+++ b/src/rabbit_node_monitor.erl
@@ -219,7 +219,7 @@ subscribe(Pid) ->
%% We could confirm something by having an HA queue see the pausing
%% state (and fail over into it) before the node monitor stops us, or
%% by using unmirrored queues and just having them vanish (and
-%% confiming messages as thrown away).
+%% confirming messages as thrown away).
%%
%% So we have channels call in here before issuing confirms, to do a
%% lightweight check that we have not entered a pausing state.
diff --git a/src/rabbit_priority_queue.erl b/src/rabbit_priority_queue.erl
index 621f42dafb..12d3291b54 100644
--- a/src/rabbit_priority_queue.erl
+++ b/src/rabbit_priority_queue.erl
@@ -22,7 +22,7 @@
-behaviour(rabbit_backing_queue).
-%% enabled unconditionally. Disabling priority queueing after
+%% enabled unconditionally. Disabling priority queuing after
%% it has been enabled is dangerous.
-rabbit_boot_step({?MODULE,
[{description, "enable priority queue"},
@@ -55,7 +55,7 @@
-define(passthrough3(F),
{Res1, Res2, BQS1} = BQ:F, {Res1, Res2, State#passthrough{bqs = BQS1}}).
-%% This module adds suport for priority queues.
+%% This module adds support for priority queues.
%%
%% Priority queues have one backing queue per priority. Backing queue functions
%% then produce a list of results for each BQ and fold over them, sorting
diff --git a/src/rabbit_queue_index.erl b/src/rabbit_queue_index.erl
index e4047a9902..61373e49c1 100644
--- a/src/rabbit_queue_index.erl
+++ b/src/rabbit_queue_index.erl
@@ -355,7 +355,7 @@ pre_publish(MsgOrId, SeqId, MsgProps, IsPersistent, IsDelivered, JournalSizeHint
State1#qistate{pre_publish_cache = PPC1,
delivered_cache = DC1})).
-%% pre_publish_cache is the entry with most elements when comapred to
+%% pre_publish_cache is the entry with most elements when compared to
%% delivered_cache so we only check the former in the guard.
maybe_flush_pre_publish_cache(JournalSizeHint,
#qistate{pre_publish_cache = PPC} = State)
diff --git a/src/rabbit_reader.erl b/src/rabbit_reader.erl
index 3a2d4650ca..c0cb9c57d5 100644
--- a/src/rabbit_reader.erl
+++ b/src/rabbit_reader.erl
@@ -240,7 +240,7 @@ server_properties(Protocol) ->
{ok, RawConfigServerProps} = application:get_env(rabbit,
server_properties),
- %% Normalize the simplifed (2-tuple) and unsimplified (3-tuple) forms
+ %% Normalize the simplified (2-tuple) and unsimplified (3-tuple) forms
%% from the config and merge them with the generated built-in properties
NormalizedConfigServerProps =
[{<<"capabilities">>, table, server_capabilities(Protocol)} |
@@ -852,7 +852,7 @@ handle_exception(State = #v1{connection = #connection{protocol = Protocol},
respond_and_close(State, Channel, Protocol, Reason,
{handshake_error, CS, Reason});
%% when negotiation fails, e.g. due to channel_max being higher than the
-%% maxiumum allowed limit
+%% maximum allowed limit
handle_exception(State = #v1{connection = #connection{protocol = Protocol,
log_name = ConnName,
user = User},
diff --git a/src/rabbit_upgrade.erl b/src/rabbit_upgrade.erl
index f452d5c92f..83ec692ff3 100644
--- a/src/rabbit_upgrade.erl
+++ b/src/rabbit_upgrade.erl
@@ -31,7 +31,7 @@
%% clusters.
%%
%% Firstly, we have two different types of upgrades to do: Mnesia and
-%% everythinq else. Mnesia upgrades must only be done by one node in
+%% everything else. Mnesia upgrades must only be done by one node in
%% the cluster (we treat a non-clustered node as a single-node
%% cluster). This is the primary upgrader. The other upgrades need to
%% be done by all nodes.
@@ -314,7 +314,7 @@ node_type_legacy() ->
%% hang), we can't look at the config file (may not include us
%% even if we're a disc node). We also can't use
%% rabbit_mnesia:node_type/0 because that will give false
- %% postivies on Rabbit up to 2.5.1.
+ %% positives on Rabbit up to 2.5.1.
case filelib:is_regular(filename:join(dir(), "rabbit_durable_exchange.DCD")) of
true -> disc;
false -> ram
diff --git a/src/rabbit_variable_queue.erl b/src/rabbit_variable_queue.erl
index 8b773f2cc2..e2b4f64a8f 100644
--- a/src/rabbit_variable_queue.erl
+++ b/src/rabbit_variable_queue.erl
@@ -1735,7 +1735,7 @@ purge_and_index_reset(State) ->
%%
%% purge_betas_and_deltas/2 loads messages from the queue index,
%% filling up q3 and in some cases moving messages form q2 to q3 while
-%% reseting q2 to an empty queue (see maybe_deltas_to_betas/2). The
+%% resetting q2 to an empty queue (see maybe_deltas_to_betas/2). The
%% messages loaded into q3 are removed by calling
%% remove_queue_entries/3 until there are no more messages to be read
%% from the queue index. Messages are read in batches from the queue
@@ -1939,7 +1939,7 @@ maybe_write_msg_to_disk(Force, MsgStatus = #msg_status {
maybe_write_msg_to_disk(_Force, MsgStatus, State) ->
{MsgStatus, State}.
-%% Due to certain optimizations made inside
+%% Due to certain optimisations made inside
%% rabbit_queue_index:pre_publish/7 we need to have two separate
%% functions for index persistence. This one is only used when paging
%% during memory pressure. We didn't want to modify
View Lorry Controller Status