[erts][socket] Add {otp, scheduler_polling} option

erts/doc/references/erl_driver.md

@@ -433,6 +433,7 @@ first:
      int nif_major_version;
      int nif_minor_version;
      int dirty_scheduler_support;
+     int scheduler_polling_support;
   } ErlDrvSysInfo;
   ```
 
@@ -487,6 +488,12 @@ first:
   - **`dirty_scheduler_support`** - A value `!= 0` if the runtime system has
     support for dirty scheduler threads; otherwise `0`.
 
+  - **`scheduler_polling_support`** - A value `!= 0` if the runtime system has
+    support for scheduler pollset migration (available on platforms with epoll
+    or kqueue); otherwise `0`. When supported, file descriptors used by NIFs
+    may be automatically migrated to scheduler-specific pollsets after repeated
+    use from the same process.
+
 - **`ErlDrvBinary`{: #ErlDrvBinary }**
 
   ```text

erts/emulator/beam/erl_driver.h

@@ -70,7 +70,7 @@
 
 #define ERL_DRV_EXTENDED_MARKER		(0xfeeeeeed)
 #define ERL_DRV_EXTENDED_MAJOR_VERSION	3
-#define ERL_DRV_EXTENDED_MINOR_VERSION	3
+#define ERL_DRV_EXTENDED_MINOR_VERSION	4
 
 /*
  * The emulator will refuse to load a driver with a major version

erts/emulator/beam/erl_drv_nif.h

@@ -39,6 +39,7 @@ typedef struct {
     int nif_major_version;
     int nif_minor_version;
     int dirty_scheduler_support;
+    int scheduler_polling_support;
 }  ErlDrvSysInfo;
 
 typedef struct {
@@ -58,7 +59,8 @@ enum ErlNifSelectFlags {
     ERL_NIF_SELECT_STOP      = (1 << 2),
     ERL_NIF_SELECT_CANCEL    = (1 << 3),
     ERL_NIF_SELECT_CUSTOM_MSG= (1 << 4),
-    ERL_NIF_SELECT_ERROR     = (1 << 5)
+    ERL_NIF_SELECT_ERROR     = (1 << 5),
+    ERL_NIF_SELECT_NO_SCHEDULER_POLLSET = (1 << 6)
 };
 
 /*

erts/emulator/beam/io.c

@@ -7486,6 +7486,17 @@ driver_system_info(ErlDrvSysInfo *sip, size_t si_size)
 	    1
 	    ;
     }
+    /*
+     * 'scheduler_polling_support' is the last field in the 5th version
+     * (driver version 3.2, NIF version 2.18)
+     */
+    if (si_size >= ERL_DRV_SYS_INFO_SIZE(scheduler_polling_support)) {
+#if ERTS_POLL_USE_SCHEDULER_POLLING
+	sip->scheduler_polling_support = 1;
+#else
+	sip->scheduler_polling_support = 0;
+#endif
+    }
 
 }
 

erts/emulator/nifs/common/prim_socket_int.h

