From 67a025ec7d8a8248222a4b6bbe9cbdfe4ed4439c Mon Sep 17 00:00:00 2001 From: Augustin Degomme Date: Fri, 7 May 2021 18:14:52 +0200 Subject: [PATCH] non blocking collectives : array->vector --- src/smpi/colls/smpi_default_selector.cpp | 11 +- src/smpi/colls/smpi_nbc_impl.cpp | 250 ++++++++--------------- src/smpi/include/smpi_request.hpp | 8 +- src/smpi/mpi/smpi_request.cpp | 41 ++-- 4 files changed, 107 insertions(+), 203 deletions(-) diff --git a/src/smpi/colls/smpi_default_selector.cpp b/src/smpi/colls/smpi_default_selector.cpp index 03c7d542c7..d5f5aac804 100644 --- a/src/smpi/colls/smpi_default_selector.cpp +++ b/src/smpi/colls/smpi_default_selector.cpp @@ -77,13 +77,10 @@ int allgatherv__default(const void *sendbuf, int sendcount, MPI_Datatype sendtyp { MPI_Request request; colls::iallgatherv(sendbuf, sendcount, sendtype, recvbuf, recvcounts, displs, recvtype, comm, &request, 0); - MPI_Request* requests = request->get_nbc_requests(); - int count = request->get_nbc_requests_size(); - Request::waitall(count, requests, MPI_STATUS_IGNORE); - for (int other = 0; other < count; other++) { - Request::unref(&requests[other]); - } - delete[] requests; + auto requests = request->get_nbc_requests(); + Request::waitall(requests.size(), &requests[0], MPI_STATUS_IGNORE); + for(auto& req: requests) + Request::unref(&req); Request::unref(&request); return MPI_SUCCESS; } diff --git a/src/smpi/colls/smpi_nbc_impl.cpp b/src/smpi/colls/smpi_nbc_impl.cpp index f3b4a4672f..a6aa5277af 100644 --- a/src/smpi/colls/smpi_nbc_impl.cpp +++ b/src/smpi/colls/smpi_nbc_impl.cpp @@ -18,24 +18,19 @@ int colls::ibarrier(MPI_Comm comm, MPI_Request* request, int external) int system_tag=COLL_TAG_BARRIER-external; (*request) = new Request( nullptr, 0, MPI_BYTE, rank,rank, system_tag, comm, MPI_REQ_PERSISTENT|MPI_REQ_NBC); + std::vector requests; + if (rank > 0) { - auto* requests = new MPI_Request[2]; - requests[0] = Request::isend (nullptr, 0, MPI_BYTE, 0, - system_tag, - comm); - requests[1] = Request::irecv (nullptr, 0, MPI_BYTE, 0, - system_tag, - comm); - (*request)->set_nbc_requests(requests, 2); + requests.push_back(Request::isend_init (nullptr, 0, MPI_BYTE, 0, system_tag, comm)); + requests.push_back(Request::irecv_init(nullptr, 0, MPI_BYTE, 0, system_tag, comm)); } else { - auto* requests = new MPI_Request[(size - 1) * 2]; for (int i = 1; i < 2 * size - 1; i += 2) { - requests[i - 1] = Request::irecv(nullptr, 0, MPI_BYTE, MPI_ANY_SOURCE, system_tag, comm); - requests[i] = Request::isend(nullptr, 0, MPI_BYTE, (i + 1) / 2, system_tag, comm); + requests.push_back(Request::irecv_init(nullptr, 0, MPI_BYTE, MPI_ANY_SOURCE, system_tag, comm)); + requests.push_back(Request::isend_init(nullptr, 0, MPI_BYTE, (i + 1) / 2, system_tag, comm)); } - (*request)->set_nbc_requests(requests, 2*(size-1)); } + (*request)->start_nbc_requests(requests); return MPI_SUCCESS; } @@ -45,29 +40,20 @@ int colls::ibcast(void* buf, int count, MPI_Datatype datatype, int root, MPI_Com int size = comm->size(); int rank = comm->rank(); int system_tag=COLL_TAG_BCAST-external; + std::vector requests; (*request) = new Request( nullptr, 0, MPI_BYTE, rank,rank, system_tag, comm, MPI_REQ_PERSISTENT|MPI_REQ_NBC); if (rank != root) { - auto* requests = new MPI_Request[1]; - requests[0] = Request::irecv (buf, count, datatype, root, - system_tag, - comm); - (*request)->set_nbc_requests(requests, 1); + requests.push_back(Request::irecv_init(buf, count, datatype, root, system_tag, comm)); } else { - auto* requests = new MPI_Request[size - 1]; - int n = 0; for (int i = 0; i < size; i++) { if(i!=root){ - requests[n] = Request::isend(buf, count, datatype, i, - system_tag, - comm - ); - n++; + requests.push_back(Request::isend_init(buf, count, datatype, i, system_tag, comm)); } } - (*request)->set_nbc_requests(requests, size-1); } + (*request)->start_nbc_requests(requests); return MPI_SUCCESS; } @@ -78,6 +64,7 @@ int colls::iallgather(const void* sendbuf, int sendcount, MPI_Datatype sendtype, const int system_tag = COLL_TAG_ALLGATHER-external; MPI_Aint lb = 0; MPI_Aint recvext = 0; + std::vector requests; int rank = comm->rank(); int size = comm->size(); @@ -89,19 +76,14 @@ int colls::iallgather(const void* sendbuf, int sendcount, MPI_Datatype sendtype, Datatype::copy(sendbuf, sendcount, sendtype, static_cast(recvbuf) + rank * recvcount * recvext, recvcount, recvtype); // Send/Recv buffers to/from others; - auto* 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, sendcount, sendtype, other, system_tag,comm); - index++; - requests[index] = Request::irecv_init(static_cast(recvbuf) + other * recvcount * recvext, recvcount, recvtype, - other, system_tag, comm); - index++; + requests.push_back(Request::isend_init(sendbuf, sendcount, sendtype, other, system_tag, comm)); + requests.push_back(Request::irecv_init(static_cast(recvbuf) + other * recvcount * recvext, + recvcount, recvtype, other, system_tag, comm)); } } - Request::startall(2 * (size - 1), requests); - (*request)->set_nbc_requests(requests, 2 * (size - 1)); + (*request)->start_nbc_requests(requests); return MPI_SUCCESS; } @@ -111,16 +93,15 @@ int colls::iscatter(const void* sendbuf, int sendcount, MPI_Datatype sendtype, v const int system_tag = COLL_TAG_SCATTER-external; MPI_Aint lb = 0; MPI_Aint sendext = 0; + std::vector requests; int rank = comm->rank(); int size = comm->size(); (*request) = new Request( nullptr, 0, MPI_BYTE, rank,rank, system_tag, comm, MPI_REQ_PERSISTENT|MPI_REQ_NBC); if(rank != root) { - auto* requests = new MPI_Request[1]; // Recv buffer from root - requests[0] = Request::irecv(recvbuf, recvcount, recvtype, root, system_tag, comm); - (*request)->set_nbc_requests(requests, 1); + requests.push_back(Request::irecv_init(recvbuf, recvcount, recvtype, root, system_tag, comm)); } else { sendtype->extent(&lb, &sendext); // Local copy from root @@ -129,19 +110,14 @@ int colls::iscatter(const void* sendbuf, int sendcount, MPI_Datatype sendtype, v sendcount, sendtype, recvbuf, recvcount, recvtype); } // Send buffers to receivers - auto* requests = new MPI_Request[size - 1]; - int index = 0; for(int dst = 0; dst < size; dst++) { if(dst != root) { - requests[index] = Request::isend_init(static_cast(sendbuf) + dst * sendcount * sendext, sendcount, sendtype, - dst, system_tag, comm); - index++; + requests.push_back(Request::isend_init(static_cast(sendbuf) + dst * sendcount * sendext, sendcount, sendtype, + dst, system_tag, comm)); } } - // Wait for completion of isend's. - Request::startall(size - 1, requests); - (*request)->set_nbc_requests(requests, size - 1); } + (*request)->start_nbc_requests(requests); return MPI_SUCCESS; } @@ -151,6 +127,7 @@ int colls::iallgatherv(const void* sendbuf, int sendcount, MPI_Datatype sendtype const int system_tag = COLL_TAG_ALLGATHERV-external; MPI_Aint lb = 0; MPI_Aint recvext = 0; + std::vector requests; int rank = comm->rank(); int size = comm->size(); @@ -161,21 +138,15 @@ int colls::iallgatherv(const void* sendbuf, int sendcount, MPI_Datatype sendtype Datatype::copy(sendbuf, sendcount, sendtype, static_cast(recvbuf) + displs[rank] * recvext,recvcounts[rank], recvtype); // Send buffers to others; - auto* 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, sendcount, sendtype, other, system_tag, comm); - index++; - requests[index] = Request::irecv_init(static_cast(recvbuf) + displs[other] * recvext, recvcounts[other], - recvtype, other, system_tag, comm); - index++; + requests.push_back(Request::isend_init(sendbuf, sendcount, sendtype, other, system_tag, comm)); + requests.push_back(Request::irecv_init(static_cast(recvbuf) + displs[other] * recvext, recvcounts[other], + recvtype, other, system_tag, comm)); } } // Wait for completion of all comms. - Request::startall(2 * (size - 1), requests); - (*request)->set_nbc_requests(requests, 2 * (size - 1)); + (*request)->start_nbc_requests(requests); return MPI_SUCCESS; } @@ -186,6 +157,7 @@ int colls::ialltoall(const void* sendbuf, int sendcount, MPI_Datatype sendtype, MPI_Aint lb = 0; MPI_Aint sendext = 0; MPI_Aint recvext = 0; + std::vector requests; /* Initialize. */ int rank = comm->rank(); @@ -199,13 +171,10 @@ int colls::ialltoall(const void* sendbuf, int sendcount, MPI_Datatype sendtype, static_cast(recvbuf) + rank * recvcount * recvext, recvcount, recvtype); if (err == MPI_SUCCESS && size > 1) { /* Initiate all send/recv to/from others. */ - auto* requests = new MPI_Request[2 * (size - 1)]; /* Post all receives first -- a simple optimization */ - int count = 0; for (int i = (rank + 1) % size; i != rank; i = (i + 1) % size) { - requests[count] = Request::irecv_init(static_cast(recvbuf) + i * recvcount * recvext, recvcount, - recvtype, i, system_tag, comm); - count++; + requests.push_back(Request::irecv_init(static_cast(recvbuf) + i * recvcount * recvext, recvcount, + recvtype, i, system_tag, comm)); } /* Now post all sends in reverse order * - We would like to minimize the search time through message queue @@ -213,13 +182,11 @@ int colls::ialltoall(const void* sendbuf, int sendcount, MPI_Datatype sendtype, * TODO: check the previous assertion */ for (int i = (rank + size - 1) % size; i != rank; i = (i + size - 1) % size) { - requests[count] = Request::isend_init(static_cast(sendbuf) + i * sendcount * sendext, sendcount, - sendtype, i, system_tag, comm); - count++; + requests.push_back(Request::isend_init(static_cast(sendbuf) + i * sendcount * sendext, sendcount, + sendtype, i, system_tag, comm)); } /* Wait for them all. */ - Request::startall(count, requests); - (*request)->set_nbc_requests(requests, count); + (*request)->start_nbc_requests(requests); } return MPI_SUCCESS; } @@ -232,6 +199,7 @@ int colls::ialltoallv(const void* sendbuf, const int* sendcounts, const int* sen MPI_Aint lb = 0; MPI_Aint sendext = 0; MPI_Aint recvext = 0; + std::vector requests; /* Initialize. */ int rank = comm->rank(); @@ -245,14 +213,11 @@ int colls::ialltoallv(const void* sendbuf, const int* sendcounts, const int* sen static_cast(recvbuf) + recvdisps[rank] * recvext, recvcounts[rank], recvtype); if (err == MPI_SUCCESS && size > 1) { /* Initiate all send/recv to/from others. */ - auto* 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(recvbuf) + recvdisps[i] * recvext, - recvcounts[i], recvtype, i, system_tag, comm); - count++; + requests.push_back(Request::irecv_init(static_cast(recvbuf) + recvdisps[i] * recvext, + recvcounts[i], recvtype, i, system_tag, comm)); }else{ XBT_DEBUG("<%d> skip request creation [src = %d, recvcounts[src] = %d]", rank, i, recvcounts[i]); } @@ -260,16 +225,14 @@ int colls::ialltoallv(const void* sendbuf, const int* sendcounts, const int* sen /* Now create all sends */ for (int i = 0; i < size; ++i) { if (i != rank) { - requests[count] = Request::isend_init(static_cast(sendbuf) + senddisps[i] * sendext, - sendcounts[i], sendtype, i, system_tag, comm); - count++; + requests.push_back(Request::isend_init(static_cast(sendbuf) + senddisps[i] * sendext, + sendcounts[i], sendtype, i, system_tag, comm)); }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); + (*request)->start_nbc_requests(requests); } return err; } @@ -279,6 +242,7 @@ int colls::ialltoallw(const void* sendbuf, const int* sendcounts, const int* sen MPI_Comm comm, MPI_Request* request, int external) { const int system_tag = COLL_TAG_ALLTOALLW-external; + std::vector requests; /* Initialize. */ int rank = comm->rank(); @@ -290,14 +254,11 @@ int colls::ialltoallw(const void* sendbuf, const int* sendcounts, const int* sen static_cast(recvbuf) + recvdisps[rank], recvcounts[rank], recvtypes[rank]): MPI_SUCCESS; if (err == MPI_SUCCESS && size > 1) { /* Initiate all send/recv to/from others. */ - auto* 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(recvbuf) + recvdisps[i], - recvcounts[i], recvtypes[i], i, system_tag, comm); - count++; + requests.push_back(Request::irecv_init(static_cast(recvbuf) + recvdisps[i], + recvcounts[i], recvtypes[i], i, system_tag, comm)); }else{ XBT_DEBUG("<%d> skip request creation [src = %d, recvcounts[src] = %d]", rank, i, recvcounts[i]); } @@ -305,16 +266,14 @@ int colls::ialltoallw(const void* sendbuf, const int* sendcounts, const int* sen /* Now create all sends */ for (int i = 0; i < size; ++i) { if (i != rank) { - requests[count] = Request::isend_init(static_cast(sendbuf) + senddisps[i] , - sendcounts[i], sendtypes[i], i, system_tag, comm); - count++; + requests.push_back(Request::isend_init(static_cast(sendbuf) + senddisps[i] , + sendcounts[i], sendtypes[i], i, system_tag, comm)); }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); + (*request)->start_nbc_requests(requests); } return err; } @@ -325,6 +284,7 @@ int colls::igather(const void* sendbuf, int sendcount, MPI_Datatype sendtype, vo const int system_tag = COLL_TAG_GATHER-external; MPI_Aint lb = 0; MPI_Aint recvext = 0; + std::vector requests; int rank = comm->rank(); int size = comm->size(); @@ -332,28 +292,21 @@ int colls::igather(const void* sendbuf, int sendcount, MPI_Datatype sendtype, vo rank,rank, system_tag, comm, MPI_REQ_PERSISTENT|MPI_REQ_NBC); if(rank != root) { // Send buffer to root - auto* requests = new MPI_Request[1]; - requests[0]=Request::isend(sendbuf, sendcount, sendtype, root, system_tag, comm); - (*request)->set_nbc_requests(requests, 1); + requests.push_back(Request::isend_init(sendbuf, sendcount, sendtype, root, system_tag, comm)); } else { recvtype->extent(&lb, &recvext); // Local copy from root Datatype::copy(sendbuf, sendcount, sendtype, static_cast(recvbuf) + root * recvcount * recvext, recvcount, recvtype); // Receive buffers from senders - auto* requests = new MPI_Request[size - 1]; - int index = 0; for (int src = 0; src < size; src++) { if(src != root) { - requests[index] = Request::irecv_init(static_cast(recvbuf) + src * recvcount * recvext, recvcount, recvtype, - src, system_tag, comm); - index++; + requests.push_back(Request::irecv_init(static_cast(recvbuf) + src * recvcount * recvext, recvcount, recvtype, + src, system_tag, comm)); } } - // Wait for completion of irecv's. - Request::startall(size - 1, requests); - (*request)->set_nbc_requests(requests, size - 1); } + (*request)->start_nbc_requests(requests); return MPI_SUCCESS; } @@ -364,35 +317,30 @@ int colls::igatherv(const void* sendbuf, int sendcount, MPI_Datatype sendtype, v int system_tag = COLL_TAG_GATHERV-external; MPI_Aint lb = 0; MPI_Aint recvext = 0; - + std::vector requests; + int rank = comm->rank(); int size = comm->size(); (*request) = new Request( nullptr, 0, MPI_BYTE, rank,rank, system_tag, comm, MPI_REQ_PERSISTENT|MPI_REQ_NBC); if (rank != root) { // Send buffer to root - auto* requests = new MPI_Request[1]; - requests[0]=Request::isend(sendbuf, sendcount, sendtype, root, system_tag, comm); - (*request)->set_nbc_requests(requests, 1); + requests.push_back(Request::isend_init(sendbuf, sendcount, sendtype, root, system_tag, comm)); } else { recvtype->extent(&lb, &recvext); // Local copy from root Datatype::copy(sendbuf, sendcount, sendtype, static_cast(recvbuf) + displs[root] * recvext, recvcounts[root], recvtype); // Receive buffers from senders - auto* requests = new MPI_Request[size - 1]; - int index = 0; for (int src = 0; src < size; src++) { if(src != root) { - requests[index] = Request::irecv_init(static_cast(recvbuf) + displs[src] * recvext, - recvcounts[src], recvtype, src, system_tag, comm); - index++; + requests.push_back(Request::irecv_init(static_cast(recvbuf) + displs[src] * recvext, + recvcounts[src], recvtype, src, system_tag, comm)); } } - // Wait for completion of irecv's. - Request::startall(size - 1, requests); - (*request)->set_nbc_requests(requests, size - 1); } + // Wait for completion of irecv's. + (*request)->start_nbc_requests(requests); return MPI_SUCCESS; } int colls::iscatterv(const void* sendbuf, const int* sendcounts, const int* displs, MPI_Datatype sendtype, @@ -402,6 +350,7 @@ int colls::iscatterv(const void* sendbuf, const int* sendcounts, const int* disp int system_tag = COLL_TAG_SCATTERV-external; MPI_Aint lb = 0; MPI_Aint sendext = 0; + std::vector requests; int rank = comm->rank(); int size = comm->size(); @@ -409,9 +358,7 @@ int colls::iscatterv(const void* sendbuf, const int* sendcounts, const int* disp rank,rank, system_tag, comm, MPI_REQ_PERSISTENT|MPI_REQ_NBC); if(rank != root) { // Recv buffer from root - auto* requests = new MPI_Request[1]; - requests[0]=Request::irecv(recvbuf, recvcount, recvtype, root, system_tag, comm); - (*request)->set_nbc_requests(requests, 1); + requests.push_back(Request::irecv_init(recvbuf, recvcount, recvtype, root, system_tag, comm)); } else { sendtype->extent(&lb, &sendext); // Local copy from root @@ -420,19 +367,14 @@ int colls::iscatterv(const void* sendbuf, const int* sendcounts, const int* disp sendtype, recvbuf, recvcount, recvtype); } // Send buffers to receivers - auto* requests = new MPI_Request[size - 1]; - int index = 0; for (int dst = 0; dst < size; dst++) { if (dst != root) { - requests[index] = Request::isend_init(static_cast(sendbuf) + displs[dst] * sendext, sendcounts[dst], - sendtype, dst, system_tag, comm); - index++; + requests.push_back(Request::isend_init(static_cast(sendbuf) + displs[dst] * sendext, sendcounts[dst], + sendtype, dst, system_tag, comm)); } } - // Wait for completion of isend's. - Request::startall(size - 1, requests); - (*request)->set_nbc_requests(requests, size - 1); } + (*request)->start_nbc_requests(requests); return MPI_SUCCESS; } @@ -442,6 +384,7 @@ int colls::ireduce(const void* sendbuf, void* recvbuf, int count, MPI_Datatype d const int system_tag = COLL_TAG_REDUCE-external; MPI_Aint lb = 0; MPI_Aint dataext = 0; + std::vector requests; const void* real_sendbuf = sendbuf; @@ -468,28 +411,20 @@ int colls::ireduce(const void* sendbuf, void* recvbuf, int count, MPI_Datatype d if(rank != root) { // Send buffer to root - auto* requests = new MPI_Request[1]; - requests[0] = Request::isend(real_sendbuf, count, datatype, root, system_tag, comm); - (*request)->set_nbc_requests(requests, 1); + requests.push_back(Request::isend_init(real_sendbuf, count, datatype, root, system_tag, comm)); } else { datatype->extent(&lb, &dataext); // Local copy from root if (real_sendbuf != nullptr && recvbuf != nullptr) Datatype::copy(real_sendbuf, count, datatype, recvbuf, count, datatype); // Receive buffers from senders - auto* 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++; + requests.push_back(Request::irecv_init(smpi_get_tmp_sendbuffer(count * dataext), count, datatype, src, system_tag, comm)); } } - // Wait for completion of irecv's. - Request::startall(size - 1, requests); - (*request)->set_nbc_requests(requests, size - 1); - } + } + (*request)->start_nbc_requests(requests); if( sendbuf == MPI_IN_PLACE ) { smpi_free_tmp_buffer(tmp_sendbuf); } @@ -503,6 +438,7 @@ int colls::iallreduce(const void* sendbuf, void* recvbuf, int count, MPI_Datatyp const int system_tag = COLL_TAG_ALLREDUCE-external; MPI_Aint lb = 0; MPI_Aint dataext = 0; + std::vector requests; int rank = comm->rank(); int size = comm->size(); @@ -513,19 +449,14 @@ int colls::iallreduce(const void* sendbuf, void* recvbuf, int count, MPI_Datatyp // Local copy from self Datatype::copy(sendbuf, count, datatype, recvbuf, count, datatype); // Send/Recv buffers to/from others; - auto* 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++; + requests.push_back(Request::isend_init(sendbuf, count, datatype, other, system_tag,comm)); + requests.push_back(Request::irecv_init(smpi_get_tmp_sendbuffer(count * dataext), count, datatype, + other, system_tag, comm)); } } - Request::startall(2 * (size - 1), requests); - (*request)->set_nbc_requests(requests, 2 * (size - 1)); + (*request)->start_nbc_requests(requests); return MPI_SUCCESS; } @@ -535,6 +466,7 @@ int colls::iscan(const void* sendbuf, void* recvbuf, int count, MPI_Datatype dat int system_tag = -888-external; MPI_Aint lb = 0; MPI_Aint dataext = 0; + std::vector requests; int rank = comm->rank(); int size = comm->size(); @@ -546,19 +478,14 @@ int colls::iscan(const void* sendbuf, void* recvbuf, int count, MPI_Datatype dat Datatype::copy(sendbuf, count, datatype, recvbuf, count, datatype); // Send/Recv buffers to/from others - auto* requests = new MPI_Request[size - 1]; - int index = 0; for (int other = 0; other < rank; other++) { - requests[index] = Request::irecv_init(smpi_get_tmp_sendbuffer(count * dataext), count, datatype, other, system_tag, comm); - index++; + requests.push_back(Request::irecv_init(smpi_get_tmp_sendbuffer(count * dataext), count, datatype, other, system_tag, comm)); } for (int other = rank + 1; other < size; other++) { - requests[index] = Request::isend_init(sendbuf, count, datatype, other, system_tag, comm); - index++; + requests.push_back(Request::isend_init(sendbuf, count, datatype, other, system_tag, comm)); } // Wait for completion of all comms. - Request::startall(size - 1, requests); - (*request)->set_nbc_requests(requests, size - 1); + (*request)->start_nbc_requests(requests); return MPI_SUCCESS; } @@ -568,6 +495,8 @@ int colls::iexscan(const void* sendbuf, void* recvbuf, int count, MPI_Datatype d int system_tag = -888-external; MPI_Aint lb = 0; MPI_Aint dataext = 0; + std::vector requests; + int rank = comm->rank(); int size = comm->size(); (*request) = new Request( recvbuf, count, datatype, @@ -577,19 +506,14 @@ int colls::iexscan(const void* sendbuf, void* recvbuf, int count, MPI_Datatype d memset(recvbuf, 0, count*dataext); // Send/Recv buffers to/from others - auto* requests = new MPI_Request[size - 1]; - int index = 0; for (int other = 0; other < rank; other++) { - requests[index] = Request::irecv_init(smpi_get_tmp_sendbuffer(count * dataext), count, datatype, other, system_tag, comm); - index++; + requests.push_back(Request::irecv_init(smpi_get_tmp_sendbuffer(count * dataext), count, datatype, other, system_tag, comm)); } for (int other = rank + 1; other < size; other++) { - requests[index] = Request::isend_init(sendbuf, count, datatype, other, system_tag, comm); - index++; + requests.push_back(Request::isend_init(sendbuf, count, datatype, other, system_tag, comm)); } // Wait for completion of all comms. - Request::startall(size - 1, requests); - (*request)->set_nbc_requests(requests, size - 1); + (*request)->start_nbc_requests(requests); return MPI_SUCCESS; } @@ -600,6 +524,7 @@ int colls::ireduce_scatter(const void* sendbuf, void* recvbuf, const int* recvco const int system_tag = COLL_TAG_REDUCE_SCATTER-external; MPI_Aint lb = 0; MPI_Aint dataext = 0; + std::vector requests; int rank = comm->rank(); int size = comm->size(); @@ -609,24 +534,19 @@ int colls::ireduce_scatter(const void* sendbuf, void* recvbuf, const int* recvco datatype->extent(&lb, &dataext); // Send/Recv buffers to/from others; - auto* 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(sendbuf) + recvdisp * dataext, recvcounts[other], datatype, other, system_tag,comm); + requests.push_back(Request::isend_init(static_cast(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++; + requests.push_back(Request::irecv_init(smpi_get_tmp_sendbuffer(count * dataext), count, datatype, + other, system_tag, comm)); }else{ Datatype::copy(static_cast(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)); + (*request)->start_nbc_requests(requests); return MPI_SUCCESS; } diff --git a/src/smpi/include/smpi_request.hpp b/src/smpi/include/smpi_request.hpp index 19bc383d52..547d93bdaa 100644 --- a/src/smpi/include/smpi_request.hpp +++ b/src/smpi/include/smpi_request.hpp @@ -49,8 +49,7 @@ class Request : public F2C { int refcount_; MPI_Op op_; std::unique_ptr generalized_funcs; - MPI_Request* nbc_requests_; - int nbc_requests_size_; + std::vector nbc_requests_; static bool match_common(MPI_Request req, MPI_Request sender, MPI_Request receiver); public: @@ -72,10 +71,9 @@ public: void cancel(); void init_buffer(int count); void ref(); - void set_nbc_requests(MPI_Request* reqs, int size); + void start_nbc_requests(std::vector reqs); static int finish_nbc_requests(MPI_Request* req, int test); - int get_nbc_requests_size() const; - MPI_Request* get_nbc_requests() const; + std::vector get_nbc_requests() const; static void finish_wait(MPI_Request* request, MPI_Status* status); static void unref(MPI_Request* request); static int wait(MPI_Request* req, MPI_Status* status); diff --git a/src/smpi/mpi/smpi_request.cpp b/src/smpi/mpi/smpi_request.cpp index a5f4ea84a4..f6783a93f1 100644 --- a/src/smpi/mpi/smpi_request.cpp +++ b/src/smpi/mpi/smpi_request.cpp @@ -63,8 +63,6 @@ Request::Request(const void* buf, int count, MPI_Datatype datatype, aid_t src, a refcount_ = 1; else refcount_ = 0; - nbc_requests_=nullptr; - nbc_requests_size_=0; init_buffer(count); this->add_f(); } @@ -856,20 +854,20 @@ int Request::finish_nbc_requests(MPI_Request* request, int test){ int flag = 1; int ret = 0; if(test == 0) - ret = waitall((*request)->nbc_requests_size_, (*request)->nbc_requests_, MPI_STATUSES_IGNORE); + ret = waitall((*request)->nbc_requests_.size(), &(*request)->nbc_requests_[0], MPI_STATUSES_IGNORE); else{ - ret = testall((*request)->nbc_requests_size_, (*request)->nbc_requests_, &flag, MPI_STATUSES_IGNORE); + ret = testall((*request)->nbc_requests_.size(), &(*request)->nbc_requests_[0], &flag, MPI_STATUSES_IGNORE); } if(ret!=MPI_SUCCESS) xbt_die("Failure when waiting on non blocking collective sub-requests"); if(flag == 1){ - XBT_DEBUG("Finishing non blocking collective request with %d sub-requests", (*request)->nbc_requests_size_); - for (int i = 0; i < (*request)->nbc_requests_size_; i++) { - if((*request)->buf_!=nullptr && (*request)->nbc_requests_[i]!=MPI_REQUEST_NULL){//reduce case - void * buf=(*request)->nbc_requests_[i]->buf_; + XBT_DEBUG("Finishing non blocking collective request with %zu sub-requests", (*request)->nbc_requests_.size()); + for(auto& req: (*request)->nbc_requests_){ + if((*request)->buf_!=nullptr && req!=MPI_REQUEST_NULL){//reduce case + void * buf=req->buf_; if((*request)->old_type_->flags() & DT_FLAG_DERIVED) - buf=(*request)->nbc_requests_[i]->old_buf_; - if((*request)->nbc_requests_[i]->flags_ & MPI_REQ_RECV ){ + buf=req->old_buf_; + if(req->flags_ & MPI_REQ_RECV ){ if((*request)->op_!=MPI_OP_NULL){ int count=(*request)->size_/ (*request)->old_type_->size(); (*request)->op_->apply(buf, (*request)->buf_, &count, (*request)->old_type_); @@ -877,11 +875,10 @@ int Request::finish_nbc_requests(MPI_Request* request, int test){ smpi_free_tmp_buffer(static_cast(buf)); } } - if((*request)->nbc_requests_[i]!=MPI_REQUEST_NULL) - Request::unref(&((*request)->nbc_requests_[i])); + if(req!=MPI_REQUEST_NULL) + Request::unref(&req); } - delete[] (*request)->nbc_requests_; - (*request)->nbc_requests_size_=0; + (*request)->nbc_requests_.clear(); } return flag; } @@ -1263,22 +1260,14 @@ int Request::grequest_complete(MPI_Request request) return MPI_SUCCESS; } -void Request::set_nbc_requests(MPI_Request* reqs, int size){ - nbc_requests_size_ = size; - if (size > 0) { +void Request::start_nbc_requests(std::vector reqs){ + if (reqs.size() > 0) { nbc_requests_ = reqs; - } else { - delete[] reqs; - nbc_requests_ = nullptr; + Request::startall(reqs.size(), &reqs[0]); } } -int Request::get_nbc_requests_size() const -{ - return nbc_requests_size_; -} - -MPI_Request* Request::get_nbc_requests() const +std::vector Request::get_nbc_requests() const { return nbc_requests_; } -- 2.20.1