coll: add coll_group to collective interfaces

maint/gen_coll.py

@@ -94,6 +94,8 @@ def dump_allcomm_auto_blocking(name):
  dump_open("MPIR_Csel_coll_sig_s coll_sig = {")
  G.out.append(".coll_type = MPIR_CSEL_COLL_TYPE__%s," % NAME)
  G.out.append(".comm_ptr = comm_ptr,")
+ if not re.match(r'i?neighbor_', func_name, re.IGNORECASE):
+ G.out.append(".coll_group = coll_group,")
  for p in func['parameters']:
  if not re.match(r'comm$', p['name']):
  G.out.append(".u.%s.%s = %s," % (func_name, p['name'], p['name']))
@@ -163,12 +165,16 @@ def dump_allcomm_sched_auto(name):
  dump_split(0, "int MPIR_%s_allcomm_sched_auto(%s)" % (Name, func_params))
  dump_open('{')
  G.out.append("int mpi_errno = MPI_SUCCESS;")
+ if re.match(r'Ineighbor_', Name):
+ G.out.append("int coll_group = MPIR_SUBGROUP_NONE;")
  G.out.append("")
 
  # -- Csel_search
  dump_open("MPIR_Csel_coll_sig_s coll_sig = {")
  G.out.append(".coll_type = MPIR_CSEL_COLL_TYPE__%s," % NAME)
  G.out.append(".comm_ptr = comm_ptr,")
+ if not re.match(r'i?neighbor_', func_name, re.IGNORECASE):
+ G.out.append(".coll_group = coll_group,")
  for p in func['parameters']:
  if not re.match(r'comm$', p['name']):
  G.out.append(".u.%s.%s = %s," % (func_name, p['name'], p['name']))
@@ -363,6 +369,8 @@ def dump_cases(commkind):
  dump_split(0, "int MPIR_%s_sched_impl(%s)" % (Name, func_params))
  dump_open('{')
  G.out.append("int mpi_errno = MPI_SUCCESS;")
+ if re.match(r'Ineighbor_', Name):
+ G.out.append("int coll_group = MPIR_SUBGROUP_NONE;")
  G.out.append("")
 
  dump_open("if (comm_ptr->comm_kind == MPIR_COMM_KIND__INTRACOMM) {")
@@ -552,20 +560,22 @@ def dump_fallback(algo):
  elif a == "noinplace":
  cond_list.append("sendbuf != MPI_IN_PLACE")
  elif a == "power-of-two":
- cond_list.append("comm_ptr->local_size == comm_ptr->coll.pof2")
+ cond_list.append("MPL_is_pof2(MPIR_Coll_size(comm_ptr, coll_group))")
  elif a == "size-ge-pof2":
- cond_list.append("count >= comm_ptr->coll.pof2")
+ cond_list.append("count >= MPL_pof2(MPIR_Coll_size(comm_ptr, coll_group))")
  elif a == "commutative":
  cond_list.append("MPIR_Op_is_commutative(op)")
  elif a== "builtin-op":
  cond_list.append("HANDLE_IS_BUILTIN(op)")
  elif a == "parent-comm":
- cond_list.append("MPIR_Comm_is_parent_comm(comm_ptr)")
+ cond_list.append("MPIR_Comm_is_parent_comm(comm_ptr, coll_group)")
  elif a == "node-consecutive":
  cond_list.append("MPII_Comm_is_node_consecutive(comm_ptr)")
  elif a == "displs-ordered":
  # assume it's allgatherv
  cond_list.append("MPII_Iallgatherv_is_displs_ordered(comm_ptr->local_size, recvcounts, displs)")
+ elif a == "nogroup":
+ cond_list.append("coll_group == MPIR_SUBGROUP_NONE")
  else:
  raise Exception("Unsupported restrictions - %s" % a)
  (func_name, commkind) = algo['func-commkind'].split('-')
@@ -644,6 +654,9 @@ def get_algo_extra_params(algo):
 # additional wrappers
 def get_algo_args(args, algo, kind):
  algo_args = args
