src/smpi/colls/reduce-ompi.c

   1 /*
   2  * Copyright (c) 2004-2005 The Trustees of Indiana University and Indiana
   3  *                         University Research and Technology
   4  *                         Corporation.  All rights reserved.
   5  * Copyright (c) 2004-2009 The University of Tennessee and The University
   6  *                         of Tennessee Research Foundation.  All rights
   7  *                         reserved.
   8  * Copyright (c) 2004-2005 High Performance Computing Center Stuttgart,
   9  *                         University of Stuttgart.  All rights reserved.
  10  * Copyright (c) 2004-2005 The Regents of the University of California.
  11  *                         All rights reserved.
  12  * $COPYRIGHT$
  13  *
  14  * Additional copyrights may follow
  15  *
  16  * $HEADER$
  17  */
  18
  19 #include "colls_private.h"
  20 #include "coll_tuned_topo.h"
  21 #define MCA_COLL_BASE_TAG_REDUCE 555
  22
  23
  24
  25 int smpi_coll_tuned_ompi_reduce_generic( void* sendbuf, void* recvbuf, int original_count,
  26                                     MPI_Datatype datatype, MPI_Op  op,
  27                                     int root, MPI_Comm comm,
  28                                     ompi_coll_tree_t* tree, int count_by_segment,
  29                                     int max_outstanding_reqs );
  30 /**
  31  * This is a generic implementation of the reduce protocol. It used the tree
  32  * provided as an argument and execute all operations using a segment of
  33  * count times a datatype.
  34  * For the last communication it will update the count in order to limit
  35  * the number of datatype to the original count (original_count)
  36  *
  37  * Note that for non-commutative operations we cannot save memory copy
  38  * for the first block: thus we must copy sendbuf to accumbuf on intermediate
  39  * to keep the optimized loop happy.
  40  */
  41 int smpi_coll_tuned_ompi_reduce_generic( void* sendbuf, void* recvbuf, int original_count,
  42                                     MPI_Datatype datatype, MPI_Op  op,
  43                                     int root, MPI_Comm comm,
  44                                     ompi_coll_tree_t* tree, int count_by_segment,
  45                                     int max_outstanding_reqs )
  46 {
  47     char *inbuf[2] = {NULL, NULL}, *inbuf_free[2] = {NULL, NULL};
  48     char *accumbuf = NULL, *accumbuf_free = NULL;
  49     char *local_op_buffer = NULL, *sendtmpbuf = NULL;
  50     ptrdiff_t extent, lower_bound, segment_increment;
  51     size_t typelng;
  52     MPI_Request  reqs[2] = {MPI_REQUEST_NULL, MPI_REQUEST_NULL};
  53     int num_segments, line, ret, segindex, i, rank;
  54     int recvcount, prevcount, inbi;
  55
  56     /**
  57      * Determine number of segments and number of elements
  58      * sent per operation
  59      */
  60     smpi_datatype_extent( datatype, &lower_bound, &extent);
  61     typelng = smpi_datatype_size( datatype );
  62     num_segments = (original_count + count_by_segment - 1) / count_by_segment;
  63     segment_increment = count_by_segment * extent;
  64
  65     sendtmpbuf = (char*) sendbuf;
  66     if( sendbuf == MPI_IN_PLACE ) {
  67         sendtmpbuf = (char *)recvbuf;
  68     }
  69
  70     XBT_DEBUG( "coll:tuned:reduce_generic count %d, msg size %ld, segsize %ld, max_requests %d", original_count, (unsigned long)(num_segments * segment_increment), (unsigned long)segment_increment, max_outstanding_reqs);
  71
  72     rank = smpi_comm_rank(comm);
  73
  74     /* non-leaf nodes - wait for children to send me data & forward up
  75        (if needed) */
  76     if( tree->tree_nextsize > 0 ) {
  77         ptrdiff_t true_extent, real_segment_size;
  78         true_extent=smpi_datatype_get_extent( datatype);
  79
  80         /* handle non existant recv buffer (i.e. its NULL) and
  81            protect the recv buffer on non-root nodes */
  82         accumbuf = (char*)recvbuf;
  83         if( (NULL == accumbuf) || (root != rank) ) {
  84             /* Allocate temporary accumulator buffer. */
  85             accumbuf_free = (char*)malloc(true_extent +
  86                                           (original_count - 1) * extent);
  87             if (accumbuf_free == NULL) {
  88                 line = __LINE__; ret = -1; goto error_hndl;
  89             }
  90             accumbuf = accumbuf_free - lower_bound;
  91         }
  92
  93         /* If this is a non-commutative operation we must copy
  94            sendbuf to the accumbuf, in order to simplfy the loops */
  95         if (!smpi_op_is_commute(op)) {
  96             smpi_datatype_copy(
  97                                                 (char*)accumbuf, original_count, datatype,
  98                                                 (char*)sendtmpbuf, original_count, datatype);
  99         }
 100         /* Allocate two buffers for incoming segments */
 101         real_segment_size = true_extent + (count_by_segment - 1) * extent;
 102         inbuf_free[0] = (char*) malloc(real_segment_size);
 103         if( inbuf_free[0] == NULL ) {
 104             line = __LINE__; ret = -1; goto error_hndl;
 105         }
 106         inbuf[0] = inbuf_free[0] - lower_bound;
 107         /* if there is chance to overlap communication -
 108            allocate second buffer */
 109         if( (num_segments > 1) || (tree->tree_nextsize > 1) ) {
 110             inbuf_free[1] = (char*) malloc(real_segment_size);
 111             if( inbuf_free[1] == NULL ) {
 112                 line = __LINE__; ret = -1; goto error_hndl;
 113             }
 114             inbuf[1] = inbuf_free[1] - lower_bound;
 115         }
 116
 117         /* reset input buffer index and receive count */
 118         inbi = 0;
 119         recvcount = 0;
 120         /* for each segment */
 121         for( segindex = 0; segindex <= num_segments; segindex++ ) {
 122             prevcount = recvcount;
 123             /* recvcount - number of elements in current segment */
 124             recvcount = count_by_segment;
 125             if( segindex == (num_segments-1) )
 126                 recvcount = original_count - count_by_segment * segindex;
 127
 128             /* for each child */
 129             for( i = 0; i < tree->tree_nextsize; i++ ) {
 130                 /**
 131                  * We try to overlap communication:
 132                  * either with next segment or with the next child
 133                  */
 134                 /* post irecv for current segindex on current child */
 135                 if( segindex < num_segments ) {
 136                     void* local_recvbuf = inbuf[inbi];
 137                     if( 0 == i ) {
 138                         /* for the first step (1st child per segment) and
 139                          * commutative operations we might be able to irecv
 140                          * directly into the accumulate buffer so that we can
 141                          * reduce(op) this with our sendbuf in one step as
 142                          * ompi_op_reduce only has two buffer pointers,
 143                          * this avoids an extra memory copy.
 144                          *
 145                          * BUT if the operation is non-commutative or
 146                          * we are root and are USING MPI_IN_PLACE this is wrong!
 147                          */
 148                         if( (smpi_op_is_commute(op)) &&
 149                             !((MPI_IN_PLACE == sendbuf) && (rank == tree->tree_root)) ) {
 150                             local_recvbuf = accumbuf + segindex * segment_increment;
 151                         }
 152                     }
 153
 154                     reqs[inbi]=smpi_mpi_irecv(local_recvbuf, recvcount, datatype,
 155                                              tree->tree_next[i],
 156                                              MCA_COLL_BASE_TAG_REDUCE, comm
 157                                              );
 158                 }
 159                 /* wait for previous req to complete, if any.
 160                    if there are no requests reqs[inbi ^1] will be
 161                    MPI_REQUEST_NULL. */
 162                 /* wait on data from last child for previous segment */
 163                 smpi_mpi_waitall( 1, &reqs[inbi ^ 1],
 164                                              MPI_STATUSES_IGNORE );
 165                 local_op_buffer = inbuf[inbi ^ 1];
 166                 if( i > 0 ) {
 167                     /* our first operation is to combine our own [sendbuf] data
 168                      * with the data we recvd from down stream (but only
 169                      * the operation is commutative and if we are not root and
 170                      * not using MPI_IN_PLACE)
 171                      */
 172                     if( 1 == i ) {
 173                         if( (smpi_op_is_commute(op)) &&
 174                             !((MPI_IN_PLACE == sendbuf) && (rank == tree->tree_root)) ) {
 175                             local_op_buffer = sendtmpbuf + segindex * segment_increment;
 176                         }
 177                     }
 178                     /* apply operation */
 179                     smpi_op_apply(op, local_op_buffer,
 180                                    accumbuf + segindex * segment_increment,
 181                                    &recvcount, &datatype );
 182                 } else if ( segindex > 0 ) {
 183                     void* accumulator = accumbuf + (segindex-1) * segment_increment;
 184                     if( tree->tree_nextsize <= 1 ) {
 185                         if( (smpi_op_is_commute(op)) &&
 186                             !((MPI_IN_PLACE == sendbuf) && (rank == tree->tree_root)) ) {
 187                             local_op_buffer = sendtmpbuf + (segindex-1) * segment_increment;
 188                         }
 189                     }
 190                     smpi_op_apply(op, local_op_buffer, accumulator, &prevcount,
 191                                    &datatype );
 192
 193                     /* all reduced on available data this step (i) complete,
 194                      * pass to the next process unless you are the root.
 195                      */
 196                     if (rank != tree->tree_root) {
 197                         /* send combined/accumulated data to parent */
 198                         smpi_mpi_send( accumulator, prevcount,
 199                                                   datatype, tree->tree_prev,
 200                                                   MCA_COLL_BASE_TAG_REDUCE,
 201                                                   comm);
 202                     }
 203
 204                     /* we stop when segindex = number of segments
 205                        (i.e. we do num_segment+1 steps for pipelining */
 206                     if (segindex == num_segments) break;
 207                 }
 208
 209                 /* update input buffer index */
 210                 inbi = inbi ^ 1;
 211             } /* end of for each child */
 212         } /* end of for each segment */
 213
 214         /* clean up */
 215         if( inbuf_free[0] != NULL) free(inbuf_free[0]);
 216         if( inbuf_free[1] != NULL) free(inbuf_free[1]);
 217         if( accumbuf_free != NULL ) free(accumbuf_free);
 218     }
 219
 220     /* leaf nodes
 221        Depending on the value of max_outstanding_reqs and
 222        the number of segments we have two options:
 223        - send all segments using blocking send to the parent, or
 224        - avoid overflooding the parent nodes by limiting the number of
 225        outstanding requests to max_oustanding_reqs.
 226        TODO/POSSIBLE IMPROVEMENT: If there is a way to determine the eager size
 227        for the current communication, synchronization should be used only
 228        when the message/segment size is smaller than the eager size.
 229     */
 230     else {
 231
 232         /* If the number of segments is less than a maximum number of oustanding
 233            requests or there is no limit on the maximum number of outstanding
 234            requests, we send data to the parent using blocking send */
 235         if ((0 == max_outstanding_reqs) ||
 236             (num_segments <= max_outstanding_reqs)) {
 237
 238             segindex = 0;
 239             while ( original_count > 0) {
 240                 if (original_count < count_by_segment) {
 241                     count_by_segment = original_count;
 242                 }
 243                 smpi_mpi_send((char*)sendbuf +
 244                                          segindex * segment_increment,
 245                                          count_by_segment, datatype,
 246                                          tree->tree_prev,
 247                                          MCA_COLL_BASE_TAG_REDUCE,
 248                                          comm) ;
 249                 segindex++;
 250                 original_count -= count_by_segment;
 251             }
 252         }
 253
 254         /* Otherwise, introduce flow control:
 255            - post max_outstanding_reqs non-blocking synchronous send,
 256            - for remaining segments
 257            - wait for a ssend to complete, and post the next one.
 258            - wait for all outstanding sends to complete.
 259         */
 260         else {
 261
 262             int creq = 0;
 263             MPI_Request* sreq = NULL;
 264
 265             sreq = (MPI_Request*) calloc( max_outstanding_reqs,
 266                                               sizeof(MPI_Request ) );
 267             if (NULL == sreq) { line = __LINE__; ret = -1; goto error_hndl; }
 268
 269             /* post first group of requests */
 270             for (segindex = 0; segindex < max_outstanding_reqs; segindex++) {
 271                 sreq[segindex]=smpi_mpi_isend((char*)sendbuf +
 272                                           segindex * segment_increment,
 273                                           count_by_segment, datatype,
 274                                           tree->tree_prev,
 275                                           MCA_COLL_BASE_TAG_REDUCE,
 276                                           comm);
 277                 original_count -= count_by_segment;
 278             }
 279
 280             creq = 0;
 281             while ( original_count > 0 ) {
 282                 /* wait on a posted request to complete */
 283                 smpi_mpi_wait(&sreq[creq], MPI_STATUS_IGNORE);
 284                 sreq[creq] = MPI_REQUEST_NULL;
 285
 286                 if( original_count < count_by_segment ) {
 287                     count_by_segment = original_count;
 288                 }
 289                 sreq[creq]=smpi_mpi_isend((char*)sendbuf +
 290                                           segindex * segment_increment,
 291                                           count_by_segment, datatype,
 292                                           tree->tree_prev,
 293                                           MCA_COLL_BASE_TAG_REDUCE,
 294                                           comm );
 295                 creq = (creq + 1) % max_outstanding_reqs;
 296                 segindex++;
 297                 original_count -= count_by_segment;
 298             }
 299
 300             /* Wait on the remaining request to complete */
 301             smpi_mpi_waitall( max_outstanding_reqs, sreq,
 302                                          MPI_STATUSES_IGNORE );
 303
 304             /* free requests */
 305             free(sreq);
 306         }
 307     }
 308     return MPI_SUCCESS;
 309
 310  error_hndl:  /* error handler */
 311     XBT_DEBUG("ERROR_HNDL: node %d file %s line %d error %d\n",
 312                    rank, __FILE__, line, ret );
 313     if( inbuf_free[0] != NULL ) free(inbuf_free[0]);
 314     if( inbuf_free[1] != NULL ) free(inbuf_free[1]);
 315     if( accumbuf_free != NULL ) free(accumbuf);
 316     return ret;
 317 }
 318
 319 /* Attention: this version of the reduce operations does not
 320    work for:
 321    - non-commutative operations
 322    - segment sizes which are not multiplies of the extent of the datatype
 323      meaning that at least one datatype must fit in the segment !
 324 */
 325
 326 int smpi_coll_tuned_reduce_ompi_chain( void *sendbuf, void *recvbuf, int count,
 327                                         MPI_Datatype datatype,
 328                                         MPI_Op  op, int root,
 329                                         MPI_Comm  comm
 330                                         )
 331 {
 332     uint32_t segsize=64*1024;
 333     int segcount = count;
 334     size_t typelng;
 335     int fanout = smpi_comm_size(comm)/2;
 336
 337     XBT_DEBUG("coll:tuned:reduce_intra_chain rank %d fo %d ss %5d", smpi_comm_rank(comm), fanout, segsize);
 338
 339     /**
 340      * Determine number of segments and number of elements
 341      * sent per operation
 342      */
 343     typelng = smpi_datatype_size( datatype);
 344
 345     COLL_TUNED_COMPUTED_SEGCOUNT( segsize, typelng, segcount );
 346
 347     return smpi_coll_tuned_ompi_reduce_generic( sendbuf, recvbuf, count, datatype,
 348                                            op, root, comm,
 349                                            ompi_coll_tuned_topo_build_chain(fanout, comm, root),
 350                                            segcount, 0 );
 351 }
 352
 353
 354 int smpi_coll_tuned_reduce_ompi_pipeline( void *sendbuf, void *recvbuf,
 355                                            int count, MPI_Datatype datatype,
 356                                            MPI_Op  op, int root,
 357                                            MPI_Comm  comm  )
 358 {
 359
 360     uint32_t segsize;
 361     int segcount = count;
 362     size_t typelng;
 363 //    COLL_TUNED_UPDATE_PIPELINE( comm, tuned_module, root );
 364
 365     /**
 366      * Determine number of segments and number of elements
 367      * sent per operation
 368      */
 369     const double a2 =  0.0410 / 1024.0; /* [1/B] */
 370     const double b2 =  9.7128;
 371     const double a4 =  0.0033 / 1024.0; /* [1/B] */
 372     const double b4 =  1.6761;
 373     typelng= smpi_datatype_size( datatype);
 374     int communicator_size = smpi_comm_size(comm);
 375     size_t message_size = typelng * count;
 376
 377     if (communicator_size > (a2 * message_size + b2)) {
 378         // Pipeline_1K
 379         segsize = 1024;
 380     }else if (communicator_size > (a4 * message_size + b4)) {
 381         // Pipeline_32K
 382         segsize = 32*1024;
 383     } else {
 384         // Pipeline_64K
 385         segsize = 64*1024;
 386     }
 387
 388     XBT_DEBUG("coll:tuned:reduce_intra_pipeline rank %d ss %5d",
 389                  smpi_comm_rank(comm), segsize);
 390
 391     COLL_TUNED_COMPUTED_SEGCOUNT( segsize, typelng, segcount );
 392
 393     return smpi_coll_tuned_ompi_reduce_generic( sendbuf, recvbuf, count, datatype,
 394                                            op, root, comm,
 395                                            ompi_coll_tuned_topo_build_chain( 1, comm, root),
 396                                            segcount, 0);
 397 }
 398
 399 int smpi_coll_tuned_reduce_ompi_binary( void *sendbuf, void *recvbuf,
 400                                          int count, MPI_Datatype datatype,
 401                                          MPI_Op  op, int root,
 402                                          MPI_Comm  comm)
 403 {
 404     uint32_t segsize;
 405     int segcount = count;
 406     size_t typelng;
 407
 408
 409
 410     /**
 411      * Determine number of segments and number of elements
 412      * sent per operation
 413      */
 414     typelng=smpi_datatype_size( datatype );
 415
 416         // Binary_32K
 417     segsize = 32*1024;
 418
 419     XBT_DEBUG("coll:tuned:reduce_intra_binary rank %d ss %5d",
 420                  smpi_comm_rank(comm), segsize);
 421
 422     COLL_TUNED_COMPUTED_SEGCOUNT( segsize, typelng, segcount );
 423
 424     return smpi_coll_tuned_ompi_reduce_generic( sendbuf, recvbuf, count, datatype,
 425                                            op, root, comm,
 426                                            ompi_coll_tuned_topo_build_tree(2, comm, root),
 427                                            segcount, 0);
 428 }
 429
 430 int smpi_coll_tuned_reduce_ompi_binomial( void *sendbuf, void *recvbuf,
 431                                            int count, MPI_Datatype datatype,
 432                                            MPI_Op  op, int root,
 433                                            MPI_Comm  comm)
 434 {
 435
 436     uint32_t segsize=0;
 437     int segcount = count;
 438     size_t typelng;
 439
 440     const double a1 =  0.6016 / 1024.0; /* [1/B] */
 441     const double b1 =  1.3496;
 442
 443 //    COLL_TUNED_UPDATE_IN_ORDER_BMTREE( comm, tuned_module, root );
 444
 445     /**
 446      * Determine number of segments and number of elements
 447      * sent per operation
 448      */
 449     typelng= smpi_datatype_size( datatype);
 450     int communicator_size = smpi_comm_size(comm);
 451     size_t message_size = typelng * count;
 452     if (((communicator_size < 8) && (message_size < 20480)) ||
 453                (message_size < 2048) || (count <= 1)) {
 454         /* Binomial_0K */
 455         segsize = 0;
 456     } else if (communicator_size > (a1 * message_size + b1)) {
 457         // Binomial_1K
 458         segsize = 1024;
 459     }
 460
 461     XBT_DEBUG("coll:tuned:reduce_intra_binomial rank %d ss %5d",
 462                  smpi_comm_rank(comm), segsize);
 463     COLL_TUNED_COMPUTED_SEGCOUNT( segsize, typelng, segcount );
 464
 465     return smpi_coll_tuned_ompi_reduce_generic( sendbuf, recvbuf, count, datatype,
 466                                            op, root, comm,
 467                                            ompi_coll_tuned_topo_build_in_order_bmtree(comm, root),
 468                                            segcount, 0);
 469 }
 470
 471 /*
 472  * reduce_intra_in_order_binary
 473  *
 474  * Function:      Logarithmic reduce operation for non-commutative operations.
 475  * Acecpts:       same as MPI_Reduce()
 476  * Returns:       MPI_SUCCESS or error code
 477  */
 478 int smpi_coll_tuned_reduce_ompi_in_order_binary( void *sendbuf, void *recvbuf,
 479                                                   int count,
 480                                                   MPI_Datatype datatype,
 481                                                   MPI_Op  op, int root,
 482                                                   MPI_Comm  comm)
 483 {
 484     uint32_t segsize=0;
 485     int ret;
 486     int rank, size, io_root;
 487     int segcount = count;
 488     void *use_this_sendbuf = NULL, *use_this_recvbuf = NULL;
 489     size_t typelng;
 490
 491     rank = smpi_comm_rank(comm);
 492     size = smpi_comm_size(comm);
 493     XBT_DEBUG("coll:tuned:reduce_intra_in_order_binary rank %d ss %5d",
 494                  rank, segsize);
 495
 496     /**
 497      * Determine number of segments and number of elements
 498      * sent per operation
 499      */
 500     typelng=smpi_datatype_size( datatype);
 501     COLL_TUNED_COMPUTED_SEGCOUNT( segsize, typelng, segcount );
 502
 503     /* An in-order binary tree must use root (size-1) to preserve the order of
 504        operations.  Thus, if root is not rank (size - 1), then we must handle
 505        1. MPI_IN_PLACE option on real root, and
 506        2. we must allocate temporary recvbuf on rank (size - 1).
 507        Note that generic function must be careful not to switch order of
 508        operations for non-commutative ops.
 509     */
 510     io_root = size - 1;
 511     use_this_sendbuf = sendbuf;
 512     use_this_recvbuf = recvbuf;
 513     if (io_root != root) {
 514         ptrdiff_t text, ext;
 515         char *tmpbuf = NULL;
 516
 517         ext=smpi_datatype_get_extent(datatype);
 518         text=smpi_datatype_get_extent(datatype);
 519
 520         if ((root == rank) && (MPI_IN_PLACE == sendbuf)) {
 521             tmpbuf = (char *) malloc(text + (count - 1) * ext);
 522             if (NULL == tmpbuf) {
 523                 return MPI_ERR_INTERN;
 524             }
 525             smpi_datatype_copy (
 526                                                 (char*)tmpbuf, count, datatype,
 527                                                 (char*)recvbuf, count, datatype);
 528             use_this_sendbuf = tmpbuf;
 529         } else if (io_root == rank) {
 530             tmpbuf = (char *) malloc(text + (count - 1) * ext);
 531             if (NULL == tmpbuf) {
 532                 return MPI_ERR_INTERN;
 533             }
 534             use_this_recvbuf = tmpbuf;
 535         }
 536     }
 537
 538     /* Use generic reduce with in-order binary tree topology and io_root */
 539     ret = smpi_coll_tuned_ompi_reduce_generic( use_this_sendbuf, use_this_recvbuf, count, datatype,
 540                                           op, io_root, comm,
 541                                           ompi_coll_tuned_topo_build_in_order_bintree(comm),
 542                                           segcount, 0 );
 543     if (MPI_SUCCESS != ret) { return ret; }
 544
 545     /* Clean up */
 546     if (io_root != root) {
 547         if (root == rank) {
 548             /* Receive result from rank io_root to recvbuf */
 549             smpi_mpi_recv(recvbuf, count, datatype, io_root,
 550                                     MCA_COLL_BASE_TAG_REDUCE, comm,
 551                                     MPI_STATUS_IGNORE);
 552             if (MPI_IN_PLACE == sendbuf) {
 553                 free(use_this_sendbuf);
 554             }
 555
 556         } else if (io_root == rank) {
 557             /* Send result from use_this_recvbuf to root */
 558             smpi_mpi_send(use_this_recvbuf, count, datatype, root,
 559                                     MCA_COLL_BASE_TAG_REDUCE,
 560                                     comm);
 561             free(use_this_recvbuf);
 562         }
 563     }
 564
 565     return MPI_SUCCESS;
 566 }
 567
 568 /*
 569  * Linear functions are copied from the BASIC coll module
 570  * they do not segment the message and are simple implementations
 571  * but for some small number of nodes and/or small data sizes they
 572  * are just as fast as tuned/tree based segmenting operations
 573  * and as such may be selected by the decision functions
 574  * These are copied into this module due to the way we select modules
 575  * in V1. i.e. in V2 we will handle this differently and so will not
 576  * have to duplicate code.
 577  * GEF Oct05 after asking Jeff.
 578  */
 579
 580 /* copied function (with appropriate renaming) starts here */
 581
 582 /*
 583  *  reduce_lin_intra
 584  *
 585  *  Function:   - reduction using O(N) algorithm
 586  *  Accepts:    - same as MPI_Reduce()
 587  *  Returns:    - MPI_SUCCESS or error code
 588  */
 589
 590 int
 591 smpi_coll_tuned_reduce_ompi_basic_linear(void *sbuf, void *rbuf, int count,
 592                                           MPI_Datatype dtype,
 593                                           MPI_Op op,
 594                                           int root,
 595                                           MPI_Comm comm)
 596 {
 597     int i, rank, size;
 598     ptrdiff_t true_extent, lb, extent;
 599     char *free_buffer = NULL;
 600     char *pml_buffer = NULL;
 601     char *inplace_temp = NULL;
 602     char *inbuf;
 603
 604     /* Initialize */
 605
 606     rank = smpi_comm_rank(comm);
 607     size = smpi_comm_size(comm);
 608
 609     XBT_DEBUG("coll:tuned:reduce_intra_basic_linear rank %d", rank);
 610
 611     /* If not root, send data to the root. */
 612
 613     if (rank != root) {
 614         smpi_mpi_send(sbuf, count, dtype, root,
 615                                 MCA_COLL_BASE_TAG_REDUCE,
 616                                 comm);
 617         return -1;
 618     }
 619
 620     /* see discussion in ompi_coll_basic_reduce_lin_intra about
 621        extent and true extent */
 622     /* for reducing buffer allocation lengths.... */
 623
 624     smpi_datatype_extent(dtype, &lb, &extent);
 625     true_extent = smpi_datatype_get_extent(dtype);
 626
 627     if (MPI_IN_PLACE == sbuf) {
 628         sbuf = rbuf;
 629         inplace_temp = (char*)malloc(true_extent + (count - 1) * extent);
 630         if (NULL == inplace_temp) {
 631             return -1;
 632         }
 633         rbuf = inplace_temp - lb;
 634     }
 635
 636     if (size > 1) {
 637         free_buffer = (char*)malloc(true_extent + (count - 1) * extent);
 638         pml_buffer = free_buffer - lb;
 639     }
 640
 641     /* Initialize the receive buffer. */
 642
 643     if (rank == (size - 1)) {
 644         smpi_datatype_copy((char*)rbuf, count, dtype,
 645                                                   (char*)sbuf, count, dtype);
 646     } else {
 647         smpi_mpi_recv(rbuf, count, dtype, size - 1,
 648                                 MCA_COLL_BASE_TAG_REDUCE, comm,
 649                                 MPI_STATUS_IGNORE);
 650     }
 651
 652     /* Loop receiving and calling reduction function (C or Fortran). */
 653
 654     for (i = size - 2; i >= 0; --i) {
 655         if (rank == i) {
 656             inbuf = (char*)sbuf;
 657         } else {
 658             smpi_mpi_recv(pml_buffer, count, dtype, i,
 659                                     MCA_COLL_BASE_TAG_REDUCE, comm,
 660                                     MPI_STATUS_IGNORE);
 661             inbuf = pml_buffer;
 662         }
 663
 664         /* Perform the reduction */
 665         smpi_op_apply(op, inbuf, rbuf, &count, &dtype);
 666     }
 667
 668     if (NULL != inplace_temp) {
 669         smpi_datatype_copy((char*)sbuf, count, dtype,
 670                                                   inplace_temp,count , dtype);
 671         free(inplace_temp);
 672     }
 673     if (NULL != free_buffer) {
 674         free(free_buffer);
 675     }
 676
 677     /* All done */
 678     return MPI_SUCCESS;
 679 }
 680
 681 /* copied function (with appropriate renaming) ends here */
 682
 683