int rank = comm->rank();
MPI_Request* requests;
(*request) = new Request( nullptr, 0, MPI_BYTE,
- rank,rank, COLL_TAG_BARRIER, comm, MPI_REQ_PERSISTENT);
+ rank,rank, COLL_TAG_BCAST, comm, MPI_REQ_PERSISTENT);
if (rank != root) {
requests = new MPI_Request[1];
requests[0] = Request::irecv (buf, count, datatype, root,
int rank = comm->rank();
int size = comm->size();
(*request) = new Request( nullptr, 0, MPI_BYTE,
- rank,rank, COLL_TAG_BARRIER, comm, MPI_REQ_PERSISTENT);
+ rank,rank, system_tag, comm, MPI_REQ_PERSISTENT);
// FIXME: check for errors
recvtype->extent(&lb, &recvext);
// Local copy from self
int rank = comm->rank();
int size = comm->size();
(*request) = new Request( nullptr, 0, MPI_BYTE,
- rank,rank, COLL_TAG_BARRIER, comm, MPI_REQ_PERSISTENT);
+ rank,rank, system_tag, comm, MPI_REQ_PERSISTENT);
if(rank != root) {
requests = new MPI_Request[1];
// Recv buffer from root
int rank = comm->rank();
int size = comm->size();
(*request) = new Request( nullptr, 0, MPI_BYTE,
- rank,rank, COLL_TAG_BARRIER, comm, MPI_REQ_PERSISTENT);
+ rank,rank, system_tag, comm, MPI_REQ_PERSISTENT);
recvtype->extent(&lb, &recvext);
// Local copy from self
Datatype::copy(sendbuf, sendcount, sendtype,
int rank = comm->rank();
int size = comm->size();
(*request) = new Request( nullptr, 0, MPI_BYTE,
- rank,rank, COLL_TAG_ALLTOALL, comm, MPI_REQ_PERSISTENT);
+ rank,rank, system_tag, comm, MPI_REQ_PERSISTENT);
sendtype->extent(&lb, &sendext);
recvtype->extent(&lb, &recvext);
/* simple optimization */
int rank = comm->rank();
int size = comm->size();
(*request) = new Request( nullptr, 0, MPI_BYTE,
- rank,rank, COLL_TAG_ALLTOALLV, comm, MPI_REQ_PERSISTENT);
+ rank,rank, system_tag, comm, MPI_REQ_PERSISTENT);
sendtype->extent(&lb, &sendext);
recvtype->extent(&lb, &recvext);
/* Local copy from self */
int count = 0;
/* Create all receives that will be posted first */
for (int i = 0; i < size; ++i) {
- if (i != rank && recvcounts[i] != 0) {
+ if (i != rank) {
requests[count] = Request::irecv_init(static_cast<char *>(recvbuf) + recvdisps[i] * recvext,
recvcounts[i], recvtype, i, system_tag, comm);
count++;
}
/* Now create all sends */
for (int i = 0; i < size; ++i) {
- if (i != rank && sendcounts[i] != 0) {
+ if (i != rank) {
requests[count] = Request::isend_init(static_cast<char *>(sendbuf) + senddisps[i] * sendext,
sendcounts[i], sendtype, i, system_tag, comm);
count++;
return err;
}
+int Colls::ialltoallw(void *sendbuf, int *sendcounts, int *senddisps, MPI_Datatype* sendtypes,
+ void *recvbuf, int *recvcounts, int *recvdisps, MPI_Datatype* recvtypes, MPI_Comm comm, MPI_Request *request){
+ const int system_tag = COLL_TAG_ALLTOALLV;
+ MPI_Request *requests;
+
+ /* Initialize. */
+ int rank = comm->rank();
+ int size = comm->size();
+ (*request) = new Request( nullptr, 0, MPI_BYTE,
+ rank,rank, system_tag, comm, MPI_REQ_PERSISTENT);
+ /* Local copy from self */
+ int err = (sendcounts[rank]>0 && recvcounts[rank]) ? Datatype::copy(static_cast<char *>(sendbuf) + senddisps[rank], sendcounts[rank], sendtypes[rank],
+ static_cast<char *>(recvbuf) + recvdisps[rank], recvcounts[rank], recvtypes[rank]): MPI_SUCCESS;
+ if (err == MPI_SUCCESS && size > 1) {
+ /* Initiate all send/recv to/from others. */
+ requests = new MPI_Request[2 * (size - 1)];
+ int count = 0;
+ /* Create all receives that will be posted first */
+ for (int i = 0; i < size; ++i) {
+ if (i != rank) {
+ requests[count] = Request::irecv_init(static_cast<char *>(recvbuf) + recvdisps[i],
+ recvcounts[i], recvtypes[i], i, system_tag, comm);
+ count++;
+ }else{
+ XBT_DEBUG("<%d> skip request creation [src = %d, recvcounts[src] = %d]", rank, i, recvcounts[i]);
+ }
+ }
+ /* Now create all sends */
+ for (int i = 0; i < size; ++i) {
+ if (i != rank) {
+ requests[count] = Request::isend_init(static_cast<char *>(sendbuf) + senddisps[i] ,
+ sendcounts[i], sendtypes[i], i, system_tag, comm);
+ count++;
+ }else{
+ XBT_DEBUG("<%d> skip request creation [dst = %d, sendcounts[dst] = %d]", rank, i, sendcounts[i]);
+ }
+ }
+ /* Wait for them all. */
+ Request::startall(count, requests);
+ (*request)->set_nbc_requests(requests, count);
+ }
+ return err;
+}
+
int Colls::igather(void *sendbuf, int sendcount, MPI_Datatype sendtype,
void *recvbuf, int recvcount, MPI_Datatype recvtype, int root, MPI_Comm comm, MPI_Request *request)
{
int rank = comm->rank();
int size = comm->size();
(*request) = new Request( nullptr, 0, MPI_BYTE,
- rank,rank, COLL_TAG_GATHER, comm, MPI_REQ_PERSISTENT);
+ rank,rank, system_tag, comm, MPI_REQ_PERSISTENT);
if(rank != root) {
// Send buffer to root
requests = new MPI_Request[1];
int rank = comm->rank();
int size = comm->size();
(*request) = new Request( nullptr, 0, MPI_BYTE,
- rank,rank, COLL_TAG_GATHERV, comm, MPI_REQ_PERSISTENT);
+ rank,rank, system_tag, comm, MPI_REQ_PERSISTENT);
if (rank != root) {
// Send buffer to root
requests = new MPI_Request[1];
int rank = comm->rank();
int size = comm->size();
(*request) = new Request( nullptr, 0, MPI_BYTE,
- rank,rank, COLL_TAG_SCATTERV, comm, MPI_REQ_PERSISTENT);
+ rank,rank, system_tag, comm, MPI_REQ_PERSISTENT);
if(rank != root) {
// Recv buffer from root
requests = new MPI_Request[1];
}
return MPI_SUCCESS;
}
+
+int Colls::ireduce(void *sendbuf, void *recvbuf, int count, MPI_Datatype datatype, MPI_Op op, int root,
+ MPI_Comm comm, MPI_Request* request)
+{
+ const int system_tag = COLL_TAG_REDUCE;
+ MPI_Aint lb = 0;
+ MPI_Aint dataext = 0;
+ MPI_Request* requests;
+
+ char* sendtmpbuf = static_cast<char *>(sendbuf);
+
+ int rank = comm->rank();
+ int size = comm->size();
+
+ if (size <= 0)
+ return MPI_ERR_COMM;
+
+ if( sendbuf == MPI_IN_PLACE ) {
+ sendtmpbuf = static_cast<char *>(smpi_get_tmp_sendbuffer(count*datatype->get_extent()));
+ Datatype::copy(recvbuf, count, datatype,sendtmpbuf, count, datatype);
+ }
+
+ if(rank == root){
+ (*request) = new Request( recvbuf, count, datatype,
+ rank,rank, system_tag, comm, MPI_REQ_PERSISTENT, op);
+ }
+ else
+ (*request) = new Request( nullptr, count, datatype,
+ rank,rank, system_tag, comm, MPI_REQ_PERSISTENT);
+
+ if(rank != root) {
+ // Send buffer to root
+ requests = new MPI_Request[1];
+ requests[0]=Request::isend(sendtmpbuf, count, datatype, root, system_tag, comm);
+ (*request)->set_nbc_requests(requests, 1);
+ } else {
+ datatype->extent(&lb, &dataext);
+ // Local copy from root
+ if (sendtmpbuf != nullptr && recvbuf != nullptr)
+ Datatype::copy(sendtmpbuf, count, datatype, recvbuf, count, datatype);
+ // Receive buffers from senders
+ MPI_Request *requests = new MPI_Request[size - 1];
+ int index = 0;
+ for (int src = 0; src < size; src++) {
+ if (src != root) {
+ requests[index] =
+ Request::irecv_init(smpi_get_tmp_sendbuffer(count * dataext), count, datatype, src, system_tag, comm);
+ index++;
+ }
+ }
+ // Wait for completion of irecv's.
+ Request::startall(size - 1, requests);
+ (*request)->set_nbc_requests(requests, size - 1);
+ }
+ if( sendbuf == MPI_IN_PLACE ) {
+ smpi_free_tmp_buffer(sendtmpbuf);
+ }
+ return MPI_SUCCESS;
+}
+
+int Colls::iallreduce(void *sendbuf, void *recvbuf, int count, MPI_Datatype datatype,
+ MPI_Op op, MPI_Comm comm, MPI_Request* request)
+{
+
+ const int system_tag = COLL_TAG_ALLREDUCE;
+ MPI_Aint lb = 0;
+ MPI_Aint dataext = 0;
+ MPI_Request *requests;
+
+ int rank = comm->rank();
+ int size = comm->size();
+ (*request) = new Request( recvbuf, count, datatype,
+ rank,rank, system_tag, comm, MPI_REQ_PERSISTENT, op);
+ // FIXME: check for errors
+ datatype->extent(&lb, &dataext);
+ // Local copy from self
+ Datatype::copy(sendbuf, count, datatype, recvbuf, count, datatype);
+ // Send/Recv buffers to/from others;
+ requests = new MPI_Request[2 * (size - 1)];
+ int index = 0;
+ for (int other = 0; other < size; other++) {
+ if(other != rank) {
+ requests[index] = Request::isend_init(sendbuf, count, datatype, other, system_tag,comm);
+ index++;
+ requests[index] = Request::irecv_init(smpi_get_tmp_sendbuffer(count * dataext), count, datatype,
+ other, system_tag, comm);
+ index++;
+ }
+ }
+ Request::startall(2 * (size - 1), requests);
+ (*request)->set_nbc_requests(requests, 2 * (size - 1));
+ return MPI_SUCCESS;
+}
+
+int Colls::iscan(void *sendbuf, void *recvbuf, int count, MPI_Datatype datatype, MPI_Op op, MPI_Comm comm, MPI_Request* request)
+{
+ int system_tag = -888;
+ MPI_Aint lb = 0;
+ MPI_Aint dataext = 0;
+
+ int rank = comm->rank();
+ int size = comm->size();
+ (*request) = new Request( recvbuf, count, datatype,
+ rank,rank, system_tag, comm, MPI_REQ_PERSISTENT, op);
+ datatype->extent(&lb, &dataext);
+
+ // Local copy from self
+ Datatype::copy(sendbuf, count, datatype, recvbuf, count, datatype);
+
+ // Send/Recv buffers to/from others
+ MPI_Request *requests = new MPI_Request[size - 1];
+ void **tmpbufs = xbt_new(void *, rank);
+ int index = 0;
+ for (int other = 0; other < rank; other++) {
+ tmpbufs[index] = smpi_get_tmp_sendbuffer(count * dataext);
+ requests[index] = Request::irecv_init(tmpbufs[index], count, datatype, other, system_tag, comm);
+ index++;
+ }
+ for (int other = rank + 1; other < size; other++) {
+ requests[index] = Request::isend_init(sendbuf, count, datatype, other, system_tag, comm);
+ index++;
+ }
+ // Wait for completion of all comms.
+ Request::startall(size - 1, requests);
+ (*request)->set_nbc_requests(requests, size - 1);
+ return MPI_SUCCESS;
+}
+
+int Colls::iexscan(void *sendbuf, void *recvbuf, int count, MPI_Datatype datatype, MPI_Op op, MPI_Comm comm, MPI_Request* request)
+{
+ int system_tag = -888;
+ MPI_Aint lb = 0;
+ MPI_Aint dataext = 0;
+ int rank = comm->rank();
+ int size = comm->size();
+ (*request) = new Request( recvbuf, count, datatype,
+ rank,rank, system_tag, comm, MPI_REQ_PERSISTENT, op);
+ datatype->extent(&lb, &dataext);
+ if(rank != 0)
+ memset(recvbuf, 0, count*dataext);
+
+ // Send/Recv buffers to/from others
+ MPI_Request *requests = new MPI_Request[size - 1];
+ void **tmpbufs = xbt_new(void *, rank);
+ int index = 0;
+ for (int other = 0; other < rank; other++) {
+ tmpbufs[index] = smpi_get_tmp_sendbuffer(count * dataext);
+ requests[index] = Request::irecv_init(tmpbufs[index], count, datatype, other, system_tag, comm);
+ index++;
+ }
+ for (int other = rank + 1; other < size; other++) {
+ requests[index] = Request::isend_init(sendbuf, count, datatype, other, system_tag, comm);
+ index++;
+ }
+ // Wait for completion of all comms.
+ Request::startall(size - 1, requests);
+ (*request)->set_nbc_requests(requests, size - 1);
+ return MPI_SUCCESS;
+}
+
+int Colls::ireduce_scatter(void *sendbuf, void *recvbuf, int *recvcounts, MPI_Datatype datatype, MPI_Op op,
+ MPI_Comm comm, MPI_Request* request){
+//Version where each process performs the reduce for its own part. Alltoall pattern for comms.
+ const int system_tag = COLL_TAG_REDUCE_SCATTER;
+ MPI_Aint lb = 0;
+ MPI_Aint dataext = 0;
+ MPI_Request *requests;
+
+ int rank = comm->rank();
+ int size = comm->size();
+ int count=recvcounts[rank];
+ (*request) = new Request( recvbuf, count, datatype,
+ rank,rank, system_tag, comm, MPI_REQ_PERSISTENT, op);
+ datatype->extent(&lb, &dataext);
+
+ // Send/Recv buffers to/from others;
+ requests = new MPI_Request[2 * (size - 1)];
+ int index = 0;
+ int recvdisp=0;
+ for (int other = 0; other < size; other++) {
+ if(other != rank) {
+ requests[index] = Request::isend_init(static_cast<char *>(sendbuf) + recvdisp * dataext, recvcounts[other], datatype, other, system_tag,comm);
+ XBT_VERB("sending with recvdisp %d", recvdisp);
+ index++;
+ requests[index] = Request::irecv_init(smpi_get_tmp_sendbuffer(count * dataext), count, datatype,
+ other, system_tag, comm);
+ index++;
+ }else{
+ Datatype::copy(static_cast<char *>(sendbuf) + recvdisp * dataext, count, datatype, recvbuf, count, datatype);
+ }
+ recvdisp+=recvcounts[other];
+ }
+ Request::startall(2 * (size - 1), requests);
+ (*request)->set_nbc_requests(requests, 2 * (size - 1));
+ return MPI_SUCCESS;
+}
+
}
}