+ if not re.match(r'i?neighbor_', algo['func-commkind']):
+ algo_args += ", coll_group"
+
  if 'extra_params' in algo:
  algo_args += ", " + get_algo_extra_args(algo, kind)
 
@@ -658,6 +671,9 @@ def get_algo_args(args, algo, kind):
 
 def get_algo_params(params, algo):
  algo_params = params
+ if not re.match(r'i?neighbor_', algo['func-commkind']):
+ algo_params += ", int coll_group"
+
  if 'extra_params' in algo:
  algo_params += ", " + get_algo_extra_params(algo)
 
@@ -681,6 +697,8 @@ def get_algo_name(algo):
 
 def get_func_params(params, name, kind):
  func_params = params
+ if not name.startswith('neighbor_'):
+ func_params += ", int coll_group"
  if kind == "blocking":
  if not name.startswith('neighbor_'):
  func_params += ", MPIR_Errflag_t errflag"
@@ -701,6 +719,8 @@ def get_func_params(params, name, kind):
 
 def get_func_args(args, name, kind):
  func_args = args
+ if not name.startswith('neighbor_'):
+ func_args += ", coll_group"
  if kind == "blocking":
  if not name.startswith('neighbor_'):
  func_args += ", errflag"

maint/local_python/binding_c.py

@@ -1686,6 +1686,8 @@ def push_impl_decl(func, impl_name=None):
  if func['_impl_param_list']:
  params = ', '.join(func['_impl_param_list'])
  if func['dir'] == 'coll':
+ if not RE.match(r'MPI_(Ineighbor|Neighbor)', func['name']):
+ params = params.replace('comm_ptr', 'comm_ptr, int coll_group')
  # block collective use an extra errflag
  if not RE.match(r'MPI_(I.*|Neighbor.*|.*_init)$', func['name']):
  params = params + ", MPIR_Errflag_t errflag"
@@ -1726,6 +1728,8 @@ def dump_body_coll(func):
  mpir_name = re.sub(r'^MPIX?_', 'MPIR_', func['name'])
 
  args = ", ".join(func['_impl_arg_list'])
+ if not RE.match(r'MPI_(Ineighbor|Neighbor)', func['name']):
+ args = args.replace('comm_ptr', 'comm_ptr, MPIR_SUBGROUP_NONE')
 
  if RE.match(r'MPI_(I.*|.*_init)$', func['name'], re.IGNORECASE):
  # non-blocking collectives
@@ -1956,6 +1960,7 @@ def dump_body_reduce_equal(func):
  args = ", ".join(func['_impl_arg_list'])
  args = re.sub(r'recvbuf, ', '', args)
  args = re.sub(r'op, ', 'recvbuf, ', args)
+ args += ", MPIR_SUBGROUP_NONE"
  dump_line_with_break("mpi_errno = %s(%s);" % (impl, args))
  dump_error_check("")
 

src/binding/c/comm_api.txt

