Workaround for independent writes to Iterations in parallel, better d…

…etection of BP5 which in turn uncovers more instances of the first issue (#1619)

* retrieveIteration: return both Series and Iteration

* Optimize implementation

* seriesFlush(): Mark containing Iteration as dirty

* Add failing test

* Backend implementation

* Add documentation

* Add ADIOS2 v2.10 define and use that for BP5 check

* Ask the engine if it is BP5 for BP5-specific features

* write_test_zero_extent: require flush to buffer

Somehow PerformDataWrite() leads to trouble with this pattern.

* Revert "write_test_zero_extent: require flush to buffer"

This reverts commit 36597bd.
No longer needed after rebasing on fix-iteration-flush

* Fix hipace_like_write test

It used Series::flush non-collectively

* Also ensure all ranks flush in group/variable encoding

docs/source/details/mpi.rst

@@ -49,6 +49,12 @@ Functionality                Behavior           Description
 .. [4] We usually open iterations delayed on first access. This first access is usually the ``flush()`` call after a ``storeChunk``/``loadChunk`` operation. If the first access is non-collective, an explicit, collective ``Iteration::open()`` can be used to have the files already open.
        Alternatively, iterations might be accessed for the first time by immediate operations such as ``::availableChunks()``.
 
+.. warning::
+
+  The openPMD-api will by default flush only those Iterations which are dirty, i.e. have been written to.
+  This is somewhat unfortunate in parallel setups since only the dirty status of the current MPI rank can be considered.
+  As a workaround, use ``Attributable::seriesFlush()`` on an Iteration (or an object contained within an Iteration) to force flush that Iteration regardless of its dirty status.
+
 .. tip::
 
    Just because an operation is independent does not mean it is allowed to be inconsistent.

docs/source/dev/design.rst

@@ -23,7 +23,7 @@ Therefore, enabling users to handle hierarchical, self-describing file formats w
 
 .. literalinclude:: IOTask.hpp
    :language: cpp
-   :lines: 48-78
+   :lines: 50-81
 
 Every task is designed to be a fully self-contained description of one such atomic operation. By describing a required minimal step of work (without any side-effect), these operations are the foundation of the unified handling mechanism across suitable file formats.
 The actual low-level exchange of data is implemented in ``IOHandlers``, one per file format (possibly two if handlingi MPI-parallel work is possible and requires different behaviour).

include/openPMD/IO/ADIOS/ADIOS2IOHandler.hpp

@@ -206,6 +206,8 @@ class ADIOS2IOHandlerImpl
     void
     deregister(Writable *, Parameter<Operation::DEREGISTER> const &) override;
 
+    void touch(Writable *, Parameter<Operation::TOUCH> const &) override;
+
     /**
      * @brief The ADIOS2 access type to chose for Engines opened
      * within this instance.

include/openPMD/IO/ADIOS/macros.hpp

@@ -19,7 +19,10 @@
 #define openPMD_HAS_ADIOS_2_9                                                  \
     (ADIOS2_VERSION_MAJOR * 100 + ADIOS2_VERSION_MINOR >= 209)
 
-#if defined(ADIOS2_HAVE_BP5) || openPMD_HAS_ADIOS_2_9
+#define openPMD_HAS_ADIOS_2_10                                                 \
+    (ADIOS2_VERSION_MAJOR * 100 + ADIOS2_VERSION_MINOR >= 210)
+
+#if defined(ADIOS2_HAVE_BP5) || openPMD_HAS_ADIOS_2_10
 // ADIOS2 v2.10 no longer defines this
 #define openPMD_HAVE_ADIOS2_BP5 1
 #else

include/openPMD/IO/AbstractIOHandlerImpl.hpp

@@ -395,6 +395,11 @@ class AbstractIOHandlerImpl
     virtual void
     deregister(Writable *, Parameter<Operation::DEREGISTER> const &param) = 0;
 
+    /** Treat this writable's file as open/active/dirty.
+     */
+    virtual void
+    touch(Writable *, Parameter<Operation::TOUCH> const &param) = 0;
+
     AbstractIOHandler *m_handler;
     bool m_verboseIOTasks = false;
 

include/openPMD/IO/HDF5/HDF5IOHandlerImpl.hpp

@@ -80,6 +80,7 @@ class HDF5IOHandlerImpl : public AbstractIOHandlerImpl
     void listAttributes(Writable *, Parameter<Operation::LIST_ATTS> &) override;
     void
     deregister(Writable *, Parameter<Operation::DEREGISTER> const &) override;
+    void touch(Writable *, Parameter<Operation::TOUCH> const &) override;
 
     std::unordered_map<Writable *, std::string> m_fileNames;
     std::unordered_map<std::string, hid_t> m_fileNamesWithID;

include/openPMD/IO/IOTask.hpp

@@ -48,20 +48,36 @@ Writable *getWritable(Attributable *);
 /** Type of IO operation between logical and persistent data.
  */
 OPENPMDAPI_EXPORT_ENUM_CLASS(Operation){
-    CREATE_FILE,      CHECK_FILE,     OPEN_FILE,     CLOSE_FILE,
+    CREATE_FILE,
+    CHECK_FILE,
+    OPEN_FILE,
+    CLOSE_FILE,
     DELETE_FILE,
 
-    CREATE_PATH,      CLOSE_PATH,     OPEN_PATH,     DELETE_PATH,
+    CREATE_PATH,
+    CLOSE_PATH,
+    OPEN_PATH,
+    DELETE_PATH,
     LIST_PATHS,
 
-    CREATE_DATASET,   EXTEND_DATASET, OPEN_DATASET,  DELETE_DATASET,
-    WRITE_DATASET,    READ_DATASET,   LIST_DATASETS, GET_BUFFER_VIEW,
+    CREATE_DATASET,
+    EXTEND_DATASET,
+    OPEN_DATASET,
+    DELETE_DATASET,
+    WRITE_DATASET,
+    READ_DATASET,
+    LIST_DATASETS,
+    GET_BUFFER_VIEW,
 
-    DELETE_ATT,       WRITE_ATT,      READ_ATT,      LIST_ATTS,
+    DELETE_ATT,
+    WRITE_ATT,
+    READ_ATT,
+    LIST_ATTS,
 
     ADVANCE,
     AVAILABLE_CHUNKS, //!< Query chunks that can be loaded in a dataset
-    DEREGISTER //!< Inform the backend that an object has been deleted.
+    DEREGISTER, //!< Inform the backend that an object has been deleted.
+    TOUCH //!< tell the backend that the file is to be considered active
 }; // note: if you change the enum members here, please update
    // docs/source/dev/design.rst
 
@@ -658,6 +674,23 @@ struct OPENPMDAPI_EXPORT Parameter<Operation::DEREGISTER>
     void const *former_parent = nullptr;
 };
 
