WIP Runtime verification of flush level

Author: franzpoeschel

include/openPMD/IO/AbstractIOHandler.hpp

@@ -26,6 +26,7 @@
 #include "openPMD/IterationEncoding.hpp"
 #include "openPMD/config.hpp"
 #include "openPMD/version.hpp"
+#include <ostream>
 
 #if openPMD_HAVE_MPI
 #include <mpi.h>
@@ -81,6 +82,8 @@ enum class FlushLevel
     CreateOrOpenFiles
 };
 
+std::ostream &operator<<(std::ostream &, FlushLevel);
+
 namespace flush_level
 {
     inline constexpr auto global_flushpoint(FlushLevel fl)

include/openPMD/IO/AbstractIOHandlerImpl.hpp

@@ -39,7 +39,7 @@ class AbstractIOHandlerImpl
 
     virtual ~AbstractIOHandlerImpl() = default;
 
-    std::future<void> flush();
+    std::future<void> flush(FlushLevel);
 
     /**
      * Close the file corresponding with the writable and release file handles.

include/openPMD/IO/IOTask.hpp

@@ -35,6 +35,7 @@
 #include <cstddef>
 #include <memory>
 #include <optional>
+#include <ostream>
 #include <string>
 #include <utility>
 #include <variant>
@@ -89,6 +90,8 @@ OPENPMDAPI_EXPORT_ENUM_CLASS(Operation){
 }; // note: if you change the enum members here, please update
    // docs/source/dev/design.rst
 
+std::ostream &operator<<(std::ostream &os, Operation op);
+
 namespace internal
 {
     /*

include/openPMD/IO/JSON/JSONIOHandlerImpl.hpp

@@ -241,7 +241,7 @@ class JSONIOHandlerImpl : public AbstractIOHandlerImpl
 
     void touch(Writable *, Parameter<Operation::TOUCH> const &) override;
 
-    std::future<void> flush();
+    std::future<void> flush(internal::ParsedFlushParams &params);
 
 private:
 #if openPMD_HAVE_MPI

include/openPMD/RecordComponent.tpp

@@ -109,9 +109,7 @@ RecordComponent::storeChunk(Offset o, Extent e, F &&createBuffer)
         dCreate.name = Attributable::get().m_writable.ownKeyWithinParent;
         IOHandler()->enqueue(IOTask(this, dCreate));
 
-        Parameter<Operation::SET_WRITTEN> sWritten;
-        sWritten.target_status = true;
-        IOHandler()->enqueue(IOTask(this, sWritten));
+        setWritten(true, EnqueueAsynchronously::Yes);
     }
 
     Parameter<Operation::GET_BUFFER_VIEW> getBufferView;

include/openPMD/Series.hpp

@@ -982,7 +982,7 @@ OPENPMD_private
      *
      * @param doFlush If true, flush the IO handler.
      */
-    void flushStep(bool doFlush);
+    void flushStep(bool doFlush, FlushLevel l);
 
     /*
      * setIterationEncoding() should only be called by users of our public API,

src/IO/ADIOS/ADIOS2IOHandler.cpp

@@ -582,7 +582,7 @@ overrideFlushTarget(FlushTarget &inplace, FlushTarget new_val)
 std::future<void>
 ADIOS2IOHandlerImpl::flush(internal::ParsedFlushParams &flushParams)
 {
-    auto res = AbstractIOHandlerImpl::flush();
+    auto res = AbstractIOHandlerImpl::flush(flushParams.flushLevel);
 
     detail::ADIOS2File::ADIOS2FlushParams adios2FlushParams{
         flushParams.flushLevel, m_flushTarget};

src/IO/AbstractIOHandler.cpp

@@ -27,6 +27,29 @@
 
 #include <utility>
 
+namespace openPMD
+{
+std::ostream &operator<<(std::ostream &os, FlushLevel l)
+{
+    switch (l)
+    {
+    case FlushLevel::UserFlush:
+        os << "UserFlush";
+        break;
+    case FlushLevel::InternalFlush:
+        os << "InternalFlush";
+        break;
+    case FlushLevel::SkeletonOnly:
+        os << "SkeletonOnly";
+        break;
+    case FlushLevel::CreateOrOpenFiles:
+        os << "CreateOrOpenFiles";
+        break;
+    }
+    return os;
+}
+} // namespace openPMD
+
 namespace openPMD::auxiliary
 {
 using pair_t = std::pair<OpenpmdStandard, char const *>;

src/IO/AbstractIOHandlerImpl.cpp

@@ -21,6 +21,8 @@
 
 #include "openPMD/IO/AbstractIOHandlerImpl.hpp"
 
+#include "openPMD/Error.hpp"
+#include "openPMD/IO/AbstractIOHandler.hpp"
 #include "openPMD/IO/IOTask.hpp"
 #include "openPMD/Streaming.hpp"
 #include "openPMD/auxiliary/Environment.hpp"
@@ -87,13 +89,74 @@ void AbstractIOHandlerImpl::writeToStderr([[maybe_unused]] Args &&...args) const
     }
 }
 
-std::future<void> AbstractIOHandlerImpl::flush()
+namespace
+{
+    void verifyFlushType(Operation op, FlushLevel l)
+    {
+        auto do_throw = [&](char const *least_flush_level) {
+            std::stringstream err;
+            err << "Operation " << op << " is not allowed below flush level "
+                << least_flush_level << ", but flush level was " << l << ".";
+            throw error::Internal(err.str());
+        };
+        switch (op)
+        {
+        case Operation::ADVANCE:
+        case Operation::CREATE_FILE:
+        case Operation::CHECK_FILE:
+        case Operation::OPEN_FILE:
+        case Operation::CLOSE_FILE:
+        case Operation::DELETE_FILE:
+        case Operation::DEREGISTER:
+        case Operation::TOUCH:
+        case Operation::LIST_ATTS:
+        case Operation::LIST_PATHS:
+        case Operation::OPEN_PATH:
+        case Operation::SET_WRITTEN:
+        case Operation::CREATE_PATH:
+            break;
+        case Operation::CLOSE_PATH:
+        case Operation::DELETE_PATH:
+        case Operation::CREATE_DATASET:
+        case Operation::EXTEND_DATASET:
+        case Operation::OPEN_DATASET:
+        case Operation::DELETE_DATASET:
+        case Operation::LIST_DATASETS:
+            if (!flush_level::flush_hierarchy(l))
+            {
+                do_throw("SkeletonOnly (hierarchy operations)");
+            }
+            break;
+        case Operation::GET_BUFFER_VIEW:
+        case Operation::DELETE_ATT:
+        case Operation::WRITE_ATT:
+        case Operation::READ_ATT:
+        case Operation::READ_ATT_ALLSTEPS:
+        case Operation::AVAILABLE_CHUNKS:
+            if (!flush_level::write_attributes(l))
+            {
+                do_throw("InternalFlush (metadata operations)");
+            }
+            break;
+        case Operation::WRITE_DATASET:
+        case Operation::READ_DATASET:
+            if (!flush_level::write_datasets(l))
+            {
+                do_throw("UserFlush (flushpoint operations)");
+            }
+            break;
+        }
+    }
+} // namespace
+
+std::future<void> AbstractIOHandlerImpl::flush(FlushLevel l)
 {
     using namespace auxiliary;
 
     while (!(*m_handler).m_work.empty())
     {
         IOTask &i = (*m_handler).m_work.front();
+        // verifyFlushType(i.operation, l);
         try
         {
             switch (i.operation)
@@ -467,6 +530,7 @@ std::future<void> AbstractIOHandlerImpl::flush()
                 break;
             }
             }
+            verifyFlushType(i.operation, l);
         }
         catch (...)
         {

src/IO/HDF5/HDF5IOHandler.cpp

@@ -3379,7 +3379,7 @@ auto HDF5IOHandlerImpl::requireFile(
 
 std::future<void> HDF5IOHandlerImpl::flush(internal::ParsedFlushParams &params)
 {
-    auto res = AbstractIOHandlerImpl::flush();
+    auto res = AbstractIOHandlerImpl::flush(params.flushLevel);
 
     if (params.backendConfig.json().contains("hdf5"))
     {