Enqueue remote reads in batches

Ref. the meeting with the Maestro team: The fabric should not be
bombarded with too many requests at once. Batch size currently hardcoded
as 10.

Author: franzpoeschel

source/adios2/engine/sst/SstReader.cpp

@@ -669,28 +669,46 @@ bool SstReader::VariableMinMax(const VariableBase &Var, const size_t Step, MinMa
     return m_BP5Deserializer->VariableMinMax(Var, Step, MinMax);
 }
 
+void *SstReader::performDeferredReadRemoteMemory(DeferredReadRemoteMemory const &params)
+{
+    return SstReadRemoteMemory(m_Input, params.rank, CurrentStep(), params.payloadStart,
+                               params.payloadSize, params.buffer, params.dp_info);
+}
+
+constexpr static size_t BATCH_SIZE = 10;
+
 void SstReader::BP5PerformGets()
 {
     size_t maxReadSize;
     auto ReadRequests = m_BP5Deserializer->GenerateReadRequests(true, &maxReadSize);
     std::vector<void *> sstReadHandlers;
-    for (const auto &Req : ReadRequests)
+
+    auto iterator = ReadRequests.cbegin();
+    auto end = ReadRequests.cend();
+    while (iterator != end)
     {
-        void *dp_info = NULL;
-        if (m_CurrentStepMetaData->DP_TimestepInfo)
+        sstReadHandlers.clear();
+        size_t counter = 0;
+        for (; counter < BATCH_SIZE && iterator != end; ++iterator, ++counter)
         {
-            dp_info = m_CurrentStepMetaData->DP_TimestepInfo[Req.WriterRank];
+            auto const &Req = *iterator;
+
+            void *dp_info = NULL;
+            if (m_CurrentStepMetaData->DP_TimestepInfo)
+            {
+                dp_info = m_CurrentStepMetaData->DP_TimestepInfo[Req.WriterRank];
+            }
+            auto ret = SstReadRemoteMemory(m_Input, Req.WriterRank, Req.Timestep, Req.StartOffset,
+                                           Req.ReadLength, Req.DestinationAddr, dp_info);
+            sstReadHandlers.push_back(ret);
         }
-        auto ret = SstReadRemoteMemory(m_Input, Req.WriterRank, Req.Timestep, Req.StartOffset,
-                                       Req.ReadLength, Req.DestinationAddr, dp_info);
-        sstReadHandlers.push_back(ret);
-    }
-    for (const auto &i : sstReadHandlers)
-    {
-        if (SstWaitForCompletion(m_Input, i) != SstSuccess)
+        for (const auto &i : sstReadHandlers)
         {
-            helper::Throw<std::runtime_error>("Engine", "SstReader", "BP5PerformGets",
-                                              "Writer failed before returning data");
+            if (SstWaitForCompletion(m_Input, i) != SstSuccess)
+            {
+                helper::Throw<std::runtime_error>("Engine", "SstReader", "BP5PerformGets",
+                                                  "Writer failed before returning data");
+            }
         }
     }
 
@@ -709,7 +727,7 @@ void SstReader::PerformGets()
     }
     else if (m_WriterMarshalMethod == SstMarshalBP)
     {
-        std::vector<void *> sstReadHandlers;
+        std::vector<DeferredReadRemoteMemory> sstReadHandlers;
         std::vector<std::vector<char>> buffers;
         size_t iter = 0;
 
@@ -738,13 +756,26 @@ void SstReader::PerformGets()
             ADIOS2_FOREACH_STDTYPE_1ARG(declare_type)
 #undef declare_type
         }
-        // wait for all SstRead requests to finish
-        for (const auto &i : sstReadHandlers)
+        // run read requests in batches and wait for them to finish
+        auto iterator = sstReadHandlers.cbegin();
+        auto end = sstReadHandlers.cend();
+        std::vector<void *> enqueuedHandlers;
+        enqueuedHandlers.reserve(BATCH_SIZE);
+        while (iterator != end)
         {
-            if (SstWaitForCompletion(m_Input, i) != SstSuccess)
+            size_t counter = 0;
+            enqueuedHandlers.clear();
+            for (; counter < BATCH_SIZE && iterator != end; ++iterator, ++counter)
             {
-                helper::Throw<std::runtime_error>("Engine", "SstReader", "PerformGets",
-                                                  "Writer failed before returning data");
+                enqueuedHandlers.push_back(performDeferredReadRemoteMemory(*iterator));
+            }
+            for (const auto &i : enqueuedHandlers)
+            {
+                if (SstWaitForCompletion(m_Input, i) != SstSuccess)
+                {
+                    helper::Throw<std::runtime_error>("Engine", "SstReader", "PerformGets",
+                                                      "Writer failed before returning data");
+                }
             }
         }
 

