ADAPT: Event-driven collective implementation

ompi/communicator/comm_init.c

@@ -40,6 +40,7 @@
 #include "ompi/constants.h"
 #include "ompi/mca/pml/pml.h"
 #include "ompi/mca/coll/base/base.h"
+#include "ompi/mca/coll/base/coll_tags.h"
 #include "ompi/mca/topo/base/base.h"
 #include "ompi/runtime/params.h"
 #include "ompi/communicator/communicator.h"
@@ -382,6 +383,7 @@ static void ompi_comm_construct(ompi_communicator_t* comm)
     comm->c_pml_comm     = NULL;
     comm->c_topo         = NULL;
     comm->c_coll         = NULL;
+    comm->c_nbc_tag      = MCA_COLL_BASE_TAG_NONBLOCKING_BASE;
 
     /* A keyhash will be created if/when an attribute is cached on
        this communicator */

ompi/communicator/communicator.h

@@ -3,7 +3,7 @@
  * Copyright (c) 2004-2005 The Trustees of Indiana University and Indiana
  *                         University Research and Technology
  *                         Corporation.  All rights reserved.
- * Copyright (c) 2004-2017 The University of Tennessee and The University
+ * Copyright (c) 2004-2020 The University of Tennessee and The University
  *                         of Tennessee Research Foundation.  All rights
  *                         reserved.
  * Copyright (c) 2004-2005 High Performance Computing Center Stuttgart,
@@ -187,6 +187,13 @@ struct ompi_communicator_t {
 
     /* Collectives module interface and data */
     mca_coll_base_comm_coll_t *c_coll;
+
+    /* Non-blocking collective tag. These tags might be shared between
+     * all non-blocking collective modules (to avoid message collision
+     * between them in the case where multiple outstanding non-blocking
+     * collective coexists using multiple backends).
+     */
+    opal_atomic_int32_t c_nbc_tag;
 };
 typedef struct ompi_communicator_t ompi_communicator_t;
 

ompi/mca/coll/adapt/Makefile.am

@@ -0,0 +1,49 @@
+#
+# Copyright (c) 2014-2020 The University of Tennessee and The University
+#                         of Tennessee Research Foundation.  All rights
+#                         reserved.
+# $COPYRIGHT$
+# 
+# Additional copyrights may follow
+# 
+# $HEADER$
+#
+
+
+sources = \
+	coll_adapt_component.c \
+	coll_adapt_module.c \
+	coll_adapt_bcast.c \
+	coll_adapt_ibcast.c \
+	coll_adapt_reduce.c \
+	coll_adapt_ireduce.c \
+	coll_adapt.h \
+	coll_adapt_algorithms.h \
+	coll_adapt_context.h \
+	coll_adapt_context.c \
+	coll_adapt_inbuf.c \
+	coll_adapt_inbuf.h \
+	coll_adapt_item.c \
+	coll_adapt_item.h 
+
+# Make the output library in this directory, and name it either
+# mca_<type>_<name>.la (for DSO builds) or libmca_<type>_<name>.la
+# (for static builds).
+
+component_noinst =
+component_install =
+if MCA_BUILD_ompi_coll_adapt_DSO
+component_install += mca_coll_adapt.la
+else
+component_noinst += libmca_coll_adapt.la
+endif
+
+mcacomponentdir = $(ompilibdir)
+mcacomponent_LTLIBRARIES = $(component_install)
+mca_coll_adapt_la_SOURCES = $(sources)
+mca_coll_adapt_la_LDFLAGS = -module -avoid-version
+mca_coll_adapt_la_LIBADD =
+
+noinst_LTLIBRARIES = $(component_noinst)
+libmca_coll_adapt_la_SOURCES =$(sources)
+libmca_coll_adapt_la_LDFLAGS = -module -avoid-version

ompi/mca/coll/adapt/coll_adapt.h

