Adapt the socket receive buffer size to traffic

erts/emulator/nifs/common/prim_socket_int.h

@@ -504,18 +504,26 @@ typedef struct {
     ESockCounter       accFails;
     /* +++ Config stuff +++ */
     size_t             rBufSz;  // Read buffer size (when data length = 0)
-    /* rNum and rNumCnt are used (together with rBufSz) when calling the recv 
+    /* rNum and rNumCnt are used (together with rBufSz) when calling the recv
      * function with the Length argument set to 0 (zero).
      * If rNum is 0 (zero), then rNumCnt is not used and only *one* read will
-     * be done. Also, when get'ing the value of the option (rcvbuf) with 
-     * getopt, the value will be reported as an integer. If the rNum has a 
+     * be done. Also, when get'ing the value of the option (rcvbuf) with
+     * getopt, the value will be reported as an integer. If the rNum has a
      * value greater then 0 (zero), then it will instead be reported as
      * {N, BufSz}.
      * On Windows, rNum and rNumCnt is *not* used!
      */
 #ifndef __WIN32__
     unsigned int       rNum;    // recv: Number of reads using rBufSz
     unsigned int       rNumCnt; // recv: Current number of reads (so far)
+    /* While rcvbuf has not been set explicitly, a length 0 recv on a
+     * stream socket reads into a buffer that adapts to the traffic
+     * (rBufSzAdapt, only used by essio_recv). An explicit rcvbuf pins
+     * the size, since it bounds the chunks such a recv may return.
+     */
+    size_t             rBufSzAdapt; // Current adaptive read buffer size
+    size_t             rBufSzAvg;   // EWMA of the read sizes
+    BOOLEAN_T          rBufAdapt;
 #endif
     size_t             rCtrlSz; // Read control buffer size
 

erts/emulator/nifs/common/prim_socket_nif.c

@@ -8224,6 +8224,11 @@ ERL_NIF_TERM esock_setopt_otp_rcvbuf(ErlNifEnv*       env,
         descP->rBufSz = ESOCK_RECV_BUFFER_SIZE_MIN;
     else
         descP->rBufSz = bufSz;
+#ifndef __WIN32__
+    descP->rBufAdapt   = FALSE;
+    descP->rBufSzAdapt = descP->rBufSz;
+    descP->rBufSzAvg   = descP->rBufSz;
+#endif
 
     SSDBG( descP,
            ("SOCKET", "esock_setopt_otp_rcvbuf {%d} -> ok"
@@ -17062,6 +17067,9 @@ ESockDescriptor* esock_alloc_descriptor(SOCKET sock)
 #ifndef __WIN32__
     descP->rNum             = ESOCK_RECV_BUFFER_COUNT_DEFAULT;
     descP->rNumCnt          = 0;
+    descP->rBufAdapt        = TRUE;
+    descP->rBufSzAdapt      = ESOCK_RECV_BUFFER_SIZE_DEFAULT;
+    descP->rBufSzAvg        = ESOCK_RECV_BUFFER_SIZE_DEFAULT;
 #endif
     descP->rCtrlSz          = ESOCK_RECV_CTRL_BUFFER_SIZE_DEFAULT;
     descP->wCtrlSz          = ESOCK_SEND_CTRL_BUFFER_SIZE_DEFAULT;

erts/emulator/nifs/unix/unix_socket_syncio.c

@@ -209,6 +209,13 @@
     if (ctrl.sctp.bindx == NULL)                             \
         return enif_raise_exception((e), MKA((e), "notsup"));
 #define sock_close(s)                   close((s))
+
+/* Adaptive read buffer: double on a filled buffer, halve back towards
+ * the configured size once an EWMA of the read sizes falls below a
+ * quarter of it.
+ */
+#define ESSIO_RECV_ADAPT_BUFFER_MAX     (1 << 18)
+#define ESSIO_RECV_ADAPT_EWMA_SHIFT     3
 // #define sock_close_event(e)             /* do nothing */
 #define sock_connect(s, addr, len)      connect((s), (addr), (len))
 #define sock_connectx(s, addrs, acnt, aidp) \
@@ -2841,6 +2848,9 @@ BOOLEAN_T essio_accept_accepted(ErlNifEnv*       env,
     MLOCK(descP->writeMtx);
 
     accDescP->rBufSz   = descP->rBufSz;  // Inherit buffer size
+    accDescP->rBufAdapt = descP->rBufAdapt;
+    accDescP->rBufSzAdapt = descP->rBufSz;
+    accDescP->rBufSzAvg = descP->rBufSz;
     accDescP->rNum     = descP->rNum;    // Inherit buffer uses
     accDescP->rNumCnt  = 0;
     accDescP->rCtrlSz  = descP->rCtrlSz; // Inherit buffer size
@@ -2944,6 +2954,9 @@ ERL_NIF_TERM essio_peeloff(ErlNifEnv*       env,
             __FUNCTION__, descP->sock, sock) );
 
     poDescP->rBufSz   = descP->rBufSz;  // Inherit buffer size
+    poDescP->rBufAdapt = descP->rBufAdapt;
+    poDescP->rBufSzAdapt = descP->rBufSz;
+    poDescP->rBufSzAvg = descP->rBufSz;
     poDescP->rNum     = descP->rNum;    // Inherit buffer uses
     poDescP->rNumCnt  = 0;
     poDescP->rCtrlSz  = descP->rCtrlSz; // Inherit buffer size
@@ -3803,7 +3816,9 @@ ERL_NIF_TERM essio_recv(ErlNifEnv*       env,
     int          saveErrno;
     ErlNifBinary bin, *bufP;
     ssize_t      readResult;
-    size_t       bufSz = (len != 0 ? len : descP->rBufSz); // 0 means default
+    size_t       bufSz = (len != 0 ? (size_t) len : // 0 means default
+                          (descP->type == SOCK_STREAM ?
+                           descP->rBufSzAdapt : descP->rBufSz));
     ERL_NIF_TERM ret;
 
     SSDBG( descP, ("UNIX-ESSIO", "essio_recv {%d} -> entry with"
@@ -3847,6 +3862,20 @@ ERL_NIF_TERM essio_recv(ErlNifEnv*       env,
     }
     /* readResult >= 0 */
 
+    if ((len == 0) && descP->rBufAdapt && (descP->type == SOCK_STREAM)) {
+        descP->rBufSzAvg -= descP->rBufSzAvg >> ESSIO_RECV_ADAPT_EWMA_SHIFT;
+        descP->rBufSzAvg += ((size_t) readResult) >> ESSIO_RECV_ADAPT_EWMA_SHIFT;
+        if ((size_t) readResult == bufP->size) {
+            if (descP->rBufSzAdapt < ESSIO_RECV_ADAPT_BUFFER_MAX)
+                descP->rBufSzAdapt <<= 1;
+        } else if ((descP->rBufSzAdapt > descP->rBufSz) &&
+                   (descP->rBufSzAvg < (descP->rBufSzAdapt >> 2))) {
+            descP->rBufSzAdapt >>= 1;
+            if (descP->rBufSzAdapt < descP->rBufSz)
+                descP->rBufSzAdapt = descP->rBufSz;
+        }
+    }
+
     ESOCK_ASSERT( recv_create_bin(bufP, readResult, &bin) );
 
     if (bin.size < bufP->size) {

lib/kernel/test/socket_api_SUITE.erl

@@ -200,6 +200,7 @@
          api_opt_simple_otp_options/1,
          api_opt_simple_otp_meta_option/1,
          api_opt_simple_otp_rcvbuf_option/1,
+         api_opt_adaptive_otp_rcvbuf_option/1,
          api_opt_simple_otp_controlling_process/1,
          api_opt_sock_acceptconn_udp/1,
          api_opt_sock_acceptconn_tcp/1,
@@ -510,6 +511,7 @@ api_options_otp_cases() ->
      api_opt_simple_otp_options,
      api_opt_simple_otp_meta_option,
      api_opt_simple_otp_rcvbuf_option,
+     api_opt_adaptive_otp_rcvbuf_option,
      api_opt_simple_otp_controlling_process
     ].
 
@@ -12024,6 +12026,137 @@ api_opt_simple_otp_rcvbuf_option() ->
 
 
 
+%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
+
+%% The buffer used by a recv with Length = 0 on a stream socket adapts
+%% to the traffic, unless the (otp) rcvbuf option has been set, in
+%% which case that size is used (and bounds the returned chunks) as
+%% before. Dgram sockets read into the configured size as before.
+%% Adaptation is not implemented on Windows.
+api_opt_adaptive_otp_rcvbuf_option(_Config) when is_list(_Config) ->
+    ?TT(?SECS(30)),
+    tc_try(?FUNCTION_NAME,
+           fun() ->
+                   has_support_ipv4(),
+                   is_not_windows()
+           end,
+           fun() ->
+                   api_opt_adaptive_otp_rcvbuf_option()
+           end).
+
+api_opt_adaptive_otp_rcvbuf_option() ->
+    LSA = which_local_socket_addr(inet),
+
+    {ok, L} = socket:open(inet, stream, tcp),
+    ok = socket:bind(L, LSA#{port => 0}),
+    ok = socket:listen(L),
+    {ok, SSA}     = socket:sockname(L),
+    {ok, Default} = socket:getopt(L, otp, rcvbuf),
+
+    i("verify the buffer adapts to bulk traffic (default rcvbuf ~w)",
+      [Default]),
+    Bulk    = 64 * 1024 * 1024,
+    Client1 = aor_stream_client(SSA, Bulk),
+    {ok, S1} = socket:accept(L),
+    MaxChunk1 = aor_drain(S1, Bulk, 0),
+    i("max chunk: ~w", [MaxChunk1]),
+    if
+        MaxChunk1 > Default ->
+            ok;
+        true ->
+            exit({no_adaptation, MaxChunk1, Default})
+    end,
+    %% The adapted size is internal; getopt reports the configured size
+    {ok, Default} = socket:getopt(S1, otp, rcvbuf),
+    aor_stop_client(Client1),
+    _ = socket:close(S1),
+
+    i("verify an explicitly set rcvbuf bounds the chunks"),
+    Pinned  = 2048,
+    Bulk2   = 8 * 1024 * 1024,
+    Client2 = aor_stream_client(SSA, Bulk2),
+    {ok, S2} = socket:accept(L),
+    ok = socket:setopt(S2, otp, rcvbuf, Pinned),
+    MaxChunk2 = aor_drain(S2, Bulk2, 0),
+    i("max chunk: ~w", [MaxChunk2]),
+    if
+        MaxChunk2 =< Pinned ->
+            ok;
+        true ->
+            exit({not_pinned, MaxChunk2, Pinned})
+    end,
+    aor_stop_client(Client2),
+    _ = socket:close(S2),
+    _ = socket:close(L),
+
+    i("verify a dgram socket does not adapt"),
+    {ok, U} = socket:open(inet, dgram, udp),
+    ok = socket:bind(U, LSA#{port => 0}),
+    {ok, USA} = socket:sockname(U),
+    {ok, C}   = socket:open(inet, dgram, udp),
+    ok = socket:setopt(C, socket, sndbuf, 64 * 1024),
+    %% Datagrams that exactly fill the buffer would grow it if
+    %% adaptation was (wrongly) applied to dgram sockets
+    Fill = binary:copy(<<$x>>, Default),
+    Send = fun(Data) ->
+                   case socket:sendto(C, Data, USA) of
+                       ok                -> ok;
+                       {error, emsgsize} -> skip("dgram size not supported")
+                   end
+           end,
+    [begin
+         ok = Send(Fill),
+         {ok, D} = socket:recv(U, 0, ?SECS(5)),
+         Default = byte_size(D)
+     end || _ <- lists:seq(1, 8)],
+    %% An oversized datagram is still truncated at the configured size
+    ok = Send(binary:copy(<<$y>>, Default + 4096)),
+    {ok, T} = socket:recv(U, 0, ?SECS(5)),
+    i("oversized dgram read back as ~w bytes", [byte_size(T)]),
+    Default = byte_size(T),
+    _ = socket:close(C),
+    _ = socket:close(U),
+    ok.
+
+aor_stream_client(SSA, Bytes) ->
+    Self = self(),
+    spawn_monitor(
+      fun() ->
+              {ok, S} = socket:open(inet, stream, tcp),
+              ok = socket:connect(S, SSA),
+              Chunk = binary:copy(<<$x>>, 1024 * 1024),
+              aor_send(S, Chunk, Bytes),
+              receive
+                  {Self, stop} ->
+                      _ = socket:close(S),
+                      exit(normal)
+              end
+      end).
+
+aor_send(_S, _Chunk, Bytes) when Bytes =< 0 ->
+    ok;
+aor_send(S, Chunk, Bytes) ->
+    ok = socket:send(S, Chunk),
+    aor_send(S, Chunk, Bytes - byte_size(Chunk)).
+
+aor_drain(_S, Bytes, MaxChunk) when Bytes =< 0 ->
+    MaxChunk;
+aor_drain(S, Bytes, MaxChunk) ->
+    {ok, Data} = socket:recv(S, 0, ?SECS(10)),
+    Sz = byte_size(Data),
+    aor_drain(S, Bytes - Sz, max(Sz, MaxChunk)).
+
+aor_stop_client({Pid, MRef}) ->
+    Pid ! {self(), stop},
+    receive
+        {'DOWN', MRef, process, Pid, normal} ->
+            ok;
+        {'DOWN', MRef, process, Pid, Reason} ->
+            exit({client, Reason})
+    end.
+
+
+
 %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
 
 %% Perform some simple getopt and setopt with the level = otp options