src/smpi/colls/allreduce/allreduce-ompi-ring-segmented.cpp

   1 /* Copyright (c) 2013-2019. The SimGrid Team.
   2  * All rights reserved.                                                     */
   3
   4 /* This program is free software; you can redistribute it and/or modify it
   5  * under the terms of the license (GNU LGPL) which comes with this package. */
   6
   7 /*
   8  * Copyright (c) 2004-2005 The Trustees of Indiana University and Indiana
   9  *                         University Research and Technology
  10  *                         Corporation.  All rights reserved.
  11  * Copyright (c) 2004-2009 The University of Tennessee and The University
  12  *                         of Tennessee Research Foundation.  All rights
  13  *                         reserved.
  14  * Copyright (c) 2004-2005 High Performance Computing Center Stuttgart,
  15  *                         University of Stuttgart.  All rights reserved.
  16  * Copyright (c) 2004-2005 The Regents of the University of California.
  17  *                         All rights reserved.
  18  * Copyright (c) 2009      University of Houston. All rights reserved.
  19  *
  20  * Additional copyrights may follow
  21  *
  22  *  Redistribution and use in source and binary forms, with or without
  23  * modification, are permitted provided that the following conditions are
  24  * met:
  25
  26  * - Redistributions of source code must retain the above copyright
  27  *   notice, this list of conditions and the following disclaimer.
  28
  29  * - Redistributions in binary form must reproduce the above copyright
  30  *   notice, this list of conditions and the following disclaimer listed
  31  *   in this license in the documentation and/or other materials
  32  *   provided with the distribution.
  33
  34  * - Neither the name of the copyright holders nor the names of its
  35  *   contributors may be used to endorse or promote products derived from
  36  *   this software without specific prior written permission.
  37
  38  * The copyright holders provide no reassurances that the source code
  39  * provided does not infringe any patent, copyright, or any other
  40  * intellectual property rights of third parties.  The copyright holders
  41  * disclaim any liability to any recipient for claims brought against
  42  * recipient by any third party for infringement of that parties
  43  * intellectual property rights.
  44
  45  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  46  * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  47  * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  48  * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  49  * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  50  * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  51  * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  52  * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  53  * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  54  * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  55  * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  56  */
  57
  58 /*
  59  *  ompi_coll_tuned_allreduce_intra_ring_segmented
  60  *
  61  *   Function:       Pipelined ring algorithm for allreduce operation
  62  *   Accepts:        Same as MPI_Allreduce(), segment size
  63  *   Returns:        MPI_SUCCESS or error code
  64  *
  65  *   Description:    Implements pipelined ring algorithm for allreduce:
  66  *                   user supplies suggested segment size for the pipelining of
  67  *                   reduce operation.
  68  *                   The segment size determines the number of phases, np, for
  69  *                   the algorithm execution.
  70  *                   The message is automatically divided into blocks of
  71  *                   approximately  (count / (np * segcount)) elements.
  72  *                   At the end of reduction phase, allgather like step is
  73  *                   executed.
  74  *                   Algorithm requires (np + 1)*(N - 1) steps.
  75  *
  76  *   Limitations:    The algorithm DOES NOT preserve order of operations so it
  77  *                   can be used only for commutative operations.
  78  *                   In addition, algorithm cannot work if the total size is
  79  *                   less than size * segment size.
  80  *         Example on 3 nodes with 2 phases
  81  *         Initial state
  82  *   #      0              1             2
  83  *        [00a]          [10a]         [20a]
  84  *        [00b]          [10b]         [20b]
  85  *        [01a]          [11a]         [21a]
  86  *        [01b]          [11b]         [21b]
  87  *        [02a]          [12a]         [22a]
  88  *        [02b]          [12b]         [22b]
  89  *
  90  *        COMPUTATION PHASE 0 (a)
  91  *         Step 0: rank r sends block ra to rank (r+1) and receives bloc (r-1)a
  92  *                 from rank (r-1) [with wraparound].
  93  *    #     0              1             2
  94  *        [00a]        [00a+10a]       [20a]
  95  *        [00b]          [10b]         [20b]
  96  *        [01a]          [11a]       [11a+21a]
  97  *        [01b]          [11b]         [21b]
  98  *      [22a+02a]        [12a]         [22a]
  99  *        [02b]          [12b]         [22b]
 100  *
 101  *         Step 1: rank r sends block (r-1)a to rank (r+1) and receives bloc
 102  *                 (r-2)a from rank (r-1) [with wraparound].
 103  *    #     0              1             2
 104  *        [00a]        [00a+10a]   [00a+10a+20a]
 105  *        [00b]          [10b]         [20b]
 106  *    [11a+21a+01a]      [11a]       [11a+21a]
 107  *        [01b]          [11b]         [21b]
 108  *      [22a+02a]    [22a+02a+12a]     [22a]
 109  *        [02b]          [12b]         [22b]
 110  *
 111  *        COMPUTATION PHASE 1 (b)
 112  *         Step 0: rank r sends block rb to rank (r+1) and receives bloc (r-1)b
 113  *                 from rank (r-1) [with wraparound].
 114  *    #     0              1             2
 115  *        [00a]        [00a+10a]       [20a]
 116  *        [00b]        [00b+10b]       [20b]
 117  *        [01a]          [11a]       [11a+21a]
 118  *        [01b]          [11b]       [11b+21b]
 119  *      [22a+02a]        [12a]         [22a]
 120  *      [22b+02b]        [12b]         [22b]
 121  *
 122  *         Step 1: rank r sends block (r-1)b to rank (r+1) and receives bloc
 123  *                 (r-2)b from rank (r-1) [with wraparound].
 124  *    #     0              1             2
 125  *        [00a]        [00a+10a]   [00a+10a+20a]
 126  *        [00b]          [10b]     [0bb+10b+20b]
 127  *    [11a+21a+01a]      [11a]       [11a+21a]
 128  *    [11b+21b+01b]      [11b]         [21b]
 129  *      [22a+02a]    [22a+02a+12a]     [22a]
 130  *        [02b]      [22b+01b+12b]     [22b]
 131  *
 132  *
 133  *        DISTRIBUTION PHASE: ring ALLGATHER with ranks shifted by 1 (same as
 134  *         in regular ring algorithm.
 135  *
 136  */
 137
 138 #define COLL_TUNED_COMPUTED_SEGCOUNT(SEGSIZE, TYPELNG, SEGCOUNT)        \
 139     if( ((SEGSIZE) >= (TYPELNG)) &&                                     \
 140         ((SEGSIZE) < ((TYPELNG) * (SEGCOUNT))) ) {                      \
 141         size_t residual;                                                \
 142         (SEGCOUNT) = (int)((SEGSIZE) / (TYPELNG));                      \
 143         residual = (SEGSIZE) - (SEGCOUNT) * (TYPELNG);                  \
 144         if( residual > ((TYPELNG) >> 1) )                               \
 145             (SEGCOUNT)++;                                               \
 146     }                                                                   \
 147
 148 #define COLL_TUNED_COMPUTE_BLOCKCOUNT( COUNT, NUM_BLOCKS, SPLIT_INDEX,       \
 149                                        EARLY_BLOCK_COUNT, LATE_BLOCK_COUNT ) \
 150     EARLY_BLOCK_COUNT = LATE_BLOCK_COUNT = COUNT / NUM_BLOCKS;               \
 151     SPLIT_INDEX = COUNT % NUM_BLOCKS;                                        \
 152     if (0 != SPLIT_INDEX) {                                                  \
 153         EARLY_BLOCK_COUNT = EARLY_BLOCK_COUNT + 1;                           \
 154     }                                                                        \
 155
 156 #include "../colls_private.hpp"
 157
 158 namespace simgrid{
 159 namespace smpi{
 160 int
 161 Coll_allreduce_ompi_ring_segmented::allreduce(void *sbuf, void *rbuf, int count,
 162                                                MPI_Datatype dtype,
 163                                                MPI_Op op,
 164                                                MPI_Comm comm)
 165 {
 166    int ret = MPI_SUCCESS;
 167    int line;
 168    int k, recv_from, send_to;
 169    int early_blockcount, late_blockcount, split_rank;
 170    int segcount, max_segcount;
 171    int num_phases, phase;
 172    int block_count;
 173    unsigned int inbi;
 174    size_t typelng;
 175    char *tmpsend = NULL, *tmprecv = NULL;
 176    char *inbuf[2] = {NULL, NULL};
 177    ptrdiff_t true_extent, extent;
 178    ptrdiff_t block_offset, max_real_segsize;
 179    MPI_Request reqs[2] = {NULL, NULL};
 180    const size_t segsize = 1 << 20; /* 1 MB */
 181    int size = comm->size();
 182    int rank = comm->rank();
 183
 184    XBT_DEBUG("coll:tuned:allreduce_intra_ring_segmented rank %d, count %d", rank, count);
 185
 186    /* Special case for size == 1 */
 187    if (1 == size) {
 188       if (MPI_IN_PLACE != sbuf) {
 189       ret= Datatype::copy(sbuf, count, dtype,rbuf, count, dtype);
 190          if (ret < 0) { line = __LINE__; goto error_hndl; }
 191       }
 192       return MPI_SUCCESS;
 193    }
 194
 195    /* Determine segment count based on the suggested segment size */
 196    extent = dtype->get_extent();
 197    if (MPI_SUCCESS != ret) { line = __LINE__; goto error_hndl; }
 198    true_extent = dtype->get_extent();
 199    if (MPI_SUCCESS != ret) { line = __LINE__; goto error_hndl; }
 200    typelng = dtype->size();
 201    if (MPI_SUCCESS != ret) { line = __LINE__; goto error_hndl; }
 202    segcount = count;
 203    COLL_TUNED_COMPUTED_SEGCOUNT(segsize, typelng, segcount)
 204
 205    /* Special case for count less than size * segcount - use regular ring */
 206    if (count < size * segcount) {
 207       XBT_DEBUG( "coll:tuned:allreduce_ring_segmented rank %d/%d, count %d, switching to regular ring", rank, size, count);
 208       return (Coll_allreduce_lr::allreduce(sbuf, rbuf, count, dtype, op,
 209                                                    comm));
 210    }
 211
 212    /* Determine the number of phases of the algorithm */
 213    num_phases = count / (size * segcount);
 214    if ((count % (size * segcount) >= size) &&
 215        (count % (size * segcount) > ((size * segcount) / 2))) {
 216       num_phases++;
 217    }
 218
 219    /* Determine the number of elements per block and corresponding
 220       block sizes.
 221       The blocks are divided into "early" and "late" ones:
 222       blocks 0 .. (split_rank - 1) are "early" and
 223       blocks (split_rank) .. (size - 1) are "late".
 224       Early blocks are at most 1 element larger than the late ones.
 225       Note, these blocks will be split into num_phases segments,
 226       out of the largest one will have max_segcount elements.
 227     */
 228    COLL_TUNED_COMPUTE_BLOCKCOUNT( count, size, split_rank,
 229                                   early_blockcount, late_blockcount )
 230    COLL_TUNED_COMPUTE_BLOCKCOUNT( early_blockcount, num_phases, inbi,
 231                                   max_segcount, k)
 232    max_real_segsize = true_extent + (max_segcount - 1) * extent;
 233
 234    /* Allocate and initialize temporary buffers */
 235    inbuf[0] = (char*)smpi_get_tmp_sendbuffer(max_real_segsize);
 236    if (NULL == inbuf[0]) { ret = -1; line = __LINE__; goto error_hndl; }
 237    if (size > 2) {
 238       inbuf[1] = (char*)smpi_get_tmp_recvbuffer(max_real_segsize);
 239       if (NULL == inbuf[1]) { ret = -1; line = __LINE__; goto error_hndl; }
 240    }
 241
 242    /* Handle MPI_IN_PLACE */
 243    if (MPI_IN_PLACE != sbuf) {
 244       ret= Datatype::copy(sbuf, count, dtype,rbuf, count, dtype);
 245       if (ret < 0) { line = __LINE__; goto error_hndl; }
 246    }
 247
 248    /* Computation loop: for each phase, repeat ring allreduce computation loop */
 249    for (phase = 0; phase < num_phases; phase ++) {
 250       ptrdiff_t phase_offset;
 251       int early_phase_segcount, late_phase_segcount, split_phase, phase_count;
 252
 253       /*
 254          For each of the remote nodes:
 255          - post irecv for block (r-1)
 256          - send block (r)
 257            To do this, first compute block offset and count, and use block offset
 258            to compute phase offset.
 259          - in loop for every step k = 2 .. n
 260            - post irecv for block (r + n - k) % n
 261            - wait on block (r + n - k + 1) % n to arrive
 262            - compute on block (r + n - k + 1) % n
 263            - send block (r + n - k + 1) % n
 264          - wait on block (r + 1)
 265          - compute on block (r + 1)
 266          - send block (r + 1) to rank (r + 1)
 267          Note that we must be careful when computing the begining of buffers and
 268          for send operations and computation we must compute the exact block size.
 269       */
 270       send_to = (rank + 1) % size;
 271       recv_from = (rank + size - 1) % size;
 272
 273       inbi = 0;
 274       /* Initialize first receive from the neighbor on the left */
 275       reqs[inbi] = Request::irecv(inbuf[inbi], max_segcount, dtype, recv_from,
 276                                666, comm);
 277       /* Send first block (my block) to the neighbor on the right:
 278          - compute my block and phase offset
 279          - send data */
 280       block_offset = ((rank < split_rank)?
 281                       (rank * early_blockcount) :
 282                       (rank * late_blockcount + split_rank));
 283       block_count = ((rank < split_rank)? early_blockcount : late_blockcount);
 284       COLL_TUNED_COMPUTE_BLOCKCOUNT(block_count, num_phases, split_phase,
 285                                     early_phase_segcount, late_phase_segcount)
 286       phase_count = ((phase < split_phase)?
 287                      (early_phase_segcount) : (late_phase_segcount));
 288       phase_offset = ((phase < split_phase)?
 289                       (phase * early_phase_segcount) :
 290                       (phase * late_phase_segcount + split_phase));
 291       tmpsend = ((char*)rbuf) + (block_offset + phase_offset) * extent;
 292       Request::send(tmpsend, phase_count, dtype, send_to,
 293                               666, comm);
 294
 295       for (k = 2; k < size; k++) {
 296          const int prevblock = (rank + size - k + 1) % size;
 297
 298          inbi = inbi ^ 0x1;
 299
 300          /* Post irecv for the current block */
 301          reqs[inbi] = Request::irecv(inbuf[inbi], max_segcount, dtype, recv_from,
 302                                666, comm);
 303          if (MPI_SUCCESS != ret) { line = __LINE__; goto error_hndl; }
 304
 305          /* Wait on previous block to arrive */
 306          Request::wait(&reqs[inbi ^ 0x1], MPI_STATUS_IGNORE);
 307
 308          /* Apply operation on previous block: result goes to rbuf
 309             rbuf[prevblock] = inbuf[inbi ^ 0x1] (op) rbuf[prevblock]
 310          */
 311          block_offset = ((prevblock < split_rank)?
 312                          (prevblock * early_blockcount) :
 313                          (prevblock * late_blockcount + split_rank));
 314          block_count = ((prevblock < split_rank)?
 315                         early_blockcount : late_blockcount);
 316          COLL_TUNED_COMPUTE_BLOCKCOUNT(block_count, num_phases, split_phase,
 317                                        early_phase_segcount, late_phase_segcount)
 318          phase_count = ((phase < split_phase)?
 319                         (early_phase_segcount) : (late_phase_segcount));
 320          phase_offset = ((phase < split_phase)?
 321                          (phase * early_phase_segcount) :
 322                          (phase * late_phase_segcount + split_phase));
 323          tmprecv = ((char*)rbuf) + (block_offset + phase_offset) * extent;
 324          if(op!=MPI_OP_NULL) op->apply( inbuf[inbi ^ 0x1], tmprecv, &phase_count, dtype);
 325          /* send previous block to send_to */
 326          Request::send(tmprecv, phase_count, dtype, send_to,
 327                               666, comm);
 328       }
 329
 330       /* Wait on the last block to arrive */
 331       Request::wait(&reqs[inbi], MPI_STATUS_IGNORE);
 332
 333
 334       /* Apply operation on the last block (from neighbor (rank + 1)
 335          rbuf[rank+1] = inbuf[inbi] (op) rbuf[rank + 1] */
 336       recv_from = (rank + 1) % size;
 337       block_offset = ((recv_from < split_rank)?
 338                       (recv_from * early_blockcount) :
 339                       (recv_from * late_blockcount + split_rank));
 340       block_count = ((recv_from < split_rank)?
 341                      early_blockcount : late_blockcount);
 342       COLL_TUNED_COMPUTE_BLOCKCOUNT(block_count, num_phases, split_phase,
 343                                     early_phase_segcount, late_phase_segcount)
 344       phase_count = ((phase < split_phase)?
 345                      (early_phase_segcount) : (late_phase_segcount));
 346       phase_offset = ((phase < split_phase)?
 347                       (phase * early_phase_segcount) :
 348                       (phase * late_phase_segcount + split_phase));
 349       tmprecv = ((char*)rbuf) + (block_offset + phase_offset) * extent;
 350       if(op!=MPI_OP_NULL) op->apply( inbuf[inbi], tmprecv, &phase_count, dtype);
 351    }
 352
 353    /* Distribution loop - variation of ring allgather */
 354    send_to = (rank + 1) % size;
 355    recv_from = (rank + size - 1) % size;
 356    for (k = 0; k < size - 1; k++) {
 357       const int recv_data_from = (rank + size - k) % size;
 358       const int send_data_from = (rank + 1 + size - k) % size;
 359       const int send_block_offset =
 360          ((send_data_from < split_rank)?
 361           (send_data_from * early_blockcount) :
 362           (send_data_from * late_blockcount + split_rank));
 363       const int recv_block_offset =
 364          ((recv_data_from < split_rank)?
 365           (recv_data_from * early_blockcount) :
 366           (recv_data_from * late_blockcount + split_rank));
 367       block_count = ((send_data_from < split_rank)?
 368                      early_blockcount : late_blockcount);
 369
 370       tmprecv = (char*)rbuf + recv_block_offset * extent;
 371       tmpsend = (char*)rbuf + send_block_offset * extent;
 372
 373       Request::sendrecv(tmpsend, block_count, dtype, send_to,
 374                                      666,
 375                                      tmprecv, early_blockcount, dtype, recv_from,
 376                                      666,
 377                                      comm, MPI_STATUS_IGNORE);
 378
 379    }
 380
 381    if (NULL != inbuf[0]) smpi_free_tmp_buffer(inbuf[0]);
 382    if (NULL != inbuf[1]) smpi_free_tmp_buffer(inbuf[1]);
 383
 384    return MPI_SUCCESS;
 385
 386  error_hndl:
 387    XBT_DEBUG("%s:%4d\tRank %d Error occurred %d\n",
 388                 __FILE__, line, rank, ret);
 389    if (NULL != inbuf[0]) smpi_free_tmp_buffer(inbuf[0]);
 390    if (NULL != inbuf[1]) smpi_free_tmp_buffer(inbuf[1]);
 391    return ret;
 392 }
 393 }
 394 }