@@ -0,0 +1,96 @@
+/*
+ * Copyright (c) 2014-2020 The University of Tennessee and The University
+ *                         of Tennessee Research Foundation.  All rights
+ *                         reserved.
+ * $COPYRIGHT$
+ * 
+ * Additional copyrights may follow
+ * 
+ * $HEADER$
+ */
+
+
+#ifndef MCA_COLL_ADAPT_EXPORT_H
+#define MCA_COLL_ADAPT_EXPORT_H
+
+#include "ompi_config.h"
+
+#include "mpi.h"
+#include "opal/mca/mca.h"
+#include "opal/datatype/opal_convertor.h"
+#include "ompi/mca/coll/coll.h"
+#include "ompi/mca/coll/base/coll_base_topo.h"
+
+BEGIN_C_DECLS
+
+typedef struct mca_coll_adapt_module_t mca_coll_adapt_module_t;
+
+/*
+ * Structure to hold the adapt coll component.  First it holds the
+ * base coll component, and then holds a bunch of
+ * adapt-coll-component-specific stuff (e.g., current MCA param
+ * values).
+ */
+typedef struct mca_coll_adapt_component_t {
+    /* Base coll component */
+    mca_coll_base_component_2_0_0_t super;
+
+    /* MCA parameter: Priority of this component */
+    int adapt_priority;
+
+    /* MCA parameter: Output stream and verbose level */
+    int adapt_output;
+    int adapt_verbose;
+
+    /* MCA parameter: Maximum number of segment in context free list */
+    int adapt_context_free_list_max;
+
+    /* MCA parameter: Minimum number of segment in context free list */
+    int adapt_context_free_list_min;
+
+    /* MCA parameter: Increasement number of segment in context free list */
+    int adapt_context_free_list_inc;
+
+    /* Bcast MCA parameter */
+    int adapt_ibcast_algorithm;
+    size_t adapt_ibcast_segment_size;
+    int adapt_ibcast_max_send_requests;
+    int adapt_ibcast_max_recv_requests;
+    /* Bcast free list */
+    opal_free_list_t *adapt_ibcast_context_free_list;
+
+    /* Reduce MCA parameter */
+    int adapt_ireduce_algorithm;
+    size_t adapt_ireduce_segment_size;
+    int adapt_ireduce_max_send_requests;
+    int adapt_ireduce_max_recv_requests;
+    int adapt_inbuf_free_list_min;
+    int adapt_inbuf_free_list_max;
+    int adapt_inbuf_free_list_inc;
+
+    /* Reduce free list */
+    opal_free_list_t *adapt_ireduce_context_free_list;
+
+} mca_coll_adapt_component_t;
+
+/* Coll adapt module per communicator*/
+struct mca_coll_adapt_module_t {
+    /* Base module */
+    mca_coll_base_module_t super;
+
+    /* Whether this module has been lazily initialized or not yet */
+    bool adapt_enabled;
+};
+OBJ_CLASS_DECLARATION(mca_coll_adapt_module_t);
+
+/* Global component instance */
+OMPI_MODULE_DECLSPEC extern mca_coll_adapt_component_t mca_coll_adapt_component;
+
+/* ADAPT module functions */
+int ompi_coll_adapt_init_query(bool enable_progress_threads, bool enable_mpi_threads);
+mca_coll_base_module_t * ompi_coll_adapt_comm_query(struct ompi_communicator_t *comm, int *priority);
+
+/* ADAPT request free */
+int ompi_coll_adapt_request_free(ompi_request_t **request);
+
+#endif /* MCA_COLL_ADAPT_EXPORT_H */

ompi/mca/coll/adapt/coll_adapt_algorithms.h

@@ -0,0 +1,39 @@
+/*
+ * Copyright (c) 2014-2020 The University of Tennessee and The University
+ *                         of Tennessee Research Foundation.  All rights
+ *                         reserved.
+ * $COPYRIGHT$
+ * 
+ * Additional copyrights may follow
+ * 
+ * $HEADER$
+ */
+
+#include "ompi/mca/coll/coll.h"
+#include "ompi/mca/coll/base/coll_base_topo.h"
+#include "ompi/mca/coll/base/coll_base_functions.h"
+#include <math.h>
+
+typedef int (*ompi_mca_coll_adapt_ibcast_function_t)(IBCAST_ARGS);
+typedef int (*ompi_mca_coll_adapt_ireduce_function_t)(IREDUCE_ARGS);
+
+typedef struct ompi_coll_adapt_algorithm_index_s {
+    int algorithm_index;
+    union {
+        ompi_mca_coll_adapt_ibcast_function_t  ibcast_fn_ptr;
+        ompi_mca_coll_adapt_ireduce_function_t ireduce_fn_ptr;
+    };
+} ompi_coll_adapt_algorithm_index_t;
+
+/* Bcast */
+int ompi_coll_adapt_ibcast_register(void);
+int ompi_coll_adapt_ibcast_fini(void);
+int ompi_coll_adapt_bcast(BCAST_ARGS);
+int ompi_coll_adapt_ibcast(IBCAST_ARGS);
+
+/* Reduce */
+int ompi_coll_adapt_ireduce_register(void);
+int ompi_coll_adapt_ireduce_fini(void);
+int ompi_coll_adapt_reduce(REDUCE_ARGS);
+int ompi_coll_adapt_ireduce(IREDUCE_ARGS);
+

ompi/mca/coll/adapt/coll_adapt_bcast.c

@@ -0,0 +1,26 @@
+/*
+ * Copyright (c) 2014-2020 The University of Tennessee and The University
+ *                         of Tennessee Research Foundation.  All rights
+ *                         reserved.
+ * $COPYRIGHT$
+ * 
+ * Additional copyrights may follow
+ * 
+ * $HEADER$
+ */
+
+#include "coll_adapt.h"
+#include "coll_adapt_algorithms.h"
+
+int ompi_coll_adapt_bcast(void *buff, int count, struct ompi_datatype_t *datatype, int root,
+                         struct ompi_communicator_t *comm, mca_coll_base_module_t * module)
+{
+    ompi_request_t *request = NULL;
+    int err = ompi_coll_adapt_ibcast(buff, count, datatype, root, comm, &request, module);
+    if( MPI_SUCCESS != err ) {
+        if( NULL == request )
+            return err;
+    }
+    ompi_request_wait(&request, MPI_STATUS_IGNORE);
+    return err;
+}