Initial version.

1 files changed, 228 insertions, 0 deletions

diff --git a/src/nxt_thread.c b/src/nxt_thread.c
new file mode 100644
index 00000000..415253c0
--- /dev/null
+++ b/src/nxt_thread.c
@@ -0,0 +1,228 @@
+
+/*
+ * Copyright (C) Igor Sysoev
+ * Copyright (C) NGINX, Inc.
+ */
+
+#include <nxt_main.h>
+
+
+static void *nxt_thread_trampoline(void *data);
+static void nxt_thread_time_cleanup(void *data);
+
+
+#if (NXT_HAVE_PTHREAD_SPECIFIC_DATA)
+
+static void nxt_thread_key_dtor(void *data);
+
+
+void
+nxt_thread_init_data(nxt_thread_specific_data_t tsd)
+{
+    void           *p;
+    nxt_err_t      err;
+    pthread_key_t  key;
+
+    while ((nxt_atomic_int_t) tsd->key < 0) {
+        /*
+         * Atomic allocation of a key number.
+         * -1 means an uninitialized key,
+         * -2 is the initializing lock to assure the single value for the key.
+         */
+        if (nxt_atomic_cmp_set(&tsd->key, -1, -2)) {
+
+            err = pthread_key_create(&key, nxt_thread_key_dtor);
+            if (err != 0) {
+                nxt_main_log_emerg("pthread_key_create() failed %E", err);
+                goto fail;
+            }
+
+            tsd->key = (nxt_atomic_t) key;
+
+            nxt_main_log_debug("pthread_key_create(): %A", tsd->key);
+        }
+    }
+
+    if (pthread_getspecific((pthread_key_t) tsd->key) != NULL) {
+        return;
+    }
+
+    p = nxt_zalloc(tsd->size);
+    if (p == NULL) {
+        goto fail;
+    }
+
+    err = pthread_setspecific((pthread_key_t) tsd->key, p);
+    if (err == 0) {
+        return;
+    }
+
+    nxt_main_log_alert("pthread_setspecific(%A) failed %E", tsd->key, err);
+
+fail:
+
+    pthread_exit(NULL);
+    nxt_unreachable();
+}
+
+
+static void
+nxt_thread_key_dtor(void *data)
+{
+    nxt_main_log_debug("pthread key dtor: %p", data);
+
+    nxt_free(data);
+}
+
+#endif
+
+
+nxt_int_t
+nxt_thread_create(nxt_thread_handle_t *handle, nxt_thread_link_t *link)
+{
+    nxt_err_t  err;
+
+    err = pthread_create(handle, NULL, nxt_thread_trampoline, link);
+
+    if (nxt_fast_path(err == 0)) {
+        nxt_thread_log_debug("pthread_create(): %PH", *handle);
+
+        return NXT_OK;
+    }
+
+    nxt_thread_log_alert("pthread_create() failed %E", err);
+
+    nxt_free(link);
+
+    return NXT_ERROR;
+}
+
+
+static void *
+nxt_thread_trampoline(void *data)
+{
+    nxt_thread_t        *thr;
+    nxt_thread_link_t   *link;
+    nxt_thread_start_t  start;
+
+    link = data;
+
+    thr = nxt_thread_init();
+
+    nxt_log_debug(thr->log, "thread trampoline: %PH", thr->handle);
+
+    pthread_cleanup_push(nxt_thread_time_cleanup, thr);
+
+    start = link->start;
+    data = link->data;
+
+    if (link->engine != NULL) {
+        thr->link = link;
+
+    } else {
+        nxt_free(link);
+    }
+
+    start(data);
+
+    /*
+     * nxt_thread_time_cleanup() should be called only if a thread
+     * would be canceled, so ignore it here because nxt_thread_exit()
+     * calls nxt_thread_time_free() as well.
+     */
+    pthread_cleanup_pop(0);
+
+    nxt_thread_exit(thr);
+    nxt_unreachable();
+    return NULL;
+}
+
+
+nxt_thread_t *
+nxt_thread_init(void)
+{
+    nxt_thread_t  *thr;
+
+    nxt_thread_init_data(nxt_thread_context);
+
+    thr = nxt_thread();
+
+    if (thr->log == NULL) {
+        thr->log = &nxt_main_log;
+        thr->handle = nxt_thread_handle();
+
+        /*
+         * Threads are never preempted by asynchronous signals, since
+         * the signals are processed synchronously by dedicated thread.
+         */
+        thr->time.signal = -1;
+
+        nxt_thread_time_update(thr);
+    }
+
+    return thr;
+}
+
+
+static void
+nxt_thread_time_cleanup(void *data)
+{
+    nxt_thread_t  *thr;
+
+    thr = data;
+
+    nxt_log_debug(thr->log, "thread time cleanup");
+
+    nxt_thread_time_free(thr);
+}
+
+
+void
+nxt_thread_exit(nxt_thread_t *thr)
+{
+    nxt_log_debug(thr->log, "thread exit");
+
+    if (thr->link != NULL) {
+        nxt_event_engine_post(thr->link->engine, thr->link->exit,
+                              (void *) (uintptr_t) thr->handle, NULL,
+                              &nxt_main_log);
+
+        nxt_free(thr->link);
+        thr->link = NULL;
+    }
+
+    nxt_thread_time_free(thr);
+
+    pthread_exit(NULL);
+    nxt_unreachable();
+}
+
+
+void
+nxt_thread_cancel(nxt_thread_handle_t handle)
+{
+    nxt_err_t  err;
+
+    nxt_thread_log_debug("thread cancel: %PH", handle);
+
+    err = pthread_cancel(handle);
+
+    if (err != 0) {
+        nxt_main_log_alert("pthread_cancel(%PH) failed %E", handle, err);
+    }
+}
+
+
+void
+nxt_thread_wait(nxt_thread_handle_t handle)
+{
+    nxt_err_t  err;
+
+    nxt_thread_log_debug("thread wait: %PH", handle);
+
+    err = pthread_join(handle, NULL);
+
+    if (err != 0) {
+        nxt_main_log_alert("pthread_join(%PH) failed %E", handle, err);
+    }
+}