Refine flush levels for span clearing

openPMD · Jan 22, 2021 · 9c830a3 · 9c830a3
1 parent d585ab0
commit 9c830a3
Show file tree

Hide file tree

Showing 7 changed files with 75 additions and 31 deletions.
diff --git a/include/openPMD/IO/ADIOS/ADIOS2IOHandler.hpp b/include/openPMD/IO/ADIOS/ADIOS2IOHandler.hpp
@@ -875,6 +875,7 @@ namespace detail
         adios2::ADIOS & m_ADIOS;
         adios2::IO m_IO;
         std::vector< std::unique_ptr< BufferedAction > > m_buffer;
+        std::vector< std::unique_ptr< BufferedAction > > m_alreadyEnqueued;
         adios2::Mode m_mode;
         std::vector< std::unique_ptr< I_UpdateSpan > > m_updateSpans;
         detail::WriteDataset const m_writeDataset;
@@ -935,15 +936,15 @@ namespace detail
          */
         template< typename F >
         void
-        flush( F && performPutsGets, bool flushUnconditionally );
+        flush( FlushLevel, F && performPutsGets, bool flushUnconditionally );
 
         /**
          * Overload of flush() that uses adios2::Engine::Perform(Puts|Gets)
          * and does not flush unconditionally.
          *
          */
         void
