Rerouting aggregator

source/adios2/CMakeLists.txt

@@ -33,6 +33,7 @@ add_library(adios2_core
   helper/adiosNetwork.cpp
   helper/adiosPartitioner.cpp
   helper/adiosPluginManager.cpp
+  helper/adiosRerouting.cpp
   helper/adiosString.cpp helper/adiosString.tcc
   helper/adiosSystem.cpp
   helper/adiosType.cpp
@@ -57,6 +58,7 @@ add_library(adios2_core
   engine/bp5/BP5Writer.tcc
   engine/bp5/BP5Writer_TwoLevelShm_Async.cpp
   engine/bp5/BP5Writer_TwoLevelShm.cpp
+  engine/bp5/BP5Writer_WithRerouting.cpp
 
   engine/timeseries/TimeSeriesReader.cpp engine/timeseries/TimeSeriesReader.tcc
 

source/adios2/engine/bp5/BP5Engine.h

@@ -150,7 +150,8 @@ class BP5Engine
     MACRO(DirectIO, Bool, bool, false)                                                             \
     MACRO(DirectIOAlignOffset, UInt, unsigned int, 512)                                            \
     MACRO(DirectIOAlignBuffer, UInt, unsigned int, 0)                                              \
-    MACRO(AggregationType, AggregationType, int, (int)AggregationType::TwoLevelShm)                \
+    MACRO(AggregationType, AggregationType, int, (int)AggregationType::DataSizeBased)              \
+    MACRO(EnableWriterRerouting, Bool, bool, true)                                                 \
     MACRO(AsyncOpen, Bool, bool, true)                                                             \
     MACRO(AsyncWrite, AsyncWrite, int, (int)AsyncWrite::Sync)                                      \
     MACRO(GrowthFactor, Float, float, DefaultBufferGrowthFactor)                                   \

source/adios2/engine/bp5/BP5Writer.cpp

@@ -115,15 +115,15 @@ helper::RankPartition BP5Writer::GetPartitionInfo(const uint64_t rankDataSize, c
 
     m_Profiler.AddTimerWatch("PartitionRanks");
     m_Profiler.Start("PartitionRanks");
-    helper::Partitioning partitioning = helper::PartitionRanks(allsizes, numPartitions);
+    m_Partitioning = helper::PartitionRanks(allsizes, numPartitions);
     m_Profiler.Stop("PartitionRanks");
 
     if (parentRank == 0 && m_Parameters.verbose > 0)
     {
-        partitioning.PrintSummary();
+        m_Partitioning.PrintSummary();
     }
 
-    return partitioning.FindPartition(parentRank);
+    return m_Partitioning.FindPartition(parentRank);
 }
 
 StepStatus BP5Writer::BeginStep(StepMode mode, const float timeoutSeconds)
@@ -390,9 +390,32 @@ void BP5Writer::WriteData(format::BufferV *Data)
             WriteData_EveryoneWrites(Data, false);
             break;
         case (int)AggregationType::EveryoneWritesSerial:
-        case (int)AggregationType::DataSizeBased:
             WriteData_EveryoneWrites(Data, true);
             break;
+        case (int)AggregationType::DataSizeBased:
+            // First initialize aggregator and transports if we haven't done it yet this step
+            if (!m_AggregatorInitializedThisStep)
+            {
+                // We can't allow ranks to change subfiles between calls to Put(), so we only
+                // do this initialization once per timestep. Consequently, partition decision
+                // could be based on incomplete step data.
+                InitAggregator(Data->Size());
+                InitTransports();
+                m_AggregatorInitializedThisStep = true;
+            }
+
+            // For rerouting to be useful, there must be multiple writers sending
+            // data to multiple subfiles.
+            if (m_Parameters.EnableWriterRerouting && m_Comm.Size() > 1 &&
+                m_Aggregator->m_SubStreams > 1)
+            {
+                WriteData_WithRerouting(Data);
+            }
+            else
+            {
+                WriteData_EveryoneWrites(Data, true);
+            }
+            break;
         case (int)AggregationType::TwoLevelShm:
             WriteData_TwoLevelShm(Data);
             break;
@@ -410,19 +433,6 @@ void BP5Writer::WriteData(format::BufferV *Data)
 
 void BP5Writer::WriteData_EveryoneWrites(format::BufferV *Data, bool SerializedWriters)
 {
-    if (m_Parameters.AggregationType == (int)AggregationType::DataSizeBased)
-    {
-        if (!m_AggregatorInitializedThisStep)
-        {
-            // We can't allow ranks to change subfiles between calls to Put(), so we only
-            // do this initialization once per timestep. Consequently, partition decision
-            // could be based on incomplete step data.
-            InitAggregator(Data->Size());
-            InitTransports();
-            m_AggregatorInitializedThisStep = true;
-        }
-    }
-
     const aggregator::MPIChain *a = dynamic_cast<aggregator::MPIChain *>(m_Aggregator);
 
     // new step writing starts at offset m_DataPos on aggregator
@@ -1020,6 +1030,33 @@ void BP5Writer::EndStep()
 
     m_Profiler.Stop("ES_AWD");
 
+    if (m_Parameters.verbose > 2)
+    {
+        std::cout << "Rank " << m_Comm.Rank() << " deciding whether new writer map is needed"
+                  << std::endl;
+        std::cout << "  m_WriterStep: " << m_WriterStep << std::endl;
+        std::cout << "  m_AppendWriterCount: " << m_AppendWriterCount
+                  << ", m_Comm.Size(): " << m_Comm.Size() << std::endl;
+        std::cout << "  m_AppendAggregatorCount: " << m_AppendAggregatorCount
+                  << ", m_Aggregator->m_NumAggregators: " << m_Aggregator->m_NumAggregators
+                  << std::endl;
+        std::cout << "  m_AppendSubfileCount: " << m_AppendSubfileCount
+                  << ", m_Aggregator->m_SubStreams: " << m_Aggregator->m_SubStreams << std::endl;
+    }
+
+    if (!m_WriterStep || m_AppendWriterCount != static_cast<unsigned int>(m_Comm.Size()) ||
+        m_AppendAggregatorCount != static_cast<unsigned int>(m_Aggregator->m_NumAggregators) ||
+        m_AppendSubfileCount != static_cast<unsigned int>(m_Aggregator->m_SubStreams))
+    {
+        // new Writer Map is needed
+        if (m_Parameters.verbose > 2)
+        {
+            std::cout << "Rank " << m_Comm.Rank() << " new writer map needed" << std::endl;
+        }
+        const uint64_t a = static_cast<uint64_t>(m_Aggregator->m_SubStreamIndex);
+        m_WriterSubfileMap = m_Comm.GatherValues(a, 0);
+    }
+
     if (m_Parameters.UseSelectiveMetadataAggregation)
     {
         SelectiveAggregationMetadata(TSInfo);
@@ -1609,29 +1646,9 @@ void BP5Writer::InitMetadataTransports()
 
 void BP5Writer::InitTransports()
 {
-    std::string cacheKey = GetCacheKey(m_Aggregator);
-    auto search = m_AggregatorSpecifics.find(cacheKey);
-    bool cacheHit = false;
+    OpenSubfile();
 
-    if (search != m_AggregatorSpecifics.end())
-    {
-        if (m_Parameters.verbose > 2)
-        {
-            std::cout << "Rank " << m_Comm.Rank() << " cache hit for aggregator key " << cacheKey
-                      << std::endl;
-        }
-        cacheHit = true;
-    }
-    else
-    {
-        // Didn't have one in the cache, add it now
-        m_AggregatorSpecifics.emplace(std::make_pair(cacheKey, AggTransportData(m_IO, m_Comm)));
-    }
-
-    AggTransportData &aggData = m_AggregatorSpecifics.at(cacheKey);
-
-    // /path/name.bp.dir/name.bp.rank
-    aggData.m_SubStreamNames = GetBPSubStreamNames({m_Name}, m_Aggregator->m_SubStreamIndex);
+    AggTransportData &aggData = m_AggregatorSpecifics.at(GetCacheKey(m_Aggregator));
 
     if (m_IAmDraining)
     {
@@ -1658,9 +1675,49 @@ void BP5Writer::InitTransports()
             aggData.m_DrainSubStreamNames, m_IO.m_TransportsParameters, m_Parameters.NodeLocal);
     }
 
+    if (m_IAmDraining)
+    {
+        if (m_DrainBB)
+        {
+            for (const auto &name : aggData.m_DrainSubStreamNames)
+            {
+                m_FileDrainer.AddOperationOpen(name, m_OpenMode);
+            }
+        }
+    }
+
+    InitBPBuffer();
+}
+
+void BP5Writer::OpenSubfile(bool useComm, bool forceAppend)
+{
+    std::string cacheKey = GetCacheKey(m_Aggregator);
+    auto search = m_AggregatorSpecifics.find(cacheKey);
+    bool cacheHit = false;
+
+    if (search != m_AggregatorSpecifics.end())
+    {
+        if (m_Parameters.verbose > 2)
+        {
+            std::cout << "Rank " << m_Comm.Rank() << " cache hit for aggregator key " << cacheKey
+                      << std::endl;
+        }
+        cacheHit = true;
+    }
+    else
+    {
+        // Didn't have one in the cache, add it now
+        m_AggregatorSpecifics.emplace(std::make_pair(cacheKey, AggTransportData(m_IO, m_Comm)));
+    }
+
+    AggTransportData &aggData = m_AggregatorSpecifics.at(cacheKey);
+
+    // /path/name.bp.dir/name.bp.rank
+    aggData.m_SubStreamNames = GetBPSubStreamNames({m_Name}, m_Aggregator->m_SubStreamIndex);
+
     helper::Comm openSyncComm;
 
-    if (m_Parameters.AggregationType == (int)AggregationType::DataSizeBased)
+    if (m_Parameters.AggregationType == (int)AggregationType::DataSizeBased && useComm)
     {
         // Split my writer chain so only ranks that actually need to open a
         // file can do so in an ordered fashion.
@@ -1679,7 +1736,7 @@ void BP5Writer::InitTransports()
 
             if (m_Parameters.AggregationType == (int)AggregationType::DataSizeBased)
             {
-                if (m_WriterStep > 0)
+                if (m_WriterStep > 0 || forceAppend)
                 {
                     // override the mode to be append if we're opening a file that
                     // was already opened by another rank.
@@ -1691,29 +1748,24 @@ void BP5Writer::InitTransports()
             {
                 std::cout << "Rank " << m_Comm.Rank() << " opening data file" << std::endl;
             }
-            aggData.m_FileDataManager.OpenFiles(aggData.m_SubStreamNames, mode,
-                                                m_IO.m_TransportsParameters, true,
-                                                *DataWritingComm);
+            if (useComm)
+            {
+                aggData.m_FileDataManager.OpenFiles(aggData.m_SubStreamNames, mode,
+                                                    m_IO.m_TransportsParameters, true,
+                                                    *DataWritingComm);
+            }
+            else
+            {
+                aggData.m_FileDataManager.OpenFiles(aggData.m_SubStreamNames, mode,
+                                                    m_IO.m_TransportsParameters, true);
+            }
         }
     }
 
-    if (m_Parameters.AggregationType == (int)AggregationType::DataSizeBased)
+    if (m_Parameters.AggregationType == (int)AggregationType::DataSizeBased && useComm)
     {
         openSyncComm.Free();
     }
-
-    if (m_IAmDraining)
-    {
-        if (m_DrainBB)
-        {
-            for (const auto &name : aggData.m_DrainSubStreamNames)
-            {
-                m_FileDrainer.AddOperationOpen(name, m_OpenMode);
-            }
-        }
-    }
-
-    this->InitBPBuffer();
 }
 
 /*generate the header for the metadata index file*/
@@ -1963,33 +2015,6 @@ void BP5Writer::InitBPBuffer()
     {
         m_WriterDataPos.resize(m_Comm.Size());
     }
-
-    if (m_Parameters.verbose > 2)
-    {
-        std::cout << "Rank " << m_Comm.Rank() << " deciding whether new writer map is needed"
-                  << std::endl;
-        std::cout << "  m_WriterStep: " << m_WriterStep << std::endl;
-        std::cout << "  m_AppendWriterCount: " << m_AppendWriterCount
-                  << ", m_Comm.Size(): " << m_Comm.Size() << std::endl;
-        std::cout << "  m_AppendAggregatorCount: " << m_AppendAggregatorCount
-                  << ", m_Aggregator->m_NumAggregators: " << m_Aggregator->m_NumAggregators
-                  << std::endl;
-        std::cout << "  m_AppendSubfileCount: " << m_AppendSubfileCount
-                  << ", m_Aggregator->m_SubStreams: " << m_Aggregator->m_SubStreams << std::endl;
-    }
-
-    if (!m_WriterStep || m_AppendWriterCount != static_cast<unsigned int>(m_Comm.Size()) ||
-        m_AppendAggregatorCount != static_cast<unsigned int>(m_Aggregator->m_NumAggregators) ||
-        m_AppendSubfileCount != static_cast<unsigned int>(m_Aggregator->m_SubStreams))
-    {
-        // new Writer Map is needed, generate now, write later
-        if (m_Parameters.verbose > 2)
-        {
-            std::cout << "Rank " << m_Comm.Rank() << " new writer map needed" << std::endl;
-        }
-        const uint64_t a = static_cast<uint64_t>(m_Aggregator->m_SubStreamIndex);
-        m_WriterSubfileMap = m_Comm.GatherValues(a, 0);
-    }
 }
 
 void BP5Writer::EnterComputationBlock() noexcept

source/adios2/engine/bp5/BP5Writer.h

@@ -78,6 +78,8 @@ class BP5Writer : public BP5Engine, public core::Engine
     std::map<std::string, AggTransportData> m_AggregatorSpecifics;
     helper::RankPartition GetPartitionInfo(const uint64_t rankDataSize, const int subStreams,
                                            helper::Comm const &parentComm);
+    void OpenSubfile(const bool useComm = true, const bool forceAppend = false);
+    helper::Partitioning m_Partitioning;
 
     /** Single object controlling BP buffering */
     format::BP5Serializer m_BP5Serializer;
@@ -196,6 +198,7 @@ class BP5Writer : public BP5Engine, public core::Engine
     /** Write Data to disk, in an aggregator chain */
     void WriteData(format::BufferV *Data);
     void WriteData_EveryoneWrites(format::BufferV *Data, bool SerializedWriters);
+    void WriteData_WithRerouting(format::BufferV *Data);
     void WriteData_EveryoneWrites_Async(format::BufferV *Data, bool SerializedWriters);
     void WriteData_TwoLevelShm(format::BufferV *Data);
     void WriteData_TwoLevelShm_Async(format::BufferV *Data);
@@ -302,6 +305,17 @@ class BP5Writer : public BP5Engine, public core::Engine
      */
     uint64_t CountStepsInMetadataIndex(format::BufferSTL &bufferSTL);
 
+    // Thread function for inter-rank communication when rerouting aggregation
+    // is enabled
+    void ReroutingCommunicationLoop();
+    int m_TargetIndex;
+    int m_TargetCoordinator;
+    std::mutex m_WriteMutex;
+    std::mutex m_NotifMutex;
+    std::condition_variable m_WriteCV;
+    bool m_ReadyToWrite;
+    bool m_FinishedWriting;
+
     /* Async write's future */
     std::future<int> m_WriteFuture;
     // variables to delay writing to index file