MCA/UCX: added common module

- implemented non-blocking routines for flush operations

Signed-off-by: Sergey Oblomov <sergeyo@mellanox.com>

VERSION

@@ -112,5 +112,6 @@ libmca_orte_common_alps_so_version=0:0:0
 libmca_opal_common_cuda_so_version=0:0:0
 libmca_opal_common_ofi_so_version=0:0:0
 libmca_opal_common_sm_so_version=0:0:0
+libmca_opal_common_ucx_so_version=0:0:0
 libmca_opal_common_ugni_so_version=0:0:0
 libmca_opal_common_verbs_so_version=0:0:0

configure.ac

@@ -162,6 +162,7 @@ AC_SUBST(libmca_opal_common_verbs_so_version)
 AC_SUBST(libmca_orte_common_alps_so_version)
 AC_SUBST(libmca_ompi_common_ompio_so_version)
 AC_SUBST(libmca_ompi_common_monitoring_so_version)
+AC_SUBST(libmca_opal_common_ucx_so_version)
 
 #
 # Get the versions of the autotools that were used to bootstrap us

ompi/mca/osc/ucx/Makefile.am

@@ -34,8 +34,8 @@ endif
 mcacomponentdir = $(pkglibdir)
 mcacomponent_LTLIBRARIES = $(component_install)
 mca_osc_ucx_la_SOURCES = $(ucx_sources)
-mca_osc_ucx_la_LIBADD = $(top_builddir)/ompi/lib@OMPI_LIBMPI_NAME@.la \
-	$(osc_ucx_LIBS)
+mca_osc_ucx_la_LIBADD = $(top_builddir)/ompi/lib@OMPI_LIBMPI_NAME@.la $(osc_ucx_LIBS) \
+                        $(OPAL_TOP_BUILDDIR)/opal/mca/common/ucx/lib@OPAL_LIB_PREFIX@mca_common_ucx.la
 mca_osc_ucx_la_LDFLAGS = -module -avoid-version $(osc_ucx_LDFLAGS)
 
 noinst_LTLIBRARIES = $(component_noinst)

ompi/mca/osc/ucx/osc_ucx_active_target.c

@@ -28,6 +28,7 @@
 #include "ompi/mca/osc/osc.h"
 #include "ompi/mca/osc/base/base.h"
 #include "ompi/mca/osc/base/osc_base_obj_convert.h"
+#include "opal/mca/common/ucx/common_ucx.h"
 
 #include "osc_ucx.h"
 
