Non-Collective Root Communicator Managed MPI Buffer

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<char*>(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);
 }
 

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.

src/axom/lumberjack/NonCollectiveRootCommunicator.cpp

@@ -14,8 +14,10 @@
  ******************************************************************************
  */
 
-#include <limits>
 #include <iostream>
+#include <cstring>
+#include <limits>
+#include <utility>
 
 #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<int>(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()

src/axom/lumberjack/NonCollectiveRootCommunicator.hpp

@@ -19,6 +19,9 @@
 #include "axom/lumberjack/Lumberjack.hpp"
 #include "axom/lumberjack/Communicator.hpp"
 
+#include <memory>
+#include <vector>
+
 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<char[]> buffer;
+  };
+
+  void drainCompletedSends();
+  void releasePendingSends();
+
   MPI_Comm m_mpiComm;
   int m_mpiCommRank;
   int m_mpiCommSize;
   int m_ranksLimit;
   double m_startTime;
+  std::vector<PendingSend> m_pendingSends;
 };
 
 }  // end namespace lumberjack