-        flush();
+        flush( FlushLevel );
 
         /**
          * @brief Begin or end an ADIOS step.

diff --git a/include/openPMD/IO/AbstractIOHandler.hpp b/include/openPMD/IO/AbstractIOHandler.hpp
@@ -62,8 +62,11 @@ class unsupported_data_error : public std::runtime_error
  */
 enum class FlushLevel : unsigned char
 {
+    UserFlush,
     /**
      * Default mode, flush everything that can be flushed
+     * Does not need to uphold user-level guarantees about clearing and filling
+     * buffers. Spans must not yet be read.
      */
     FlushEverything,
     /**

diff --git a/include/openPMD/Series.hpp b/include/openPMD/Series.hpp
@@ -63,6 +63,7 @@ class WriteIterations;
  */
 class Series : public Attributable
 {
+    friend class Attributable;
     friend class Iteration;
     friend class SeriesIterator;
 
@@ -303,7 +304,9 @@ class Series : public Attributable
     void init(std::shared_ptr< AbstractIOHandler >, std::unique_ptr< ParsedInput >);
     void initDefaults();
     std::future< void > flush_impl(
-        iterations_iterator begin, iterations_iterator end );
+        iterations_iterator begin,
+        iterations_iterator end,
+        FlushLevel );
     void flushFileBased( iterations_iterator begin, iterations_iterator end );
     void flushGroupBased( iterations_iterator begin, iterations_iterator end );
     void flushMeshesPath();

diff --git a/src/IO/ADIOS/ADIOS2IOHandler.cpp b/src/IO/ADIOS/ADIOS2IOHandler.cpp
@@ -231,10 +231,7 @@ ADIOS2IOHandlerImpl::flush()
     {
         if ( m_dirty.find( p.first ) != m_dirty.end( ) )
         {
-            if( m_handler->m_flushLevel == FlushLevel::FlushEverything )
-            {
-                p.second->flush();
-            }
+            p.second->flush( m_handler->m_flushLevel );
         }
         else
         {
@@ -446,6 +443,7 @@ ADIOS2IOHandlerImpl::closeFile(
              * of it.
              */
             it->second->flush(
+                FlushLevel::UserFlush,
                 []( detail::BufferedActions & ba, adios2::Engine & ) {
                     ba.finalize();
                 },
@@ -1906,7 +1904,8 @@ namespace detail
 
     template< typename F >
     void
-    BufferedActions::flush( F && performPutGets, bool flushUnconditionally )
+    BufferedActions::flush(
+        FlushLevel level, F && performPutGets, bool flushUnconditionally )
     {
         if( streamStatus == StreamStatus::StreamOver )
         {
@@ -1941,17 +1940,36 @@ namespace detail
             ba->run( *this );
         }
 
-        performPutGets( *this, eng );
-
-        m_updateSpans.clear();
+        switch( level )
+        {
+            case FlushLevel::UserFlush:
+                performPutGets( *this, eng );
+                m_updateSpans.clear();
+                m_buffer.clear();
+                m_alreadyEnqueued.clear();
+                break;
 
-        m_buffer.clear();
+            case FlushLevel::FlushEverything:
+            case FlushLevel::SkeletonOnly:
+                /*
+                 * Tasks have been given to ADIOS2, but we don't flush them
+                 * yet. So, move everything to m_alreadyEnqueued to avoid
+                 * use-after-free.
+                 */
+                for( auto & task : m_buffer )
+                {
+                    m_alreadyEnqueued.emplace_back( std::move( task ) );
+                }
+                m_buffer.clear();
+                break;
+        }
     }
 
     void
-    BufferedActions::flush()
+    BufferedActions::flush( FlushLevel level )
     {
         flush(
+            level,
             []( BufferedActions & ba, adios2::Engine & eng ) {
                 switch( ba.m_mode )
                 {
@@ -1984,7 +2002,7 @@ namespace detail
         // sic! no else
         if( streamStatus == StreamStatus::NoStream )
         {
-            flush();
+            flush( FlushLevel::UserFlush );
             return AdvanceStatus::OK;
         }
         switch( mode )
@@ -2005,6 +2023,7 @@ namespace detail
                     getEngine().BeginStep();
                 }
                 flush(
+                    FlushLevel::UserFlush,
                     []( BufferedActions &, adios2::Engine & eng ) {
                         eng.EndStep();
                     },
@@ -2025,6 +2044,7 @@ namespace detail
                 if( streamStatus != StreamStatus::DuringStep )
                 {
                     flush(
+                        FlushLevel::UserFlush,
                         [ &adiosStatus ](
                             BufferedActions &, adios2::Engine & engine ) {
                             adiosStatus = engine.BeginStep();

diff --git a/src/Iteration.cpp b/src/Iteration.cpp
@@ -161,7 +161,7 @@ Iteration::close( bool _flush )
             auto end = begin;
             ++end;
 
-            s->flush_impl( begin, end );
+            s->flush_impl( begin, end, FlushLevel::UserFlush );
         }
     }
     else
@@ -186,7 +186,7 @@ Iteration::open()
     ++end;
     // set dirty, so Series::flush will open the file
     this->dirty() = true;
-    s->flush_impl( begin, end );
+    s->flush_impl( begin, end, FlushLevel::UserFlush );
     this->dirty() = false;
 
     return *this;

diff --git a/src/Series.cpp b/src/Series.cpp
@@ -358,7 +358,7 @@ Series::backend() const
 void
 Series::flush()
 {
-    flush_impl( iterations.begin(), iterations.end() );
+    flush_impl( iterations.begin(), iterations.end(), FlushLevel::UserFlush );
 }
 
 ReadIterations
@@ -506,7 +506,10 @@ Series::initDefaults()
 }
 
 std::future< void >
-Series::flush_impl( iterations_iterator begin, iterations_iterator end )
+Series::flush_impl(
+    iterations_iterator begin,
+    iterations_iterator end,
+    FlushLevel level )
 {
     switch( *m_iterationEncoding )
     {
@@ -519,7 +522,18 @@ Series::flush_impl( iterations_iterator begin, iterations_iterator end )
             break;
     }
 
-    return IOHandler->flush();
+    IOHandler->m_flushLevel = level;
+    try
+    {
+        auto res = IOHandler->flush();
+        IOHandler->m_flushLevel = FlushLevel::FlushEverything;
+        return res;
+    }
+    catch( ... )
+    {
+        IOHandler->m_flushLevel = FlushLevel::FlushEverything;
+        throw;
+    }
 }
 
 void
@@ -1139,7 +1153,17 @@ Series::advance(
     // We cannot call Series::flush now, since the IO handler is still filled
     // from calling flush(Group|File)based, but has not been emptied yet
     // Do that manually
-    IOHandler->flush();
+    IOHandler->m_flushLevel = FlushLevel::UserFlush;
+    try
+    {
+        IOHandler->flush();
+    }
+    catch( ... )
+    {
+        IOHandler->m_flushLevel = FlushLevel::FlushEverything;
+        throw;
+    }
+    IOHandler->m_flushLevel = FlushLevel::FlushEverything;
 
     return *param.status;
 }

diff --git a/src/backend/Attributable.cpp b/src/backend/Attributable.cpp
@@ -158,17 +158,10 @@ Attributable::flushSeries( FlushLevel flushLevel)
     }
     Series & series =
         auxiliary::deref_dynamic_cast< Series >( findSeries->attributable );
-    IOHandler->m_flushLevel = flushLevel;
-    try
-    {
-        series.flush();
-    }
-    catch(...)
-    {
-        IOHandler->m_flushLevel = FlushLevel::FlushEverything;
-        throw;
-    }
-    IOHandler->m_flushLevel = FlushLevel::FlushEverything;
+    series.flush_impl(
+        series.iterations.begin(),
+        series.iterations.end(),
+        flushLevel );
 }
 
 void