Merge bug 25850

3 files changed, 26 insertions, 2 deletions

diff --git a/src/rabbit_basic.erl b/src/rabbit_basic.erl
index 85f9d56e..67109e7d 100644
--- a/src/rabbit_basic.erl
+++ b/src/rabbit_basic.erl
@@ -22,7 +22,7 @@
          message/3, message/4, properties/1, prepend_table_header/3,
          extract_headers/1, map_headers/2, delivery/4, header_routes/1,
          parse_expiration/1]).
--export([build_content/2, from_content/1, msg_size/1]).
+-export([build_content/2, from_content/1, msg_size/1, maybe_gc_large_msg/1]).
 
 %%----------------------------------------------------------------------------
 
@@ -79,6 +79,9 @@
 -spec(msg_size/1 :: (rabbit_types:content() | rabbit_types:message()) ->
                          non_neg_integer()).
 
+-spec(maybe_gc_large_msg/1 ::
+        (rabbit_types:content() | rabbit_types:message()) -> non_neg_integer()).
+
 -endif.
 
 %%----------------------------------------------------------------------------
@@ -276,5 +279,24 @@ parse_expiration(#'P_basic'{expiration = Expiration}) ->
             {error, {leftover_string, S}}
     end.
 
+%% Some processes (channel, writer) can get huge amounts of binary
+%% garbage when processing huge messages at high speed (since we only
+%% do enough reductions to GC every few hundred messages, and if each
+%% message is 1MB then that's ugly). So count how many bytes of
+%% message we have processed, and force a GC every so often.
+maybe_gc_large_msg(Content) ->
+    Size = msg_size(Content),
+    Current = case get(msg_size_for_gc) of
+                  undefined -> 0;
+                  C         -> C
+              end,
+    New = Current + Size,
+    put(msg_size_for_gc, case New > 1000000 of
+                             true  -> erlang:garbage_collect(),
+                                      0;
+                             false -> New
+                         end),
+    Size.
+
 msg_size(#content{payload_fragments_rev = PFR}) -> iolist_size(PFR);
 msg_size(#basic_message{content = Content})     -> msg_size(Content).
diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl
index fc433898..8632e1b3 100644
--- a/src/rabbit_channel.erl
+++ b/src/rabbit_channel.erl
@@ -375,6 +375,7 @@ handle_cast({deliver, ConsumerTag, AckRequired,
                             exchange     = ExchangeName#resource.name,
                             routing_key  = RoutingKey},
            Content),
+    rabbit_basic:maybe_gc_large_msg(Content),
     noreply(record_sent(ConsumerTag, AckRequired, Msg, State));
 
 handle_cast({deliver_reply, _K, _Del}, State = #ch{state = closing}) ->
@@ -607,7 +608,7 @@ check_internal_exchange(_) ->
     ok.
 
 check_msg_size(Content) ->
-    Size = rabbit_basic:msg_size(Content),
+    Size = rabbit_basic:maybe_gc_large_msg(Content),
     case Size > ?MAX_MSG_SIZE of
         true  -> precondition_failed("message size ~B larger than max size ~B",
                                      [Size, ?MAX_MSG_SIZE]);
diff --git a/src/rabbit_writer.erl b/src/rabbit_writer.erl
index 1882696e..d42d7050 100644
--- a/src/rabbit_writer.erl
+++ b/src/rabbit_writer.erl
@@ -308,6 +308,7 @@ internal_send_command_async(MethodRecord, Content,
                                             pending   = Pending}) ->
     Frames = assemble_frames(Channel, MethodRecord, Content, FrameMax,
                              Protocol),
+    rabbit_basic:maybe_gc_large_msg(Content),
     maybe_flush(State#wstate{pending = [Frames | Pending]}).
 
 %% This magic number is the tcp-over-ethernet MSS (1460) minus the