@@ -301,7 +301,7 @@ MPI_Intercomm_merge:
  * error to make */
  acthigh = high ? 1 : 0; /* Clamp high into 1 or 0 */
  mpi_errno = MPIR_Allreduce(MPI_IN_PLACE, &acthigh, 1, MPI_INT,
-  MPI_SUM, intercomm_ptr->local_comm, MPIR_ERR_NONE);
+ MPI_SUM, intercomm_ptr->local_comm, MPIR_SUBGROUP_NONE, MPIR_ERR_NONE);
  MPIR_ERR_CHECK(mpi_errno);
  /* acthigh must either == 0 or the size of the local comm */
  if (acthigh != 0 && acthigh != intercomm_ptr->local_size) {

src/include/mpir_coll.h

@@ -8,6 +8,52 @@
 
 #include "coll_impl.h"
 #include "coll_algos.h"
+#include "mpir_threadcomm.h"
+
+#ifdef ENABLE_THREADCOMM
+#define MPIR_THREADCOMM_RANK_SIZE(comm, rank_, size_) do { \
+ MPIR_Threadcomm *threadcomm = (comm)->threadcomm; \
+ MPIR_Assert(threadcomm); \
+ int intracomm_size = (comm)->local_size; \
+ size_ = threadcomm->rank_offset_table[intracomm_size - 1]; \
+ rank_ = MPIR_THREADCOMM_TID_TO_RANK(threadcomm, MPIR_threadcomm_get_tid(threadcomm)); \
+ } while (0)
+#else
+#define MPIR_THREADCOMM_RANK_SIZE(comm, rank_, size_) do { \
+ MPIR_Assert(0); \
+ size_ = 0; \
+ rank_ = -1; \
+ } while (0)
+#endif
+
+#define MPIR_COLL_RANK_SIZE(comm, coll_group, rank_, size_) do { \
+ if (coll_group == MPIR_SUBGROUP_NONE) { \
+ rank_ = (comm)->rank; \
+ size_ = (comm)->local_size; \
+ } else if (coll_group == MPIR_SUBGROUP_THREADCOMM) { \
+ MPIR_THREADCOMM_RANK_SIZE(comm, rank_, size_); \
+ } else { \
+ rank_ = (comm)->subgroups[coll_group].rank; \
+ size_ = (comm)->subgroups[coll_group].size; \
+ } \
+ } while (0)
+
+/* sometime it is convenient to just get the rank or size */
+static inline int MPIR_Coll_size(MPIR_Comm * comm, int coll_group)
+{
+ int rank, size;
+ MPIR_COLL_RANK_SIZE(comm, coll_group, rank, size);
+ (void) rank;
+ return size;
+}
+
+static inline int MPIR_Coll_rank(MPIR_Comm * comm, int coll_group)
+{
+ int rank, size;
+ MPIR_COLL_RANK_SIZE(comm, coll_group, rank, size);
+ (void) size;
+ return rank;
+}
 
 /* During init, not all algorithms are safe to use. For example, the csel
  * may not have been initialized. We define a set of fallback routines that
@@ -28,36 +74,41 @@ int MPIC_Wait(MPIR_Request * request_ptr);
 int MPIC_Probe(int source, int tag, MPI_Comm comm, MPI_Status * status);
 
 int MPIC_Send(const void *buf, MPI_Aint count, MPI_Datatype datatype, int dest, int tag,
- MPIR_Comm * comm_ptr, MPIR_Errflag_t errflag);
+ MPIR_Comm * comm_ptr, int coll_group, MPIR_Errflag_t errflag);
 int MPIC_Recv(void *buf, MPI_Aint count, MPI_Datatype datatype, int source, int tag,
- MPIR_Comm * comm_ptr, MPI_Status * status);
+ MPIR_Comm * comm_ptr, int coll_group, MPI_Status * status);
 int MPIC_Sendrecv(const void *sendbuf, MPI_Aint sendcount, MPI_Datatype sendtype,
  int dest, int sendtag, void *recvbuf, MPI_Aint recvcount,
  MPI_Datatype recvtype, int source, int recvtag,
- MPIR_Comm * comm_ptr, MPI_Status * status, MPIR_Errflag_t errflag);
-int MPIC_Sendrecv_replace(void *buf, MPI_Aint count, MPI_Datatype datatype,
-  int dest, int sendtag,
- int source, int recvtag,
- MPIR_Comm * comm_ptr, MPI_Status * status, MPIR_Errflag_t errflag);
+ MPIR_Comm * comm_ptr, int coll_group, MPI_Status * status,
+ MPIR_Errflag_t errflag);
+int MPIC_Sendrecv_replace(void *buf, MPI_Aint count, MPI_Datatype datatype, int dest, int sendtag,
+ int source, int recvtag, MPIR_Comm * comm_ptr, int coll_group,
+ MPI_Status * status, MPIR_Errflag_t errflag);
 int MPIC_Isend(const void *buf, MPI_Aint count, MPI_Datatype datatype, int dest, int tag,
- MPIR_Comm * comm_ptr, MPIR_Request ** request, MPIR_Errflag_t errflag);
-int MPIC_Irecv(void *buf, MPI_Aint count, MPI_Datatype datatype, int source,
- int tag, MPIR_Comm * comm_ptr, MPIR_Request ** request);
+ MPIR_Comm * comm_ptr, int coll_group, MPIR_Request ** request,
+ MPIR_Errflag_t errflag);
+int MPIC_Irecv(void *buf, MPI_Aint count, MPI_Datatype datatype, int source, int tag,
+ MPIR_Comm * comm_ptr, int coll_group, MPIR_Request ** request);
 int MPIC_Waitall(int numreq, MPIR_Request * requests[], MPI_Status * statuses);
 
 int MPIR_Reduce_local(const void *inbuf, void *inoutbuf, MPI_Aint count, MPI_Datatype datatype,
  MPI_Op op);
 
-int MPIR_Barrier_intra_dissemination(MPIR_Comm * comm_ptr, MPIR_Errflag_t errflag);
+int MPIR_Barrier_intra_dissemination(MPIR_Comm * comm_ptr, int coll_group, MPIR_Errflag_t errflag);
 
 /* TSP auto */
 int MPIR_TSP_Iallreduce_sched_intra_tsp_auto(const void *sendbuf, void *recvbuf, MPI_Aint count,
  MPI_Datatype datatype, MPI_Op op,
- MPIR_Comm * comm, MPIR_TSP_sched_t sched);
+ MPIR_Comm * comm, int coll_group,
+ MPIR_TSP_sched_t sched);
 int MPIR_TSP_Ibcast_sched_intra_tsp_auto(void *buffer, MPI_Aint count, MPI_Datatype datatype,
- int root, MPIR_Comm * comm_ptr, MPIR_TSP_sched_t sched);
-int MPIR_TSP_Ibarrier_sched_intra_tsp_auto(MPIR_Comm * comm, MPIR_TSP_sched_t sched);
+ int root, MPIR_Comm * comm_ptr, int coll_group,
+ MPIR_TSP_sched_t sched);
+int MPIR_TSP_Ibarrier_sched_intra_tsp_auto(MPIR_Comm * comm, int coll_group,
+ MPIR_TSP_sched_t sched);
 int MPIR_TSP_Ireduce_sched_intra_tsp_auto(const void *sendbuf, void *recvbuf, MPI_Aint count,
  MPI_Datatype datatype, MPI_Op op, int root,
- MPIR_Comm * comm_ptr, MPIR_TSP_sched_t sched);
+ MPIR_Comm * comm_ptr, int coll_group,
+ MPIR_TSP_sched_t sched);
 #endif /* MPIR_COLL_H_INCLUDED */

