Support for shared memory provider of libfabric in SST + simple support for manual data progress via threading

source/adios2/engine/sst/SstReader.cpp

@@ -669,29 +669,70 @@ 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, (int)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)
-    {
+    std::vector<void *> nextSstReadHandlers;
+    sstReadHandlers.reserve(BATCH_SIZE);
+    nextSstReadHandlers.reserve(BATCH_SIZE);
+
+    auto iterator = ReadRequests.cbegin();
+    auto end = ReadRequests.cend();
+
+    auto enqueue_next = [&](std::vector<void *>& sstReadHandlers_lambda) {
+        if (iterator == end)
+        {
+            return false;
+        }
+        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, (int)Req.WriterRank, Req.Timestep, Req.StartOffset,
                                        Req.ReadLength, Req.DestinationAddr, dp_info);
-        sstReadHandlers.push_back(ret);
+        sstReadHandlers_lambda.push_back(ret);
+        ++iterator;
+        return true;
+    };
+
+    // Initiate request queue with first BATCH_SIZE requests
+    for (size_t i = 0; i < BATCH_SIZE; ++i)
+    {
+        if (!enqueue_next(sstReadHandlers))
+        {
+            break;
+        }
     }
-    for (const auto &i : sstReadHandlers)
+
+    // Drain current request queue
+    // For each fulfilled request, enqueue the next into the next queue
+    // poor man's asynchrony
+    while (!sstReadHandlers.empty())
     {
-        if (SstWaitForCompletion(m_Input, i) != SstSuccess)
+        nextSstReadHandlers.clear();
+        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");
+            }
+            enqueue_next(nextSstReadHandlers);
         }
+        sstReadHandlers.swap(nextSstReadHandlers);
     }
 
     m_BP5Deserializer->FinalizeGets(ReadRequests);
@@ -710,7 +751,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;
 
@@ -739,14 +780,46 @@ 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;
+        std::vector<void *> nextEnqueuedHandlers;
+        enqueuedHandlers.reserve(BATCH_SIZE);
+        nextEnqueuedHandlers.reserve(BATCH_SIZE);
+
+        auto enqueue_next = [&](std::vector<void *> &enqueuedHandlers) {
+            if (iterator == end)
+            {
+                return false;
+            }
+            enqueuedHandlers.push_back(performDeferredReadRemoteMemory(*iterator));
+            ++iterator;
+            return true;
+        };
+
+        // Initiate request queue with first BATCH_SIZE requests
+        for (size_t i = 0; i < BATCH_SIZE; ++i)
         {
-            if (SstWaitForCompletion(m_Input, i) != SstSuccess)
+            if (!enqueue_next(enqueuedHandlers))
             {
-                helper::Throw<std::runtime_error>("Engine", "SstReader", "PerformGets",
-                                                  "Writer failed before returning data");
+                break;
+            }
+        }
+
+        while (!enqueuedHandlers.empty())
+        {
+            nextEnqueuedHandlers.clear();
+            for (const auto &i : enqueuedHandlers)
+            {
+                if (SstWaitForCompletion(m_Input, i) != SstSuccess)
+                {
+                    helper::Throw<std::runtime_error>("Engine", "SstReader", "PerformGets",
+                                                      "Writer failed before returning data");
+                }
+                enqueue_next(nextEnqueuedHandlers);
             }
+            enqueuedHandlers.swap(nextEnqueuedHandlers);
         }
 
         for (const std::string &name : m_BP3Deserializer->m_DeferredVariables)

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

@@ -25,7 +25,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 +65,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, (int)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 +86,9 @@ void SstReader::ReadVariableBlocksRequests(Variable<T> &variable,
                             subStreamInfo.IntersectionBox, m_BP3Deserializer->m_IsRowMajor,
                             elementOffset))
                     {
-                        auto ret = SstReadRemoteMemory(m_Input, (int)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 +97,9 @@ void SstReader::ReadVariableBlocksRequests(Variable<T> &variable,
                         // batch all read requests
                         buffers.emplace_back();
                         buffers.back().resize(writerBlockSize);
-                        auto ret =
-                            SstReadRemoteMemory(m_Input, (int)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;