Redirecting buffer completion handler to specific engine.

There is a case in router where we use port in router connection thread.
Buffers are allocated within connection memory pool which can be used only in
this router thread. sendmsg() can be postponed into main router thread and
completion handler will compare current engine and post itself to correct
engine.

9 files changed, 236 insertions, 79 deletions

diff --git a/src/nxt_buf.c b/src/nxt_buf.c
index cb0faf7c..826cd017 100644
--- a/src/nxt_buf.c
+++ b/src/nxt_buf.c
@@ -8,13 +8,18 @@
 
 
 static void nxt_buf_completion(nxt_task_t *task, void *obj, void *data);
+static void nxt_buf_ts_completion(nxt_task_t *task, void *obj, void *data);
+
+
+typedef struct {
+    nxt_work_t          work;
+    nxt_event_engine_t  *engine;
+} nxt_buf_ts_t;
 
 
 void
 nxt_buf_mem_init(nxt_buf_t *b, void *start, size_t size)
 {
-    b->size = NXT_BUF_MEM_SIZE;
-
     b->mem.start = start;
     b->mem.pos = start;
     b->mem.free = start;
@@ -27,26 +32,60 @@ nxt_buf_mem_alloc(nxt_mp_t *mp, size_t size, nxt_uint_t flags)
 {
     nxt_buf_t  *b;
 
-    b = nxt_mp_zalloc(mp, NXT_BUF_MEM_SIZE);
+    b = nxt_mp_alloc(mp, NXT_BUF_MEM_SIZE + size);
     if (nxt_slow_path(b == NULL)) {
         return NULL;
     }
 
+    nxt_memzero(b, NXT_BUF_MEM_SIZE);
+
     b->data = mp;
     b->completion_handler = nxt_buf_completion;
-    b->size = NXT_BUF_MEM_SIZE;
 
     if (size != 0) {
-        b->mem.start = nxt_mp_alloc(mp, size);
-        if (nxt_slow_path(b->mem.start == NULL)) {
-            return NULL;
-        }
+        b->mem.start = nxt_pointer_to(b, NXT_BUF_MEM_SIZE);
+        b->mem.pos = b->mem.start;
+        b->mem.free = b->mem.start;
+        b->mem.end = b->mem.start + size;
+    }
+
+    return b;
+}
+
+
+nxt_buf_t *
+nxt_buf_mem_ts_alloc(nxt_task_t *task, nxt_mp_t *mp, size_t size)
+{
+    nxt_buf_t     *b;
+    nxt_buf_ts_t  *ts;
+
+    b = nxt_mp_retain(mp, NXT_BUF_MEM_SIZE + sizeof(nxt_buf_ts_t) + size);
+    if (nxt_slow_path(b == NULL)) {
+        return NULL;
+    }
+
+    nxt_memzero(b, NXT_BUF_MEM_SIZE + sizeof(nxt_buf_ts_t));
 
+    b->data = mp;
+    b->completion_handler = nxt_buf_ts_completion;
+    b->is_ts = 1;
+
+    if (size != 0) {
+        b->mem.start = nxt_pointer_to(b, NXT_BUF_MEM_SIZE +
+                                      sizeof(nxt_buf_ts_t));
         b->mem.pos = b->mem.start;
         b->mem.free = b->mem.start;
         b->mem.end = b->mem.start + size;
     }
 
+    ts = nxt_pointer_to(b, NXT_BUF_MEM_SIZE);
+    ts->engine = task->thread->engine;
+
+    ts->work.handler = nxt_buf_ts_completion;
+    ts->work.task = task;
+    ts->work.obj = b;
+    ts->work.data = b->parent;
+
     return b;
 }
 
@@ -56,22 +95,19 @@ nxt_buf_file_alloc(nxt_mp_t *mp, size_t size, nxt_uint_t flags)
 {
     nxt_buf_t  *b;
 
-    b = nxt_mp_zalloc(mp, NXT_BUF_FILE_SIZE);
+    b = nxt_mp_alloc(mp, NXT_BUF_FILE_SIZE + size);
     if (nxt_slow_path(b == NULL)) {
         return NULL;
     }
 
+    nxt_memzero(b, NXT_BUF_FILE_SIZE);
+
     b->data = mp;
     b->completion_handler = nxt_buf_completion;
-    b->size = NXT_BUF_FILE_SIZE;
     nxt_buf_set_file(b);
 
     if (size != 0) {
-        b->mem.start = nxt_mp_alloc(mp, size);
-        if (nxt_slow_path(b->mem.start == NULL)) {
-            return NULL;
-        }
-
+        b->mem.start = nxt_pointer_to(b, NXT_BUF_FILE_SIZE);
         b->mem.pos = b->mem.start;
         b->mem.free = b->mem.start;
         b->mem.end = b->mem.start + size;
@@ -91,7 +127,6 @@ nxt_buf_mmap_alloc(nxt_mp_t *mp, size_t size)
     if (nxt_fast_path(b != NULL)) {
         b->data = mp;
         b->completion_handler = nxt_buf_completion;
-        b->size = NXT_BUF_MMAP_SIZE;
 
         nxt_buf_set_file(b);
         nxt_buf_set_mmap(b);
@@ -112,7 +147,6 @@ nxt_buf_sync_alloc(nxt_mp_t *mp, nxt_uint_t flags)
     if (nxt_fast_path(b != NULL)) {
         b->data = mp;
         b->completion_handler = nxt_buf_completion;
-        b->size = NXT_BUF_SYNC_SIZE;
 
         nxt_buf_set_sync(b);
         b->is_nobuf = ((flags & NXT_BUF_SYNC_NOBUF) != 0);
@@ -166,8 +200,91 @@ nxt_buf_completion(nxt_task_t *task, void *obj, void *data)
 
     nxt_debug(task, "buf completion: %p %p", b, b->mem.start);
 
+#if (NXT_DEBUG)
+    if (nxt_slow_path(data != b->parent)) {
+        nxt_log_alert(task->log, "completion data (%p) != b->parent (%p)",
+                      data, b->parent);
+        nxt_abort();
+    }
+#endif
+
+    mp = b->data;
+    nxt_mp_free(mp, b);
+
+    if (parent != NULL) {
+        nxt_debug(task, "parent retain:%uD", parent->retain);
+
+        parent->retain--;
+
+        if (parent->retain == 0) {
+            parent->mem.pos = parent->mem.free;
+
+            parent->completion_handler(task, parent, parent->parent);
+        }
+    }
+}
+
+
+nxt_int_t
+nxt_buf_ts_handle(nxt_task_t *task, void *obj, void *data)
+{
+    nxt_buf_t     *b;
+    nxt_buf_ts_t  *ts;
+
+    b = obj;
+
+#if (NXT_DEBUG)
+    if (nxt_slow_path(b->is_ts == 0)) {
+        nxt_log_alert(task->log, "not a thread safe buf (%p) completed", b);
+        nxt_abort();
+    }
+#endif
+
+    ts = nxt_pointer_to(b, NXT_BUF_MEM_SIZE);
+
+    if (ts->engine != task->thread->engine) {
+
+        nxt_debug(task, "buf ts: %p current engine is %p, expected %p",
+                  b, task->thread->engine, ts->engine);
+
+        ts->work.handler = b->completion_handler;
+        ts->work.obj = obj;
+        ts->work.data = data;
+
+        nxt_event_engine_post(ts->engine, &ts->work);
+
+        return 1;
+    }
+
+    return 0;
+}
+
+
+static void
+nxt_buf_ts_completion(nxt_task_t *task, void *obj, void *data)
+{
+    nxt_mp_t   *mp;
+    nxt_buf_t  *b, *parent;
+
+    b = obj;
+    parent = data;
+
+    if (nxt_buf_ts_handle(task, obj, data)) {
+        return;
+    }
+
+    nxt_debug(task, "buf ts completion: %p %p", b, b->mem.start);
+
+#if (NXT_DEBUG)
+    if (nxt_slow_path(data != b->parent)) {
+        nxt_log_alert(task->log, "completion data (%p) != b->parent (%p)",
+                      data, b->parent);
+        nxt_abort();
+    }
+#endif
+
     mp = b->data;
-    nxt_buf_free(mp, b);
+    nxt_mp_release(mp, b);
 
     if (parent != NULL) {
         nxt_debug(task, "parent retain:%uD", parent->retain);
diff --git a/src/nxt_buf.h b/src/nxt_buf.h
index 43c0c453..ae96b24a 100644
--- a/src/nxt_buf.h
+++ b/src/nxt_buf.h
@@ -76,11 +76,6 @@ struct nxt_buf_s {
     nxt_buf_t               *next;
 
     uint32_t                retain;
-    /*
-     * Used by nxt_mem_cache_free() to return buffer
-     * in appropriate memory pool cache.
-     */
-    uint8_t                 size;
 
     uint8_t                 is_file;     /* 1 bit */
 
@@ -92,6 +87,7 @@ struct nxt_buf_s {
     uint16_t                is_flush:1;
     uint16_t                is_last:1;
     uint16_t                is_port_mmap_sent:1;
+    uint16_t                is_ts:1;
 
     nxt_buf_mem_t           mem;
 
@@ -245,11 +241,14 @@ nxt_buf_used_size(b)                                                          \
 NXT_EXPORT void nxt_buf_mem_init(nxt_buf_t *b, void *start, size_t size);
 NXT_EXPORT nxt_buf_t *nxt_buf_mem_alloc(nxt_mp_t *mp, size_t size,
     nxt_uint_t flags);
+NXT_EXPORT nxt_buf_t *nxt_buf_mem_ts_alloc(nxt_task_t *task, nxt_mp_t *mp,
+    size_t size);
 NXT_EXPORT nxt_buf_t *nxt_buf_file_alloc(nxt_mp_t *mp, size_t size,
     nxt_uint_t flags);
 NXT_EXPORT nxt_buf_t *nxt_buf_mmap_alloc(nxt_mp_t *mp, size_t size);
 NXT_EXPORT nxt_buf_t *nxt_buf_sync_alloc(nxt_mp_t *mp, nxt_uint_t flags);
 
+NXT_EXPORT nxt_int_t nxt_buf_ts_handle(nxt_task_t *task, void *obj, void *data);
 
 #define                                                                       \
 nxt_buf_free(mp, b)                                                           \
diff --git a/src/nxt_buf_pool.c b/src/nxt_buf_pool.c
index 7ef9def5..8259c60a 100644
--- a/src/nxt_buf_pool.c
+++ b/src/nxt_buf_pool.c
@@ -147,10 +147,6 @@ nxt_buf_pool_free(nxt_buf_pool_t *bp, nxt_buf_t *b)
             bp->current = NULL;
         }
 
-        if (!bp->mmap) {
-            nxt_mp_free(bp->mem_pool, b->mem.start);
-        }
-
         nxt_buf_free(bp->mem_pool, b);
 
         return;
@@ -176,15 +172,12 @@ nxt_buf_pool_free(nxt_buf_pool_t *bp, nxt_buf_t *b)
 void
 nxt_buf_pool_destroy(nxt_buf_pool_t *bp)
 {
-    u_char     *p;
     nxt_buf_t  *b;
 
     bp->destroy = 1;
 
     for (b = bp->free; b != NULL; b = b->next) {
-        p = b->mem.start;
         nxt_buf_free(bp->mem_pool, b);
-        nxt_mp_free(bp->mem_pool, p);
     }
 
     bp->free = b; /* NULL */
diff --git a/src/nxt_log.h b/src/nxt_log.h
index e1b69442..74260576 100644
--- a/src/nxt_log.h
+++ b/src/nxt_log.h
@@ -122,7 +122,8 @@ nxt_log_debug(_log, ...)                                                      \
     } while (0)
 
 
-#define nxt_debug_alloc(...)
+#define nxt_debug_alloc(_fmt, ARGS...)                                        \
+    nxt_thread_log_debug("%08p: " _fmt, mp, ##ARGS)
 
 #else
 
diff --git a/src/nxt_port.c b/src/nxt_port.c
index 91f0fe31..249a287c 100644
--- a/src/nxt_port.c
+++ b/src/nxt_port.c
@@ -10,8 +10,6 @@
 
 
 static void nxt_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg);
-static void nxt_port_new_port_buf_completion(nxt_task_t *task, void *obj,
-    void *data);
 
 
 void
@@ -102,13 +100,10 @@ nxt_port_send_new_port(nxt_task_t *task, nxt_runtime_t *rt,
 nxt_int_t
 nxt_port_send_port(nxt_task_t *task, nxt_port_t *port, nxt_port_t *new_port)
 {
-    nxt_mp_t                 *mp;
     nxt_buf_t                *b;
     nxt_port_msg_new_port_t  *msg;
 
-    mp = port->mem_pool;
-
-    b = nxt_buf_mem_alloc(mp, sizeof(nxt_port_data_t), 0);
+    b = nxt_buf_mem_ts_alloc(task, port->mem_pool, sizeof(nxt_port_data_t));
     if (nxt_slow_path(b == NULL)) {
         return NXT_ERROR;
     }
@@ -116,8 +111,6 @@ nxt_port_send_port(nxt_task_t *task, nxt_port_t *port, nxt_port_t *new_port)
     nxt_debug(task, "send port %FD to process %PI",
               new_port->pair[1], port->pid);
 
-    b->data = mp;
-    b->completion_handler = nxt_port_new_port_buf_completion;
     b->mem.free += sizeof(nxt_port_msg_new_port_t);
     msg = (nxt_port_msg_new_port_t *) b->mem.pos;
 
@@ -133,19 +126,6 @@ nxt_port_send_port(nxt_task_t *task, nxt_port_t *port, nxt_port_t *new_port)
 }
 
 
-static void
-nxt_port_new_port_buf_completion(nxt_task_t *task, void *obj, void *data)
-{
-    nxt_mp_t   *mp;
-    nxt_buf_t  *b;
-
-    b = obj;
-    mp = b->data;
-
-    nxt_buf_free(mp, b);
-}
-
-
 void
 nxt_port_new_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
 {
diff --git a/src/nxt_port.h b/src/nxt_port.h
index 66534cdc..927e0e2e 100644
--- a/src/nxt_port.h
+++ b/src/nxt_port.h
@@ -36,10 +36,13 @@ typedef struct {
 typedef struct {
     nxt_queue_link_t    link;
     nxt_buf_t           *buf;
-    nxt_mp_t            *mem_pool;
     size_t              share;
     nxt_fd_t            fd;
     nxt_port_msg_t      port_msg;
+
+    nxt_work_t          work;
+    nxt_event_engine_t  *engine;
+    nxt_mp_t            *mem_pool;
 } nxt_port_send_msg_t;
 
 
diff --git a/src/nxt_port_memory.c b/src/nxt_port_memory.c
index ba39c9c1..cf69a758 100644
--- a/src/nxt_port_memory.c
+++ b/src/nxt_port_memory.c
@@ -35,6 +35,10 @@ nxt_port_mmap_buf_completion(nxt_task_t *task, void *obj, void *data)
     nxt_chunk_id_t          c;
     nxt_port_mmap_header_t  *hdr;
 
+    if (nxt_buf_ts_handle(task, obj, data)) {
+        return;
+    }
+
     b = obj;
 
     nxt_debug(task, "mmap buf completion: %p %p", b, b->mem.start);
@@ -71,7 +75,7 @@ nxt_port_mmap_buf_completion(nxt_task_t *task, void *obj, void *data)
         c++;
     }
 
-    nxt_buf_free(mp, b);
+    nxt_mp_release(mp, b);
 }
 
 
@@ -150,6 +154,10 @@ nxt_port_mmap_send_fd_buf_completion(nxt_task_t *task, void *obj, void *data)
     nxt_buf_t  *b;
     nxt_mp_t   *mp;
 
+    if (nxt_buf_ts_handle(task, obj, data)) {
+        return;
+    }
+
     b = obj;
     mp = b->data;
     fd = (nxt_fd_t) (intptr_t) data;
@@ -166,7 +174,7 @@ nxt_port_mmap_send_fd_buf_completion(nxt_task_t *task, void *obj, void *data)
 
     nxt_fd_close(fd);
 
-    nxt_buf_free(mp, b);
+    nxt_mp_release(mp, b);
 }
 
 
@@ -252,13 +260,12 @@ nxt_port_new_port_mmap(nxt_task_t *task, nxt_process_t *process,
 
     port_mmap->hdr = mem;
 
-    b = nxt_buf_mem_alloc(port->mem_pool, 0, 0);
+    b = nxt_buf_mem_ts_alloc(task, port->mem_pool, 0);
     if (nxt_slow_path(b == NULL)) {
         goto remove_fail;
     }
 
     b->completion_handler = nxt_port_mmap_send_fd_buf_completion;
-    b->data = port->mem_pool;
     b->parent = (void *) (intptr_t) fd;
 
     /* Init segment header. */
@@ -396,24 +403,22 @@ nxt_port_mmap_get_buf(nxt_task_t *task, nxt_port_t *port, size_t size)
         return NULL;
     }
 
-    b = nxt_mp_zalloc(port->mem_pool, NXT_BUF_PORT_MMAP_SIZE);
+    b = nxt_buf_mem_ts_alloc(task, port->mem_pool, 0);
     if (nxt_slow_path(b == NULL)) {
         return NULL;
     }
 
-    b->data = port->mem_pool;
     b->completion_handler = nxt_port_mmap_buf_completion;
-    b->size = NXT_BUF_PORT_MMAP_SIZE;
-
     nxt_buf_set_port_mmap(b);
 
     hdr = nxt_port_mmap_get(task, port, &c, size);
     if (nxt_slow_path(hdr == NULL)) {
-        nxt_buf_free(port->mem_pool, b);
+        nxt_mp_release(port->mem_pool, b);
         return NULL;
     }
 
     b->parent = hdr;
+
     b->mem.start = nxt_port_mmap_chunk_start(hdr, c);
     b->mem.pos = b->mem.start;
     b->mem.free = b->mem.start;
@@ -517,14 +522,12 @@ nxt_port_mmap_get_incoming_buf(nxt_task_t *task, nxt_port_t *port,
         return NULL;
     }
 
-    b = nxt_mp_zalloc(port->mem_pool, NXT_BUF_PORT_MMAP_SIZE);
+    b = nxt_buf_mem_ts_alloc(task, port->mem_pool, 0);
     if (nxt_slow_path(b == NULL)) {
         return NULL;
     }
 
-    b->data = port->mem_pool;
     b->completion_handler = nxt_port_mmap_buf_completion;
-    b->size = NXT_BUF_PORT_MMAP_SIZE;
 
     nxt_buf_set_port_mmap(b);
 
@@ -559,7 +562,7 @@ nxt_port_mmap_write(nxt_task_t *task, nxt_port_t *port,
 
     bsize = sb->niov * sizeof(nxt_port_mmap_msg_t);
 
-    b = nxt_buf_mem_alloc(port->mem_pool, bsize, 0);
+    b = nxt_buf_mem_ts_alloc(task, port->mem_pool, bsize);
     if (nxt_slow_path(b == NULL)) {
         return;
     }
diff --git a/src/nxt_port_socket.c b/src/nxt_port_socket.c
index 3bf57479..73bb44bd 100644
--- a/src/nxt_port_socket.c
+++ b/src/nxt_port_socket.c
@@ -135,6 +135,37 @@ nxt_port_write_close(nxt_port_t *port)
 }
 
 
+static void
+nxt_port_release_send_msg(nxt_task_t *task, void *obj, void *data)
+{
+    nxt_event_engine_t   *engine;
+    nxt_port_send_msg_t  *msg;
+
+    msg = obj;
+    engine = data;
+
+#if (NXT_DEBUG)
+    if (nxt_slow_path(data != msg->engine)) {
+        nxt_log_alert(task->log, "release msg data (%p) != msg->engine (%p)",
+                      data, msg->engine);
+        nxt_abort();
+    }
+#endif
+
+    if (engine != task->thread->engine) {
+
+        nxt_debug(task, "current thread is %PT, expected %PT",
+                  task->thread->tid, engine->task.thread->tid);
+
+        nxt_event_engine_post(engine, &msg->work);
+
+        return;
+    }
+
+    nxt_mp_release(msg->mem_pool, obj);
+}
+
+
 nxt_int_t
 nxt_port_socket_write(nxt_task_t *task, nxt_port_t *port, nxt_uint_t type,
     nxt_fd_t fd, uint32_t stream, nxt_port_id_t reply_port, nxt_buf_t *b)
@@ -160,14 +191,25 @@ nxt_port_socket_write(nxt_task_t *task, nxt_port_t *port, nxt_uint_t type,
         }
     }
 
-    msg = nxt_mp_zalloc(port->mem_pool, sizeof(nxt_port_send_msg_t));
+    msg = nxt_mp_retain(port->mem_pool, sizeof(nxt_port_send_msg_t));
     if (nxt_slow_path(msg == NULL)) {
         return NXT_ERROR;
     }
 
+    msg->link.next = NULL;
+    msg->link.prev = NULL;
+
     msg->buf = b;
     msg->fd = fd;
     msg->share = 0;
+
+    msg->work.next = NULL;
+    msg->work.handler = nxt_port_release_send_msg;
+    msg->work.task = task;
+    msg->work.obj = msg;
+    msg->work.data = task->thread->engine;
+
+    msg->engine = task->thread->engine;
     msg->mem_pool = port->mem_pool;
 
     msg->port_msg.stream = stream;
@@ -292,7 +334,7 @@ nxt_port_write_handler(nxt_task_t *task, void *obj, void *data)
 
             } else {
                 nxt_queue_remove(link);
-                nxt_mp_free(msg->mem_pool, msg);
+                nxt_port_release_send_msg(task, msg, msg->engine);
             }
 
         } else if (nxt_slow_path(n == NXT_ERROR)) {
diff --git a/src/nxt_router.c b/src/nxt_router.c
index f063d87d..8971fb5b 100644
--- a/src/nxt_router.c
+++ b/src/nxt_router.c
@@ -788,7 +788,6 @@ static nxt_int_t
 nxt_router_thread_create(nxt_task_t *task, nxt_runtime_t *rt,
     nxt_event_engine_t *engine)
 {
-    nxt_mp_t             *mp;
     nxt_int_t            ret;
     nxt_port_t           *port;
     nxt_process_t        *process;
@@ -824,12 +823,6 @@ nxt_router_thread_create(nxt_task_t *task, nxt_runtime_t *rt,
         return ret;
     }
 
-    mp = nxt_mp_create(1024, 128, 256, 32);
-    if (nxt_slow_path(mp == NULL)) {
-        return NXT_ERROR;
-    }
-
-    port->mem_pool = mp;
     port->engine = 0;
     port->type = NXT_PROCESS_ROUTER;
 
@@ -1441,6 +1434,8 @@ static void
 nxt_router_process_http_request(nxt_task_t *task, nxt_conn_t *c,
     nxt_app_parse_ctx_t *ap)
 {
+    nxt_mp_t             *port_mp;
+    nxt_int_t            res;
     nxt_port_t           *port, *c_port;
     nxt_req_id_t         req_id;
     nxt_app_wmsg_t       wmsg;
@@ -1466,6 +1461,7 @@ nxt_router_process_http_request(nxt_task_t *task, nxt_conn_t *c,
     } while (nxt_event_engine_request_find(engine, req_id) != NULL);
 
     rc = nxt_conn_request_add(c, req_id);
+
     if (nxt_slow_path(rc == NULL)) {
         // 500 Failed to allocate req->conn link
         nxt_log_alert(task->log, "failed to allocate req->conn link");
@@ -1476,11 +1472,20 @@ nxt_router_process_http_request(nxt_task_t *task, nxt_conn_t *c,
     nxt_debug(task, "req_id %uxD linked to conn %p at engine %p",
               req_id, c, engine);
 
+    port_mp = port->mem_pool;
+    port->mem_pool = c->mem_pool;
+
     c_port = nxt_process_connected_port_find(port->process,
                                              engine->port->pid,
                                              engine->port->id);
     if (nxt_slow_path(c_port != engine->port)) {
-        (void) nxt_port_send_port(task, port, engine->port);
+        res = nxt_port_send_port(task, port, engine->port);
+
+        if (nxt_slow_path(res != NXT_OK)) {
+            // 500 Failed to send reply port
+            nxt_log_alert(task->log, "failed to send reply port to application");
+        }
+
         nxt_process_connected_port_add(port->process, engine->port);
     }
 
@@ -1489,14 +1494,26 @@ nxt_router_process_http_request(nxt_task_t *task, nxt_conn_t *c,
     wmsg.buf = &wmsg.write;
     wmsg.stream = req_id;
 
-    (void)nxt_app->prepare_msg(task, &ap->r, &wmsg);
+    res = nxt_app->prepare_msg(task, &ap->r, &wmsg);
+
+    if (nxt_slow_path(res != NXT_OK)) {
+        // 500 Failed to prepare message
+        nxt_log_alert(task->log, "failed to prepare message for application");
+    }
 
     nxt_debug(task, "about to send %d bytes buffer to worker port %d",
                     nxt_buf_used_size(wmsg.write),
                     wmsg.port->socket.fd);
 
-    (void) nxt_port_socket_write(task, wmsg.port, NXT_PORT_MSG_DATA,
+    res = nxt_port_socket_write(task, wmsg.port, NXT_PORT_MSG_DATA,
                                  -1, req_id, engine->port->id, wmsg.write);
+
+    if (nxt_slow_path(res != NXT_OK)) {
+        // 500 Failed to send message
+        nxt_log_alert(task->log, "failed to send message to application");
+    }
+
+    port->mem_pool = port_mp;
 }
 
 
@@ -1597,7 +1614,9 @@ nxt_router_conn_free(nxt_task_t *task, void *obj, void *data)
 
     } nxt_queue_loop;
 
-    nxt_mp_destroy(c->mem_pool);
+    nxt_queue_remove(&c->link);
+
+    nxt_mp_release(c->mem_pool, c);
 }