Add dynamic buffer functionality to rabbit_reader

Author: lhoguin

deps/rabbit/src/rabbit_networking.erl

@@ -570,18 +570,19 @@ handshake(Ref, ProxyProtocolEnabled) ->
     end.
 
 tune_buffer_size(Sock) ->
-    case tune_buffer_size1(Sock) of
+%    case tune_buffer_size1(Sock) of
+    case rabbit_net:setopts(Sock, [{buffer, 1024}]) of
         ok         -> ok;
         {error, _} -> rabbit_net:fast_close(Sock),
                       exit(normal)
     end.
 
-tune_buffer_size1(Sock) ->
-    case rabbit_net:getopts(Sock, [sndbuf, recbuf, buffer]) of
-        {ok, BufSizes} -> BufSz = lists:max([Sz || {_Opt, Sz} <- BufSizes]),
-                          rabbit_net:setopts(Sock, [{buffer, BufSz}]);
-        Error          -> Error
-    end.
+%tune_buffer_size1(Sock) ->
+%    case rabbit_net:getopts(Sock, [sndbuf, recbuf, buffer]) of
+%        {ok, BufSizes} -> BufSz = lists:max([Sz || {_Opt, Sz} <- BufSizes]),
+%                          rabbit_net:setopts(Sock, [{buffer, BufSz}]);
+%        Error          -> Error
+%    end.
 
 %%--------------------------------------------------------------------
 

deps/rabbit/src/rabbit_reader.erl

@@ -99,7 +99,11 @@
           %% throttling state, for both
           %% credit- and resource-driven flow control
           throttle,
-          proxy_socket}).
+          proxy_socket,
+          %% dynamic buffer
+          dynamic_buffer_size,
+          dynamic_buffer_moving_average
+}).
 
 -record(throttle, {
   %% never | timestamp()
@@ -512,8 +516,9 @@ mainloop(Deb, Buf, BufLen, State = #v1{sock = Sock,
     end,
     case Recv of
         {data, Data} ->
+            State1 = maybe_resize_buffer(State, Data),
             recvloop(Deb, [Data | Buf], BufLen + size(Data),
-                     State#v1{pending_recv = false});
+                     State1#v1{pending_recv = false});
         closed when State#v1.connection_state =:= closed ->
             State;
         closed when CS =:= pre_init andalso Buf =:= [] ->
@@ -536,6 +541,37 @@ mainloop(Deb, Buf, BufLen, State = #v1{sock = Sock,
             end
     end.
 
+maybe_resize_buffer(State=#v1{sock=Sock, dynamic_buffer_size=BufferSize0,
+        dynamic_buffer_moving_average=MovingAvg0}, Data) ->
+    LowDynamicBuffer = 1024,
+    HighDynamicBuffer = 131072,
+    DataLen = byte_size(Data),
+    MovingAvg = (MovingAvg0 + DataLen) div 2,
+    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 ->
+                    stop(Error, State)
+            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 ->
+                    stop(Error, State)
+            end;
+        true ->
+            State#v1{dynamic_buffer_moving_average=MovingAvg}
+    end.
+
 -spec stop(_, #v1{}) -> no_return().
 stop(tcp_healthcheck, State) ->
     %% The connection was closed before any packet was received. It's