erts: Shrink the adaptive esock rcvbuf via an EWMA

The consecutive-reads counter made shrinking statistically
unreachable on mixed traffic: a single read over a quarter of the
buffer reset it, so a buffer at the cap could stay there through
tens of thousands of smaller reads. Track an EWMA of the read sizes
instead (alpha 1/8, shift arithmetic) and halve the buffer whenever
the average falls below a quarter of it. Bulk bursts still ramp up
unchanged since every filled read doubles the buffer, and once
traffic turns small the average converges within a couple dozen
reads and the buffer halves per read back to the configured size.
An isolated filled read among small ones now also shrinks back
instead of ratcheting.

Author: lpgauth

erts/emulator/nifs/common/prim_socket_int.h

@@ -522,7 +522,7 @@ typedef struct {
      * the size, since it bounds the chunks such a recv may return.
      */
     size_t             rBufSzAdapt; // Current adaptive read buffer size
-    unsigned int       rBufShrinkCnt;
+    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

@@ -8225,9 +8225,9 @@ ERL_NIF_TERM esock_setopt_otp_rcvbuf(ErlNifEnv*       env,
     else
         descP->rBufSz = bufSz;
 #ifndef __WIN32__
-    descP->rBufAdapt     = FALSE;
-    descP->rBufSzAdapt   = descP->rBufSz;
-    descP->rBufShrinkCnt = 0;
+    descP->rBufAdapt   = FALSE;
+    descP->rBufSzAdapt = descP->rBufSz;
+    descP->rBufSzAvg   = descP->rBufSz;
 #endif
 
     SSDBG( descP,
@@ -17069,7 +17069,7 @@ ESockDescriptor* esock_alloc_descriptor(SOCKET sock)
     descP->rNumCnt          = 0;
     descP->rBufAdapt        = TRUE;
     descP->rBufSzAdapt      = ESOCK_RECV_BUFFER_SIZE_DEFAULT;
-    descP->rBufShrinkCnt    = 0;
+    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

@@ -211,11 +211,11 @@
 #define sock_close(s)                   close((s))
 
 /* Adaptive read buffer: double on a filled buffer, halve back towards
- * the configured size after this many consecutive reads that used
- * less than a quarter of it.
+ * the configured size once an EWMA of the read sizes falls below a
+ * quarter of it.
  */
 #define ESSIO_RECV_ADAPT_BUFFER_MAX     (1 << 20)
-#define ESSIO_RECV_ADAPT_SHRINK_COUNT   32
+#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) \
@@ -2850,7 +2850,7 @@ BOOLEAN_T essio_accept_accepted(ErlNifEnv*       env,
     accDescP->rBufSz   = descP->rBufSz;  // Inherit buffer size
     accDescP->rBufAdapt = descP->rBufAdapt;
     accDescP->rBufSzAdapt = descP->rBufSz;
-    accDescP->rBufShrinkCnt = 0;
+    accDescP->rBufSzAvg = descP->rBufSz;
     accDescP->rNum     = descP->rNum;    // Inherit buffer uses
     accDescP->rNumCnt  = 0;
     accDescP->rCtrlSz  = descP->rCtrlSz; // Inherit buffer size
@@ -2956,7 +2956,7 @@ ERL_NIF_TERM essio_peeloff(ErlNifEnv*       env,
     poDescP->rBufSz   = descP->rBufSz;  // Inherit buffer size
     poDescP->rBufAdapt = descP->rBufAdapt;
     poDescP->rBufSzAdapt = descP->rBufSz;
-    poDescP->rBufShrinkCnt = 0;
+    poDescP->rBufSzAvg = descP->rBufSz;
     poDescP->rNum     = descP->rNum;    // Inherit buffer uses
     poDescP->rNumCnt  = 0;
     poDescP->rCtrlSz  = descP->rCtrlSz; // Inherit buffer size
@@ -3863,20 +3863,16 @@ 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;
-            descP->rBufShrinkCnt = 0;
         } else if ((descP->rBufSzAdapt > descP->rBufSz) &&
-                   ((size_t) readResult < (descP->rBufSzAdapt >> 2))) {
-            if (++descP->rBufShrinkCnt >= ESSIO_RECV_ADAPT_SHRINK_COUNT) {
-                descP->rBufSzAdapt >>= 1;
-                if (descP->rBufSzAdapt < descP->rBufSz)
-                    descP->rBufSzAdapt = descP->rBufSz;
-                descP->rBufShrinkCnt = 0;
-            }
-        } else {
-            descP->rBufShrinkCnt = 0;
+                   (descP->rBufSzAvg < (descP->rBufSzAdapt >> 2))) {
+            descP->rBufSzAdapt >>= 1;
+            if (descP->rBufSzAdapt < descP->rBufSz)
+                descP->rBufSzAdapt = descP->rBufSz;
         }
     }