Asynchronous IO in BP5: Guided and Naive implementations. Guided uses…

bindings/C/adios2/c/adios2_c_adios.cpp

@@ -233,6 +233,44 @@ adios2_error adios2_finalize(adios2_adios *adios)
     }
 }
 
+/** Inform ADIOS about entering communication-free computation block
+ * in main thread. Useful when using Async IO */
+adios2_error adios2_enter_computation_block(adios2_adios *adios)
+{
+    try
+    {
+        adios2::helper::CheckForNullptr(
+            adios,
+            "for adios2_adios, in call to adios2_enter_computation_block");
+        reinterpret_cast<adios2::core::ADIOS *>(adios)->EnterComputationBlock();
+        return adios2_error_none;
+    }
+    catch (...)
+    {
+        return static_cast<adios2_error>(
+            adios2::helper::ExceptionToError("adios2_enter_computation_block"));
+    }
+}
+
+/** Inform ADIOS about exiting communication-free computation block
+ * in main thread. Useful when using Async IO */
+adios2_error adios2_exit_computation_block(adios2_adios *adios)
+{
+    try
+    {
+        adios2::helper::CheckForNullptr(
+            adios,
+            "for adios2_adios, in call to adios2_exit_computation_block");
+        reinterpret_cast<adios2::core::ADIOS *>(adios)->ExitComputationBlock();
+        return adios2_error_none;
+    }
+    catch (...)
+    {
+        return static_cast<adios2_error>(
+            adios2::helper::ExceptionToError("adios2_exit_computation_block"));
+    }
+}
+
 #ifdef __cplusplus
 }
 // end extern C

bindings/C/adios2/c/adios2_c_adios.h

@@ -147,6 +147,14 @@ adios2_error adios2_remove_io(adios2_bool *result, adios2_adios *adios,
  */
 adios2_error adios2_remove_all_ios(adios2_adios *adios);
 
+/** Inform ADIOS about entering communication-free computation block
+ * in main thread. Useful when using Async IO */
+adios2_error adios2_enter_computation_block(adios2_adios *adios);
+
+/** Inform ADIOS about exiting communication-free computation block
+ * in main thread. Useful when using Async IO */
+adios2_error adios2_exit_computation_block(adios2_adios *adios);
+
 #ifdef __cplusplus
 } // end extern C
 #endif

bindings/CXX11/adios2/cxx11/ADIOS.cpp

@@ -49,6 +49,18 @@ void ADIOS::FlushAll()
     m_ADIOS->FlushAll();
 }
 
+void ADIOS::EnterComputationBlock() noexcept
+{
+    CheckPointer("in call to ADIOS::EnterComputationBlock()");
+    m_ADIOS->EnterComputationBlock();
+}
+
+void ADIOS::ExitComputationBlock() noexcept
+{
+    CheckPointer("in call to ADIOS::ExitComputationBlock()");
+    m_ADIOS->ExitComputationBlock();
+}
+
 Operator ADIOS::DefineOperator(const std::string name, const std::string type,
                                const Params &parameters)
 {

bindings/CXX11/adios2/cxx11/ADIOS.h

@@ -235,6 +235,14 @@ class ADIOS
      */
     void RemoveAllIOs() noexcept;
 
+    /** Inform ADIOS about entering communication-free computation block
+     * in main thread. Useful when using Async IO */
+    void EnterComputationBlock() noexcept;
+
+    /** Inform ADIOS about exiting communication-free computation block
+     * in main thread. Useful when using Async IO */
+    void ExitComputationBlock() noexcept;
+
 protected:
     std::shared_ptr<core::ADIOS> m_ADIOS;
 

bindings/Fortran/f2c/adios2_f2c_adios.cpp

@@ -109,6 +109,20 @@ void FC_GLOBAL(adios2_finalize_f2c, ADIOS2_FINALIZE_F2C)(adios2_adios **adios,
     *ierr = static_cast<int>(adios2_finalize(*adios));
 }
 
+void FC_GLOBAL(adios2_enter_computation_block_f2c,
+               ADIOS2_ENTER_COMPUTATION_BLOCK_F2C)(adios2_adios **adios,
+                                                   int *ierr)
+{
+    *ierr = static_cast<int>(adios2_enter_computation_block(*adios));
+}
+
+void FC_GLOBAL(adios2_exit_computation_block_f2c,
+               ADIOS2_EXIT_COMPUTATION_BLOCK_F2C)(adios2_adios **adios,
+                                                  int *ierr)
+{
+    *ierr = static_cast<int>(adios2_exit_computation_block(*adios));
+}
+
 #ifdef __cplusplus
 }
 #endif

bindings/Fortran/modules/adios2_adios_mod.f90

@@ -24,6 +24,8 @@ module adios2_adios_mod
     external adios2_remove_io_f2c
     external adios2_remove_all_ios_f2c
     external adios2_finalize_f2c
+    external adios2_enter_computation_block_f2c
+    external adios2_exit_computation_block_f2c
 contains
 
     subroutine adios2_declare_io(io, adios, io_name, ierr)
