Skip to content

Take reject into account when sending confirms and vice-versa. #1719

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 4 commits into from
Oct 3, 2018
Merged

Take reject into account when sending confirms and vice-versa. #1719

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
8b37501
Take reject into account when sending confirms and vice-versa.
hairyhum Oct 1, 2018
25801d7
Naming; use erlang:min/2 instead of lists:min/1
michaelklishin Oct 2, 2018
4e2dc48
Fix min function to lists:min instead of erlang:min.
hairyhum Oct 2, 2018
08075cf
removed TODO
hairyhum Oct 3, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
86 changes: 48 additions & 38 deletions src/rabbit_channel.erl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1579,9 +1579,9 @@ handle_publishing_queue_down(QPid, Reason, State = #ch{unconfirmed = UC,
State1 = State#ch{mandatory = Mand1},
case rabbit_misc:is_abnormal_exit(Reason) of
true -> {MXs, UC1} = dtree:take_all(QPid, UC),
send_nacks(MXs, State1#ch{unconfirmed = UC1});
record_rejects(MXs, State1#ch{unconfirmed = UC1});
false -> {MXs, UC1} = dtree:take(QPid, UC),
record_confirms(MXs, State1#ch{unconfirmed = UC1})
record_confirms(MXs, State1#ch{unconfirmed = UC1})

end.

Expand Down Expand Up @@ -1911,37 +1911,29 @@ process_routing_confirm(true, QPids, MsgSeqNo, XName, State) ->
State#ch{unconfirmed = dtree:insert(MsgSeqNo, QPids, XName,
State#ch.unconfirmed)}.

send_nacks([], State) ->
State;
send_nacks(_MXs, State = #ch{state = closing,
tx = none}) -> %% optimisation
State;
send_nacks(MXs, State = #ch{tx = none}) ->
coalesce_and_send([MsgSeqNo || {MsgSeqNo, _} <- MXs],
fun(MsgSeqNo, Multiple) ->
#'basic.nack'{delivery_tag = MsgSeqNo,
multiple = Multiple}
end, State);
send_nacks(_MXs, State = #ch{state = closing}) -> %% optimisation
State#ch{tx = failed};
send_nacks(_, State) ->
maybe_complete_tx(State#ch{tx = failed}).

send_confirms_and_nacks(State = #ch{tx = none, confirmed = [], rejected = []}) ->
State;
send_confirms_and_nacks(State = #ch{tx = none, confirmed = C, rejected = R}) ->
case rabbit_node_monitor:pause_partition_guard() of
ok -> ConfirmMsgSeqNos =
lists:foldl(
fun ({MsgSeqNo, XName}, MSNs) ->
?INCR_STATS(exchange_stats, XName, 1,
confirm, State),
[MsgSeqNo | MSNs]
end, [], lists:append(C)),
State1 = send_confirms(ConfirmMsgSeqNos, State#ch{confirmed = []}),
%% TODO: msg seq nos, same as for confirms. Need to implement
%% nack rates first.
send_nacks(lists:append(R), State1#ch{rejected = []});
ok ->
Confirms = lists:append(C),
Rejects = lists:append(R),
ConfirmMsgSeqNos =
lists:foldl(
fun ({MsgSeqNo, XName}, MSNs) ->
?INCR_STATS(exchange_stats, XName, 1, confirm, State),
[MsgSeqNo | MSNs]
end, [], Confirms),
RejectMsgSeqNos = [MsgSeqNo || {MsgSeqNo, _} <- Rejects],

State1 = send_confirms(ConfirmMsgSeqNos,
RejectMsgSeqNos,
State#ch{confirmed = []}),
%% TODO: msg seq nos, same as for confirms. Need to implement
%% nack rates first.
send_nacks(RejectMsgSeqNos,
ConfirmMsgSeqNos,
State1#ch{rejected = []});
pausing -> State
end;
send_confirms_and_nacks(State) ->
Expand All @@ -1950,26 +1942,44 @@ send_confirms_and_nacks(State) ->
pausing -> State
end.

send_confirms([], State) ->
send_nacks([], _, State) ->
State;
send_confirms(_Cs, State = #ch{state = closing}) -> %% optimisation
send_nacks(_Rs, _, State = #ch{state = closing,
tx = none}) -> %% optimisation
State;
send_confirms([MsgSeqNo], State) ->
send_nacks(Rs, Cs, State = #ch{tx = none}) ->
coalesce_and_send(Rs, Cs,
fun(MsgSeqNo, Multiple) ->
#'basic.nack'{delivery_tag = MsgSeqNo,
multiple = Multiple}
end, State);
send_nacks(_MXs, _, State = #ch{state = closing}) -> %% optimisation
State#ch{tx = failed};
send_nacks(_, _, State) ->
maybe_complete_tx(State#ch{tx = failed}).

send_confirms([], _, State) ->
State;
send_confirms(_Cs, _, State = #ch{state = closing}) -> %% optimisation
State;
send_confirms([MsgSeqNo], _, State) ->
ok = send(#'basic.ack'{delivery_tag = MsgSeqNo}, State),
State;
send_confirms(Cs, State) ->
coalesce_and_send(Cs, fun(MsgSeqNo, Multiple) ->
send_confirms(Cs, Rs, State) ->
coalesce_and_send(Cs, Rs,
fun(MsgSeqNo, Multiple) ->
#'basic.ack'{delivery_tag = MsgSeqNo,
multiple = Multiple}
end, State).
end, State).

coalesce_and_send(MsgSeqNos, MkMsgFun, State = #ch{unconfirmed = UC}) ->
coalesce_and_send(MsgSeqNos, NegativeMsgSeqNos, MkMsgFun, State = #ch{unconfirmed = UC}) ->
SMsgSeqNos = lists:usort(MsgSeqNos),
CutOff = case dtree:is_empty(UC) of
UnconfirmedCutoff = case dtree:is_empty(UC) of
true -> lists:last(SMsgSeqNos) + 1;
false -> {SeqNo, _XName} = dtree:smallest(UC), SeqNo
end,
{Ms, Ss} = lists:splitwith(fun(X) -> X < CutOff end, SMsgSeqNos),
Cutoff = lists:min([UnconfirmedCutoff | NegativeMsgSeqNos]),
{Ms, Ss} = lists:splitwith(fun(X) -> X < Cutoff end, SMsgSeqNos),
case Ms of
[] -> ok;
_ -> ok = send(MkMsgFun(lists:last(Ms), true), State)
Expand Down