diff options
| author | Thuan Duong Ba <thuandb@amazon.com> | 2021-12-08 14:52:11 -0800 |
|---|---|---|
| committer | Thuan Duong-Ba <59986634+thuandb@users.noreply.github.com> | 2021-12-20 17:39:06 -0800 |
| commit | 1ab485b44c5dc3e5a8ceacbc89579ca2cf706faf (patch) | |
| tree | 898bf5785807ebf587aaf00594a761286d305c2d | |
| parent | 157bffa3325626961c94038a312317330e3a0d67 (diff) | |
| download | rabbitmq-server-git-1ab485b44c5dc3e5a8ceacbc89579ca2cf706faf.tar.gz | |
minor update for batching messages when syncthroughput is 0
| -rw-r--r-- | deps/rabbit/src/rabbit_mirror_queue_sync.erl | 4 | ||||
| -rw-r--r-- | deps/rabbit/test/rabbit_mirror_queue_sync_SUITE.erl | 31 |
2 files changed, 33 insertions, 2 deletions
diff --git a/deps/rabbit/src/rabbit_mirror_queue_sync.erl b/deps/rabbit/src/rabbit_mirror_queue_sync.erl index f7e1900a29..0519836902 100644 --- a/deps/rabbit/src/rabbit_mirror_queue_sync.erl +++ b/deps/rabbit/src/rabbit_mirror_queue_sync.erl @@ -12,7 +12,7 @@ -export([master_prepare/4, master_go/9, slave/7, conserve_resources/3]). %% Export for UTs --export([maybe_master_batch_send/2, get_time_diff/3]). +-export([maybe_master_batch_send/2, get_time_diff/3, append_to_acc/4]). -define(SYNC_PROGRESS_INTERVAL, 1000000). @@ -130,6 +130,8 @@ bq_fold(FoldFun, FoldAcc, Args, BQ, BQS) -> {_, BQS1} -> master_done(Args, BQS1) end. +append_to_acc(Msg, MsgProps, Unacked, {Batch, I, {_, _, 0}, {Curr, Len}, T}) -> + {[{Msg, MsgProps, Unacked} | Batch], I, {-1, 0, 0}, {Curr + 1, Len}, T}; append_to_acc(Msg, MsgProps, Unacked, {Batch, I, {TotalBytes, LastCheck, SyncThroughput}, {Curr, Len}, T}) -> {[{Msg, MsgProps, Unacked} | Batch], I, {TotalBytes + rabbit_basic:msg_size(Msg), LastCheck, SyncThroughput}, {Curr + 1, Len}, T}. diff --git a/deps/rabbit/test/rabbit_mirror_queue_sync_SUITE.erl b/deps/rabbit/test/rabbit_mirror_queue_sync_SUITE.erl index eb8cc7e918..f9ce4a99b6 100644 --- a/deps/rabbit/test/rabbit_mirror_queue_sync_SUITE.erl +++ b/deps/rabbit/test/rabbit_mirror_queue_sync_SUITE.erl @@ -2,13 +2,15 @@ -include_lib("common_test/include/ct.hrl"). -include_lib("eunit/include/eunit.hrl"). +-include_lib("amqp_client/include/amqp_client.hrl"). -compile(export_all). all() -> [ maybe_master_batch_send, - get_time_diff + get_time_diff, + append_to_acc ]. maybe_master_batch_send(_Config) -> @@ -55,3 +57,30 @@ get_time_diff(_Config) -> 1000, %% => pause queue sync for 1000 ms rabbit_mirror_queue_sync:get_time_diff(TotalBytes2, Interval2, MaxSyncThroughput2)), ok. + +append_to_acc(_Config) -> + Msg = #basic_message{ + id = 1, + content = #content{ + properties = #'P_basic'{ + priority = 2 + }, + payload_fragments_rev = [[<<"1234567890">>]] %% 10 bytes + }, + is_persistent = true + }, + BQDepth = 10, + SyncThroughput_0 = 0, + FoldAcc1 = {[], 0, {0, erlang:monotonic_time(), SyncThroughput_0}, {0, BQDepth}, erlang:monotonic_time()}, + {_, _, {TotalBytes1, _, _}, _, _} = rabbit_mirror_queue_sync:append_to_acc(Msg, {}, false, FoldAcc1), + ?assertEqual(-1, TotalBytes1), %% Skipping calculating TotalBytes for the pending batch as SyncThroughput is 0. + + SyncThroughput = 100, + FoldAcc2 = {[], 0, {0, erlang:monotonic_time(), SyncThroughput}, {0, BQDepth}, erlang:monotonic_time()}, + {_, _, {TotalBytes2, _, _}, _, _} = rabbit_mirror_queue_sync:append_to_acc(Msg, {}, false, FoldAcc2), + ?assertEqual(10, TotalBytes2), %% Message size is added to existing TotalBytes + + FoldAcc3 = {[], 0, {TotalBytes2, erlang:monotonic_time(), SyncThroughput}, {0, BQDepth}, erlang:monotonic_time()}, + {_, _, {TotalBytes3, _, _}, _, _} = rabbit_mirror_queue_sync:append_to_acc(Msg, {}, false, FoldAcc3), + ?assertEqual(TotalBytes2 + 10, TotalBytes3), %% Message size is added to existing TotalBytes + ok.
\ No newline at end of file |