Modify example to support/enforce MPI streaming backend

Will fail to build in non-MPI-enabled openPMD Will fail to execute in ADIOS2 released versions, only works with master branch Will fail to execute on some systems Add block for Series in writer Same thing in reader Run Streaming test under MPI
franzpoeschel · Aug 10, 2023 · 053d429 · 053d429
1 parent 15feda4
commit 053d429
Show file tree

Hide file tree

Showing 3 changed files with 66 additions and 8 deletions.
diff --git a/CMakeLists.txt b/CMakeLists.txt
@@ -1281,9 +1281,17 @@ if(openPMD_BUILD_TESTING)
             endif()
         endif()
         if(openPMD_HAVE_ADIOS2)
-            add_test(NAME Asynchronous.10_streaming
-                     COMMAND sh -c "$<TARGET_FILE:10_streaming_write> & sleep 1; $<TARGET_FILE:10_streaming_read>"
-                     WORKING_DIRECTORY ${openPMD_RUNTIME_OUTPUT_DIRECTORY})
+            if(openPMD_HAVE_MPI)
+                add_test(NAME Asynchronous.10_streaming
+                        COMMAND sh -c "${MPIEXEC_EXECUTABLE} ${MPIEXEC_NUMPROC_FLAG} 1 $<TARGET_FILE:10_streaming_write> & \
+                            sleep 1; ${MPIEXEC_EXECUTABLE} ${MPIEXEC_NUMPROC_FLAG} 1 $<TARGET_FILE:10_streaming_read>"
+                        WORKING_DIRECTORY ${openPMD_RUNTIME_OUTPUT_DIRECTORY})
+            else()
+                add_test(NAME Asynchronous.10_streaming
+                        COMMAND sh -c "$<TARGET_FILE:10_streaming_write> & \
+                            sleep 1; $<TARGET_FILE:10_streaming_read>"
+                        WORKING_DIRECTORY ${openPMD_RUNTIME_OUTPUT_DIRECTORY})
+            endif()
         endif()
     endif()
 

diff --git a/examples/10_streaming_read.cpp b/examples/10_streaming_read.cpp
@@ -5,6 +5,8 @@
 #include <iostream>
 #include <memory>
 
+#include <mpi.h>
+
 using std::cout;
 using namespace openPMD;
 
@@ -19,7 +21,30 @@ int main()
         return 0;
     }
 
-    Series series = Series("electrons.sst", Access::READ_LINEAR);
+    {
+        int provided{};
+        MPI_Init_thread(nullptr, nullptr, MPI_THREAD_MULTIPLE, &provided);
+        switch (provided)
+        {
+        case MPI_THREAD_SINGLE:
+            throw std::runtime_error("MPI_THREAD_SINGLE");
+        case MPI_THREAD_FUNNELED:
+            throw std::runtime_error("MPI_THREAD_FUNNELED");
+        case MPI_THREAD_SERIALIZED:
+            throw std::runtime_error("MPI_THREAD_SERIALIZED");
+        case MPI_THREAD_MULTIPLE:
+            std::cout << "MPI_THREAD_MULTIPLE" << std::endl;
+            break;
+        default:
+            throw std::runtime_error("???????????????");
+        }
+    }
+
+    Series series = Series(
+        "electrons.sst",
+        Access::READ_ONLY,
+        MPI_COMM_WORLD,
+        R"(adios2.engine.parameters.DataTransport = "MPI")");
 
     // `Series::writeIterations()` and `Series::readIterations()` are
     // intentionally restricted APIs that ensure a workflow which also works
@@ -44,9 +69,6 @@ int main()
             extents[i] = rc.getExtent();
         }
 
-        // The iteration can be closed in order to help free up resources.
-        // The iteration's content will be flushed automatically.
-        // An iteration once closed cannot (yet) be reopened.
         iteration.close();
 
         for (size_t i = 0; i < 3; ++i)
@@ -71,6 +93,8 @@ int main()
      */
     series.close();
 
+    MPI_Finalize();
+
     return 0;
 #else
     std::cout << "The streaming example requires that openPMD has been built "

diff --git a/examples/10_streaming_write.cpp b/examples/10_streaming_write.cpp
@@ -5,6 +5,8 @@
 #include <memory>
 #include <numeric> // std::iota
 
+#include <mpi.h>
+
 using std::cout;
 using namespace openPMD;
 
@@ -19,8 +21,31 @@ int main()
         return 0;
     }
 
+    {
+        int provided{};
+        MPI_Init_thread(nullptr, nullptr, MPI_THREAD_MULTIPLE, &provided);
+        switch (provided)
+        {
+        case MPI_THREAD_SINGLE:
+            throw std::runtime_error("MPI_THREAD_SINGLE");
+        case MPI_THREAD_FUNNELED:
+            throw std::runtime_error("MPI_THREAD_FUNNELED");
+        case MPI_THREAD_SERIALIZED:
+            throw std::runtime_error("MPI_THREAD_SERIALIZED");
+        case MPI_THREAD_MULTIPLE:
+            std::cout << "MPI_THREAD_MULTIPLE" << std::endl;
+            break;
+        default:
+            throw std::runtime_error("???????????????");
+        }
+    }
+
     // open file for writing
-    Series series = Series("electrons.sst", Access::CREATE);
+    Series series = Series(
+        "electrons.sst",
+        Access::CREATE,
+        MPI_COMM_WORLD,
+        R"(adios2.engine.parameters.DataTransport = "MPI")");
 
     Datatype datatype = determineDatatype<position_t>();
     constexpr unsigned long length = 10ul;
@@ -57,6 +82,7 @@ int main()
      * calling the destructor, including the release of file handles.
      */
     series.close();
+    MPI_Finalize();
 
     return 0;
 #else