Cleanup Pthread code. (#298)

Most notably plug memory leaks and another potential race.

Signed-off-by: Samuel K. Gutierrez <[email protected]>

Author: samuelkgutierrez

src/qvi-group-pthread.cc

@@ -41,20 +41,21 @@ qvi_group_pthread::qvi_group_pthread(
     assert(ctx && thread_group);
     m_context = ctx;
     m_context->retain();
-    //
+    // No retain() here because the calling side takes care of this. See
+    // qvi_pthread_group::split() by way of qvi_group_pthread::split().
     thgroup = thread_group;
 }
 
 qvi_group_pthread::~qvi_group_pthread(void)
 {
-    qvi_delete(&thgroup);
+    thgroup->release();
+    m_context->release();
 }
 
 int
 qvi_group_pthread::self(
     qvi_group **
 ) {
-    // TODO(skg)
     return QV_ERR_NOT_SUPPORTED;
 }
 
@@ -75,7 +76,6 @@ qvi_group_pthread::split(
     rc = qvi_new(&ichild, m_context, ithgroup);
 out:
     if (qvi_unlikely(rc != QV_SUCCESS)) {
-        qvi_delete(&ithgroup);
         qvi_delete(&ichild);
     }
     *child = ichild;

src/qvi-group-pthread.h

@@ -77,11 +77,11 @@ struct qvi_group_pthread : public qvi_group {
     make_intrinsic(
         qv_scope_intrinsic_t
     ) {
-        // Nothing to do here because a Pthread group cannot be created outside
-        // of another group. For example, a thread_split can be called from a
+        // Not supported because a Pthread group cannot be created outside of
+        // another group. For example, a thread_split can be called from a
         // process context, which can be an intrinsic group, but not from a
         // threaded context alone.
-        return QV_SUCCESS;
+        return QV_ERR_NOT_SUPPORTED;
     }
 
     virtual int

src/qvi-pthread.cc

@@ -17,26 +17,27 @@
 #include "qvi-utils.h"
 
 int
-qvi_pthread_group::m_init_by_a_single_thread(
+qvi_pthread_group::m_start_init_by_a_single_thread(
     qvi_pthread_group_context *ctx,
     int group_size
 ) {
-    m_context = ctx;
+    assert(ctx && group_size > 0);
 
+    m_context = ctx;
     m_size = group_size;
 
-    int rc = pthread_barrier_init(&m_barrier, NULL, m_size);
-    if (qvi_unlikely(rc != 0)) throw qvi_runtime_error();
+    int rc = pthread_barrier_init(&m_barrier, nullptr, m_size);
+    if (qvi_unlikely(rc != 0)) return rc;
 
-    m_data_g = new qvi_bbuff *[m_size]();
+    m_gather_data = new qvi_bbuff *[m_size]();
     for (int i = 0 ; i < group_size ; i++) {
-        const int rc = qvi_bbuff_new(&m_data_g[i]);
-        if (qvi_unlikely(rc != 0)) throw qvi_runtime_error();
+        rc = qvi_bbuff_new(&m_gather_data[i]);
+        if (qvi_unlikely(rc != QV_SUCCESS)) return rc;
     }
-    m_data_s = new qvi_bbuff**();
+    m_scatter_data = new qvi_bbuff**();
     m_ckrs = new qvi_subgroup_color_key_rank[m_size]();
 
-    return rc;
+    return QV_SUCCESS;
 }
 
 /* static */ int
@@ -52,7 +53,7 @@ qvi_pthread_group::m_finish_init_by_all_threads(
         group->m_tids.push_back(mytid);
     }
     // Make sure they all contribute before continuing.
-    pthread_barrier_wait(&group->m_barrier);
+    group->barrier();
     // Elect one thread to be the worker.
     bool worker = false;
     {
@@ -61,14 +62,15 @@ qvi_pthread_group::m_finish_init_by_all_threads(
     }
     // The worker populates the TID to rank mapping, while the others wait.
     if (worker) {
+        std::lock_guard<std::mutex> guard(group->m_mutex);
         std::sort(group->m_tids.begin(), group->m_tids.end());
 
         for (int i = 0; i < group->m_size; ++i) {
             const pid_t tidi = group->m_tids[i];
             group->m_tid2rank.insert({tidi, i});
         }
     }
-    pthread_barrier_wait(&group->m_barrier);
+    group->barrier();
     // Everyone can now create their task and populate the mapping table.
     {
         std::lock_guard<std::mutex> guard(group->m_mutex);
@@ -78,27 +80,28 @@ qvi_pthread_group::m_finish_init_by_all_threads(
         group->m_tid2task.insert({mytid, task});
     }
     // Make sure they all finish before returning.
-    pthread_barrier_wait(&group->m_barrier);
+    group->barrier();
     return rc;
 }
 
 qvi_pthread_group::qvi_pthread_group(
     qvi_pthread_group_context *ctx,
     int group_size
 ) {
-    const int rc = m_init_by_a_single_thread(ctx, group_size);
+    const int rc = m_start_init_by_a_single_thread(ctx, group_size);
     if (qvi_unlikely(rc != QV_SUCCESS)) throw qvi_runtime_error();
-    // TODO(skg) Add to group table context.
 }
 
 qvi_pthread_group::qvi_pthread_group(
     qvi_pthread_group *parent_group,
     const qvi_subgroup_info &sginfo
 ) {
-    assert(sginfo.rank == 0);
+    assert(sginfo.rank == qvi_subgroup_info::master_rank);
 
     std::lock_guard<std::mutex> guard(parent_group->m_mutex);
-    const int rc = m_init_by_a_single_thread(parent_group->m_context, sginfo.size);
+    const int rc = m_start_init_by_a_single_thread(
+        parent_group->m_context, sginfo.size
+    );
     if (qvi_unlikely(rc != QV_SUCCESS)) throw qvi_runtime_error();
 
     const qvi_group_id_t mygid = parent_group->m_subgroup_gids[sginfo.index];
@@ -124,30 +127,23 @@ qvi_pthread_group::call_first_from_pthread_create(
 
 qvi_pthread_group::~qvi_pthread_group(void)
 {
-    // TODO(skg)
-    return;
-
     std::lock_guard<std::mutex> guard(m_mutex);
+
     for (auto &tt : m_tid2task) {
         qvi_delete(&tt.second);
     }
-    pthread_barrier_destroy(&m_barrier);
 
-    if (m_data_g) {
+    if (m_gather_data) {
         for (int i = 0; i < m_size; ++i) {
-            qvi_bbuff_delete(&m_data_g[i]);
+            qvi_bbuff_delete(&m_gather_data[i]);
         }
-        delete[] m_data_g;
+        delete[] m_gather_data;
     }
 
-    //C++ flavor
-    /*
-    for (auto &tt : m_data) {
-        qvi_delete(&tt);
-    }
-    */
-    delete m_data_s;
+    delete m_scatter_data;
     delete[] m_ckrs;
+
+    pthread_barrier_destroy(&m_barrier);
 }
 
 int
@@ -216,8 +212,8 @@ qvi_pthread_group::m_subgroup_info(
         const size_t ncolors = color_set.size();
         m_ckrs[my_rank].ncolors = ncolors;
         // Now that we know the number of distinct colors, populate the
-        // sub-group IDs. Set rc here and continue until return so we don't hang
-        // on an error path.
+        // sub-group IDs. Set rc here and continue until after the next barrier
+        // so we don't hang on an error path.
         rc = qvi_group::next_ids(ncolors, m_subgroup_gids);
     }
     // All threads wait for the number of colors to be computed.
@@ -265,7 +261,9 @@ qvi_pthread_group::split(
     qvi_subgroup_info sginfo;
     int rc = m_subgroup_info(color, key, sginfo);
     if (qvi_unlikely(rc != QV_SUCCESS)) goto out;
-    if (sginfo.rank == 0) {
+    // One thread creates the child group. The rest wait for the instance and
+    // later grab a pointer to their group based on the sub-group index.
+    if (sginfo.rank == qvi_subgroup_info::master_rank) {
         // Recall this is the parent group.
         rc = qvi_new(&ichild, this, sginfo);
         barrier();
@@ -274,10 +272,11 @@ qvi_pthread_group::split(
         barrier();
         const qvi_group_id_t mygid = m_subgroup_gids[sginfo.index];
         ichild = m_context->groupid2thgroup.at(mygid);
+        ichild->retain();
     }
     // Now we can check if errors happened above.
     if (qvi_unlikely(rc != QV_SUCCESS)) goto out;
-
+    // Collectively finish child instance initialization.
     rc = m_finish_init_by_all_threads(ichild);
 out:
     if (qvi_unlikely(rc != QV_SUCCESS)) {
@@ -300,7 +299,7 @@ qvi_pthread_group::gather(
     int rc = QV_SUCCESS;
     {
         std::lock_guard<std::mutex> guard(m_mutex);
-        rc = qvi_bbuff_copy(*txbuff, m_data_g[myrank]);
+        rc = qvi_bbuff_copy(*txbuff, m_gather_data[myrank]);
         *alloc_type = QVI_BBUFF_ALLOC_SHARED_GLOBAL;
     }
     // Need to ensure that all threads have contributed to m_data_g
@@ -310,7 +309,7 @@ qvi_pthread_group::gather(
         *rxbuffs = nullptr;
         return QV_ERR_INTERNAL;
     }
-    *rxbuffs = m_data_g;
+    *rxbuffs = m_gather_data;
     return rc;
 }
 
@@ -325,13 +324,13 @@ qvi_pthread_group::scatter(
     qvi_bbuff *mybbuff = nullptr;
 
     if (rootid == myrank) {
-        *m_data_s = txbuffs;
+        *m_scatter_data = txbuffs;
     }
 
     barrier();
     {
         std::lock_guard<std::mutex> guard(m_mutex);
-        rc = qvi_bbuff_dup(*((*m_data_s)[myrank]), &mybbuff);
+        rc = qvi_bbuff_dup(*((*m_scatter_data)[myrank]), &mybbuff);
     }
     barrier();
 

src/qvi-pthread.h

@@ -28,7 +28,7 @@ struct qvi_pthread_group;
  * all threads that are spawned by a common parent process.
  */
 struct qvi_pthread_group_context : public qvi_refc {
-    /** */
+    /** Maps group IDs to Pthread group instance pointers. */
     std::unordered_map<qvi_group_id_t, qvi_pthread_group *> groupid2thgroup;
 };
 
@@ -51,7 +51,7 @@ struct qvi_pthread_group_pthread_create_args {
       , throutine_argp(throutine_argp_a) { }
 };
 
-struct qvi_pthread_group {
+struct qvi_pthread_group : qvi_refc {
 private:
     /** Context information. */
     qvi_pthread_group_context *m_context = nullptr;
@@ -70,16 +70,16 @@ struct qvi_pthread_group {
     /** Used for barrier things. */
     pthread_barrier_t m_barrier;
     /** Used for gather exchanges. */
-    qvi_bbuff **m_data_g = nullptr;
+    qvi_bbuff **m_gather_data = nullptr;
     /** Used for scatter exchanges. */
-    qvi_bbuff ***m_data_s = nullptr;
-    /** Shared color, key, rank scratch pad used for splitting. */
+    qvi_bbuff ***m_scatter_data = nullptr;
+    /** Shared color, key, rank data used for splitting. */
     qvi_subgroup_color_key_rank *m_ckrs = nullptr;
     /** Shared sub-group IDs. */
     std::vector<qvi_group_id_t> m_subgroup_gids;
     /** */
     int
-    m_init_by_a_single_thread(
+    m_start_init_by_a_single_thread(
         qvi_pthread_group_context *ctx,
         int group_size
     );

src/qvi-subgroup.h

@@ -27,6 +27,8 @@
  * doesn't have native support for creating sub-groups.
  */
 struct qvi_subgroup_info {
+    /** The rank of the master task. */
+    static constexpr int master_rank = 0;
     /** Number of sub-groups created from split. */
     int ngroups = 0;
     /** My sub-group index (from 0 to ngroups - 1). */

tests/test-pthread-split.c

@@ -47,14 +47,11 @@ thread_work(
     ctu_scope_report(pthread_subscope, "thread_subscope");
     ctu_emit_task_bind(pthread_subscope);
 
-#if 0
     rc = qv_scope_free(pthread_subscope);
     if (rc != QV_SUCCESS) {
         ers = "qv_scope_free failed";
         ctu_panic("%s (rc=%s)", ers, qv_strerr(rc));
     }
-#endif
-
     return NULL;
 }