@@ -154,4 +156,18 @@ subroutine adios2_finalize(adios, ierr)
 
     end subroutine
 
+
+    subroutine adios2_enter_computation_block(adios, ierr)
+        type(adios2_adios), intent(in) :: adios
+        integer, intent(out) :: ierr
+        call adios2_enter_computation_block_f2c(adios, ierr)
+    end subroutine
+
+    subroutine adios2_exit_computation_block(adios, ierr)
+        type(adios2_adios), intent(in) :: adios
+        integer, intent(out) :: ierr
+        call adios2_exit_computation_block_f2c(adios, ierr)
+    end subroutine
+
+
 end module

examples/basics/globalArray/globalArray_write.cpp

@@ -28,7 +28,9 @@
  *      Author: pnorbert
  */
 
+#include <chrono>
 #include <iostream>
+#include <thread>
 #include <vector>
 
 #include <adios2.h>
@@ -64,7 +66,10 @@ int main(int argc, char *argv[])
         // create one with default settings here
         adios2::IO io = adios.DeclareIO("Output");
         io.SetEngine("BP5");
+        io.SetParameter("AggregationType", "TwoLevelShm");
         io.SetParameter("NumAggregators", "1");
+        io.SetParameter("NumSubFiles", "1");
+        io.SetParameter("AsyncWrite", "Guided");
 
         /*
          * Define global array: type, name, global dimensions
@@ -98,6 +103,9 @@ int main(int argc, char *argv[])
             // Disk I/O will be performed during this call unless
             // time aggregation postpones all of that to some later step
             writer.EndStep();
+            adios.EnterComputationBlock();
+            std::this_thread::sleep_for(std::chrono::duration<double>(1.0));
+            adios.ExitComputationBlock();
         }
 
         // Called once: indicate that we are done with this output for the run

source/adios2/CMakeLists.txt

@@ -91,6 +91,10 @@ add_library(adios2_core
 
   toolkit/transportman/TransportMan.cpp
 
+  toolkit/shm/Spinlock.cpp
+  toolkit/shm/SerializeProcesses.cpp
+  toolkit/shm/TokenChain.h
+
   toolkit/aggregator/mpi/MPIAggregator.cpp
   toolkit/aggregator/mpi/MPIChain.cpp
   toolkit/aggregator/mpi/MPIShmChain.cpp
@@ -125,7 +129,7 @@ if (ADIOS2_HAVE_BP5)
   target_sources(adios2_core PRIVATE
     engine/bp5/BP5Engine.cpp
     engine/bp5/BP5Reader.cpp engine/bp5/BP5Reader.tcc
-    engine/bp5/BP5Writer.cpp engine/bp5/BP5Writer.tcc engine/bp5/BP5Writer_TwoLevelShm.cpp
+    engine/bp5/BP5Writer.cpp engine/bp5/BP5Writer.tcc engine/bp5/BP5Writer_TwoLevelShm.cpp engine/bp5/BP5Writer_TwoLevelShm_Async.cpp engine/bp5/BP5Writer_EveryoneWrites_Async.cpp
   )
 endif()
 

source/adios2/common/ADIOSTypes.h

@@ -214,8 +214,8 @@ constexpr size_t DefaultMinDeferredSize = 4 * 1024 * 1024;
 constexpr size_t DefaultMaxFileBatchSize = 2147381248;
 
 /** default maximum shared memory segment size
- *  128Mb */
-constexpr uint64_t DefaultMaxShmSize = 128 * 1024 * 1024;
+ *  2 blocks of MaxFileBatchSize */
+constexpr uint64_t DefaultMaxShmSize = 2 * DefaultMaxFileBatchSize;
 
 constexpr char PathSeparator =
 #ifdef _WIN32

source/adios2/core/ADIOS.cpp

@@ -151,6 +151,27 @@ void ADIOS::FlushAll()
     }
 }
 
+void ADIOS::EnterComputationBlock() noexcept
+{
+    enteredComputationBlock = true;
+    for (auto &ioPair : m_IOs)
+    {
+        ioPair.second.EnterComputationBlock();
+    }
+}
+
+void ADIOS::ExitComputationBlock() noexcept
+{
+    if (enteredComputationBlock)
+    {
+        enteredComputationBlock = false;
+        for (auto &ioPair : m_IOs)
+        {
+            ioPair.second.ExitComputationBlock();
+        }
+    }
+}
+
 std::pair<std::string, Params> &ADIOS::DefineOperator(const std::string &name,
                                                       const std::string type,
                                                       const Params &parameters)

source/adios2/core/ADIOS.h

@@ -20,6 +20,7 @@
 
 #include "adios2/common/ADIOSConfig.h"
 #include "adios2/common/ADIOSTypes.h"
+#include "adios2/core/CoreTypes.h"
 #include "adios2/core/Operator.h"
 #include "adios2/helper/adiosComm.h"
 
