added threading in ssc for hiding metadata manipulation

ornladios · Jan 14, 2021 · f5202b0 · f5202b0
1 parent a32272a
commit f5202b0
Show file tree

Hide file tree

Showing 2 changed files with 47 additions and 3 deletions.
diff --git a/source/adios2/engine/ssc/SscWriter.cpp b/source/adios2/engine/ssc/SscWriter.cpp
@@ -28,9 +28,26 @@ SscWriter::SscWriter(IO &io, const std::string &name, const Mode mode,
 
     helper::GetParameter(m_IO.m_Parameters, "MpiMode", m_MpiMode);
     helper::GetParameter(m_IO.m_Parameters, "Verbose", m_Verbosity);
+    helper::GetParameter(m_IO.m_Parameters, "Threading", m_Threading);
     helper::GetParameter(m_IO.m_Parameters, "OpenTimeoutSecs",
                          m_OpenTimeoutSecs);
 
+    int providedMpiMode;
+    MPI_Query_thread(&providedMpiMode);
+    if (providedMpiMode != MPI_THREAD_MULTIPLE)
+    {
+        if (m_Threading == true)
+        {
+            m_Threading = false;
+            if (m_WriterRank == 0)
+            {
+                std::cout << "SSC Threading disabled as MPI is not initialized "
+                             "with multi-threads"
+                          << std::endl;
+            }
+        }
+    }
+
     SyncMpiPattern();
     m_WriterRank = m_Comm.Rank();
     m_WriterSize = m_Comm.Size();
@@ -42,6 +59,11 @@ StepStatus SscWriter::BeginStep(StepMode mode, const float timeoutSeconds)
 {
     TAU_SCOPED_TIMER_FUNC();
 
+    if (m_Threading && m_EndStepThread.joinable())
+    {
+        m_EndStepThread.join();
+    }
+
     ++m_CurrentStep;
 
     if (m_Verbosity >= 5)
@@ -160,7 +182,14 @@ void SscWriter::EndStep()
 
     if (m_CurrentStep == 0)
     {
-        EndStepFirst();
+        if (m_Threading)
+        {
+            m_EndStepThread = std::thread(&SscWriter::EndStepFirst, this);
+        }
+        else
+        {
+            EndStepFirst();
+        }
     }
     else
     {
@@ -170,7 +199,15 @@ void SscWriter::EndStep()
         }
         else
         {
-            EndStepConsequentFlexible();
+            if (m_Threading)
+            {
+                m_EndStepThread =
+                    std::thread(&SscWriter::EndStepConsequentFlexible, this);
+            }
+            else
+            {
+                EndStepConsequentFlexible();
+            }
         }
     }
 }
@@ -393,6 +430,11 @@ void SscWriter::DoClose(const int transportIndex)
                   << ", Writer Rank " << m_WriterRank << std::endl;
     }
 
+    if (m_Threading && m_EndStepThread.joinable())
+    {
+        m_EndStepThread.join();
+    }
+
     if (m_WriterDefinitionsLocked && m_ReaderSelectionsLocked)
     {
         if (m_CurrentStep > 0)

diff --git a/source/adios2/engine/ssc/SscWriter.h b/source/adios2/engine/ssc/SscWriter.h
@@ -52,8 +52,8 @@ class SscWriter : public Engine
     MPI_Win m_MpiWin;
     MPI_Group m_MpiAllReadersGroup;
     MPI_Comm m_StreamComm;
-    std::string m_MpiMode = "twosided";
     std::vector<MPI_Request> m_MpiRequests;
+    std::thread m_EndStepThread;
 
     int m_StreamRank;
     int m_StreamSize;
@@ -85,6 +85,8 @@ class SscWriter : public Engine
 
     int m_Verbosity = 0;
     int m_OpenTimeoutSecs = 10;
+    bool m_Threading = false;
+    std::string m_MpiMode = "twosided";
 };
 
 } // end namespace engine