src/include/mpir_comm.h

@@ -101,6 +101,51 @@ enum MPIR_COMM_HINT_PREDEFINED_t {
  MPIR_COMM_HINT_PREDEFINED_COUNT
 };
 
+/* MPIR_Subgroup is similar to MPIR_Group, but only used to describe subgroups within
+ * an intra communicator. The proc_table refers to ranks within the communicator.
+ * It is only used internally for group collectives.
+ */
+typedef struct MPIR_Subgroup {
+ int size;
+ int rank;
+ int *proc_table; /* can be NULL if the group is trivial */
+} MPIR_Subgroup;
+
+#define MPIR_MAX_SUBGROUPS 16
+
+/* reserved subgroup indexes */
+enum {
+ MPIR_SUBGROUP_THREADCOMM = -1,
+ MPIR_SUBGROUP_NONE = 0,
+ MPIR_SUBGROUP_NODE, /* i.e. nodecomm */
+ MPIR_SUBGROUP_NODE_CROSS, /* node_roots_comm, node_rank_1_comm, ... */
+ MPIR_SUBGROUP_NUMA1, /* 1-level below node in topology */
+ MPIR_SUBGROUP_NUMA1_CROSS, /* cross-link group at NUMA1 within NODE */
+ MPIR_SUBGROUP_NUMA2, /* and so on */
+ MPIR_SUBGROUP_NUMA2_CROSS,
+ MPIR_SUBGROUP_NUM_RESERVED,
+};
+
+/* macros to create dynamic subgroups.
+ * It is expected to fillout the proc_table after MPIR_COMM_PUSH_SUBGROUP.
+ */
+#define MPIR_COMM_PUSH_SUBGROUP(comm, _size, _rank, newgrp, proc_table_out) \
+ do { \
+ (newgrp) = (comm)->num_subgroups++; \
+ MPIR_Assert((comm)->num_subgroups < MPIR_MAX_SUBGROUPS); \
+ (comm)->subgroups[newgrp].size = _size; \
+ (comm)->subgroups[newgrp].rank = _rank; \
+ (proc_table_out) = MPL_malloc((_size) * sizeof(int), MPL_MEM_OTHER); \
+ (comm)->subgroups[newgrp].proc_table = (proc_table_out); \
+ } while (0)
+
+#define MPIR_COMM_POP_SUBGROUP(comm) \
+ do { \
+ int i = --(comm)->num_subgroups; \
+ MPIR_Assert(i > 0); \
+ MPL_free((comm)->subgroups[i].proc_table); \
+ } while (0)
+
 /*S
  MPIR_Comm - Description of the Communicator data structure
 
@@ -187,7 +232,8 @@ struct MPIR_Comm {
  int *internode_table; /* internode_table[i] gives the rank in
  * node_roots_comm of rank i in this comm.
  * It is of size 'local_size'. */