@@ -522,6 +522,7 @@ typedef struct {
     SOCKET             origFD; // A 'socket' created from this FD
     BOOLEAN_T          closeOnClose; // Have we dup'ed or not
     BOOLEAN_T          selectRead; // Try to have read select active
+    BOOLEAN_T          schedulerPolling; // Allow scheduler pollset migration
     /* +++ The dbg flag for SSDBG +++ */
     BOOLEAN_T          dbg;
     BOOLEAN_T          useReg;

erts/emulator/nifs/common/prim_socket_nif.c

@@ -869,6 +869,7 @@ const int esock_ioctl_flags_length = NUM(esock_ioctl_flags);
 #define ESOCK_OPT_OTP_META         1009
 #define ESOCK_OPT_OTP_USE_REGISTRY 1010
 #define ESOCK_OPT_OTP_SELECT_READ  1011
+#define ESOCK_OPT_OTP_SCHEDULER_POLLING 1012
 /**/
 #define ESOCK_OPT_OTP_DOMAIN       1999 // INTERNAL AND ONLY GET
 #if 0
@@ -1243,6 +1244,7 @@ static ERL_NIF_TERM esock_setopt_otp(ErlNifEnv*       env,
     ESOCK_SETOPT_OTP_FUNC_DEF(iow);             \
     ESOCK_SETOPT_OTP_FUNC_DEF(ctrl_proc);       \
     ESOCK_SETOPT_OTP_FUNC_DEF(select_read);     \
+    ESOCK_SETOPT_OTP_FUNC_DEF(scheduler_polling); \
     ESOCK_SETOPT_OTP_FUNC_DEF(rcvbuf);          \
     ESOCK_SETOPT_OTP_FUNC_DEF(rcvctrlbuf);      \
     ESOCK_SETOPT_OTP_FUNC_DEF(sndctrlbuf);      \
@@ -1278,6 +1280,7 @@ static ERL_NIF_TERM esock_getopt_otp(ErlNifEnv*       env,
     ESOCK_GETOPT_OTP_FUNC_DEF(iow);             \
     ESOCK_GETOPT_OTP_FUNC_DEF(ctrl_proc);       \
     ESOCK_GETOPT_OTP_FUNC_DEF(select_read);     \
+    ESOCK_GETOPT_OTP_FUNC_DEF(scheduler_polling); \
     ESOCK_GETOPT_OTP_FUNC_DEF(rcvbuf);          \
     ESOCK_GETOPT_OTP_FUNC_DEF(rcvctrlbuf);      \
     ESOCK_GETOPT_OTP_FUNC_DEF(sndctrlbuf);      \
@@ -6792,6 +6795,12 @@ ERL_NIF_TERM esock_setopt_otp(ErlNifEnv*       env,
         MUNLOCK(descP->readMtx);
         break;
 
+    case ESOCK_OPT_OTP_SCHEDULER_POLLING:
+        MLOCK(descP->readMtx);
+        result = esock_setopt_otp_scheduler_polling(env, descP, eVal);
+        MUNLOCK(descP->readMtx);
+        break;
+
     case ESOCK_OPT_OTP_RCVBUF:
         MLOCK(descP->readMtx);
         result = esock_setopt_otp_rcvbuf(env, descP, eVal);
@@ -6929,6 +6938,51 @@ ERL_NIF_TERM esock_setopt_otp_select_read(ErlNifEnv*       env,
 
 
 
+/* esock_setopt_otp_scheduler_polling - Handle the OTP (level) scheduler_polling option
+ */
+
+static
+ERL_NIF_TERM esock_setopt_otp_scheduler_polling(ErlNifEnv*       env,
+                                                ESockDescriptor* descP,
+                                                ERL_NIF_TERM     eVal)
+{
+    BOOLEAN_T schedulerPolling;
+
+    if (! IS_OPEN(descP->readState)) {
+        SSDBG( descP,
+               ("SOCKET", "esock_setopt_otp_scheduler_polling {%d} -> closed\r\n",
+                descP->sock) );
+        return esock_make_error_closed(env);
+    }
+
+    if (! esock_decode_bool(eVal, &schedulerPolling))
+      return esock_make_invalid(env, esock_atom_value);
+
+    /* Check if scheduler polling is supported on this platform */
+    {
+        ErlNifSysInfo si;
+        enif_system_info(&si, sizeof(ErlNifSysInfo));
+        if (!si.scheduler_polling_support && schedulerPolling) {
+            SSDBG( descP,
+                   ("SOCKET", "esock_setopt_otp_scheduler_polling {%d} -> notsup"
+                    "\r\n   scheduler polling not available on this platform"
+                    "\r\n", descP->sock) );
+            return esock_make_error(env, esock_atom_enotsup);
+        }
+    }
+
+    descP->schedulerPolling = schedulerPolling;
+
+    SSDBG( descP,
+           ("SOCKET", "esock_setopt_otp_scheduler_polling {%d} -> ok"
+            "\r\n   eVal: %T"
+            "\r\n", descP->sock, eVal) );
+
+    return esock_atom_ok;
+}
+
+
+
 /* esock_setopt_otp_ctrl_proc - Handle the OTP (level)
  * controlling_process options
  */
@@ -8589,6 +8643,12 @@ ERL_NIF_TERM esock_getopt_otp(ErlNifEnv*       env,
         MUNLOCK(descP->readMtx);
         break;
 
+    case ESOCK_OPT_OTP_SCHEDULER_POLLING:
+        MLOCK(descP->readMtx);
+        result = esock_getopt_otp_scheduler_polling(env, descP);
+        MUNLOCK(descP->readMtx);
+        break;
+
     case ESOCK_OPT_OTP_RCVBUF:
         MLOCK(descP->readMtx);
         result = esock_getopt_otp_rcvbuf(env, descP);
@@ -8751,6 +8811,34 @@ ERL_NIF_TERM esock_getopt_otp_select_read(ErlNifEnv*       env,
 
 
 
+/* esock_getopt_otp_scheduler_polling - Handle the OTP (level) scheduler_polling option
+ */
+
+static
+ERL_NIF_TERM esock_getopt_otp_scheduler_polling(ErlNifEnv*       env,
+                                                ESockDescriptor* descP)
+{
+    ERL_NIF_TERM eVal;
+
+    if (! IS_OPEN(descP->readState)) {
+        SSDBG( descP,
+               ("SOCKET", "esock_getopt_otp_scheduler_polling {%d} -> done closed\r\n",
+                descP->sock) );
+        return esock_make_error_closed(env);
+    }
+
+    eVal = esock_encode_bool(descP->schedulerPolling);
+
+    SSDBG( descP,
+           ("SOCKET", "esock_getopt_otp_scheduler_polling {%d} ->"
+            "\r\n   eVal: %T"
+            "\r\n", descP->sock, eVal) );
+
+    return esock_make_ok2(env, eVal);
+}
+
+
+
 /* esock_getopt_otp_ctrl_proc - Handle the OTP (level) controlling_process option
  */
 
@@ -12057,6 +12145,12 @@ ESockDescriptor* esock_alloc_descriptor(SOCKET sock)
     descP->iow              = FALSE;
     descP->dbg              = ESOCK_DEBUG_DEFAULT;      // Overwritten by caller
     descP->selectRead       = FALSE;
+    {
+        /* Query scheduler polling support at runtime */
+        ErlNifSysInfo si;
+        enif_system_info(&si, sizeof(ErlNifSysInfo));
+        descP->schedulerPolling = si.scheduler_polling_support ? TRUE : FALSE;
+    }
     descP->useReg           = ESOCK_USE_SOCKET_REGISTRY;// Overwritten by caller
     descP->meta.env         = esock_alloc_env("esock_alloc_descriptor - "
                                               "meta-env");
@@ -12692,9 +12786,15 @@ int esock_select_read(ErlNifEnv*       env,
                       ERL_NIF_TERM     sockRef,   // Socket
                       ERL_NIF_TERM     selectRef) // "ID" of the operation
 {
+    ESockDescriptor* descP = (ESockDescriptor*) obj;
     ERL_NIF_TERM selectMsg = mk_select_msg(env, sockRef, selectRef);
+    enum ErlNifSelectFlags flags = ERL_NIF_SELECT_READ | ERL_NIF_SELECT_CUSTOM_MSG;
+
+    if (!descP->schedulerPolling) {
+        flags |= ERL_NIF_SELECT_NO_SCHEDULER_POLLSET;
+    }
 
-    return enif_select_read(env, event, obj, pidP, selectMsg, NULL);
+    return enif_select_x(env, event, flags, obj, pidP, selectMsg, NULL);
 
 }
 #endif // #ifndef __WIN32__
@@ -12716,9 +12816,15 @@ int esock_select_write(ErlNifEnv*       env,
                        ERL_NIF_TERM     sockRef,   // Socket
                        ERL_NIF_TERM     selectRef) // "ID" of the operation
 {
+    ESockDescriptor* descP = (ESockDescriptor*) obj;
     ERL_NIF_TERM selectMsg = mk_select_msg(env, sockRef, selectRef);
+    enum ErlNifSelectFlags flags = ERL_NIF_SELECT_WRITE | ERL_NIF_SELECT_CUSTOM_MSG;
+
+    if (!descP->schedulerPolling) {
+        flags |= ERL_NIF_SELECT_NO_SCHEDULER_POLLSET;
+    }
 
-    return enif_select_write(env, event, obj, pidP, selectMsg, NULL);
+    return enif_select_x(env, event, flags, obj, pidP, selectMsg, NULL);
 }
 #endif // #ifndef __WIN32__