+template <>
+struct OPENPMDAPI_EXPORT Parameter<Operation::TOUCH> : public AbstractParameter
+{
+    explicit Parameter() = default;
+
+    Parameter(Parameter const &) = default;
+    Parameter(Parameter &&) = default;
+
+    Parameter &operator=(Parameter const &) = default;
+    Parameter &operator=(Parameter &&) = default;
+
+    std::unique_ptr<AbstractParameter> to_heap() && override
+    {
+        return std::make_unique<Parameter<Operation::TOUCH>>(std::move(*this));
+    }
+};
+
 /** @brief Self-contained description of a single IO operation.
  *
  * Contained are

include/openPMD/IO/JSON/JSONIOHandlerImpl.hpp

@@ -240,6 +240,8 @@ class JSONIOHandlerImpl : public AbstractIOHandlerImpl
     void
     deregister(Writable *, Parameter<Operation::DEREGISTER> const &) override;
 
+    void touch(Writable *, Parameter<Operation::TOUCH> const &) override;
+
     std::future<void> flush();
 
 private:

include/openPMD/Iteration.hpp

@@ -130,6 +130,7 @@ class Iteration : public Attributable
     friend class Series;
     friend class WriteIterations;
     friend class SeriesIterator;
+    friend class internal::AttributableData;
 
 public:
     Iteration(Iteration const &) = default;

include/openPMD/Series.hpp

@@ -254,6 +254,7 @@ class Series : public Attributable
     friend class ReadIterations;
     friend class SeriesIterator;
     friend class internal::SeriesData;
+    friend class internal::AttributableData;
     friend class WriteIterations;
 
 public:

include/openPMD/backend/Attributable.hpp

@@ -53,6 +53,9 @@ class Series;
 
 namespace internal
 {
+    class IterationData;
+    class SeriesData;
+
     class AttributableData
     {
         friend class openPMD::Attributable;
@@ -74,6 +77,42 @@ namespace internal
          */
         Writable m_writable;
 
