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
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.
14 * Additional copyrights may follow
19 #include "colls_private.h"
20 #include "coll_tuned_topo.h"
21 #define MCA_COLL_BASE_TAG_REDUCE 555
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 );
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)
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.
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 )
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 MPI_Request reqs[2] = {MPI_REQUEST_NULL, MPI_REQUEST_NULL};
52 int num_segments, line, ret, segindex, i, rank;
53 int recvcount, prevcount, inbi;
56 * Determine number of segments and number of elements
59 smpi_datatype_extent( datatype, &lower_bound, &extent);
60 num_segments = (original_count + count_by_segment - 1) / count_by_segment;
61 segment_increment = count_by_segment * extent;
63 sendtmpbuf = (char*) sendbuf;
64 if( sendbuf == MPI_IN_PLACE ) {
65 sendtmpbuf = (char *)recvbuf;
68 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);
70 rank = smpi_comm_rank(comm);
72 /* non-leaf nodes - wait for children to send me data & forward up
74 if( tree->tree_nextsize > 0 ) {
75 ptrdiff_t true_extent, real_segment_size;
76 true_extent=smpi_datatype_get_extent( datatype);
78 /* handle non existant recv buffer (i.e. its NULL) and
79 protect the recv buffer on non-root nodes */
80 accumbuf = (char*)recvbuf;
81 if( (NULL == accumbuf) || (root != rank) ) {
82 /* Allocate temporary accumulator buffer. */
83 accumbuf_free = (char*)malloc(true_extent +
84 (original_count - 1) * extent);
85 if (accumbuf_free == NULL) {
86 line = __LINE__; ret = -1; goto error_hndl;
88 accumbuf = accumbuf_free - lower_bound;
91 /* If this is a non-commutative operation we must copy
92 sendbuf to the accumbuf, in order to simplfy the loops */
93 if (!smpi_op_is_commute(op)) {
95 (char*)sendtmpbuf, original_count, datatype,
96 (char*)accumbuf, original_count, datatype);
98 /* Allocate two buffers for incoming segments */
99 real_segment_size = true_extent + (count_by_segment - 1) * extent;
100 inbuf_free[0] = (char*) malloc(real_segment_size);
101 if( inbuf_free[0] == NULL ) {
102 line = __LINE__; ret = -1; goto error_hndl;
104 inbuf[0] = inbuf_free[0] - lower_bound;
105 /* if there is chance to overlap communication -
106 allocate second buffer */
107 if( (num_segments > 1) || (tree->tree_nextsize > 1) ) {
108 inbuf_free[1] = (char*) malloc(real_segment_size);
109 if( inbuf_free[1] == NULL ) {
110 line = __LINE__; ret = -1; goto error_hndl;
112 inbuf[1] = inbuf_free[1] - lower_bound;
115 /* reset input buffer index and receive count */
118 /* for each segment */
119 for( segindex = 0; segindex <= num_segments; segindex++ ) {
120 prevcount = recvcount;
121 /* recvcount - number of elements in current segment */
122 recvcount = count_by_segment;
123 if( segindex == (num_segments-1) )
124 recvcount = original_count - count_by_segment * segindex;
127 for( i = 0; i < tree->tree_nextsize; i++ ) {
129 * We try to overlap communication:
130 * either with next segment or with the next child
132 /* post irecv for current segindex on current child */
133 if( segindex < num_segments ) {
134 void* local_recvbuf = inbuf[inbi];
136 /* for the first step (1st child per segment) and
137 * commutative operations we might be able to irecv
138 * directly into the accumulate buffer so that we can
139 * reduce(op) this with our sendbuf in one step as
140 * ompi_op_reduce only has two buffer pointers,
141 * this avoids an extra memory copy.
143 * BUT if the operation is non-commutative or
144 * we are root and are USING MPI_IN_PLACE this is wrong!
146 if( (smpi_op_is_commute(op)) &&
147 !((MPI_IN_PLACE == sendbuf) && (rank == tree->tree_root)) ) {
148 local_recvbuf = accumbuf + segindex * segment_increment;
152 reqs[inbi]=smpi_mpi_irecv(local_recvbuf, recvcount, datatype,
154 MCA_COLL_BASE_TAG_REDUCE, comm
157 /* wait for previous req to complete, if any.
158 if there are no requests reqs[inbi ^1] will be
160 /* wait on data from last child for previous segment */
161 smpi_mpi_waitall( 1, &reqs[inbi ^ 1],
162 MPI_STATUSES_IGNORE );
163 local_op_buffer = inbuf[inbi ^ 1];
165 /* our first operation is to combine our own [sendbuf] data
166 * with the data we recvd from down stream (but only
167 * the operation is commutative and if we are not root and
168 * not using MPI_IN_PLACE)
171 if( (smpi_op_is_commute(op)) &&
172 !((MPI_IN_PLACE == sendbuf) && (rank == tree->tree_root)) ) {
173 local_op_buffer = sendtmpbuf + segindex * segment_increment;
176 /* apply operation */
177 smpi_op_apply(op, local_op_buffer,
178 accumbuf + segindex * segment_increment,
179 &recvcount, &datatype );
180 } else if ( segindex > 0 ) {
181 void* accumulator = accumbuf + (segindex-1) * segment_increment;
182 if( tree->tree_nextsize <= 1 ) {
183 if( (smpi_op_is_commute(op)) &&
184 !((MPI_IN_PLACE == sendbuf) && (rank == tree->tree_root)) ) {
185 local_op_buffer = sendtmpbuf + (segindex-1) * segment_increment;
188 smpi_op_apply(op, local_op_buffer, accumulator, &prevcount,
191 /* all reduced on available data this step (i) complete,
192 * pass to the next process unless you are the root.
194 if (rank != tree->tree_root) {
195 /* send combined/accumulated data to parent */
196 smpi_mpi_send( accumulator, prevcount,
197 datatype, tree->tree_prev,
198 MCA_COLL_BASE_TAG_REDUCE,
202 /* we stop when segindex = number of segments
203 (i.e. we do num_segment+1 steps for pipelining */
204 if (segindex == num_segments) break;
207 /* update input buffer index */
209 } /* end of for each child */
210 } /* end of for each segment */
213 if( inbuf_free[0] != NULL) free(inbuf_free[0]);
214 if( inbuf_free[1] != NULL) free(inbuf_free[1]);
215 if( accumbuf_free != NULL ) free(accumbuf_free);
219 Depending on the value of max_outstanding_reqs and
220 the number of segments we have two options:
221 - send all segments using blocking send to the parent, or
222 - avoid overflooding the parent nodes by limiting the number of
223 outstanding requests to max_oustanding_reqs.
224 TODO/POSSIBLE IMPROVEMENT: If there is a way to determine the eager size
225 for the current communication, synchronization should be used only
226 when the message/segment size is smaller than the eager size.
230 /* If the number of segments is less than a maximum number of oustanding
231 requests or there is no limit on the maximum number of outstanding
232 requests, we send data to the parent using blocking send */
233 if ((0 == max_outstanding_reqs) ||
234 (num_segments <= max_outstanding_reqs)) {
237 while ( original_count > 0) {
238 if (original_count < count_by_segment) {
239 count_by_segment = original_count;
241 smpi_mpi_send((char*)sendbuf +
242 segindex * segment_increment,
243 count_by_segment, datatype,
245 MCA_COLL_BASE_TAG_REDUCE,
248 original_count -= count_by_segment;
252 /* Otherwise, introduce flow control:
253 - post max_outstanding_reqs non-blocking synchronous send,
254 - for remaining segments
255 - wait for a ssend to complete, and post the next one.
256 - wait for all outstanding sends to complete.
261 MPI_Request* sreq = NULL;
263 sreq = (MPI_Request*) calloc( max_outstanding_reqs,
264 sizeof(MPI_Request ) );
265 if (NULL == sreq) { line = __LINE__; ret = -1; goto error_hndl; }
267 /* post first group of requests */
268 for (segindex = 0; segindex < max_outstanding_reqs; segindex++) {
269 sreq[segindex]=smpi_mpi_isend((char*)sendbuf +
270 segindex * segment_increment,
271 count_by_segment, datatype,
273 MCA_COLL_BASE_TAG_REDUCE,
275 original_count -= count_by_segment;
279 while ( original_count > 0 ) {
280 /* wait on a posted request to complete */
281 smpi_mpi_wait(&sreq[creq], MPI_STATUS_IGNORE);
282 sreq[creq] = MPI_REQUEST_NULL;
284 if( original_count < count_by_segment ) {
285 count_by_segment = original_count;
287 sreq[creq]=smpi_mpi_isend((char*)sendbuf +
288 segindex * segment_increment,
289 count_by_segment, datatype,
291 MCA_COLL_BASE_TAG_REDUCE,
293 creq = (creq + 1) % max_outstanding_reqs;
295 original_count -= count_by_segment;
298 /* Wait on the remaining request to complete */
299 smpi_mpi_waitall( max_outstanding_reqs, sreq,
300 MPI_STATUSES_IGNORE );
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);
317 /* Attention: this version of the reduce operations does not
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 !
324 int smpi_coll_tuned_reduce_ompi_chain( void *sendbuf, void *recvbuf, int count,
325 MPI_Datatype datatype,
330 uint32_t segsize=64*1024;
331 int segcount = count;
333 int fanout = smpi_comm_size(comm)/2;
335 XBT_DEBUG("coll:tuned:reduce_intra_chain rank %d fo %d ss %5d", smpi_comm_rank(comm), fanout, segsize);
338 * Determine number of segments and number of elements
341 typelng = smpi_datatype_size( datatype);
343 COLL_TUNED_COMPUTED_SEGCOUNT( segsize, typelng, segcount );
345 return smpi_coll_tuned_ompi_reduce_generic( sendbuf, recvbuf, count, datatype,
347 ompi_coll_tuned_topo_build_chain(fanout, comm, root),
352 int smpi_coll_tuned_reduce_ompi_pipeline( void *sendbuf, void *recvbuf,
353 int count, MPI_Datatype datatype,
359 int segcount = count;
361 // COLL_TUNED_UPDATE_PIPELINE( comm, tuned_module, root );
364 * Determine number of segments and number of elements
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;
375 if (communicator_size > (a2 * message_size + b2)) {
378 }else if (communicator_size > (a4 * message_size + b4)) {
386 XBT_DEBUG("coll:tuned:reduce_intra_pipeline rank %d ss %5d",
387 smpi_comm_rank(comm), segsize);
389 COLL_TUNED_COMPUTED_SEGCOUNT( segsize, typelng, segcount );
391 return smpi_coll_tuned_ompi_reduce_generic( sendbuf, recvbuf, count, datatype,
393 ompi_coll_tuned_topo_build_chain( 1, comm, root),
397 int smpi_coll_tuned_reduce_ompi_binary( void *sendbuf, void *recvbuf,
398 int count, MPI_Datatype datatype,
403 int segcount = count;
409 * Determine number of segments and number of elements
412 typelng=smpi_datatype_size( datatype );
417 XBT_DEBUG("coll:tuned:reduce_intra_binary rank %d ss %5d",
418 smpi_comm_rank(comm), segsize);
420 COLL_TUNED_COMPUTED_SEGCOUNT( segsize, typelng, segcount );
422 return smpi_coll_tuned_ompi_reduce_generic( sendbuf, recvbuf, count, datatype,
424 ompi_coll_tuned_topo_build_tree(2, comm, root),
428 int smpi_coll_tuned_reduce_ompi_binomial( void *sendbuf, void *recvbuf,
429 int count, MPI_Datatype datatype,
435 int segcount = count;
438 const double a1 = 0.6016 / 1024.0; /* [1/B] */
439 const double b1 = 1.3496;
441 // COLL_TUNED_UPDATE_IN_ORDER_BMTREE( comm, tuned_module, root );
444 * Determine number of segments and number of elements
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)) {
454 } else if (communicator_size > (a1 * message_size + b1)) {
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 );
463 return smpi_coll_tuned_ompi_reduce_generic( sendbuf, recvbuf, count, datatype,
465 ompi_coll_tuned_topo_build_in_order_bmtree(comm, root),
470 * reduce_intra_in_order_binary
472 * Function: Logarithmic reduce operation for non-commutative operations.
473 * Acecpts: same as MPI_Reduce()
474 * Returns: MPI_SUCCESS or error code
476 int smpi_coll_tuned_reduce_ompi_in_order_binary( void *sendbuf, void *recvbuf,
478 MPI_Datatype datatype,
484 int rank, size, io_root;
485 int segcount = count;
486 void *use_this_sendbuf = NULL, *use_this_recvbuf = NULL;
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",
495 * Determine number of segments and number of elements
498 typelng=smpi_datatype_size( datatype);
499 COLL_TUNED_COMPUTED_SEGCOUNT( segsize, typelng, segcount );
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.
509 use_this_sendbuf = sendbuf;
510 use_this_recvbuf = recvbuf;
511 if (io_root != root) {
515 ext=smpi_datatype_get_extent(datatype);
516 text=smpi_datatype_get_extent(datatype);
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;
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;
532 use_this_recvbuf = tmpbuf;
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,
539 ompi_coll_tuned_topo_build_in_order_bintree(comm),
541 if (MPI_SUCCESS != ret) { return ret; }
544 if (io_root != root) {
546 /* Receive result from rank io_root to recvbuf */
547 smpi_mpi_recv(recvbuf, count, datatype, io_root,
548 MCA_COLL_BASE_TAG_REDUCE, comm,
550 if (MPI_IN_PLACE == sendbuf) {
551 free(use_this_sendbuf);
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 MCA_COLL_BASE_TAG_REDUCE,
559 free(use_this_recvbuf);
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.
578 /* copied function (with appropriate renaming) starts here */
583 * Function: - reduction using O(N) algorithm
584 * Accepts: - same as MPI_Reduce()
585 * Returns: - MPI_SUCCESS or error code
589 smpi_coll_tuned_reduce_ompi_basic_linear(void *sbuf, void *rbuf, int count,
596 ptrdiff_t true_extent, lb, extent;
597 char *free_buffer = NULL;
598 char *pml_buffer = NULL;
599 char *inplace_temp = NULL;
604 rank = smpi_comm_rank(comm);
605 size = smpi_comm_size(comm);
607 XBT_DEBUG("coll:tuned:reduce_intra_basic_linear rank %d", rank);
609 /* If not root, send data to the root. */
612 smpi_mpi_send(sbuf, count, dtype, root,
613 MCA_COLL_BASE_TAG_REDUCE,
618 /* see discussion in ompi_coll_basic_reduce_lin_intra about
619 extent and true extent */
620 /* for reducing buffer allocation lengths.... */
622 smpi_datatype_extent(dtype, &lb, &extent);
623 true_extent = smpi_datatype_get_extent(dtype);
625 if (MPI_IN_PLACE == sbuf) {
627 inplace_temp = (char*)malloc(true_extent + (count - 1) * extent);
628 if (NULL == inplace_temp) {
631 rbuf = inplace_temp - lb;
635 free_buffer = (char*)malloc(true_extent + (count - 1) * extent);
636 pml_buffer = free_buffer - lb;
639 /* Initialize the receive buffer. */
641 if (rank == (size - 1)) {
642 smpi_datatype_copy((char*)sbuf, count, dtype,(char*)rbuf, count, dtype);
644 smpi_mpi_recv(rbuf, count, dtype, size - 1,
645 MCA_COLL_BASE_TAG_REDUCE, comm,
649 /* Loop receiving and calling reduction function (C or Fortran). */
651 for (i = size - 2; i >= 0; --i) {
655 smpi_mpi_recv(pml_buffer, count, dtype, i,
656 MCA_COLL_BASE_TAG_REDUCE, comm,
661 /* Perform the reduction */
662 smpi_op_apply(op, inbuf, rbuf, &count, &dtype);
665 if (NULL != inplace_temp) {
666 smpi_datatype_copy(inplace_temp, count, dtype,(char*)sbuf
670 if (NULL != free_buffer) {
678 /* copied function (with appropriate renaming) ends here */