push_chunk: distinguish immediate from deferred flushes

Author: franzpoeschel

include/openPMD/IO/AbstractIOHandler.hpp

@@ -61,7 +61,7 @@ enum class FlushLevel
     UserFlush,
     /**
      * Flush triggered by storeChunk in immediate flush mode.
-     * Must not perform storeChunk operations of the Span API.
+     * Must not perform operations enqueued in m_chunks.
      */
     ImmediateFlush,
     /**

include/openPMD/RecordComponent.hpp

@@ -74,7 +74,8 @@ namespace internal
          */
         std::queue<IOTask> m_chunks;
 
-        [[nodiscard]] auto push_chunk(IOTask &&task) -> bool;
+        void push_chunk(
+            IOTask &&task, std::optional<bool> immediate_flush = std::nullopt);
         /**
          * Stores the value for constant record components.
          * Ignored otherwise.

include/openPMD/backend/PatchRecordComponent.hpp

@@ -157,10 +157,7 @@ inline void PatchRecordComponent::load(std::shared_ptr<T> data)
     dRead.dtype = getDatatype();
     dRead.data = std::static_pointer_cast<void>(data);
     auto &rc = get();
-    if (rc.push_chunk(IOTask(this, dRead)))
-    {
-        seriesFlush();
-    }
+    rc.push_chunk(IOTask(this, dRead));
 }
 
 template <typename T>
@@ -199,10 +196,7 @@ inline void PatchRecordComponent::store(uint64_t idx, T data)
     dWrite.dtype = dtype;
     dWrite.data = std::make_shared<T>(data);
     auto &rc = get();
-    if (rc.push_chunk(IOTask(this, std::move(dWrite))))
-    {
-        seriesFlush();
-    }
+    rc.push_chunk(IOTask(this, std::move(dWrite)));
 }
 
 template <typename T>
@@ -231,9 +225,6 @@ inline void PatchRecordComponent::store(T data)
     dWrite.dtype = dtype;
     dWrite.data = std::make_shared<T>(data);
     auto &rc = get();
-    if (rc.push_chunk(IOTask(this, std::move(dWrite))))
-    {
-        seriesFlush();
-    }
+    rc.push_chunk(IOTask(this, std::move(dWrite)));
 }
 } // namespace openPMD

src/RecordComponent.cpp

@@ -22,6 +22,7 @@
 #include "openPMD/Dataset.hpp"
 #include "openPMD/DatatypeHelpers.hpp"
 #include "openPMD/Error.hpp"
+#include "openPMD/IO/AbstractIOHandler.hpp"
 #include "openPMD/IO/Format.hpp"
 #include "openPMD/LoadStoreChunk.hpp"
 #include "openPMD/Series.hpp"
@@ -52,7 +53,8 @@ namespace openPMD
 namespace internal
 {
     RecordComponentData::RecordComponentData() = default;
-    auto RecordComponentData::push_chunk(IOTask &&task) -> bool
+    void RecordComponentData::push_chunk(
+        IOTask &&task, std::optional<bool> immediate_flush)
     {
         Attributable a;
         a.setData(std::shared_ptr<AttributableData>{this, [](auto const &) {}});
@@ -74,9 +76,29 @@ namespace internal
                 "Cannot write/read chunks to/from closed Iterations.");
         }
 #endif
+        bool immediate_flush_resolved = [&]() {
+            if (immediate_flush.has_value())
+            {
+                return *immediate_flush;
+            }
+            else
+            {
+                return a.IOHandler()->m_flush_immediately;
+            }
+        }();
         a.setDirtyRecursive(true);
-        m_chunks.push(std::move(task));
-        return a.IOHandler()->m_flush_immediately;
+        if (immediate_flush_resolved)
+        {
+            a.seriesFlush_impl<false>(
+                internal::FlushParams{FlushLevel::ImmediateFlush});
+            auto IOHandler = a.IOHandler();
+            IOHandler->enqueue(task);
+            IOHandler->flush(FlushLevel::UserFlush);
+        }
+        else
+        {
+            m_chunks.push(std::move(task));
+        }
     }
 
     static constexpr char const *note_on_deactivating_this_check = R"(
@@ -465,7 +487,8 @@ void RecordComponent::flush(
     {
         return;
     }
-    if (access::readOnly(IOHandler()->m_frontendAccess))
+    if (access::readOnly(IOHandler()->m_frontendAccess) &&
+        flush_level::global_flushpoint(flushParams.flushLevel))
     {
         while (!rc.m_chunks.empty())
         {
@@ -692,10 +715,7 @@ void RecordComponent::storeChunk_impl(
     /* std::static_pointer_cast correctly reference-counts the pointer */
     dWrite.data = std::move(buffer);
     auto &rc = get();
-    if (rc.push_chunk(IOTask(this, std::move(dWrite))))
-    {
-        seriesFlush();
-    }
+    rc.push_chunk(IOTask(this, std::move(dWrite)));
 }
 
 void RecordComponent::verifyChunk(
@@ -956,10 +976,7 @@ void RecordComponent::loadChunk_impl(
         dRead.extent = extent;
         dRead.dtype = getDatatype();
         dRead.data = std::static_pointer_cast<void>(data);
-        if (rc.push_chunk(IOTask(this, dRead)))
-        {
-            seriesFlush();
-        }
+        rc.push_chunk(IOTask(this, dRead));
     }
 }