Work queue thread assertions. Reset thread after fork.

6 files changed, 54 insertions, 3 deletions

diff --git a/src/nxt_event_engine.c b/src/nxt_event_engine.c
index b91f0e50..2ec7df1e 100644
--- a/src/nxt_event_engine.c
+++ b/src/nxt_event_engine.c
@@ -660,3 +660,19 @@ nxt_event_engine_request_find_remove(nxt_event_engine_t *engine,
     return NULL;
 }
 
+
+#if (NXT_DEBUG)
+
+void nxt_event_engine_thread_adopt(nxt_event_engine_t *engine)
+{
+    nxt_work_queue_thread_adopt(&engine->fast_work_queue);
+    nxt_work_queue_thread_adopt(&engine->accept_work_queue);
+    nxt_work_queue_thread_adopt(&engine->read_work_queue);
+    nxt_work_queue_thread_adopt(&engine->socket_work_queue);
+    nxt_work_queue_thread_adopt(&engine->connect_work_queue);
+    nxt_work_queue_thread_adopt(&engine->write_work_queue);
+    nxt_work_queue_thread_adopt(&engine->shutdown_work_queue);
+    nxt_work_queue_thread_adopt(&engine->close_work_queue);
+}
+
+#endif
diff --git a/src/nxt_event_engine.h b/src/nxt_event_engine.h
index d8741669..6c0adafd 100644
--- a/src/nxt_event_engine.h
+++ b/src/nxt_event_engine.h
@@ -531,5 +531,15 @@ nxt_thread_event_engine(void)
     return thr->engine;
 }
 
+#if (NXT_DEBUG)
+
+NXT_EXPORT void nxt_event_engine_thread_adopt(nxt_event_engine_t *engine);
+
+#else
+
+#define nxt_event_engine_thread_adopt(_engine)
+
+#endif
+
 
 #endif /* _NXT_EVENT_ENGINE_H_INCLUDED_ */
diff --git a/src/nxt_process.c b/src/nxt_process.c
index 9561fb84..9ab3eaf3 100644
--- a/src/nxt_process.c
+++ b/src/nxt_process.c
@@ -50,6 +50,8 @@ nxt_process_create(nxt_task_t *task, nxt_process_t *process)
 
         nxt_port_reset_next_id();
 
+        nxt_event_engine_thread_adopt(task->thread->engine);
+
         /* Remove not ready processes */
         nxt_runtime_process_each(rt, p) {
 
diff --git a/src/nxt_router.c b/src/nxt_router.c
index c103a861..77d2ad8d 100644
--- a/src/nxt_router.c
+++ b/src/nxt_router.c
@@ -1273,6 +1273,8 @@ nxt_router_thread_start(void *data)
 
     thread = nxt_thread();
 
+    nxt_event_engine_thread_adopt(engine);
+
     /* STUB */
     thread->runtime = engine->task.thread->runtime;
 
diff --git a/src/nxt_work_queue.c b/src/nxt_work_queue.c
index 85f5cc3e..1b989eab 100644
--- a/src/nxt_work_queue.c
+++ b/src/nxt_work_queue.c
@@ -58,6 +58,26 @@ nxt_work_queue_thread_assert(nxt_work_queue_t *wq)
     nxt_abort();
 }
 
+
+void nxt_work_queue_thread_adopt(nxt_work_queue_t *wq)
+{
+    nxt_thread_t  *thread;
+
+    thread = nxt_thread();
+
+    wq->pid = nxt_pid;
+    wq->tid = nxt_thread_tid(thread);
+}
+
+
+void
+nxt_work_queue_name(nxt_work_queue_t *wq, const char* name)
+{
+    nxt_work_queue_thread_assert(wq);
+
+    wq->name = name;
+}
+
 #else
 
 #define nxt_work_queue_thread_assert(wq)
diff --git a/src/nxt_work_queue.h b/src/nxt_work_queue.h
index 9819882e..9694bfd2 100644
--- a/src/nxt_work_queue.h
+++ b/src/nxt_work_queue.h
@@ -104,15 +104,16 @@ NXT_EXPORT nxt_work_handler_t nxt_work_queue_pop(nxt_work_queue_t *wq,
 
 #if (NXT_DEBUG)
 
-#define                                                                       \
-nxt_work_queue_name(_wq, _name)                                               \
-    (_wq)->name = _name
+NXT_EXPORT void nxt_work_queue_name(nxt_work_queue_t *wq, const char* name);
+NXT_EXPORT void nxt_work_queue_thread_adopt(nxt_work_queue_t *wq);
 
 #else
 
 #define                                                                       \
 nxt_work_queue_name(_wq, _name)
 
+#define nxt_work_queue_thread_adopt(_wq)
+
 #endif