mca/threads/qthreads: implement missing functionalities

Signed-off-by: Shintaro Iwasaki <siwasaki@anl.gov>

Author: shintaro-iwasaki

opal/mca/threads/qthreads/Makefile.am

@@ -27,6 +27,7 @@ sources = \
     threads_qthreads_condition.c \
     threads_qthreads_module.c \
     threads_qthreads_mutex.c \
+    threads_qthreads_wait_sync.c \
     threads_qthreads_mutex.h \
     threads_qthreads_threads.h \
     threads_qthreads_tsd.h \

opal/mca/threads/qthreads/configure.m4

@@ -25,12 +25,6 @@
 
 AC_DEFUN([OPAL_CONFIG_QTHREADS],[
 
-    AC_CHECK_HEADERS([qthread/qthread.h],
-                     [AC_CHECK_LIB([qthread],[qthread_initialize],
-                                    [threads_qthreads_happy="yes"],
-                                    [threads_qthreads_happy="no"])],
-                     [threads_qthreads_happy="no"])
-
     AC_ARG_WITH([qthreads],
                 [AC_HELP_STRING([--with-qthreads=DIR],
                                 [Specify location of qthreads installation.  Error if qthreads support cannot be found.])])
@@ -60,7 +54,7 @@ AC_DEFUN([OPAL_CONFIG_QTHREADS],[
 
     AS_IF([test $opal_qthreads_happy = yes],
           [OPAL_CHECK_PACKAGE([opal_qthreads],
-                              [qthread/qthread.h],
+                              [qthread.h],
                               [qthread],
                               [qthread_initialize],
                               [],

opal/mca/threads/qthreads/threads_qthreads.h.in

@@ -26,9 +26,10 @@
 #ifndef OPAL_MCA_THREADS_QTHREADS_THREADS_QTHREADS_H
 #define OPAL_MCA_THREADS_QTHREADS_THREADS_QTHREADS_H 1
 
-#include "@OPAL_QTHREADS_INCLUDE_PATH@qthread/qthread.h"
+#include "@OPAL_QTHREADS_INCLUDE_PATH@qthread.h"
+#include "@OPAL_QTHREADS_INCLUDE_PATH@qthread/tls.h"
 
-static inline void ensure_init_qthreads(void)
+static inline void opal_threads_ensure_init_qthreads(void)
 {
     qthread_initialize();
 }

opal/mca/threads/qthreads/threads_qthreads_component.c

@@ -23,6 +23,7 @@
 
 #include "opal_config.h"
 
+#include "opal/mca/threads/qthreads/threads_qthreads.h"
 #include "opal/mca/threads/thread.h"
 #include "opal/mca/threads/threads.h"
 #include "opal/constants.h"

opal/mca/threads/qthreads/threads_qthreads_module.c

@@ -19,13 +19,8 @@
  * $HEADER$
  */
 
-#include <unistd.h>
-
 #include "opal/mca/threads/qthreads/threads_qthreads.h"
 #include "opal/constants.h"
-#include "opal/util/sys_limits.h"
-#include "opal/util/output.h"
-#include "opal/prefetch.h"
 #include "opal/mca/threads/threads.h"
 #include "opal/mca/threads/tsd.h"
 
@@ -34,40 +29,82 @@ struct opal_tsd_key_value {
     opal_tsd_destructor_t destructor;
 };
 
-static int opal_main_thread;
-struct opal_tsd_key_value *opal_tsd_key_values = NULL;
-static int opal_tsd_key_values_count = 0;
+/* false: uninitialized, true: initialized. */
+static opal_atomic_lock_t opal_thread_self_key_lock = OPAL_ATOMIC_LOCK_INIT;
+static bool opal_thread_self_key_init = false;
+static opal_tsd_key_t opal_thread_self_key;
+
+static inline void self_key_ensure_init(void)
+{
+    if (false == opal_thread_self_key_init) {
+        /* not initialized yet. */
+        opal_atomic_lock(&opal_thread_self_key_lock);
+        /* check again. */
+        if (false == opal_thread_self_key_init) {
+            /* This thread is responsible for initializing this key. */
+            qthread_key_create(&opal_thread_self_key, NULL);
+            opal_atomic_mb();
+            opal_thread_self_key_init = true;
+        }
+        opal_atomic_unlock(&opal_thread_self_key_lock);
+    }
+    /* opal_thread_self_key has been already initialized. */
+}
 
 /*
  * Constructor
  */
 static void opal_thread_construct(opal_thread_t *t)
 {
+    t->t_run = 0;
+    t->t_thread_ret = 0;
 }
 
 OBJ_CLASS_INSTANCE(opal_thread_t,
                    opal_object_t,
                    opal_thread_construct, NULL);
 
+static inline aligned_t *opal_thread_get_qthreads_self(void)
+{
+    self_key_ensure_init();
+    void *ptr = qthread_getspecific(opal_thread_self_key);
+    return (aligned_t *)ptr;
+}
 
-opal_thread_t *opal_thread_get_self(void)
+static aligned_t opal_thread_qthreads_wrapper(void *arg)
 {
-    return NULL;
+    opal_thread_t *t = (opal_thread_t *)arg;
+
+    /* Register itself. */
+    self_key_ensure_init();
+    qthread_setspecific(opal_thread_self_key, t->t_thread_ret_ptr);
+
+    t->t_ret = ((void *(*)(void *))t->t_run)(t);
+    return 0;
 }
 
-bool opal_thread_self_compare(opal_thread_t *t)
+opal_thread_t *opal_thread_get_self(void)
 {
-    return OPAL_ERR_NOT_IMPLEMENTED;
+    opal_threads_ensure_init_qthreads();
+    opal_thread_t *t = OBJ_NEW(opal_thread_t);
+    t->t_thread_ret_ptr = opal_thread_get_qthreads_self();
+    return t;
 }
 
-int sync_wait_mt(void *p)
+bool opal_thread_self_compare(opal_thread_t *t)
 {
-    return OPAL_ERR_NOT_IMPLEMENTED;
+    opal_threads_ensure_init_qthreads();
+    return opal_thread_get_qthreads_self() == &t->t_thread_ret;
 }
 
 int opal_thread_join(opal_thread_t *t, void **thr_return)
 {
-    return OPAL_ERR_NOT_IMPLEMENTED;
+    qthread_readFF(NULL, t->t_thread_ret_ptr);
+    if (thr_return) {
+        *thr_return = t->t_ret;
+    }
+    t->t_thread_ret = 0;
+    return OPAL_SUCCESS;
 }
 
 void opal_thread_set_main(void)
@@ -76,12 +113,17 @@ void opal_thread_set_main(void)
 
 int opal_thread_start(opal_thread_t *t)
 {
-    return OPAL_ERR_NOT_IMPLEMENTED;
+    opal_threads_ensure_init_qthreads();
+    t->t_thread_ret_ptr = &t->t_thread_ret;
+    qthread_fork(opal_thread_qthreads_wrapper, t, &t->t_thread_ret);
+    return OPAL_SUCCESS;
 }
 
 OBJ_CLASS_DECLARATION(opal_thread_t);
 
 int opal_tsd_key_create(opal_tsd_key_t *key, opal_tsd_destructor_t destructor)
 {
-    return OPAL_ERR_NOT_IMPLEMENTED;
+    opal_threads_ensure_init_qthreads();
+    qthread_key_create(key, destructor);
+    return OPAL_SUCCESS;
 }

opal/mca/threads/qthreads/threads_qthreads_mutex.c

@@ -36,6 +36,8 @@ bool opal_uses_threads = false;
 
 static void opal_mutex_construct(opal_mutex_t *m)
 {
+    opal_threads_ensure_init_qthreads();
+    opal_mutex_create(m);
 }
 
 static void opal_mutex_destruct(opal_mutex_t *m)
@@ -56,36 +58,82 @@ OBJ_CLASS_INSTANCE(opal_recursive_mutex_t,
                    opal_recursive_mutex_construct,
                    opal_mutex_destruct);
 
-static void opal_cond_create(opal_cond_t *cond)
-{
-    ensure_init_qthreads();
-}
-
 int opal_cond_init(opal_cond_t *cond)
 {
-    return OPAL_ERR_NOT_IMPLEMENTED;
+    opal_atomic_lock_init(&cond->m_lock, 0);
+    cond->m_waiter_head = NULL;
+    cond->m_waiter_tail = NULL;
+    return OPAL_SUCCESS;
 }
 
+typedef struct {
+    int m_signaled;
+    void *m_prev;
+} cond_waiter_t;
+
 int opal_cond_wait(opal_cond_t *cond, opal_mutex_t *lock)
 {
-    ensure_init_qthreads();
-    return 0 == qthread_readFE(*cond, &lock->m_lock_qthreads) ? OPAL_SUCCESS
-                                                              : OPAL_ERROR;
+    opal_threads_ensure_init_qthreads();
+    /* This thread is taking "lock", so only this thread can access this
+     * condition variable.  */
+    opal_atomic_lock(&cond->m_lock);
+    cond_waiter_t waiter = { 0, NULL };
+    if (NULL == cond->m_waiter_head) {
+        cond->m_waiter_tail = (void *)&waiter;
+    } else {
+        ((cond_waiter_t *)cond->m_waiter_head)->m_prev = (void *)&waiter;
+    }
+    cond->m_waiter_head = (void *)&waiter;
+    opal_atomic_unlock(&cond->m_lock);
+
+    while (1) {
+        opal_mutex_unlock(lock);
+        qthread_yield();
+        opal_mutex_lock(lock);
+        /* Check if someone woke me up. */
+        opal_atomic_lock(&cond->m_lock);
+        int signaled = waiter.m_signaled;
+        opal_atomic_unlock(&cond->m_lock);
+        if (1 == signaled) {
+            break;
+        }
+        /* Unlock the lock again. */
+    }
+    return OPAL_SUCCESS;
 }
 
 int opal_cond_broadcast(opal_cond_t *cond)
 {
-    ensure_init_qthreads();
-    return 0 == qthread_fill(*cond) ? OPAL_SUCCESS : OPAL_ERROR;
+    opal_atomic_lock(&cond->m_lock);
+    while (NULL != cond->m_waiter_tail) {
+        cond_waiter_t *p_cur_tail = (cond_waiter_t *)cond->m_waiter_tail;
+        cond->m_waiter_tail = p_cur_tail->m_prev;
+        /* Awaken one of threads in a FIFO manner. */
+        p_cur_tail->m_signaled = 1;
+    }
+    /* No waiters. */
+    cond->m_waiter_head = NULL;
+    opal_atomic_unlock(&cond->m_lock);
+    return OPAL_SUCCESS;
 }
 
 int opal_cond_signal(opal_cond_t *cond)
 {
-    ensure_init_qthreads();
-    return 0 == qthread_fill(*cond) ? OPAL_SUCCESS : OPAL_ERROR;
+    opal_atomic_lock(&cond->m_lock);
+    if (NULL != cond->m_waiter_tail) {
+        cond_waiter_t *p_cur_tail = (cond_waiter_t *)cond->m_waiter_tail;
+        cond->m_waiter_tail = p_cur_tail->m_prev;
+        /* Awaken one of threads. */
+        p_cur_tail->m_signaled = 1;
+        if (NULL == cond->m_waiter_tail) {
+            cond->m_waiter_head = NULL;
+        }
+    }
+    opal_atomic_unlock(&cond->m_lock);
+    return OPAL_SUCCESS;
 }
 
 int opal_cond_destroy(opal_cond_t *cond)
 {
-    return OPAL_ERR_NOT_IMPLEMENTED;
+    return OPAL_SUCCESS;
 }