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

chore: cap replayq seg size under total size

This commit is contained in:
Zaiming (Stone) Shi 2022-12-19 23:16:05 +01:00
parent 7aee1a08aa
commit f611cbab45
1 changed files with 5 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
apps/emqx_resource/src/emqx_resource_worker.erl
View File

@ -117,13 +117,16 @@ init({Id, Index, Opts}) ->
true = gproc_pool:connect_worker(Id, {Id, Index}), true = gproc_pool:connect_worker(Id, {Id, Index}),
Name = name(Id, Index), Name = name(Id, Index),
BatchSize = maps:get(batch_size, Opts, ?DEFAULT_BATCH_SIZE), BatchSize = maps:get(batch_size, Opts, ?DEFAULT_BATCH_SIZE),
SegBytes0 = maps:get(queue_seg_bytes, Opts, ?DEFAULT_QUEUE_SEG_SIZE),
TotalBytes = maps:get(max_queue_bytes, Opts, ?DEFAULT_QUEUE_SIZE),
SegBytes = min(SegBytes0, TotalBytes),
Queue = Queue =
case maps:get(enable_queue, Opts, false) of case maps:get(enable_queue, Opts, false) of
true -> true ->
replayq:open(#{ replayq:open(#{
dir => disk_queue_dir(Id, Index), dir => disk_queue_dir(Id, Index),
seg_bytes => maps:get(queue_seg_bytes, Opts, ?DEFAULT_QUEUE_SEG_SIZE), seg_bytes => SegBytes,
max_total_bytes => maps:get(max_queue_bytes, Opts, ?DEFAULT_QUEUE_SIZE), max_total_bytes => TotalBytes,
sizer => fun ?MODULE:estimate_size/1, sizer => fun ?MODULE:estimate_size/1,
marshaller => fun ?MODULE:queue_item_marshaller/1 marshaller => fun ?MODULE:queue_item_marshaller/1
}); });