Store /data/snapshot attribute in groupbased layout too

Renamed from /data/__snapshot__

Only store it if the backend supports step.
If the attribute is not found while reading, the old Streaming API logic
is used as a fallback (proceed through the iterations linearly).
If it is found, the iteration to be returned in the current streaming
step is dictated by it.

Consistent naming: snapshot instead of __step__

Avoid infinite loops if there is no stream active

The backends will explicitly report in their status if it is currently
random-accessing. The frontend will then fall back to stepping linearly.

Author: franzpoeschel

include/openPMD/IO/ADIOS/ADIOS2IOHandler.hpp

@@ -1216,13 +1216,6 @@ namespace detail
         void
         invalidateVariablesMap();
 
-    private:
-        ADIOS2IOHandlerImpl * m_impl;
-        auxiliary::Option< adios2::Engine > m_engine; //! ADIOS engine
-        /**
-         * The ADIOS2 engine type, to be passed to adios2::IO::SetEngine
-         */
-        std::string m_engineType;
         /*
          * streamStatus is NoStream for file-based ADIOS engines.
          * This is relevant for the method BufferedActions::requireActiveStep,
@@ -1313,6 +1306,14 @@ namespace detail
             Undecided
         };
         StreamStatus streamStatus = StreamStatus::OutsideOfStep;
+
+    private:
+        ADIOS2IOHandlerImpl * m_impl;
+        auxiliary::Option< adios2::Engine > m_engine; //! ADIOS engine
+        /**
+         * The ADIOS2 engine type, to be passed to adios2::IO::SetEngine
+         */
+        std::string m_engineType;
         adios2::StepStatus m_lastStepStatus = adios2::StepStatus::OK;
 
         /**

include/openPMD/IO/AbstractIOHandlerImpl.hpp

@@ -156,8 +156,10 @@ class AbstractIOHandlerImpl
    * The return status code shall be stored as parameters.status.
    */
   virtual void
-  advance( Writable *, Parameter< Operation::ADVANCE > & )
-  {}
+  advance( Writable *, Parameter< Operation::ADVANCE > & parameters )
+  {
+    *parameters.status = AdvanceStatus::RANDOMACCESS;
+  }
 
   /** Close an openPMD group.
    *
@@ -336,6 +338,11 @@ class AbstractIOHandlerImpl
    * The attribute should be of datatype parameters.dtype.
    * Any existing attribute with the same name should be overwritten. If possible, only the value should be changed if the datatype stays the same.
    * The attribute should be written to physical storage after the operation completes successfully.
+   * If the parameter changesOverSteps is true, then the attribute must be able
+   * to hold different values across IO steps. If the backend does not support
+   * IO steps in such a way, the attribute should not be written.
+   * (IO steps are an optional backend feature and the frontend must implement
+   * fallback measures in such a case)
    * All datatypes of Datatype should be supported in a type-safe way.
    */
   virtual void writeAttribute(Writable*, Parameter< Operation::WRITE_ATT > const&) = 0;

include/openPMD/IO/IOTask.hpp

@@ -468,7 +468,8 @@ struct OPENPMDAPI_EXPORT Parameter< Operation::WRITE_ATT > : public AbstractPara
 {
     Parameter() = default;
     Parameter(Parameter const & p) : AbstractParameter(),
-        name(p.name), dtype(p.dtype), resource(p.resource) {}
+        name(p.name), dtype(p.dtype), changesOverSteps(p.changesOverSteps),
+        resource(p.resource) {}
 
     std::unique_ptr< AbstractParameter >
     clone() const override
@@ -479,6 +480,7 @@ struct OPENPMDAPI_EXPORT Parameter< Operation::WRITE_ATT > : public AbstractPara
 
     std::string name = "";
     Datatype dtype = Datatype::UNDEFINED;
+    bool changesOverSteps = false;
     Attribute::resource resource;
 };
 

include/openPMD/Iteration.hpp

@@ -29,6 +29,8 @@
 #include "openPMD/ParticleSpecies.hpp"
 #include "openPMD/Streaming.hpp"
 
