diff --git a/ompi/mca/coll/libnbc/coll_libnbc.h b/ompi/mca/coll/libnbc/coll_libnbc.h index ac944321fc..f61229d2cc 100644 --- a/ompi/mca/coll/libnbc/coll_libnbc.h +++ b/ompi/mca/coll/libnbc/coll_libnbc.h @@ -74,6 +74,7 @@ extern int libnbc_iallgather_algorithm; extern int libnbc_ibcast_algorithm; extern int libnbc_ibcast_knomial_radix; extern int libnbc_iexscan_algorithm; +extern int libnbc_ireduce_algorithm; extern int libnbc_iscan_algorithm; struct ompi_coll_libnbc_component_t { diff --git a/ompi/mca/coll/libnbc/coll_libnbc_component.c b/ompi/mca/coll/libnbc/coll_libnbc_component.c index ef67918027..6b859b5874 100644 --- a/ompi/mca/coll/libnbc/coll_libnbc_component.c +++ b/ompi/mca/coll/libnbc/coll_libnbc_component.c @@ -73,6 +73,15 @@ static mca_base_var_enum_value_t iexscan_algorithms[] = { {0, NULL} }; +int libnbc_ireduce_algorithm = 0; /* ireduce user forced algorithm */ +static mca_base_var_enum_value_t ireduce_algorithms[] = { + {0, "ignore"}, + {1, "chain"}, + {2, "binomial"}, + {3, "rabenseifner"}, + {0, NULL} +}; + int libnbc_iscan_algorithm = 0; /* iscan user forced algorithm */ static mca_base_var_enum_value_t iscan_algorithms[] = { {0, "ignore"}, @@ -232,6 +241,16 @@ libnbc_register(void) &libnbc_iexscan_algorithm); OBJ_RELEASE(new_enum); + libnbc_ireduce_algorithm = 0; + (void) mca_base_var_enum_create("coll_libnbc_ireduce_algorithms", ireduce_algorithms, &new_enum); + mca_base_component_var_register(&mca_coll_libnbc_component.super.collm_version, + "ireduce_algorithm", + "Which ireduce algorithm is used: 0 ignore, 1 chain, 2 binomial, 3 rabenseifner", + MCA_BASE_VAR_TYPE_INT, new_enum, 0, MCA_BASE_VAR_FLAG_SETTABLE, + OPAL_INFO_LVL_5, MCA_BASE_VAR_SCOPE_ALL, + &libnbc_ireduce_algorithm); + OBJ_RELEASE(new_enum); + libnbc_iscan_algorithm = 0; (void) mca_base_var_enum_create("coll_libnbc_iscan_algorithms", iscan_algorithms, &new_enum); mca_base_component_var_register(&mca_coll_libnbc_component.super.collm_version, diff --git a/ompi/mca/coll/libnbc/nbc_ireduce.c b/ompi/mca/coll/libnbc/nbc_ireduce.c index c222fa3a7f..52521a0217 100644 --- a/ompi/mca/coll/libnbc/nbc_ireduce.c +++ b/ompi/mca/coll/libnbc/nbc_ireduce.c @@ -20,6 +20,7 @@ */ #include "opal/include/opal/align.h" +#include "opal/util/bit_ops.h" #include "ompi/op/op.h" #include "nbc_internal.h" @@ -31,6 +32,10 @@ static inline int red_sched_chain (int rank, int p, int root, const void *sendbu static inline int red_sched_linear (int rank, int rsize, int root, const void *sendbuf, void *recvbuf, void *tmpbuf, int count, MPI_Datatype datatype, MPI_Op op, NBC_Schedule *schedule); +static inline int red_sched_redscat_gather( + int rank, int comm_size, int root, const void *sbuf, void *rbuf, + char tmpredbuf, int count, MPI_Datatype datatype, MPI_Op op, char inplace, + NBC_Schedule *schedule, void *tmp_buf, struct ompi_communicator_t *comm); #ifdef NBC_CACHE_SCHEDULE /* tree comparison function for schedule cache */ @@ -63,7 +68,7 @@ static int nbc_reduce_init(const void* sendbuf, void* recvbuf, int count, MPI_Da char *redbuf=NULL, inplace; void *tmpbuf; char tmpredbuf = 0; - enum { NBC_RED_BINOMIAL, NBC_RED_CHAIN } alg; + enum { NBC_RED_BINOMIAL, NBC_RED_CHAIN, NBC_RED_REDSCAT_GATHER} alg; ompi_coll_libnbc_module_t *libnbc_module = (ompi_coll_libnbc_module_t*) module; ptrdiff_t span, gap; @@ -98,22 +103,42 @@ static int nbc_reduce_init(const void* sendbuf, void* recvbuf, int count, MPI_Da span = opal_datatype_span(&datatype->super, count, &gap); /* algorithm selection */ - if (p > 4 || size * count < 65536 || !ompi_op_is_commute(op)) { - alg = NBC_RED_BINOMIAL; - if(rank == root) { - /* root reduces in receivebuffer */ - tmpbuf = malloc (span); + int nprocs_pof2 = opal_next_poweroftwo(p) >> 1; + if (libnbc_ireduce_algorithm == 0) { + if (ompi_op_is_commute(op) && p > 2 && count >= nprocs_pof2) { + alg = NBC_RED_REDSCAT_GATHER; + } else if (p > 4 || size * count < 65536 || !ompi_op_is_commute(op)) { + alg = NBC_RED_BINOMIAL; + } else { + alg = NBC_RED_CHAIN; + } + } else { + if (libnbc_ireduce_algorithm == 1) { + alg = NBC_RED_CHAIN; + } else if (libnbc_ireduce_algorithm == 2) { + alg = NBC_RED_BINOMIAL; + } else if (libnbc_ireduce_algorithm == 3 && ompi_op_is_commute(op) && p > 2 && count >= nprocs_pof2) { + alg = NBC_RED_REDSCAT_GATHER; + } else { + alg = NBC_RED_CHAIN; + } + } + + /* allocate temporary buffers */ + if (alg == NBC_RED_REDSCAT_GATHER || alg == NBC_RED_BINOMIAL) { + if (rank == root) { + /* root reduces in receive buffer */ + tmpbuf = malloc(span); redbuf = recvbuf; } else { /* recvbuf may not be valid on non-root nodes */ ptrdiff_t span_align = OPAL_ALIGN(span, datatype->super.align, ptrdiff_t); - tmpbuf = malloc (span_align + span); - redbuf = (char*)span_align - gap; + tmpbuf = malloc(span_align + span); + redbuf = (char *)span_align - gap; tmpredbuf = 1; } } else { tmpbuf = malloc (span); - alg = NBC_RED_CHAIN; segsize = 16384/2; } @@ -151,6 +176,9 @@ static int nbc_reduce_init(const void* sendbuf, void* recvbuf, int count, MPI_Da case NBC_RED_CHAIN: res = red_sched_chain(rank, p, root, sendbuf, recvbuf, count, datatype, op, ext, size, schedule, tmpbuf, segsize); break; + case NBC_RED_REDSCAT_GATHER: + res = red_sched_redscat_gather(rank, p, root, sendbuf, redbuf, tmpredbuf, count, datatype, op, inplace, schedule, tmpbuf, comm); + break; } } @@ -560,6 +588,350 @@ static inline int red_sched_linear (int rank, int rsize, int root, const void *s return OMPI_SUCCESS; } +/* + * red_sched_redscat_gather: + * + * Description: an implementation of Rabenseifner's Reduce algorithm [1, 2]. + * [1] Rajeev Thakur, Rolf Rabenseifner and William Gropp. + * Optimization of Collective Communication Operations in MPICH // + * The Int. Journal of High Performance Computing Applications. Vol 19, + * Issue 1, pp. 49--66. + * [2] http://www.hlrs.de/mpi/myreduce.html. + * + * This algorithm is a combination of a reduce-scatter implemented with + * recursive vector halving and recursive distance doubling, followed either + * by a binomial tree gather. + * + * Step 1. If the number of processes is not a power of two, reduce it to + * the nearest lower power of two (p' = 2^{\floor{\log_2 p}}) + * by removing r = p - p' extra processes as follows. In the first 2r processes + * (ranks 0 to 2r - 1), all the even ranks send the second half of the input + * vector to their right neighbor (rank + 1), and all the odd ranks send + * the first half of the input vector to their left neighbor (rank - 1). + * The even ranks compute the reduction on the first half of the vector and + * the odd ranks compute the reduction on the second half. The odd ranks then + * send the result to their left neighbors (the even ranks). As a result, + * the even ranks among the first 2r processes now contain the reduction with + * the input vector on their right neighbors (the odd ranks). These odd ranks + * do not participate in the rest of the algorithm, which leaves behind + * a power-of-two number of processes. The first r even-ranked processes and + * the last p - 2r processes are now renumbered from 0 to p' - 1. + * + * Step 2. The remaining processes now perform a reduce-scatter by using + * recursive vector halving and recursive distance doubling. The even-ranked + * processes send the second half of their buffer to rank + 1 and the odd-ranked + * processes send the first half of their buffer to rank - 1. All processes + * then compute the reduction between the local buffer and the received buffer. + * In the next log_2(p') - 1 steps, the buffers are recursively halved, and the + * distance is doubled. At the end, each of the p' processes has 1 / p' of the + * total reduction result. + * + * Step 3. A binomial tree gather is performed by using recursive vector + * doubling and distance halving. In the non-power-of-two case, if the root + * happens to be one of those odd-ranked processes that would normally + * be removed in the first step, then the role of this process and process 0 + * are interchanged. + * + * Limitations: + * count >= 2^{\floor{\log_2 p}} + * commutative operations only + * intra-communicators only + * + * Memory requirements (per process): + * rank != root: 2 * count * typesize + 4 * \log_2(p) * sizeof(int) = O(count) + * rank == root: count * typesize + 4 * \log_2(p) * sizeof(int) = O(count) + * + * Schedule length (rounds): O(\log(p)) + * Recommendations: root = 0, otherwise it is required additional steps + * in the root process. + */ +static inline int red_sched_redscat_gather( + int rank, int comm_size, int root, const void *sbuf, void *rbuf, + char tmpredbuf, int count, MPI_Datatype datatype, MPI_Op op, char inplace, + NBC_Schedule *schedule, void *tmp_buf, struct ompi_communicator_t *comm) +{ + int res = OMPI_SUCCESS; + int *rindex = NULL, *rcount = NULL, *sindex = NULL, *scount = NULL; + + /* Find nearest power-of-two less than or equal to comm_size */ + int nsteps = opal_hibit(comm_size, comm->c_cube_dim + 1); /* ilog2(comm_size) */ + int nprocs_pof2 = 1 << nsteps; /* flp2(comm_size) */ + + ptrdiff_t lb, extent; + ompi_datatype_get_extent(datatype, &lb, &extent); + + if ((rank != root) || !inplace) { + res = NBC_Sched_copy((char *)sbuf, false, count, datatype, + rbuf, tmpredbuf, count, datatype, schedule, true); + if (OPAL_UNLIKELY(OMPI_SUCCESS != res)) { goto cleanup_and_return; } + } + + /* + * Step 1. Reduce the number of processes to the nearest lower power of two + * p' = 2^{\floor{\log_2 p}} by removing r = p - p' processes. + * 1. In the first 2r processes (ranks 0 to 2r - 1), all the even ranks send + * the second half of the input vector to their right neighbor (rank + 1) + * and all the odd ranks send the first half of the input vector to their + * left neighbor (rank - 1). + * 2. All 2r processes compute the reduction on their half. + * 3. The odd ranks then send the result to their left neighbors + * (the even ranks). + * + * The even ranks (0 to 2r - 1) now contain the reduction with the input + * vector on their right neighbors (the odd ranks). The first r even + * processes and the p - 2r last processes are renumbered from + * 0 to 2^{\floor{\log_2 p}} - 1. These odd ranks do not participate in the + * rest of the algorithm. + */ + + int vrank, step, wsize; + int nprocs_rem = comm_size - nprocs_pof2; + + if (rank < 2 * nprocs_rem) { + int count_lhalf = count / 2; + int count_rhalf = count - count_lhalf; + + if (rank % 2 != 0) { + /* + * Odd process -- exchange with rank - 1 + * Send the left half of the input vector to the left neighbor, + * Recv the right half of the input vector from the left neighbor + */ + res = NBC_Sched_send(rbuf, tmpredbuf, count_lhalf, datatype, rank - 1, + schedule, false); + if (OPAL_UNLIKELY(OMPI_SUCCESS != res)) { goto cleanup_and_return; } + + res = NBC_Sched_recv((char *)tmp_buf + (ptrdiff_t)count_lhalf * extent, + false, count_rhalf, datatype, rank - 1, schedule, true); + if (OPAL_UNLIKELY(OMPI_SUCCESS != res)) { goto cleanup_and_return; } + + res = NBC_Sched_op((char *)tmp_buf + (ptrdiff_t)count_lhalf * extent, + false, (char *)rbuf + (ptrdiff_t)count_lhalf * extent, + tmpredbuf, count_rhalf, datatype, op, schedule, true); + if (OPAL_UNLIKELY(OMPI_SUCCESS != res)) { goto cleanup_and_return; } + + /* Send the right half to the left neighbor */ + res = NBC_Sched_send((char *)rbuf + (ptrdiff_t)count_lhalf * extent, + tmpredbuf, count_rhalf, datatype, rank - 1, schedule, true); + if (OPAL_UNLIKELY(OMPI_SUCCESS != res)) { goto cleanup_and_return; } + + /* This process does not participate in recursive doubling phase */ + vrank = -1; + + } else { + /* + * Even process -- exchange with rank + 1 + * Send the right half of the input vector to the right neighbor, + * Recv the left half of the input vector from the right neighbor + */ + res = NBC_Sched_send((char *)rbuf + (ptrdiff_t)count_lhalf * extent, + tmpredbuf, count_rhalf, datatype, rank + 1, schedule, false); + if (OPAL_UNLIKELY(OMPI_SUCCESS != res)) { goto cleanup_and_return; } + + res = NBC_Sched_recv((char *)tmp_buf, false, count_lhalf, datatype, rank + 1, + schedule, true); + if (OPAL_UNLIKELY(OMPI_SUCCESS != res)) { goto cleanup_and_return; } + + res = NBC_Sched_op(tmp_buf, false, rbuf, tmpredbuf, count_lhalf, + datatype, op, schedule, true); + if (OPAL_UNLIKELY(OMPI_SUCCESS != res)) { goto cleanup_and_return; } + + /* Recv the right half from the right neighbor */ + res = NBC_Sched_recv((char *)rbuf + (ptrdiff_t)count_lhalf * extent, + tmpredbuf, count_rhalf, datatype, rank + 1, schedule, true); + if (OPAL_UNLIKELY(OMPI_SUCCESS != res)) { goto cleanup_and_return; } + + vrank = rank / 2; + } + } else { /* rank >= 2 * nprocs_rem */ + vrank = rank - nprocs_rem; + } + + /* + * Step 2. Reduce-scatter implemented with recursive vector halving and + * recursive distance doubling. We have p' = 2^{\floor{\log_2 p}} + * power-of-two number of processes with new ranks (vrank) and result in rbuf. + * + * The even-ranked processes send the right half of their buffer to rank + 1 + * and the odd-ranked processes send the left half of their buffer to + * rank - 1. All processes then compute the reduction between the local + * buffer and the received buffer. In the next \log_2(p') - 1 steps, the + * buffers are recursively halved, and the distance is doubled. At the end, + * each of the p' processes has 1 / p' of the total reduction result. + */ + + rindex = malloc(sizeof(*rindex) * nsteps); /* O(\log_2(p)) */ + sindex = malloc(sizeof(*sindex) * nsteps); + rcount = malloc(sizeof(*rcount) * nsteps); + scount = malloc(sizeof(*scount) * nsteps); + if (NULL == rindex || NULL == sindex || NULL == rcount || NULL == scount) { + res = OMPI_ERR_OUT_OF_RESOURCE; + goto cleanup_and_return; + } + + if (vrank != -1) { + step = 0; + wsize = count; + sindex[0] = rindex[0] = 0; + + for (int mask = 1; mask < nprocs_pof2; mask <<= 1) { + /* + * On each iteration: rindex[step] = sindex[step] -- begining of the + * current window. Length of the current window is storded in wsize. + */ + int vdest = vrank ^ mask; + /* Translate vdest virtual rank to real rank */ + int dest = (vdest < nprocs_rem) ? vdest * 2 : vdest + nprocs_rem; + + if (rank < dest) { + /* + * Recv into the left half of the current window, send the right + * half of the window to the peer (perform reduce on the left + * half of the current window) + */ + rcount[step] = wsize / 2; + scount[step] = wsize - rcount[step]; + sindex[step] = rindex[step] + rcount[step]; + } else { + /* + * Recv into the right half of the current window, send the left + * half of the window to the peer (perform reduce on the right + * half of the current window) + */ + scount[step] = wsize / 2; + rcount[step] = wsize - scount[step]; + rindex[step] = sindex[step] + scount[step]; + } + + /* Send part of data from the rbuf, recv into the tmp_buf */ + res = NBC_Sched_send((char *)rbuf + (ptrdiff_t)sindex[step] * extent, + tmpredbuf, scount[step], datatype, dest, schedule, false); + if (OPAL_UNLIKELY(OMPI_SUCCESS != res)) { goto cleanup_and_return; } + res = NBC_Sched_recv((char *)tmp_buf + (ptrdiff_t)rindex[step] * extent, + false, rcount[step], datatype, dest, schedule, true); + if (OPAL_UNLIKELY(OMPI_SUCCESS != res)) { goto cleanup_and_return; } + + /* Local reduce: rbuf[] = tmp_buf[] rbuf[] */ + res = NBC_Sched_op((char *)tmp_buf + (ptrdiff_t)rindex[step] * extent, + false, (char *)rbuf + (ptrdiff_t)rindex[step] * extent, + tmpredbuf, rcount[step], datatype, op, schedule, true); + if (OPAL_UNLIKELY(OMPI_SUCCESS != res)) { goto cleanup_and_return; } + + /* Move the current window to the received message */ + if (step + 1 < nsteps) { + rindex[step + 1] = rindex[step]; + sindex[step + 1] = rindex[step]; + wsize = rcount[step]; + step++; + } + } + } + /* + * Assertion: each process has 1 / p' of the total reduction result: + * rcount[nsteps - 1] elements in the rbuf[rindex[nsteps - 1], ...]. + */ + + /* + * Setup the root process for gather operation. + * Case 1: root < 2r and root is odd -- root process was excluded on step 1 + * Recv data from process 0, vroot = 0, vrank = 0 + * Case 2: root < 2r and root is even: vroot = root / 2 + * Case 3: root >= 2r: vroot = root - r + */ + int vroot = 0; + if (root < 2 * nprocs_rem) { + if (root % 2 != 0) { + vroot = 0; + if (rank == root) { + /* + * Case 1: root < 2r and root is odd -- root process was + * excluded on step 1 (newrank == -1). + * Recv a data from the process 0. + */ + rindex[0] = 0; + step = 0, wsize = count; + for (int mask = 1; mask < nprocs_pof2; mask *= 2) { + rcount[step] = wsize / 2; + scount[step] = wsize - rcount[step]; + rindex[step] = 0; + sindex[step] = rcount[step]; + step++; + wsize /= 2; + } + + res = NBC_Sched_recv(rbuf, tmpredbuf, rcount[nsteps - 1], datatype, + 0, schedule, true); + if (OPAL_UNLIKELY(OMPI_SUCCESS != res)) { goto cleanup_and_return; } + vrank = 0; + + } else if (vrank == 0) { + /* Send a data to the root */ + res = NBC_Sched_send(rbuf, tmpredbuf, rcount[nsteps - 1], datatype, + root, schedule, true); + if (OPAL_UNLIKELY(OMPI_SUCCESS != res)) { goto cleanup_and_return; } + vrank = -1; + } + } else { + /* Case 2: root < 2r and a root is even: vroot = root / 2 */ + vroot = root / 2; + } + } else { + /* Case 3: root >= 2r: newroot = root - r */ + vroot = root - nprocs_rem; + } + + /* + * Step 3. Gather result at the vroot by the binomial tree algorithm. + * Each process has 1 / p' of the total reduction result: + * rcount[nsteps - 1] elements in the rbuf[rindex[nsteps - 1], ...]. + * All exchanges are executed in reverse order relative + * to recursive doubling (previous step). + */ + + if (vrank != -1) { + int vdest_tree, vroot_tree; + step = nsteps - 1; /* step = ilog2(p') - 1 */ + + for (int mask = nprocs_pof2 >> 1; mask > 0; mask >>= 1) { + int vdest = vrank ^ mask; + /* Translate vdest virtual rank to real rank */ + int dest = (vdest < nprocs_rem) ? vdest * 2 : vdest + nprocs_rem; + if ((vdest == 0) && (root < 2 * nprocs_rem) && (root % 2 != 0)) + dest = root; + + vdest_tree = vdest >> step; + vdest_tree <<= step; + vroot_tree = vroot >> step; + vroot_tree <<= step; + if (vdest_tree == vroot_tree) { + /* Send data from rbuf and exit */ + + res = NBC_Sched_send((char *)rbuf + (ptrdiff_t)rindex[step] * extent, + tmpredbuf, rcount[step], datatype, dest, schedule, false); + if (OPAL_UNLIKELY(OMPI_SUCCESS != res)) { goto cleanup_and_return; } + break; + } else { + /* Recv and continue */ + res = NBC_Sched_recv((char *)rbuf + (ptrdiff_t)sindex[step] * extent, + tmpredbuf, scount[step], datatype, dest, schedule, true); + if (OPAL_UNLIKELY(OMPI_SUCCESS != res)) { goto cleanup_and_return; } + } + step--; + } + } + + cleanup_and_return: + if (NULL != rindex) + free(rindex); + if (NULL != sindex) + free(sindex); + if (NULL != rcount) + free(rcount); + if (NULL != scount) + free(scount); + return res; +} + int ompi_coll_libnbc_reduce_init(const void* sendbuf, void* recvbuf, int count, MPI_Datatype datatype, MPI_Op op, int root, struct ompi_communicator_t *comm, MPI_Info info, ompi_request_t ** request, struct mca_coll_base_module_2_3_0_t *module) {