Merged with the default branch.1.19.0-1

1 files changed, 93 insertions, 99 deletions

diff --git a/src/nxt_port_memory.c b/src/nxt_port_memory.c
index f4d2125c..1e01629e 100644
--- a/src/nxt_port_memory.c
+++ b/src/nxt_port_memory.c
@@ -282,11 +282,10 @@ fail:
 
 
 static nxt_port_mmap_handler_t *
-nxt_port_new_port_mmap(nxt_task_t *task, nxt_process_t *process,
-    nxt_port_t *port, nxt_bool_t tracking, nxt_int_t n)
+nxt_port_new_port_mmap(nxt_task_t *task, nxt_port_mmaps_t *mmaps,
+    nxt_bool_t tracking, nxt_int_t n)
 {
     void                     *mem;
-    u_char                   *p, name[64];
     nxt_fd_t                 fd;
     nxt_int_t                i;
     nxt_free_map_t           *free_map;
@@ -296,24 +295,87 @@ nxt_port_new_port_mmap(nxt_task_t *task, nxt_process_t *process,
 
     mmap_handler = nxt_zalloc(sizeof(nxt_port_mmap_handler_t));
     if (nxt_slow_path(mmap_handler == NULL)) {
-        nxt_log(task, NXT_LOG_WARN, "failed to allocate mmap_handler");
+        nxt_alert(task, "failed to allocate mmap_handler");
 
         return NULL;
     }
 
-    port_mmap = nxt_port_mmap_at(&process->outgoing, process->outgoing.size);
+    port_mmap = nxt_port_mmap_at(mmaps, mmaps->size);
     if (nxt_slow_path(port_mmap == NULL)) {
-        nxt_log(task, NXT_LOG_WARN,
-                "failed to add port mmap to outgoing array");
+        nxt_alert(task, "failed to add port mmap to mmaps array");
 
         nxt_free(mmap_handler);
         return NULL;
     }
 
+    fd = nxt_shm_open(task, PORT_MMAP_SIZE);
+    if (nxt_slow_path(fd == -1)) {
+        goto remove_fail;
+    }
+
+    mem = nxt_mem_mmap(NULL, PORT_MMAP_SIZE, PROT_READ | PROT_WRITE,
+                       MAP_SHARED, fd, 0);
+
+    if (nxt_slow_path(mem == MAP_FAILED)) {
+        goto remove_fail;
+    }
+
+    mmap_handler->hdr = mem;
+    mmap_handler->fd = fd;
+    port_mmap->mmap_handler = mmap_handler;
+    nxt_port_mmap_handler_use(mmap_handler, 1);
+
+    /* Init segment header. */
+    hdr = mmap_handler->hdr;
+
+    nxt_memset(hdr->free_map, 0xFFU, sizeof(hdr->free_map));
+    nxt_memset(hdr->free_tracking_map, 0xFFU, sizeof(hdr->free_tracking_map));
+
+    hdr->id = mmaps->size - 1;
+    hdr->src_pid = nxt_pid;
+    hdr->sent_over = 0xFFFFu;
+
+    /* Mark first chunk as busy */
+    free_map = tracking ? hdr->free_tracking_map : hdr->free_map;
+
+    for (i = 0; i < n; i++) {
+        nxt_port_mmap_set_chunk_busy(free_map, i);
+    }
+
+    /* Mark as busy chunk followed the last available chunk. */
+    nxt_port_mmap_set_chunk_busy(hdr->free_map, PORT_MMAP_CHUNK_COUNT);
+    nxt_port_mmap_set_chunk_busy(hdr->free_tracking_map, PORT_MMAP_CHUNK_COUNT);
+
+    nxt_log(task, NXT_LOG_DEBUG, "new mmap #%D created for %PI -> ...",
+            hdr->id, nxt_pid);
+
+    return mmap_handler;
+
+remove_fail:
+
+    nxt_free(mmap_handler);
+
+    mmaps->size--;
+
+    return NULL;
+}
+
+
+nxt_int_t
+nxt_shm_open(nxt_task_t *task, size_t size)
+{
+    nxt_fd_t  fd;
+
+#if (NXT_HAVE_MEMFD_CREATE || NXT_HAVE_SHM_OPEN)
+
+    u_char    *p, name[64];
+
     p = nxt_sprintf(name, name + sizeof(name), NXT_SHM_PREFIX "unit.%PI.%uxD",
                     nxt_pid, nxt_random(&task->thread->random));
     *p = '\0';
 
+#endif
+
 #if (NXT_HAVE_MEMFD_CREATE)
 
     fd = syscall(SYS_memfd_create, name, MFD_CLOEXEC);
@@ -321,7 +383,7 @@ nxt_port_new_port_mmap(nxt_task_t *task, nxt_process_t *process,
     if (nxt_slow_path(fd == -1)) {
         nxt_alert(task, "memfd_create(%s) failed %E", name, nxt_errno);
 
-        goto remove_fail;
+        return -1;
     }
 
     nxt_debug(task, "memfd_create(%s): %FD", name, fd);
@@ -330,14 +392,14 @@ nxt_port_new_port_mmap(nxt_task_t *task, nxt_process_t *process,
 
     fd = shm_open(SHM_ANON, O_RDWR, S_IRUSR | S_IWUSR);
 
-    nxt_debug(task, "shm_open(SHM_ANON): %FD", fd);
-
     if (nxt_slow_path(fd == -1)) {
         nxt_alert(task, "shm_open(SHM_ANON) failed %E", nxt_errno);
 
-        goto remove_fail;
+        return -1;
     }
 
+    nxt_debug(task, "shm_open(SHM_ANON): %FD", fd);
+
 #elif (NXT_HAVE_SHM_OPEN)
 
     /* Just in case. */
@@ -345,14 +407,14 @@ nxt_port_new_port_mmap(nxt_task_t *task, nxt_process_t *process,
 
     fd = shm_open((char *) name, O_CREAT | O_EXCL | O_RDWR, S_IRUSR | S_IWUSR);
 
-    nxt_debug(task, "shm_open(%s): %FD", name, fd);
-
     if (nxt_slow_path(fd == -1)) {
         nxt_alert(task, "shm_open(%s) failed %E", name, nxt_errno);
 
-        goto remove_fail;
+        return -1;
     }
 
+    nxt_debug(task, "shm_open(%s): %FD", name, fd);
+
     if (nxt_slow_path(shm_unlink((char *) name) == -1)) {
         nxt_log(task, NXT_LOG_WARN, "shm_unlink(%s) failed %E", name,
                 nxt_errno);
@@ -364,94 +426,41 @@ nxt_port_new_port_mmap(nxt_task_t *task, nxt_process_t *process,
 
 #endif
 
-    if (nxt_slow_path(ftruncate(fd, PORT_MMAP_SIZE) == -1)) {
-        nxt_log(task, NXT_LOG_WARN, "ftruncate() failed %E", nxt_errno);
-
-        goto remove_fail;
-    }
-
-    mem = nxt_mem_mmap(NULL, PORT_MMAP_SIZE, PROT_READ | PROT_WRITE,
-                       MAP_SHARED, fd, 0);
+    if (nxt_slow_path(ftruncate(fd, size) == -1)) {
+        nxt_alert(task, "ftruncate() failed %E", nxt_errno);
 
-    if (nxt_slow_path(mem == MAP_FAILED)) {
-        goto remove_fail;
-    }
+        nxt_fd_close(fd);
 
-    mmap_handler->hdr = mem;
-    port_mmap->mmap_handler = mmap_handler;
-    nxt_port_mmap_handler_use(mmap_handler, 1);
-
-    /* Init segment header. */
-    hdr = mmap_handler->hdr;
-
-    nxt_memset(hdr->free_map, 0xFFU, sizeof(hdr->free_map));
-    nxt_memset(hdr->free_tracking_map, 0xFFU, sizeof(hdr->free_tracking_map));
-
-    hdr->id = process->outgoing.size - 1;
-    hdr->src_pid = nxt_pid;
-    hdr->dst_pid = process->pid;
-    hdr->sent_over = port->id;
-
-    /* Mark first chunk as busy */
-    free_map = tracking ? hdr->free_tracking_map : hdr->free_map;
-
-    for (i = 0; i < n; i++) {
-        nxt_port_mmap_set_chunk_busy(free_map, i);
+        return -1;
     }
 
-    /* Mark as busy chunk followed the last available chunk. */
-    nxt_port_mmap_set_chunk_busy(hdr->free_map, PORT_MMAP_CHUNK_COUNT);
-    nxt_port_mmap_set_chunk_busy(hdr->free_tracking_map, PORT_MMAP_CHUNK_COUNT);
-
-    nxt_debug(task, "send mmap fd %FD to process %PI", fd, port->pid);
-
-    /* TODO handle error */
-    (void) nxt_port_socket_write(task, port, NXT_PORT_MSG_MMAP, fd, 0, 0, NULL);
-
-    nxt_log(task, NXT_LOG_DEBUG, "new mmap #%D created for %PI -> %PI",
-            hdr->id, nxt_pid, process->pid);
-
-    return mmap_handler;
-
-remove_fail:
-
-    nxt_free(mmap_handler);
-
-    process->outgoing.size--;
-
-    return NULL;
+    return fd;
 }
 
 
 static nxt_port_mmap_handler_t *
-nxt_port_mmap_get(nxt_task_t *task, nxt_port_t *port, nxt_chunk_id_t *c,
+nxt_port_mmap_get(nxt_task_t *task, nxt_port_mmaps_t *mmaps, nxt_chunk_id_t *c,
     nxt_int_t n, nxt_bool_t tracking)
 {
     nxt_int_t                i, res, nchunks;
-    nxt_process_t            *process;
     nxt_free_map_t           *free_map;
     nxt_port_mmap_t          *port_mmap;
     nxt_port_mmap_t          *end_port_mmap;
     nxt_port_mmap_header_t   *hdr;
     nxt_port_mmap_handler_t  *mmap_handler;
 
-    process = port->process;
-    if (nxt_slow_path(process == NULL)) {
-        return NULL;
-    }
-
-    nxt_thread_mutex_lock(&process->outgoing.mutex);
+    nxt_thread_mutex_lock(&mmaps->mutex);
 
-    end_port_mmap = process->outgoing.elts + process->outgoing.size;
+    end_port_mmap = mmaps->elts + mmaps->size;
 
-    for (port_mmap = process->outgoing.elts;
+    for (port_mmap = mmaps->elts;
          port_mmap < end_port_mmap;
          port_mmap++)
     {
         mmap_handler = port_mmap->mmap_handler;
         hdr = mmap_handler->hdr;
 
-        if (hdr->sent_over != 0xFFFFu && hdr->sent_over != port->id) {
+        if (hdr->sent_over != 0xFFFFu) {
             continue;
         }
 
@@ -489,11 +498,11 @@ nxt_port_mmap_get(nxt_task_t *task, nxt_port_t *port, nxt_chunk_id_t *c,
     /* TODO introduce port_mmap limit and release wait. */
 
     *c = 0;
-    mmap_handler = nxt_port_new_port_mmap(task, process, port, tracking, n);
+    mmap_handler = nxt_port_new_port_mmap(task, mmaps, tracking, n);
 
 unlock_return:
 
-    nxt_thread_mutex_unlock(&process->outgoing.mutex);
+    nxt_thread_mutex_unlock(&mmaps->mutex);
 
     return mmap_handler;
 }
@@ -528,7 +537,7 @@ nxt_port_get_port_incoming_mmap(nxt_task_t *task, nxt_pid_t spid, uint32_t id)
 
 
 nxt_int_t
-nxt_port_mmap_get_tracking(nxt_task_t *task, nxt_port_t *port,
+nxt_port_mmap_get_tracking(nxt_task_t *task, nxt_port_mmaps_t *mmaps,
     nxt_port_mmap_tracking_t *tracking, uint32_t stream)
 {
     nxt_chunk_id_t           c;
@@ -537,7 +546,7 @@ nxt_port_mmap_get_tracking(nxt_task_t *task, nxt_port_t *port,
 
     nxt_debug(task, "request tracking for stream #%uD", stream);
 
-    mmap_handler = nxt_port_mmap_get(task, port, &c, 1, 1);
+    mmap_handler = nxt_port_mmap_get(task, mmaps, &c, 1, 1);
     if (nxt_slow_path(mmap_handler == NULL)) {
         return NXT_ERROR;
     }
@@ -659,7 +668,7 @@ nxt_port_mmap_tracking_read(nxt_task_t *task, nxt_port_recv_msg_t *msg)
 
 
 nxt_buf_t *
-nxt_port_mmap_get_buf(nxt_task_t *task, nxt_port_t *port, size_t size)
+nxt_port_mmap_get_buf(nxt_task_t *task, nxt_port_mmaps_t *mmaps, size_t size)
 {
     nxt_mp_t                 *mp;
     nxt_buf_t                *b;
@@ -686,7 +695,7 @@ nxt_port_mmap_get_buf(nxt_task_t *task, nxt_port_t *port, size_t size)
     b->completion_handler = nxt_port_mmap_buf_completion;
     nxt_buf_set_port_mmap(b);
 
-    mmap_handler = nxt_port_mmap_get(task, port, &c, nchunks, 0);
+    mmap_handler = nxt_port_mmap_get(task, mmaps, &c, nchunks, 0);
     if (nxt_slow_path(mmap_handler == NULL)) {
         mp = task->thread->engine->mem_pool;
         nxt_mp_free(mp, b);
@@ -922,9 +931,7 @@ nxt_port_mmap_read(nxt_task_t *task, nxt_port_recv_msg_t *msg)
 nxt_port_method_t
 nxt_port_mmap_get_method(nxt_task_t *task, nxt_port_t *port, nxt_buf_t *b)
 {
-    nxt_port_method_t        m;
-    nxt_port_mmap_header_t   *hdr;
-    nxt_port_mmap_handler_t  *mmap_handler;
+    nxt_port_method_t  m;
 
     m = NXT_PORT_METHOD_ANY;
 
@@ -935,9 +942,6 @@ nxt_port_mmap_get_method(nxt_task_t *task, nxt_port_t *port, nxt_buf_t *b)
         }
 
         if (nxt_buf_is_port_mmap(b)) {
-            mmap_handler = b->parent;
-            hdr = mmap_handler->hdr;
-
             if (m == NXT_PORT_METHOD_PLAIN) {
                 nxt_log_error(NXT_LOG_ERR, task->log,
                               "mixing plain and mmap buffers, "
@@ -946,16 +950,6 @@ nxt_port_mmap_get_method(nxt_task_t *task, nxt_port_t *port, nxt_buf_t *b)
                 break;
             }
 
-            if (port->pid != hdr->dst_pid) {
-                nxt_log_error(NXT_LOG_ERR, task->log,
-                              "send mmap buffer for %PI to %PI, "
-                              "using plain mode", hdr->dst_pid, port->pid);
-
-                m = NXT_PORT_METHOD_PLAIN;
-
-                break;
-            }
-
             if (m == NXT_PORT_METHOD_ANY) {
                 nxt_debug(task, "using mmap mode");