Port message extended to transfer 2 file descriptors.

5 files changed, 81 insertions, 27 deletions

diff --git a/src/nxt_port.h b/src/nxt_port.h
index 9a933e75..ab455f92 100644
--- a/src/nxt_port.h
+++ b/src/nxt_port.h
@@ -165,6 +165,7 @@ typedef struct {
     nxt_buf_t           *buf;
     size_t              share;
     nxt_fd_t            fd;
+    nxt_fd_t            fd2;
     nxt_port_msg_t      port_msg;
     uint32_t            tracking_msg[2];
     uint8_t             close_fd;   /* 1 bit */
@@ -174,6 +175,7 @@ typedef struct {
 
 struct nxt_port_recv_msg_s {
     nxt_fd_t            fd;
+    nxt_fd_t            fd2;
     nxt_buf_t           *buf;
     nxt_port_t          *port;
     nxt_port_msg_t      port_msg;
diff --git a/src/nxt_port_socket.c b/src/nxt_port_socket.c
index 4e3eaef6..844b65ca 100644
--- a/src/nxt_port_socket.c
+++ b/src/nxt_port_socket.c
@@ -156,6 +156,7 @@ nxt_port_socket_twrite(nxt_task_t *task, nxt_port_t *port, nxt_uint_t type,
     msg.buf = b;
     msg.share = 0;
     msg.fd = fd;
+    msg.fd2 = -1;
     msg.close_fd = (type & NXT_PORT_MSG_CLOSE_FD) != 0;
     msg.allocated = 0;
 
@@ -331,7 +332,7 @@ next_fragment:
         msg->port_msg.last |= sb.last;
         msg->port_msg.mf = sb.limit_reached || sb.nmax_reached;
 
-        n = nxt_socketpair_send(&port->socket, msg->fd, iov, sb.niov + 1);
+        n = nxt_socketpair_send(&port->socket, &msg->fd, iov, sb.niov + 1);
 
         if (n > 0) {
             if (nxt_slow_path((size_t) n != sb.size + iov[0].iov_len)) {
@@ -346,6 +347,12 @@ next_fragment:
                 msg->fd = -1;
             }
 
+            if (msg->fd2 != -1 && msg->close_fd != 0) {
+                nxt_fd_close(msg->fd2);
+
+                msg->fd2 = -1;
+            }
+
             msg->buf = nxt_port_buf_completion(task, wq, msg->buf, plain_size,
                                                m == NXT_PORT_METHOD_MMAP);
 
@@ -358,6 +365,7 @@ next_fragment:
                  * in the first message of a stream.
                  */
                 msg->fd = -1;
+                msg->fd2 = -1;
                 msg->share += n;
                 msg->port_msg.nf = 1;
                 msg->port_msg.tracking = 0;
@@ -810,6 +818,10 @@ nxt_port_read_msg_process(nxt_task_t *task, nxt_port_t *port,
             nxt_fd_close(msg->fd);
         }
 
+        if (msg->fd2 != -1) {
+            nxt_fd_close(msg->fd2);
+        }
+
         return;
     }
 
@@ -854,6 +866,7 @@ nxt_port_read_msg_process(nxt_task_t *task, nxt_port_t *port,
 
                 msg->buf = fmsg->buf;
                 msg->fd = fmsg->fd;
+                msg->fd2 = fmsg->fd2;
 
                 /*
                  * To disable instant completion or buffer re-usage,
@@ -888,12 +901,17 @@ nxt_port_read_msg_process(nxt_task_t *task, nxt_port_t *port,
             if (nxt_fast_path(msg->cancelled == 0)) {
                 msg->buf = NULL;
                 msg->fd = -1;
+                msg->fd2 = -1;
                 b = NULL;
 
             } else {
                 if (msg->fd != -1) {
                     nxt_fd_close(msg->fd);
                 }
+
+                if (msg->fd2 != -1) {
+                    nxt_fd_close(msg->fd2);
+                }
             }
         } else {
             if (nxt_fast_path(msg->cancelled == 0)) {
@@ -999,6 +1017,12 @@ nxt_port_error_handler(nxt_task_t *task, void *obj, void *data)
             msg->fd = -1;
         }
 
+        if (msg->fd2 != -1 && msg->close_fd != 0) {
+            nxt_fd_close(msg->fd2);
+
+            msg->fd2 = -1;
+        }
+
         for (b = msg->buf; b != NULL; b = next) {
             next = b->next;
             b->next = NULL;
diff --git a/src/nxt_socket.h b/src/nxt_socket.h
index 718ad398..7403de3d 100644
--- a/src/nxt_socket.h
+++ b/src/nxt_socket.h
@@ -112,7 +112,7 @@ nxt_uint_t nxt_socket_error_level(nxt_err_t err);
 NXT_EXPORT nxt_int_t nxt_socketpair_create(nxt_task_t *task,
     nxt_socket_t *pair);
 NXT_EXPORT void nxt_socketpair_close(nxt_task_t *task, nxt_socket_t *pair);
-NXT_EXPORT ssize_t nxt_socketpair_send(nxt_fd_event_t *ev, nxt_fd_t fd,
+NXT_EXPORT ssize_t nxt_socketpair_send(nxt_fd_event_t *ev, nxt_fd_t *fd,
     nxt_iobuf_t *iob, nxt_uint_t niob);
 NXT_EXPORT ssize_t nxt_socketpair_recv(nxt_fd_event_t *ev, nxt_fd_t *fd,
     nxt_iobuf_t *iob, nxt_uint_t niob);
diff --git a/src/nxt_socketpair.c b/src/nxt_socketpair.c
index 10ea562e..8b9d12bf 100644
--- a/src/nxt_socketpair.c
+++ b/src/nxt_socketpair.c
@@ -20,7 +20,7 @@
 #endif
 
 
-static ssize_t nxt_sendmsg(nxt_socket_t s, nxt_fd_t fd, nxt_iobuf_t *iob,
+static ssize_t nxt_sendmsg(nxt_socket_t s, nxt_fd_t *fd, nxt_iobuf_t *iob,
     nxt_uint_t niob);
 static ssize_t nxt_recvmsg(nxt_socket_t s, nxt_fd_t *fd, nxt_iobuf_t *iob,
     nxt_uint_t niob);
@@ -71,7 +71,7 @@ nxt_socketpair_close(nxt_task_t *task, nxt_socket_t *pair)
 
 
 ssize_t
-nxt_socketpair_send(nxt_fd_event_t *ev, nxt_fd_t fd, nxt_iobuf_t *iob,
+nxt_socketpair_send(nxt_fd_event_t *ev, nxt_fd_t *fd, nxt_iobuf_t *iob,
     nxt_uint_t niob)
 {
     ssize_t    n;
@@ -82,7 +82,8 @@ nxt_socketpair_send(nxt_fd_event_t *ev, nxt_fd_t fd, nxt_iobuf_t *iob,
 
         err = (n == -1) ? nxt_socket_errno : 0;
 
-        nxt_debug(ev->task, "sendmsg(%d, %FD, %ui): %z", ev->fd, fd, niob, n);
+        nxt_debug(ev->task, "sendmsg(%d, %FD, %FD, %ui): %z", ev->fd, fd[0],
+                  fd[1], niob, n);
 
         if (n > 0) {
             return n;
@@ -108,8 +109,8 @@ nxt_socketpair_send(nxt_fd_event_t *ev, nxt_fd_t fd, nxt_iobuf_t *iob,
             continue;
 
         default:
-            nxt_alert(ev->task, "sendmsg(%d, %FD, %ui) failed %E",
-                      ev->fd, fd, niob, err);
+            nxt_alert(ev->task, "sendmsg(%d, %FD, %FD, %ui) failed %E",
+                      ev->fd, fd[0], fd[1], niob, err);
 
             return NXT_ERROR;
         }
@@ -133,7 +134,8 @@ nxt_socketpair_recv(nxt_fd_event_t *ev, nxt_fd_t *fd, nxt_iobuf_t *iob,
 
         err = (n == -1) ? nxt_socket_errno : 0;
 
-        nxt_debug(ev->task, "recvmsg(%d, %FD, %ui): %z", ev->fd, *fd, niob, n);
+        nxt_debug(ev->task, "recvmsg(%d, %FD, %FD, %ui): %z", ev->fd, fd[0],
+                  fd[1], niob, n);
 
         if (n > 0) {
             return n;
@@ -178,12 +180,13 @@ nxt_socketpair_recv(nxt_fd_event_t *ev, nxt_fd_t *fd, nxt_iobuf_t *iob,
  */
 
 static ssize_t
-nxt_sendmsg(nxt_socket_t s, nxt_fd_t fd, nxt_iobuf_t *iob, nxt_uint_t niob)
+nxt_sendmsg(nxt_socket_t s, nxt_fd_t *fd, nxt_iobuf_t *iob, nxt_uint_t niob)
 {
+    size_t              csize;
     struct msghdr       msg;
     union {
         struct cmsghdr  cm;
-        char            space[CMSG_SPACE(sizeof(int))];
+        char            space[CMSG_SPACE(sizeof(int) * 2)];
     } cmsg;
 
     msg.msg_name = NULL;
@@ -193,15 +196,17 @@ nxt_sendmsg(nxt_socket_t s, nxt_fd_t fd, nxt_iobuf_t *iob, nxt_uint_t niob)
     /* Flags are cleared just to suppress valgrind warning. */
     msg.msg_flags = 0;
 
-    if (fd != -1) {
+    if (fd[0] != -1) {
+        csize = (fd[1] == -1) ? sizeof(int) : sizeof(int) * 2;
+
         msg.msg_control = (caddr_t) &cmsg;
-        msg.msg_controllen = sizeof(cmsg);
+        msg.msg_controllen = CMSG_SPACE(csize);
 
 #if (NXT_VALGRIND)
         nxt_memzero(&cmsg, sizeof(cmsg));
 #endif
 
-        cmsg.cm.cmsg_len = CMSG_LEN(sizeof(int));
+        cmsg.cm.cmsg_len = CMSG_LEN(csize);
         cmsg.cm.cmsg_level = SOL_SOCKET;
         cmsg.cm.cmsg_type = SCM_RIGHTS;
 
@@ -214,7 +219,7 @@ nxt_sendmsg(nxt_socket_t s, nxt_fd_t fd, nxt_iobuf_t *iob, nxt_uint_t niob)
          * Fortunately, GCC with -O1 compiles this nxt_memcpy()
          * in the same simple assignment as in the code above.
          */
-        nxt_memcpy(CMSG_DATA(&cmsg.cm), &fd, sizeof(int));
+        nxt_memcpy(CMSG_DATA(&cmsg.cm), fd, csize);
 
     } else {
         msg.msg_control = NULL;
@@ -232,7 +237,7 @@ nxt_recvmsg(nxt_socket_t s, nxt_fd_t *fd, nxt_iobuf_t *iob, nxt_uint_t niob)
     struct msghdr       msg;
     union {
         struct cmsghdr  cm;
-        char            space[CMSG_SPACE(sizeof(int))];
+        char            space[CMSG_SPACE(sizeof(int) * 2)];
     } cmsg;
 
     msg.msg_name = NULL;
@@ -242,7 +247,8 @@ nxt_recvmsg(nxt_socket_t s, nxt_fd_t *fd, nxt_iobuf_t *iob, nxt_uint_t niob)
     msg.msg_control = (caddr_t) &cmsg;
     msg.msg_controllen = sizeof(cmsg);
 
-    *fd = -1;
+    fd[0] = -1;
+    fd[1] = -1;
 
 #if (NXT_VALGRIND)
     nxt_memzero(&cmsg, sizeof(cmsg));
@@ -251,12 +257,16 @@ nxt_recvmsg(nxt_socket_t s, nxt_fd_t *fd, nxt_iobuf_t *iob, nxt_uint_t niob)
     n = recvmsg(s, &msg, 0);
 
     if (n > 0
-        && cmsg.cm.cmsg_len == CMSG_LEN(sizeof(int))
         && cmsg.cm.cmsg_level == SOL_SOCKET
         && cmsg.cm.cmsg_type == SCM_RIGHTS)
     {
-        /* (*fd) = *(int *) CMSG_DATA(&cmsg.cm); */
-        nxt_memcpy(fd, CMSG_DATA(&cmsg.cm), sizeof(int));
+        if (cmsg.cm.cmsg_len == CMSG_LEN(sizeof(int))) {
+            nxt_memcpy(fd, CMSG_DATA(&cmsg.cm), sizeof(int));
+        }
+
+        if (cmsg.cm.cmsg_len == CMSG_LEN(sizeof(int) * 2)) {
+            nxt_memcpy(fd, CMSG_DATA(&cmsg.cm), sizeof(int) * 2);
+        }
     }
 
     return n;
@@ -267,7 +277,7 @@ nxt_recvmsg(nxt_socket_t s, nxt_fd_t *fd, nxt_iobuf_t *iob, nxt_uint_t niob)
 /* Solaris 4.3BSD sockets. */
 
 static ssize_t
-nxt_sendmsg(nxt_socket_t s, nxt_fd_t fd, nxt_iobuf_t *iob, nxt_uint_t niob)
+nxt_sendmsg(nxt_socket_t s, nxt_fd_t *fd, nxt_iobuf_t *iob, nxt_uint_t niob)
 {
     struct msghdr  msg;
 
@@ -276,10 +286,14 @@ nxt_sendmsg(nxt_socket_t s, nxt_fd_t fd, nxt_iobuf_t *iob, nxt_uint_t niob)
     msg.msg_iov = iob;
     msg.msg_iovlen = niob;
 
-    if (fd != -1) {
-        msg.msg_accrights = (caddr_t) &fd;
+    if (fd[0] != -1) {
+        msg.msg_accrights = (caddr_t) fd;
         msg.msg_accrightslen = sizeof(int);
 
+        if (fd[1] != -1) {
+            msg.msg_accrightslen += sizeof(int);
+        }
+
     } else {
         msg.msg_accrights = NULL;
         msg.msg_accrightslen = 0;
@@ -294,14 +308,15 @@ nxt_recvmsg(nxt_socket_t s, nxt_fd_t *fd, nxt_iobuf_t *iob, nxt_uint_t niob)
 {
     struct msghdr  msg;
 
-    *fd = -1;
+    fd[0] = -1;
+    fd[1] = -1;
 
     msg.msg_name = NULL;
     msg.msg_namelen = 0;
     msg.msg_iov = iob;
     msg.msg_iovlen = niob;
     msg.msg_accrights = (caddr_t) fd;
-    msg.msg_accrightslen = sizeof(int);
+    msg.msg_accrightslen = sizeof(int) * 2;
 
     return recvmsg(s, &msg, 0);
 }
diff --git a/src/nxt_unit.c b/src/nxt_unit.c
index 154fd480..66aadd98 100644
--- a/src/nxt_unit.c
+++ b/src/nxt_unit.c
@@ -192,6 +192,7 @@ struct nxt_unit_recv_msg_s {
     uint32_t                 size;
 
     int                      fd;
+    int                      fd2;
 
     nxt_unit_mmap_buf_t      *incoming_buf;
 };
@@ -805,14 +806,20 @@ nxt_unit_process_msg(nxt_unit_ctx_t *ctx, nxt_unit_read_buf_t *rbuf)
 
     rc = NXT_UNIT_ERROR;
     recv_msg.fd = -1;
+    recv_msg.fd2 = -1;
     port_msg = (nxt_port_msg_t *) rbuf->buf;
     cm = (struct cmsghdr *) rbuf->oob;
 
-    if (cm->cmsg_len == CMSG_LEN(sizeof(int))
-        && cm->cmsg_level == SOL_SOCKET
+    if (cm->cmsg_level == SOL_SOCKET
         && cm->cmsg_type == SCM_RIGHTS)
     {
-        memcpy(&recv_msg.fd, CMSG_DATA(cm), sizeof(int));
+        if (cm->cmsg_len == CMSG_LEN(sizeof(int))) {
+            memcpy(&recv_msg.fd, CMSG_DATA(cm), sizeof(int));
+        }
+
+        if (cm->cmsg_len == CMSG_LEN(sizeof(int) * 2)) {
+            memcpy(&recv_msg.fd, CMSG_DATA(cm), sizeof(int) * 2);
+        }
     }
 
     recv_msg.incoming_buf = NULL;
@@ -852,6 +859,7 @@ nxt_unit_process_msg(nxt_unit_ctx_t *ctx, nxt_unit_read_buf_t *rbuf)
         if (nxt_slow_path(rc != NXT_UNIT_OK)) {
             if (rc == NXT_UNIT_AGAIN) {
                 recv_msg.fd = -1;
+                recv_msg.fd2 = -1;
             }
 
             goto fail;
@@ -871,6 +879,7 @@ nxt_unit_process_msg(nxt_unit_ctx_t *ctx, nxt_unit_read_buf_t *rbuf)
         if (nxt_slow_path(rc != NXT_UNIT_OK)) {
             if (rc == NXT_UNIT_AGAIN) {
                 recv_msg.fd = -1;
+                recv_msg.fd2 = -1;
             }
 
             goto fail;
@@ -960,6 +969,10 @@ fail:
         close(recv_msg.fd);
     }
 
+    if (recv_msg.fd2 != -1) {
+        close(recv_msg.fd2);
+    }
+
     while (recv_msg.incoming_buf != NULL) {
         nxt_unit_mmap_buf_free(recv_msg.incoming_buf);
     }