Encapsulate multi-process communication

source/adios2/CMakeLists.txt

@@ -32,10 +32,10 @@ add_library(adios2
   operator/callback/Signature2.cpp
 
 #helper
+  helper/adiosComm.h  helper/adiosComm.cpp
   helper/adiosDynamicBinder.h  helper/adiosDynamicBinder.cpp
   helper/adiosMath.cpp
   helper/adiosMemory.cpp
-  helper/adiosMPIFunctions.cpp
   helper/adiosNetwork.cpp
   helper/adiosString.cpp helper/adiosString.tcc
   helper/adiosSystem.cpp

source/adios2/core/ADIOS.cpp

@@ -59,10 +59,9 @@ namespace core
 
 ADIOS::ADIOS(const std::string configFile, MPI_Comm mpiComm,
              const bool debugMode, const std::string hostLanguage)
-: m_ConfigFile(configFile), m_DebugMode(debugMode), m_HostLanguage(hostLanguage)
+: m_ConfigFile(configFile), m_DebugMode(debugMode),
+  m_HostLanguage(hostLanguage), m_Comm(helper::Comm::Duplicate(mpiComm))
 {
-    SMPI_Comm_dup(mpiComm, &m_MPIComm);
-
     if (!configFile.empty())
     {
         if (configFile.substr(configFile.size() - 3) == "xml")
@@ -90,17 +89,7 @@ ADIOS::ADIOS(const bool debugMode, const std::string hostLanguage)
 {
 }
 
-ADIOS::~ADIOS()
-{
-    // Handle the case where MPI is finalized before the ADIOS destructor is
-    // called, which happens, e.g., with global / static ADIOS objects
-    int flag;
-    MPI_Finalized(&flag);
-    if (!flag)
-    {
-        SMPI_Comm_free(&m_MPIComm);
-    }
-}
+ADIOS::~ADIOS() = default;
 
 IO &ADIOS::DeclareIO(const std::string name)
 {
@@ -128,7 +117,7 @@ IO &ADIOS::DeclareIO(const std::string name)
     }
 
     auto ioPair = m_IOs.emplace(
-        name, IO(*this, name, m_MPIComm, false, m_HostLanguage, m_DebugMode));
+        name, IO(*this, name, false, m_HostLanguage, m_DebugMode));
     IO &io = ioPair.first->second;
     io.SetDeclared();
     return io;

source/adios2/core/ADIOS.h

@@ -22,6 +22,7 @@
 #include "adios2/common/ADIOSMPI.h"
 #include "adios2/common/ADIOSTypes.h"
 #include "adios2/core/Operator.h"
+#include "adios2/helper/adiosComm.h"
 
 namespace adios2
 {
@@ -38,8 +39,8 @@ class ADIOS
     /** if true will do more checks, exceptions, warnings, expect slower code */
     const bool m_DebugMode = true;
 
-    /** Passed from parallel constructor, MPI_Comm is a pointer itself. */
-    MPI_Comm m_MPIComm;
+    /** Get the communicator passed to constructor for parallel case.  */
+    helper::Comm const &GetComm() const { return m_Comm; }
 
     /** Changed by language bindings in constructor */
     const std::string m_HostLanguage = "C++";
@@ -184,6 +185,9 @@ class ADIOS
     void RemoveAllIOs() noexcept;
 
 private:
+    /** Communicator given to parallel constructor. */
+    helper::Comm m_Comm;
+
     /** XML File to be read containing configuration information */
     const std::string m_ConfigFile;
 

source/adios2/core/Engine.cpp

@@ -12,6 +12,7 @@
 #include "Engine.tcc"
 
 #include <stdexcept>
+#include <utility>
 
 #include "adios2/core/IO.h"
 
@@ -21,9 +22,9 @@ namespace core
 {
 
 Engine::Engine(const std::string engineType, IO &io, const std::string &name,
-               const Mode openMode, MPI_Comm mpiComm)
+               const Mode openMode, helper::Comm comm)
 : m_EngineType(engineType), m_IO(io), m_Name(name), m_OpenMode(openMode),
-  m_MPIComm(mpiComm), m_DebugMode(io.m_DebugMode)
+  m_Comm(std::move(comm)), m_DebugMode(io.m_DebugMode)
 {
 }
 
@@ -69,9 +70,7 @@ void Engine::Close(const int transportIndex)
 
     if (transportIndex == -1)
     {
-        helper::CheckMPIReturn(SMPI_Comm_free(&m_MPIComm),
-                               "freeing comm in Engine " + m_Name +
-                                   ", in call to Close");
+        m_Comm.Free("freeing comm in Engine " + m_Name + ", in call to Close");
         m_IsClosed = true;
     }
 }

source/adios2/core/Engine.h

@@ -29,6 +29,7 @@
 #include "adios2/core/IO.h"
 #include "adios2/core/Variable.h"
 #include "adios2/core/VariableCompound.h"
+#include "adios2/helper/adiosComm.h"
 
 namespace adios2
 {
@@ -62,10 +63,10 @@ class Engine
      * @param io object that generates this Engine
      * @param name unique engine name within IO class object
      * @param mode  open mode from ADIOSTypes.h Mode
-     * @param mpiComm new communicator passed at Open or from ADIOS class
+     * @param comm  communicator passed at Open or from ADIOS class
      */
     Engine(const std::string engineType, IO &io, const std::string &name,
-           const Mode mode, MPI_Comm mpiComm);
+           const Mode mode, helper::Comm comm);
 
     virtual ~Engine();
 
@@ -446,8 +447,8 @@ class Engine
 
 protected:
     /** from ADIOS class passed to Engine created with Open
-     *  if no new communicator is passed */
-    MPI_Comm m_MPIComm;
+     *  if no communicator is passed */
+    helper::Comm m_Comm;
 
     /** true: additional exceptions */
     const bool m_DebugMode = false;

source/adios2/core/IO.cpp

@@ -27,6 +27,7 @@
 #include "adios2/engine/skeleton/SkeletonReader.h"
 #include "adios2/engine/skeleton/SkeletonWriter.h"
 
+#include "adios2/helper/adiosComm.h"
 #include "adios2/helper/adiosFunctions.h" //BuildParametersMap
 #include "adios2/toolkit/profiling/taustubs/tautimer.hpp"
 #include <adios2sys/SystemTools.hxx> // FileIsDirectory()
@@ -73,12 +74,10 @@ namespace adios2
 namespace core
 {
 
-IO::IO(ADIOS &adios, const std::string name, MPI_Comm mpiComm,
-       const bool inConfigFile, const std::string hostLanguage,
-       const bool debugMode)
-: m_ADIOS(adios), m_Name(name), m_MPIComm(mpiComm),
-  m_InConfigFile(inConfigFile), m_HostLanguage(hostLanguage),
-  m_DebugMode(debugMode)
+IO::IO(ADIOS &adios, const std::string name, const bool inConfigFile,
+       const std::string hostLanguage, const bool debugMode)
+: m_ADIOS(adios), m_Name(name), m_InConfigFile(inConfigFile),
+  m_HostLanguage(hostLanguage), m_DebugMode(debugMode)
 {
 }
 
@@ -426,8 +425,7 @@ size_t IO::AddOperation(Operator &op, const Params &parameters) noexcept
     return m_Operations.size() - 1;
 }
 
-Engine &IO::Open(const std::string &name, const Mode mode,
-                 MPI_Comm mpiComm_orig)
+Engine &IO::Open(const std::string &name, const Mode mode, MPI_Comm mpiComm)
 {
     TAU_SCOPED_TIMER("IO::Open");
     auto itEngineFound = m_Engines.find(name);
@@ -459,8 +457,7 @@ Engine &IO::Open(const std::string &name, const Mode mode,
         }
     }
 
-    MPI_Comm mpiComm;
-    SMPI_Comm_dup(mpiComm_orig, &mpiComm);
+    auto comm = helper::Comm::Duplicate(mpiComm);
     std::shared_ptr<Engine> engine;
     const bool isDefaultEngine = m_EngineType.empty() ? true : false;
     std::string engineTypeLC = m_EngineType;
@@ -495,26 +492,26 @@ Engine &IO::Open(const std::string &name, const Mode mode,
     {
         if (mode == Mode::Read)
         {
-            engine =
-                std::make_shared<engine::BP3Reader>(*this, name, mode, mpiComm);
+            engine = std::make_shared<engine::BP3Reader>(*this, name, mode,
+                                                         std::move(comm));
         }
         else
         {
-            engine =
-                std::make_shared<engine::BP3Writer>(*this, name, mode, mpiComm);
+            engine = std::make_shared<engine::BP3Writer>(*this, name, mode,
+                                                         std::move(comm));
         }
     }
     else if (engineTypeLC == "bp4")
     {
         if (mode == Mode::Read)
         {
-            engine =
-                std::make_shared<engine::BP4Reader>(*this, name, mode, mpiComm);
+            engine = std::make_shared<engine::BP4Reader>(*this, name, mode,
+                                                         std::move(comm));
         }
         else
         {
-            engine =
-                std::make_shared<engine::BP4Writer>(*this, name, mode, mpiComm);
+            engine = std::make_shared<engine::BP4Writer>(*this, name, mode,
+                                                         std::move(comm));
         }
     }
     else if (engineTypeLC == "hdfmixer")
@@ -523,10 +520,10 @@ Engine &IO::Open(const std::string &name, const Mode mode,
 #if H5_VERSION_GE(1, 11, 0)
         if (mode == Mode::Read)
             engine = std::make_shared<engine::HDF5ReaderP>(*this, name, mode,
-                                                           mpiComm);
+                                                           std::move(comm));
         else
-            engine =
-                std::make_shared<engine::HDFMixer>(*this, name, mode, mpiComm);
+            engine = std::make_shared<engine::HDFMixer>(*this, name, mode,
+                                                        std::move(comm));
 #else
         throw std::invalid_argument(
             "ERROR: update HDF5 >= 1.11 to support VDS.");
@@ -541,10 +538,10 @@ Engine &IO::Open(const std::string &name, const Mode mode,
 #ifdef ADIOS2_HAVE_DATAMAN
         if (mode == Mode::Read)
             engine = std::make_shared<engine::DataManReader>(*this, name, mode,
-                                                             mpiComm);
+                                                             std::move(comm));
         else
             engine = std::make_shared<engine::DataManWriter>(*this, name, mode,
-                                                             mpiComm);
+                                                             std::move(comm));
 #else
         throw std::invalid_argument(
             "ERROR: this version didn't compile with "
@@ -555,11 +552,11 @@ Engine &IO::Open(const std::string &name, const Mode mode,
     {
 #ifdef ADIOS2_HAVE_SSC
         if (mode == Mode::Read)
-            engine =
-                std::make_shared<engine::SscReader>(*this, name, mode, mpiComm);
+            engine = std::make_shared<engine::SscReader>(*this, name, mode,
+                                                         std::move(comm));
         else
-            engine =
-                std::make_shared<engine::SscWriter>(*this, name, mode, mpiComm);
+            engine = std::make_shared<engine::SscWriter>(*this, name, mode,
+                                                         std::move(comm));
 #else
         throw std::invalid_argument("ERROR: this version didn't compile with "
                                     "SSC library, can't use SSC engine\n");
@@ -570,7 +567,7 @@ Engine &IO::Open(const std::string &name, const Mode mode,
 #ifdef ADIOS2_HAVE_TABLE
         if (mode == Mode::Write)
             engine = std::make_shared<engine::TableWriter>(*this, name, mode,
-                                                           mpiComm);
+                                                           std::move(comm));
         else
             throw std::invalid_argument(
                 "ERROR: Table engine only supports Write. It uses other "
@@ -585,11 +582,11 @@ Engine &IO::Open(const std::string &name, const Mode mode,
     {
 #ifdef ADIOS2_HAVE_SST
         if (mode == Mode::Read)
-            engine =
-                std::make_shared<engine::SstReader>(*this, name, mode, mpiComm);
+            engine = std::make_shared<engine::SstReader>(*this, name, mode,
+                                                         std::move(comm));
         else
-            engine =
-                std::make_shared<engine::SstWriter>(*this, name, mode, mpiComm);
+            engine = std::make_shared<engine::SstWriter>(*this, name, mode,
+                                                         std::move(comm));
 #else
         throw std::invalid_argument("ERROR: this version didn't compile with "
                                     "Sst library, can't use Sst engine\n");
@@ -599,11 +596,11 @@ Engine &IO::Open(const std::string &name, const Mode mode,
     {
 #ifdef ADIOS2_HAVE_DATASPACES
         if (mode == Mode::Read)
-            engine = std::make_shared<engine::DataSpacesReader>(*this, name,
-                                                                mode, mpiComm);
+            engine = std::make_shared<engine::DataSpacesReader>(
+                *this, name, mode, std::move(comm));
         else
-            engine = std::make_shared<engine::DataSpacesWriter>(*this, name,
-                                                                mode, mpiComm);
+            engine = std::make_shared<engine::DataSpacesWriter>(
+                *this, name, mode, std::move(comm));
 #else
         throw std::invalid_argument(
             "ERROR: this version didn't compile with "
@@ -615,10 +612,10 @@ Engine &IO::Open(const std::string &name, const Mode mode,
 #ifdef ADIOS2_HAVE_HDF5
         if (mode == Mode::Read)
             engine = std::make_shared<engine::HDF5ReaderP>(*this, name, mode,
-                                                           mpiComm);
+                                                           std::move(comm));
         else
             engine = std::make_shared<engine::HDF5WriterP>(*this, name, mode,
-                                                           mpiComm);
+                                                           std::move(comm));
 #else
         throw std::invalid_argument("ERROR: this version didn't compile with "
                                     "HDF5 library, can't use HDF5 engine\n");
@@ -628,11 +625,11 @@ Engine &IO::Open(const std::string &name, const Mode mode,
     {
 #ifdef ADIOS2_HAVE_MPI
         if (mode == Mode::Read)
-            engine = std::make_shared<engine::InSituMPIReader>(*this, name,
-                                                               mode, mpiComm);
+            engine = std::make_shared<engine::InSituMPIReader>(
+                *this, name, mode, std::move(comm));
         else
-            engine = std::make_shared<engine::InSituMPIWriter>(*this, name,
-                                                               mode, mpiComm);
+            engine = std::make_shared<engine::InSituMPIWriter>(
+                *this, name, mode, std::move(comm));
 #else
         throw std::invalid_argument("ERROR: this version didn't compile with "
                                     "MPI, can't use InSituMPI engine\n");
@@ -642,24 +639,24 @@ Engine &IO::Open(const std::string &name, const Mode mode,
     {
         if (mode == Mode::Read)
             engine = std::make_shared<engine::SkeletonReader>(*this, name, mode,
-                                                              mpiComm);
+                                                              std::move(comm));
         else
             engine = std::make_shared<engine::SkeletonWriter>(*this, name, mode,
-                                                              mpiComm);
+                                                              std::move(comm));
     }
     else if (engineTypeLC == "inline")
     {
         if (mode == Mode::Read)
             engine = std::make_shared<engine::InlineReader>(*this, name, mode,
-                                                            mpiComm);
+                                                            std::move(comm));
         else
             engine = std::make_shared<engine::InlineWriter>(*this, name, mode,
-                                                            mpiComm);
+                                                            std::move(comm));
     }
     else if (engineTypeLC == "null")
     {
-        engine =
-            std::make_shared<engine::NullEngine>(*this, name, mode, mpiComm);
+        engine = std::make_shared<engine::NullEngine>(*this, name, mode,
+                                                      std::move(comm));
     }
     else if (engineTypeLC == "nullcore")
     {
@@ -668,7 +665,7 @@ Engine &IO::Open(const std::string &name, const Mode mode,
                 "ERROR: nullcore engine does not support read mode");
         else
             engine = std::make_shared<engine::NullCoreWriter>(*this, name, mode,
-                                                              mpiComm);
+                                                              std::move(comm));
     }
     else
     {
@@ -698,7 +695,7 @@ Engine &IO::Open(const std::string &name, const Mode mode,
 
 Engine &IO::Open(const std::string &name, const Mode mode)
 {
-    return Open(name, mode, m_MPIComm);
+    return Open(name, mode, m_ADIOS.GetComm().AsMPI());
 }
 
 Engine &IO::GetEngine(const std::string &name)