root
/
rabbitmq-server
mirror of https://github.com/rabbitmq/rabbitmq-server.git
1
0
Fork 0
Code Issues Actions 8 Packages Projects Releases Wiki Activity

make garbage collector threshold configurable

This commit is contained in:
ANycz 2019-12-11 23:11:15 +01:00
parent 178757932e
commit 3daa6be55c
3 changed files with 24 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
Makefile
View File

@ -129,7 +129,9 @@ define PROJECT_ENV
%% interval at which the channel can perform periodic actions %% interval at which the channel can perform periodic actions
{channel_tick_interval, 60000}, {channel_tick_interval, 60000},
%% Default max message size is 128 MB %% Default max message size is 128 MB
{max_message_size, 134217728} {max_message_size, 134217728},
%% Default is ~ 1MB
{gc_threshold, 1000000}
] ]
endef endef

8
src/rabbit_basic.erl
View File

@ -22,7 +22,8 @@
message/3, message/4, properties/1, prepend_table_header/3, message/3, message/4, properties/1, prepend_table_header/3,
extract_headers/1, extract_timestamp/1, map_headers/2, delivery/4, extract_headers/1, extract_timestamp/1, map_headers/2, delivery/4,
header_routes/1, parse_expiration/1, header/2, header/3]). header_routes/1, parse_expiration/1, header/2, header/3]).
-export([build_content/2, from_content/1, msg_size/1, maybe_gc_large_msg/1]). -export([build_content/2, from_content/1, msg_size/1,
maybe_gc_large_msg/1, maybe_gc_large_msg/2]).
-export([add_header/4]). -export([add_header/4]).
%%---------------------------------------------------------------------------- %%----------------------------------------------------------------------------
@ -311,6 +312,11 @@ parse_expiration(#'P_basic'{expiration = Expiration}) ->
maybe_gc_large_msg(Content) -> maybe_gc_large_msg(Content) ->
rabbit_writer:maybe_gc_large_msg(Content). rabbit_writer:maybe_gc_large_msg(Content).
maybe_gc_large_msg(Content, undefined) ->
rabbit_writer:msg_size(Content);
maybe_gc_large_msg(Content, GCThreshold) ->
rabbit_writer:maybe_gc_large_msg(Content, GCThreshold).
msg_size(Content) -> msg_size(Content) ->
rabbit_writer:msg_size(Content). rabbit_writer:msg_size(Content).

22
src/rabbit_channel.erl
View File

@ -169,7 +169,9 @@
delivery_flow, delivery_flow,
interceptor_state, interceptor_state,
queue_states, queue_states,
tick_timer tick_timer,
%% defines how ofter gc will be executed
gc_threshold
}). }).
-define(QUEUE, lqueue). -define(QUEUE, lqueue).
@ -508,6 +510,7 @@ init([Channel, ReaderPid, WriterPid, ConnPid, ConnName, Protocol, User, VHost,
MaxMessageSize = get_max_message_size(), MaxMessageSize = get_max_message_size(),
ConsumerTimeout = get_consumer_timeout(), ConsumerTimeout = get_consumer_timeout(),
OptionalVariables = extract_variable_map_from_amqp_params(AmqpParams), OptionalVariables = extract_variable_map_from_amqp_params(AmqpParams),
{ok, GCThreshold} = application:get_env(rabbit, gc_threshold),
State = #ch{cfg = #conf{state = starting, State = #ch{cfg = #conf{state = starting,
protocol = Protocol, protocol = Protocol,
channel = Channel, channel = Channel,
@ -543,7 +546,8 @@ init([Channel, ReaderPid, WriterPid, ConnPid, ConnName, Protocol, User, VHost,
reply_consumer = none, reply_consumer = none,
delivery_flow = Flow, delivery_flow = Flow,
interceptor_state = undefined, interceptor_state = undefined,
queue_states = #{} queue_states = #{},
gc_threshold = GCThreshold
}, },
State1 = State#ch{ State1 = State#ch{
interceptor_state = rabbit_channel_interceptor:init(State)}, interceptor_state = rabbit_channel_interceptor:init(State)},
@ -1110,8 +1114,8 @@ extract_variable_map_from_amqp_params([Value]) ->
extract_variable_map_from_amqp_params(_) -> extract_variable_map_from_amqp_params(_) ->
#{}. #{}.
check_msg_size(Content, MaxMessageSize) -> check_msg_size(Content, MaxMessageSize, GCThreshold) ->
Size = rabbit_basic:maybe_gc_large_msg(Content), Size = rabbit_basic:maybe_gc_large_msg(Content, GCThreshold),
case Size of case Size of
S when S > MaxMessageSize -> S when S > MaxMessageSize ->
ErrorMessage = case MaxMessageSize of ErrorMessage = case MaxMessageSize of
@ -1309,9 +1313,10 @@ handle_method(#'basic.publish'{exchange = ExchangeNameBin,
}, },
tx = Tx, tx = Tx,
confirm_enabled = ConfirmEnabled, confirm_enabled = ConfirmEnabled,
delivery_flow = Flow delivery_flow = Flow,
gc_threshold = GCThreshold
}) -> }) ->
check_msg_size(Content, MaxMessageSize), check_msg_size(Content, MaxMessageSize, GCThreshold),
ExchangeName = rabbit_misc:r(VHostPath, exchange, ExchangeNameBin), ExchangeName = rabbit_misc:r(VHostPath, exchange, ExchangeNameBin),
check_write_permitted(ExchangeName, User, AuthzContext), check_write_permitted(ExchangeName, User, AuthzContext),
Exchange = rabbit_exchange:lookup_or_die(ExchangeName), Exchange = rabbit_exchange:lookup_or_die(ExchangeName),
@ -2689,7 +2694,8 @@ handle_deliver(ConsumerTag, AckRequired,
routing_keys = [RoutingKey | _CcRoutes], routing_keys = [RoutingKey | _CcRoutes],
content = Content}}, content = Content}},
State = #ch{cfg = #conf{writer_pid = WriterPid}, State = #ch{cfg = #conf{writer_pid = WriterPid},
next_tag = DeliveryTag}) -> next_tag = DeliveryTag,
gc_threshold = GCThreshold}) ->
Deliver = #'basic.deliver'{consumer_tag = ConsumerTag, Deliver = #'basic.deliver'{consumer_tag = ConsumerTag,
delivery_tag = DeliveryTag, delivery_tag = DeliveryTag,
redelivered = Redelivered, redelivered = Redelivered,
@ -2702,7 +2708,7 @@ handle_deliver(ConsumerTag, AckRequired,
false -> false ->
ok = rabbit_writer:send_command(WriterPid, Deliver, Content) ok = rabbit_writer:send_command(WriterPid, Deliver, Content)
end, end,
rabbit_basic:maybe_gc_large_msg(Content), rabbit_basic:maybe_gc_large_msg(Content, GCThreshold),
record_sent(deliver, ConsumerTag, AckRequired, Msg, State). record_sent(deliver, ConsumerTag, AckRequired, Msg, State).
handle_basic_get(WriterPid, DeliveryTag, NoAck, MessageCount, handle_basic_get(WriterPid, DeliveryTag, NoAck, MessageCount,