technocloud-public/nginx-unit
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Port message extended to transfer 2 file descriptors.

Browse Source
This commit is contained in:
Max Romanov
2020-08-11 19:20:30 +03:00
parent 72475ee11c
commit a1e9df2aef
5 changed files with 81 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/nxt_port.h
View File

@@ -165,6 +165,7 @@ typedef struct {
nxt_buf_t *buf; nxt_buf_t *buf;
size_t share; size_t share;
nxt_fd_t fd; nxt_fd_t fd;
nxt_fd_t fd2;
nxt_port_msg_t port_msg; nxt_port_msg_t port_msg;
uint32_t tracking_msg[2]; uint32_t tracking_msg[2];
uint8_t close_fd; /* 1 bit */ uint8_t close_fd; /* 1 bit */
@@ -174,6 +175,7 @@ typedef struct {
struct nxt_port_recv_msg_s { struct nxt_port_recv_msg_s {
nxt_fd_t fd; nxt_fd_t fd;
nxt_fd_t fd2;
nxt_buf_t *buf; nxt_buf_t *buf;
nxt_port_t *port; nxt_port_t *port;
nxt_port_msg_t port_msg; nxt_port_msg_t port_msg;

26
src/nxt_port_socket.c
View File

@@ -156,6 +156,7 @@ nxt_port_socket_twrite(nxt_task_t *task, nxt_port_t *port, nxt_uint_t type,
msg.buf = b; msg.buf = b;
msg.share = 0; msg.share = 0;
msg.fd = fd; msg.fd = fd;
msg.fd2 = -1;
msg.close_fd = (type & NXT_PORT_MSG_CLOSE_FD) != 0; msg.close_fd = (type & NXT_PORT_MSG_CLOSE_FD) != 0;
msg.allocated = 0; msg.allocated = 0;
@@ -331,7 +332,7 @@ next_fragment:
msg->port_msg.last |= sb.last; msg->port_msg.last |= sb.last;
msg->port_msg.mf = sb.limit_reached || sb.nmax_reached; 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 (n > 0) {
if (nxt_slow_path((size_t) n != sb.size + iov[0].iov_len)) { if (nxt_slow_path((size_t) n != sb.size + iov[0].iov_len)) {
@@ -346,6 +347,12 @@ next_fragment:
msg->fd = -1; 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, msg->buf = nxt_port_buf_completion(task, wq, msg->buf, plain_size,
m == NXT_PORT_METHOD_MMAP); m == NXT_PORT_METHOD_MMAP);
@@ -358,6 +365,7 @@ next_fragment:
* in the first message of a stream. * in the first message of a stream.
*/ */
msg->fd = -1; msg->fd = -1;
msg->fd2 = -1;
msg->share += n; msg->share += n;
msg->port_msg.nf = 1; msg->port_msg.nf = 1;
msg->port_msg.tracking = 0; 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); nxt_fd_close(msg->fd);
} }
if (msg->fd2 != -1) {
nxt_fd_close(msg->fd2);
}
return; return;
} }
@@ -854,6 +866,7 @@ nxt_port_read_msg_process(nxt_task_t *task, nxt_port_t *port,
msg->buf = fmsg->buf; msg->buf = fmsg->buf;
msg->fd = fmsg->fd; msg->fd = fmsg->fd;
msg->fd2 = fmsg->fd2;
/* /*
* To disable instant completion or buffer re-usage, * 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)) { if (nxt_fast_path(msg->cancelled == 0)) {
msg->buf = NULL; msg->buf = NULL;
msg->fd = -1; msg->fd = -1;
msg->fd2 = -1;
b = NULL; b = NULL;
} else { } else {
if (msg->fd != -1) { if (msg->fd != -1) {
nxt_fd_close(msg->fd); nxt_fd_close(msg->fd);
} }
if (msg->fd2 != -1) {
nxt_fd_close(msg->fd2);
}
} }
} else { } else {
if (nxt_fast_path(msg->cancelled == 0)) { 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; 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) { for (b = msg->buf; b != NULL; b = next) {
next = b->next; next = b->next;
b->next = NULL; b->next = NULL;

2
src/nxt_socket.h
View File

@@ -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_EXPORT nxt_int_t nxt_socketpair_create(nxt_task_t *task,
nxt_socket_t *pair); nxt_socket_t *pair);
NXT_EXPORT void nxt_socketpair_close(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_iobuf_t *iob, nxt_uint_t niob);
NXT_EXPORT ssize_t nxt_socketpair_recv(nxt_fd_event_t *ev, nxt_fd_t *fd, NXT_EXPORT ssize_t nxt_socketpair_recv(nxt_fd_event_t *ev, nxt_fd_t *fd,
nxt_iobuf_t *iob, nxt_uint_t niob); nxt_iobuf_t *iob, nxt_uint_t niob);

59
src/nxt_socketpair.c
View File

@@ -20,7 +20,7 @@
#endif #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); nxt_uint_t niob);
static ssize_t nxt_recvmsg(nxt_socket_t s, nxt_fd_t *fd, nxt_iobuf_t *iob, static ssize_t nxt_recvmsg(nxt_socket_t s, nxt_fd_t *fd, nxt_iobuf_t *iob,
nxt_uint_t niob); nxt_uint_t niob);
@@ -71,7 +71,7 @@ nxt_socketpair_close(nxt_task_t *task, nxt_socket_t *pair)
ssize_t 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) nxt_uint_t niob)
{ {
ssize_t n; 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; 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) { if (n > 0) {
return n; return n;
@@ -108,8 +109,8 @@ nxt_socketpair_send(nxt_fd_event_t *ev, nxt_fd_t fd, nxt_iobuf_t *iob,
continue; continue;
default: default:
nxt_alert(ev->task, "sendmsg(%d, %FD, %ui) failed %E", nxt_alert(ev->task, "sendmsg(%d, %FD, %FD, %ui) failed %E",
ev->fd, fd, niob, err); ev->fd, fd[0], fd[1], niob, err);
return NXT_ERROR; 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; 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) { if (n > 0) {
return n; 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 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; struct msghdr msg;
union { union {
struct cmsghdr cm; struct cmsghdr cm;
char space[CMSG_SPACE(sizeof(int))]; char space[CMSG_SPACE(sizeof(int) * 2)];
} cmsg; } cmsg;
msg.msg_name = NULL; 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. */ /* Flags are cleared just to suppress valgrind warning. */
msg.msg_flags = 0; 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_control = (caddr_t) &cmsg;
msg.msg_controllen = sizeof(cmsg); msg.msg_controllen = CMSG_SPACE(csize);
#if (NXT_VALGRIND) #if (NXT_VALGRIND)
nxt_memzero(&cmsg, sizeof(cmsg)); nxt_memzero(&cmsg, sizeof(cmsg));
#endif #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_level = SOL_SOCKET;
cmsg.cm.cmsg_type = SCM_RIGHTS; 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() * Fortunately, GCC with -O1 compiles this nxt_memcpy()
* in the same simple assignment as in the code above. * 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 { } else {
msg.msg_control = NULL; 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; struct msghdr msg;
union { union {
struct cmsghdr cm; struct cmsghdr cm;
char space[CMSG_SPACE(sizeof(int))]; char space[CMSG_SPACE(sizeof(int) * 2)];
} cmsg; } cmsg;
msg.msg_name = NULL; 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_control = (caddr_t) &cmsg;
msg.msg_controllen = sizeof(cmsg); msg.msg_controllen = sizeof(cmsg);
*fd = -1; fd[0] = -1;
fd[1] = -1;
#if (NXT_VALGRIND) #if (NXT_VALGRIND)
nxt_memzero(&cmsg, sizeof(cmsg)); 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); n = recvmsg(s, &msg, 0);
if (n > 0 if (n > 0
&& cmsg.cm.cmsg_len == CMSG_LEN(sizeof(int))
&& cmsg.cm.cmsg_level == SOL_SOCKET && cmsg.cm.cmsg_level == SOL_SOCKET
&& cmsg.cm.cmsg_type == SCM_RIGHTS) && cmsg.cm.cmsg_type == SCM_RIGHTS)
{ {
/* (*fd) = *(int *) CMSG_DATA(&cmsg.cm); */ if (cmsg.cm.cmsg_len == CMSG_LEN(sizeof(int))) {
nxt_memcpy(fd, CMSG_DATA(&cmsg.cm), 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; 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. */ /* Solaris 4.3BSD sockets. */
static ssize_t 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; 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_iov = iob;
msg.msg_iovlen = niob; msg.msg_iovlen = niob;
if (fd != -1) { if (fd[0] != -1) {
msg.msg_accrights = (caddr_t) &fd; msg.msg_accrights = (caddr_t) fd;
msg.msg_accrightslen = sizeof(int); msg.msg_accrightslen = sizeof(int);
if (fd[1] != -1) {
msg.msg_accrightslen += sizeof(int);
}
} else { } else {
msg.msg_accrights = NULL; msg.msg_accrights = NULL;
msg.msg_accrightslen = 0; 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; struct msghdr msg;
*fd = -1; fd[0] = -1;
fd[1] = -1;
msg.msg_name = NULL; msg.msg_name = NULL;
msg.msg_namelen = 0; msg.msg_namelen = 0;
msg.msg_iov = iob; msg.msg_iov = iob;
msg.msg_iovlen = niob; msg.msg_iovlen = niob;
msg.msg_accrights = (caddr_t) fd; msg.msg_accrights = (caddr_t) fd;
msg.msg_accrightslen = sizeof(int); msg.msg_accrightslen = sizeof(int) * 2;
return recvmsg(s, &msg, 0); return recvmsg(s, &msg, 0);
} }

19
src/nxt_unit.c
View File

@@ -192,6 +192,7 @@ struct nxt_unit_recv_msg_s {
uint32_t size; uint32_t size;
int fd; int fd;
int fd2;
nxt_unit_mmap_buf_t *incoming_buf; 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; rc = NXT_UNIT_ERROR;
recv_msg.fd = -1; recv_msg.fd = -1;
recv_msg.fd2 = -1;
port_msg = (nxt_port_msg_t *) rbuf->buf; port_msg = (nxt_port_msg_t *) rbuf->buf;
cm = (struct cmsghdr *) rbuf->oob; cm = (struct cmsghdr *) rbuf->oob;
if (cm->cmsg_len == CMSG_LEN(sizeof(int)) if (cm->cmsg_level == SOL_SOCKET
&& cm->cmsg_level == SOL_SOCKET
&& cm->cmsg_type == SCM_RIGHTS) && 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; 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 (nxt_slow_path(rc != NXT_UNIT_OK)) {
if (rc == NXT_UNIT_AGAIN) { if (rc == NXT_UNIT_AGAIN) {
recv_msg.fd = -1; recv_msg.fd = -1;
recv_msg.fd2 = -1;
} }
goto fail; 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 (nxt_slow_path(rc != NXT_UNIT_OK)) {
if (rc == NXT_UNIT_AGAIN) { if (rc == NXT_UNIT_AGAIN) {
recv_msg.fd = -1; recv_msg.fd = -1;
recv_msg.fd2 = -1;
} }
goto fail; goto fail;
@@ -960,6 +969,10 @@ fail:
close(recv_msg.fd); close(recv_msg.fd);
} }
if (recv_msg.fd2 != -1) {
close(recv_msg.fd2);
}
while (recv_msg.incoming_buf != NULL) { while (recv_msg.incoming_buf != NULL) {
nxt_unit_mmap_buf_free(recv_msg.incoming_buf); nxt_unit_mmap_buf_free(recv_msg.incoming_buf);
} }