+        template <typename T>
+        T asInternalCopyOf()
+        {
+            auto *self = dynamic_cast<typename T::Data_t *>(this);
+            if (!self)
+            {
+                if constexpr (std::is_same_v<Series, T>)
+                {
+                    throw std::runtime_error(
+                        "[Attributable::retrieveSeries] Error when trying to "
+                        "retrieve the Series object. Note: An instance of the "
+                        "Series object must still exist when flushing. A "
+                        "common cause for this error is using a flush call on "
+                        "a handle (e.g. `Iteration::seriesFlush()`) when the "
+                        "original Series object has already gone out of "
+                        "scope.");
+                }
+                else
+                {
+                    throw std::runtime_error(
+
+                        "[AttributableData::asInternalCopyOf<T>] Error when "
+                        "trying to retrieve a containing object. Note: An "
+                        "instance of the Series object must still exist when "
+                        "flushing. A common cause for this error is using a "
+                        "flush call on a handle (e.g. "
+                        "`Iteration::seriesFlush()`) when the original Series "
+                        "object has already gone out of scope.");
+                }
+            }
+            T res;
+            res.setData(
+                std::shared_ptr<typename T::Data_t>(self, [](auto const *) {}));
+            return res;
+        }
+
     private:
         /**
          * The attributes defined by this Attributable.
@@ -207,6 +246,8 @@ class Attributable
      * of parents. This method will walk up the parent list until it reaches
      * an object that has no parent, which is the Series object, and flush()-es
      * it.
+     * If the Attributable is an Iteration or any object contained in an
+     * Iteration, that Iteration will be flushed regardless of its dirty status.
      *
      * @param backendConfig Further backend-specific instructions on how to
      *                      implement this flush call.
@@ -263,8 +304,13 @@ OPENPMD_protected
      * Throws an error otherwise, e.g., for Series objects.
      * @{
      */
-    Iteration const &containingIteration() const;
-    Iteration &containingIteration();
+    [[nodiscard]] auto containingIteration() const
+        -> std::pair<
+            std::optional<internal::IterationData const *>,
+            internal::SeriesData const *>;
+    auto containingIteration() -> std::pair<
+                                   std::optional<internal::IterationData *>,
+                                   internal::SeriesData *>;
     /** @} */
 
     void seriesFlush(internal::FlushParams const &);

src/IO/ADIOS/ADIOS2File.cpp

@@ -23,6 +23,7 @@
 #include "openPMD/Error.hpp"
 #include "openPMD/IO/ADIOS/ADIOS2IOHandler.hpp"
 #include "openPMD/auxiliary/Environment.hpp"
+#include "openPMD/auxiliary/StringManip.hpp"
 
 #if openPMD_USE_VERIFY
 #define VERIFY(CONDITION, TEXT)                                                \
@@ -1044,7 +1045,14 @@ void ADIOS2File::flush_impl(ADIOS2FlushParams flushParams, bool writeLatePuts)
                 performDataWrite = false;
                 break;
             }
