added non-blocking root communicator

src/axom/lumberjack/CMakeLists.txt

@@ -17,6 +17,7 @@ set(lumberjack_headers
     Message.hpp
     MPIUtility.hpp
     RootCommunicator.hpp
+    NonCollectiveRootCommunicator.hpp
     TextEqualityCombiner.hpp
     TextTagCombiner.hpp
     )
@@ -27,6 +28,7 @@ set(lumberjack_sources
     Message.cpp
     MPIUtility.cpp
     RootCommunicator.cpp
+    NonCollectiveRootCommunicator.cpp
     )
 
 

src/axom/lumberjack/MPIUtility.cpp

@@ -14,7 +14,6 @@
  */
 
 #include "axom/lumberjack/MPIUtility.hpp"
-
 #include <cstring>
 
 namespace axom
@@ -49,16 +48,50 @@ const char* mpiBlockingReceiveMessages(MPI_Comm comm)
   return charArray;
 }
 
+const char* mpiNonBlockingReceiveMessages(MPI_Comm comm, int tag)
+{
+  const int mpiTag = (tag == 0) ? LJ_TAG : tag;
+  char* charArray = nullptr;
+  int messageSize = -1;
+  MPI_Status mpiStatus;
+
+  // Get size and source of MPI message
+  int mpiFlag = 0;
+  MPI_Iprobe(MPI_ANY_SOURCE, tag, comm, &mpiFlag, &mpiStatus);
+
+  if(mpiFlag == 1)
+  {
+    MPI_Get_count(&mpiStatus, MPI_CHAR, &messageSize);
+
+    // Setup where to receive the char array
+    charArray = new char[messageSize + 1];
+    charArray[messageSize] = '\0';
+
+    // Receive packed Message
+    MPI_Recv(charArray,
+             messageSize,
+             MPI_CHAR,
+             mpiStatus.MPI_SOURCE,
+             mpiTag,
+             comm,
+             &mpiStatus);
+  }
+
+  return charArray;
+}
+
 void mpiNonBlockingSendMessages(MPI_Comm comm,
                                 int destinationRank,
-                                const char* packedMessagesToBeSent)
+                                const char* packedMessagesToBeSent,
+                                int tag)
 {
+  const int mpiTag = (tag == 0) ? LJ_TAG : tag;
   MPI_Request mpiRequest;
   MPI_Isend(const_cast<char*>(packedMessagesToBeSent),
             strlen(packedMessagesToBeSent),
             MPI_CHAR,
             destinationRank,
-            LJ_TAG,
+            mpiTag,
             comm,
             &mpiRequest);
   MPI_Request_free(&mpiRequest);

src/axom/lumberjack/MPIUtility.hpp

@@ -30,6 +30,18 @@ namespace lumberjack
  */
 const char* mpiBlockingReceiveMessages(MPI_Comm comm);
 
+/*!
+ *****************************************************************************
+ * \brief Receives any Message sent to this rank, if there are any messages 
+ *  that are sent. Returns null if no messages are sent.
+ *
+ * \param [in] comm The MPI Communicator.
+ * \param [in] tag The MPI tag to use for communication.  When set to zero, 
+ *  MPI communication uses default LJ_Tag. 
+ *****************************************************************************
+ */
+const char* mpiNonBlockingReceiveMessages(MPI_Comm comm, int tag = 0);
+
 /*!
  *****************************************************************************
  * \brief Sends all Message sent to the given rank.
@@ -40,11 +52,14 @@ const char* mpiBlockingReceiveMessages(MPI_Comm comm);
  * \param [in] destinationRank Where the Message classes is being sent.
  * \param [in,out] packedMessagesToBeSent All of the Message classes to be sent
  *  packed together.
+ * \param [in] tag The MPI tag to use for communication.  When set to zero, 
+ *  MPI communication uses the default Lumberjack Tag.
  *****************************************************************************
  */
 void mpiNonBlockingSendMessages(MPI_Comm comm,
                                 int destinationRank,
-                                const char* packedMessagesToBeSent);
+                                const char* packedMessagesToBeSent,
+                                int tag = 0);
 }  // end namespace lumberjack
 }  // end namespace axom
 

src/axom/lumberjack/NonCollectiveRootCommunicator.cpp

