OSC/UCX: Fix coverity issues and missing extend var in nb acc

Signed-off-by: Mamzi Bayatpour  <mbayatpour@nvidia.com>
Co-authored-by: Tomislav Janjusic <tomislavj@nvidia.com>

Author: Mamzi Bayatpour mbayatpour@nvidia.com ()

ompi/mca/osc/ucx/osc_ucx.h

@@ -49,12 +49,12 @@ OMPI_DECLSPEC extern ompi_osc_ucx_component_t mca_osc_ucx_component;
 
 #define OSC_UCX_INCREMENT_OUTSTANDING_NB_OPS(_module)                               \
     do {                                                                            \
-        opal_atomic_add_fetch_size_t(&_module->ctx->num_incomplete_req_ops, 1);     \
+        opal_atomic_add_fetch_64(&_module->ctx->num_incomplete_req_ops, 1);         \
     } while(0);
 
 #define OSC_UCX_DECREMENT_OUTSTANDING_NB_OPS(_module)                               \
     do {                                                                            \
-        opal_atomic_add_fetch_size_t(&_module->ctx->num_incomplete_req_ops, -1);    \
+        opal_atomic_add_fetch_64(&_module->ctx->num_incomplete_req_ops, -1);        \
     } while(0);
 
 typedef enum ompi_osc_ucx_epoch {

ompi/mca/osc/ucx/osc_ucx_comm.c

@@ -263,7 +263,7 @@ static inline int get_dynamic_win_info(uint64_t remote_addr,
     OSC_UCX_GET_DEFAULT_EP(ep, module, target);
     uint64_t remote_state_addr = (module->state_addrs)[target] + OSC_UCX_STATE_DYNAMIC_WIN_CNT_OFFSET;
     size_t remote_state_len = sizeof(uint64_t) + sizeof(ompi_osc_dynamic_win_info_t) * OMPI_OSC_UCX_ATTACH_MAX;
-    char *temp_buf;
+    char *temp_buf = NULL;
     ompi_osc_dynamic_win_info_t *temp_dynamic_wins;
     int contain = 0;
     uint64_t win_count;
@@ -870,11 +870,10 @@ static inline int ompi_osc_ucx_acc_rputget(void *stage_addr, int stage_count,
             }
         }
         ucx_req->target = target;
-        if (target_dt != NULL) {
-            ucx_req->target_dt = target_dt;
-            if (!ompi_datatype_is_predefined(target_dt)) {
-                OBJ_RETAIN(ucx_req->target_dt);
-            }
+        assert(target_dt != NULL);
+        ucx_req->target_dt = target_dt;
+        if (!ompi_datatype_is_predefined(target_dt)) {
+            OBJ_RETAIN(ucx_req->target_dt);
         }
         ucx_req->target_disp = target_disp;
         ucx_req->target_count = target_count;
@@ -884,9 +883,11 @@ static inline int ompi_osc_ucx_acc_rputget(void *stage_addr, int stage_count,
     module->skip_sync_check = true; /* we already hold the acc lock, so no need for sync check*/
 
     if (is_put) {
+        assert(origin_dt != NULL);
         ret = ompi_osc_ucx_put(origin_addr, origin_count, origin_dt, target, target_disp,
                                target_count, target_dt, win);
     } else {
+        assert(stage_dt != NULL);
         ret = ompi_osc_ucx_get(stage_addr, stage_count, stage_dt, target, target_disp,
                 target_count, target_dt, win);
     }
@@ -1615,7 +1616,7 @@ void ompi_osc_ucx_req_completion(void *request) {
         assert(req->phase != ACC_INIT);
         void *free_addr = NULL;
         bool release_lock = false;
-        ptrdiff_t temp_extent;
+        ptrdiff_t temp_lb, temp_extent;
         const void *origin_addr = req->origin_addr;
         int origin_count = req->origin_count;
         struct ompi_datatype_t *origin_dt = req->origin_dt;
@@ -1723,6 +1724,7 @@ void ompi_osc_ucx_req_completion(void *request) {
                     } else {
                         int i;
                         void *curr_origin_addr = origin_ucx_iov[origin_ucx_iov_idx].addr;
+                        ompi_datatype_get_true_extent(temp_dt, &temp_lb, &temp_extent);
                         for (i = 0; i < (int)temp_count; i++) {
                             ompi_op_reduce(op, curr_origin_addr,
                                            (void *)((char *)temp_addr + i * temp_extent),

ompi/mca/osc/ucx/osc_ucx_component.c

@@ -650,6 +650,7 @@ static int component_select(struct ompi_win_t *win, void **base, size_t size, in
         module->noncontig_shared_win = false;
         if (OMPI_SUCCESS != opal_info_get_bool(info, "alloc_shared_noncontig",
                                                &module->noncontig_shared_win, &flag)) {
+            free(rbuf);
             goto error;
         }
 
@@ -879,6 +880,7 @@ static int component_select(struct ompi_win_t *win, void **base, size_t size, in
     if (module->disp_units) free(module->disp_units);
     if (module->comm) ompi_comm_free(&module->comm);
     free(module);
+    module = NULL;
 
 error_nomem:
     if (env_initialized == true) {
@@ -887,7 +889,7 @@ static int component_select(struct ompi_win_t *win, void **base, size_t size, in
         mca_osc_ucx_component.env_initialized = false;
     }
 
-    if (0 == ompi_comm_rank (module->comm) && unlink_needed) {
+    if ((NULL != module) && (0 == ompi_comm_rank (module->comm)) && unlink_needed) {
         opal_shmem_unlink (&module->seg_ds);
     }
     ompi_osc_ucx_unregister_progress();
@@ -1082,7 +1084,10 @@ int ompi_osc_ucx_free(struct ompi_win_t *win) {
     }
     OBJ_DESTRUCT(&module->pending_posts);
 
-    opal_common_ucx_ctx_flush(module->ctx, OPAL_COMMON_UCX_SCOPE_WORKER, 0);
+    ret = opal_common_ucx_ctx_flush(module->ctx, OPAL_COMMON_UCX_SCOPE_WORKER, 0);
+    if (OPAL_SUCCESS != ret) {
+        return ret;
+    }
 
     ret = module->comm->c_coll->coll_barrier(module->comm,
                                              module->comm->c_coll->coll_barrier_module);

opal/mca/common/ucx/common_ucx_wpool.c

@@ -872,7 +872,7 @@ OPAL_DECLSPEC int opal_common_ucx_ctx_flush(opal_common_ucx_ctx_t *ctx,
     }
 
     /* progress the nonblocking operations */
-    while (ctx->num_incomplete_req_ops != 0) {
+    while (ctx->num_incomplete_req_ops > 0) {
         spin++;
         rc = ctx_flush(ctx, OPAL_COMMON_UCX_SCOPE_WORKER, 0);
         if (rc != OPAL_SUCCESS) {

opal/mca/common/ucx/common_ucx_wpool.h

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