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