@@ -0,0 +1,112 @@
+// Copyright (c) 2017-2024, Lawrence Livermore National Security, LLC and
+// other Axom Project Developers. See the top-level LICENSE file for details.
+//
+// SPDX-License-Identifier: (BSD-3-Clause)
+
+/*!
+ ******************************************************************************
+ *
+ * \file NonCollectiveRootCommunicator.cpp
+ *
+ * \brief Implementation of the NonCollectiveRootCommunicator class.
+ *
+ ******************************************************************************
+ */
+
+#include <limits>
+
+#include "axom/lumberjack/NonCollectiveRootCommunicator.hpp"
+#include "axom/lumberjack/MPIUtility.hpp"
+
+namespace axom
+{
+namespace lumberjack
+{
+void NonCollectiveRootCommunicator::initialize(MPI_Comm comm, int ranksLimit)
+{
+  static int mpiTag = 32767;
+  m_mpiComm = comm;
+  MPI_Comm_rank(m_mpiComm, &m_mpiCommRank);
+  MPI_Comm_size(m_mpiComm, &m_mpiCommSize);
+  m_ranksLimit = ranksLimit;
+  m_mpiTag = mpiTag;
+  /*
+    Each communicator that sends/receives messages non-collectively needs its 
+    own mpiTag in order to not interfere with other communicators.
+   */
+  if(mpiTag == std::numeric_limits<int>::max())
+  {
+    mpiTag = 32767;
+  }
+  else
+  {
+    ++mpiTag;
+  }
+}
+
+void NonCollectiveRootCommunicator::finalize() { }
+
+int NonCollectiveRootCommunicator::rank() { return m_mpiCommRank; }
+
+void NonCollectiveRootCommunicator::ranksLimit(int value)
+{
+  m_ranksLimit = value;
+}
+
+int NonCollectiveRootCommunicator::ranksLimit() { return m_ranksLimit; }
+
+int NonCollectiveRootCommunicator::numPushesToFlush() { return 1; }
+
+void NonCollectiveRootCommunicator::push(
+  const char* packedMessagesToBeSent,
+  std::vector<const char*>& receivedPackedMessages)
+{
+  if(m_mpiCommRank == 0)
+  {
+    const char* currPackedMessages = nullptr;
+    bool receive_messages = true;
+    while(receive_messages)
+    {
+      currPackedMessages = mpiNonBlockingReceiveMessages(m_mpiComm, m_mpiTag);
+
+      if(isPackedMessagesEmpty(currPackedMessages))
+      {
+        if(currPackedMessages == nullptr)
+        {
+          receive_messages = false;
+        }
+        else
+        {
+          delete[] currPackedMessages;
+        }
+      }
+      else
+      {
+        receivedPackedMessages.push_back(currPackedMessages);
+      }
+
+      currPackedMessages = nullptr;
+    }
+  }
+  else
+  {
+    if(isPackedMessagesEmpty(packedMessagesToBeSent) == false)
+    {
+      mpiNonBlockingSendMessages(m_mpiComm, 0, packedMessagesToBeSent, m_mpiTag);
+    }
+  }
+}
+
+bool NonCollectiveRootCommunicator::isOutputNode()
+{
+  if(m_mpiCommRank == 0)
+  {
+    return true;
+  }
+  return false;
+}
+
+int NonCollectiveRootCommunicator::mpiTag() const { return m_mpiTag; }
+
+}  // end namespace lumberjack
+}  // end namespace axom

src/axom/lumberjack/NonCollectiveRootCommunicator.hpp