@@ -145,6 +146,14 @@ class ADIOS
      */
     void RemoveAllIOs() noexcept;
 
+    /** Inform ADIOS about entering communication-free computation block
+     * in main thread. Useful when using Async IO */
+    void EnterComputationBlock() noexcept;
+
+    /** Inform ADIOS about exiting communication-free computation block
+     * in main thread. Useful when using Async IO */
+    void ExitComputationBlock() noexcept;
+
 private:
     /** Communicator given to parallel constructor. */
     helper::Comm m_Comm;
@@ -166,6 +175,9 @@ class ADIOS
     /** operators created with DefineOperator */
     std::unordered_map<std::string, std::pair<std::string, Params>> m_Operators;
 
+    /** Flag for Enter/ExitComputationBlock */
+    bool enteredComputationBlock = false;
+
     void CheckOperator(const std::string name) const;
 
     void XMLInit(const std::string &configFileXML);

source/adios2/core/Engine.cpp

@@ -95,6 +95,9 @@ size_t Engine::DebugGetDataBufferSize() const
     return 0;
 }
 
+void Engine::EnterComputationBlock() noexcept {}
+void Engine::ExitComputationBlock() noexcept {}
+
 // PROTECTED
 void Engine::Init() {}
 void Engine::InitParameters() {}

source/adios2/core/Engine.h

@@ -511,6 +511,11 @@ class Engine
     virtual void NotifyEngineAttribute(std::string name,
                                        DataType type) noexcept;
 
+    /** Inform about computation block through User->ADIOS->IO */
+    virtual void EnterComputationBlock() noexcept;
+    /** Inform about computation block through User->ADIOS->IO */
+    virtual void ExitComputationBlock() noexcept;
+
 protected:
     /** from ADIOS class passed to Engine created with Open
      *  if no communicator is passed */

source/adios2/core/IO.cpp

@@ -738,6 +738,30 @@ void IO::RemoveEngine(const std::string &name)
     }
 }
 
+void IO::EnterComputationBlock() noexcept
+{
+    for (auto &enginePair : m_Engines)
+    {
+        auto &engine = enginePair.second;
+        if (engine->OpenMode() != Mode::Read)
+        {
+            enginePair.second->EnterComputationBlock();
+        }
+    }
+}
+
+void IO::ExitComputationBlock() noexcept
+{
+    for (auto &enginePair : m_Engines)
+    {
+        auto &engine = enginePair.second;
+        if (engine->OpenMode() != Mode::Read)
+        {
+            enginePair.second->ExitComputationBlock();
+        }
+    }
+}
+
 void IO::FlushAll()
 {
     PERFSTUBS_SCOPED_TIMER("IO::FlushAll");

source/adios2/core/IO.h

@@ -25,6 +25,7 @@
 #include "adios2/common/ADIOSTypes.h"
 #include "adios2/core/ADIOS.h"
 #include "adios2/core/Attribute.h"
+#include "adios2/core/CoreTypes.h"
 #include "adios2/core/Group.h"
 #include "adios2/core/Variable.h"
 #include "adios2/core/VariableCompound.h"
@@ -487,6 +488,11 @@ class IO
         return m_Engines;
     }
 
+    /** Inform about computation block through User->ADIOS */
+    void EnterComputationBlock() noexcept;
+    /** Inform about computation block through User->ADIOS */
+    void ExitComputationBlock() noexcept;
+
 private:
     /** true: exist in config file (XML) */
     const bool m_InConfigFile = false;

source/adios2/engine/bp3/BP3Writer.cpp

@@ -196,11 +196,11 @@ void BP3Writer::InitTransports()
 
     if (m_BP3Serializer.m_Aggregator.m_IsAggregator)
     {
-        if (m_BP3Serializer.m_Parameters.AsyncTasks)
+        if (m_BP3Serializer.m_Parameters.AsyncOpen)
         {
             for (size_t i = 0; i < m_IO.m_TransportsParameters.size(); ++i)
             {
-                m_IO.m_TransportsParameters[i]["asynctasks"] = "true";
+                m_IO.m_TransportsParameters[i]["asyncopen"] = "true";
             }
         }
         m_FileDataManager.OpenFiles(bpSubStreamNames, m_OpenMode,

source/adios2/engine/bp4/BP4Writer.cpp

@@ -248,11 +248,11 @@ void BP4Writer::InitTransports()
 
     if (m_BP4Serializer.m_Aggregator.m_IsAggregator)
     {
-        if (m_BP4Serializer.m_Parameters.AsyncTasks)
+        if (m_BP4Serializer.m_Parameters.AsyncOpen)
         {
             for (size_t i = 0; i < m_IO.m_TransportsParameters.size(); ++i)
             {
-                m_IO.m_TransportsParameters[i]["asynctasks"] = "true";
+                m_IO.m_TransportsParameters[i]["asyncopen"] = "true";
             }
         }
         for (size_t i = 0; i < m_IO.m_TransportsParameters.size(); ++i)