Logo AND Algorithmique Numérique Distribuée

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