diff --git a/src/axom/lumberjack/MPIUtility.cpp b/src/axom/lumberjack/MPIUtility.cpp index 202473d372..68e8030536 100644 --- a/src/axom/lumberjack/MPIUtility.cpp +++ b/src/axom/lumberjack/MPIUtility.cpp @@ -68,16 +68,25 @@ const char* mpiBlockingReceiveIfMessagesExist(MPI_Comm comm) return charArray; } -void mpiNonBlockingSendMessages(MPI_Comm comm, int destinationRank, const char* packedMessagesToBeSent) +MPI_Request mpiNonBlockingSendMessagesWithRequest(MPI_Comm comm, + int destinationRank, + const char* packedMessagesToBeSent) { MPI_Request mpiRequest; MPI_Isend(const_cast(packedMessagesToBeSent), - strlen(packedMessagesToBeSent), + std::strlen(packedMessagesToBeSent), MPI_CHAR, destinationRank, LJ_TAG, comm, &mpiRequest); + return mpiRequest; +} + +void mpiNonBlockingSendMessages(MPI_Comm comm, int destinationRank, const char* packedMessagesToBeSent) +{ + MPI_Request mpiRequest = + mpiNonBlockingSendMessagesWithRequest(comm, destinationRank, packedMessagesToBeSent); MPI_Request_free(&mpiRequest); } diff --git a/src/axom/lumberjack/MPIUtility.hpp b/src/axom/lumberjack/MPIUtility.hpp index 166b9503f2..2878d38a02 100644 --- a/src/axom/lumberjack/MPIUtility.hpp +++ b/src/axom/lumberjack/MPIUtility.hpp @@ -45,6 +45,25 @@ const char* mpiBlockingReceiveMessages(MPI_Comm comm); */ const char* mpiBlockingReceiveIfMessagesExist(MPI_Comm comm); +/*! + ***************************************************************************** + * \brief Starts a non-blocking send to the given rank. + * + * The caller owns the returned MPI_Request and must keep + * packedMessagesToBeSent valid until the request completes. + * + * \param [in] comm The MPI Communicator. + * \param [in] destinationRank Where the Message classes is being sent. + * \param [in,out] packedMessagesToBeSent All of the Message classes to be sent + * packed together. + * + * \return Request associated with the non-blocking send. + ***************************************************************************** + */ +MPI_Request mpiNonBlockingSendMessagesWithRequest(MPI_Comm comm, + int destinationRank, + const char* packedMessagesToBeSent); + /*! ***************************************************************************** * \brief Sends all Message sent to the given rank. diff --git a/src/axom/lumberjack/NonCollectiveRootCommunicator.cpp b/src/axom/lumberjack/NonCollectiveRootCommunicator.cpp index 554a0489a0..80cdf86d51 100644 --- a/src/axom/lumberjack/NonCollectiveRootCommunicator.cpp +++ b/src/axom/lumberjack/NonCollectiveRootCommunicator.cpp @@ -14,8 +14,10 @@ ****************************************************************************** */ -#include #include +#include +#include +#include #include "axom/lumberjack/NonCollectiveRootCommunicator.hpp" #include "axom/lumberjack/MPIUtility.hpp" @@ -26,6 +28,7 @@ namespace lumberjack { void NonCollectiveRootCommunicator::initialize(MPI_Comm comm, int ranksLimit) { + m_pendingSends.clear(); if(ranksLimit < 1) { std::cerr << "Error: Ranks limit passed to NonCollectiveRootCommunicator " @@ -41,7 +44,11 @@ void NonCollectiveRootCommunicator::initialize(MPI_Comm comm, int ranksLimit) m_ranksLimit = ranksLimit; } -void NonCollectiveRootCommunicator::finalize() { MPI_Comm_free(&m_mpiComm); } +void NonCollectiveRootCommunicator::finalize() +{ + releasePendingSends(); + MPI_Comm_free(&m_mpiComm); +} MPI_Comm NonCollectiveRootCommunicator::comm() { return m_mpiComm; } @@ -85,11 +92,61 @@ void NonCollectiveRootCommunicator::push(const char* packedMessagesToBeSent, } else { + drainCompletedSends(); if(isPackedMessagesEmpty(packedMessagesToBeSent) == false) { - mpiNonBlockingSendMessages(m_mpiComm, 0, packedMessagesToBeSent); + const int messageSize = static_cast(std::strlen(packedMessagesToBeSent)); + PendingSend pendingSend; + pendingSend.request = MPI_REQUEST_NULL; + pendingSend.buffer.reset(new char[messageSize + 1]); + std::memcpy(pendingSend.buffer.get(), packedMessagesToBeSent, messageSize + 1); + + pendingSend.request = + mpiNonBlockingSendMessagesWithRequest(m_mpiComm, 0, pendingSend.buffer.get()); + m_pendingSends.push_back(std::move(pendingSend)); + } + drainCompletedSends(); + } +} + +void NonCollectiveRootCommunicator::drainCompletedSends() +{ + for(auto it = m_pendingSends.begin(); it != m_pendingSends.end();) + { + int complete = 0; + MPI_Test(&it->request, &complete, MPI_STATUS_IGNORE); + if(complete) + { + it = m_pendingSends.erase(it); + } + else + { + ++it; + } + } +} + +void NonCollectiveRootCommunicator::releasePendingSends() +{ + for(auto& pendingSend : m_pendingSends) + { + if(pendingSend.request == MPI_REQUEST_NULL) + { + continue; + } + + int complete = 0; + MPI_Test(&pendingSend.request, &complete, MPI_STATUS_IGNORE); + if(!complete) + { + // Keep the send buffer valid without blocking finalization. This + // communicator is used for best-effort non-collective error reporting; + // waiting here could hang if root is no longer receiving. + MPI_Request_free(&pendingSend.request); + pendingSend.buffer.release(); } } + m_pendingSends.clear(); } bool NonCollectiveRootCommunicator::isOutputNode() diff --git a/src/axom/lumberjack/NonCollectiveRootCommunicator.hpp b/src/axom/lumberjack/NonCollectiveRootCommunicator.hpp index 1c0345b1d8..6298c27c2b 100644 --- a/src/axom/lumberjack/NonCollectiveRootCommunicator.hpp +++ b/src/axom/lumberjack/NonCollectiveRootCommunicator.hpp @@ -19,6 +19,9 @@ #include "axom/lumberjack/Lumberjack.hpp" #include "axom/lumberjack/Communicator.hpp" +#include +#include + namespace axom { namespace lumberjack @@ -140,11 +143,21 @@ class NonCollectiveRootCommunicator : public axom::lumberjack::Communicator double startTime(); private: + struct PendingSend + { + MPI_Request request; + std::unique_ptr buffer; + }; + + void drainCompletedSends(); + void releasePendingSends(); + MPI_Comm m_mpiComm; int m_mpiCommRank; int m_mpiCommSize; int m_ranksLimit; double m_startTime; + std::vector m_pendingSends; }; } // end namespace lumberjack