Merge pull request #7716 from bosilca/coll/adapt
ADAPT: Event-driven collective implementation
Этот коммит содержится в:
Jeff Squyres
GitHub
Коммит
560ebc5780
@ -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 */
|
||||
|
||||
@ -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;
|
||||
|
||||
|
||||
49
ompi/mca/coll/adapt/Makefile.am
Обычный файл
49
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
|
||||
96
ompi/mca/coll/adapt/coll_adapt.h
Обычный файл
96
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 */
|
||||
39
ompi/mca/coll/adapt/coll_adapt_algorithms.h
Обычный файл
39
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);
|
||||
|
||||
26
ompi/mca/coll/adapt/coll_adapt_bcast.c
Обычный файл
26
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;
|
||||
}
|
||||
155
ompi/mca/coll/adapt/coll_adapt_component.c
Обычный файл
155
ompi/mca/coll/adapt/coll_adapt_component.c
Обычный файл
@ -0,0 +1,155 @@
|
||||
/*
|
||||
* 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_config.h"
|
||||
|
||||
#include "opal/util/show_help.h"
|
||||
#include "ompi/constants.h"
|
||||
#include "ompi/mca/coll/coll.h"
|
||||
#include "coll_adapt.h"
|
||||
#include "coll_adapt_algorithms.h"
|
||||
|
||||
/*
|
||||
* Public string showing the coll ompi_adapt component version number
|
||||
*/
|
||||
const char *mca_coll_adapt_component_version_string =
|
||||
"Open MPI ADAPT collective MCA component version " OMPI_VERSION;
|
||||
|
||||
/*
|
||||
* Local functions
|
||||
*/
|
||||
static int adapt_open(void);
|
||||
static int adapt_close(void);
|
||||
static int adapt_register(void);
|
||||
|
||||
/*
|
||||
* Instantiate the public struct with all of our public information
|
||||
* and pointers to our public functions in it
|
||||
*/
|
||||
|
||||
mca_coll_adapt_component_t mca_coll_adapt_component = {
|
||||
/* First, fill in the super */
|
||||
{
|
||||
/* First, the mca_component_t struct containing meta
|
||||
information about the component itself */
|
||||
.collm_version = {
|
||||
MCA_COLL_BASE_VERSION_2_0_0,
|
||||
|
||||
/* Component name and version */
|
||||
.mca_component_name = "adapt",
|
||||
MCA_BASE_MAKE_VERSION(component, OMPI_MAJOR_VERSION, OMPI_MINOR_VERSION,
|
||||
OMPI_RELEASE_VERSION),
|
||||
|
||||
/* Component functions */
|
||||
.mca_open_component = adapt_open,
|
||||
.mca_close_component = adapt_close,
|
||||
.mca_register_component_params = adapt_register,
|
||||
},
|
||||
.collm_data = {
|
||||
/* The component is not checkpoint ready */
|
||||
MCA_BASE_METADATA_PARAM_NONE
|
||||
},
|
||||
|
||||
/* Initialization / querying functions */
|
||||
.collm_init_query = ompi_coll_adapt_init_query,
|
||||
.collm_comm_query = ompi_coll_adapt_comm_query,
|
||||
},
|
||||
|
||||
/* adapt-component specific information */
|
||||
|
||||
0, /* (default) priority */
|
||||
|
||||
0, /* (default) output stream */
|
||||
0, /* (default) verbose level */
|
||||
|
||||
/* default values for non-MCA parameters */
|
||||
/* Not specifying values here gives us all 0's */
|
||||
};
|
||||
|
||||
/* Open the component */
|
||||
static int adapt_open(void)
|
||||
{
|
||||
mca_coll_adapt_component_t *cs = &mca_coll_adapt_component;
|
||||
|
||||
if (cs->adapt_verbose > 0) {
|
||||
cs->adapt_output = opal_output_open(NULL);
|
||||
opal_output_set_verbosity(cs->adapt_output, cs->adapt_verbose);
|
||||
}
|
||||
|
||||
return OMPI_SUCCESS;
|
||||
}
|
||||
|
||||
|
||||
/* Shut down the component */
|
||||
static int adapt_close(void)
|
||||
{
|
||||
ompi_coll_adapt_ibcast_fini();
|
||||
ompi_coll_adapt_ireduce_fini();
|
||||
|
||||
return OMPI_SUCCESS;
|
||||
}
|
||||
|
||||
static int adapt_verify_mca_variables(void)
|
||||
{
|
||||
return OMPI_SUCCESS;
|
||||
}
|
||||
|
||||
/*
|
||||
* Register MCA params
|
||||
*/
|
||||
static int adapt_register(void)
|
||||
{
|
||||
mca_base_component_t *c = &mca_coll_adapt_component.super.collm_version;
|
||||
mca_coll_adapt_component_t *cs = &mca_coll_adapt_component;
|
||||
|
||||
/* If we want to be selected (i.e., all procs on one node), then
|
||||
we should have a high priority */
|
||||
cs->adapt_priority = 0;
|
||||
(void) mca_base_component_var_register(c, "priority", "Priority of the adapt coll component",
|
||||
MCA_BASE_VAR_TYPE_INT, NULL, 0, 0,
|
||||
OPAL_INFO_LVL_9,
|
||||
MCA_BASE_VAR_SCOPE_READONLY, &cs->adapt_priority);
|
||||
|
||||
cs->adapt_verbose = ompi_coll_base_framework.framework_verbose;
|
||||
(void) mca_base_component_var_register(c, "verbose",
|
||||
"Verbose level (default set to the collective framework verbosity)",
|
||||
MCA_BASE_VAR_TYPE_INT, NULL, 0, 0,
|
||||
OPAL_INFO_LVL_9,
|
||||
MCA_BASE_VAR_SCOPE_READONLY, &cs->adapt_verbose);
|
||||
|
||||
cs->adapt_context_free_list_min = 64;
|
||||
(void) mca_base_component_var_register(c, "context_free_list_min",
|
||||
"Minimum number of segments in context free list",
|
||||
MCA_BASE_VAR_TYPE_INT, NULL, 0, 0,
|
||||
OPAL_INFO_LVL_9,
|
||||
MCA_BASE_VAR_SCOPE_READONLY,
|
||||
&cs->adapt_context_free_list_min);
|
||||
|
||||
cs->adapt_context_free_list_max = 1024;
|
||||
(void) mca_base_component_var_register(c, "context_free_list_max",
|
||||
"Maximum number of segments in context free list",
|
||||
MCA_BASE_VAR_TYPE_INT, NULL, 0, 0,
|
||||
OPAL_INFO_LVL_9,
|
||||
MCA_BASE_VAR_SCOPE_READONLY,
|
||||
&cs->adapt_context_free_list_max);
|
||||
|
||||
cs->adapt_context_free_list_inc = 32;
|
||||
(void) mca_base_component_var_register(c, "context_free_list_inc",
|
||||
"Increasement number of segments in context free list",
|
||||
MCA_BASE_VAR_TYPE_INT, NULL, 0, 0,
|
||||
OPAL_INFO_LVL_9,
|
||||
MCA_BASE_VAR_SCOPE_READONLY,
|
||||
&cs->adapt_context_free_list_inc);
|
||||
ompi_coll_adapt_ibcast_register();
|
||||
ompi_coll_adapt_ireduce_register();
|
||||
|
||||
return adapt_verify_mca_variables();
|
||||
}
|
||||
26
ompi/mca/coll/adapt/coll_adapt_context.c
Обычный файл
26
ompi/mca/coll/adapt/coll_adapt_context.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 "ompi/mca/coll/coll.h"
|
||||
#include "coll_adapt_context.h"
|
||||
|
||||
|
||||
OBJ_CLASS_INSTANCE(ompi_coll_adapt_bcast_context_t, opal_free_list_item_t,
|
||||
NULL, NULL);
|
||||
|
||||
OBJ_CLASS_INSTANCE(ompi_coll_adapt_constant_bcast_context_t, opal_object_t,
|
||||
NULL, NULL);
|
||||
|
||||
OBJ_CLASS_INSTANCE(ompi_coll_adapt_reduce_context_t, opal_free_list_item_t,
|
||||
NULL, NULL);
|
||||
|
||||
OBJ_CLASS_INSTANCE(ompi_coll_adapt_constant_reduce_context_t, opal_object_t,
|
||||
NULL, NULL);
|
||||
134
ompi/mca/coll/adapt/coll_adapt_context.h
Обычный файл
134
ompi/mca/coll/adapt/coll_adapt_context.h
Обычный файл
@ -0,0 +1,134 @@
|
||||
/*
|
||||
* 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 "opal/class/opal_free_list.h"
|
||||
#include "opal/class/opal_list.h"
|
||||
#include "ompi/datatype/ompi_datatype.h"
|
||||
#include "ompi/communicator/communicator.h"
|
||||
#include "ompi/op/op.h"
|
||||
#include "ompi/mca/coll/base/coll_base_topo.h"
|
||||
#include "coll_adapt_inbuf.h"
|
||||
|
||||
/* Bcast constant context in bcast context */
|
||||
struct ompi_coll_adapt_constant_bcast_context_s {
|
||||
opal_object_t super;
|
||||
int root;
|
||||
size_t count;
|
||||
size_t seg_count;
|
||||
ompi_datatype_t *datatype;
|
||||
ompi_communicator_t *comm;
|
||||
int real_seg_size;
|
||||
int num_segs;
|
||||
ompi_request_t *request;
|
||||
opal_mutex_t *mutex;
|
||||
int *recv_array;
|
||||
int *send_array;
|
||||
/* Length of the fragment array, which is the number of recevied segments */
|
||||
int num_recv_segs;
|
||||
/* Number of segments that is finishing recving */
|
||||
int num_recv_fini;
|
||||
/* Store the number of sent segments */
|
||||
int num_sent_segs;
|
||||
ompi_coll_tree_t *tree;
|
||||
int ibcast_tag;
|
||||
};
|
||||
|
||||
typedef struct ompi_coll_adapt_constant_bcast_context_s ompi_coll_adapt_constant_bcast_context_t;
|
||||
|
||||
OBJ_CLASS_DECLARATION(ompi_coll_adapt_constant_bcast_context_t);
|
||||
|
||||
|
||||
/* Bcast context of each segment*/
|
||||
typedef struct ompi_coll_adapt_bcast_context_s ompi_coll_adapt_bcast_context_t;
|
||||
|
||||
typedef int (*ompi_coll_adapt_bcast_cuda_callback_fn_t) (ompi_coll_adapt_bcast_context_t * context);
|
||||
|
||||
struct ompi_coll_adapt_bcast_context_s {
|
||||
opal_free_list_item_t super;
|
||||
char *buff;
|
||||
int frag_id;
|
||||
int child_id;
|
||||
int peer;
|
||||
ompi_coll_adapt_constant_bcast_context_t *con;
|
||||
};
|
||||
|
||||
OBJ_CLASS_DECLARATION(ompi_coll_adapt_bcast_context_t);
|
||||
|
||||
/* Reduce constant context in reduce context */
|
||||
struct ompi_coll_adapt_constant_reduce_context_s {
|
||||
opal_object_t super;
|
||||
size_t count;
|
||||
size_t seg_count;
|
||||
ompi_datatype_t *datatype;
|
||||
ompi_communicator_t *comm;
|
||||
size_t real_seg_size;
|
||||
/* Increment of each segment */
|
||||
int segment_increment;
|
||||
int num_segs;
|
||||
ompi_request_t *request;
|
||||
int rank;
|
||||
/* Length of the fragment array, which is the number of recevied segments */
|
||||
int32_t num_recv_segs;
|
||||
/* Number of sent segments */
|
||||
int32_t num_sent_segs;
|
||||
/* Next seg need to be received for every children */
|
||||
opal_atomic_int32_t *next_recv_segs;
|
||||
/* Mutex to protect recv_list */
|
||||
opal_mutex_t *mutex_recv_list;
|
||||
/* Mutex to protect num_recv_segs */
|
||||
opal_mutex_t *mutex_num_recv_segs;
|
||||
/* Mutex to protect num_sent */
|
||||
opal_mutex_t *mutex_num_sent;
|
||||
/* Mutex to protect each segment when do the reduce op */
|
||||
opal_mutex_t *mutex_op_list;
|
||||
/* Reduce operation */
|
||||
ompi_op_t *op;
|
||||
ompi_coll_tree_t *tree;
|
||||
/* Accumulate buff */
|
||||
char **accumbuf;
|
||||
/* inbuf list address of accumbuf */
|
||||
ompi_coll_adapt_inbuf_t ** accumbuf_to_inbuf;
|
||||
opal_free_list_t *inbuf_list;
|
||||
/* A list to store the segments which are received and not yet be sent */
|
||||
opal_list_t *recv_list;
|
||||
ptrdiff_t lower_bound;
|
||||
/* How many sends are posted but not finished */
|
||||
opal_atomic_int32_t ongoing_send;
|
||||
char *sbuf;
|
||||
char *rbuf;
|
||||
int root;
|
||||
/* The distance between the address of inbuf->buff and the address of inbuf */
|
||||
int distance;
|
||||
int ireduce_tag;
|
||||
};
|
||||
|
||||
typedef struct ompi_coll_adapt_constant_reduce_context_s ompi_coll_adapt_constant_reduce_context_t;
|
||||
|
||||
OBJ_CLASS_DECLARATION(ompi_coll_adapt_constant_reduce_context_t);
|
||||
|
||||
/* Reduce context of each segment */
|
||||
typedef struct ompi_coll_adapt_reduce_context_s ompi_coll_adapt_reduce_context_t;
|
||||
|
||||
typedef int (*ompi_coll_adapt_reduce_cuda_callback_fn_t) (ompi_coll_adapt_reduce_context_t * context);
|
||||
|
||||
struct ompi_coll_adapt_reduce_context_s {
|
||||
opal_free_list_item_t super;
|
||||
char *buff;
|
||||
int frag_id;
|
||||
int child_id;
|
||||
int peer;
|
||||
ompi_coll_adapt_constant_reduce_context_t *con;
|
||||
/* store the incoming segment */
|
||||
ompi_coll_adapt_inbuf_t *inbuf;
|
||||
};
|
||||
|
||||
OBJ_CLASS_DECLARATION(ompi_coll_adapt_reduce_context_t);
|
||||
674
ompi/mca/coll/adapt/coll_adapt_ibcast.c
Обычный файл
674
ompi/mca/coll/adapt/coll_adapt_ibcast.c
Обычный файл
@ -0,0 +1,674 @@
|
||||
/*
|
||||
* 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_config.h"
|
||||
#include "ompi/mca/pml/pml.h"
|
||||
#include "coll_adapt.h"
|
||||
#include "coll_adapt_algorithms.h"
|
||||
#include "coll_adapt_context.h"
|
||||
#include "ompi/mca/coll/base/coll_base_util.h"
|
||||
#include "ompi/mca/coll/base/coll_base_functions.h"
|
||||
#include "opal/util/bit_ops.h"
|
||||
#include "opal/sys/atomic.h"
|
||||
#include "ompi/mca/pml/ob1/pml_ob1.h"
|
||||
|
||||
static int ompi_coll_adapt_ibcast_generic(IBCAST_ARGS,
|
||||
ompi_coll_tree_t * tree, size_t seg_size);
|
||||
static int ompi_coll_adapt_ibcast_binomial(IBCAST_ARGS);
|
||||
static int ompi_coll_adapt_ibcast_in_order_binomial(IBCAST_ARGS);
|
||||
static int ompi_coll_adapt_ibcast_binary(IBCAST_ARGS);
|
||||
static int ompi_coll_adapt_ibcast_pipeline(IBCAST_ARGS);
|
||||
static int ompi_coll_adapt_ibcast_chain(IBCAST_ARGS);
|
||||
static int ompi_coll_adapt_ibcast_linear(IBCAST_ARGS);
|
||||
static int ompi_coll_adapt_ibcast_tuned(IBCAST_ARGS);
|
||||
|
||||
typedef int (*ompi_coll_adapt_ibcast_fn_t) (void *buff,
|
||||
int count,
|
||||
struct ompi_datatype_t * datatype,
|
||||
int root,
|
||||
struct ompi_communicator_t * comm,
|
||||
ompi_request_t ** request,
|
||||
mca_coll_base_module_t * module);
|
||||
|
||||
static ompi_coll_adapt_algorithm_index_t ompi_coll_adapt_ibcast_algorithm_index[] = {
|
||||
{0, {ompi_coll_adapt_ibcast_tuned}},
|
||||
{1, {ompi_coll_adapt_ibcast_binomial}},
|
||||
{2, {ompi_coll_adapt_ibcast_in_order_binomial}},
|
||||
{3, {ompi_coll_adapt_ibcast_binary}},
|
||||
{4, {ompi_coll_adapt_ibcast_pipeline}},
|
||||
{5, {ompi_coll_adapt_ibcast_chain}},
|
||||
{6, {ompi_coll_adapt_ibcast_linear}},
|
||||
};
|
||||
|
||||
/*
|
||||
* Set up MCA parameters of MPI_Bcast and MPI_IBcast
|
||||
*/
|
||||
int ompi_coll_adapt_ibcast_register(void)
|
||||
{
|
||||
mca_base_component_t *c = &mca_coll_adapt_component.super.collm_version;
|
||||
|
||||
mca_coll_adapt_component.adapt_ibcast_algorithm = 1;
|
||||
mca_base_component_var_register(c, "bcast_algorithm",
|
||||
"Algorithm of broadcast, 0: tuned, 1: binomial, 2: in_order_binomial, 3: binary, 4: pipeline, 5: chain, 6: linear", MCA_BASE_VAR_TYPE_INT, NULL, 0, 0,
|
||||
OPAL_INFO_LVL_5, MCA_BASE_VAR_SCOPE_READONLY,
|
||||
&mca_coll_adapt_component.adapt_ibcast_algorithm);
|
||||
if( (mca_coll_adapt_component.adapt_ibcast_algorithm < 0) ||
|
||||
(mca_coll_adapt_component.adapt_ibcast_algorithm > (int32_t)(sizeof(ompi_coll_adapt_ibcast_algorithm_index) / sizeof(ompi_coll_adapt_algorithm_index_t))) ) {
|
||||
mca_coll_adapt_component.adapt_ibcast_algorithm = 1;
|
||||
}
|
||||
|
||||
mca_coll_adapt_component.adapt_ibcast_segment_size = 0;
|
||||
mca_base_component_var_register(c, "bcast_segment_size",
|
||||
"Segment size in bytes used by default for bcast algorithms. Only has meaning if algorithm is forced and supports segmenting. 0 bytes means no segmentation.",
|
||||
MCA_BASE_VAR_TYPE_INT, NULL, 0, 0,
|
||||
OPAL_INFO_LVL_5,
|
||||
MCA_BASE_VAR_SCOPE_READONLY,
|
||||
&mca_coll_adapt_component.adapt_ibcast_segment_size);
|
||||
|
||||
mca_coll_adapt_component.adapt_ibcast_max_send_requests = 2;
|
||||
mca_base_component_var_register(c, "bcast_max_send_requests",
|
||||
"Maximum number of send requests",
|
||||
MCA_BASE_VAR_TYPE_INT, NULL, 0, 0,
|
||||
OPAL_INFO_LVL_5,
|
||||
MCA_BASE_VAR_SCOPE_READONLY,
|
||||
&mca_coll_adapt_component.adapt_ibcast_max_send_requests);
|
||||
|
||||
mca_coll_adapt_component.adapt_ibcast_max_recv_requests = 3;
|
||||
mca_base_component_var_register(c, "bcast_max_recv_requests",
|
||||
"Maximum number of receive requests",
|
||||
MCA_BASE_VAR_TYPE_INT, NULL, 0, 0,
|
||||
OPAL_INFO_LVL_5,
|
||||
MCA_BASE_VAR_SCOPE_READONLY,
|
||||
&mca_coll_adapt_component.adapt_ibcast_max_recv_requests);
|
||||
|
||||
mca_coll_adapt_component.adapt_ibcast_context_free_list = NULL;
|
||||
return OMPI_SUCCESS;
|
||||
}
|
||||
|
||||
/*
|
||||
* Release the free list created in ompi_coll_adapt_ibcast_generic
|
||||
*/
|
||||
int ompi_coll_adapt_ibcast_fini(void)
|
||||
{
|
||||
if (NULL != mca_coll_adapt_component.adapt_ibcast_context_free_list) {
|
||||
OBJ_RELEASE(mca_coll_adapt_component.adapt_ibcast_context_free_list);
|
||||
mca_coll_adapt_component.adapt_ibcast_context_free_list = NULL;
|
||||
OPAL_OUTPUT_VERBOSE((10, mca_coll_adapt_component.adapt_output, "ibcast fini\n"));
|
||||
}
|
||||
return OMPI_SUCCESS;
|
||||
}
|
||||
|
||||
/*
|
||||
* Finish a ibcast request
|
||||
*/
|
||||
static int ibcast_request_fini(ompi_coll_adapt_bcast_context_t * context)
|
||||
{
|
||||
ompi_request_t *temp_req = context->con->request;
|
||||
if (context->con->tree->tree_nextsize != 0) {
|
||||
free(context->con->send_array);
|
||||
}
|
||||
if (context->con->num_segs != 0) {
|
||||
free(context->con->recv_array);
|
||||
}
|
||||
OBJ_RELEASE(context->con->mutex);
|
||||
OBJ_RELEASE(context->con);
|
||||
opal_free_list_return(mca_coll_adapt_component.adapt_ibcast_context_free_list,
|
||||
(opal_free_list_item_t *) context);
|
||||
ompi_request_complete(temp_req, 1);
|
||||
|
||||
return OMPI_SUCCESS;
|
||||
}
|
||||
|
||||
/*
|
||||
* Callback function of isend
|
||||
*/
|
||||
static int send_cb(ompi_request_t * req)
|
||||
{
|
||||
ompi_coll_adapt_bcast_context_t *context =
|
||||
(ompi_coll_adapt_bcast_context_t *) req->req_complete_cb_data;
|
||||
int err;
|
||||
|
||||
OPAL_OUTPUT_VERBOSE((30, mca_coll_adapt_component.adapt_output,
|
||||
"[%d]: Send(cb): segment %d to %d at buff %p root %d\n",
|
||||
ompi_comm_rank(context->con->comm), context->frag_id,
|
||||
context->peer, (void *) context->buff, context->con->root));
|
||||
|
||||
OPAL_THREAD_LOCK(context->con->mutex);
|
||||
int sent_id = context->con->send_array[context->child_id];
|
||||
/* If the current process has fragments in recv_array can be sent */
|
||||
if (sent_id < context->con->num_recv_segs) {
|
||||
ompi_coll_adapt_bcast_context_t *send_context;
|
||||
int new_id = context->con->recv_array[sent_id];
|
||||
ompi_request_t *send_req;
|
||||
|
||||
send_context = (ompi_coll_adapt_bcast_context_t *) opal_free_list_wait(mca_coll_adapt_component.adapt_ibcast_context_free_list);
|
||||
send_context->buff =
|
||||
context->buff + (new_id - context->frag_id) * context->con->real_seg_size;
|
||||
send_context->frag_id = new_id;
|
||||
send_context->child_id = context->child_id;
|
||||
send_context->peer = context->peer;
|
||||
send_context->con = context->con;
|
||||
int send_count = send_context->con->seg_count;
|
||||
if (new_id == (send_context->con->num_segs - 1)) {
|
||||
send_count = send_context->con->count - new_id * send_context->con->seg_count;
|
||||
}
|
||||
++(send_context->con->send_array[send_context->child_id]);
|
||||
char *send_buff = send_context->buff;
|
||||
OPAL_OUTPUT_VERBOSE((30, mca_coll_adapt_component.adapt_output,
|
||||
"[%d]: Send(start in send cb): segment %d to %d at buff %p send_count %d tag %d\n",
|
||||
ompi_comm_rank(send_context->con->comm), send_context->frag_id,
|
||||
send_context->peer, (void *) send_context->buff, send_count,
|
||||
send_context->con->ibcast_tag - new_id));
|
||||
err = MCA_PML_CALL(isend
|
||||
(send_buff, send_count, send_context->con->datatype, send_context->peer,
|
||||
send_context->con->ibcast_tag - new_id,
|
||||
MCA_PML_BASE_SEND_SYNCHRONOUS, send_context->con->comm, &send_req));
|
||||
if (MPI_SUCCESS != err) {
|
||||
opal_free_list_return(mca_coll_adapt_component.adapt_ibcast_context_free_list,
|
||||
(opal_free_list_item_t *)send_context);
|
||||
OPAL_THREAD_UNLOCK(context->con->mutex);
|
||||
OBJ_RELEASE(context->con);
|
||||
return err;
|
||||
}
|
||||
/* Set send callback */
|
||||
OPAL_THREAD_UNLOCK(context->con->mutex);
|
||||
ompi_request_set_callback(send_req, send_cb, send_context);
|
||||
OPAL_THREAD_LOCK(context->con->mutex);
|
||||
} else {
|
||||
/* No future send here, we can release the ref */
|
||||
OBJ_RELEASE(context->con);
|
||||
}
|
||||
|
||||
int num_sent = ++(context->con->num_sent_segs);
|
||||
int num_recv_fini = context->con->num_recv_fini;
|
||||
int rank = ompi_comm_rank(context->con->comm);
|
||||
/* Check whether signal the condition */
|
||||
if ((rank == context->con->root
|
||||
&& num_sent == context->con->tree->tree_nextsize * context->con->num_segs)
|
||||
|| (context->con->tree->tree_nextsize > 0 && rank != context->con->root
|
||||
&& num_sent == context->con->tree->tree_nextsize * context->con->num_segs
|
||||
&& num_recv_fini == context->con->num_segs)
|
||||
|| (context->con->tree->tree_nextsize == 0
|
||||
&& num_recv_fini == context->con->num_segs)) {
|
||||
OPAL_OUTPUT_VERBOSE((30, mca_coll_adapt_component.adapt_output, "[%d]: Singal in send\n",
|
||||
ompi_comm_rank(context->con->comm)));
|
||||
OPAL_THREAD_UNLOCK(context->con->mutex);
|
||||
ibcast_request_fini(context);
|
||||
} else {
|
||||
opal_free_list_return(mca_coll_adapt_component.adapt_ibcast_context_free_list,
|
||||
(opal_free_list_item_t *) context);
|
||||
OPAL_THREAD_UNLOCK(context->con->mutex);
|
||||
}
|
||||
req->req_free(&req);
|
||||
/* Call back function return 1, which means successful */
|
||||
return 1;
|
||||
}
|
||||
|
||||
/*
|
||||
* Callback function of irecv
|
||||
*/
|
||||
static int recv_cb(ompi_request_t * req)
|
||||
{
|
||||
/* Get necessary info from request */
|
||||
ompi_coll_adapt_bcast_context_t *context =
|
||||
(ompi_coll_adapt_bcast_context_t *) req->req_complete_cb_data;
|
||||
|
||||
int err, i;
|
||||
OPAL_OUTPUT_VERBOSE((30, mca_coll_adapt_component.adapt_output,
|
||||
"[%d]: Recv(cb): segment %d from %d at buff %p root %d\n",
|
||||
ompi_comm_rank(context->con->comm), context->frag_id,
|
||||
context->peer, (void *) context->buff, context->con->root));
|
||||
|
||||
/* Store the frag_id to seg array */
|
||||
OPAL_THREAD_LOCK(context->con->mutex);
|
||||
int num_recv_segs = ++(context->con->num_recv_segs);
|
||||
context->con->recv_array[num_recv_segs - 1] = context->frag_id;
|
||||
|
||||
int new_id = num_recv_segs + mca_coll_adapt_component.adapt_ibcast_max_recv_requests - 1;
|
||||
/* Receive new segment */
|
||||
if (new_id < context->con->num_segs) {
|
||||
ompi_request_t *recv_req;
|
||||
ompi_coll_adapt_bcast_context_t *recv_context;
|
||||
/* Get new context item from free list */
|
||||
recv_context = (ompi_coll_adapt_bcast_context_t *) opal_free_list_wait(mca_coll_adapt_component.adapt_ibcast_context_free_list);
|
||||
recv_context->buff =
|
||||
context->buff + (new_id - context->frag_id) * context->con->real_seg_size;
|
||||
recv_context->frag_id = new_id;
|
||||
recv_context->child_id = context->child_id;
|
||||
recv_context->peer = context->peer;
|
||||
recv_context->con = context->con;
|
||||
OBJ_RETAIN(context->con);
|
||||
int recv_count = recv_context->con->seg_count;
|
||||
if (new_id == (recv_context->con->num_segs - 1)) {
|
||||
recv_count = recv_context->con->count - new_id * recv_context->con->seg_count;
|
||||
}
|
||||
char *recv_buff = recv_context->buff;
|
||||
OPAL_OUTPUT_VERBOSE((30, mca_coll_adapt_component.adapt_output,
|
||||
"[%d]: Recv(start in recv cb): segment %d from %d at buff %p recv_count %d tag %d\n",
|
||||
ompi_comm_rank(context->con->comm), context->frag_id, context->peer,
|
||||
(void *) recv_buff, recv_count,
|
||||
recv_context->con->ibcast_tag - recv_context->frag_id));
|
||||
MCA_PML_CALL(irecv
|
||||
(recv_buff, recv_count, recv_context->con->datatype, recv_context->peer,
|
||||
recv_context->con->ibcast_tag - recv_context->frag_id,
|
||||
recv_context->con->comm, &recv_req));
|
||||
|
||||
/* Set the receive callback */
|
||||
OPAL_THREAD_UNLOCK(context->con->mutex);
|
||||
ompi_request_set_callback(recv_req, recv_cb, recv_context);
|
||||
OPAL_THREAD_LOCK(context->con->mutex);
|
||||
}
|
||||
|
||||
/* Propagate segment to all children */
|
||||
for (i = 0; i < context->con->tree->tree_nextsize; i++) {
|
||||
/* If the current process can send the segment now, which means the only segment need to be sent is the just arrived one */
|
||||
if (num_recv_segs - 1 == context->con->send_array[i]) {
|
||||
ompi_request_t *send_req;
|
||||
int send_count = context->con->seg_count;
|
||||
if (context->frag_id == (context->con->num_segs - 1)) {
|
||||
send_count = context->con->count - context->frag_id * context->con->seg_count;
|
||||
}
|
||||
|
||||
ompi_coll_adapt_bcast_context_t *send_context;
|
||||
send_context = (ompi_coll_adapt_bcast_context_t *) opal_free_list_wait(mca_coll_adapt_component.adapt_ibcast_context_free_list);
|
||||
send_context->buff = context->buff;
|
||||
send_context->frag_id = context->frag_id;
|
||||
send_context->child_id = i;
|
||||
send_context->peer = context->con->tree->tree_next[i];
|
||||
send_context->con = context->con;
|
||||
OBJ_RETAIN(context->con);
|
||||
++(send_context->con->send_array[i]);
|
||||
char *send_buff = send_context->buff;
|
||||
OPAL_OUTPUT_VERBOSE((30, mca_coll_adapt_component.adapt_output,
|
||||
"[%d]: Send(start in recv cb): segment %d to %d at buff %p send_count %d tag %d\n",
|
||||
ompi_comm_rank(send_context->con->comm), send_context->frag_id,
|
||||
send_context->peer, (void *) send_context->buff, send_count,
|
||||
send_context->con->ibcast_tag - send_context->frag_id));
|
||||
err =
|
||||
MCA_PML_CALL(isend
|
||||
(send_buff, send_count, send_context->con->datatype,
|
||||
send_context->peer,
|
||||
send_context->con->ibcast_tag - send_context->frag_id,
|
||||
MCA_PML_BASE_SEND_SYNCHRONOUS, send_context->con->comm, &send_req));
|
||||
if (MPI_SUCCESS != err) {
|
||||
opal_free_list_return(mca_coll_adapt_component.adapt_ibcast_context_free_list,
|
||||
(opal_free_list_item_t *)send_context);
|
||||
OPAL_THREAD_UNLOCK(context->con->mutex);
|
||||
OBJ_RELEASE(context->con);
|
||||
return err;
|
||||
}
|
||||
/* Set send callback */
|
||||
OPAL_THREAD_UNLOCK(context->con->mutex);
|
||||
ompi_request_set_callback(send_req, send_cb, send_context);
|
||||
OPAL_THREAD_LOCK(context->con->mutex);
|
||||
}
|
||||
}
|
||||
|
||||
OBJ_RELEASE(context->con);
|
||||
|
||||
int num_sent = context->con->num_sent_segs;
|
||||
int num_recv_fini = ++(context->con->num_recv_fini);
|
||||
int rank = ompi_comm_rank(context->con->comm);
|
||||
|
||||
/* If this process is leaf and has received all the segments */
|
||||
if ((rank == context->con->root
|
||||
&& num_sent == context->con->tree->tree_nextsize * context->con->num_segs)
|
||||
|| (context->con->tree->tree_nextsize > 0 && rank != context->con->root
|
||||
&& num_sent == context->con->tree->tree_nextsize * context->con->num_segs
|
||||
&& num_recv_fini == context->con->num_segs)
|
||||
|| (context->con->tree->tree_nextsize == 0
|
||||
&& num_recv_fini == context->con->num_segs)) {
|
||||
OPAL_OUTPUT_VERBOSE((30, mca_coll_adapt_component.adapt_output, "[%d]: Singal in recv\n",
|
||||
ompi_comm_rank(context->con->comm)));
|
||||
OPAL_THREAD_UNLOCK(context->con->mutex);
|
||||
ibcast_request_fini(context);
|
||||
} else {
|
||||
opal_free_list_return(mca_coll_adapt_component.adapt_ibcast_context_free_list,
|
||||
(opal_free_list_item_t *) context);
|
||||
OPAL_THREAD_UNLOCK(context->con->mutex);
|
||||
}
|
||||
req->req_free(&req);
|
||||
return 1;
|
||||
}
|
||||
|
||||
int ompi_coll_adapt_ibcast(void *buff, int count, struct ompi_datatype_t *datatype, int root,
|
||||
struct ompi_communicator_t *comm, ompi_request_t ** request,
|
||||
mca_coll_base_module_t * module)
|
||||
{
|
||||
OPAL_OUTPUT_VERBOSE((10, mca_coll_adapt_component.adapt_output,
|
||||
"ibcast root %d, algorithm %d, coll_adapt_ibcast_segment_size %zu, coll_adapt_ibcast_max_send_requests %d, coll_adapt_ibcast_max_recv_requests %d\n",
|
||||
root, mca_coll_adapt_component.adapt_ibcast_algorithm,
|
||||
mca_coll_adapt_component.adapt_ibcast_segment_size,
|
||||
mca_coll_adapt_component.adapt_ibcast_max_send_requests,
|
||||
mca_coll_adapt_component.adapt_ibcast_max_recv_requests));
|
||||
|
||||
ompi_coll_adapt_ibcast_fn_t bcast_func =
|
||||
ompi_coll_adapt_ibcast_algorithm_index[mca_coll_adapt_component.adapt_ibcast_algorithm].ibcast_fn_ptr;
|
||||
return bcast_func(buff, count, datatype, root, comm, request, module);
|
||||
}
|
||||
|
||||
/*
|
||||
* Ibcast functions with different algorithms
|
||||
*/
|
||||
static int
|
||||
ompi_coll_adapt_ibcast_tuned(void *buff, int count, struct ompi_datatype_t *datatype,
|
||||
int root, struct ompi_communicator_t *comm,
|
||||
ompi_request_t ** request,
|
||||
mca_coll_base_module_t *module)
|
||||
{
|
||||
OPAL_OUTPUT_VERBOSE((10, mca_coll_adapt_component.adapt_output, "tuned not implemented\n"));
|
||||
return OMPI_ERR_NOT_IMPLEMENTED;
|
||||
}
|
||||
|
||||
static int
|
||||
ompi_coll_adapt_ibcast_binomial(void *buff, int count, struct ompi_datatype_t *datatype,
|
||||
int root, struct ompi_communicator_t *comm,
|
||||
ompi_request_t ** request, mca_coll_base_module_t * module)
|
||||
{
|
||||
ompi_coll_tree_t *tree = ompi_coll_base_topo_build_bmtree(comm, root);
|
||||
return ompi_coll_adapt_ibcast_generic(buff, count, datatype, root, comm, request, module, tree,
|
||||
mca_coll_adapt_component.adapt_ibcast_segment_size);
|
||||
}
|
||||
|
||||
static int
|
||||
ompi_coll_adapt_ibcast_in_order_binomial(void *buff, int count, struct ompi_datatype_t *datatype,
|
||||
int root, struct ompi_communicator_t *comm,
|
||||
ompi_request_t ** request,
|
||||
mca_coll_base_module_t * module)
|
||||
{
|
||||
ompi_coll_tree_t *tree = ompi_coll_base_topo_build_in_order_bmtree(comm, root);
|
||||
return ompi_coll_adapt_ibcast_generic(buff, count, datatype, root, comm, request, module, tree,
|
||||
mca_coll_adapt_component.adapt_ibcast_segment_size);
|
||||
}
|
||||
|
||||
|
||||
static int
|
||||
ompi_coll_adapt_ibcast_binary(void *buff, int count, struct ompi_datatype_t *datatype, int root,
|
||||
struct ompi_communicator_t *comm, ompi_request_t ** request,
|
||||
mca_coll_base_module_t * module)
|
||||
{
|
||||
ompi_coll_tree_t *tree = ompi_coll_base_topo_build_tree(2, comm, root);
|
||||
return ompi_coll_adapt_ibcast_generic(buff, count, datatype, root, comm, request, module, tree,
|
||||
mca_coll_adapt_component.adapt_ibcast_segment_size);
|
||||
}
|
||||
|
||||
static int
|
||||
ompi_coll_adapt_ibcast_pipeline(void *buff, int count, struct ompi_datatype_t *datatype,
|
||||
int root, struct ompi_communicator_t *comm,
|
||||
ompi_request_t ** request, mca_coll_base_module_t * module)
|
||||
{
|
||||
ompi_coll_tree_t *tree = ompi_coll_base_topo_build_chain(1, comm, root);
|
||||
return ompi_coll_adapt_ibcast_generic(buff, count, datatype, root, comm, request, module, tree,
|
||||
mca_coll_adapt_component.adapt_ibcast_segment_size);
|
||||
}
|
||||
|
||||
|
||||
static int
|
||||
ompi_coll_adapt_ibcast_chain(void *buff, int count, struct ompi_datatype_t *datatype, int root,
|
||||
struct ompi_communicator_t *comm, ompi_request_t ** request,
|
||||
mca_coll_base_module_t * module)
|
||||
{
|
||||
ompi_coll_tree_t *tree = ompi_coll_base_topo_build_chain(4, comm, root);
|
||||
return ompi_coll_adapt_ibcast_generic(buff, count, datatype, root, comm, request, module, tree,
|
||||
mca_coll_adapt_component.adapt_ibcast_segment_size);
|
||||
}
|
||||
|
||||
static int
|
||||
ompi_coll_adapt_ibcast_linear(void *buff, int count, struct ompi_datatype_t *datatype, int root,
|
||||
struct ompi_communicator_t *comm, ompi_request_t ** request,
|
||||
mca_coll_base_module_t * module)
|
||||
{
|
||||
int fanout = ompi_comm_size(comm) - 1;
|
||||
ompi_coll_tree_t *tree;
|
||||
if (fanout < 1) {
|
||||
tree = ompi_coll_base_topo_build_chain(1, comm, root);
|
||||
} else if (fanout <= MAXTREEFANOUT) {
|
||||
tree = ompi_coll_base_topo_build_tree(ompi_comm_size(comm) - 1, comm, root);
|
||||
} else {
|
||||
tree = ompi_coll_base_topo_build_tree(MAXTREEFANOUT, comm, root);
|
||||
}
|
||||
return ompi_coll_adapt_ibcast_generic(buff, count, datatype, root, comm, request, module, tree,
|
||||
mca_coll_adapt_component.adapt_ibcast_segment_size);
|
||||
}
|
||||
|
||||
|
||||
int ompi_coll_adapt_ibcast_generic(void *buff, int count, struct ompi_datatype_t *datatype, int root,
|
||||
struct ompi_communicator_t *comm, ompi_request_t ** request,
|
||||
mca_coll_base_module_t * module, ompi_coll_tree_t * tree,
|
||||
size_t seg_size)
|
||||
{
|
||||
int i, j, rank, err;
|
||||
/* The min of num_segs and SEND_NUM or RECV_NUM, in case the num_segs is less than SEND_NUM or RECV_NUM */
|
||||
int min;
|
||||
|
||||
/* Number of datatype in a segment */
|
||||
int seg_count = count;
|
||||
/* Size of a datatype */
|
||||
size_t type_size;
|
||||
/* Real size of a segment */
|
||||
size_t real_seg_size;
|
||||
ptrdiff_t extent, lb;
|
||||
/* Number of segments */
|
||||
int num_segs;
|
||||
|
||||
/* The request passed outside */
|
||||
ompi_request_t *temp_request = NULL;
|
||||
opal_mutex_t *mutex;
|
||||
/* Store the segments which are received */
|
||||
int *recv_array = NULL;
|
||||
/* Record how many isends have been issued for every child */
|
||||
int *send_array = NULL;
|
||||
|
||||
/* Atomically set up free list */
|
||||
if (NULL == mca_coll_adapt_component.adapt_ibcast_context_free_list) {
|
||||
opal_free_list_t* fl = OBJ_NEW(opal_free_list_t);
|
||||
opal_free_list_init(fl,
|
||||
sizeof(ompi_coll_adapt_bcast_context_t),
|
||||
opal_cache_line_size,
|
||||
OBJ_CLASS(ompi_coll_adapt_bcast_context_t),
|
||||
0, opal_cache_line_size,
|
||||
mca_coll_adapt_component.adapt_context_free_list_min,
|
||||
mca_coll_adapt_component.adapt_context_free_list_max,
|
||||
mca_coll_adapt_component.adapt_context_free_list_inc,
|
||||
NULL, 0, NULL, NULL, NULL);
|
||||
if( !OPAL_ATOMIC_COMPARE_EXCHANGE_STRONG_PTR((opal_atomic_intptr_t *)&mca_coll_adapt_component.adapt_ibcast_context_free_list,
|
||||
&(intptr_t){0}, fl) ) {
|
||||
OBJ_RELEASE(fl);
|
||||
}
|
||||
}
|
||||
|
||||
/* Set up request */
|
||||
temp_request = OBJ_NEW(ompi_request_t);
|
||||
OMPI_REQUEST_INIT(temp_request, false);
|
||||
temp_request->req_state = OMPI_REQUEST_ACTIVE;
|
||||
temp_request->req_type = OMPI_REQUEST_COLL;
|
||||
temp_request->req_free = ompi_coll_adapt_request_free;
|
||||
temp_request->req_status.MPI_SOURCE = 0;
|
||||
temp_request->req_status.MPI_TAG = 0;
|
||||
temp_request->req_status.MPI_ERROR = 0;
|
||||
temp_request->req_status._cancelled = 0;
|
||||
temp_request->req_status._ucount = 0;
|
||||
*request = temp_request;
|
||||
|
||||
/* Set up mutex */
|
||||
mutex = OBJ_NEW(opal_mutex_t);
|
||||
|
||||
rank = ompi_comm_rank(comm);
|
||||
|
||||
/* Determine number of elements sent per operation */
|
||||
ompi_datatype_type_size(datatype, &type_size);
|
||||
COLL_BASE_COMPUTED_SEGCOUNT(seg_size, type_size, seg_count);
|
||||
|
||||
ompi_datatype_get_extent(datatype, &lb, &extent);
|
||||
num_segs = (count + seg_count - 1) / seg_count;
|
||||
real_seg_size = (ptrdiff_t) seg_count *extent;
|
||||
|
||||
/* Set memory for recv_array and send_array, created on heap becasue they are needed to be accessed by other functions (callback functions) */
|
||||
if (num_segs != 0) {
|
||||
recv_array = (int *) malloc(sizeof(int) * num_segs);
|
||||
}
|
||||
if (tree->tree_nextsize != 0) {
|
||||
send_array = (int *) malloc(sizeof(int) * tree->tree_nextsize);
|
||||
}
|
||||
|
||||
/* Set constant context for send and recv call back */
|
||||
ompi_coll_adapt_constant_bcast_context_t *con = OBJ_NEW(ompi_coll_adapt_constant_bcast_context_t);
|
||||
con->root = root;
|
||||
con->count = count;
|
||||
con->seg_count = seg_count;
|
||||
con->datatype = datatype;
|
||||
con->comm = comm;
|
||||
con->real_seg_size = real_seg_size;
|
||||
con->num_segs = num_segs;
|
||||
con->recv_array = recv_array;
|
||||
con->num_recv_segs = 0;
|
||||
con->num_recv_fini = 0;
|
||||
con->send_array = send_array;
|
||||
con->num_sent_segs = 0;
|
||||
con->mutex = mutex;
|
||||
con->request = temp_request;
|
||||
con->tree = tree;
|
||||
con->ibcast_tag = ompi_coll_base_nbc_reserve_tags(comm, num_segs);
|
||||
|
||||
OPAL_OUTPUT_VERBOSE((30, mca_coll_adapt_component.adapt_output,
|
||||
"[%d]: Ibcast, root %d, tag %d\n", rank, root,
|
||||
con->ibcast_tag));
|
||||
OPAL_OUTPUT_VERBOSE((30, mca_coll_adapt_component.adapt_output,
|
||||
"[%d]: con->mutex = %p, num_children = %d, num_segs = %d, real_seg_size = %d, seg_count = %d, tree_adreess = %p\n",
|
||||
rank, (void *) con->mutex, tree->tree_nextsize, num_segs,
|
||||
(int) real_seg_size, seg_count, (void *) con->tree));
|
||||
|
||||
OPAL_THREAD_LOCK(mutex);
|
||||
|
||||
/* If the current process is root, it sends segment to every children */
|
||||
if (rank == root) {
|
||||
/* Handle the situation when num_segs < SEND_NUM */
|
||||
if (num_segs <= mca_coll_adapt_component.adapt_ibcast_max_send_requests) {
|
||||
min = num_segs;
|
||||
} else {
|
||||
min = mca_coll_adapt_component.adapt_ibcast_max_send_requests;
|
||||
}
|
||||
|
||||
/* Set recv_array, root has already had all the segments */
|
||||
for (i = 0; i < num_segs; i++) {
|
||||
recv_array[i] = i;
|
||||
}
|
||||
con->num_recv_segs = num_segs;
|
||||
/* Set send_array, will send ompi_coll_adapt_ibcast_max_send_requests segments */
|
||||
for (i = 0; i < tree->tree_nextsize; i++) {
|
||||
send_array[i] = mca_coll_adapt_component.adapt_ibcast_max_send_requests;
|
||||
}
|
||||
|
||||
ompi_request_t *send_req;
|
||||
/* Number of datatypes in each send */
|
||||
int send_count = seg_count;
|
||||
for (i = 0; i < min; i++) {
|
||||
if (i == (num_segs - 1)) {
|
||||
send_count = count - i * seg_count;
|
||||
}
|
||||
for (j = 0; j < tree->tree_nextsize; j++) {
|
||||
ompi_coll_adapt_bcast_context_t *context =
|
||||
(ompi_coll_adapt_bcast_context_t *) opal_free_list_wait(mca_coll_adapt_component.
|
||||
adapt_ibcast_context_free_list);
|
||||
context->buff = (char *) buff + i * real_seg_size;
|
||||
context->frag_id = i;
|
||||
/* The id of peer in in children_list */
|
||||
context->child_id = j;
|
||||
/* Actural rank of the peer */
|
||||
context->peer = tree->tree_next[j];
|
||||
context->con = con;
|
||||
OBJ_RETAIN(con);
|
||||
|
||||
char *send_buff = context->buff;
|
||||
OPAL_OUTPUT_VERBOSE((30, mca_coll_adapt_component.adapt_output,
|
||||
"[%d]: Send(start in main): segment %d to %d at buff %p send_count %d tag %d\n",
|
||||
rank, context->frag_id, context->peer,
|
||||
(void *) send_buff, send_count, con->ibcast_tag - i));
|
||||
err =
|
||||
MCA_PML_CALL(isend
|
||||
(send_buff, send_count, datatype, context->peer,
|
||||
con->ibcast_tag - i, MCA_PML_BASE_SEND_SYNCHRONOUS, comm,
|
||||
&send_req));
|
||||
if (MPI_SUCCESS != err) {
|
||||
return err;
|
||||
}
|
||||
/* Set send callback */
|
||||
OPAL_THREAD_UNLOCK(mutex);
|
||||
ompi_request_set_callback(send_req, send_cb, context);
|
||||
OPAL_THREAD_LOCK(mutex);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
/* If the current process is not root, it receives data from parent in the tree. */
|
||||
else {
|
||||
/* Handle the situation when num_segs < RECV_NUM */
|
||||
if (num_segs <= mca_coll_adapt_component.adapt_ibcast_max_recv_requests) {
|
||||
min = num_segs;
|
||||
} else {
|
||||
min = mca_coll_adapt_component.adapt_ibcast_max_recv_requests;
|
||||
}
|
||||
|
||||
/* Set recv_array, recv_array is empty */
|
||||
for (i = 0; i < num_segs; i++) {
|
||||
recv_array[i] = 0;
|
||||
}
|
||||
/* Set send_array to empty */
|
||||
for (i = 0; i < tree->tree_nextsize; i++) {
|
||||
send_array[i] = 0;
|
||||
}
|
||||
|
||||
/* Create a recv request */
|
||||
ompi_request_t *recv_req;
|
||||
|
||||
/* Recevice some segments from its parent */
|
||||
int recv_count = seg_count;
|
||||
for (i = 0; i < min; i++) {
|
||||
if (i == (num_segs - 1)) {
|
||||
recv_count = count - i * seg_count;
|
||||
}
|
||||
ompi_coll_adapt_bcast_context_t *context =
|
||||
(ompi_coll_adapt_bcast_context_t *) opal_free_list_wait(mca_coll_adapt_component.
|
||||
adapt_ibcast_context_free_list);
|
||||
context->buff = (char *) buff + i * real_seg_size;
|
||||
context->frag_id = i;
|
||||
context->peer = tree->tree_prev;
|
||||
context->con = con;
|
||||
OBJ_RETAIN(con);
|
||||
char *recv_buff = context->buff;
|
||||
OPAL_OUTPUT_VERBOSE((30, mca_coll_adapt_component.adapt_output,
|
||||
"[%d]: Recv(start in main): segment %d from %d at buff %p recv_count %d tag %d\n",
|
||||
ompi_comm_rank(context->con->comm), context->frag_id,
|
||||
context->peer, (void *) recv_buff, recv_count,
|
||||
con->ibcast_tag - i));
|
||||
err =
|
||||
MCA_PML_CALL(irecv
|
||||
(recv_buff, recv_count, datatype, context->peer,
|
||||
con->ibcast_tag - i, comm, &recv_req));
|
||||
if (MPI_SUCCESS != err) {
|
||||
return err;
|
||||
}
|
||||
/* Set receive callback */
|
||||
OPAL_THREAD_UNLOCK(mutex);
|
||||
ompi_request_set_callback(recv_req, recv_cb, context);
|
||||
OPAL_THREAD_LOCK(mutex);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
OPAL_THREAD_UNLOCK(mutex);
|
||||
|
||||
OPAL_OUTPUT_VERBOSE((30, mca_coll_adapt_component.adapt_output,
|
||||
"[%d]: End of Ibcast\n", rank));
|
||||
|
||||
return MPI_SUCCESS;
|
||||
}
|
||||
16
ompi/mca/coll/adapt/coll_adapt_inbuf.c
Обычный файл
16
ompi/mca/coll/adapt/coll_adapt_inbuf.c
Обычный файл
@ -0,0 +1,16 @@
|
||||
/*
|
||||
* 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_inbuf.h"
|
||||
|
||||
OBJ_CLASS_INSTANCE(ompi_coll_adapt_inbuf_t, opal_free_list_item_t,
|
||||
NULL, NULL);
|
||||
26
ompi/mca/coll/adapt/coll_adapt_inbuf.h
Обычный файл
26
ompi/mca/coll/adapt/coll_adapt_inbuf.h
Обычный файл
@ -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$
|
||||
*/
|
||||
|
||||
#ifndef MCA_COLL_ADAPT_INBUF_H
|
||||
#define MCA_COLL_ADAPT_INBUF_H
|
||||
|
||||
#include "opal/class/opal_free_list.h"
|
||||
|
||||
struct ompi_coll_adapt_inbuf_s {
|
||||
opal_free_list_item_t super;
|
||||
char buff[];
|
||||
};
|
||||
|
||||
typedef struct ompi_coll_adapt_inbuf_s ompi_coll_adapt_inbuf_t;
|
||||
|
||||
OBJ_CLASS_DECLARATION(ompi_coll_adapt_inbuf_t);
|
||||
|
||||
#endif /* MCA_COLL_ADAPT_INBUF_H */
|
||||
929
ompi/mca/coll/adapt/coll_adapt_ireduce.c
Обычный файл
929
ompi/mca/coll/adapt/coll_adapt_ireduce.c
Обычный файл
@ -0,0 +1,929 @@
|
||||
/*
|
||||
* 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_config.h"
|
||||
#include "ompi/communicator/communicator.h"
|
||||
#include "coll_adapt.h"
|
||||
#include "coll_adapt_algorithms.h"
|
||||
#include "coll_adapt_context.h"
|
||||
#include "coll_adapt_item.h"
|
||||
#include "ompi/constants.h"
|
||||
#include "ompi/mca/coll/coll.h"
|
||||
#include "ompi/mca/coll/base/coll_base_util.h"
|
||||
#include "ompi/mca/pml/pml.h"
|
||||
#include "ompi/mca/coll/base/coll_base_functions.h"
|
||||
#include "ompi/mca/coll/base/coll_base_topo.h"
|
||||
|
||||
static int ompi_coll_adapt_ireduce_tuned(IREDUCE_ARGS);
|
||||
static int ompi_coll_adapt_ireduce_binomial(IREDUCE_ARGS);
|
||||
static int ompi_coll_adapt_ireduce_in_order_binomial(IREDUCE_ARGS);
|
||||
static int ompi_coll_adapt_ireduce_binary(IREDUCE_ARGS);
|
||||
static int ompi_coll_adapt_ireduce_pipeline(IREDUCE_ARGS);
|
||||
static int ompi_coll_adapt_ireduce_chain(IREDUCE_ARGS);
|
||||
static int ompi_coll_adapt_ireduce_linear(IREDUCE_ARGS);
|
||||
static int ompi_coll_adapt_ireduce_generic(IREDUCE_ARGS,
|
||||
ompi_coll_tree_t * tree, size_t seg_size);
|
||||
|
||||
/* MPI_Reduce and MPI_Ireduce in the ADAPT module only work for commutative operations */
|
||||
|
||||
typedef int (*ompi_coll_adapt_ireduce_fn_t) (const void *sbuf,
|
||||
void *rbuf,
|
||||
int count,
|
||||
struct ompi_datatype_t * datatype,
|
||||
struct ompi_op_t * op,
|
||||
int root,
|
||||
struct ompi_communicator_t * comm,
|
||||
ompi_request_t ** request,
|
||||
mca_coll_base_module_t * module);
|
||||
|
||||
static ompi_coll_adapt_algorithm_index_t ompi_coll_adapt_ireduce_algorithm_index[] = {
|
||||
{0, {.ireduce_fn_ptr = ompi_coll_adapt_ireduce_tuned}},
|
||||
{1, {.ireduce_fn_ptr = ompi_coll_adapt_ireduce_binomial}},
|
||||
{2, {.ireduce_fn_ptr = ompi_coll_adapt_ireduce_in_order_binomial}},
|
||||
{3, {.ireduce_fn_ptr = ompi_coll_adapt_ireduce_binary}},
|
||||
{4, {.ireduce_fn_ptr = ompi_coll_adapt_ireduce_pipeline}},
|
||||
{5, {.ireduce_fn_ptr = ompi_coll_adapt_ireduce_chain}},
|
||||
{6, {.ireduce_fn_ptr = ompi_coll_adapt_ireduce_linear}},
|
||||
};
|
||||
|
||||
/*
|
||||
* Set up MCA parameters of MPI_Reduce and MPI_Ireduce
|
||||
*/
|
||||
int ompi_coll_adapt_ireduce_register(void)
|
||||
{
|
||||
mca_base_component_t *c = &mca_coll_adapt_component.super.collm_version;
|
||||
|
||||
mca_coll_adapt_component.adapt_ireduce_algorithm = 1;
|
||||
mca_base_component_var_register(c, "reduce_algorithm",
|
||||
"Algorithm of reduce, 1: binomial, 2: in_order_binomial, 3: binary, 4: pipeline, 5: chain, 6: linear", MCA_BASE_VAR_TYPE_INT, NULL, 0, 0,
|
||||
OPAL_INFO_LVL_5, MCA_BASE_VAR_SCOPE_READONLY,
|
||||
&mca_coll_adapt_component.adapt_ireduce_algorithm);
|
||||
if( (mca_coll_adapt_component.adapt_ireduce_algorithm < 0) ||
|
||||
(mca_coll_adapt_component.adapt_ireduce_algorithm > (int32_t)(sizeof(ompi_coll_adapt_ireduce_algorithm_index) / sizeof(ompi_coll_adapt_algorithm_index_t))) ) {
|
||||
mca_coll_adapt_component.adapt_ireduce_algorithm = 1;
|
||||
}
|
||||
|
||||
mca_coll_adapt_component.adapt_ireduce_segment_size = 163740;
|
||||
mca_base_component_var_register(c, "reduce_segment_size",
|
||||
"Segment size in bytes used by default for reduce algorithms. Only has meaning if algorithm is forced and supports segmenting. 0 bytes means no segmentation.",
|
||||
MCA_BASE_VAR_TYPE_SIZE_T, NULL, 0, 0,
|
||||
OPAL_INFO_LVL_5,
|
||||
MCA_BASE_VAR_SCOPE_READONLY,
|
||||
&mca_coll_adapt_component.adapt_ireduce_segment_size);
|
||||
|
||||
mca_coll_adapt_component.adapt_ireduce_max_send_requests = 2;
|
||||
mca_base_component_var_register(c, "reduce_max_send_requests",
|
||||
"Maximum number of send requests",
|
||||
MCA_BASE_VAR_TYPE_INT, NULL, 0, 0,
|
||||
OPAL_INFO_LVL_5,
|
||||
MCA_BASE_VAR_SCOPE_READONLY,
|
||||
&mca_coll_adapt_component.adapt_ireduce_max_send_requests);
|
||||
|
||||
mca_coll_adapt_component.adapt_ireduce_max_recv_requests = 3;
|
||||
mca_base_component_var_register(c, "reduce_max_recv_requests",
|
||||
"Maximum number of receive requests",
|
||||
MCA_BASE_VAR_TYPE_INT, NULL, 0, 0,
|
||||
OPAL_INFO_LVL_5,
|
||||
MCA_BASE_VAR_SCOPE_READONLY,
|
||||
&mca_coll_adapt_component.adapt_ireduce_max_recv_requests);
|
||||
|
||||
mca_coll_adapt_component.adapt_inbuf_free_list_min = 10;
|
||||
mca_base_component_var_register(c, "inbuf_free_list_min",
|
||||
"Minimum number of segment in inbuf free list",
|
||||
MCA_BASE_VAR_TYPE_INT, NULL, 0, 0,
|
||||
OPAL_INFO_LVL_5,
|
||||
MCA_BASE_VAR_SCOPE_READONLY,
|
||||
&mca_coll_adapt_component.adapt_inbuf_free_list_min);
|
||||
|
||||
mca_coll_adapt_component.adapt_inbuf_free_list_max = 10000;
|
||||
mca_base_component_var_register(c, "inbuf_free_list_max",
|
||||
"Maximum number of segment in inbuf free list",
|
||||
MCA_BASE_VAR_TYPE_INT, NULL, 0, 0,
|
||||
OPAL_INFO_LVL_5,
|
||||
MCA_BASE_VAR_SCOPE_READONLY,
|
||||
&mca_coll_adapt_component.adapt_inbuf_free_list_max);
|
||||
|
||||
|
||||
mca_coll_adapt_component.adapt_inbuf_free_list_inc = 10;
|
||||
mca_base_component_var_register(c, "inbuf_free_list_inc",
|
||||
"Maximum number of segment in inbuf free list",
|
||||
MCA_BASE_VAR_TYPE_INT, NULL, 0, 0,
|
||||
OPAL_INFO_LVL_5,
|
||||
MCA_BASE_VAR_SCOPE_READONLY,
|
||||
&mca_coll_adapt_component.adapt_inbuf_free_list_inc);
|
||||
|
||||
mca_coll_adapt_component.adapt_ireduce_context_free_list = NULL;
|
||||
return OMPI_SUCCESS;
|
||||
}
|
||||
|
||||
/*
|
||||
* Release the free list created in ompi_coll_adapt_ireduce_generic
|
||||
*/
|
||||
int ompi_coll_adapt_ireduce_fini(void)
|
||||
{
|
||||
if (NULL != mca_coll_adapt_component.adapt_ireduce_context_free_list) {
|
||||
OBJ_RELEASE(mca_coll_adapt_component.adapt_ireduce_context_free_list);
|
||||
mca_coll_adapt_component.adapt_ireduce_context_free_list = NULL;
|
||||
OPAL_OUTPUT_VERBOSE((10, mca_coll_adapt_component.adapt_output, "ireduce fini\n"));
|
||||
}
|
||||
return OMPI_SUCCESS;
|
||||
}
|
||||
|
||||
/*
|
||||
* Functions to access list
|
||||
*/
|
||||
static ompi_coll_adapt_item_t *get_next_ready_item(opal_list_t * list, int num_children)
|
||||
{
|
||||
ompi_coll_adapt_item_t *item;
|
||||
if (opal_list_is_empty(list)) {
|
||||
return NULL;
|
||||
}
|
||||
for (item = (ompi_coll_adapt_item_t *) opal_list_get_first(list);
|
||||
item != (ompi_coll_adapt_item_t *) opal_list_get_end(list);
|
||||
item = (ompi_coll_adapt_item_t *) ((opal_list_item_t *) item)->opal_list_next) {
|
||||
if (item->count == num_children) {
|
||||
opal_list_remove_item(list, (opal_list_item_t *) item);
|
||||
return item;
|
||||
}
|
||||
}
|
||||
return NULL;
|
||||
}
|
||||
|
||||
static int add_to_list(opal_list_t * list, int id)
|
||||
{
|
||||
ompi_coll_adapt_item_t *item;
|
||||
for (item = (ompi_coll_adapt_item_t *) opal_list_get_first(list);
|
||||
item != (ompi_coll_adapt_item_t *) opal_list_get_end(list);
|
||||
item = (ompi_coll_adapt_item_t *) ((opal_list_item_t *) item)->opal_list_next) {
|
||||
if (item->id == id) {
|
||||
(item->count)++;
|
||||
OPAL_OUTPUT_VERBOSE((30, mca_coll_adapt_component.adapt_output, "add_to_list_return 1\n"));
|
||||
return 1;
|
||||
}
|
||||
}
|
||||
/* Add a new object to the list with count set to 1 */
|
||||
item = OBJ_NEW(ompi_coll_adapt_item_t);
|
||||
item->id = id;
|
||||
item->count = 1;
|
||||
opal_list_append(list, (opal_list_item_t *) item);
|
||||
OPAL_OUTPUT_VERBOSE((30, mca_coll_adapt_component.adapt_output, "add_to_list_return 1\n"));
|
||||
return 2;
|
||||
}
|
||||
|
||||
/*
|
||||
* Get the inbuf address
|
||||
*/
|
||||
static ompi_coll_adapt_inbuf_t *to_inbuf(char *buf, int distance)
|
||||
{
|
||||
return (ompi_coll_adapt_inbuf_t *) (buf - distance);
|
||||
}
|
||||
|
||||
/*
|
||||
* Finish a ireduce request
|
||||
*/
|
||||
static int ireduce_request_fini(ompi_coll_adapt_reduce_context_t * context)
|
||||
{
|
||||
/* Return the allocated recourses */
|
||||
int i;
|
||||
ompi_request_t *temp_req = context->con->request;
|
||||
if (context->con->accumbuf != NULL) {
|
||||
if (context->con->rank != context->con->root) {
|
||||
for (i = 0; i < context->con->num_segs; i++) {
|
||||
OPAL_OUTPUT_VERBOSE((30, mca_coll_adapt_component.adapt_output,
|
||||
"[%d]: Return accumbuf %d %p\n",
|
||||
ompi_comm_rank(context->con->comm), i,
|
||||
(void *) to_inbuf(context->con->accumbuf[i],
|
||||
context->con->distance)));
|
||||
opal_free_list_return(context->con->inbuf_list,
|
||||
(opal_free_list_item_t *) to_inbuf(context->con->accumbuf[i],
|
||||
context->con->distance));
|
||||
}
|
||||
}
|
||||
free(context->con->accumbuf);
|
||||
}
|
||||
OBJ_RELEASE(context->con->recv_list);
|
||||
for (i = 0; i < context->con->num_segs; i++) {
|
||||
OBJ_DESTRUCT(&context->con->mutex_op_list[i]);
|
||||
}
|
||||
free(context->con->mutex_op_list);
|
||||
OBJ_RELEASE(context->con->mutex_num_recv_segs);
|
||||
OBJ_RELEASE(context->con->mutex_recv_list);
|
||||
OBJ_RELEASE(context->con->mutex_num_sent);
|
||||
if (context->con->tree->tree_nextsize > 0) {
|
||||
OBJ_RELEASE(context->con->inbuf_list);
|
||||
free(context->con->next_recv_segs);
|
||||
}
|
||||
OBJ_RELEASE(context->con);
|
||||
OBJ_RELEASE(context->con);
|
||||
OPAL_OUTPUT_VERBOSE((30, mca_coll_adapt_component.adapt_output, "return context_list\n"));
|
||||
opal_free_list_return(mca_coll_adapt_component.adapt_ireduce_context_free_list,
|
||||
(opal_free_list_item_t *) context);
|
||||
/* Complete the request */
|
||||
ompi_request_complete(temp_req, 1);
|
||||
return OMPI_SUCCESS;
|
||||
}
|
||||
|
||||
/*
|
||||
* Callback function of isend
|
||||
*/
|
||||
static int send_cb(ompi_request_t * req)
|
||||
{
|
||||
ompi_coll_adapt_reduce_context_t *context =
|
||||
(ompi_coll_adapt_reduce_context_t *) req->req_complete_cb_data;
|
||||
OPAL_OUTPUT_VERBOSE((30, mca_coll_adapt_component.adapt_output,
|
||||
"[%d]: ireduce_send_cb, peer %d, seg_id %d\n", context->con->rank,
|
||||
context->peer, context->frag_id));
|
||||
int err;
|
||||
|
||||
opal_atomic_sub_fetch_32(&(context->con->ongoing_send), 1);
|
||||
|
||||
/* Send a new segment */
|
||||
OPAL_THREAD_LOCK(context->con->mutex_recv_list);
|
||||
ompi_coll_adapt_item_t *item =
|
||||
get_next_ready_item(context->con->recv_list, context->con->tree->tree_nextsize);
|
||||
OPAL_THREAD_UNLOCK(context->con->mutex_recv_list);
|
||||
|
||||
if (item != NULL) {
|
||||
/* Get new context item from free list */
|
||||
ompi_coll_adapt_reduce_context_t *send_context =
|
||||
(ompi_coll_adapt_reduce_context_t *) opal_free_list_wait(mca_coll_adapt_component.
|
||||
adapt_ireduce_context_free_list);
|
||||
if (context->con->tree->tree_nextsize > 0) {
|
||||
send_context->buff = context->con->accumbuf[item->id];
|
||||
} else {
|
||||
send_context->buff =
|
||||
context->buff + (item->id - context->frag_id) * context->con->segment_increment;
|
||||
}
|
||||
send_context->frag_id = item->id;
|
||||
send_context->peer = context->peer;
|
||||
send_context->con = context->con;
|
||||
OBJ_RETAIN(context->con);
|
||||
|
||||
opal_atomic_add_fetch_32(&(context->con->ongoing_send), 1);
|
||||
|
||||
int send_count = send_context->con->seg_count;
|
||||
if (item->id == (send_context->con->num_segs - 1)) {
|
||||
send_count = send_context->con->count - item->id * send_context->con->seg_count;
|
||||
}
|
||||
|
||||
OPAL_OUTPUT_VERBOSE((30, mca_coll_adapt_component.adapt_output,
|
||||
"[%d]: In send_cb, create isend to seg %d, peer %d, tag %d\n",
|
||||
send_context->con->rank, send_context->frag_id, send_context->peer,
|
||||
send_context->con->ireduce_tag - send_context->frag_id));
|
||||
|
||||
ompi_request_t *send_req;
|
||||
err =
|
||||
MCA_PML_CALL(isend
|
||||
(send_context->buff, send_count, send_context->con->datatype,
|
||||
send_context->peer,
|
||||
context->con->ireduce_tag - send_context->frag_id,
|
||||
MCA_PML_BASE_SEND_SYNCHRONOUS, send_context->con->comm, &send_req));
|
||||
if (MPI_SUCCESS != err) {
|
||||
return err;
|
||||
}
|
||||
|
||||
/* Release the item */
|
||||
OBJ_RELEASE(item);
|
||||
|
||||
/* Set the send callback */
|
||||
ompi_request_set_callback(send_req, send_cb, send_context);
|
||||
}
|
||||
|
||||
OPAL_THREAD_LOCK(context->con->mutex_num_sent);
|
||||
int32_t num_sent = ++(context->con->num_sent_segs);
|
||||
OPAL_OUTPUT_VERBOSE((30, mca_coll_adapt_component.adapt_output,
|
||||
"[%d]: In send_cb, root = %d, num_sent = %d, num_segs = %d\n",
|
||||
context->con->rank, context->con->tree->tree_root, num_sent,
|
||||
context->con->num_segs));
|
||||
/* Check whether signal the condition, non root and sent all the segments */
|
||||
if (context->con->tree->tree_root != context->con->rank && num_sent == context->con->num_segs) {
|
||||
OPAL_THREAD_UNLOCK(context->con->mutex_num_sent);
|
||||
ireduce_request_fini(context);
|
||||
} else {
|
||||
OPAL_THREAD_UNLOCK(context->con->mutex_num_sent);
|
||||
OBJ_RELEASE(context->con);
|
||||
OPAL_OUTPUT_VERBOSE((30, mca_coll_adapt_component.adapt_output, "return context_list\n"));
|
||||
opal_free_list_return(mca_coll_adapt_component.adapt_ireduce_context_free_list,
|
||||
(opal_free_list_item_t *) context);
|
||||
}
|
||||
/* Call back function return 1, which means successful */
|
||||
req->req_free(&req);
|
||||
return 1;
|
||||
}
|
||||
|
||||
/*
|
||||
* Callback function of irecv
|
||||
*/
|
||||
static int recv_cb(ompi_request_t * req)
|
||||
{
|
||||
ompi_coll_adapt_reduce_context_t *context =
|
||||
(ompi_coll_adapt_reduce_context_t *) req->req_complete_cb_data;
|
||||
OPAL_OUTPUT_VERBOSE((30, mca_coll_adapt_component.adapt_output,
|
||||
"[%d]: ireduce_recv_cb, peer %d, seg_id %d\n", context->con->rank,
|
||||
context->peer, context->frag_id));
|
||||
|
||||
int err;
|
||||
int32_t new_id =
|
||||
opal_atomic_add_fetch_32(&(context->con->next_recv_segs[context->child_id]), 1);
|
||||
|
||||
/* Receive new segment */
|
||||
if (new_id < context->con->num_segs) {
|
||||
char *temp_recv_buf = NULL;
|
||||
ompi_coll_adapt_inbuf_t *inbuf = NULL;
|
||||
/* Set inbuf, if it it first child, recv on rbuf, else recv on inbuf */
|
||||
if (context->child_id == 0 && context->con->sbuf != MPI_IN_PLACE
|
||||
&& context->con->root == context->con->rank) {
|
||||
temp_recv_buf =
|
||||
(char *) context->con->rbuf +
|
||||
(ptrdiff_t) new_id *(ptrdiff_t) context->con->segment_increment;
|
||||
} else {
|
||||
OPAL_OUTPUT_VERBOSE((30, mca_coll_adapt_component.adapt_output,
|
||||
"[%d]: In recv_cb, alloc inbuf\n", context->con->rank));
|
||||
inbuf = (ompi_coll_adapt_inbuf_t *) opal_free_list_wait(context->con->inbuf_list);
|
||||
temp_recv_buf = inbuf->buff - context->con->lower_bound;
|
||||
}
|
||||
/* Get new context item from free list */
|
||||
ompi_coll_adapt_reduce_context_t *recv_context =
|
||||
(ompi_coll_adapt_reduce_context_t *) opal_free_list_wait(mca_coll_adapt_component.
|
||||
adapt_ireduce_context_free_list);
|
||||
recv_context->buff = temp_recv_buf;
|
||||
recv_context->frag_id = new_id;
|
||||
recv_context->child_id = context->child_id;
|
||||
recv_context->peer = context->peer;
|
||||
recv_context->con = context->con;
|
||||
OBJ_RETAIN(context->con);
|
||||
recv_context->inbuf = inbuf;
|
||||
int recv_count = recv_context->con->seg_count;
|
||||
if (new_id == (recv_context->con->num_segs - 1)) {
|
||||
recv_count = recv_context->con->count - new_id * recv_context->con->seg_count;
|
||||
}
|
||||
OPAL_OUTPUT_VERBOSE((30, mca_coll_adapt_component.adapt_output,
|
||||
"[%d]: In recv_cb, create irecv for seg %d, peer %d, inbuf %p, tag %d\n",
|
||||
context->con->rank, recv_context->frag_id, recv_context->peer,
|
||||
(void *) inbuf,
|
||||
recv_context->con->ireduce_tag - recv_context->frag_id));
|
||||
ompi_request_t *recv_req;
|
||||
err =
|
||||
MCA_PML_CALL(irecv
|
||||
(temp_recv_buf, recv_count, recv_context->con->datatype,
|
||||
recv_context->peer,
|
||||
recv_context->con->ireduce_tag - recv_context->frag_id,
|
||||
recv_context->con->comm, &recv_req));
|
||||
if (MPI_SUCCESS != err) {
|
||||
return err;
|
||||
}
|
||||
/* Set the receive callback */
|
||||
ompi_request_set_callback(recv_req, recv_cb, recv_context);
|
||||
}
|
||||
|
||||
/* Do the op */
|
||||
int op_count = context->con->seg_count;
|
||||
if (context->frag_id == (context->con->num_segs - 1)) {
|
||||
op_count = context->con->count - context->frag_id * context->con->seg_count;
|
||||
}
|
||||
|
||||
int keep_inbuf = 0;
|
||||
OPAL_THREAD_LOCK(&context->con->mutex_op_list[context->frag_id]);
|
||||
if (context->con->accumbuf[context->frag_id] == NULL) {
|
||||
if (context->inbuf == NULL) {
|
||||
OPAL_OUTPUT_VERBOSE((30, mca_coll_adapt_component.adapt_output,
|
||||
"[%d]: set accumbuf to rbuf\n", context->con->rank));
|
||||
context->con->accumbuf[context->frag_id] = context->buff;
|
||||
} else {
|
||||
keep_inbuf = 1;
|
||||
OPAL_OUTPUT_VERBOSE((30, mca_coll_adapt_component.adapt_output,
|
||||
"[%d]: set accumbuf to inbuf\n", context->con->rank));
|
||||
context->con->accumbuf[context->frag_id] =
|
||||
context->inbuf->buff - context->con->lower_bound;
|
||||
}
|
||||
/* Op sbuf and accmbuf to accumbuf */
|
||||
ompi_op_reduce(context->con->op,
|
||||
context->con->sbuf +
|
||||
(ptrdiff_t) context->frag_id * (ptrdiff_t) context->con->segment_increment,
|
||||
context->con->accumbuf[context->frag_id], op_count, context->con->datatype);
|
||||
|
||||
} else {
|
||||
if (context->inbuf == NULL) {
|
||||
/* Op rbuf and accumbuf to rbuf */
|
||||
OPAL_OUTPUT_VERBOSE((30, mca_coll_adapt_component.adapt_output,
|
||||
"[%d]: op rbuf and accumbuf to rbuf\n", context->con->rank));
|
||||
ompi_op_reduce(context->con->op, context->con->accumbuf[context->frag_id],
|
||||
context->buff, op_count, context->con->datatype);
|
||||
/* Free old accumbuf */
|
||||
OPAL_OUTPUT_VERBOSE((30, mca_coll_adapt_component.adapt_output,
|
||||
"[%d]: free old accumbuf %p\n", context->con->rank,
|
||||
(void *) to_inbuf(context->con->accumbuf[context->frag_id],
|
||||
context->con->distance)));
|
||||
opal_free_list_return(context->con->inbuf_list,
|
||||
(opal_free_list_item_t *) to_inbuf(context->con->
|
||||
accumbuf[context->frag_id],
|
||||
context->con->distance));
|
||||
/* Set accumbut to rbuf */
|
||||
context->con->accumbuf[context->frag_id] = context->buff;
|
||||
} else {
|
||||
/* Op inbuf and accmbuf to accumbuf */
|
||||
OPAL_OUTPUT_VERBOSE((30, mca_coll_adapt_component.adapt_output,
|
||||
"[%d]: op inbuf and accmbuf to accumbuf\n", context->con->rank));
|
||||
ompi_op_reduce(context->con->op, context->inbuf->buff - context->con->lower_bound,
|
||||
context->con->accumbuf[context->frag_id], op_count,
|
||||
context->con->datatype);
|
||||
}
|
||||
}
|
||||
|
||||
OPAL_THREAD_UNLOCK(&context->con->mutex_op_list[context->frag_id]);
|
||||
|
||||
/* Set recv list */
|
||||
if (context->con->rank != context->con->tree->tree_root) {
|
||||
OPAL_THREAD_LOCK(context->con->mutex_recv_list);
|
||||
add_to_list(context->con->recv_list, context->frag_id);
|
||||
OPAL_THREAD_UNLOCK(context->con->mutex_recv_list);
|
||||
}
|
||||
|
||||
/* Send to parent */
|
||||
if (context->con->rank != context->con->tree->tree_root
|
||||
&& context->con->ongoing_send < mca_coll_adapt_component.adapt_ireduce_max_send_requests) {
|
||||
OPAL_THREAD_LOCK(context->con->mutex_recv_list);
|
||||
ompi_coll_adapt_item_t *item =
|
||||
get_next_ready_item(context->con->recv_list, context->con->tree->tree_nextsize);
|
||||
OPAL_THREAD_UNLOCK(context->con->mutex_recv_list);
|
||||
|
||||
if (item != NULL) {
|
||||
/* Get new context item from free list */
|
||||
ompi_coll_adapt_reduce_context_t *send_context =
|
||||
(ompi_coll_adapt_reduce_context_t *) opal_free_list_wait(mca_coll_adapt_component.
|
||||
adapt_ireduce_context_free_list);
|
||||
send_context->buff = context->con->accumbuf[context->frag_id];
|
||||
send_context->frag_id = item->id;
|
||||
send_context->peer = context->con->tree->tree_prev;
|
||||
send_context->con = context->con;
|
||||
OBJ_RETAIN(context->con);
|
||||
opal_atomic_add_fetch_32(&(context->con->ongoing_send), 1);
|
||||
|
||||
int send_count = send_context->con->seg_count;
|
||||
if (item->id == (send_context->con->num_segs - 1)) {
|
||||
send_count = send_context->con->count - item->id * send_context->con->seg_count;
|
||||
}
|
||||
OPAL_OUTPUT_VERBOSE((30, mca_coll_adapt_component.adapt_output,
|
||||
"[%d]: In recv_cb, create isend to seg %d, peer %d, tag %d\n",
|
||||
send_context->con->rank, send_context->frag_id, send_context->peer,
|
||||
send_context->con->ireduce_tag - send_context->frag_id));
|
||||
|
||||
ompi_request_t *send_req;
|
||||
err =
|
||||
MCA_PML_CALL(isend
|
||||
(send_context->buff, send_count, send_context->con->datatype,
|
||||
send_context->peer,
|
||||
send_context->con->ireduce_tag - send_context->frag_id,
|
||||
MCA_PML_BASE_SEND_SYNCHRONOUS, send_context->con->comm, &send_req));
|
||||
if (MPI_SUCCESS != err) {
|
||||
return err;
|
||||
}
|
||||
OBJ_RELEASE(item);
|
||||
|
||||
/* Set the send callback */
|
||||
ompi_request_set_callback(send_req, send_cb, send_context);
|
||||
}
|
||||
}
|
||||
|
||||
OPAL_THREAD_LOCK(context->con->mutex_num_recv_segs);
|
||||
int num_recv_segs_t = ++(context->con->num_recv_segs);
|
||||
OPAL_OUTPUT_VERBOSE((30, mca_coll_adapt_component.adapt_output,
|
||||
"[%d]: In recv_cb, tree = %p, root = %d, num_recv = %d, num_segs = %d, num_child = %d\n",
|
||||
context->con->rank, (void *) context->con->tree,
|
||||
context->con->tree->tree_root, num_recv_segs_t, context->con->num_segs,
|
||||
context->con->tree->tree_nextsize));
|
||||
/* If this is root and has received all the segments */
|
||||
if (context->con->tree->tree_root == context->con->rank
|
||||
&& num_recv_segs_t == context->con->num_segs * context->con->tree->tree_nextsize) {
|
||||
OPAL_THREAD_UNLOCK(context->con->mutex_num_recv_segs);
|
||||
if (!keep_inbuf && context->inbuf != NULL) {
|
||||
OPAL_OUTPUT_VERBOSE((30, mca_coll_adapt_component.adapt_output,
|
||||
"[%d]: root free context inbuf %p", context->con->rank,
|
||||
(void *) context->inbuf));
|
||||
opal_free_list_return(context->con->inbuf_list,
|
||||
(opal_free_list_item_t *) context->inbuf);
|
||||
}
|
||||
ireduce_request_fini(context);
|
||||
} else {
|
||||
OPAL_THREAD_UNLOCK(context->con->mutex_num_recv_segs);
|
||||
if (!keep_inbuf && context->inbuf != NULL) {
|
||||
OPAL_OUTPUT_VERBOSE((30, mca_coll_adapt_component.adapt_output,
|
||||
"[%d]: free context inbuf %p", context->con->rank,
|
||||
(void *) context->inbuf));
|
||||
opal_free_list_return(context->con->inbuf_list,
|
||||
(opal_free_list_item_t *) context->inbuf);
|
||||
}
|
||||
OBJ_RELEASE(context->con);
|
||||
OPAL_OUTPUT_VERBOSE((30, mca_coll_adapt_component.adapt_output, "[%d]: return context_list",
|
||||
context->con->rank));
|
||||
opal_free_list_return(mca_coll_adapt_component.adapt_ireduce_context_free_list,
|
||||
(opal_free_list_item_t *) context);
|
||||
}
|
||||
req->req_free(&req);
|
||||
return 1;
|
||||
}
|
||||
|
||||
int ompi_coll_adapt_ireduce(const void *sbuf, void *rbuf, int count, struct ompi_datatype_t *dtype,
|
||||
struct ompi_op_t *op, int root, struct ompi_communicator_t *comm,
|
||||
ompi_request_t ** request, mca_coll_base_module_t * module)
|
||||
{
|
||||
OPAL_OUTPUT_VERBOSE((10, mca_coll_adapt_component.adapt_output,
|
||||
"ireduce root %d, algorithm %d, coll_adapt_ireduce_segment_size %zu, coll_adapt_ireduce_max_send_requests %d, coll_adapt_ireduce_max_recv_requests %d\n",
|
||||
root, mca_coll_adapt_component.adapt_ireduce_algorithm,
|
||||
mca_coll_adapt_component.adapt_ireduce_segment_size,
|
||||
mca_coll_adapt_component.adapt_ireduce_max_send_requests,
|
||||
mca_coll_adapt_component.adapt_ireduce_max_recv_requests));
|
||||
|
||||
ompi_coll_adapt_ireduce_fn_t reduce_func =
|
||||
ompi_coll_adapt_ireduce_algorithm_index[mca_coll_adapt_component.adapt_ireduce_algorithm].ireduce_fn_ptr;
|
||||
return reduce_func(sbuf, rbuf, count, dtype, op, root, comm, request, module);
|
||||
}
|
||||
|
||||
/*
|
||||
* Ireduce functions with different algorithms
|
||||
*/
|
||||
static int
|
||||
ompi_coll_adapt_ireduce_tuned(const void *sbuf, void *rbuf, int count,
|
||||
struct ompi_datatype_t *dtype, struct ompi_op_t *op,
|
||||
int root, struct ompi_communicator_t *comm,
|
||||
ompi_request_t ** request,
|
||||
mca_coll_base_module_t *module)
|
||||
{
|
||||
OPAL_OUTPUT_VERBOSE((10, mca_coll_adapt_component.adapt_output, "tuned not implemented\n"));
|
||||
return OMPI_ERR_NOT_IMPLEMENTED;
|
||||
}
|
||||
|
||||
static int
|
||||
ompi_coll_adapt_ireduce_binomial(const void *sbuf, void *rbuf, int count,
|
||||
struct ompi_datatype_t *dtype, struct ompi_op_t *op, int root,
|
||||
struct ompi_communicator_t *comm, ompi_request_t ** request,
|
||||
mca_coll_base_module_t * module)
|
||||
{
|
||||
return ompi_coll_adapt_ireduce_generic(sbuf, rbuf, count, dtype, op, root, comm,
|
||||
request, module, ompi_coll_base_topo_build_bmtree(comm, root),
|
||||
mca_coll_adapt_component.adapt_ireduce_segment_size);
|
||||
}
|
||||
|
||||
static int
|
||||
ompi_coll_adapt_ireduce_in_order_binomial(const void *sbuf, void *rbuf, int count,
|
||||
struct ompi_datatype_t *dtype, struct ompi_op_t *op,
|
||||
int root, struct ompi_communicator_t *comm,
|
||||
ompi_request_t ** request,
|
||||
mca_coll_base_module_t * module)
|
||||
{
|
||||
return ompi_coll_adapt_ireduce_generic(sbuf, rbuf, count, dtype, op, root, comm,
|
||||
request, module, ompi_coll_base_topo_build_in_order_bmtree(comm, root),
|
||||
mca_coll_adapt_component.adapt_ireduce_segment_size);
|
||||
}
|
||||
|
||||
static int
|
||||
ompi_coll_adapt_ireduce_binary(const void *sbuf, void *rbuf, int count,
|
||||
struct ompi_datatype_t *dtype, struct ompi_op_t *op, int root,
|
||||
struct ompi_communicator_t *comm, ompi_request_t ** request,
|
||||
mca_coll_base_module_t * module)
|
||||
{
|
||||
return ompi_coll_adapt_ireduce_generic(sbuf, rbuf, count, dtype, op, root, comm,
|
||||
request, module, ompi_coll_base_topo_build_tree(2, comm, root),
|
||||
mca_coll_adapt_component.adapt_ireduce_segment_size);
|
||||
}
|
||||
|
||||
static int
|
||||
ompi_coll_adapt_ireduce_pipeline(const void *sbuf, void *rbuf, int count,
|
||||
struct ompi_datatype_t *dtype, struct ompi_op_t *op, int root,
|
||||
struct ompi_communicator_t *comm, ompi_request_t ** request,
|
||||
mca_coll_base_module_t * module)
|
||||
{
|
||||
return ompi_coll_adapt_ireduce_generic(sbuf, rbuf, count, dtype, op, root, comm,
|
||||
request, module, ompi_coll_base_topo_build_chain(1, comm, root),
|
||||
mca_coll_adapt_component.adapt_ireduce_segment_size);
|
||||
}
|
||||
|
||||
|
||||
static int
|
||||
ompi_coll_adapt_ireduce_chain(const void *sbuf, void *rbuf, int count,
|
||||
struct ompi_datatype_t *dtype, struct ompi_op_t *op, int root,
|
||||
struct ompi_communicator_t *comm, ompi_request_t ** request,
|
||||
mca_coll_base_module_t * module)
|
||||
{
|
||||
return ompi_coll_adapt_ireduce_generic(sbuf, rbuf, count, dtype, op, root, comm,
|
||||
request, module, ompi_coll_base_topo_build_chain(4, comm, root),
|
||||
mca_coll_adapt_component.adapt_ireduce_segment_size);
|
||||
}
|
||||
|
||||
static int
|
||||
ompi_coll_adapt_ireduce_linear(const void *sbuf, void *rbuf, int count,
|
||||
struct ompi_datatype_t *dtype, struct ompi_op_t *op, int root,
|
||||
struct ompi_communicator_t *comm, ompi_request_t ** request,
|
||||
mca_coll_base_module_t * module)
|
||||
{
|
||||
int fanout = ompi_comm_size(comm) - 1;
|
||||
ompi_coll_tree_t *tree;
|
||||
if (fanout < 1) {
|
||||
tree = ompi_coll_base_topo_build_chain(1, comm, root);
|
||||
} else if (fanout <= MAXTREEFANOUT) {
|
||||
tree = ompi_coll_base_topo_build_tree(ompi_comm_size(comm) - 1, comm, root);
|
||||
} else {
|
||||
tree = ompi_coll_base_topo_build_tree(MAXTREEFANOUT, comm, root);
|
||||
}
|
||||
return ompi_coll_adapt_ireduce_generic(sbuf, rbuf, count, dtype, op, root, comm,
|
||||
request, module, tree,
|
||||
mca_coll_adapt_component.adapt_ireduce_segment_size);
|
||||
}
|
||||
|
||||
|
||||
int ompi_coll_adapt_ireduce_generic(const void *sbuf, void *rbuf, int count,
|
||||
struct ompi_datatype_t *dtype, struct ompi_op_t *op, int root,
|
||||
struct ompi_communicator_t *comm, ompi_request_t ** request,
|
||||
mca_coll_base_module_t * module, ompi_coll_tree_t * tree,
|
||||
size_t seg_size)
|
||||
{
|
||||
|
||||
ptrdiff_t extent, lower_bound, segment_increment;
|
||||
ptrdiff_t true_lower_bound, true_extent, real_seg_size;
|
||||
size_t typelng;
|
||||
int seg_count = count, num_segs, rank, recv_count, send_count, i, j, err, min, distance = 0;
|
||||
int32_t seg_index;
|
||||
opal_atomic_int_t *next_recv_segs = NULL;
|
||||
/* Used to store the accumuate result, pointer to every segment */
|
||||
char **accumbuf = NULL;
|
||||
/* A free list contains all recv data */
|
||||
opal_free_list_t *inbuf_list;
|
||||
opal_mutex_t *mutex_recv_list;
|
||||
opal_mutex_t *mutex_num_recv_segs;
|
||||
opal_mutex_t *mutex_num_sent;
|
||||
opal_mutex_t *mutex_op_list;
|
||||
/* A list to store the segments need to be sent */
|
||||
opal_list_t *recv_list;
|
||||
|
||||
/* Determine number of segments and number of elements sent per operation */
|
||||
rank = ompi_comm_rank(comm);
|
||||
ompi_datatype_get_extent(dtype, &lower_bound, &extent);
|
||||
ompi_datatype_type_size(dtype, &typelng);
|
||||
COLL_BASE_COMPUTED_SEGCOUNT(seg_size, typelng, seg_count);
|
||||
num_segs = (count + seg_count - 1) / seg_count;
|
||||
segment_increment = (ptrdiff_t) seg_count *extent;
|
||||
ompi_datatype_get_true_extent(dtype, &true_lower_bound, &true_extent);
|
||||
real_seg_size = true_extent + (ptrdiff_t) (seg_count - 1) * extent;
|
||||
|
||||
/* Atomically set up free list */
|
||||
if (NULL == mca_coll_adapt_component.adapt_ireduce_context_free_list) {
|
||||
opal_free_list_t* fl = OBJ_NEW(opal_free_list_t);
|
||||
opal_free_list_init(fl,
|
||||
sizeof(ompi_coll_adapt_reduce_context_t),
|
||||
opal_cache_line_size,
|
||||
OBJ_CLASS(ompi_coll_adapt_reduce_context_t),
|
||||
0, opal_cache_line_size,
|
||||
mca_coll_adapt_component.adapt_context_free_list_min,
|
||||
mca_coll_adapt_component.adapt_context_free_list_max,
|
||||
mca_coll_adapt_component.adapt_context_free_list_inc,
|
||||
NULL, 0, NULL, NULL, NULL);
|
||||
if( !OPAL_ATOMIC_COMPARE_EXCHANGE_STRONG_PTR((opal_atomic_intptr_t *)&mca_coll_adapt_component.adapt_ireduce_context_free_list,
|
||||
&(intptr_t){0}, fl) ) {
|
||||
OBJ_RELEASE(fl);
|
||||
}
|
||||
}
|
||||
|
||||
/* If the current process is not leaf */
|
||||
if (tree->tree_nextsize > 0) {
|
||||
inbuf_list = OBJ_NEW(opal_free_list_t);
|
||||
opal_free_list_init(inbuf_list,
|
||||
sizeof(ompi_coll_adapt_inbuf_t) + real_seg_size,
|
||||
opal_cache_line_size,
|
||||
OBJ_CLASS(ompi_coll_adapt_inbuf_t),
|
||||
0, opal_cache_line_size,
|
||||
mca_coll_adapt_component.adapt_inbuf_free_list_min,
|
||||
mca_coll_adapt_component.adapt_inbuf_free_list_max,
|
||||
mca_coll_adapt_component.adapt_inbuf_free_list_inc,
|
||||
NULL, 0, NULL, NULL, NULL);
|
||||
/* Set up next_recv_segs */
|
||||
next_recv_segs = (opal_atomic_int32_t *) malloc(sizeof(int32_t) * tree->tree_nextsize);
|
||||
ompi_coll_adapt_inbuf_t *temp_inbuf =
|
||||
(ompi_coll_adapt_inbuf_t *) opal_free_list_wait(inbuf_list);
|
||||
distance = (char *) temp_inbuf->buff - lower_bound - (char *) temp_inbuf; //address of inbuf->buff to address of inbuf
|
||||
OPAL_OUTPUT_VERBOSE((30, mca_coll_adapt_component.adapt_output,
|
||||
"[%d]: distance %d, inbuf %p, inbuf->buff %p, inbuf->buff-lb %p, to_inbuf %p, inbuf_list %p\n",
|
||||
rank, distance, (void *) temp_inbuf, (void *) temp_inbuf->buff,
|
||||
(char *) temp_inbuf->buff - lower_bound,
|
||||
(void *) to_inbuf((char *) temp_inbuf->buff - lower_bound, distance),
|
||||
(void *) inbuf_list));
|
||||
opal_free_list_return(inbuf_list, (opal_free_list_item_t *) temp_inbuf);
|
||||
} else {
|
||||
inbuf_list = NULL;
|
||||
next_recv_segs = NULL;
|
||||
}
|
||||
|
||||
ompi_request_t *temp_request = NULL;
|
||||
/* Set up request */
|
||||
temp_request = OBJ_NEW(ompi_request_t);
|
||||
OMPI_REQUEST_INIT(temp_request, false);
|
||||
temp_request->req_state = OMPI_REQUEST_ACTIVE;
|
||||
temp_request->req_type = OMPI_REQUEST_COLL;
|
||||
temp_request->req_free = ompi_coll_adapt_request_free;
|
||||
temp_request->req_status.MPI_SOURCE = 0;
|
||||
temp_request->req_status.MPI_TAG = 0;
|
||||
temp_request->req_status.MPI_ERROR = 0;
|
||||
temp_request->req_status._cancelled = 0;
|
||||
temp_request->req_status._ucount = 0;
|
||||
*request = temp_request;
|
||||
|
||||
/* Set up mutex */
|
||||
mutex_recv_list = OBJ_NEW(opal_mutex_t);
|
||||
mutex_num_recv_segs = OBJ_NEW(opal_mutex_t);
|
||||
mutex_op_list = (opal_mutex_t *) malloc(sizeof(opal_mutex_t) * num_segs);
|
||||
for (i = 0; i < num_segs; i++) {
|
||||
OBJ_CONSTRUCT(&mutex_op_list[i], opal_mutex_t);
|
||||
}
|
||||
mutex_num_sent = OBJ_NEW(opal_mutex_t);
|
||||
/* Create recv_list */
|
||||
recv_list = OBJ_NEW(opal_list_t);
|
||||
|
||||
/* Set constant context for send and recv call back */
|
||||
ompi_coll_adapt_constant_reduce_context_t *con =
|
||||
OBJ_NEW(ompi_coll_adapt_constant_reduce_context_t);
|
||||
con->count = count;
|
||||
con->seg_count = seg_count;
|
||||
con->datatype = dtype;
|
||||
con->comm = comm;
|
||||
con->segment_increment = segment_increment;
|
||||
con->num_segs = num_segs;
|
||||
con->request = temp_request;
|
||||
con->rank = rank;
|
||||
con->num_recv_segs = 0;
|
||||
con->num_sent_segs = 0;
|
||||
con->next_recv_segs = next_recv_segs;
|
||||
con->mutex_recv_list = mutex_recv_list;
|
||||
con->mutex_num_recv_segs = mutex_num_recv_segs;
|
||||
con->mutex_num_sent = mutex_num_sent;
|
||||
con->mutex_op_list = mutex_op_list;
|
||||
con->op = op;
|
||||
con->tree = tree;
|
||||
con->inbuf_list = inbuf_list;
|
||||
con->recv_list = recv_list;
|
||||
con->lower_bound = lower_bound;
|
||||
con->ongoing_send = 0;
|
||||
con->sbuf = (char *) sbuf;
|
||||
con->rbuf = (char *) rbuf;
|
||||
con->root = root;
|
||||
con->distance = distance;
|
||||
con->ireduce_tag = ompi_coll_base_nbc_reserve_tags(comm, num_segs);
|
||||
con->real_seg_size = real_seg_size;
|
||||
|
||||
OPAL_OUTPUT_VERBOSE((30, mca_coll_adapt_component.adapt_output,
|
||||
"[%d]: start ireduce root %d tag %d\n", rank, tree->tree_root,
|
||||
con->ireduce_tag));
|
||||
|
||||
/* If the current process is not leaf node */
|
||||
if (tree->tree_nextsize > 0) {
|
||||
/* Set up accumbuf */
|
||||
accumbuf = (char **) malloc(sizeof(char *) * num_segs);
|
||||
if (root == rank && sbuf == MPI_IN_PLACE) {
|
||||
for (i = 0; i < num_segs; i++) {
|
||||
accumbuf[i] = (char *) rbuf + (ptrdiff_t) i *(ptrdiff_t) segment_increment;
|
||||
}
|
||||
} else {
|
||||
for (i = 0; i < num_segs; i++) {
|
||||
accumbuf[i] = NULL;
|
||||
}
|
||||
}
|
||||
|
||||
con->accumbuf = accumbuf;
|
||||
|
||||
/* For the first batch of segments */
|
||||
if (num_segs <= mca_coll_adapt_component.adapt_ireduce_max_recv_requests) {
|
||||
min = num_segs;
|
||||
} else {
|
||||
min = mca_coll_adapt_component.adapt_ireduce_max_recv_requests;
|
||||
}
|
||||
for (i = 0; i < tree->tree_nextsize; i++) {
|
||||
next_recv_segs[i] = min - 1;
|
||||
}
|
||||
|
||||
for (j = 0; j < min; j++) {
|
||||
/* For each child */
|
||||
for (i = 0; i < tree->tree_nextsize; i++) {
|
||||
seg_index = j;
|
||||
if (seg_index < num_segs) {
|
||||
recv_count = seg_count;
|
||||
if (seg_index == (num_segs - 1)) {
|
||||
recv_count = count - (ptrdiff_t) seg_count *(ptrdiff_t) seg_index;
|
||||
}
|
||||
char *temp_recv_buf = NULL;
|
||||
ompi_coll_adapt_inbuf_t *inbuf = NULL;
|
||||
/* Set inbuf, if it it first child, recv on rbuf, else recv on inbuf */
|
||||
if (i == 0 && sbuf != MPI_IN_PLACE && root == rank) {
|
||||
temp_recv_buf =
|
||||
(char *) rbuf + (ptrdiff_t) j *(ptrdiff_t) segment_increment;
|
||||
} else {
|
||||
inbuf = (ompi_coll_adapt_inbuf_t *) opal_free_list_wait(inbuf_list);
|
||||
OPAL_OUTPUT_VERBOSE((30, mca_coll_adapt_component.adapt_output,
|
||||
"[%d]: In ireduce, alloc inbuf %p\n", rank,
|
||||
(void *) inbuf));
|
||||
temp_recv_buf = inbuf->buff - lower_bound;
|
||||
}
|
||||
/* Get context */
|
||||
ompi_coll_adapt_reduce_context_t *context =
|
||||
(ompi_coll_adapt_reduce_context_t *)
|
||||
opal_free_list_wait(mca_coll_adapt_component.
|
||||
adapt_ireduce_context_free_list);
|
||||
context->buff = temp_recv_buf;
|
||||
context->frag_id = seg_index;
|
||||
context->child_id = i; //the id of peer in in the tree
|
||||
context->peer = tree->tree_next[i]; //the actural rank of the peer
|
||||
context->con = con;
|
||||
OBJ_RETAIN(con);
|
||||
context->inbuf = inbuf;
|
||||
|
||||
OPAL_OUTPUT_VERBOSE((30, mca_coll_adapt_component.adapt_output,
|
||||
"[%d]: In ireduce, create irecv for seg %d, peer %d, recv_count %d, inbuf %p tag %d\n",
|
||||
context->con->rank, context->frag_id, context->peer,
|
||||
recv_count, (void *) inbuf,
|
||||
con->ireduce_tag - seg_index));
|
||||
|
||||
/* Create a recv request */
|
||||
ompi_request_t *recv_req;
|
||||
err =
|
||||
MCA_PML_CALL(irecv
|
||||
(temp_recv_buf, recv_count, dtype, tree->tree_next[i],
|
||||
con->ireduce_tag - seg_index, comm, &recv_req));
|
||||
if (MPI_SUCCESS != err) {
|
||||
return err;
|
||||
}
|
||||
/* Set the recv callback */
|
||||
ompi_request_set_callback(recv_req, recv_cb, context);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/* Leaf nodes */
|
||||
else {
|
||||
ompi_coll_adapt_item_t *item;
|
||||
/* Set up recv_list */
|
||||
for (seg_index = 0; seg_index < num_segs; seg_index++) {
|
||||
item = OBJ_NEW(ompi_coll_adapt_item_t);
|
||||
item->id = seg_index;
|
||||
item->count = tree->tree_nextsize;
|
||||
opal_list_append(recv_list, (opal_list_item_t *) item);
|
||||
}
|
||||
if (num_segs <= mca_coll_adapt_component.adapt_ireduce_max_send_requests) {
|
||||
min = num_segs;
|
||||
} else {
|
||||
min = mca_coll_adapt_component.adapt_ireduce_max_send_requests;
|
||||
}
|
||||
con->accumbuf = accumbuf;
|
||||
for (i = 0; i < min; i++) {
|
||||
OPAL_THREAD_LOCK(mutex_recv_list);
|
||||
item = get_next_ready_item(recv_list, tree->tree_nextsize);
|
||||
OPAL_THREAD_UNLOCK(mutex_recv_list);
|
||||
if (item != NULL) {
|
||||
send_count = seg_count;
|
||||
if (item->id == (num_segs - 1)) {
|
||||
send_count = count - (ptrdiff_t) seg_count *(ptrdiff_t) item->id;
|
||||
}
|
||||
ompi_coll_adapt_reduce_context_t *context =
|
||||
(ompi_coll_adapt_reduce_context_t *)
|
||||
opal_free_list_wait(mca_coll_adapt_component.adapt_ireduce_context_free_list);
|
||||
context->buff =
|
||||
(char *) sbuf + (ptrdiff_t) item->id * (ptrdiff_t) segment_increment;
|
||||
context->frag_id = item->id;
|
||||
/* Actural rank of the peer */
|
||||
context->peer = tree->tree_prev;
|
||||
context->con = con;
|
||||
OBJ_RETAIN(con);
|
||||
context->inbuf = NULL;
|
||||
|
||||
opal_atomic_add_fetch_32(&(context->con->ongoing_send), 1);
|
||||
OPAL_OUTPUT_VERBOSE((30, mca_coll_adapt_component.adapt_output,
|
||||
"[%d]: In ireduce, create isend to seg %d, peer %d, send_count %d tag %d\n",
|
||||
context->con->rank, context->frag_id, context->peer,
|
||||
send_count, con->ireduce_tag - context->frag_id));
|
||||
|
||||
/* Create send request */
|
||||
ompi_request_t *send_req;
|
||||
err =
|
||||
MCA_PML_CALL(isend
|
||||
(context->buff, send_count, dtype, tree->tree_prev,
|
||||
con->ireduce_tag - context->frag_id,
|
||||
MCA_PML_BASE_SEND_SYNCHRONOUS, comm, &send_req));
|
||||
if (MPI_SUCCESS != err) {
|
||||
return err;
|
||||
}
|
||||
OBJ_RELEASE(item);
|
||||
|
||||
/* Set the send callback */
|
||||
ompi_request_set_callback(send_req, send_cb, context);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
return MPI_SUCCESS;
|
||||
}
|
||||
15
ompi/mca/coll/adapt/coll_adapt_item.c
Обычный файл
15
ompi/mca/coll/adapt/coll_adapt_item.c
Обычный файл
@ -0,0 +1,15 @@
|
||||
/*
|
||||
* 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_item.h"
|
||||
|
||||
OBJ_CLASS_INSTANCE(ompi_coll_adapt_item_t, opal_list_item_t,
|
||||
NULL, NULL);
|
||||
25
ompi/mca/coll/adapt/coll_adapt_item.h
Обычный файл
25
ompi/mca/coll/adapt/coll_adapt_item.h
Обычный файл
@ -0,0 +1,25 @@
|
||||
/*
|
||||
* 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 "opal/class/opal_list.h"
|
||||
#include "coll_adapt_inbuf.h"
|
||||
|
||||
struct ompi_coll_adapt_item_s {
|
||||
opal_list_item_t super;
|
||||
/* Fragment id */
|
||||
int id;
|
||||
/* The number of children which have received the current segment */
|
||||
int count;
|
||||
};
|
||||
|
||||
typedef struct ompi_coll_adapt_item_s ompi_coll_adapt_item_t;
|
||||
|
||||
OBJ_CLASS_DECLARATION(ompi_coll_adapt_item_t);
|
||||
164
ompi/mca/coll/adapt/coll_adapt_module.c
Обычный файл
164
ompi/mca/coll/adapt/coll_adapt_module.c
Обычный файл
@ -0,0 +1,164 @@
|
||||
/*
|
||||
* 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_config.h"
|
||||
|
||||
#include <stdio.h>
|
||||
#ifdef HAVE_STRING_H
|
||||
#include <string.h>
|
||||
#endif /* HAVE_STRING_H */
|
||||
#ifdef HAVE_SCHED_H
|
||||
#include <sched.h>
|
||||
#endif /* HAVE_SCHED_H */
|
||||
#include <sys/types.h>
|
||||
#ifdef HAVE_SYS_MMAN_H
|
||||
#include <sys/mman.h>
|
||||
#endif /* HAVE_SYS_MMAN_H */
|
||||
#ifdef HAVE_UNISTD_H
|
||||
#include <unistd.h>
|
||||
#endif /* HAVE_UNISTD_H */
|
||||
|
||||
#include "mpi.h"
|
||||
#include "opal_stdint.h"
|
||||
#include "opal/util/os_path.h"
|
||||
|
||||
#include "ompi/communicator/communicator.h"
|
||||
#include "ompi/group/group.h"
|
||||
#include "ompi/mca/coll/coll.h"
|
||||
#include "ompi/mca/coll/base/base.h"
|
||||
#include "ompi/mca/coll/base/coll_base_functions.h"
|
||||
#include "ompi/proc/proc.h"
|
||||
#include "coll_adapt.h"
|
||||
|
||||
#include "ompi/mca/coll/base/coll_tags.h"
|
||||
#include "ompi/mca/pml/pml.h"
|
||||
#include "coll_adapt_algorithms.h"
|
||||
|
||||
|
||||
/*
|
||||
* Local functions
|
||||
*/
|
||||
|
||||
/*
|
||||
* Module constructor
|
||||
*/
|
||||
static void adapt_module_construct(mca_coll_adapt_module_t * module)
|
||||
{
|
||||
module->adapt_enabled = false;
|
||||
}
|
||||
|
||||
/*
|
||||
* Module destructor
|
||||
*/
|
||||
static void adapt_module_destruct(mca_coll_adapt_module_t * module)
|
||||
{
|
||||
module->adapt_enabled = false;
|
||||
}
|
||||
|
||||
|
||||
OBJ_CLASS_INSTANCE(mca_coll_adapt_module_t,
|
||||
mca_coll_base_module_t,
|
||||
adapt_module_construct,
|
||||
adapt_module_destruct);
|
||||
|
||||
/*
|
||||
* Init module on the communicator
|
||||
*/
|
||||
static int adapt_module_enable(mca_coll_base_module_t * module,
|
||||
struct ompi_communicator_t *comm)
|
||||
{
|
||||
return OMPI_SUCCESS;
|
||||
}
|
||||
|
||||
/*
|
||||
* Initial query function that is invoked during MPI_INIT, allowing
|
||||
* this component to disqualify itself if it doesn't support the
|
||||
* required level of thread support. This function is invoked exactly
|
||||
* once.
|
||||
*/
|
||||
int ompi_coll_adapt_init_query(bool enable_progress_threads, bool enable_mpi_threads)
|
||||
{
|
||||
return OMPI_SUCCESS;
|
||||
}
|
||||
|
||||
/*
|
||||
* Invoked when there's a new communicator that has been created.
|
||||
* Look at the communicator and decide which set of functions and
|
||||
* priority we want to return.
|
||||
*/
|
||||
mca_coll_base_module_t *ompi_coll_adapt_comm_query(struct ompi_communicator_t * comm,
|
||||
int *priority)
|
||||
{
|
||||
mca_coll_adapt_module_t *adapt_module;
|
||||
|
||||
/* If we're intercomm, or if there's only one process in the communicator */
|
||||
if (OMPI_COMM_IS_INTER(comm) || 1 == ompi_comm_size(comm)) {
|
||||
opal_output_verbose(10, ompi_coll_base_framework.framework_output,
|
||||
"coll:adapt:comm_query (%d/%s): intercomm, "
|
||||
"comm is too small; disqualifying myself",
|
||||
comm->c_contextid, comm->c_name);
|
||||
return NULL;
|
||||
}
|
||||
|
||||
/* Get the priority level attached to this module.
|
||||
If priority is less than or equal to 0, then the module is unavailable. */
|
||||
*priority = mca_coll_adapt_component.adapt_priority;
|
||||
if (mca_coll_adapt_component.adapt_priority <= 0) {
|
||||
opal_output_verbose(10, ompi_coll_base_framework.framework_output,
|
||||
"coll:adapt:comm_query (%d/%s): priority too low; "
|
||||
"disqualifying myself",
|
||||
comm->c_contextid, comm->c_name);
|
||||
return NULL;
|
||||
}
|
||||
|
||||
adapt_module = OBJ_NEW(mca_coll_adapt_module_t);
|
||||
if (NULL == adapt_module) {
|
||||
return NULL;
|
||||
}
|
||||
|
||||
/* All is good -- return a module */
|
||||
adapt_module->super.coll_module_enable = adapt_module_enable;
|
||||
adapt_module->super.ft_event = NULL;
|
||||
adapt_module->super.coll_allgather = NULL;
|
||||
adapt_module->super.coll_allgatherv = NULL;
|
||||
adapt_module->super.coll_allreduce = NULL;
|
||||
adapt_module->super.coll_alltoall = NULL;
|
||||
adapt_module->super.coll_alltoallw = NULL;
|
||||
adapt_module->super.coll_barrier = NULL;
|
||||
adapt_module->super.coll_bcast = ompi_coll_adapt_bcast;
|
||||
adapt_module->super.coll_exscan = NULL;
|
||||
adapt_module->super.coll_gather = NULL;
|
||||
adapt_module->super.coll_gatherv = NULL;
|
||||
adapt_module->super.coll_reduce = ompi_coll_adapt_reduce;
|
||||
adapt_module->super.coll_reduce_scatter = NULL;
|
||||
adapt_module->super.coll_scan = NULL;
|
||||
adapt_module->super.coll_scatter = NULL;
|
||||
adapt_module->super.coll_scatterv = NULL;
|
||||
adapt_module->super.coll_ibcast = ompi_coll_adapt_ibcast;
|
||||
adapt_module->super.coll_ireduce = ompi_coll_adapt_ireduce;
|
||||
adapt_module->super.coll_iallreduce = NULL;
|
||||
|
||||
opal_output_verbose(10, ompi_coll_base_framework.framework_output,
|
||||
"coll:adapt:comm_query (%d/%s): pick me! pick me!",
|
||||
comm->c_contextid, comm->c_name);
|
||||
return &(adapt_module->super);
|
||||
}
|
||||
|
||||
/*
|
||||
* Free ADAPT request
|
||||
*/
|
||||
int ompi_coll_adapt_request_free(ompi_request_t ** request)
|
||||
{
|
||||
(*request)->req_state = OMPI_REQUEST_INVALID;
|
||||
OBJ_RELEASE(*request);
|
||||
*request = MPI_REQUEST_NULL;
|
||||
return OMPI_SUCCESS;
|
||||
}
|
||||
28
ompi/mca/coll/adapt/coll_adapt_reduce.c
Обычный файл
28
ompi/mca/coll/adapt/coll_adapt_reduce.c
Обычный файл
@ -0,0 +1,28 @@
|
||||
/*
|
||||
* 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"
|
||||
|
||||
/* MPI_Reduce and MPI_Ireduce in the ADAPT module only work for commutative operations */
|
||||
int ompi_coll_adapt_reduce(const void *sbuf, void *rbuf, int count, struct ompi_datatype_t *dtype,
|
||||
struct ompi_op_t *op, int root, struct ompi_communicator_t *comm,
|
||||
mca_coll_base_module_t * module)
|
||||
{
|
||||
ompi_request_t *request = NULL;
|
||||
int err = ompi_coll_adapt_ireduce(sbuf, rbuf, count, dtype, op, root, comm, &request, module);
|
||||
if( MPI_SUCCESS != err ) {
|
||||
if( NULL == request )
|
||||
return err;
|
||||
}
|
||||
ompi_request_wait(&request, MPI_STATUS_IGNORE);
|
||||
return err;
|
||||
}
|
||||
@ -27,6 +27,8 @@
|
||||
#include "ompi/mca/mca.h"
|
||||
#include "ompi/datatype/ompi_datatype.h"
|
||||
#include "ompi/request/request.h"
|
||||
#include "ompi/communicator/communicator.h"
|
||||
#include "ompi/mca/coll/base/coll_tags.h"
|
||||
#include "ompi/op/op.h"
|
||||
#include "ompi/mca/pml/pml.h"
|
||||
|
||||
@ -60,6 +62,22 @@ struct ompi_coll_base_nbc_request_t {
|
||||
|
||||
OMPI_DECLSPEC OBJ_CLASS_DECLARATION(ompi_coll_base_nbc_request_t);
|
||||
|
||||
static inline int32_t
|
||||
ompi_coll_base_nbc_reserve_tags(ompi_communicator_t* comm, int32_t reserve)
|
||||
{
|
||||
int32_t tag, old_tag;
|
||||
assert( reserve > 0 );
|
||||
reread_tag: /* In case we fail to atomically update the tag */
|
||||
tag = old_tag = comm->c_nbc_tag;
|
||||
if ((tag - reserve) < MCA_COLL_BASE_TAG_NONBLOCKING_END) {
|
||||
tag = MCA_COLL_BASE_TAG_NONBLOCKING_BASE;
|
||||
}
|
||||
if( !OPAL_ATOMIC_COMPARE_EXCHANGE_STRONG_32(&comm->c_nbc_tag, &old_tag, tag - reserve) ) {
|
||||
goto reread_tag;
|
||||
}
|
||||
return tag;
|
||||
}
|
||||
|
||||
typedef struct ompi_coll_base_nbc_request_t ompi_coll_base_nbc_request_t;
|
||||
|
||||
/**
|
||||
|
||||
@ -94,7 +94,6 @@ struct ompi_coll_libnbc_module_t {
|
||||
mca_coll_base_module_t super;
|
||||
opal_mutex_t mutex;
|
||||
bool comm_registered;
|
||||
int tag;
|
||||
#ifdef NBC_CACHE_SCHEDULE
|
||||
void *NBC_Dict[NBC_NUM_COLL]; /* this should point to a struct
|
||||
hb_tree, but since this is a
|
||||
|
||||
@ -25,7 +25,7 @@
|
||||
* Additional copyrights may follow
|
||||
*/
|
||||
#include "nbc_internal.h"
|
||||
#include "ompi/mca/coll/base/coll_tags.h"
|
||||
#include "ompi/mca/coll/base/coll_base_util.h"
|
||||
#include "ompi/op/op.h"
|
||||
#include "ompi/mca/pml/pml.h"
|
||||
|
||||
@ -595,7 +595,6 @@ void NBC_Return_handle(ompi_coll_libnbc_request_t *request) {
|
||||
}
|
||||
|
||||
int NBC_Init_comm(MPI_Comm comm, NBC_Comminfo *comminfo) {
|
||||
comminfo->tag= MCA_COLL_BASE_TAG_NONBLOCKING_BASE;
|
||||
|
||||
#ifdef NBC_CACHE_SCHEDULE
|
||||
/* initialize the NBC_ALLTOALL SchedCache tree */
|
||||
@ -672,7 +671,7 @@ int NBC_Start(NBC_Handle *handle) {
|
||||
int NBC_Schedule_request(NBC_Schedule *schedule, ompi_communicator_t *comm,
|
||||
ompi_coll_libnbc_module_t *module, bool persistent,
|
||||
ompi_request_t **request, void *tmpbuf) {
|
||||
int ret, tmp_tag;
|
||||
int ret;
|
||||
bool need_register = false;
|
||||
ompi_coll_libnbc_request_t *handle;
|
||||
|
||||
@ -685,13 +684,7 @@ int NBC_Schedule_request(NBC_Schedule *schedule, ompi_communicator_t *comm,
|
||||
|
||||
/* update the module->tag here because other processes may have operations
|
||||
* and they may update the module->tag */
|
||||
OPAL_THREAD_LOCK(&module->mutex);
|
||||
tmp_tag = module->tag--;
|
||||
if (tmp_tag == MCA_COLL_BASE_TAG_NONBLOCKING_END) {
|
||||
tmp_tag = module->tag = MCA_COLL_BASE_TAG_NONBLOCKING_BASE;
|
||||
NBC_DEBUG(2,"resetting tags ...\n");
|
||||
}
|
||||
OPAL_THREAD_UNLOCK(&module->mutex);
|
||||
(void)ompi_coll_base_nbc_reserve_tags(comm, 1);
|
||||
|
||||
OBJ_RELEASE(schedule);
|
||||
free(tmpbuf);
|
||||
@ -712,20 +705,15 @@ int NBC_Schedule_request(NBC_Schedule *schedule, ompi_communicator_t *comm,
|
||||
|
||||
/******************** Do the tag and shadow comm administration ... ***************/
|
||||
|
||||
OPAL_THREAD_LOCK(&module->mutex);
|
||||
tmp_tag = module->tag--;
|
||||
if (tmp_tag == MCA_COLL_BASE_TAG_NONBLOCKING_END) {
|
||||
tmp_tag = module->tag = MCA_COLL_BASE_TAG_NONBLOCKING_BASE;
|
||||
NBC_DEBUG(2,"resetting tags ...\n");
|
||||
}
|
||||
handle->tag = ompi_coll_base_nbc_reserve_tags(comm, 1);
|
||||
|
||||
OPAL_THREAD_LOCK(&module->mutex);
|
||||
if (true != module->comm_registered) {
|
||||
module->comm_registered = true;
|
||||
need_register = true;
|
||||
}
|
||||
OPAL_THREAD_UNLOCK(&module->mutex);
|
||||
|
||||
handle->tag = tmp_tag;
|
||||
|
||||
/* register progress */
|
||||
if (need_register) {
|
||||
@ -737,7 +725,6 @@ int NBC_Schedule_request(NBC_Schedule *schedule, ompi_communicator_t *comm,
|
||||
}
|
||||
|
||||
handle->comm=comm;
|
||||
/*printf("got module: %lu tag: %i\n", module, module->tag);*/
|
||||
|
||||
/******************** end of tag and shadow comm administration ... ***************/
|
||||
handle->comminfo = module;
|
||||
|
||||
@ -215,17 +215,8 @@ static int tuned_open(void)
|
||||
int rc;
|
||||
|
||||
#if OPAL_ENABLE_DEBUG
|
||||
{
|
||||
int param;
|
||||
|
||||
param = mca_base_var_find("ompi", "coll", "base", "verbose");
|
||||
if (param >= 0) {
|
||||
const int *verbose = NULL;
|
||||
mca_base_var_get_value(param, &verbose, NULL, NULL);
|
||||
if (verbose && verbose[0] > 0) {
|
||||
ompi_coll_tuned_stream = opal_output_open(NULL);
|
||||
}
|
||||
}
|
||||
if (ompi_coll_base_framework.framework_verbose) {
|
||||
ompi_coll_tuned_stream = opal_output_open(NULL);
|
||||
}
|
||||
#endif /* OPAL_ENABLE_DEBUG */
|
||||
|
||||
|
||||
@ -96,6 +96,14 @@ int MPI_Ibcast(void *buffer, int count, MPI_Datatype datatype,
|
||||
}
|
||||
}
|
||||
|
||||
/* If there's only one node, or if the count is 0, we're done */
|
||||
|
||||
if ((OMPI_COMM_IS_INTRA(comm) && ompi_comm_size(comm) <= 1) ||
|
||||
0 == count) {
|
||||
*request = &ompi_request_empty;
|
||||
return MPI_SUCCESS;
|
||||
}
|
||||
|
||||
OPAL_CR_ENTER_LIBRARY();
|
||||
|
||||
/* Invoke the coll component to perform the back-end operation */
|
||||
|
||||
@ -97,6 +97,11 @@ int MPI_Ireduce_scatter_block(const void *sendbuf, void *recvbuf, int recvcount,
|
||||
OMPI_ERRHANDLER_CHECK(err, comm, err, FUNC_NAME);
|
||||
}
|
||||
|
||||
if (0 == recvcount) {
|
||||
*request = &ompi_request_empty;
|
||||
return MPI_SUCCESS;
|
||||
}
|
||||
|
||||
OPAL_CR_ENTER_LIBRARY();
|
||||
|
||||
/* Invoke the coll component to perform the back-end operation */
|
||||
|
||||
@ -94,6 +94,9 @@ int MPI_Reduce_scatter_block(const void *sendbuf, void *recvbuf, int recvcount,
|
||||
OMPI_CHECK_DATATYPE_FOR_SEND(err, datatype, recvcount);
|
||||
OMPI_ERRHANDLER_CHECK(err, comm, err, FUNC_NAME);
|
||||
}
|
||||
if (0 == recvcount) {
|
||||
return MPI_SUCCESS;
|
||||
}
|
||||
|
||||
OPAL_CR_ENTER_LIBRARY();
|
||||
|
||||
|
||||
@ -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-2016 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,
|
||||
@ -437,9 +437,11 @@ static inline int ompi_request_complete(ompi_request_t* request, bool with_signa
|
||||
{
|
||||
int rc = 0;
|
||||
|
||||
if( NULL != request->req_complete_cb) {
|
||||
rc = request->req_complete_cb( request );
|
||||
if(NULL != request->req_complete_cb) {
|
||||
/* Set the request cb to NULL to allow resetting in the callback */
|
||||
ompi_request_complete_fn_t fct = request->req_complete_cb;
|
||||
request->req_complete_cb = NULL;
|
||||
rc = fct( request );
|
||||
}
|
||||
|
||||
if (0 == rc) {
|
||||
@ -460,6 +462,21 @@ static inline int ompi_request_complete(ompi_request_t* request, bool with_signa
|
||||
return OMPI_SUCCESS;
|
||||
}
|
||||
|
||||
static inline int ompi_request_set_callback(ompi_request_t* request,
|
||||
ompi_request_complete_fn_t cb,
|
||||
void* cb_data)
|
||||
{
|
||||
request->req_complete_cb_data = cb_data;
|
||||
request->req_complete_cb = cb;
|
||||
/* If request is completed and the callback is not called, need to call callback */
|
||||
if ((NULL != request->req_complete_cb) && (request->req_complete == REQUEST_COMPLETED)) {
|
||||
ompi_request_complete_fn_t fct = request->req_complete_cb;
|
||||
request->req_complete_cb = NULL;
|
||||
return fct( request );
|
||||
}
|
||||
return OMPI_SUCCESS;
|
||||
}
|
||||
|
||||
END_C_DECLS
|
||||
|
||||
#endif
|
||||
|
||||
Загрузка…
x
Ссылка в новой задаче
Block a user