pthread gather/scatter/split support MkI (#273)

Signed-off-by: Guillaume Mercier <[email protected]>

Author: GuillaumeMercier

include/quo-vadis-pthread.h

@@ -27,6 +27,21 @@
 extern "C" {
 #endif
 
+/**
+ * Mapping policies types.
+ */
+typedef enum qv_policy_s {
+  QV_POLICY_PACKED     = 1,
+  QV_POLICY_COMPACT    = 1,
+  QV_POLICY_CLOSE      = 1,
+  QV_POLICY_SPREAD     = 2,
+  QV_POLICY_DISTRIBUTE = 3,
+  QV_POLICY_ALTERNATE  = 3,
+  QV_POLICY_CORESFIRST = 3,
+  QV_POLICY_SCATTER    = 4,
+  QV_POLICY_CHOOSE     = 5,
+} qv_policy_t;
+
 /**
  * Similar to pthread_create(3).
  */
@@ -66,6 +81,16 @@ qv_pthread_scopes_free(
     qv_scope_t **scopes
 );
 
+/**
+ * Fills color array used in qv_pthread_scope_split*.
+ */
+int
+qv_pthread_colors_fill(
+    int *color_array,
+    int array_size,
+    qv_policy_t policy
+);
+
 #ifdef __cplusplus
 }
 #endif

src/quo-vadis-pthread.cc

@@ -1,5 +1,5 @@
 /* -*- Mode: C++; c-basic-offset:4; indent-tabs-mode:nil -*- */
-/*
+/*qv_policy_t
  * Copyright (c) 2022-2024 Triad National Security, LLC
  *                         All rights reserved.
  *
@@ -116,7 +116,7 @@ qv_pthread_create(
         return ENOMEM;
     }
     return pthread_create(
-        thread, attr, qvi_pthread_group_s::call_first_from_pthread_create, cargs
+        thread, attr, qvi_pthread_group::call_first_from_pthread_create, cargs
     );
 }
 
@@ -135,6 +135,16 @@ qv_pthread_scopes_free(
     qvi_catch_and_return();
 }
 
+int
+qv_pthread_colors_fill(
+   int *,//color_array,
+   int, // array_size,
+   qv_policy_t //policy
+){
+    //TODO(GM) implement
+    return QV_ERR_NOT_SUPPORTED;
+}
+
 /*
  * vim: ft=cpp ts=4 sts=4 sw=4 expandtab
  */

src/quo-vadis.cc

@@ -179,6 +179,7 @@ qv_scope_split(
         // We use the sentinel value QV_HW_OBJ_LAST to differentiate between
         // calls from split() and split_at(). Since this call doesn't have a
         // hardware type argument, we use QV_HW_OBJ_LAST as the hardware type.
+        // return
         return scope->split(npieces, color, QV_HW_OBJ_LAST, subscope);
     }
     qvi_catch_and_return();

src/qvi-bbuff.cc

@@ -112,6 +112,14 @@ qvi_bbuff_dup(
     return qvi_dup(src, buff);
 }
 
+int
+qvi_bbuff_copy(
+    const qvi_bbuff &src,
+    qvi_bbuff *buff
+) {
+    return qvi_copy(src, buff);
+}
+
 void
 qvi_bbuff_delete(
     qvi_bbuff **buff

src/qvi-bbuff.h

@@ -73,11 +73,25 @@ qvi_bbuff_dup(
     qvi_bbuff **buff
 );
 
+int
+qvi_bbuff_copy(
+    const qvi_bbuff &src,
+    qvi_bbuff *buff
+);
+
 void
 qvi_bbuff_delete(
     qvi_bbuff **buff
 );
 
+enum qvi_alloc_s {
+    ALLOC_SHARED = 0,
+    ALLOC_SHARED_GLOBAL,
+    ALLOC_PRIVATE,
+};
+
+typedef enum qvi_alloc_s qvi_alloc_type_t;
+
 #endif
 
 /*

src/qvi-common.h

@@ -48,6 +48,7 @@
 #include <chrono>
 #include <map>
 #include <mutex>
+#include <condition_variable>
 #include <new>
 #include <numeric>
 #include <set>

src/qvi-group-mpi.h

@@ -20,6 +20,7 @@
 #include "qvi-common.h"
 #include "qvi-group.h"
 #include "qvi-mpi.h"
+#include "qvi-bbuff.h"
 
 struct qvi_group_mpi : public qvi_group {
 protected:
@@ -84,7 +85,7 @@ struct qvi_group_mpi : public qvi_group {
     gather(
         qvi_bbuff *txbuff,
         int root,
-        bool *shared,
+        qvi_alloc_type_t *shared,
         qvi_bbuff ***rxbuffs
     ) {
         return qvi_mpi_group_gather_bbuffs(

src/qvi-group-omp.h

@@ -23,6 +23,7 @@
 #include "qvi-common.h"
 #include "qvi-group.h"
 #include "qvi-omp.h"
+#include "qvi-bbuff.h"
 
 struct qvi_group_omp : public qvi_group {
 private:
@@ -90,7 +91,7 @@ struct qvi_group_omp : public qvi_group {
     gather(
         qvi_bbuff *txbuff,
         int root,
-        bool *shared,
+        qvi_alloc_type_t *shared,
         qvi_bbuff ***rxbuffs
     ) {
         return m_ompgroup->gather(txbuff, root, shared, rxbuffs);

src/qvi-group-process.h

@@ -17,6 +17,7 @@
 #include "qvi-common.h"
 #include "qvi-group.h"
 #include "qvi-process.h"
+#include "qvi-bbuff.h"
 
 struct qvi_group_process : public qvi_group {
 protected:
@@ -85,7 +86,7 @@ struct qvi_group_process : public qvi_group {
     gather(
         qvi_bbuff *txbuff,
         int root,
-        bool *shared,
+        qvi_alloc_type_t  *shared,
         qvi_bbuff ***rxbuffs
     ) {
         return qvi_process_group_gather_bbuffs(

src/qvi-group-pthread.cc

@@ -17,7 +17,7 @@
 qvi_group_pthread::qvi_group_pthread(
     int group_size
 ) {
-    const int rc = qvi_new(&thgroup, group_size);
+    const int rc = qvi_new(&thgroup, group_size, 0);
     if (qvi_unlikely(rc != QV_SUCCESS)) throw qvi_runtime_error();
 }
 
@@ -36,12 +36,21 @@ qvi_group_pthread::self(
 
 int
 qvi_group_pthread::split(
-    int,
-    int,
-    qvi_group **
+    int color ,
+    int key,
+    qvi_group ** child
 ) {
-    // TODO(skg)
-    return QV_ERR_NOT_SUPPORTED;
+    qvi_group_pthread *ichild = nullptr;
+    int rc  = qvi_new(&ichild);
+    if (qvi_unlikely(rc != QV_SUCCESS)) goto out;
+
+    rc = thgroup->split(color, key, &ichild->thgroup);
+out:
+    if (qvi_unlikely(rc != QV_SUCCESS)) {
+        qvi_delete(&ichild);
+    }
+    *child = ichild;
+    return rc;
 }
 
 /*

src/qvi-group-pthread.h

@@ -17,12 +17,13 @@
 #include "qvi-common.h"
 #include "qvi-group.h"
 #include "qvi-pthread.h"
+#include "qvi-bbuff.h"
 
 struct qvi_group_pthread : public qvi_group {
     /** Underlying group instance. */
     qvi_pthread_group_t *thgroup = nullptr;
     /** Constructor. */
-    qvi_group_pthread(void) = delete;
+    qvi_group_pthread(void) = default;
     /** Constructor. */
     qvi_group_pthread(
         int group_size
@@ -87,7 +88,7 @@ struct qvi_group_pthread : public qvi_group {
     gather(
         qvi_bbuff *txbuff,
         int root,
-        bool *shared,
+        qvi_alloc_type_t  *shared,
         qvi_bbuff ***rxbuffs
     ) {
         return thgroup->gather(

src/qvi-group.h

@@ -19,6 +19,7 @@
 
 #include "qvi-common.h"
 #include "qvi-utils.h"
+#include "qvi-bbuff.h"
 
 /** Group ID type. */
 using qvi_group_id_t = uint64_t;
@@ -82,7 +83,7 @@ struct qvi_group : qvi_refc {
     gather(
         qvi_bbuff *txbuff,
         int root,
-        bool *shared,
+        qvi_alloc_type_t  *shared,
         qvi_bbuff ***rxbuffs
     ) = 0;
     /** Scatters bbuffs from specified root. */

src/qvi-hwsplit.cc

@@ -641,7 +641,9 @@ qvi_hwsplit_coll::scatter_values(
     }
 
     rc = group->scatter(txbuffs.data(), rootid, &rxbuff);
-    if (qvi_unlikely(rc != QV_SUCCESS)) goto out;
+    if (qvi_unlikely(rc != QV_SUCCESS)) {
+        goto out;
+    }
 
     *value = *(TYPE *)rxbuff->data();
 out:
@@ -692,7 +694,7 @@ qvi_hwsplit_coll::gather_values(
         return rc;
     }
     // Gather the values to the root.
-    bool shared = false;
+    qvi_alloc_type_t shared = ALLOC_PRIVATE;
     qvi_bbuff **bbuffs = nullptr;
     rc = group->gather(txbuff, rootid, &shared, &bbuffs);
     if (qvi_unlikely(rc != QV_SUCCESS)) goto out;
@@ -705,14 +707,15 @@ qvi_hwsplit_coll::gather_values(
         }
     }
 out:
-    if (!shared || (shared && (group->rank() == rootid))) {
+    if ((ALLOC_PRIVATE == shared) || ((ALLOC_SHARED == shared) && (group->rank() == rootid))) {
         if (bbuffs) {
-            for (uint_t i = 0; i < group_size; ++i) {
+           for (uint_t i = 0; i < group_size; ++i) {
                 qvi_bbuff_delete(&bbuffs[i]);
             }
             delete[] bbuffs;
         }
     }
+
     qvi_bbuff_delete(&txbuff);
     if (qvi_unlikely(rc != QV_SUCCESS)) {
         // If something went wrong, just zero-initialize the values.
@@ -733,7 +736,7 @@ qvi_hwsplit_coll::gather_hwpools(
     int rc = txpool->packinto(&txbuff);
     if (qvi_unlikely(rc != QV_SUCCESS)) return rc;
     // Gather the values to the root.
-    bool shared = false;
+    qvi_alloc_type_t shared = ALLOC_PRIVATE;
     qvi_bbuff **bbuffs = nullptr;
     rc = group->gather(&txbuff, rootid, &shared, &bbuffs);
     if (rc != QV_SUCCESS) goto out;
@@ -749,14 +752,15 @@ qvi_hwsplit_coll::gather_hwpools(
         }
     }
 out:
-    if (!shared || (shared && (group->rank() == rootid))) {
+    if ((ALLOC_PRIVATE == shared) || ((ALLOC_SHARED == shared) && (group->rank() == rootid))) {
         if (bbuffs) {
             for (uint_t i = 0; i < group_size; ++i) {
                 qvi_bbuff_delete(&bbuffs[i]);
             }
             delete[] bbuffs;
         }
     }
+
     if (rc != QV_SUCCESS) {
         // If something went wrong, just zero-initialize the pools.
         rxpools = {};

src/qvi-mpi.cc

@@ -365,7 +365,7 @@ qvi_mpi_group_gather_bbuffs(
     qvi_mpi_group_t *group,
     qvi_bbuff *txbuff,
     int root,
-    bool *shared_alloc,
+    qvi_alloc_type_t *shared_alloc,
     qvi_bbuff ***rxbuffs
 ) {
     const int send_count = (int)txbuff->size();
@@ -434,7 +434,7 @@ qvi_mpi_group_gather_bbuffs(
         bbuffs = nullptr;
     }
     *rxbuffs = bbuffs;
-    *shared_alloc = false;
+    *shared_alloc = ALLOC_PRIVATE;
     return rc;
 }
 

src/qvi-mpi.h

@@ -20,6 +20,7 @@
 #include "qvi-common.h"
 #include "qvi-group.h"
 #include "quo-vadis-mpi.h" // IWYU pragma: keep
+#include "qvi-bbuff.h"
 
 // Forward declarations.
 struct qvi_mpi_s;
@@ -151,7 +152,7 @@ qvi_mpi_group_gather_bbuffs(
     qvi_mpi_group_t *group,
     qvi_bbuff *txbuff,
     int root,
-    bool *shared_alloc,
+    qvi_alloc_type_t *shared_alloc,
     qvi_bbuff ***rxbuffs
 );