1 /* Copyright (c) 2013-2018. The SimGrid Team.
2 * All rights reserved. */
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. */
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
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.
19 * Additional copyrights may follow
22 #include "../coll_tuned_topo.hpp"
23 #include "../colls_private.hpp"
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 );
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)
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.
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 )
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;
59 * Determine number of segments and number of elements
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;
66 sendtmpbuf = (char*) sendbuf;
67 if( sendbuf == MPI_IN_PLACE ) {
68 sendtmpbuf = (char *)recvbuf;
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);
77 /* non-leaf nodes - wait for children to send me data & forward up
79 if( tree->tree_nextsize > 0 ) {
80 ptrdiff_t true_extent, real_segment_size;
81 true_extent=datatype->get_extent();
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;
93 accumbuf = accumbuf_free - lower_bound;
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);
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;
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;
115 inbuf[1] = inbuf_free[1] - lower_bound;
118 /* reset input buffer index and receive count */
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;
130 for( i = 0; i < tree->tree_nextsize; i++ ) {
132 * We try to overlap communication:
133 * either with next segment or with the next child
135 /* post irecv for current segindex on current child */
136 if( segindex < num_segments ) {
137 void* local_recvbuf = inbuf[inbi];
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.
146 * BUT if the operation is non-commutative or
147 * we are root and are USING MPI_IN_PLACE this is wrong!
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;
155 reqs[inbi]=Request::irecv(local_recvbuf, recvcount, datatype,
157 COLL_TAG_REDUCE, comm
160 /* wait for previous req to complete, if any.
161 if there are no requests reqs[inbi ^1] will be
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];
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)
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;
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;
191 if(op!=MPI_OP_NULL) op->apply( local_op_buffer, accumulator, &prevcount,
194 /* all reduced on available data this step (i) complete,
195 * pass to the next process unless you are the root.
197 if (rank != tree->tree_root) {
198 /* send combined/accumulated data to parent */
199 Request::send( accumulator, prevcount,
200 datatype, tree->tree_prev,
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;
210 /* update input buffer index */
212 } /* end of for each child */
213 } /* end of for each segment */
216 smpi_free_tmp_buffer(inbuf_free[0]);
217 smpi_free_tmp_buffer(inbuf_free[1]);
218 smpi_free_tmp_buffer(accumbuf_free);
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.
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)) {
240 while ( original_count > 0) {
241 if (original_count < count_by_segment) {
242 count_by_segment = original_count;
244 Request::send((char*)sendbuf +
245 segindex * segment_increment,
246 count_by_segment, datatype,
251 original_count -= count_by_segment;
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.
264 MPI_Request* sreq = new (std::nothrow) MPI_Request[max_outstanding_reqs];
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]=Request::isend((char*)sendbuf +
270 segindex * segment_increment,
271 count_by_segment, datatype,
275 original_count -= count_by_segment;
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;
284 if( original_count < count_by_segment ) {
285 count_by_segment = original_count;
287 sreq[creq]=Request::isend((char*)sendbuf +
288 segindex * segment_increment,
289 count_by_segment, datatype,
293 creq = (creq + 1) % max_outstanding_reqs;
295 original_count -= count_by_segment;
298 /* Wait on the remaining request to complete */
299 Request::waitall( max_outstanding_reqs, sreq,
300 MPI_STATUSES_IGNORE );
306 ompi_coll_tuned_topo_destroy_tree(&tree);
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);
318 /* Attention: this version of the reduce operations does not
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 !
326 int Coll_reduce_ompi_chain::reduce( void *sendbuf, void *recvbuf, int count,
327 MPI_Datatype datatype,
332 uint32_t segsize=64*1024;
333 int segcount = count;
335 int fanout = comm->size()/2;
337 XBT_DEBUG("coll:tuned:reduce_intra_chain rank %d fo %d ss %5u", comm->rank(), fanout, segsize);
340 * Determine number of segments and number of elements
343 typelng = datatype->size();
345 COLL_TUNED_COMPUTED_SEGCOUNT( segsize, typelng, segcount );
347 return smpi_coll_tuned_ompi_reduce_generic( sendbuf, recvbuf, count, datatype,
349 ompi_coll_tuned_topo_build_chain(fanout, comm, root),
354 int Coll_reduce_ompi_pipeline::reduce( void *sendbuf, void *recvbuf,
355 int count, MPI_Datatype datatype,
361 int segcount = count;
363 // COLL_TUNED_UPDATE_PIPELINE( comm, tuned_module, root );
366 * Determine number of segments and number of elements
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;
377 if (communicator_size > (a2 * message_size + b2)) {
380 }else if (communicator_size > (a4 * message_size + b4)) {
388 XBT_DEBUG("coll:tuned:reduce_intra_pipeline rank %d ss %5u", comm->rank(), segsize);
390 COLL_TUNED_COMPUTED_SEGCOUNT( segsize, typelng, segcount );
392 return smpi_coll_tuned_ompi_reduce_generic( sendbuf, recvbuf, count, datatype,
394 ompi_coll_tuned_topo_build_chain( 1, comm, root),
398 int Coll_reduce_ompi_binary::reduce( void *sendbuf, void *recvbuf,
399 int count, MPI_Datatype datatype,
404 int segcount = count;
410 * Determine number of segments and number of elements
413 typelng=datatype->size();
418 XBT_DEBUG("coll:tuned:reduce_intra_binary rank %d ss %5u", comm->rank(), 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 Coll_reduce_ompi_binomial::reduce( 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= 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)) {
454 } else if (communicator_size > (a1 * message_size + b1)) {
459 XBT_DEBUG("coll:tuned:reduce_intra_binomial rank %d ss %5u", comm->rank(), segsize);
460 COLL_TUNED_COMPUTED_SEGCOUNT( segsize, typelng, segcount );
462 return smpi_coll_tuned_ompi_reduce_generic( sendbuf, recvbuf, count, datatype,
464 ompi_coll_tuned_topo_build_in_order_bmtree(comm, root),
469 * reduce_intra_in_order_binary
471 * Function: Logarithmic reduce operation for non-commutative operations.
472 * Acecpts: same as MPI_Reduce()
473 * Returns: MPI_SUCCESS or error code
475 int Coll_reduce_ompi_in_order_binary::reduce( void *sendbuf, void *recvbuf,
477 MPI_Datatype datatype,
483 int rank, size, io_root;
484 int segcount = count;
485 void *use_this_sendbuf = NULL, *use_this_recvbuf = NULL;
490 XBT_DEBUG("coll:tuned:reduce_intra_in_order_binary rank %d ss %5u", rank, segsize);
493 * Determine number of segments and number of elements
496 typelng=datatype->size();
497 COLL_TUNED_COMPUTED_SEGCOUNT( segsize, typelng, segcount );
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.
507 use_this_sendbuf = sendbuf;
508 use_this_recvbuf = recvbuf;
509 if (io_root != root) {
513 ext=datatype->get_extent();
514 text=datatype->get_extent();
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;
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;
530 use_this_recvbuf = tmpbuf;
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,
537 ompi_coll_tuned_topo_build_in_order_bintree(comm),
539 if (MPI_SUCCESS != ret) { return ret; }
542 if (io_root != root) {
544 /* Receive result from rank io_root to recvbuf */
545 Request::recv(recvbuf, count, datatype, io_root,
546 COLL_TAG_REDUCE, comm,
548 if (MPI_IN_PLACE == sendbuf) {
549 smpi_free_tmp_buffer(use_this_sendbuf);
552 } else if (io_root == rank) {
553 /* Send result from use_this_recvbuf to root */
554 Request::send(use_this_recvbuf, count, datatype, root,
557 smpi_free_tmp_buffer(use_this_recvbuf);
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.
576 /* copied function (with appropriate renaming) starts here */
581 * Function: - reduction using O(N) algorithm
582 * Accepts: - same as MPI_Reduce()
583 * Returns: - MPI_SUCCESS or error code
587 Coll_reduce_ompi_basic_linear::reduce(void *sbuf, void *rbuf, int count,
594 ptrdiff_t true_extent, lb, extent;
595 char *free_buffer = NULL;
596 char *pml_buffer = NULL;
597 char *inplace_temp = NULL;
605 XBT_DEBUG("coll:tuned:reduce_intra_basic_linear rank %d", rank);
607 /* If not root, send data to the root. */
610 Request::send(sbuf, count, dtype, root,
616 /* see discussion in ompi_coll_basic_reduce_lin_intra about
617 extent and true extent */
618 /* for reducing buffer allocation lengths.... */
620 dtype->extent(&lb, &extent);
621 true_extent = dtype->get_extent();
623 if (MPI_IN_PLACE == sbuf) {
625 inplace_temp = (char*)smpi_get_tmp_recvbuffer(true_extent + (count - 1) * extent);
626 if (NULL == inplace_temp) {
629 rbuf = inplace_temp - lb;
633 free_buffer = (char*)smpi_get_tmp_recvbuffer(true_extent + (count - 1) * extent);
634 pml_buffer = free_buffer - lb;
637 /* Initialize the receive buffer. */
639 if (rank == (size - 1)) {
640 Datatype::copy((char*)sbuf, count, dtype,(char*)rbuf, count, dtype);
642 Request::recv(rbuf, count, dtype, size - 1,
643 COLL_TAG_REDUCE, comm,
647 /* Loop receiving and calling reduction function (C or Fortran). */
649 for (i = size - 2; i >= 0; --i) {
653 Request::recv(pml_buffer, count, dtype, i,
654 COLL_TAG_REDUCE, comm,
659 /* Perform the reduction */
660 if(op!=MPI_OP_NULL) op->apply( inbuf, rbuf, &count, dtype);
663 if (NULL != inplace_temp) {
664 Datatype::copy(inplace_temp, count, dtype,(char*)sbuf
666 smpi_free_tmp_buffer(inplace_temp);
668 if (NULL != free_buffer) {
669 smpi_free_tmp_buffer(free_buffer);
676 /* copied function (with appropriate renaming) ends here */