#include "colls_private.h"
#include "coll_tuned_topo.h"
-#include "xbt/replay.h"
/*
* Recursive-halving function is (*mostly*) copied from the BASIC coll module.
size = comm->size();
XBT_DEBUG("coll:tuned:reduce_scatter_ompi_basic_recursivehalving, rank %d", rank);
- if(!smpi_op_is_commute(op))
+ if( (op!=MPI_OP_NULL && !op->is_commutative()))
THROWF(arg_error,0, " reduce_scatter ompi_basic_recursivehalving can only be used for commutative operations! ");
/* Find displacements and the like */
}
/* get datatype information */
- smpi_datatype_extent(dtype, &lb, &extent);
- smpi_datatype_extent(dtype, &true_lb, &true_extent);
+ dtype->extent(&lb, &extent);
+ dtype->extent(&true_lb, &true_extent);
buf_size = true_extent + (ptrdiff_t)(count - 1) * extent;
/* Handle MPI_IN_PLACE */
result_buf = result_buf_free - lb;
/* copy local buffer into the temporary results */
- err =smpi_datatype_copy(sbuf, count, dtype, result_buf, count, dtype);
+ err =Datatype::copy(sbuf, count, dtype, result_buf, count, dtype);
if (MPI_SUCCESS != err) goto cleanup;
/* figure out power of two mapping: grow until larger than
two procs to do the rest of the algorithm */
if (rank < 2 * remain) {
if ((rank & 1) == 0) {
- smpi_mpi_send(result_buf, count, dtype, rank + 1,
+ Request::send(result_buf, count, dtype, rank + 1,
COLL_TAG_REDUCE_SCATTER,
comm);
/* we don't participate from here on out */
tmp_rank = -1;
} else {
- smpi_mpi_recv(recv_buf, count, dtype, rank - 1,
+ Request::recv(recv_buf, count, dtype, rank - 1,
COLL_TAG_REDUCE_SCATTER,
comm, MPI_STATUS_IGNORE);
/* integrate their results into our temp results */
- smpi_op_apply(op, recv_buf, result_buf, &count, &dtype);
+ if(op!=MPI_OP_NULL) op->apply( recv_buf, result_buf, &count, dtype);
/* adjust rank to be the bottom "remain" ranks */
tmp_rank = rank / 2;
/* actual data transfer. Send from result_buf,
receive into recv_buf */
if (send_count > 0 && recv_count != 0) {
- request=smpi_mpi_irecv(recv_buf + (ptrdiff_t)tmp_disps[recv_index] * extent,
+ request=Request::irecv(recv_buf + (ptrdiff_t)tmp_disps[recv_index] * extent,
recv_count, dtype, peer,
COLL_TAG_REDUCE_SCATTER,
comm);
}
}
if (recv_count > 0 && send_count != 0) {
- smpi_mpi_send(result_buf + (ptrdiff_t)tmp_disps[send_index] * extent,
+ Request::send(result_buf + (ptrdiff_t)tmp_disps[send_index] * extent,
send_count, dtype, peer,
COLL_TAG_REDUCE_SCATTER,
comm);
}
}
if (send_count > 0 && recv_count != 0) {
- smpi_mpi_wait(&request, MPI_STATUS_IGNORE);
+ Request::wait(&request, MPI_STATUS_IGNORE);
}
/* if we received something on this step, push it into
the results buffer */
if (recv_count > 0) {
- smpi_op_apply(op,
+ if(op!=MPI_OP_NULL) op->apply(
recv_buf + (ptrdiff_t)tmp_disps[recv_index] * extent,
result_buf + (ptrdiff_t)tmp_disps[recv_index] * extent,
- &recv_count, &dtype);
+ &recv_count, dtype);
}
/* update for next iteration */
/* copy local results from results buffer into real receive buffer */
if (0 != rcounts[rank]) {
- err = smpi_datatype_copy(result_buf + disps[rank] * extent,
+ err = Datatype::copy(result_buf + disps[rank] * extent,
rcounts[rank], dtype,
rbuf, rcounts[rank], dtype);
if (MPI_SUCCESS != err) {
if (rank < (2 * remain)) {
if ((rank & 1) == 0) {
if (rcounts[rank]) {
- smpi_mpi_recv(rbuf, rcounts[rank], dtype, rank + 1,
+ Request::recv(rbuf, rcounts[rank], dtype, rank + 1,
COLL_TAG_REDUCE_SCATTER,
comm, MPI_STATUS_IGNORE);
}
} else {
if (rcounts[rank - 1]) {
- smpi_mpi_send(result_buf + disps[rank - 1] * extent,
+ Request::send(result_buf + disps[rank - 1] * extent,
rcounts[rank - 1], dtype, rank - 1,
COLL_TAG_REDUCE_SCATTER,
comm);
/* Special case for size == 1 */
if (1 == size) {
if (MPI_IN_PLACE != sbuf) {
- ret = smpi_datatype_copy((char*)sbuf, total_count, dtype, (char*)rbuf, total_count, dtype);
+ ret = Datatype::copy((char*)sbuf, total_count, dtype, (char*)rbuf, total_count, dtype);
if (ret < 0) { line = __LINE__; goto error_hndl; }
}
xbt_free(displs);
rbuf can be of rcounts[rank] size.
- up to two temporary buffers used for communication/computation overlap.
*/
- smpi_datatype_extent(dtype, &lb, &extent);
- smpi_datatype_extent(dtype, &true_lb, &true_extent);
+ dtype->extent(&lb, &extent);
+ dtype->extent(&true_lb, &true_extent);
max_real_segsize = true_extent + (ptrdiff_t)(max_block_count - 1) * extent;
sbuf = rbuf;
}
- ret = smpi_datatype_copy((char*)sbuf, total_count, dtype, accumbuf, total_count, dtype);
+ ret = Datatype::copy((char*)sbuf, total_count, dtype, accumbuf, total_count, dtype);
if (ret < 0) { line = __LINE__; goto error_hndl; }
/* Computation loop */
inbi = 0;
/* Initialize first receive from the neighbor on the left */
- reqs[inbi]=smpi_mpi_irecv(inbuf[inbi], max_block_count, dtype, recv_from,
+ reqs[inbi]=Request::irecv(inbuf[inbi], max_block_count, dtype, recv_from,
COLL_TAG_REDUCE_SCATTER, comm
);
tmpsend = accumbuf + (ptrdiff_t)displs[recv_from] * extent;
- smpi_mpi_send(tmpsend, rcounts[recv_from], dtype, send_to,
+ Request::send(tmpsend, rcounts[recv_from], dtype, send_to,
COLL_TAG_REDUCE_SCATTER,
comm);
inbi = inbi ^ 0x1;
/* Post irecv for the current block */
- reqs[inbi]=smpi_mpi_irecv(inbuf[inbi], max_block_count, dtype, recv_from,
+ reqs[inbi]=Request::irecv(inbuf[inbi], max_block_count, dtype, recv_from,
COLL_TAG_REDUCE_SCATTER, comm
);
/* Wait on previous block to arrive */
- smpi_mpi_wait(&reqs[inbi ^ 0x1], MPI_STATUS_IGNORE);
+ Request::wait(&reqs[inbi ^ 0x1], MPI_STATUS_IGNORE);
/* Apply operation on previous block: result goes to rbuf
rbuf[prevblock] = inbuf[inbi ^ 0x1] (op) rbuf[prevblock]
*/
tmprecv = accumbuf + (ptrdiff_t)displs[prevblock] * extent;
- smpi_op_apply(op, inbuf[inbi ^ 0x1], tmprecv, &(rcounts[prevblock]), &dtype);
+ if(op!=MPI_OP_NULL) op->apply( inbuf[inbi ^ 0x1], tmprecv, &(rcounts[prevblock]), dtype);
/* send previous block to send_to */
- smpi_mpi_send(tmprecv, rcounts[prevblock], dtype, send_to,
+ Request::send(tmprecv, rcounts[prevblock], dtype, send_to,
COLL_TAG_REDUCE_SCATTER,
comm);
}
/* Wait on the last block to arrive */
- smpi_mpi_wait(&reqs[inbi], MPI_STATUS_IGNORE);
+ Request::wait(&reqs[inbi], MPI_STATUS_IGNORE);
/* Apply operation on the last block (my block)
rbuf[rank] = inbuf[inbi] (op) rbuf[rank] */
tmprecv = accumbuf + (ptrdiff_t)displs[rank] * extent;
- smpi_op_apply(op, inbuf[inbi], tmprecv, &(rcounts[rank]), &dtype);
+ if(op!=MPI_OP_NULL) op->apply( inbuf[inbi], tmprecv, &(rcounts[rank]), dtype);
/* Copy result from tmprecv to rbuf */
- ret = smpi_datatype_copy(tmprecv, rcounts[rank], dtype, (char*)rbuf, rcounts[rank], dtype);
+ ret = Datatype::copy(tmprecv, rcounts[rank], dtype, (char*)rbuf, rcounts[rank], dtype);
if (ret < 0) { line = __LINE__; goto error_hndl; }
if (NULL != displs) xbt_free(displs);
