wip

include/openPMD/IO/IOTask.hpp

@@ -48,20 +48,36 @@ Writable *getWritable(Attributable *);
 /** Type of IO operation between logical and persistent data.
  */
 OPENPMDAPI_EXPORT_ENUM_CLASS(Operation){
-    CREATE_FILE,      CHECK_FILE,     OPEN_FILE,     CLOSE_FILE,
+    CREATE_FILE,
+    CHECK_FILE,
+    OPEN_FILE,
+    CLOSE_FILE,
     DELETE_FILE,
 
-    CREATE_PATH,      CLOSE_PATH,     OPEN_PATH,     DELETE_PATH,
+    CREATE_PATH,
+    CLOSE_PATH,
+    OPEN_PATH,
+    DELETE_PATH,
     LIST_PATHS,
 
-    CREATE_DATASET,   EXTEND_DATASET, OPEN_DATASET,  DELETE_DATASET,
-    WRITE_DATASET,    READ_DATASET,   LIST_DATASETS, GET_BUFFER_VIEW,
+    CREATE_DATASET,
+    EXTEND_DATASET,
+    OPEN_DATASET,
+    DELETE_DATASET,
+    WRITE_DATASET,
+    READ_DATASET,
+    LIST_DATASETS,
+    GET_BUFFER_VIEW,
 
-    DELETE_ATT,       WRITE_ATT,      READ_ATT,      LIST_ATTS,
+    DELETE_ATT,
+    WRITE_ATT,
+    READ_ATT,
+    LIST_ATTS,
 
     ADVANCE,
     AVAILABLE_CHUNKS, //!< Query chunks that can be loaded in a dataset
-    DEREGISTER //!< Inform the backend that an object has been deleted.
+    DEREGISTER, //!< Inform the backend that an object has been deleted.
+    TOUCH //!< tell the backend that the file is to be considered active
 }; // note: if you change the enum members here, please update
    // docs/source/dev/design.rst
 

src/IO/ADIOS/ADIOS2File.cpp

@@ -23,6 +23,7 @@
 #include "openPMD/Error.hpp"
 #include "openPMD/IO/ADIOS/ADIOS2IOHandler.hpp"
 #include "openPMD/auxiliary/Environment.hpp"
+#include <mpi.h>
 
 #if openPMD_USE_VERIFY
 #define VERIFY(CONDITION, TEXT)                                                \
@@ -1045,6 +1046,11 @@ void ADIOS2File::flush_impl(ADIOS2FlushParams flushParams, bool writeLatePuts)
                 break;
             }
             performDataWrite = performDataWrite && m_engineType == "bp5";
+            int rank;
+            MPI_Comm_rank(MPI_COMM_WORLD, &rank);
+            std::cout << "[" << rank
+                      << "] Performing data write: " << performDataWrite
+                      << std::endl;
 
             if (performDataWrite)
             {

src/IO/ADIOS/ADIOS2IOHandler.cpp

@@ -485,6 +485,9 @@ overrideFlushTarget(FlushTarget &inplace, FlushTarget new_val)
 std::future<void>
 ADIOS2IOHandlerImpl::flush(internal::ParsedFlushParams &flushParams)
 {
+    int rank;
+    MPI_Comm_rank(MPI_COMM_WORLD, &rank);
+    std::cout << "[" << rank << "] Flushing ADIOS2" << std::endl;
     auto res = AbstractIOHandlerImpl::flush();
 
     detail::ADIOS2File::ADIOS2FlushParams adios2FlushParams{
@@ -536,10 +539,14 @@ ADIOS2IOHandlerImpl::flush(internal::ParsedFlushParams &flushParams)
     {
         if (m_dirty.find(p.first) != m_dirty.end())
         {
+            std::cout << "[" << rank << "] Flush " << p.second->m_file
+                      << std::endl;
             p.second->flush(adios2FlushParams, /* writeLatePuts = */ false);
         }
         else
         {
+            std::cout << "[" << rank << "] Drop " << p.second->m_file
+                      << std::endl;
             p.second->drop();
         }
     }
@@ -895,6 +902,7 @@ void ADIOS2IOHandlerImpl::openFile(
     // lazy opening is deathly in parallel situations
     auto &fileData = getFileData(file, IfFileNotOpen::OpenImplicitly);
     *parameters.out_parsePreference = fileData.parsePreference;
+    m_dirty.emplace(std::move(file));
 }
 
 void ADIOS2IOHandlerImpl::closeFile(

test/ParallelIOTest.cpp

@@ -1049,6 +1049,41 @@ TEST_CASE("hipace_like_write", "[parallel]")
 }
 #endif
 
+#if openPMD_HAVE_ADIOS2 && openPMD_HAS_ADIOS_2_9
+TEST_CASE("independent_write_with_collective_flush", "[parallel]")
+{
+    Series write(
+        "../samples/independent_write_with_collective_flush.bp5",
+        Access::CREATE,
+        MPI_COMM_WORLD,
+        "adios2.engine.preferred_flush_target = \"buffer\"");
+    int size, rank;
+    MPI_Comm_rank(MPI_COMM_WORLD, &rank);
+    MPI_Comm_size(MPI_COMM_WORLD, &size);
+    auto iteration = write.iterations[0];
+    auto E_x = iteration.meshes["E"]["x"];
+    E_x.resetDataset({Datatype::DOUBLE, {10}});
+    write.flush();
+    if (rank == 1)
+    {
+        E_x.storeChunk(
+            std::unique_ptr<double[]>{new double[10]{4.2}}, {0}, {10});
+    }
+    /*
+     * Now, the iteration is dirty only on rank 1. But the following flush must
+     * run collectively anyway. The test has been designed in such a way that
+     * the PerformDataWrite() call required by the disk flush target will
+     * conflict with the default buffer target that will run in the destructor,
+     * unless the flush in the next line really is collective.
+     */
+    std::cout << "ENTER" << std::endl;
+    MPI_Barrier(MPI_COMM_WORLD);
+    iteration.seriesFlush("adios2.engine.preferred_flush_target = \"disk\"");
+    MPI_Barrier(MPI_COMM_WORLD);
+    std::cout << "LEAVE" << std::endl;
+}
+#endif
+
 #if openPMD_HAVE_ADIOS2 && openPMD_HAVE_MPI
 
 void adios2_streaming(bool variableBasedLayout)