Skip to content

Instantly share code, notes, and snippets.

@goodell

goodell/stdinfile1.diff

Created April 19, 2012 23:16
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save goodell/2424812 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save goodell/2424812 to your computer and use it in GitHub Desktop.
Download ZIP
Raw
stdinfile1.diff
commit a5566493e8acfca25718cbf6539c562c52ddbf3e
Author: Dave Goodell <[email protected]>
Date: Thu Apr 19 18:07:54 2012 -0500
use better algorithms in MPIR_Allreduce_group
This is a copy of the non-SMP-aware code in MPIR_Allreduce_intra that
has been modified to perform rank translation immediately prior to any
communication calls. For cases where incremental group rank translation
is abnormally slow compared to batch translation, this is a non-optimal
method. A better approach would be to compute the list of ranks that we
communicate with and then translate them all at once. Such an approach
makes the code much uglier though, so I'm punting on that until we
experience such a case.
No reviewer.
---
src/mpi/coll/allred_group.c | 439 +++++++++++++++++++++++++++++++++----------
1 file changed, 338 insertions(+), 101 deletions(-)
diff --git a/src/mpi/coll/allred_group.c b/src/mpi/coll/allred_group.c
index adab806..f0ecf2e 100644
--- a/src/mpi/coll/allred_group.c
+++ b/src/mpi/coll/allred_group.c
@@ -5,139 +5,376 @@
*/
#include "mpiimpl.h"
+#include "collutil.h"
-/* Map process IDs onto a binary tree.
- *
- * [in] comm_ptr Parent communicator
- * [in] group_ptr Group on which to map the tree (must be a subgroup of comm_ptr's group)
- * [out] root Process id of the root
- * [out] up Process id of my parent
- * [out] left Process id of my left child
- * [out] right Process id of my right child
- */
-static void MPIR_Allreduce_group_bintree(MPID_Comm *comm_ptr, MPID_Group *group_ptr, int *root,
- int *up, int *left, int *right)
+/* Local utility macro: takes an two args and sets lvalue cr_ equal to the rank
+ * in comm_ptr corresponding to rvalue gr_ */
+#define to_comm_rank(cr_, gr_) \
+ do { \
+ int gr_tmp_ = (gr_); \
+ mpi_errno = MPIR_Group_translate_ranks_impl(group_ptr, 1, &(gr_tmp_), comm_ptr->local_group, &(cr_)); \
+ if (mpi_errno) MPIU_ERR_POP(mpi_errno); \
+ MPIU_Assert((cr_) != MPI_UNDEFINED); \
+ } while (0)
+
+#undef FUNCNAME
+#define FUNCNAME MPIR_Allreduce_group_intra
+#undef FCNAME
+#define FCNAME MPIU_QUOTE(FUNCNAME)
+int MPIR_Allreduce_group_intra(void *sendbuf, void *recvbuf, int count,
+ MPI_Datatype datatype, MPI_Op op, MPID_Comm *comm_ptr,
+ MPID_Group *group_ptr, int tag, int *errflag)
{
- int ranks_in[4], ranks_out[4];
- int me, nproc;
+ int comm_size, comm_rank, type_size;
+ int mpi_errno = MPI_SUCCESS;
+ int mpi_errno_ret = MPI_SUCCESS;
+ /* newrank is a rank in group_ptr */
+ int mask, dst, is_commutative, pof2, newrank, rem, newdst, i,
+ send_idx, recv_idx, last_idx, send_cnt, recv_cnt, *cnts, *disps;
+ MPI_Aint true_extent, true_lb, extent;
+ void *tmp_buf;
+ MPI_Comm comm;
+ int group_rank, group_size;
+ int cdst, csrc;
+ MPIU_CHKLMEM_DECL(3);
- me = group_ptr->rank;
- nproc = group_ptr->size;
+#ifdef MPID_HAS_HETERO
+ if (comm_ptr->is_hetero)
+ MPIU_Assert_fmt_msg(FALSE,("heterogeneous support for Allreduce_group_intra not yet implemented"));
+#endif
- *root = 0;
- *up = (me == 0) ? MPI_PROC_NULL : (me - 1) / 2;
+ /* homogeneous case */
- *left = 2*me + 1;
- if (*left >= nproc) *left = MPI_PROC_NULL;
+ comm_size = comm_ptr->local_size;
+ comm_rank = comm_ptr->rank;
+ group_rank = group_ptr->rank;
+ MPIU_ERR_CHKANDJUMP(group_rank == MPI_UNDEFINED, mpi_errno, MPI_ERR_OTHER, "**rank");
- *right = 2*me + 2;
- if (*right >= nproc) *right = MPI_PROC_NULL;
+ is_commutative = MPIR_Op_is_commutative(op);
- ranks_in[0] = *root;
- ranks_in[1] = *up;
- ranks_in[2] = *left;
- ranks_in[3] = *right;
+ /* need to allocate temporary buffer to store incoming data*/
+ MPIR_Type_get_true_extent_impl(datatype, &true_lb, &true_extent);
+ MPID_Datatype_get_extent_macro(datatype, extent);
- MPIR_Group_translate_ranks_impl(group_ptr, 4, ranks_in, comm_ptr->local_group, ranks_out);
+ MPID_Ensure_Aint_fits_in_pointer(count * MPIR_MAX(extent, true_extent));
+ MPIU_CHKLMEM_MALLOC(tmp_buf, void *, count*(MPIR_MAX(extent,true_extent)), mpi_errno, "temporary buffer");
- *root = ranks_out[0];
- *up = ranks_out[1];
- *left = ranks_out[2];
- *right= ranks_out[3];
-}
+ /* adjust for potential negative lower bound in datatype */
+ tmp_buf = (void *)((char*)tmp_buf - true_lb);
+ /* copy local data into recvbuf */
+ if (sendbuf != MPI_IN_PLACE) {
+ mpi_errno = MPIR_Localcopy(sendbuf, count, datatype, recvbuf,
+ count, datatype);
+ if (mpi_errno) MPIU_ERR_POP(mpi_errno);
+ }
+ MPID_Datatype_get_size_macro(datatype, type_size);
-/* Perform group-collective allreduce (binary tree reduce-broadcast algorithm).
- *
- * [in] sendbuf Input buffer
- * [out] recvbuf Output buffer
- * [in] count Number of data elements
- * [in] datatype Must be MPI_INT
- * [in] op One of: MPI_BAND, MPI_MAX
- * [in] comm_ptr Parent communicator
- * [in] group_ptr Group of processes that will participate in the allreduce
- * (must be a subgroup of comm_ptr's group)
- * [in] tag Tag to be used for allreduce messages
- * [out] errflag Error flag
- */
-int MPIR_Allreduce_group(void *sendbuf, void *recvbuf, int count,
- MPI_Datatype datatype, MPI_Op op, MPID_Comm *comm_ptr,
- MPID_Group *group_ptr, int tag, int *errflag)
-{
+ /* find nearest power-of-two less than or equal to comm_size */
+ pof2 = 1;
+ while (pof2 <= group_size) pof2 <<= 1;
+ pof2 >>=1;
- MPI_Status status;
- int i, bin_root, bin_parent, bin_lchild, bin_rchild;
- int *buf_left, *buf_right, *in_buf, *out_buf;
+ rem = group_size - pof2;
- MPIU_Assert(datatype == MPI_INT);
+ /* In the non-power-of-two case, all even-numbered
+ processes of rank < 2*rem send their data to
+ (rank+1). These even-numbered processes no longer
+ participate in the algorithm until the very end. The
+ remaining processes form a nice power-of-two. */
- buf_left = (int*) MPIU_Malloc(sizeof(int)*count);
- buf_right = (int*) MPIU_Malloc(sizeof(int)*count);
- out_buf = (int*) recvbuf;
- in_buf = (int*) ((sendbuf == MPI_IN_PLACE) ? recvbuf : sendbuf);
+ if (group_rank < 2*rem) {
+ if (group_rank % 2 == 0) { /* even */
+ to_comm_rank(cdst, group_rank+1);
+ mpi_errno = MPIC_Send_ft(recvbuf, count,
+ datatype, cdst,
+ MPIR_ALLREDUCE_TAG, comm, errflag);
+ if (mpi_errno) {
+ /* for communication errors, just record the error but continue */
+ *errflag = TRUE;
+ MPIU_ERR_SET(mpi_errno, MPI_ERR_OTHER, "**fail");
+ MPIU_ERR_ADD(mpi_errno_ret, mpi_errno);
+ }
- /* Map a binary spanning tree on members of group in comm. Resulting
- * ranks will be MPI_PROC_NULL if, e.g., a left child does not exist, so
- * it's fine to unconditionally send/recv since these will be no-ops. */
- MPIR_Allreduce_group_bintree(comm_ptr, group_ptr, &bin_root, &bin_parent, &bin_lchild, &bin_rchild);
+ /* temporarily set the rank to -1 so that this
+ process does not pariticipate in recursive
+ doubling */
+ newrank = -1;
+ }
+ else { /* odd */
+ to_comm_rank(csrc, group_rank-1);
+ mpi_errno = MPIC_Recv_ft(tmp_buf, count,
+ datatype, csrc,
+ MPIR_ALLREDUCE_TAG, comm,
+ MPI_STATUS_IGNORE, errflag);
+ if (mpi_errno) {
+ /* for communication errors, just record the error but continue */
+ *errflag = TRUE;
+ MPIU_ERR_SET(mpi_errno, MPI_ERR_OTHER, "**fail");
+ MPIU_ERR_ADD(mpi_errno_ret, mpi_errno);
+ }
- /** REDUCE **/
+ /* do the reduction on received data. since the
+ ordering is right, it doesn't matter whether
+ the operation is commutative or not. */
+ mpi_errno = MPIR_Reduce_local_impl(tmp_buf, recvbuf, count, datatype, op);
+ if (mpi_errno) MPIU_ERR_POP(mpi_errno);
- /* Receive from left child */
- MPIC_Recv(buf_left, count, datatype, bin_lchild, tag, comm_ptr->handle, &status);
+ /* change the rank */
+ newrank = group_rank / 2;
+ }
+ }
+ else /* rank >= 2*rem */
+ newrank = group_rank - rem;
- /* Receive from right child */
- MPIC_Recv(buf_right, count, datatype, bin_rchild, tag, comm_ptr->handle, &status);
+ /* If op is user-defined or count is less than pof2, use
+ recursive doubling algorithm. Otherwise do a reduce-scatter
+ followed by allgather. (If op is user-defined,
+ derived datatypes are allowed and the user could pass basic
+ datatypes on one process and derived on another as long as
+ the type maps are the same. Breaking up derived
+ datatypes to do the reduce-scatter is tricky, therefore
+ using recursive doubling in that case.) */
- /* Initialize local reduction output to input values */
- if (sendbuf != MPI_IN_PLACE)
- for (i = 0; i < count; i++)
- out_buf[i] = in_buf[i];
+ if (newrank != -1) {
+ if ((count*type_size <= MPIR_PARAM_ALLREDUCE_SHORT_MSG_SIZE) ||
+ (HANDLE_GET_KIND(op) != HANDLE_KIND_BUILTIN) ||
+ (count < pof2))
+ {
+ /* use recursive doubling */
+ mask = 0x1;
+ while (mask < pof2) {
+ newdst = newrank ^ mask;
+ /* find real rank of dest */
+ dst = (newdst < rem) ? newdst*2 + 1 : newdst + rem;
+ to_comm_rank(cdst, dst);
- /* Perform local reduction on left-, and right-child values if they exist */
- if (op == MPI_BAND) {
- if (bin_lchild != MPI_PROC_NULL) {
- for (i = 0; i < count; i++)
- out_buf[i] = out_buf[i] & buf_left[i];
+ /* Send the most current data, which is in recvbuf. Recv
+ into tmp_buf */
+ mpi_errno = MPIC_Sendrecv_ft(recvbuf, count, datatype,
+ cdst, MPIR_ALLREDUCE_TAG, tmp_buf,
+ count, datatype, cdst,
+ MPIR_ALLREDUCE_TAG, comm,
+ MPI_STATUS_IGNORE, errflag);
+ if (mpi_errno) {
+ /* for communication errors, just record the error but continue */
+ *errflag = TRUE;
+ MPIU_ERR_SET(mpi_errno, MPI_ERR_OTHER, "**fail");
+ MPIU_ERR_ADD(mpi_errno_ret, mpi_errno);
+ }
+
+ /* tmp_buf contains data received in this step.
+ recvbuf contains data accumulated so far */
+
+ if (is_commutative || (dst < group_rank)) {
+ /* op is commutative OR the order is already right */
+ mpi_errno = MPIR_Reduce_local_impl(tmp_buf, recvbuf, count, datatype, op);
+ if (mpi_errno) MPIU_ERR_POP(mpi_errno);
+ }
+ else {
+ /* op is noncommutative and the order is not right */
+ mpi_errno = MPIR_Reduce_local_impl(recvbuf, tmp_buf, count, datatype, op);
+ if (mpi_errno) MPIU_ERR_POP(mpi_errno);
+
+ /* copy result back into recvbuf */
+ mpi_errno = MPIR_Localcopy(tmp_buf, count, datatype,
+ recvbuf, count, datatype);
+ if (mpi_errno) MPIU_ERR_POP(mpi_errno);
+ }
+ mask <<= 1;
+ }
}
- if (bin_rchild != MPI_PROC_NULL) {
- for (i = 0; i < count; i++)
- out_buf[i] = out_buf[i] & buf_right[i];
+ else {
+
+ /* do a reduce-scatter followed by allgather */
+
+ /* for the reduce-scatter, calculate the count that
+ each process receives and the displacement within
+ the buffer */
+
+ MPIU_CHKLMEM_MALLOC(cnts, int *, pof2*sizeof(int), mpi_errno, "counts");
+ MPIU_CHKLMEM_MALLOC(disps, int *, pof2*sizeof(int), mpi_errno, "displacements");
+
+ for (i=0; i<(pof2-1); i++)
+ cnts[i] = count/pof2;
+ cnts[pof2-1] = count - (count/pof2)*(pof2-1);
+
+ disps[0] = 0;
+ for (i=1; i<pof2; i++)
+ disps[i] = disps[i-1] + cnts[i-1];
+
+ mask = 0x1;
+ send_idx = recv_idx = 0;
+ last_idx = pof2;
+ while (mask < pof2) {
+ newdst = newrank ^ mask;
+ /* find real rank of dest */
+ dst = (newdst < rem) ? newdst*2 + 1 : newdst + rem;
+ to_comm_rank(cdst, dst);
+
+ send_cnt = recv_cnt = 0;
+ if (newrank < newdst) {
+ send_idx = recv_idx + pof2/(mask*2);
+ for (i=send_idx; i<last_idx; i++)
+ send_cnt += cnts[i];
+ for (i=recv_idx; i<send_idx; i++)
+ recv_cnt += cnts[i];
+ }
+ else {
+ recv_idx = send_idx + pof2/(mask*2);
+ for (i=send_idx; i<recv_idx; i++)
+ send_cnt += cnts[i];
+ for (i=recv_idx; i<last_idx; i++)
+ recv_cnt += cnts[i];
+ }
+
+ /* Send data from recvbuf. Recv into tmp_buf */
+ mpi_errno = MPIC_Sendrecv_ft((char *) recvbuf +
+ disps[send_idx]*extent,
+ send_cnt, datatype,
+ cdst, MPIR_ALLREDUCE_TAG,
+ (char *) tmp_buf +
+ disps[recv_idx]*extent,
+ recv_cnt, datatype, cdst,
+ MPIR_ALLREDUCE_TAG, comm,
+ MPI_STATUS_IGNORE, errflag);
+ if (mpi_errno) {
+ /* for communication errors, just record the error but continue */
+ *errflag = TRUE;
+ MPIU_ERR_SET(mpi_errno, MPI_ERR_OTHER, "**fail");
+ MPIU_ERR_ADD(mpi_errno_ret, mpi_errno);
+ }
+
+ /* tmp_buf contains data received in this step.
+ recvbuf contains data accumulated so far */
+
+ /* This algorithm is used only for predefined ops
+ and predefined ops are always commutative. */
+ mpi_errno = MPIR_Reduce_local_impl(((char *) tmp_buf + disps[recv_idx]*extent),
+ ((char *) recvbuf + disps[recv_idx]*extent),
+ recv_cnt, datatype, op);
+ if (mpi_errno) MPIU_ERR_POP(mpi_errno);
+
+ /* update send_idx for next iteration */
+ send_idx = recv_idx;
+ mask <<= 1;
+
+ /* update last_idx, but not in last iteration
+ because the value is needed in the allgather
+ step below. */
+ if (mask < pof2)
+ last_idx = recv_idx + pof2/mask;
+ }
+
+ /* now do the allgather */
+
+ mask >>= 1;
+ while (mask > 0) {
+ newdst = newrank ^ mask;
+ /* find real rank of dest */
+ dst = (newdst < rem) ? newdst*2 + 1 : newdst + rem;
+ to_comm_rank(cdst, dst);
+
+ send_cnt = recv_cnt = 0;
+ if (newrank < newdst) {
+ /* update last_idx except on first iteration */
+ if (mask != pof2/2)
+ last_idx = last_idx + pof2/(mask*2);
+
+ recv_idx = send_idx + pof2/(mask*2);
+ for (i=send_idx; i<recv_idx; i++)
+ send_cnt += cnts[i];
+ for (i=recv_idx; i<last_idx; i++)
+ recv_cnt += cnts[i];
+ }
+ else {
+ recv_idx = send_idx - pof2/(mask*2);
+ for (i=send_idx; i<last_idx; i++)
+ send_cnt += cnts[i];
+ for (i=recv_idx; i<send_idx; i++)
+ recv_cnt += cnts[i];
+ }
+
+ mpi_errno = MPIC_Sendrecv_ft((char *) recvbuf +
+ disps[send_idx]*extent,
+ send_cnt, datatype,
+ cdst, MPIR_ALLREDUCE_TAG,
+ (char *) recvbuf +
+ disps[recv_idx]*extent,
+ recv_cnt, datatype, cdst,
+ MPIR_ALLREDUCE_TAG, comm,
+ MPI_STATUS_IGNORE, errflag);
+ if (mpi_errno) {
+ /* for communication errors, just record the error but continue */
+ *errflag = TRUE;
+ MPIU_ERR_SET(mpi_errno, MPI_ERR_OTHER, "**fail");
+ MPIU_ERR_ADD(mpi_errno_ret, mpi_errno);
+ }
+
+ if (newrank > newdst) send_idx = recv_idx;
+
+ mask >>= 1;
+ }
}
}
- else if (op == MPI_MAX) {
- if (bin_lchild != MPI_PROC_NULL) {
- for (i = 0; i < count; i++)
- out_buf[i] = (out_buf[i] > buf_left[i]) ? out_buf[i] : buf_left[i];
+
+ /* In the non-power-of-two case, all odd-numbered
+ processes of rank < 2*rem send the result to
+ (rank-1), the ranks who didn't participate above. */
+ if (group_rank < 2*rem) {
+ if (group_rank % 2) { /* odd */
+ to_comm_rank(cdst, group_rank-1);
+ mpi_errno = MPIC_Send_ft(recvbuf, count,
+ datatype, cdst,
+ MPIR_ALLREDUCE_TAG, comm, errflag);
}
-
- if (bin_rchild != MPI_PROC_NULL) {
- for (i = 0; i < count; i++)
- out_buf[i] = (out_buf[i] > buf_right[i]) ? out_buf[i] : buf_right[i];
+ else { /* even */
+ to_comm_rank(csrc, group_rank+1);
+ mpi_errno = MPIC_Recv_ft(recvbuf, count,
+ datatype, csrc,
+ MPIR_ALLREDUCE_TAG, comm,
+ MPI_STATUS_IGNORE, errflag);
+ }
+ if (mpi_errno) {
+ /* for communication errors, just record the error but continue */
+ *errflag = TRUE;
+ MPIU_ERR_SET(mpi_errno, MPI_ERR_OTHER, "**fail");
+ MPIU_ERR_ADD(mpi_errno_ret, mpi_errno);
}
- }
- else {
- MPIU_Assert(FALSE);
}
- /* Send to parent */
- MPIC_Send(recvbuf, count, datatype, bin_parent, tag, comm_ptr->handle);
+ fn_exit:
+ MPIU_CHKLMEM_FREEALL();
+ if (mpi_errno_ret)
+ mpi_errno = mpi_errno_ret;
+ else if (*errflag)
+ MPIU_ERR_SET(mpi_errno, MPI_ERR_OTHER, "**coll_fail");
+ return (mpi_errno);
- /** BROADCAST **/
-
- /* Receive from parent */
- MPIC_Recv(recvbuf, count, datatype, bin_parent, tag, comm_ptr->handle, &status);
+ fn_fail:
+ goto fn_exit;
+}
- /* Send to left child */
- MPIC_Send(recvbuf, count, datatype, bin_lchild, tag, comm_ptr->handle);
+#undef FUNCNAME
+#define FUNCNAME MPIR_Allreduce_group
+#undef FCNAME
+#define FCNAME MPIU_QUOTE(FUNCNAME)
+int MPIR_Allreduce_group(void *sendbuf, void *recvbuf, int count,
+ MPI_Datatype datatype, MPI_Op op, MPID_Comm *comm_ptr,
+ MPID_Group *group_ptr, int tag, int *errflag)
+{
+ int mpi_errno = MPI_SUCCESS;
- /* Send to right child */
- MPIC_Send(recvbuf, count, datatype, bin_rchild, tag, comm_ptr->handle);
+ MPIU_ERR_CHKANDJUMP(comm_ptr->comm_kind != MPID_INTRACOMM, mpi_errno, MPI_ERR_OTHER, "**commnotintra");
- MPIU_Free(buf_left);
- MPIU_Free(buf_right);
+ mpi_errno = MPIR_Allreduce_group_intra(sendbuf, recvbuf, count, datatype,
+ op, comm_ptr, group_ptr, tag, errflag);
+ if (mpi_errno) MPIU_ERR_POP(mpi_errno);
- return MPI_SUCCESS;
+ fn_exit:
+ return mpi_errno;
+ fn_fail:
+ goto fn_exit;
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment