First attempt at immediate flushing

Author: franzpoeschel

include/openPMD/IO/AbstractIOHandler.hpp

@@ -230,6 +230,8 @@ namespace internal
         OpenpmdStandard m_standard =
             auxiliary::parseStandard(getStandardDefault());
         bool m_verify_homogeneous_extents = true;
+        // If true, then flush directly upon storeChunk
+        bool m_flush_immediately = false;
 
     protected:
         explicit GlobalParameters();

include/openPMD/RecordComponent.hpp

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

include/openPMD/backend/PatchRecordComponent.hpp

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

src/RecordComponent.cpp

@@ -52,7 +52,7 @@ namespace openPMD
 namespace internal
 {
     RecordComponentData::RecordComponentData() = default;
-    auto RecordComponentData::push_chunk(IOTask &&task) -> void
+    auto RecordComponentData::push_chunk(IOTask &&task) -> bool
     {
         Attributable a;
         a.setData(std::shared_ptr<AttributableData>{this, [](auto const &) {}});
@@ -76,6 +76,7 @@ namespace internal
 #endif
         a.setDirtyRecursive(true);
         m_chunks.push(std::move(task));
+        return a.IOHandler()->m_flush_immediately;
     }
 
     static constexpr char const *note_on_deactivating_this_check = R"(
@@ -691,7 +692,10 @@ void RecordComponent::storeChunk_impl(
     /* std::static_pointer_cast correctly reference-counts the pointer */
     dWrite.data = std::move(buffer);
     auto &rc = get();
-    rc.push_chunk(IOTask(this, std::move(dWrite)));
+    if (rc.push_chunk(IOTask(this, std::move(dWrite))))
+    {
+        seriesFlush();
+    }
 }
 
 void RecordComponent::verifyChunk(
@@ -952,7 +956,10 @@ void RecordComponent::loadChunk_impl(
         dRead.extent = extent;
         dRead.dtype = getDatatype();
         dRead.data = std::static_pointer_cast<void>(data);
-        rc.push_chunk(IOTask(this, dRead));
+        if (rc.push_chunk(IOTask(this, dRead)))
+        {
+            seriesFlush();
+        }
     }
 }
 

src/Series.cpp

@@ -3174,6 +3174,11 @@ void Series::parseJsonOptions(
         "verify_homogeneous_extents",
         gp.m_verify_homogeneous_extents,
         "OPENPMD_VERIFY_HOMOGENEOUS_EXTENTS");
+    getJsonOption<bool>(
+        options,
+        "flush_immediately",
+        gp.m_flush_immediately,
+        "OPENPMD_FLUSH_IMMEDIATELY");
     internal::SeriesData::SourceSpecifiedViaJSON rankTableSource;
     if (getJsonOptionLowerCase(options, "rank_table", rankTableSource.value))
     {