Merge pull request #2828 from JasonRuonanWang/mhs

added runtime control of numbers of tiers to read in mhs

source/adios2/engine/mhs/MhsReader.cpp

@@ -23,7 +23,7 @@ MhsReader::MhsReader(IO &io, const std::string &name, const Mode mode,
 {
     helper::GetParameter(io.m_Parameters, "Tiers", m_Tiers);
     Params params = {{"tiers", std::to_string(m_Tiers)}};
-    m_Compressor = std::make_shared<compress::CompressSirius>(params);
+    m_SiriusCompressor = std::make_shared<compress::CompressSirius>(params);
     io.SetEngine("");
     m_SubIOs.emplace_back(&io);
     m_SubEngines.emplace_back(&io.Open(m_Name + ".tier0", adios2::Mode::Read));

source/adios2/engine/mhs/MhsReader.h

@@ -41,7 +41,7 @@ class MhsReader : public Engine
 private:
     std::vector<IO *> m_SubIOs;
     std::vector<Engine *> m_SubEngines;
-    std::shared_ptr<compress::CompressSirius> m_Compressor;
+    std::shared_ptr<compress::CompressSirius> m_SiriusCompressor;
     int m_Tiers;
 
 #define declare_type(T)                                                        \

source/adios2/engine/mhs/MhsReader.tcc

@@ -46,6 +46,10 @@ void MhsReader::GetDeferredCommon(Variable<T> &variable, T *data)
         }
         var->SetSelection({variable.m_Start, variable.m_Count});
         m_SubEngines[i]->Get(*var, data, Mode::Sync);
+        if (m_SiriusCompressor->m_CurrentReadFinished)
+        {
+            break;
+        }
     }
 }
 

source/adios2/operator/compress/CompressSirius.cpp

@@ -24,6 +24,7 @@ std::vector<std::unordered_map<std::string, std::vector<char>>>
 int CompressSirius::m_CurrentTier;
 std::vector<std::vector<char>> CompressSirius::m_TierBuffers;
 int CompressSirius::m_Tiers = 0;
+bool CompressSirius::m_CurrentReadFinished = false;
 
 CompressSirius::CompressSirius(const Params &parameters)
 : Operator("sirius", parameters)
@@ -88,6 +89,12 @@ size_t CompressSirius::Decompress(const void *bufferIn, const size_t sizeIn,
     // if called from the final tier, then merge all tier buffers and copy back
     // to dataOut
     size_t accumulatedBytes = 0;
+    // TODO: it currently only copies output data back when the final tier is
+    // read. However, the real Sirius algorithm should instead decide when to
+    // copy back decompressed data based on required acuracy level. Once it's
+    // done, it should set m_CurrentReadFinished to true to inform the MHS
+    // engine that the current read is finished so that it won't read the next
+    // tier.
     if (currentTier == m_Tiers - 1)
     {
         for (auto &bmap : m_TierBuffersMap)
@@ -97,8 +104,16 @@ size_t CompressSirius::Decompress(const void *bufferIn, const size_t sizeIn,
                         b.data(), b.size());
             accumulatedBytes += b.size();
         }
+        // set m_CurrentReadFinished to true if after the current call, the
+        // required acuracy is already satisfied, so that the MHS engine knows
+        // it shouldn't continue reading the next tier.
+        m_CurrentReadFinished = true;
     }
 
+    // set m_CurrentReadFinished to false if the current tier does not satisfy
+    // the required acuracy, so the MHS engine will read the next tier.
+    m_CurrentReadFinished = false;
+
     currentTier++;
     if (currentTier % m_Tiers == 0)
     {

source/adios2/operator/compress/CompressSirius.h

@@ -41,6 +41,8 @@ class CompressSirius : public Operator
 
     bool IsDataTypeValid(const DataType type) const final;
 
+    static bool m_CurrentReadFinished;
+
 private:
     static int m_Tiers;