@@ -0,0 +1,143 @@
+// Copyright (c) 2017-2024, Lawrence Livermore National Security, LLC and
+// other Axom Project Developers. See the top-level LICENSE file for details.
+//
+// SPDX-License-Identifier: (BSD-3-Clause)
+
+/*!
+ *******************************************************************************
+ * \file NonCollectiveRootCommunicator.hpp
+ *
+ * \brief This file contains the class definition of the 
+ *  NonCollectiveRootCommunicator.
+ *******************************************************************************
+ */
+
+#ifndef NONCOLLECTIVEROOTCOMMUNICATOR_HPP
+#define NONCOLLECTIVEROOTCOMMUNICATOR_HPP
+
+#include "axom/lumberjack/Lumberjack.hpp"
+#include "axom/lumberjack/Communicator.hpp"
+
+namespace axom
+{
+namespace lumberjack
+{
+/*!
+ *******************************************************************************
+ * \class NonCollectiveRootCommunicator
+ *
+ * \brief Based off of RootCommunicator. This communicator pushes
+    messages from any rank to root non-collectively, if any messages are sent.
+ *******************************************************************************
+ */
+class NonCollectiveRootCommunicator : public axom::lumberjack::Communicator
+{
+public:
+  /*!
+   *****************************************************************************
+   * \brief Called to initialize the Communicator.
+   *
+   * This performs any setup work the Communicator needs before doing any work.
+   * It is required that this is called before using the Communicator.
+   *
+   * \param [in] comm The MPI Communicator
+   * \param [in] ranksLimit Limit on how many ranks are individually tracked per
+   *  Message.
+   *****************************************************************************
+   */
+  void initialize(MPI_Comm comm, int ranksLimit);
+
+  /*!
+   *****************************************************************************
+   * \brief Called to finalize the Communicator.
+   *
+   * This performs any cleanup work the Communicator needs to do before going
+   * away.It is required that this is the last function called by the
+   * Communicator.
+   *****************************************************************************
+   */
+  void finalize();
+
+  /*!
+   *****************************************************************************
+   * \brief Returns the MPI rank of this node
+   *****************************************************************************
+   */
+  int rank();
+
+  /*!
+   *****************************************************************************
+   * \brief Sets the rank limit.
+   *
+   * This is the limit on how many ranks generated a given message are
+   * individually tracked per Message.  After the limit has been reached, only
+   * the Message::rankCount is incremented.
+   *
+   * \param [in] value Limits how many ranks are tracked per Message.
+   *****************************************************************************
+   */
+  void ranksLimit(int value);
+
+  /*!
+   *****************************************************************************
+   * \brief Returns the rank limit.
+   *
+   * This is the limit on how many ranks generated a given message are
+   * individually tracked per Message.  After the limit has been reached, only
+   * the Message::rankCount is incremented.
+   *****************************************************************************
+   */
+  int ranksLimit();
+
+  /*!
+   *****************************************************************************
+   * \brief Function used by the Lumberjack class to indicate how many
+   * individual pushes fully flush all currently held Message classes to the
+   * root node. The Communicator class's tree structure dictates this.
+   *****************************************************************************
+   */
+  int numPushesToFlush();
+
+  /*!
+   *****************************************************************************
+   * \brief This pushes all messages to the root node.
+   *
+   * All messages are pushed to the root node. This is the same as
+   * RootCommunicator::pushMessagesFully for this Communicator.
+   *
+   * \param [in] packedMessagesToBeSent All of this rank's Message classes
+   *  packed into a single buffer.
+   * \param [in,out] receivedPackedMessages Received packed message buffers from
+   *  this nodes children.
+   *****************************************************************************
+   */
+  void push(const char* packedMessagesToBeSent,
+            std::vector<const char*>& receivedPackedMessages);
+
+  /*!
+   *****************************************************************************
+   * \brief Function used by the Lumberjack to indicate whether this node should
+   *  be outputting messages. Only the root node outputs messages.
+   *****************************************************************************
+   */
+  bool isOutputNode();
+
+  /*!
+   *****************************************************************************
+   * \brief Returns the MPI tag used for this communicator.
+   *****************************************************************************
+   */
+  int mpiTag() const;
+
+private:
+  MPI_Comm m_mpiComm;
+  int m_mpiCommRank;
+  int m_mpiCommSize;
+  int m_ranksLimit;
+  int m_mpiTag;
+};
+
+}  // end namespace lumberjack
+}  // end namespace axom
+
+#endif

src/axom/lumberjack/tests/CMakeLists.txt

@@ -34,7 +34,8 @@ endforeach()
 #------------------------------------------------------------------------------
 set(lumberjack_mpi_tests
     lumberjack_BinaryCommunicator.hpp
-    lumberjack_RootCommunicator.hpp )
+    lumberjack_RootCommunicator.hpp
+    lumberjack_NonCollectiveRootCommunicator.hpp )
 
 axom_add_executable(NAME       lumberjack_mpi_tests
                     SOURCES    lumberjack_mpi_main.cpp