Open new instance of ADIOS per File

Author: franzpoeschel

include/openPMD/IO/ADIOS/ADIOS2Auxiliary.hpp

@@ -81,6 +81,12 @@ namespace adios_defs
         Open,
         ReopenFileThatWeCreated
     };
+
+    struct ParameterizedOperator
+    {
+        adios2::Operator op;
+        adios2::Params params;
+    };
 } // namespace adios_defs
 
 /*

include/openPMD/IO/ADIOS/ADIOS2File.hpp

@@ -177,7 +177,7 @@ class ADIOS2File
      * IO.
      */
     std::string m_IOName;
-    adios2::ADIOS &m_ADIOS;
+    adios2::ADIOS m_ADIOS;
     adios2::IO m_IO;
     /**
      * The default queue for deferred actions.
@@ -421,6 +421,14 @@ class ADIOS2File
         return m_attributes;
     }
 
+    // read operators can (currently) not be specified per dataset, so parse
+    // them once and then buffer them
+    std::vector<adios_defs::ParameterizedOperator> readOperators;
+    std::map<std::string, adios2::Operator> m_operators;
+
+    std::optional<adios2::Operator>
+    getCompressionOperator(std::string const &compression);
+
 private:
     ADIOS2IOHandlerImpl *m_impl;
     std::optional<adios2::Engine> m_engine; //! ADIOS engine

include/openPMD/IO/ADIOS/ADIOS2IOHandler.hpp

@@ -221,7 +221,6 @@ class ADIOS2IOHandlerImpl
     FlushTarget m_flushTarget = FlushTarget::Disk;
 
 private:
-    adios2::ADIOS m_ADIOS;
 #if openPMD_HAVE_MPI
     std::optional<MPI_Comm> m_communicator;
 #endif
@@ -296,16 +295,6 @@ class ADIOS2IOHandlerImpl
 
     bool m_writeAttributesFromThisRank = true;
 
-    struct ParameterizedOperator
-    {
-        adios2::Operator op;
-        adios2::Params params;
-    };
-
-    // read operators can (currently) not be specified per dataset, so parse
-    // them once and then buffer them
-    std::vector<ParameterizedOperator> readOperators;
-
     json::TracingJSON m_config;
     std::optional<nlohmann::json> m_buffered_dataset_config;
     static json::TracingJSON nullvalue;
@@ -340,11 +329,12 @@ class ADIOS2IOHandlerImpl
      * @return first parameter: the operators, second parameters: whether
      * operators have been configured
      */
-    std::optional<std::vector<ParameterizedOperator>>
-    getOperators(json::TracingJSON config);
+    std::optional<std::vector<adios_defs::ParameterizedOperator>>
+    getOperators(detail::ADIOS2File &, json::TracingJSON config);
 
     // use m_config
-    std::optional<std::vector<ParameterizedOperator>> getOperators();
+    std::optional<std::vector<adios_defs::ParameterizedOperator>>
+    getOperators(detail::ADIOS2File &);
 
     enum class Shape
     {
@@ -356,9 +346,10 @@ class ADIOS2IOHandlerImpl
     auto parseDatasetConfig(
         Parameter const &,
         Writable *,
+        detail::ADIOS2File &filedata,
         std::string const &varName,
-        std::vector<ParameterizedOperator> default_operators = {})
-        -> std::tuple<std::vector<ParameterizedOperator>, Shape>;
+        std::vector<adios_defs::ParameterizedOperator> default_operators = {})
+        -> std::tuple<std::vector<adios_defs::ParameterizedOperator>, Shape>;
 
     std::string fileSuffix(bool verbose = true) const;
 
@@ -385,8 +376,6 @@ class ADIOS2IOHandlerImpl
     std::unordered_map<InvalidatableFile, std::unique_ptr<detail::ADIOS2File>>
         m_fileData;
 
-    std::map<std::string, adios2::Operator> m_operators;
-
     // Overrides from AbstractIOHandlerImplCommon.
 
     std::string
@@ -398,9 +387,6 @@ class ADIOS2IOHandlerImpl
 
     // Helper methods.
 
