Writing changes: Write current step(s) to snapshot attribute

Only set snapshot attribute if Iteration is not yet written

For v-based iteration encoding, the snapshot attribute is already being
set before this PR. Just add a comment there.

Also add missing <cstdint> includes

Co-authored-by: Axel Huebl <axel.huebl@plasma.ninja>

Author: franzpoeschel

include/openPMD/Series.hpp

@@ -37,8 +37,10 @@
 #include <mpi.h>
 #endif
 
+#include <cstdint>
 #include <map>
 #include <optional>
+#include <set>
 #include <string>
 
 // expose private and protected members for invasive testing
@@ -82,6 +84,12 @@ namespace internal
          * the same instance.
          */
         std::optional<WriteIterations> m_writeIterations;
+        /**
+         * For writing: Remember which iterations have been written in the
+         * currently active output step. Use this later when writing the
+         * snapshot attribute.
+         */
+        std::set<uint64_t> m_currentlyActiveIterations;
         /**
          * Needed if reading a single iteration of a file-based series.
          * Users may specify the concrete filename of one iteration instead of
@@ -627,6 +635,14 @@ OPENPMD_private
         internal::AttributableData &file,
         iterations_iterator it,
         Iteration &iteration);
+
+    /**
+     * @brief Called at the end of an IO step to store the iterations defined
+     *        in the IO step to the snapshot attribute.
+     *
+     * @param doFlush If true, flush the IO handler.
+     */
+    void flushStep(bool doFlush);
 }; // Series
 } // namespace openPMD
 

src/Iteration.cpp

@@ -282,6 +282,13 @@ void Iteration::flushVariableBased(
         Parameter<Operation::OPEN_PATH> pOpen;
         pOpen.path = "";
         IOHandler()->enqueue(IOTask(this, pOpen));
+        /*
+         * In v-based encoding, the snapshot attribute must always be written,
+         * so don't set the `changesOverSteps` flag of the IOTask here.
+         * Reason: Even in backends that don't support changing attributes,
+         * variable-based iteration encoding can be used to write one single
+         * iteration. Then, this attribute determines which iteration it is.
+         */
         this->setAttribute("snapshot", i);
     }
 

src/Series.cpp

@@ -732,11 +732,16 @@ void Series::flushFileBased(
             switch (openIterationIfDirty(it->first, it->second))
             {
                 using IO = IterationOpened;
+            case IO::RemainsClosed:
+                // we might need to proceed further if the close status is
+                // ClosedInFrontend
+                // hence no continue here
+                // otherwise, we might forget to close files physically
+                break;
             case IO::HasBeenOpened:
+                // continue below
                 it->second.flush(flushParams);
                 break;
-            case IO::RemainsClosed:
-                break;
             }
 
             // Phase 2
@@ -769,13 +774,21 @@ void Series::flushFileBased(
             case IO::HasBeenOpened: {
                 /* as there is only one series,
                  * emulate the file belonging to each iteration as not yet
-                 * written
+                 * written, even if the iteration itself is already written
+                 * (to ensure that the Series gets reassociated with the
+                 * current iteration)
                  */
                 written() = false;
                 series.iterations.written() = false;
 
                 dirty() |= it->second.dirty();
                 std::string filename = iterationFilename(it->first);
+
+                if(!it->second.written())
+                {
+                    series.m_currentlyActiveIterations.emplace(it->first);
+                }
+
                 it->second.flushFileBased(filename, it->first, flushParams);
 
                 series.iterations.flush(
@@ -831,11 +844,15 @@ void Series::flushGorVBased(
             switch (openIterationIfDirty(it->first, it->second))
             {
                 using IO = IterationOpened;
+            case IO::RemainsClosed:
+                // we might need to proceed further if the close status is
+                // ClosedInFrontend
+                // hence no continue here
+                break;
             case IO::HasBeenOpened:
+                // continue below
                 it->second.flush(flushParams);
                 break;
-            case IO::RemainsClosed:
-                break;
             }
 
             // Phase 2
@@ -895,6 +912,7 @@ void Series::flushGorVBased(
                 if (!it->second.written())
                 {
                     it->second.parent() = getWritable(&series.iterations);
+                    series.m_currentlyActiveIterations.emplace(it->first);
                 }
                 switch (iterationEncoding())
                 {
@@ -1458,9 +1476,15 @@ AdvanceStatus Series::advance(
          * opening an iteration's file by beginning a step on it.
          * So, return now.
          */
+        iteration.get().m_closed = internal::CloseStatus::ClosedInBackend;
         return AdvanceStatus::OK;
     }
 
+    if (mode == AdvanceMode::ENDSTEP)
+    {
+        flushStep(/* doFlush = */ false);
+    }
+
     Parameter<Operation::ADVANCE> param;
     if (itData.m_closed == internal::CloseStatus::ClosedTemporarily &&
         series.m_iterationEncoding == IterationEncoding::fileBased)
@@ -1517,6 +1541,32 @@ AdvanceStatus Series::advance(
     return *param.status;
 }
 
+void Series::flushStep(bool doFlush)
+{
+    auto &series = get();
+    if (!series.m_currentlyActiveIterations.empty())
+    {
+        /*
+         * Warning: changing attribute extents over time (probably) unsupported
+         * by this so far.
+         * Not (yet) needed as there is no way to pack several iterations within
+         * one IO step.
+         */
+        Parameter<Operation::WRITE_ATT> wAttr;
+        wAttr.changesOverSteps = true;
+        wAttr.name = "snapshot";
+        wAttr.resource = std::vector<unsigned long long>{
+            series.m_currentlyActiveIterations.begin(),
+            series.m_currentlyActiveIterations.end()};
+        wAttr.dtype = Datatype::VEC_ULONGLONG;
+        IOHandler()->enqueue(IOTask(&series.iterations, wAttr));
+        if (doFlush)
+        {
+            IOHandler()->flush(internal::defaultFlushParams);
+        }
+    }
+}
+
 auto Series::openIterationIfDirty(uint64_t index, Iteration iteration)
     -> IterationOpened
 {
@@ -1795,6 +1845,7 @@ namespace internal
             {
                 Series impl{{this, [](auto const *) {}}};
                 impl.flush();
+                impl.flushStep(/* doFlush = */ true);
             }
             if (m_writeIterations.has_value())
             {