+#include <deque>
+#include <tuple>
 
 namespace openPMD
 {
@@ -252,15 +254,38 @@ class Iteration : public LegacyAttributable
             std::make_shared< auxiliary::Option< DeferredParseAccess > >(
                 auxiliary::Option< DeferredParseAccess >() );
 
+    struct BeginStepStatus
+    {
+        using AvailableIterations_t =
+            auxiliary::Option< std::deque< uint64_t > >;
+
+        AdvanceStatus stepStatus{};
+        /*
+         * If the iteration attribute `snapshot` is present, the value of that
+         * attribute. Otherwise empty.
+         */
+        AvailableIterations_t iterationsInOpenedStep;
+
+        inline operator AdvanceStatus() const
+        {
+            return stepStatus;
+        }
+        inline
+        operator std::tuple< AdvanceStatus &, AvailableIterations_t & >()
+        {
+            return std::tuple< AdvanceStatus &, AvailableIterations_t & >{
+                stepStatus, iterationsInOpenedStep };
+        }
+    };
+
     /**
      * @brief Begin an IO step on the IO file (or file-like object)
      *        containing this iteration. In case of group-based iteration
      *        layout, this will be the complete Series.
      *
      * @return AdvanceStatus
      */
-    AdvanceStatus
-    beginStep();
+    BeginStepStatus beginStep();
 
     /**
      * @brief End an IO step on the IO file (or file-like object)

include/openPMD/ReadIterations.hpp

@@ -23,6 +23,9 @@
 #include "openPMD/Iteration.hpp"
 #include "openPMD/Series.hpp"
 
+#include <deque>
+#include <iostream>
+
 namespace openPMD
 {
 /**
@@ -54,7 +57,8 @@ class SeriesIterator
     using maybe_series_t = auxiliary::Option< Series >;
 
     maybe_series_t m_series;
-    iteration_index_t m_currentIteration = 0;
+    std::deque< iteration_index_t > m_iterationsInCurrentStep;
+    uint64_t m_currentIteration{};
 
 public:
     //! construct the end() iterator
@@ -71,6 +75,21 @@ class SeriesIterator
     bool operator!=( SeriesIterator const & other ) const;
 
     static SeriesIterator end();
+
+private:
+    inline bool setCurrentIteration()
+    {
+        if( m_iterationsInCurrentStep.empty() )
+        {
+            std::cerr << "[ReadIterations] Encountered a step without "
+                         "iterations. Closing the Series."
+                      << std::endl;
+            *this = end();
+            return false;
+        }
+        m_currentIteration = *m_iterationsInCurrentStep.begin();
+        return true;
+    }
 };
 
 /**

include/openPMD/Series.hpp

@@ -41,7 +41,9 @@
 #   include <mpi.h>
 #endif
 
+#include <deque>
 #include <map>
+#include <set>
 #include <string>
 
 // expose private and protected members for invasive testing
@@ -80,6 +82,11 @@ class SeriesData : public AttributableData
     Container< Iteration, uint64_t > iterations{};
 
     auxiliary::Option< 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;
     auxiliary::Option< std::string > m_overrideFilebasedFilename;
     std::string m_name;
     std::string m_filenamePrefix;
@@ -383,8 +390,11 @@ class SeriesInterface : public AttributableInterface
      * Note on re-parsing of a Series:
      * If init == false, the parsing process will seek for new
      * Iterations/Records/Record Components etc.
+     * If series.iterations contains the attribute `snapshot`, returns its
+     * value (will only be true in variable-based iteration encoding).
      */
-    void readGorVBased( bool init = true );
+    auxiliary::Option< std::deque< uint64_t > >
+    readGorVBased( bool init = true );
     void readBase();
     std::string iterationFilename( uint64_t i );
 
@@ -436,6 +446,8 @@ class SeriesInterface : public AttributableInterface
         internal::AttributableData & file,
         iterations_iterator it,
         Iteration & iteration );
+
+    void flushStep( bool doFlush );
 }; // SeriesInterface
 
 namespace internal

include/openPMD/Streaming.hpp

@@ -19,8 +19,9 @@ namespace openPMD
  */
 enum class AdvanceStatus : unsigned char
 {
-    OK,  /* stream goes on */
-    OVER /* stream is over */
+    OK,   /* stream goes on */
+    OVER, /* stream is over */
+    RANDOMACCESS /* there is no stream, it will never be over */
 };
 
 /**

src/IO/ADIOS/ADIOS2IOHandler.cpp

@@ -602,6 +602,11 @@ void ADIOS2IOHandlerImpl::writeAttribute(
     switch( attributeLayout() )
     {
         case AttributeLayout::ByAdiosAttributes:
+            if( parameters.changesOverSteps )
+            {
+                // cannot do this
+                return;
+            }
             switchType< detail::OldAttributeWriter >(
                 parameters.dtype,
                 this,
@@ -619,6 +624,13 @@ void ADIOS2IOHandlerImpl::writeAttribute(
             auto prefix = filePositionToString( pos );
 
             auto & filedata = getFileData( file, IfFileNotOpen::ThrowError );
+            if( parameters.changesOverSteps &&
+                filedata.streamStatus ==
+                    detail::BufferedActions::StreamStatus::NoStream )
+            {
+                // cannot do this
+                return;
+            }
             filedata.invalidateAttributesMap();
             m_dirty.emplace( std::move( file ) );
 
@@ -2721,7 +2733,7 @@ namespace detail
             m_IO.DefineAttribute< bool_representation >(
                 ADIOS2Defaults::str_usesstepsAttribute, 0 );
             flush( FlushLevel::UserFlush, /* writeAttributes = */ false );
-            return AdvanceStatus::OK;
+            return AdvanceStatus::RANDOMACCESS;
         }
 
         m_IO.DefineAttribute< bool_representation >(

src/IO/ADIOS/CommonADIOS1IOHandler.cpp

@@ -1002,6 +1002,11 @@ void
 CommonADIOS1IOHandlerImpl< ChildClass >::writeAttribute(Writable* writable,
                                     Parameter< Operation::WRITE_ATT > const& parameters)
 {
+    if( parameters.changesOverSteps )
+    {
+        // cannot do this
+        return;
+    }
     if( m_handler->m_backendAccess == Access::READ_ONLY )
         throw std::runtime_error("[ADIOS1] Writing an attribute in a file opened as read only is not possible.");
 

src/IO/HDF5/HDF5IOHandler.cpp

@@ -1033,6 +1033,11 @@ void
 HDF5IOHandlerImpl::writeAttribute(Writable* writable,
                                   Parameter< Operation::WRITE_ATT > const& parameters)
 {
+    if( parameters.changesOverSteps )
+    {
+        // cannot do this
+        return;
+    }
     if( m_handler->m_backendAccess == Access::READ_ONLY )
         throw std::runtime_error("[HDF5] Writing an attribute in a file opened as read only is not possible.");