yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(ds): add caller to pending replies before flushing

This commit is contained in:
Thales Macedo Garitezi 2024-03-21 14:39:21 -03:00
parent e2a2295c99
commit 70737a437a
1 changed files with 10 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
apps/emqx_durable_storage/src/emqx_ds_replication_layer_egress.erl
View File

@ -193,14 +193,6 @@ do_enqueue(From, Sync, MsgOrBatch, S0 = #s{n = N, batch = Batch, pending_replies
Msg -> Msg ->
S0#s{n = N + 1, batch = [Msg | Batch]} S0#s{n = N + 1, batch = [Msg | Batch]}
end, end,
S2 =
case N >= NMax of
true ->
_ = erlang:cancel_timer(S0#s.tref),
do_flush(S1);
false ->
S1
end,
%% TODO: later we may want to delay the reply until the message is %% TODO: later we may want to delay the reply until the message is
%% replicated, but it requies changes to the PUBACK/PUBREC flow to %% replicated, but it requies changes to the PUBACK/PUBREC flow to
%% allow for async replies. For now, we ack when the message is %% allow for async replies. For now, we ack when the message is
@ -208,12 +200,20 @@ do_enqueue(From, Sync, MsgOrBatch, S0 = #s{n = N, batch = Batch, pending_replies
%% %%
%% Otherwise, the client would freeze for at least flush interval, %% Otherwise, the client would freeze for at least flush interval,
%% or until the buffer is filled. %% or until the buffer is filled.
S = S2 =
case Sync of case Sync of
true -> true ->
S2#s{pending_replies = [From | Replies]}; S1#s{pending_replies = [From | Replies]};
false -> false ->
gen_server:reply(From, ok), gen_server:reply(From, ok),
S1
end,
S =
case N >= NMax of
true ->
_ = erlang:cancel_timer(S2#s.tref),
do_flush(S2);
false ->
S2 S2
end, end,
%% TODO: add a backpressure mechanism for the server to avoid %% TODO: add a backpressure mechanism for the server to avoid