-    std::optional<adios2::Operator>
-    getCompressionOperator(std::string const &compression);
-
     /*
      * The name of the ADIOS2 variable associated with this Writable.
      * To be used for Writables that represent a dataset.
@@ -630,8 +616,7 @@ namespace detail
             std::string const &varName,
             Parameter<Operation::OPEN_DATASET> &parameters,
             std::optional<size_t> stepSelection,
-            std::vector<ADIOS2IOHandlerImpl::ParameterizedOperator> const
-                &operators);
+            std::vector<adios_defs::ParameterizedOperator> const &operators);
 
         static constexpr char const *errorMsg = "ADIOS2: openDataset()";
     };
@@ -657,8 +642,7 @@ namespace detail
         static void call(
             adios2::IO &IO,
             std::string const &name,
-            std::vector<ADIOS2IOHandlerImpl::ParameterizedOperator> const
-                &compressions,
+            std::vector<adios_defs::ParameterizedOperator> const &compressions,
             adios2::Dims const &shape = adios2::Dims(),
             adios2::Dims const &start = adios2::Dims(),
             adios2::Dims const &count = adios2::Dims(),

src/IO/ADIOS/ADIOS2File.cpp

@@ -217,25 +217,43 @@ ADIOS2File::ADIOS2File(
     InvalidatableFile file,
     adios_defs::OpenFileAs openFileAs)
     : m_file(impl.fullPath(std::move(file)))
-    , m_ADIOS(impl.m_ADIOS)
+#if openPMD_HAVE_MPI
+    , m_ADIOS([&]() {
+        if (impl.m_communicator.has_value())
+        {
+            return adios2::ADIOS(*impl.m_communicator);
+        }
+        else
+        {
+            return adios2::ADIOS();
+        }
+    }())
+#else
+    , m_ADIOS2()
+#endif
     , m_impl(&impl)
 {
     // Declaring these members in the constructor body to avoid
     // initialization order hazards. Need the IO_ prefix since in some
     // situation there seems to be trouble with number-only IO names
     m_mode = impl.adios2AccessMode(m_file, openFileAs);
     create_IO();
+
     if (!m_IO)
     {
         throw std::runtime_error(
             "[ADIOS2] Internal error: Failed declaring ADIOS2 IO object "
             "for file " +
             m_file);
     }
-    else
+
+    auto operators = impl.getOperators(*this);
+    if (operators)
     {
-        configure_IO();
+        readOperators = std::move(operators.value());
     }
+
+    configure_IO();
 }
 
 auto ADIOS2File::useGroupTable() const -> UseGroupTable
@@ -246,7 +264,7 @@ auto ADIOS2File::useGroupTable() const -> UseGroupTable
 void ADIOS2File::create_IO()
 {
     m_IOName = std::to_string(m_impl->nameCounter++);
-    m_IO = m_impl->m_ADIOS.DeclareIO("IO_" + m_IOName);
+    m_IO = m_ADIOS.DeclareIO("IO_" + m_IOName);
 }
 
 ADIOS2File::~ADIOS2File()
@@ -426,6 +444,42 @@ std::optional<size_t> ADIOS2File::stepSelection() const
     }
 }
 
+std::optional<adios2::Operator>
+ADIOS2File::getCompressionOperator(std::string const &compression)
+{
+    adios2::Operator res;
+    auto it = m_operators.find(compression);
+    if (it == m_operators.end())
+    {
+        try
+        {
+            res = m_ADIOS.DefineOperator(compression, compression);
+        }
+        catch (std::invalid_argument const &e)
+        {
+            std::cerr << "Warning: ADIOS2 backend does not support compression "
+                         "method "
+                      << compression << ". Continuing without compression."
+                      << "\nOriginal error: " << e.what() << std::endl;
+            return std::optional<adios2::Operator>();
+        }
+        catch (std::string const &s)
+        {
+            std::cerr << "Warning: ADIOS2 backend does not support compression "
+                         "method "
+                      << compression << ". Continuing without compression."
+                      << "\nOriginal error: " << s << std::endl;
+            return std::optional<adios2::Operator>();
+        }
+        m_operators.emplace(compression, res);
+    }
+    else
+    {
+        res = it->second;
+    }
+    return std::make_optional(adios2::Operator(res));
+}
+
 void ADIOS2File::configure_IO_Read()
 {
     bool upfrontParsing = supportsUpfrontParsing(