source/adios2/engine/sst/SstReader.h

@@ -54,8 +54,19 @@ class SstReader : public Engine
     bool VariableMinMax(const VariableBase &, const size_t Step, MinMaxStruct &MinMax);
 
 private:
+    struct DeferredReadRemoteMemory
+    {
+        size_t rank;
+        size_t payloadStart;
+        size_t payloadSize;
+        char *buffer;
+        void *dp_info;
+    };
+    void * performDeferredReadRemoteMemory(DeferredReadRemoteMemory const &);
+
     template <class T>
-    void ReadVariableBlocksRequests(Variable<T> &variable, std::vector<void *> &sstReadHandlers,
+    void ReadVariableBlocksRequests(Variable<T> &variable,
+                                    std::vector<DeferredReadRemoteMemory> &sstReadHandlers,
                                     std::vector<std::vector<char>> &buffers);
 
     template <class T>

source/adios2/engine/sst/SstReader.tcc

@@ -15,6 +15,7 @@
 
 #include "adios2/helper/adiosFunctions.h" //GetDataType<T>
 #include <adios2-perfstubs-interface.h>
+#include <initializer_list>
 
 namespace adios2
 {
@@ -25,7 +26,7 @@ namespace engine
 
 template <class T>
 void SstReader::ReadVariableBlocksRequests(Variable<T> &variable,
-                                           std::vector<void *> &sstReadHandlers,
+                                           std::vector<DeferredReadRemoteMemory> &sstReadHandlers,
                                            std::vector<std::vector<char>> &buffers)
 {
     PERFSTUBS_SCOPED_TIMER_FUNC();
@@ -65,9 +66,8 @@ void SstReader::ReadVariableBlocksRequests(Variable<T> &variable,
                     std::stringstream ss;
                     ss << "SST Bytes Read from remote rank " << rank;
                     PERFSTUBS_SAMPLE_COUNTER(ss.str().c_str(), payloadSize);
-                    auto ret = SstReadRemoteMemory(m_Input, rank, CurrentStep(), payloadStart,
-                                                   payloadSize, buffer, dp_info);
-                    sstReadHandlers.push_back(ret);
+                    sstReadHandlers.push_back(
+                        DeferredReadRemoteMemory{rank, payloadStart, payloadSize, buffer, dp_info});
                 }
                 // if remote data buffer is not compressed
                 else
@@ -87,10 +87,9 @@ void SstReader::ReadVariableBlocksRequests(Variable<T> &variable,
                             subStreamInfo.IntersectionBox, m_BP3Deserializer->m_IsRowMajor,
                             elementOffset))
                     {
-                        auto ret = SstReadRemoteMemory(m_Input, rank, CurrentStep(),
-                                                       writerBlockStart, writerBlockSize,
-                                                       blockInfo.Data + elementOffset, dp_info);
-                        sstReadHandlers.push_back(ret);
+                        sstReadHandlers.push_back(DeferredReadRemoteMemory{
+                            rank, writerBlockStart, writerBlockSize,
+                            reinterpret_cast<char *>(blockInfo.Data + elementOffset), dp_info});
                     }
                     // if either input or output is not contiguous memory then
                     // find all contiguous parts.
@@ -99,10 +98,9 @@ void SstReader::ReadVariableBlocksRequests(Variable<T> &variable,
                         // batch all read requests
                         buffers.emplace_back();
                         buffers.back().resize(writerBlockSize);
-                        auto ret =
-                            SstReadRemoteMemory(m_Input, rank, CurrentStep(), writerBlockStart,
-                                                writerBlockSize, buffers.back().data(), dp_info);
-                        sstReadHandlers.push_back(ret);
+                        sstReadHandlers.push_back(
+                            DeferredReadRemoteMemory{rank, writerBlockStart, writerBlockSize,
+                                                     buffers.back().data(), dp_info});
                     }
                 }
                 ++threadID;