erts: Add forker StartAck for port start flowcontrol

An acknowledgement of the Start command has to be managed
as we have to make sure that packages are not dropped and
also that the close calls do not happen too early.

3 files changed, 51 insertions, 30 deletions

diff --git a/erts/emulator/sys/unix/erl_child_setup.c b/erts/emulator/sys/unix/erl_child_setup.c
index f74cb0f356..fdffc7f119 100644
--- a/erts/emulator/sys/unix/erl_child_setup.c
+++ b/erts/emulator/sys/unix/erl_child_setup.c
@@ -426,6 +426,10 @@ main(int argc, char *argv[])
             while (write(pipes[1], &proto, sizeof(proto)) < 0 && errno == EINTR)
                 ; /* remove gcc warning */
 
+            proto.action = ErtsSysForkerProtoAction_StartAck;
+            while (write(uds_fd, &proto, sizeof(proto)) < 0 && errno == EINTR)
+                ; /* remove gcc warning */
+
             sys_sigrelease(SIGCHLD);
             close(pipes[0]);
             close(pipes[1]);
diff --git a/erts/emulator/sys/unix/erl_child_setup.h b/erts/emulator/sys/unix/erl_child_setup.h
index c3258f7cbe..93b39d46b2 100644
--- a/erts/emulator/sys/unix/erl_child_setup.h
+++ b/erts/emulator/sys/unix/erl_child_setup.h
@@ -45,6 +45,7 @@ typedef unsigned long long ErtsSysPortId;
 typedef struct ErtsSysForkerProto_ {
     enum {
         ErtsSysForkerProtoAction_Start,
+        ErtsSysForkerProtoAction_StartAck,
         ErtsSysForkerProtoAction_Go,
         ErtsSysForkerProtoAction_SigChld,
         ErtsSysForkerProtoAction_Ack
diff --git a/erts/emulator/sys/unix/sys_drivers.c b/erts/emulator/sys/unix/sys_drivers.c
index 82096e5779..dc1c00ff96 100644
--- a/erts/emulator/sys/unix/sys_drivers.c
+++ b/erts/emulator/sys/unix/sys_drivers.c
@@ -1761,37 +1761,59 @@ static void forker_ready_input(ErlDrvData e, ErlDrvEvent fd)
         erl_exit(ERTS_DUMP_EXIT, "erl_child_setup closed\n");
 
     ASSERT(res == sizeof(*proto));
-    ASSERT(proto->action == ErtsSysForkerProtoAction_SigChld);
-
-    /* ideally this would be a port_command call, but as command is
-       already used by the spawn_driver, we use control instead.
-       Note that when using erl_drv_port_control it is an asynchronous
-       control. */
-    erl_drv_port_control(proto->u.sigchld.port_id, 'S',
-                         (char*)proto, sizeof(*proto));
-
-}
 
-static void forker_ready_output(ErlDrvData e, ErlDrvEvent fd)
-{
-    ErlDrvPort port_num = (ErlDrvPort)e;
-
-    while(driver_sizeq(port_num) > 0) {
+    if (proto->action == ErtsSysForkerProtoAction_StartAck) {
+        /* Ideally we would like to not have to ack each Start
+           command being sent over the uds, but it would seem
+           that some operating systems (only observed on FreeBSD)
+           throw away data on the uds when the socket becomes full,
+           so we have to.
+
+           Also the freebsd manual explicitly states that
+           you should not close fds before they are known
+           to have reached the other side, so this Ack protects
+           against that as well.
+        */
+        ErlDrvPort port_num = (ErlDrvPort)e;
         int vlen;
         SysIOVec *iov = driver_peekq(port_num, &vlen);
         ErtsSysForkerProto *proto = (ErtsSysForkerProto *)iov[0].iov_base;
-        ASSERT(iov[0].iov_len >= (sizeof(*proto)));
-        if (sys_uds_write(forker_fd, (char*)proto, sizeof(*proto),
-                          proto->u.start.fds, 3, 0) < 0) {
-            if (errno == ERRNO_BLOCK)
-                return;
-            erl_exit(ERTS_DUMP_EXIT, "Failed to write to erl_child_setup: %d\n", errno);
-        }
+
         close(proto->u.start.fds[0]);
         close(proto->u.start.fds[1]);
         if (proto->u.start.fds[1] != proto->u.start.fds[2])
             close(proto->u.start.fds[2]);
+
         driver_deq(port_num, sizeof(*proto));
+
+        if (driver_sizeq(port_num) > 0)
+            driver_select(port_num, forker_fd, ERL_DRV_WRITE|ERL_DRV_USE, 1);
+    } else {
+        ASSERT(proto->action == ErtsSysForkerProtoAction_SigChld);
+
+        /* ideally this would be a port_command call, but as command is
+           already used by the spawn_driver, we use control instead.
+           Note that when using erl_drv_port_control it is an asynchronous
+           control. */
+        erl_drv_port_control(proto->u.sigchld.port_id, 'S',
+                             (char*)proto, sizeof(*proto));
+    }
+
+}
+
+static void forker_ready_output(ErlDrvData e, ErlDrvEvent fd)
+{
+    ErlDrvPort port_num = (ErlDrvPort)e;
+
+    int vlen;
+    SysIOVec *iov = driver_peekq(port_num, &vlen);
+    ErtsSysForkerProto *proto = (ErtsSysForkerProto *)iov[0].iov_base;
+    ASSERT(iov[0].iov_len >= (sizeof(*proto)));
+    if (sys_uds_write(forker_fd, (char*)proto, sizeof(*proto),
+                      proto->u.start.fds, 3, 0) < 0) {
+        if (errno == ERRNO_BLOCK)
+            return;
+        erl_exit(ERTS_DUMP_EXIT, "Failed to write to erl_child_setup uds: %d", errno);
     }
 
     driver_select(port_num, forker_fd, ERL_DRV_WRITE, 0);
@@ -1804,25 +1826,19 @@ static ErlDrvSSizeT forker_control(ErlDrvData e, unsigned int cmd, char *buf,
     ErlDrvPort port_num = (ErlDrvPort)e;
     int res;
 
-    if (driver_sizeq(port_num) > 0) {
-        driver_enq(port_num, buf, len);
+    driver_enq(port_num, buf, len);
+    if (driver_sizeq(port_num) > sizeof(*proto)) {
         return 0;
     }
 
     if ((res = sys_uds_write(forker_fd, (char*)proto, sizeof(*proto),
                              proto->u.start.fds, 3, 0)) < 0) {
         if (errno == ERRNO_BLOCK) {
-            driver_enq(port_num, buf, len);
             driver_select(port_num, forker_fd, ERL_DRV_WRITE|ERL_DRV_USE, 1);
             return 0;
         }
         erl_exit(ERTS_DUMP_EXIT, "Failed to write to erl_child_setup: %d\n", errno);
     }
 
-    ASSERT(res == sizeof(*proto));
-    close(proto->u.start.fds[0]);
-    close(proto->u.start.fds[1]);
-    if (proto->u.start.fds[1] != proto->u.start.fds[2])
-        close(proto->u.start.fds[2]);
     return 0;
 }