@@ -73,7 +74,7 @@ int ompi_osc_ucx_fence(int assert, struct ompi_win_t *win) {
     }
 
     if (!(assert & MPI_MODE_NOPRECEDE)) {
-        status = ucp_worker_flush(mca_osc_ucx_component.ucp_worker);
+        status = opal_common_ucx_worker_flush(mca_osc_ucx_component.ucp_worker);
         if (status != UCS_OK) {
             opal_output_verbose(1, ompi_osc_base_framework.framework_output,
                                 "%s:%d: ucp_worker_flush failed: %d\n",
@@ -175,7 +176,7 @@ int ompi_osc_ucx_complete(struct ompi_win_t *win) {
 
     module->epoch_type.access = NONE_EPOCH;
 
-    status = ucp_worker_flush(mca_osc_ucx_component.ucp_worker);
+    status = opal_common_ucx_worker_flush(mca_osc_ucx_component.ucp_worker);
     if (status != UCS_OK) {
         opal_output_verbose(1, ompi_osc_base_framework.framework_output,
                             "%s:%d: ucp_worker_flush failed: %d\n",
@@ -200,7 +201,7 @@ int ompi_osc_ucx_complete(struct ompi_win_t *win) {
                                 __FILE__, __LINE__, status);
         }
 
-        status = ucp_ep_flush(ep);
+        status = opal_common_ucx_ep_flush(ep, mca_osc_ucx_component.ucp_worker);
         if (status != UCS_OK) {
             opal_output_verbose(1, ompi_osc_base_framework.framework_output,
                                 "%s:%d: ucp_ep_flush failed: %d\n",

ompi/mca/osc/ucx/osc_ucx_comm.c

@@ -12,6 +12,7 @@
 #include "ompi/mca/osc/osc.h"
 #include "ompi/mca/osc/base/base.h"
 #include "ompi/mca/osc/base/osc_base_obj_convert.h"
+#include "opal/mca/common/ucx/common_ucx.h"
 
 #include "osc_ucx.h"
 #include "osc_ucx_request.h"
@@ -65,9 +66,7 @@ static inline int incr_and_check_ops_num(ompi_osc_ucx_module_t *module, int targ
     module->global_ops_num++;
     module->per_target_ops_nums[target]++;
     if (module->global_ops_num >= OSC_UCX_OPS_THRESHOLD) {
-        /* TODO: ucp_ep_flush needs to be replaced with its non-blocking counterpart
-         * when it is implemented in UCX */
-        status = ucp_ep_flush(ep);
+        status = opal_common_ucx_ep_flush(ep, mca_osc_ucx_component.ucp_worker);
         if (status != UCS_OK) {
             opal_output_verbose(1, ompi_osc_base_framework.framework_output,
                                 "%s:%d: ucp_ep_flush failed: %d\n",
@@ -348,7 +347,7 @@ static inline int get_dynamic_win_info(uint64_t remote_addr, ompi_osc_ucx_module
         return OMPI_ERROR;
     }
 
-    status = ucp_ep_flush(ep);
+    status = opal_common_ucx_ep_flush(ep, mca_osc_ucx_component.ucp_worker);
     if (status != UCS_OK) {
         opal_output_verbose(1, ompi_osc_base_framework.framework_output,
                             "%s:%d: ucp_ep_flush failed: %d\n",
@@ -550,7 +549,7 @@ int ompi_osc_ucx_accumulate(const void *origin_addr, int origin_count,
             return ret;
         }
 
-        status = ucp_ep_flush(ep);
+        status = opal_common_ucx_ep_flush(ep, mca_osc_ucx_component.ucp_worker);
         if (status != UCS_OK) {
             opal_output_verbose(1, ompi_osc_base_framework.framework_output,
                                 "%s:%d: ucp_ep_flush failed: %d\n",
@@ -607,7 +606,7 @@ int ompi_osc_ucx_accumulate(const void *origin_addr, int origin_count,
             return ret;
         }
 
-        status = ucp_ep_flush(ep);
+        status = opal_common_ucx_ep_flush(ep, mca_osc_ucx_component.ucp_worker);
         if (status != UCS_OK) {
             opal_output_verbose(1, ompi_osc_base_framework.framework_output,
                                 "%s:%d: ucp_ep_flush failed: %d\n",
@@ -801,7 +800,7 @@ int ompi_osc_ucx_get_accumulate(const void *origin_addr, int origin_count,
                 return ret;
             }
 
-            status = ucp_ep_flush(ep);
+            status = opal_common_ucx_ep_flush(ep, mca_osc_ucx_component.ucp_worker);
             if (status != UCS_OK) {
                 opal_output_verbose(1, ompi_osc_base_framework.framework_output,
                                     "%s:%d: ucp_ep_flush failed: %d\n",
@@ -857,7 +856,7 @@ int ompi_osc_ucx_get_accumulate(const void *origin_addr, int origin_count,
                 return ret;
             }
 
-            status = ucp_ep_flush(ep);
+            status = opal_common_ucx_ep_flush(ep, mca_osc_ucx_component.ucp_worker);
             if (status != UCS_OK) {
                 opal_output_verbose(1, ompi_osc_base_framework.framework_output,
                                     "%s:%d: ucp_ep_flush failed: %d\n",

ompi/mca/osc/ucx/osc_ucx_passive_target.c

@@ -12,6 +12,7 @@
 #include "ompi/mca/osc/osc.h"
 #include "ompi/mca/osc/base/base.h"
 #include "ompi/mca/osc/base/osc_base_obj_convert.h"
+#include "opal/mca/common/ucx/common_ucx.h"
 
 #include "osc_ucx.h"
 
@@ -179,7 +180,7 @@ int ompi_osc_ucx_unlock(int target, struct ompi_win_t *win) {
                                         (uint32_t)target);
 
     ep = OSC_UCX_GET_EP(module->comm, target);
-    status = ucp_ep_flush(ep);
+    status = opal_common_ucx_ep_flush(ep, mca_osc_ucx_component.ucp_worker);
     if (status != UCS_OK) {
         opal_output_verbose(1, ompi_osc_base_framework.framework_output,
                             "%s:%d: ucp_ep_flush failed: %d\n",
@@ -258,7 +259,7 @@ int ompi_osc_ucx_unlock_all(struct ompi_win_t *win) {
 
     assert(module->lock_count == 0);
 
-    status = ucp_worker_flush(mca_osc_ucx_component.ucp_worker);
+    status = opal_common_ucx_worker_flush(mca_osc_ucx_component.ucp_worker);
     if (status != UCS_OK) {
         opal_output_verbose(1, ompi_osc_base_framework.framework_output,
                             "%s:%d: ucp_worker_flush failed: %d\n",
@@ -314,7 +315,7 @@ int ompi_osc_ucx_flush(int target, struct ompi_win_t *win) {
     }
 
     ep = OSC_UCX_GET_EP(module->comm, target);
-    status = ucp_ep_flush(ep);
+    status = opal_common_ucx_ep_flush(ep, mca_osc_ucx_component.ucp_worker);
     if (status != UCS_OK) {
         opal_output_verbose(1, ompi_osc_base_framework.framework_output,
                             "%s:%d: ucp_ep_flush failed: %d\n",
@@ -337,7 +338,7 @@ int ompi_osc_ucx_flush_all(struct ompi_win_t *win) {
         return OMPI_ERR_RMA_SYNC;
     }
 
-    status = ucp_worker_flush(mca_osc_ucx_component.ucp_worker);
+    status = opal_common_ucx_worker_flush(mca_osc_ucx_component.ucp_worker);
     if (status != UCS_OK) {
         opal_output_verbose(1, ompi_osc_base_framework.framework_output,
                             "%s:%d: ucp_worker_flush failed: %d\n",

ompi/mca/pml/ucx/Makefile.am

@@ -37,8 +37,8 @@ endif
 mcacomponentdir = $(ompilibdir)
 mcacomponent_LTLIBRARIES = $(component_install)
 mca_pml_ucx_la_SOURCES = $(local_sources)
-mca_pml_ucx_la_LIBADD = $(top_builddir)/ompi/lib@OMPI_LIBMPI_NAME@.la \
-	$(pml_ucx_LIBS)
+mca_pml_ucx_la_LIBADD = $(top_builddir)/ompi/lib@OMPI_LIBMPI_NAME@.la $(pml_ucx_LIBS) \
+                        $(OPAL_TOP_BUILDDIR)/opal/mca/common/ucx/lib@OPAL_LIB_PREFIX@mca_common_ucx.la
 mca_pml_ucx_la_LDFLAGS = -module -avoid-version $(pml_ucx_LDFLAGS)
 
 noinst_LTLIBRARIES = $(component_noinst)

opal/mca/common/ucx/Makefile.am

@@ -0,0 +1,92 @@
+#
+# Copyright (c) 2018      Mellanox Technologies.  All rights reserved.
+# $COPYRIGHT$
+#
+# Additional copyrights may follow
+#
+# $HEADER$
+#
+
+# Note that building this common component statically and linking
+# against other dynamic components is *not* supported!
+
+# Header files
+
+headers = \
+        common_ucx.h
+
+# Source files
+
+sources = \
+        common_ucx.c
+
+# Help file
+
+# As per above, we'll either have an installable or noinst result.
+# The installable one should follow the same MCA prefix naming rules
+# (i.e., libmca_<type>_<name>.la).  The noinst one can be named
+# whatever it wants, although libmca_<type>_<name>_noinst.la is
+# recommended.
+
+# To simplify components that link to this library, we will *always*
+# have an output libtool library named libmca_<type>_<name>.la -- even
+# for case 2) described above (i.e., so there's no conditional logic
+# necessary in component Makefile.am's that link to this library).
+# Hence, if we're creating a noinst version of this library (i.e.,
+# case 2), we sym link it to the libmca_<type>_<name>.la name
+# (libtool will do the Right Things under the covers).  See the
+# all-local and clean-local rules, below, for how this is effected.
+
+lib_LTLIBRARIES =
+noinst_LTLIBRARIES =
+comp_inst = lib@OPAL_LIB_PREFIX@mca_common_ucx.la
+comp_noinst = lib@OPAL_LIB_PREFIX@mca_common_ucx_noinst.la
+
+if MCA_BUILD_opal_common_ucx_DSO
+lib_LTLIBRARIES += $(comp_inst)
+else
+noinst_LTLIBRARIES += $(comp_noinst)
+endif
+
+lib@OPAL_LIB_PREFIX@mca_common_ucx_la_SOURCES = \
+        $(headers) $(sources)
+lib@OPAL_LIB_PREFIX@mca_common_ucx_la_LDFLAGS = \
+        -version-info $(libmca_opal_common_ucx_so_version) \
+		$(common_ucx_LDFLAGS)
+lib@OPAL_LIB_PREFIX@mca_common_ucx_noinst_la_LDFLAGS = \
+		$(common_ucx_LDFLAGS)
+lib@OPAL_LIB_PREFIX@mca_common_ucx_la_LIBADD = \
+		$(common_ucx_LIBS) \
+		$(OMPI_TOP_BUILDDIR)/opal/lib@OPAL_LIB_PREFIX@open-pal.la
+lib@OPAL_LIB_PREFIX@mca_common_ucx_noinst_la_LIBADD = \
+		$(common_ucx_LIBS)
+lib@OPAL_LIB_PREFIX@mca_common_ucx_noinst_la_SOURCES = \
+        $(headers) $(sources)
+
+# Conditionally install the header files
+
+if WANT_INSTALL_HEADERS
+opaldir = $(opalincludedir)/$(subdir)
+opal_HEADERS = $(headers)
+endif
+
+# These two rules will sym link the "noinst" libtool library filename
+# to the installable libtool library filename in the case where we are
+# compiling this component statically (case 2), described above).
+
+# See Makefile.ompi-rules for an explanation of the "V" macros, below
+V=0
+OMPI_V_LN_SCOMP = $(ompi__v_LN_SCOMP_$V)
+ompi__v_LN_SCOMP_ = $(ompi__v_LN_SCOMP_$AM_DEFAULT_VERBOSITY)
+ompi__v_LN_SCOMP_0 = @echo "  LN_S    " `basename $(comp_inst)`;
+
+all-local:
+	$(OMPI_V_LN_SCOMP) if test -z "$(lib_LTLIBRARIES)"; then \
+	  rm -f "$(comp_inst)"; \
+	  $(LN_S) "$(comp_noinst)" "$(comp_inst)"; \
+	fi
+
+clean-local:
+	if test -z "$(lib_LTLIBRARIES)"; then \
+	  rm -f "$(comp_inst)"; \
+	fi

opal/mca/common/ucx/common_ucx.c

@@ -0,0 +1,34 @@
+/*
+ * Copyright (C) Mellanox Technologies Ltd. 2018. ALL RIGHTS RESERVED.
+ * $COPYRIGHT$
+ *
+ * Additional copyrights may follow
+ *
+ * $HEADER$
+ */
+
+#include "opal_config.h"
+
+#include "common_ucx.h"
+
+/***********************************************************************/
+
+static void opal_common_ucp_send_cb(void *request, ucs_status_t status)
+{
+}
+
+ucs_status_t opal_common_ucx_ep_flush(ucp_ep_h ep, ucp_worker_h worker)
+{
+    ucs_status_ptr_t status;
+
+    status = ucp_ep_flush_nb(ep, 0, opal_common_ucp_send_cb);
+    return opal_common_ucx_wait_request(status, worker);
+}
+
+ucs_status_t opal_common_ucx_worker_flush(ucp_worker_h worker)
+{
+    ucs_status_ptr_t status;
+
+    status = ucp_worker_flush_nb(worker, 0, opal_common_ucp_send_cb);
+    return opal_common_ucx_wait_request(status, worker);
+}

opal/mca/common/ucx/common_ucx.h

@@ -0,0 +1,60 @@
+/*
+ * Copyright (c) 2018      Mellanox Technologies.  All rights reserved.
+ *                         All rights reserved.
+ * $COPYRIGHT$
+ *
+ * Additional copyrights may follow
+ *
+ * $HEADER$
+ */
+
+#ifndef _COMMON_UCX_H_
+#define _COMMON_UCX_H_
+
+#include "opal_config.h"
+
+#include <stdint.h>
+
+#include <ucp/api/ucp.h>
+
+#include "opal/mca/mca.h"
+#include "opal/runtime/opal_progress.h"
+#include "opal/class/opal_list.h"
+
+BEGIN_C_DECLS
+
+OPAL_DECLSPEC ucs_status_t opal_common_ucx_ep_flush(ucp_ep_h ep, ucp_worker_h worker);
+OPAL_DECLSPEC ucs_status_t opal_common_ucx_worker_flush(ucp_worker_h worker);
+
+static inline
+ucs_status_t opal_common_ucx_wait_request(ucs_status_ptr_t request, ucp_worker_h worker)
+{
+    ucs_status_t status;
+    int i;
+
+    /* check for request completed or failed */
+    if (UCS_OK == request) {
+        return UCS_OK;
+    } else if (UCS_PTR_IS_ERR(request)) {
+        return UCS_PTR_STATUS(request);
+    }
+
+    while (1) {
+        /* call UCX progress */
+        for (i = 0; i < 100; i++) {
+            if (UCS_INPROGRESS != (status = ucp_request_check_status(request))) {
+                ucp_request_free(request);
+                return status;
+            }
+            ucp_worker_progress(worker);
+        }
+        /* call OPAL progress on every 100 call to UCX progress */
+        opal_progress();
+    }
+}
+
+END_C_DECLS
+
+#endif
+
+

opal/mca/common/ucx/configure.m4

@@ -0,0 +1,33 @@
+# -*- shell-script -*-
+#
+# Copyright (c) 2018      Mellanox Technologies.  All rights reserved.
+# $COPYRIGHT$
+#
+# Additional copyrights may follow
+#
+# $HEADER$
+#
+
+# MCA_opal_common_sm_CONFIG([action-if-can-compile],
+#                           [action-if-cant-compile])
+# ------------------------------------------------
+AC_DEFUN([MCA_opal_common_ucx_CONFIG],[
+    AC_CONFIG_FILES([opal/mca/common/ucx/Makefile])
+    common_ucx_happy="no"
+    OMPI_CHECK_UCX([common_ucx],
+               [common_ucx_happy="yes"],
+               [common_ucx_happy="no"])
+
+    AS_IF([test "$common_ucx_happy" = "yes"],
+          [$1],
+          [$2])
+
+
+    # substitute in the things needed to build openib
+    AC_SUBST([common_ucx_CFLAGS])
+    AC_SUBST([common_ucx_CPPFLAGS])
+    AC_SUBST([common_ucx_LDFLAGS])
+    AC_SUBST([common_ucx_LIBS])
+])dnl
+
+

opal/mca/common/ucx/owner.txt

@@ -0,0 +1,8 @@
+#
+# owner/status file
+# owner: institution that is responsible for this package
+# status: e.g. active, maintenance, unmaintained
+#
+owner: MELLANOX
+status: maintenance
+

oshmem/mca/atomic/ucx/atomic_ucx.h

@@ -13,6 +13,7 @@
 
 #include "oshmem_config.h"
 
+#include "opal/mca/common/ucx/common_ucx.h"
 #include "opal/mca/mca.h"
 #include "oshmem/mca/atomic/atomic.h"
 #include "oshmem/util/oshmem_util.h"
@@ -66,28 +67,7 @@ void mca_atomic_ucx_complete_cb(void *request, ucs_status_t status);
 static inline
 ucs_status_t mca_atomic_ucx_wait_request(ucs_status_ptr_t request)
 {
-    ucs_status_t status;
-    int i;
-
-    /* check for request completed or failed */
-    if (UCS_OK == request) {
-        return UCS_OK;
-    } else if (UCS_PTR_IS_ERR(request)) {
-        return UCS_PTR_STATUS(request);
-    }
-
-    while (1) {
-        /* call UCX progress */
-        for (i = 0; i < 100; i++) {
-            if (UCS_INPROGRESS != (status = ucp_request_check_status(request))) {
-                ucp_request_free(request);
-                return status;
-            }
-            ucp_worker_progress(mca_spml_self->ucp_worker);
-        }
-        /* call OPAL progress on every 100 call to UCX progress */
-        opal_progress();
-    }
+    return opal_common_ucx_wait_request(request, mca_spml_self->ucp_worker);
 }
 END_C_DECLS
 

oshmem/mca/spml/ucx/Makefile.am

@@ -34,7 +34,8 @@ mcacomponentdir = $(ompilibdir)
 mcacomponent_LTLIBRARIES = $(component_install)
 mca_spml_ucx_la_SOURCES = $(ucx_sources)
 mca_spml_ucx_la_LIBADD = $(top_builddir)/oshmem/liboshmem.la \
-	$(spml_ucx_LIBS)
+	$(spml_ucx_LIBS) \
+	$(OPAL_TOP_BUILDDIR)/opal/mca/common/ucx/lib@OPAL_LIB_PREFIX@mca_common_ucx.la
 mca_spml_ucx_la_LDFLAGS = -module -avoid-version $(spml_ucx_LDFLAGS)
 
 noinst_LTLIBRARIES = $(component_noinst)

oshmem/mca/spml/ucx/spml_ucx.c

@@ -20,6 +20,7 @@
 
 #include "oshmem_config.h"
 #include "opal/datatype/opal_convertor.h"
+#include "opal/mca/common/ucx/common_ucx.h"
 #include "orte/include/orte/types.h"
 #include "orte/runtime/orte_globals.h"
 #include "ompi/datatype/ompi_datatype.h"
@@ -611,7 +612,7 @@ int mca_spml_ucx_quiet(void)
 {
     ucs_status_t err;
 
-    err = ucp_worker_flush(mca_spml_ucx.ucp_worker);
+    err = opal_common_ucx_worker_flush(mca_spml_ucx.ucp_worker);
     if (UCS_OK != err) {
          SPML_ERROR("quiet failed: %s", ucs_status_string(err));
          oshmem_shmem_abort(-1);