Implement in 1.0

Author: lhoguin

deps/rabbit/include/rabbit_amqp_reader.hrl

@@ -59,7 +59,10 @@
          buf :: list(),
          buf_len :: non_neg_integer(),
          tracked_channels = maps:new() :: #{channel_number() => Session :: pid()},
-         stats_timer :: rabbit_event:state()
+         stats_timer :: rabbit_event:state(),
+         %% dynamic buffer
+         dynamic_buffer_size = 128,
+         dynamic_buffer_moving_average = 0.0
         }).
 
 -type state() :: #v1{}.

deps/rabbit/src/rabbit_amqp_reader.erl

@@ -111,9 +111,10 @@ recvloop(Deb, State0 = #v1{recv_len = RecvLen,
 mainloop(Deb, State = #v1{sock = Sock, buf = Buf, buf_len = BufLen}) ->
     case rabbit_net:recv(Sock) of
         {data, Data} ->
-            recvloop(Deb, State#v1{buf = [Data | Buf],
-                                   buf_len = BufLen + size(Data),
-                                   pending_recv = false});
+            State1 = maybe_resize_buffer(State, Data),
+            recvloop(Deb, State1#v1{buf = [Data | Buf],
+                                    buf_len = BufLen + size(Data),
+                                    pending_recv = false});
         closed when State#v1.connection_state =:= closed ->
             ok;
         closed ->
@@ -130,6 +131,37 @@ mainloop(Deb, State = #v1{sock = Sock, buf = Buf, buf_len = BufLen}) ->
             end
     end.
 
+maybe_resize_buffer(State=#v1{sock=Sock, dynamic_buffer_size=BufferSize0,
+        dynamic_buffer_moving_average=MovingAvg0}, Data) ->
+    LowDynamicBuffer = 128,
+    HighDynamicBuffer = 131072,
+    DataLen = byte_size(Data),
+    MovingAvg = (MovingAvg0 * 7 + DataLen) / 8,
+    if
+        BufferSize0 < HighDynamicBuffer andalso MovingAvg > BufferSize0 * 0.9 ->
+            BufferSize = min(BufferSize0 * 2, HighDynamicBuffer),
+            case rabbit_net:setopts(Sock, [{buffer, BufferSize}]) of
+                ok -> State#v1{
+                    dynamic_buffer_size=BufferSize,
+                    dynamic_buffer_moving_average=MovingAvg
+                };
+                {error, Reason} ->
+                    throw({inet_error, Reason})
+            end;
+        BufferSize0 > LowDynamicBuffer andalso MovingAvg < BufferSize0 * 0.4 ->
+            BufferSize = max(BufferSize0 div 2, LowDynamicBuffer),
+            case rabbit_net:setopts(Sock, [{buffer, BufferSize}]) of
+                ok -> State#v1{
+                    dynamic_buffer_size=BufferSize,
+                    dynamic_buffer_moving_average=MovingAvg
+                };
+                {error, Reason} ->
+                    throw({inet_error, Reason})
+            end;
+        true ->
+            State#v1{dynamic_buffer_moving_average=MovingAvg}
+    end.
+
 -spec handle_other(any(), state()) -> state() | stop.
 handle_other(emit_stats, State) ->
     emit_stats(State);

deps/rabbit/src/rabbit_reader.erl

@@ -551,7 +551,6 @@ maybe_resize_buffer(State=#v1{sock=Sock, dynamic_buffer_size=BufferSize0,
     if
         BufferSize0 < HighDynamicBuffer andalso MovingAvg > BufferSize0 * 0.9 ->
             BufferSize = min(BufferSize0 * 2, HighDynamicBuffer),
-%            erlang:display({up,BufferSize}),
             case rabbit_net:setopts(Sock, [{buffer, BufferSize}]) of
                 ok -> State#v1{
                     dynamic_buffer_size=BufferSize,
@@ -562,7 +561,6 @@ maybe_resize_buffer(State=#v1{sock=Sock, dynamic_buffer_size=BufferSize0,
             end;
         BufferSize0 > LowDynamicBuffer andalso MovingAvg < BufferSize0 * 0.4 ->
             BufferSize = max(BufferSize0 div 2, LowDynamicBuffer),
-%            erlang:display({down,BufferSize}),
             case rabbit_net:setopts(Sock, [{buffer, BufferSize}]) of
                 ok -> State#v1{
                     dynamic_buffer_size=BufferSize,