Memory pool thread safety checks in DEBUG build and usage fixes.

6 files changed, 100 insertions, 16 deletions

diff --git a/src/nxt_mp.c b/src/nxt_mp.c
index f1ddd86b..464df1c1 100644
--- a/src/nxt_mp.c
+++ b/src/nxt_mp.c
@@ -109,6 +109,11 @@ struct nxt_mp_s {
     uint32_t             cluster_size;
     uint32_t             retain;
 
+#if (NXT_DEBUG)
+    nxt_pid_t            pid;
+    nxt_tid_t            tid;
+#endif
+
     /* Lists of nxt_mp_page_t. */
     nxt_queue_t          free_pages;
     nxt_queue_t          nget_pages;
@@ -186,6 +191,49 @@ nxt_lg2(uint64_t v)
 #endif
 
 
+#if (NXT_DEBUG)
+
+nxt_inline void
+nxt_mp_thread_assert(nxt_mp_t *mp)
+{
+    nxt_tid_t     tid;
+    nxt_thread_t  *thread;
+
+    thread = nxt_thread();
+    tid = nxt_thread_tid(thread);
+
+    if (nxt_fast_path(mp->tid == tid)) {
+        return;
+    }
+
+    if (nxt_slow_path(nxt_pid != mp->pid)) {
+        mp->pid = nxt_pid;
+        mp->tid = tid;
+
+        return;
+    }
+
+    nxt_log_alert(thread->log, "mem_pool locked by thread %PT", mp->tid);
+    nxt_abort();
+}
+
+#else
+
+#define nxt_mp_thread_assert(mp)
+
+#endif
+
+
+void
+nxt_mp_thread_adopt(nxt_mp_t *mp)
+{
+#if (NXT_DEBUG)
+    mp->pid = nxt_pid;
+    mp->tid = nxt_thread_tid(NULL);
+#endif
+}
+
+
 nxt_mp_t *
 nxt_mp_create(size_t cluster_size, size_t page_alignment, size_t page_size,
     size_t min_chunk_size)
@@ -417,6 +465,8 @@ nxt_mp_alloc_small(nxt_mp_t *mp, size_t size)
     nxt_mp_page_t     *page;
     nxt_queue_link_t  *link;
 
+    nxt_mp_thread_assert(mp);
+
     p = NULL;
 
     if (size <= mp->page_size / 2) {
@@ -489,6 +539,8 @@ nxt_mp_get_small(nxt_mp_t *mp, nxt_queue_t *pages, size_t size)
     nxt_mp_page_t     *page;
     nxt_queue_link_t  *link, *next;
 
+    nxt_mp_thread_assert(mp);
+
     for (link = nxt_queue_first(pages);
          link != nxt_queue_tail(pages);
          link = next)
@@ -604,6 +656,8 @@ nxt_mp_alloc_large(nxt_mp_t *mp, size_t alignment, size_t size)
     uint8_t         type;
     nxt_mp_block_t  *block;
 
+    nxt_mp_thread_assert(mp);
+
     /* Allocation must be less than 4G. */
     if (nxt_slow_path(size >= 0xFFFFFFFF)) {
         return NULL;
@@ -664,6 +718,8 @@ nxt_mp_free(nxt_mp_t *mp, void *p)
     nxt_thread_t    *thread;
     nxt_mp_block_t  *block;
 
+    nxt_mp_thread_assert(mp);
+
     nxt_debug_alloc("mp free %p", p);
 
     block = nxt_mp_find_block(&mp->blocks, p);
diff --git a/src/nxt_mp.h b/src/nxt_mp.h
index 40abc707..78040df2 100644
--- a/src/nxt_mp.h
+++ b/src/nxt_mp.h
@@ -109,4 +109,6 @@ NXT_EXPORT void *nxt_mp_zget(nxt_mp_t *mp, size_t size)
     NXT_MALLOC_LIKE;
 
 
+NXT_EXPORT void nxt_mp_thread_adopt(nxt_mp_t *mp);
+
 #endif /* _NXT_MP_H_INCLUDED_ */
diff --git a/src/nxt_port_memory.c b/src/nxt_port_memory.c
index cf69a758..6aaff5cf 100644
--- a/src/nxt_port_memory.c
+++ b/src/nxt_port_memory.c
@@ -103,7 +103,8 @@ nxt_port_incoming_port_mmap(nxt_task_t *task, nxt_process_t *process,
     nxt_thread_mutex_lock(&process->incoming_mutex);
 
     if (process->incoming == NULL) {
-        process->incoming = nxt_array_create(process->mem_pool, 1,
+        process->incoming_mp = nxt_mp_create(1024, 128, 256, 32);
+        process->incoming = nxt_array_create(process->incoming_mp, 1,
             sizeof(nxt_port_mmap_t));
     }
 
@@ -113,6 +114,8 @@ nxt_port_incoming_port_mmap(nxt_task_t *task, nxt_process_t *process,
         goto fail;
     }
 
+    nxt_mp_thread_adopt(process->incoming_mp);
+
     port_mmap = nxt_array_zero_add(process->incoming);
     if (nxt_slow_path(port_mmap == NULL)) {
         nxt_log(task, NXT_LOG_WARN, "failed to add mmap to incoming array");
@@ -192,7 +195,8 @@ nxt_port_new_port_mmap(nxt_task_t *task, nxt_process_t *process,
     port_mmap = NULL;
 
     if (process->outgoing == NULL) {
-        process->outgoing = nxt_array_create(process->mem_pool, 1,
+        process->outgoing_mp = nxt_mp_create(1024, 128, 256, 32);
+        process->outgoing = nxt_array_create(process->outgoing_mp, 1,
                                              sizeof(nxt_port_mmap_t));
     }
 
@@ -202,6 +206,8 @@ nxt_port_new_port_mmap(nxt_task_t *task, nxt_process_t *process,
         return NULL;
     }
 
+    nxt_mp_thread_adopt(process->outgoing_mp);
+
     port_mmap = nxt_array_zero_add(process->outgoing);
     if (nxt_slow_path(port_mmap == NULL)) {
         nxt_log(task, NXT_LOG_WARN,
diff --git a/src/nxt_process.h b/src/nxt_process.h
index d0ed7343..28824bd0 100644
--- a/src/nxt_process.h
+++ b/src/nxt_process.h
@@ -57,8 +57,10 @@ typedef struct {
 
     nxt_process_init_t  *init;
     nxt_thread_mutex_t  incoming_mutex;
+    nxt_mp_t            *incoming_mp;
     nxt_array_t         *incoming;  /* of nxt_port_mmap_t */
     nxt_thread_mutex_t  outgoing_mutex;
+    nxt_mp_t            *outgoing_mp;
     nxt_array_t         *outgoing;  /* of nxt_port_mmap_t */
 
     nxt_lvlhsh_t        connected_ports; /* of nxt_port_t */
diff --git a/src/nxt_router.c b/src/nxt_router.c
index 9753d289..a0c5a17d 100644
--- a/src/nxt_router.c
+++ b/src/nxt_router.c
@@ -926,6 +926,8 @@ nxt_router_thread_start(void *data)
     thread->task = &engine->task;
     thread->fiber = &engine->fibers->fiber;
 
+    nxt_mp_thread_adopt(engine->port->mem_pool);
+
     engine->port->socket.task = task;
     nxt_port_create(task, engine->port, nxt_router_app_port_handlers);
 
diff --git a/src/nxt_runtime.c b/src/nxt_runtime.c
index 9d7a3fd7..91c4ba70 100644
--- a/src/nxt_runtime.c
+++ b/src/nxt_runtime.c
@@ -1522,8 +1522,6 @@ nxt_runtime_process_find(nxt_runtime_t *rt, nxt_pid_t pid)
     lhq.key.start = (u_char *) &pid;
     lhq.proto = &lvlhsh_processes_proto;
 
-    /* TODO lock processes */
-
     if (nxt_lvlhsh_find(&rt->processes, &lhq) == NXT_OK) {
         nxt_thread_log_debug("process %PI found", pid);
         return lhq.value;
@@ -1546,8 +1544,6 @@ nxt_runtime_process_get(nxt_runtime_t *rt, nxt_pid_t pid)
     lhq.key.start = (u_char *) &pid;
     lhq.proto = &lvlhsh_processes_proto;
 
-    /* TODO lock processes */
-
     if (nxt_lvlhsh_find(&rt->processes, &lhq) == NXT_OK) {
         nxt_thread_log_debug("process %PI found", pid);
         return lhq.value;
@@ -1599,8 +1595,6 @@ nxt_runtime_process_add(nxt_runtime_t *rt, nxt_process_t *process)
     lhq.value = process;
     lhq.pool = rt->mem_pool;
 
-    /* TODO lock processes */
-
     switch (nxt_lvlhsh_insert(&rt->processes, &lhq)) {
 
     case NXT_OK:
@@ -1627,7 +1621,9 @@ nxt_runtime_process_add(nxt_runtime_t *rt, nxt_process_t *process)
 void
 nxt_runtime_process_remove(nxt_runtime_t *rt, nxt_process_t *process)
 {
+    uint32_t            i;
     nxt_port_t          *port;
+    nxt_port_mmap_t     *port_mmap;
     nxt_lvlhsh_query_t  lhq;
 
     lhq.key_hash = nxt_murmur_hash2(&process->pid, sizeof(process->pid));
@@ -1638,8 +1634,6 @@ nxt_runtime_process_remove(nxt_runtime_t *rt, nxt_process_t *process)
     lhq.value = process;
     lhq.pool = rt->mem_pool;
 
-    /* TODO lock processes */
-
     switch (nxt_lvlhsh_delete(&rt->processes, &lhq)) {
 
     case NXT_OK:
@@ -1651,6 +1645,34 @@ nxt_runtime_process_remove(nxt_runtime_t *rt, nxt_process_t *process)
 
         } nxt_process_port_loop;
 
+        if (process->incoming) {
+            nxt_mp_thread_adopt(process->incoming_mp);
+
+            port_mmap = process->incoming->elts;
+
+            for (i = 0; i < process->incoming->nelts; i++) {
+                nxt_port_mmap_destroy(port_mmap);
+            }
+
+            nxt_thread_mutex_destroy(&process->incoming_mutex);
+
+            nxt_mp_destroy(process->incoming_mp);
+        }
+
+        if (process->outgoing) {
+            nxt_mp_thread_adopt(process->outgoing_mp);
+
+            port_mmap = process->outgoing->elts;
+
+            for (i = 0; i < process->outgoing->nelts; i++) {
+                nxt_port_mmap_destroy(port_mmap);
+            }
+
+            nxt_thread_mutex_destroy(&process->outgoing_mutex);
+
+            nxt_mp_destroy(process->outgoing_mp);
+        }
+
         nxt_mp_free(rt->mem_pool, process);
         break;
 
@@ -1681,8 +1703,6 @@ nxt_runtime_port_first(nxt_runtime_t *rt, nxt_lvlhsh_each_t *lhe)
 void
 nxt_runtime_port_add(nxt_runtime_t *rt, nxt_port_t *port)
 {
-    /* TODO lock ports */
-
     nxt_port_hash_add(&rt->ports, rt->mem_pool, port);
 }
 
@@ -1690,8 +1710,6 @@ nxt_runtime_port_add(nxt_runtime_t *rt, nxt_port_t *port)
 void
 nxt_runtime_port_remove(nxt_runtime_t *rt, nxt_port_t *port)
 {
-    /* TODO lock ports */
-
     nxt_port_hash_remove(&rt->ports, rt->mem_pool, port);
 
     if (port->pair[0] != -1) {
@@ -1714,7 +1732,5 @@ nxt_port_t *
 nxt_runtime_port_find(nxt_runtime_t *rt, nxt_pid_t pid,
     nxt_port_id_t port_id)
 {
-    /* TODO lock ports */
-
     return nxt_port_hash_find(&rt->ports, pid, port_id);
 }