X-Git-Url: http://bilbo.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/08e7455d67920bbd7a87f440d00f2c1e071314a0..db41dc7d999a5792aca42198b5bc87edb2dbfcce:/src/smpi/colls/reduce_scatter/reduce_scatter-ompi.cpp diff --git a/src/smpi/colls/reduce_scatter/reduce_scatter-ompi.cpp b/src/smpi/colls/reduce_scatter/reduce_scatter-ompi.cpp index 3fe4ae1e7d..a2d3687682 100644 --- a/src/smpi/colls/reduce_scatter/reduce_scatter-ompi.cpp +++ b/src/smpi/colls/reduce_scatter/reduce_scatter-ompi.cpp @@ -1,4 +1,4 @@ -/* Copyright (c) 2013-2021. The SimGrid Team. +/* Copyright (c) 2013-2023. The SimGrid Team. * All rights reserved. */ /* This program is free software; you can redistribute it and/or modify it @@ -41,8 +41,7 @@ * Returns: - MPI_SUCCESS or error code * Limitation: - Works only for commutative operations. */ -namespace simgrid { -namespace smpi { +namespace simgrid::smpi { int reduce_scatter__ompi_basic_recursivehalving(const void *sbuf, void *rbuf, const int *rcounts, @@ -526,6 +525,281 @@ int reduce_scatter__ompi_ring(const void *sbuf, void *rbuf, const int *rcounts, smpi_free_tmp_buffer(inbuf_free[1]); return ret; } + +static int ompi_sum_counts(const int *counts, int *displs, int nprocs_rem, int lo, int hi) +{ + /* Adjust lo and hi for taking into account blocks of excluded processes */ + lo = (lo < nprocs_rem) ? lo * 2 : lo + nprocs_rem; + hi = (hi < nprocs_rem) ? hi * 2 + 1 : hi + nprocs_rem; + return displs[hi] + counts[hi] - displs[lo]; } + +/* + * ompi_mirror_perm: Returns mirror permutation of nbits low-order bits + * of x [*]. + * [*] Warren Jr., Henry S. Hacker's Delight (2ed). 2013. + * Chapter 7. Rearranging Bits and Bytes. + */ +static unsigned int ompi_mirror_perm(unsigned int x, int nbits) +{ + x = (((x & 0xaaaaaaaa) >> 1) | ((x & 0x55555555) << 1)); + x = (((x & 0xcccccccc) >> 2) | ((x & 0x33333333) << 2)); + x = (((x & 0xf0f0f0f0) >> 4) | ((x & 0x0f0f0f0f) << 4)); + x = (((x & 0xff00ff00) >> 8) | ((x & 0x00ff00ff) << 8)); + x = ((x >> 16) | (x << 16)); + return x >> (sizeof(x) * 8 - nbits); } +/* + * ompi_coll_base_reduce_scatter_intra_butterfly + * + * Function: Butterfly algorithm for reduce_scatter + * Accepts: Same as MPI_Reduce_scatter + * Returns: MPI_SUCCESS or error code + * + * Description: Implements butterfly algorithm for MPI_Reduce_scatter [*]. + * The algorithm can be used both by commutative and non-commutative + * operations, for power-of-two and non-power-of-two number of processes. + * + * [*] J.L. Traff. An improved Algorithm for (non-commutative) Reduce-scatter + * with an Application // Proc. of EuroPVM/MPI, 2005. -- pp. 129-137. + * + * Time complexity: O(m\lambda + log(p)\alpha + m\beta + m\gamma), + * where m = sum of rcounts[], p = comm_size + * Memory requirements (per process): 2 * m * typesize + comm_size + * + * Example: comm_size=6, nprocs_pof2=4, nprocs_rem=2, rcounts[]=1, sbuf=[0,1,...,5] + * Step 1. Reduce the number of processes to 4 + * rank 0: [0|1|2|3|4|5]: send to 1: vrank -1 + * rank 1: [0|1|2|3|4|5]: recv from 0, op: vrank 0: [0|2|4|6|8|10] + * rank 2: [0|1|2|3|4|5]: send to 3: vrank -1 + * rank 3: [0|1|2|3|4|5]: recv from 2, op: vrank 1: [0|2|4|6|8|10] + * rank 4: [0|1|2|3|4|5]: vrank 2: [0|1|2|3|4|5] + * rank 5: [0|1|2|3|4|5]: vrank 3: [0|1|2|3|4|5] + * + * Step 2. Butterfly. Buffer of 6 elements is divided into 4 blocks. + * Round 1 (mask=1, nblocks=2) + * 0: vrank -1 + * 1: vrank 0 [0 2|4 6|8|10]: exch with 1: send [2,3], recv [0,1]: [0 4|8 12|*|*] + * 2: vrank -1 + * 3: vrank 1 [0 2|4 6|8|10]: exch with 0: send [0,1], recv [2,3]: [**|**|16|20] + * 4: vrank 2 [0 1|2 3|4|5] : exch with 3: send [2,3], recv [0,1]: [0 2|4 6|*|*] + * 5: vrank 3 [0 1|2 3|4|5] : exch with 2: send [0,1], recv [2,3]: [**|**|8|10] + * + * Round 2 (mask=2, nblocks=1) + * 0: vrank -1 + * 1: vrank 0 [0 4|8 12|*|*]: exch with 2: send [1], recv [0]: [0 6|**|*|*] + * 2: vrank -1 + * 3: vrank 1 [**|**|16|20] : exch with 3: send [3], recv [2]: [**|**|24|*] + * 4: vrank 2 [0 2|4 6|*|*] : exch with 0: send [0], recv [1]: [**|12 18|*|*] + * 5: vrank 3 [**|**|8|10] : exch with 1: send [2], recv [3]: [**|**|*|30] + * + * Step 3. Exchange with remote process according to a mirror permutation: + * mperm(0)=0, mperm(1)=2, mperm(2)=1, mperm(3)=3 + * 0: vrank -1: recv "0" from process 0 + * 1: vrank 0 [0 6|**|*|*]: send "0" to 0, copy "6" to rbuf (mperm(0)=0) + * 2: vrank -1: recv result "12" from process 4 + * 3: vrank 1 [**|**|24|*] + * 4: vrank 2 [**|12 18|*|*]: send "12" to 2, send "18" to 3, recv "24" from 3 + * 5: vrank 3 [**|**|*|30]: copy "30" to rbuf (mperm(3)=3) + */ +int reduce_scatter__ompi_butterfly( + const void *sbuf, void *rbuf, const int *rcounts, MPI_Datatype dtype, + MPI_Op op, MPI_Comm comm) +{ + char *tmpbuf[2] = {NULL, NULL}, *psend, *precv; + int *displs = NULL, index; + ptrdiff_t span, gap, totalcount, extent; + int err = MPI_SUCCESS; + int comm_size = comm->size(); + int rank = comm->rank(); + int vrank = -1; + int nprocs_rem = 0; + + XBT_DEBUG("coll:base:reduce_scatter_intra_butterfly: rank %d/%d", + rank, comm_size); + if (comm_size < 2) + return MPI_SUCCESS; + + displs = (int*)malloc(sizeof(*displs) * comm_size); + if (NULL == displs) { + err = MPI_ERR_OTHER; + goto cleanup_and_return; + } + displs[0] = 0; + for (int i = 1; i < comm_size; i++) { + displs[i] = displs[i - 1] + rcounts[i - 1]; + } + totalcount = displs[comm_size - 1] + rcounts[comm_size - 1]; + dtype->extent(&gap, &extent); + span = extent * totalcount; + tmpbuf[0] = (char*)malloc(span); + tmpbuf[1] = (char*)malloc(span); + if (NULL == tmpbuf[0] || NULL == tmpbuf[1]) { + err = MPI_ERR_OTHER; + goto cleanup_and_return; + } + psend = tmpbuf[0] - gap; + precv = tmpbuf[1] - gap; + + if (sbuf != MPI_IN_PLACE) { + err = Datatype::copy(sbuf, totalcount, dtype, psend, totalcount, dtype); + if (MPI_SUCCESS != err) { goto cleanup_and_return; } + } else { + err = Datatype::copy(rbuf, totalcount, dtype, psend, totalcount, dtype); + if (MPI_SUCCESS != err) { 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. + * In the first 2r processes (ranks 0 to 2r - 1), all the even ranks send + * the input vector to their neighbor (rank + 1) and all the odd ranks recv + * the input vector and perform local reduction. + * The odd ranks (0 to 2r - 1) contain the reduction with the input + * vector on their neighbors (the even ranks). The first r odd + * processes and the p - 2r last processes are renumbered from + * 0 to 2^{\floor{\log_2 p}} - 1. Even ranks do not participate in the + * rest of the algorithm. + */ + + /* Find nearest power-of-two less than or equal to comm_size */ + int nprocs_pof2, size; + for( nprocs_pof2 = 1, size = comm_size; size > 0; size >>= 1, nprocs_pof2 <<= 1 ); + nprocs_pof2 >>= 1; + + nprocs_rem = comm_size - nprocs_pof2; + int log2_size; + for (log2_size = 0, size = 1; size < nprocs_pof2; ++log2_size, size <<= 1); + + if (rank < 2 * nprocs_rem) { + if ((rank % 2) == 0) { + /* Even process */ + Request::send(psend, totalcount, dtype, rank + 1, + COLL_TAG_REDUCE_SCATTER, comm); + /* This process does not participate in the rest of the algorithm */ + vrank = -1; + } else { + /* Odd process */ + Request::recv(precv, totalcount, dtype, rank - 1, + COLL_TAG_REDUCE_SCATTER, comm, MPI_STATUS_IGNORE); + op->apply(precv, psend, (int*)&totalcount, dtype); + /* Adjust rank to be the bottom "remain" ranks */ + vrank = rank / 2; + } + } else { + /* Adjust rank to show that the bottom "even remain" ranks dropped out */ + vrank = rank - nprocs_rem; + } + + if (vrank != -1) { + /* + * Now, psend vector of size totalcount is divided into nprocs_pof2 blocks: + * block 0: rcounts[0] and rcounts[1] -- for process 0 and 1 + * block 1: rcounts[2] and rcounts[3] -- for process 2 and 3 + * ... + * block r-1: rcounts[2*(r-1)] and rcounts[2*(r-1)+1] + * block r: rcounts[r+r] + * block r+1: rcounts[r+r+1] + * ... + * block nprocs_pof2 - 1: rcounts[r+nprocs_pof2-1] + */ + int nblocks = nprocs_pof2, send_index = 0, recv_index = 0; + for (int mask = 1; mask < nprocs_pof2; mask <<= 1) { + int vpeer = vrank ^ mask; + int peer = (vpeer < nprocs_rem) ? vpeer * 2 + 1 : vpeer + nprocs_rem; + + nblocks /= 2; + if ((vrank & mask) == 0) { + /* Send the upper half of reduction buffer, recv the lower half */ + send_index += nblocks; + } else { + /* Send the upper half of reduction buffer, recv the lower half */ + recv_index += nblocks; + } + + /* Send blocks: [send_index, send_index + nblocks - 1] */ + int send_count = ompi_sum_counts(rcounts, displs, nprocs_rem, + send_index, send_index + nblocks - 1); + index = (send_index < nprocs_rem) ? 2 * send_index : nprocs_rem + send_index; + ptrdiff_t sdispl = displs[index]; + + /* Recv blocks: [recv_index, recv_index + nblocks - 1] */ + int recv_count = ompi_sum_counts(rcounts, displs, nprocs_rem, + recv_index, recv_index + nblocks - 1); + index = (recv_index < nprocs_rem) ? 2 * recv_index : nprocs_rem + recv_index; + ptrdiff_t rdispl = displs[index]; + + Request::sendrecv(psend + (ptrdiff_t)sdispl * extent, send_count, + dtype, peer, COLL_TAG_REDUCE_SCATTER, + precv + (ptrdiff_t)rdispl * extent, recv_count, + dtype, peer, COLL_TAG_REDUCE_SCATTER, + comm, MPI_STATUS_IGNORE); + + if (vrank < vpeer) { + /* precv = psend precv */ + op->apply(psend + (ptrdiff_t)rdispl * extent, + precv + (ptrdiff_t)rdispl * extent, &recv_count, dtype); + char *p = psend; + psend = precv; + precv = p; + } else { + /* psend = precv psend */ + op->apply(precv + (ptrdiff_t)rdispl * extent, + psend + (ptrdiff_t)rdispl * extent, &recv_count, dtype); + } + send_index = recv_index; + } + /* + * psend points to the result block [send_index] + * Exchange results with remote process according to a mirror permutation. + */ + int vpeer = ompi_mirror_perm(vrank, log2_size); + int peer = (vpeer < nprocs_rem) ? vpeer * 2 + 1 : vpeer + nprocs_rem; + index = (send_index < nprocs_rem) ? 2 * send_index : nprocs_rem + send_index; + + if (vpeer < nprocs_rem) { + /* + * Process has two blocks: for excluded process and own. + * Send the first block to excluded process. + */ + Request::send(psend + (ptrdiff_t)displs[index] * extent, + rcounts[index], dtype, peer - 1, + COLL_TAG_REDUCE_SCATTER, + comm); + } + + /* If process has two blocks, then send the second block (own block) */ + if (vpeer < nprocs_rem) + index++; + if (vpeer != vrank) { + Request::sendrecv(psend + (ptrdiff_t)displs[index] * extent, + rcounts[index], dtype, peer, + COLL_TAG_REDUCE_SCATTER, + rbuf, rcounts[rank], dtype, peer, + COLL_TAG_REDUCE_SCATTER, + comm, MPI_STATUS_IGNORE); + } else { + err = Datatype::copy(psend + (ptrdiff_t)displs[rank] * extent, rcounts[rank], dtype, + rbuf, rcounts[rank], dtype); + if (MPI_SUCCESS != err) { goto cleanup_and_return; } + } + + } else { + /* Excluded process: receive result */ + int vpeer = ompi_mirror_perm((rank + 1) / 2, log2_size); + int peer = (vpeer < nprocs_rem) ? vpeer * 2 + 1 : vpeer + nprocs_rem; + Request::recv(rbuf, rcounts[rank], dtype, peer, + COLL_TAG_REDUCE_SCATTER, comm, + MPI_STATUS_IGNORE); + } + +cleanup_and_return: + if (displs) + free(displs); + if (tmpbuf[0]) + free(tmpbuf[0]); + if (tmpbuf[1]) + free(tmpbuf[1]); + return err; +} +} // namespace simgrid::smpi