diff options
author | Simon MacMullen <simon@rabbitmq.com> | 2014-10-08 12:43:26 +0100 |
---|---|---|
committer | Simon MacMullen <simon@rabbitmq.com> | 2014-10-08 12:43:26 +0100 |
commit | 26fb4517fe9fdb5918b9f5c2665663a7f89be5e7 (patch) | |
tree | afe5155d3b6326860fbfc54a8560a7693f056bed | |
parent | cce2494b91bdf4509e977665620faf759c47582f (diff) | |
parent | 5760b4ef12634123b0b568c36105e0368e6caeb2 (diff) | |
download | rabbitmq-server-26fb4517fe9fdb5918b9f5c2665663a7f89be5e7.tar.gz |
Merge bug 25850
-rw-r--r-- | src/rabbit_basic.erl | 24 | ||||
-rw-r--r-- | src/rabbit_channel.erl | 3 | ||||
-rw-r--r-- | src/rabbit_writer.erl | 1 |
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 |