-            performDataWrite = performDataWrite && m_engineType == "bp5";
+            performDataWrite = performDataWrite &&
+                (m_engineType == "bp5" ||
+                 /* this second check should be sufficient, but we leave the
+                    first check in as a safeguard against renamings in ADIOS2.
+                    Also do a lowerCase transform since the docstring of
+                    `Engine::Type()` claims that the return value is in
+                    lowercase, but for BP5 this does not seem true. */
+                 auxiliary::lowerCase(engine.Type()) == "bp5writer");
 
             if (performDataWrite)
             {

src/IO/ADIOS/ADIOS2IOHandler.cpp

@@ -895,6 +895,7 @@ void ADIOS2IOHandlerImpl::openFile(
     // lazy opening is deathly in parallel situations
     auto &fileData = getFileData(file, IfFileNotOpen::OpenImplicitly);
     *parameters.out_parsePreference = fileData.parsePreference;
+    m_dirty.emplace(std::move(file));
 }
 
 void ADIOS2IOHandlerImpl::closeFile(
@@ -1482,6 +1483,13 @@ void ADIOS2IOHandlerImpl::deregister(
     m_files.erase(writable);
 }
 
+void ADIOS2IOHandlerImpl::touch(
+    Writable *writable, Parameter<Operation::TOUCH> const &)
+{
+    auto file = refreshFileFromParent(writable, /* preferParentFile = */ false);
+    m_dirty.emplace(std::move(file));
+}
+
 adios2::Mode ADIOS2IOHandlerImpl::adios2AccessMode(std::string const &fullPath)
 {
     switch (m_handler->m_backendAccess)

src/IO/AbstractIOHandlerImpl.cpp

@@ -421,6 +421,14 @@ std::future<void> AbstractIOHandlerImpl::flush()
                 deregister(i.writable, parameter);
                 break;
             }
+            case O::TOUCH: {
+                auto &parameter =
+                    deref_dynamic_cast<Parameter<O::TOUCH>>(i.parameter.get());
+                writeToStderr(
+                    "[", i.writable->parent, "->", i.writable, "] DEREGISTER");
+                touch(i.writable, parameter);
+                break;
+            }
             }
         }
         catch (...)

src/IO/HDF5/HDF5IOHandler.cpp

@@ -2910,6 +2910,11 @@ void HDF5IOHandlerImpl::deregister(
     m_fileNames.erase(writable);
 }
 
+void HDF5IOHandlerImpl::touch(Writable *, Parameter<Operation::TOUCH> const &)
+{
+    // no-op
+}
+
 std::optional<HDF5IOHandlerImpl::File>
 HDF5IOHandlerImpl::getFile(Writable *writable)
 {

src/IO/JSON/JSONIOHandlerImpl.cpp

@@ -1039,6 +1039,13 @@ void JSONIOHandlerImpl::deregister(
     m_files.erase(writable);
 }
 
+void JSONIOHandlerImpl::touch(
+    Writable *writable, Parameter<Operation::TOUCH> const &)
+{
+    auto file = refreshFileFromParent(writable);
+    m_dirty.emplace(std::move(file));
+}
+
 auto JSONIOHandlerImpl::getFilehandle(File const &fileName, Access access)
     -> std::tuple<std::unique_ptr<FILEHANDLE>, std::istream *, std::ostream *>
 {

src/Iteration.cpp

@@ -22,6 +22,7 @@
 #include "openPMD/Dataset.hpp"
 #include "openPMD/Datatype.hpp"
 #include "openPMD/IO/AbstractIOHandler.hpp"
+#include "openPMD/IO/IOTask.hpp"
 #include "openPMD/Series.hpp"
 #include "openPMD/auxiliary/DerefDynamicCast.hpp"
 #include "openPMD/auxiliary/Filesystem.hpp"
@@ -315,6 +316,8 @@ void Iteration::flushVariableBased(
 
 void Iteration::flush(internal::FlushParams const &flushParams)
 {
+    Parameter<Operation::TOUCH> touch;
+    IOHandler()->enqueue(IOTask(&writable(), touch));
     if (access::readOnly(IOHandler()->m_frontendAccess))
     {
         for (auto &m : meshes)

src/Series.cpp

@@ -1403,6 +1403,7 @@ void Series::flushGorVBased(
     bool flushIOHandler)
 {
     auto &series = get();
+
     if (access::readOnly(IOHandler()->m_frontendAccess))
     {
         for (auto it = begin; it != end; ++it)
@@ -1432,6 +1433,8 @@ void Series::flushGorVBased(
             }
 
             // Phase 3
+            Parameter<Operation::TOUCH> touch;
+            IOHandler()->enqueue(IOTask(&writable(), touch));
             if (flushIOHandler)
             {
                 IOHandler()->flush(flushParams);
@@ -1510,6 +1513,8 @@ void Series::flushGorVBased(
         }
 
         flushAttributes(flushParams);
+        Parameter<Operation::TOUCH> touch;
+        IOHandler()->enqueue(IOTask(&writable(), touch));
         if (flushIOHandler)
         {
             IOHandler()->flush(flushParams);