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

Merge pull request #12903 from ieQu1/dev/ds-egress-flush-condition 

fix(ds): Fix egress flush condition
This commit is contained in:
ieQu1 2024-04-22 10:17:22 +02:00 committed by GitHub
parent 170809362e 4c76a2574d
commit f9eda1883f
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 12 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -127,7 +127,7 @@ init([DB, Shard]) ->
metrics_id = MetricsId, metrics_id = MetricsId,
queue = queue:new() queue = queue:new()
}, },
{ok, start_timer(S)}. {ok, S}.
handle_call( handle_call(
#enqueue_req{ #enqueue_req{
@ -195,7 +195,6 @@ enqueue(
true -> true ->
%% Adding this batch would cause buffer to overflow. Flush %% Adding this batch would cause buffer to overflow. Flush
%% it now, and retry: %% it now, and retry:
cancel_timer(S0),
S1 = flush(S0), S1 = flush(S0),
enqueue(Sync, Atomic, Msgs, BatchSize, BatchBytes, S1); enqueue(Sync, Atomic, Msgs, BatchSize, BatchBytes, S1);
false -> false ->
@ -203,12 +202,11 @@ enqueue(
%% entirety: %% entirety:
Q1 = lists:foldl(fun queue:in/2, Q0, Msgs), Q1 = lists:foldl(fun queue:in/2, Q0, Msgs),
S1 = S0#s{n = NMsgs, n_bytes = NBytes, queue = Q1}, S1 = S0#s{n = NMsgs, n_bytes = NBytes, queue = Q1},
case NMsgs >= NMax orelse NBytes >= NBytes of case NMsgs >= NMax orelse NBytes >= NBytesMax of
true -> true ->
cancel_timer(S1),
flush(S1); flush(S1);
false -> false ->
S1 ensure_timer(S1)
end end
end. end.
@ -216,7 +214,7 @@ enqueue(
-define(COOLDOWN_MAX, 5000). -define(COOLDOWN_MAX, 5000).
flush(S) -> flush(S) ->
start_timer(do_flush(S)). do_flush(cancel_timer(S)).
do_flush(S0 = #s{n = 0}) -> do_flush(S0 = #s{n = 0}) ->
S0; S0;
@ -372,16 +370,18 @@ compose_errors({error, recoverable, _}, {error, unrecoverable, Err}) ->
compose_errors(ErrAcc, _Err) -> compose_errors(ErrAcc, _Err) ->
ErrAcc. ErrAcc.
start_timer(S) -> ensure_timer(S = #s{tref = undefined}) ->
Interval = application:get_env(emqx_durable_storage, egress_flush_interval, 100), Interval = application:get_env(emqx_durable_storage, egress_flush_interval, 100),
Tref = erlang:send_after(Interval, self(), ?flush), Tref = erlang:send_after(Interval, self(), ?flush),
S#s{tref = Tref}. S#s{tref = Tref};
ensure_timer(S) ->
S.
cancel_timer(#s{tref = undefined}) -> cancel_timer(S = #s{tref = undefined}) ->
ok; S;
cancel_timer(#s{tref = TRef}) -> cancel_timer(S = #s{tref = TRef}) ->
_ = erlang:cancel_timer(TRef), _ = erlang:cancel_timer(TRef),
ok. S#s{tref = undefined}.
%% @doc Return approximate size of the MQTT message (it doesn't take %% @doc Return approximate size of the MQTT message (it doesn't take
%% all things into account, for example headers and extras) %% all things into account, for example headers and extras)