From 09b910e0d247c9cb022a024d02b87749c0e3f73c Mon Sep 17 00:00:00 2001 From: Norbert Podhorszki Date: Mon, 8 Jul 2019 10:54:26 -0400 Subject: [PATCH] Rewrite BP4Serializer::DeserializeIndicesPerRankThreads hoping XL compiler will work with it. --- .../toolkit/format/bp4/BP4Serializer.cpp | 171 +++++++++++++++--- .../adios2/toolkit/format/bp4/BP4Serializer.h | 8 + 2 files changed, 156 insertions(+), 23 deletions(-) diff --git a/source/adios2/toolkit/format/bp4/BP4Serializer.cpp b/source/adios2/toolkit/format/bp4/BP4Serializer.cpp index 368f24cd02..07be2cc856 100644 --- a/source/adios2/toolkit/format/bp4/BP4Serializer.cpp +++ b/source/adios2/toolkit/format/bp4/BP4Serializer.cpp @@ -1081,13 +1081,14 @@ std::vector BP4Serializer::SerializeIndices( } std::unordered_map> -BP4Serializer::DeserializeIndicesPerRankThreads( +BP4Serializer::DeserializeIndicesPerRankSingleThread( const std::vector &serialized, MPI_Comm comm, const bool isRankConstant) const noexcept { std::unordered_map> deserialized; + /* auto lf_Deserialize_no_mutex = [&](const int rankSource, const size_t serializedPosition, const bool isRankConstant) { @@ -1139,7 +1140,152 @@ BP4Serializer::DeserializeIndicesPerRankThreads( SerialElementIndex &index = deserializedIndexes->at(rankSource); helper::InsertToBuffer(index.Buffer, &serialized[serializedPosition], bufferSize); - }; + };*/ + + // BODY OF FUNCTION starts here + const size_t serializedSize = serialized.size(); + int rank; + SMPI_Comm_rank(comm, &rank); + + if (rank != 0 || serializedSize < 8) + { + return deserialized; + } + + size_t serializedPosition = 0; + + while (serializedPosition < serializedSize - 4) + { + const int rankSource = static_cast( + helper::ReadValue(serialized, serializedPosition)); + + if (serializedPosition <= serializedSize) + { + // lf_Deserialize_no_mutex(rankSource, serializedPosition, + // isRankConstant); + size_t localPosition = serializedPosition; + + ElementIndexHeader header = + ReadElementIndexHeader(serialized, localPosition); + + if (!isRankConstant || deserialized.count(header.Name) != 1) + { + + std::vector *deserializedIndexes; + + // deserializedIndexes = + // &(deserialized.emplace(std::piecewise_construct, + // std::forward_as_tuple(header.Name), + // std::forward_as_tuple( + // m_SizeMPI, + // SerialElementIndex(header.MemberID, + // 0))).first->second); + + auto search = deserialized.find(header.Name); + if (search == deserialized.end()) + { + // variable does not exist, we need to add it + deserializedIndexes = + &(deserialized + .emplace(std::piecewise_construct, + std::forward_as_tuple(header.Name), + std::forward_as_tuple( + m_SizeMPI, SerialElementIndex( + header.MemberID, 0))) + .first->second); + // std::cout << "rank " << rankSource << ": did not find " + // << header.Name << ", added it" << std::endl; + } + else + { + deserializedIndexes = &(search->second); + // std::cout << "rank " << rankSource << ": found " << + // header.Name + // << std::endl; + } + + const size_t bufferSize = + static_cast(header.Length) + 4; + SerialElementIndex &index = deserializedIndexes->at(rankSource); + helper::InsertToBuffer( + index.Buffer, &serialized[serializedPosition], bufferSize); + } + } + + const size_t bufferSize = static_cast( + helper::ReadValue(serialized, serializedPosition)); + serializedPosition += bufferSize; + } + + return deserialized; +} + +std::unordered_map> +BP4Serializer::DeserializeIndicesPerRankThreads( + const std::vector &serialized, MPI_Comm comm, + const bool isRankConstant) const noexcept +{ + if (m_Threads == 1) + { + return BP4Serializer::DeserializeIndicesPerRankSingleThread( + serialized, comm, isRankConstant); + } + + std::unordered_map> + deserialized; + /* + auto lf_Deserialize_no_mutex = [&](const int rankSource, + const size_t serializedPosition, + const bool isRankConstant) { + size_t localPosition = serializedPosition; + + ElementIndexHeader header = + ReadElementIndexHeader(serialized, localPosition); + + if (isRankConstant) + { + if (deserialized.count(header.Name) == 1) + { + return; + } + } + + std::vector *deserializedIndexes; + + // deserializedIndexes = + // &(deserialized.emplace(std::piecewise_construct, + // std::forward_as_tuple(header.Name), + // std::forward_as_tuple( + // m_SizeMPI, SerialElementIndex(header.MemberID, + // 0))).first->second); + + auto search = deserialized.find(header.Name); + if (search == deserialized.end()) + { + // variable does not exist, we need to add it + deserializedIndexes = + &(deserialized + .emplace(std::piecewise_construct, + std::forward_as_tuple(header.Name), + std::forward_as_tuple( + m_SizeMPI, + SerialElementIndex(header.MemberID, 0))) + .first->second); + // std::cout << "rank " << rankSource << ": did not find " << + // header.Name << ", added it" << std::endl; + } + else + { + deserializedIndexes = &(search->second); + // std::cout << "rank " << rankSource << ": found " << header.Name + // << std::endl; + } + + const size_t bufferSize = static_cast(header.Length) + 4; + SerialElementIndex &index = deserializedIndexes->at(rankSource); + helper::InsertToBuffer(index.Buffer, &serialized[serializedPosition], + bufferSize); + };*/ auto lf_Deserialize = [&](const int rankSource, const size_t serializedPosition, @@ -1188,27 +1334,6 @@ BP4Serializer::DeserializeIndicesPerRankThreads( size_t serializedPosition = 0; - if (m_Threads == 1) - { - while (serializedPosition < serializedSize) - { - const int rankSource = static_cast( - helper::ReadValue(serialized, serializedPosition)); - - if (serializedPosition <= serializedSize) - { - lf_Deserialize_no_mutex(rankSource, serializedPosition, - isRankConstant); - } - - const size_t bufferSize = static_cast( - helper::ReadValue(serialized, serializedPosition)); - serializedPosition += bufferSize; - } - - return deserialized; - } - std::vector> asyncs(m_Threads); std::vector asyncPositions(m_Threads); std::vector asyncRankSources(m_Threads); diff --git a/source/adios2/toolkit/format/bp4/BP4Serializer.h b/source/adios2/toolkit/format/bp4/BP4Serializer.h index d02bd61d06..82fb8ca222 100644 --- a/source/adios2/toolkit/format/bp4/BP4Serializer.h +++ b/source/adios2/toolkit/format/bp4/BP4Serializer.h @@ -416,6 +416,14 @@ class BP4Serializer : public BP4Base MPI_Comm comm, const bool isRankConstant) const noexcept; + /** private function called by DeserializeIndicesPerRankThreads + * in case of a single thread + */ + std::unordered_map> + DeserializeIndicesPerRankSingleThread(const std::vector &serialized, + MPI_Comm comm, + const bool isRankConstant) const + noexcept; /** * Merge indices by time step (default) and write to m_HeapBuffer.m_Metadata * @param nameRankIndices