- int node_count; /* number of nodes this comm is spread over */
+ int num_local; /* number of procs in this comm on local node */
+ int num_external; /* number of nodes this comm is spread over */
 
  int is_low_group; /* For intercomms only, this boolean is
  * set for all members of one of the
@@ -196,6 +242,8 @@ struct MPIR_Comm {
  * intercommunicator collective operations
  * that wish to use half-duplex operations
  * to implement a full-duplex operation */
+ MPIR_Subgroup subgroups[MPIR_MAX_SUBGROUPS];
+ int num_subgroups;
 
  struct MPIR_Comm *comm_next; /* Provides a chain through all active
  * communicators */
@@ -222,9 +270,6 @@ struct MPIR_Comm {
  * use int array for fast access */
 
  struct {
- int pof2; /* Nearest (smaller than or equal to) power of 2
- * to the number of ranks in the communicator.
- * To be used during collective communication */
  int pofk[MAX_RADIX - 1];
  int k[MAX_RADIX - 1];
  int step1_sendto[MAX_RADIX - 1];
@@ -234,18 +279,9 @@ struct MPIR_Comm {
  int **step2_nbrs[MAX_RADIX - 1];
  int nbrs_defined[MAX_RADIX - 1];
  void **recexch_allreduce_nbr_buffer;
- int topo_aware_tree_root;
- int topo_aware_tree_k;
- MPIR_Treealgo_tree_t *topo_aware_tree;
- int topo_aware_k_tree_root;
- int topo_aware_k_tree_k;
- MPIR_Treealgo_tree_t *topo_aware_k_tree;
- int topo_wave_tree_root;
- int topo_wave_tree_overhead;
- int topo_wave_tree_lat_diff_groups;
- int topo_wave_tree_lat_diff_switches;
- int topo_wave_tree_lat_same_switches;
- MPIR_Treealgo_tree_t *topo_wave_tree;
+
+ MPIR_Treealgo_tree_t *cached_tree;
+ MPIR_Treealgo_param_t cached_tree_param;
  } coll;
 
  void *csel_comm; /* collective selector handle */
@@ -374,7 +410,7 @@ int MPIR_Comm_create_inter(MPIR_Comm * comm_ptr, MPIR_Group * group_ptr, MPIR_Co
 int MPIR_Comm_create_subcomms(MPIR_Comm * comm);
 int MPIR_Comm_commit(MPIR_Comm *);
 
-int MPIR_Comm_is_parent_comm(MPIR_Comm *);
+int MPIR_Comm_is_parent_comm(MPIR_Comm * comm, int coll_group);
 
 /* peer intercomm is an internal 1-to-1 intercomm used for connecting dynamic processes */
 int MPIR_peer_intercomm_create(MPIR_Context_id_t context_id, MPIR_Context_id_t recvcontext_id,

src/include/mpir_csel.h

@@ -60,6 +60,7 @@ typedef enum {
 typedef struct {
  MPIR_Csel_coll_type_e coll_type;
  MPIR_Comm *comm_ptr;
+ int coll_group;
 
  union {
  struct {