atomic add for nb outstanding ops and renaming the thread enabled var

Author: Mamzi Bayatpour [email protected] ()

ompi/mca/osc/ucx/osc_ucx.h

@@ -46,14 +46,14 @@ typedef struct ompi_osc_ucx_component {
 
 OMPI_DECLSPEC extern ompi_osc_ucx_component_t mca_osc_ucx_component;
 
-#define OSC_UCX_INCREMENT_OUTSTANDING_NB_OPS(_module)               \
-    do {                                                            \
-        _module->ctx->num_incomplete_req_ops++;                     \
+#define OSC_UCX_INCREMENT_OUTSTANDING_NB_OPS(_module)                               \
+    do {                                                                            \
+        opal_atomic_add_fetch_size_t(&_module->ctx->num_incomplete_req_ops, 1);     \
     } while(0);
 
-#define OSC_UCX_DECREMENT_OUTSTANDING_NB_OPS(_module)               \
-    do {                                                            \
-        _module->ctx->num_incomplete_req_ops--;                     \
+#define OSC_UCX_DECREMENT_OUTSTANDING_NB_OPS(_module)                               \
+    do {                                                                            \
+        opal_atomic_add_fetch_size_t(&_module->ctx->num_incomplete_req_ops, -1);    \
     } while(0);
 
 typedef enum ompi_osc_ucx_epoch {
@@ -164,10 +164,10 @@ typedef struct ompi_osc_ucx_lock {
 #define OSC_UCX_GET_EP(_module, rank_) (mca_osc_ucx_component.endpoints[_module->comm_world_ranks[rank_]])
 #define OSC_UCX_GET_DISP(module_, rank_) ((module_->disp_unit < 0) ? module_->disp_units[rank_] : module_->disp_unit)
 
-extern bool opal_mca_common_ucx_mpi_thread_multiple_enabled;
+extern bool thread_enabled;
 
 #define OSC_UCX_GET_DEFAULT_EP(_ep_ptr, _module, _target)                   \
-    if (opal_mca_common_ucx_mpi_thread_multiple_enabled) {                  \
+    if (thread_enabled) {                  \
         _ep_ptr = NULL;                                                     \
     } else {                                                                \
         _ep_ptr = (ucp_ep_h *)&(OSC_UCX_GET_EP(_module, _target));          \

ompi/mca/osc/ucx/osc_ucx_component.c

@@ -175,7 +175,7 @@ static int component_register(void) {
                                            MCA_BASE_VAR_SCOPE_GROUP, &mca_osc_ucx_component.no_locks);
     free(description_str);
 
-    opal_mca_common_ucx_mpi_thread_multiple_enabled = opal_using_threads();
+    thread_enabled = opal_using_threads();
     mca_osc_ucx_component.acc_single_intrinsic = false;
     opal_asprintf(&description_str, "Enable optimizations for MPI_Fetch_and_op, MPI_Accumulate, etc for codes "
              "that will not use anything more than a single predefined datatype (default: %s)",
@@ -188,7 +188,7 @@ static int component_register(void) {
                                            MCA_BASE_VAR_SCOPE_GROUP, &enable_nonblocking_accumulate);
     (void) mca_base_component_var_register(&mca_osc_ucx_component.super.osc_version, "enable_wpool_thread_multiple",
                                            description_str, MCA_BASE_VAR_TYPE_BOOL, NULL, 0, 0, OPAL_INFO_LVL_5,
-                                           MCA_BASE_VAR_SCOPE_GROUP, &opal_mca_common_ucx_mpi_thread_multiple_enabled);
+                                           MCA_BASE_VAR_SCOPE_GROUP, &thread_enabled);
     (void) mca_base_component_var_register(&mca_osc_ucx_component.super.osc_version, "outstanding_ops_flush_threshold",
                                            description_str, MCA_BASE_VAR_TYPE_INT, NULL, 0, 0, OPAL_INFO_LVL_5,
                                            MCA_BASE_VAR_SCOPE_GROUP, &ompi_osc_ucx_outstanding_ops_flush_threshold);
@@ -298,7 +298,7 @@ static int component_init(bool enable_progress_threads, bool enable_mpi_threads)
 }
 
 static int component_finalize(void) {
-    if (!opal_mca_common_ucx_mpi_thread_multiple_enabled) {
+    if (!thread_enabled) {
         int i;
         for (i = 0; i < mca_osc_ucx_component.comm_world_size; i++) {
             ucp_ep_h ep = mca_osc_ucx_component.endpoints[i];
@@ -497,7 +497,7 @@ static int component_select(struct ompi_win_t *win, void **base, size_t size, in
             OSC_UCX_VERBOSE(1, "opal_common_ucx_wpool_init failed: %d", ret);
             goto select_unlock;
         }
-        if (!opal_mca_common_ucx_mpi_thread_multiple_enabled) {
+        if (!thread_enabled) {
             mca_osc_ucx_component.comm_world_size = ompi_proc_world_size();
             mca_osc_ucx_component.endpoints = calloc(mca_osc_ucx_component.comm_world_size, sizeof(ucp_ep_h));
         }

opal/mca/common/ucx/common_ucx_wpool.c

@@ -31,7 +31,7 @@ __thread FILE *tls_pf = NULL;
 __thread int initialized = 0;
 #endif
 
-bool opal_mca_common_ucx_mpi_thread_multiple_enabled;
+bool thread_enabled;
 
 static _ctx_record_t *_tlocal_add_ctx_rec(opal_common_ucx_ctx_t *ctx);
 static inline _ctx_record_t *_tlocal_get_ctx_rec(opal_tsd_tracked_key_t tls_key);
@@ -50,7 +50,7 @@ static opal_common_ucx_winfo_t *_winfo_create(opal_common_ucx_wpool_t *wpool)
     ucs_status_t status;
     opal_common_ucx_winfo_t *winfo = NULL;
 
-    if (opal_mca_common_ucx_mpi_thread_multiple_enabled || wpool->dflt_winfo == NULL) {
+    if (thread_enabled || wpool->dflt_winfo == NULL) {
         memset(&worker_params, 0, sizeof(worker_params));
         worker_params.field_mask = UCP_WORKER_PARAM_FIELD_THREAD_MODE;
         worker_params.thread_mode = UCS_THREAD_MODE_SINGLE;
@@ -98,7 +98,7 @@ static void _winfo_destructor(opal_common_ucx_winfo_t *winfo)
 
     if (winfo->comm_size != 0) {
         size_t i;
-        if (opal_mca_common_ucx_mpi_thread_multiple_enabled) {
+        if (thread_enabled) {
             for (i = 0; i < winfo->comm_size; i++) {
                 if (NULL != winfo->endpoints[i]) {
                     ucp_ep_destroy(winfo->endpoints[i]);
@@ -113,7 +113,7 @@ static void _winfo_destructor(opal_common_ucx_winfo_t *winfo)
     winfo->comm_size = 0;
 
     OBJ_DESTRUCT(&winfo->mutex);
-    if (opal_mca_common_ucx_mpi_thread_multiple_enabled || winfo->is_dflt_winfo) {
+    if (thread_enabled || winfo->is_dflt_winfo) {
         ucp_worker_destroy(winfo->worker);
     }
 
@@ -732,13 +732,13 @@ OPAL_DECLSPEC int opal_common_ucx_tlocal_fetch_spath(opal_common_ucx_wpmem_t *me
 
     /* Obtain the endpoint */
     if (OPAL_UNLIKELY(NULL == winfo->endpoints[target])) {
-        if (opal_mca_common_ucx_mpi_thread_multiple_enabled || (dflt_ep == NULL) ||
+        if (thread_enabled || (dflt_ep == NULL) ||
                 (*dflt_ep == NULL)) {
             rc = _tlocal_ctx_connect(ctx_rec, target);
             if (rc != OPAL_SUCCESS) {
                 return rc;
             }
-            if (!opal_mca_common_ucx_mpi_thread_multiple_enabled && (dflt_ep != NULL) &&
+            if (!thread_enabled && (dflt_ep != NULL) &&
                     (*dflt_ep == NULL)) {
                 /* set the proc ep */
                 *dflt_ep = winfo->endpoints[target];

opal/mca/common/ucx/common_ucx_wpool.h

@@ -86,7 +86,7 @@ typedef struct {
     char *recv_worker_addrs;
     int *recv_worker_displs;
     size_t comm_size;
-    int num_incomplete_req_ops;
+    opal_atomic_size_t num_incomplete_req_ops;
 } opal_common_ucx_ctx_t;
 
 /* Worker Pool memory (wpmem) is an object that represents a remotely accessible