Merge branch 'nyczol-configurable_gc_threshold'

(cherry picked from commit e86b5a6)

Conflicts:
	Makefile
	src/rabbit_basic.erl
	src/rabbit_channel.erl

Author: michaelklishin

Makefile

@@ -123,7 +123,9 @@ define PROJECT_ENV
 	    %% vhost had to shut down, see server#1158 and server#1280
 	    {vhost_restart_strategy, continue},
 	    %% {global, prefetch count}
-	    {default_consumer_prefetch, {false, 0}}
+	    {default_consumer_prefetch, {false, 0}},
+			%% Socket writer will run GC every 1 GB of outgoing data
+	    {gc_threshold, 1000000000}
 	  ]
 endef
 

src/rabbit_basic.erl

@@ -22,7 +22,8 @@
          message/3, message/4, properties/1, prepend_table_header/3,
          extract_headers/1, extract_timestamp/1, map_headers/2, delivery/4,
          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]).
 
 %%----------------------------------------------------------------------------
 
@@ -298,5 +299,10 @@ parse_expiration(#'P_basic'{expiration = Expiration}) ->
 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) ->
     rabbit_writer:msg_size(Content).

src/rabbit_channel.erl

@@ -161,7 +161,9 @@
   %% flow | noflow, see rabbitmq-server#114
   delivery_flow,
   interceptor_state,
-  authz_context
+  authz_context,
+  %% defines how ofter gc will be executed
+  gc_threshold
 }).
 
 
@@ -456,6 +458,7 @@ init([Channel, ReaderPid, WriterPid, ConnPid, ConnName, Protocol, User, VHost,
                       Limiter0
               end,
     OptionalVariables = extract_topic_variable_map_from_amqp_params(AmqpParams),
+    {ok, GCThreshold} = application:get_env(rabbit, gc_threshold),
     State = #ch{state                   = starting,
                 protocol                = Protocol,
                 channel                 = Channel,
@@ -488,7 +491,8 @@ init([Channel, ReaderPid, WriterPid, ConnPid, ConnName, Protocol, User, VHost,
                 reply_consumer          = none,
                 delivery_flow           = Flow,
                 interceptor_state       = undefined,
-                authz_context           = OptionalVariables},
+                authz_context           = OptionalVariables,
+                gc_threshold            = GCThreshold},
     State1 = State#ch{
                interceptor_state = rabbit_channel_interceptor:init(State)},
     State2 = rabbit_event:init_stats_timer(State1, #ch.stats_timer),
@@ -931,8 +935,8 @@ extract_topic_variable_map_from_amqp_params([Value]) ->
 extract_topic_variable_map_from_amqp_params(_) ->
     #{}.
 
-check_msg_size(Content) ->
-    Size = rabbit_basic:maybe_gc_large_msg(Content),
+check_msg_size(Content, GCThreshold) ->
+    Size = rabbit_basic:maybe_gc_large_msg(Content, GCThreshold),
     case Size > ?MAX_MSG_SIZE of
         true  -> precondition_failed("message size ~B larger than max size ~B",
                                      [Size, ?MAX_MSG_SIZE]);
@@ -1118,8 +1122,9 @@ handle_method(#'basic.publish'{exchange    = ExchangeNameBin,
                                    user            = #user{username = Username} = User,
                                    conn_name       = ConnName,
                                    delivery_flow   = Flow,
-                                   authz_context   = AuthzContext}) ->
-    check_msg_size(Content),
+                                   authz_context   = AuthzContext,
+                                   gc_threshold     = GCThreshold}) ->
+    check_msg_size(Content, GCThreshold),
     ExchangeName = rabbit_misc:r(VHostPath, exchange, ExchangeNameBin),
     check_write_permitted(ExchangeName, User),
     Exchange = rabbit_exchange:lookup_or_die(ExchangeName),