Store /data/__step__ attribute in groupbased layout too

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.

Author: franzpoeschel

include/openPMD/IO/ADIOS/ADIOS2IOHandler.hpp

@@ -1156,13 +1156,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,
@@ -1253,6 +1246,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/IOTask.hpp

@@ -432,7 +432,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
@@ -443,6 +444,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
 {
@@ -164,7 +166,7 @@ class Iteration : public LegacyAttributable
 
     void flushFileBased(std::string const&, uint64_t);
     void flushGroupBased(uint64_t);
-    void flushVariableBased(uint64_t);
+    void flushVariableBased();
     void flush();
     void deferRead( DeferredRead );
     /*
@@ -227,15 +229,34 @@ class Iteration : public LegacyAttributable
         std::make_shared< auxiliary::Option< DeferredRead > >(
             auxiliary::Option< DeferredRead >() );
 
+    struct BeginStepStatus
+    {
+        using AvailableIterations_t =
+            auxiliary::Option< std::deque< uint64_t > >;
+
+        AdvanceStatus stepStatus{};
+        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{};
 
     //! construct the end() iterator
     SeriesIterator();
@@ -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
@@ -86,6 +88,7 @@ class SeriesData : public AttributableData
 
 OPENPMD_private :
     auxiliary::Option< WriteIterations > m_writeIterations;
+    std::set< uint64_t > m_currentlyActiveIterations;
     auxiliary::Option< std::string > m_overrideFilebasedFilename;
     std::string m_name;
     std::string m_filenamePrefix;
@@ -355,7 +358,8 @@ class SeriesImpl : public AttributableImpl
      * as of yet. Such a facility will be required upon implementing things such
      * as resizable datasets.
      */
-    void readGorVBased( bool init = true );
+    auxiliary::Option< std::deque< uint64_t > >
+    readGorVBased( bool init = true );
     void readBase();
     std::string iterationFilename( uint64_t i );
     void openIteration( uint64_t index, Iteration iteration );
@@ -388,6 +392,8 @@ class SeriesImpl : public AttributableImpl
         internal::AttributableData & file,
         iterations_iterator it,
         Iteration & iteration );
+
+    void flushStep( bool doFlush );
 }; // SeriesImpl
 
 namespace internal

src/IO/ADIOS/ADIOS2IOHandler.cpp

@@ -596,6 +596,11 @@ void ADIOS2IOHandlerImpl::writeAttribute(
     switch( attributeLayout() )
     {
         case AttributeLayout::ByAdiosAttributes:
+            if( parameters.changesOverSteps )
+            {
+                // cannot do this
+                return;
+            }
             switchType(
                 parameters.dtype,
                 detail::OldAttributeWriter(),
@@ -614,6 +619,13 @@ void ADIOS2IOHandlerImpl::writeAttribute(
             auto prefix = filePositionToString( pos );
 
             auto & filedata = getFileData( file );
+            if( parameters.changesOverSteps &&
+                filedata.streamStatus ==
+                    detail::BufferedActions::StreamStatus::NoStream )
+            {
+                // cannot do this
+                return;
+            }
             filedata.invalidateAttributesMap();
             m_dirty.emplace( std::move( file ) );
 

src/IO/ADIOS/CommonADIOS1IOHandler.cpp

@@ -960,6 +960,11 @@ void
 CommonADIOS1IOHandlerImpl::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

@@ -904,6 +904,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.");
 

src/IO/JSON/JSONIOHandlerImpl.cpp

@@ -914,6 +914,11 @@ namespace openPMD
         Parameter< Operation::WRITE_ATT > const & parameter
     )
     {
+        if( parameter.changesOverSteps )
+        {
+            // cannot do this
+            return;
+        }
         if(m_handler->m_backendAccess == Access::READ_ONLY )
         {
             throw std::runtime_error( "[JSON] Creating a dataset in a file opened as read only is not possible." );

src/Iteration.cpp

@@ -259,15 +259,14 @@ Iteration::flushGroupBased(uint64_t i)
 }
 
 void
-Iteration::flushVariableBased( uint64_t i )
+Iteration::flushVariableBased()
 {
     if( !written() )
     {
         /* create iteration path */
         Parameter< Operation::OPEN_PATH > pOpen;
         pOpen.path = "";
         IOHandler()->enqueue( IOTask( this, pOpen ) );
-        this->setAttribute( "__step__", i );
     }
 
     flush();
@@ -511,9 +510,9 @@ void Iteration::read_impl( std::string const & groupPath )
     readAttributes();
 }
 
-AdvanceStatus
-Iteration::beginStep()
+auto Iteration::beginStep() -> BeginStepStatus
 {
+    BeginStepStatus res;
     using IE = IterationEncoding;
     auto & series = retrieveSeries();
     // Initialize file with this to quiet warnings
@@ -533,7 +532,8 @@ Iteration::beginStep()
         AdvanceMode::BEGINSTEP, *file, series.indexOf( *this ), *this );
     if( status != AdvanceStatus::OK )
     {
-        return status;
+        res.stepStatus = status;
+        return res;
     }
 
     // re-read -> new datasets might be available
@@ -548,12 +548,13 @@ Iteration::beginStep()
         auto newType =
             const_cast< Access * >( &this->IOHandler()->m_frontendAccess );
         *newType = Access::READ_WRITE;
-        series.readGorVBased( false );
+        res.iterationsInOpenedStep = series.readGorVBased( false );
         *newType = oldType;
         series.iterations.written() = previous;
     }
 
-    return status;
+    res.stepStatus = status;
+    return res;
 }
 
 void
@@ -577,6 +578,7 @@ Iteration::endStep()
     // @todo filebased check
     series.advance(
         AdvanceMode::ENDSTEP, *file, series.indexOf( *this ), *this );
+    series.m_currentlyActiveIterations.clear();
 }
 
 StepStatus