diff --git a/source/adios2/engine/bp5/BP5Reader.cpp b/source/adios2/engine/bp5/BP5Reader.cpp index 815ae752e4..d498997b9a 100644 --- a/source/adios2/engine/bp5/BP5Reader.cpp +++ b/source/adios2/engine/bp5/BP5Reader.cpp @@ -227,8 +227,8 @@ void BP5Reader::ReadData(const size_t WriterRank, const size_t Timestep, } ThisDataPos = helper::ReadValue( m_MetadataIndex.m_Buffer, ThisFlushInfo, m_Minifooter.IsLittleEndian); - m_DataFileManager.ReadFile(Destination, RemainingLength, ThisDataPos, - SubfileNum); + m_DataFileManager.ReadFile(Destination, RemainingLength, + ThisDataPos + Offset, SubfileNum); } void BP5Reader::PerformGets() diff --git a/source/adios2/helper/adiosMemory.h b/source/adios2/helper/adiosMemory.h index f01fc42fcd..75bced3a7d 100644 --- a/source/adios2/helper/adiosMemory.h +++ b/source/adios2/helper/adiosMemory.h @@ -24,6 +24,8 @@ namespace adios2 namespace helper { +const size_t MAX_DIMS = 32; + #ifdef ADIOS2_HAVE_ENDIAN_REVERSE template void CopyEndianReverse(const char *src, const size_t payloadStride, T *dest); diff --git a/source/adios2/toolkit/format/bp5/BP5Deserializer.cpp b/source/adios2/toolkit/format/bp5/BP5Deserializer.cpp index 8fccc7032d..eb367de40d 100644 --- a/source/adios2/toolkit/format/bp5/BP5Deserializer.cpp +++ b/source/adios2/toolkit/format/bp5/BP5Deserializer.cpp @@ -30,6 +30,7 @@ #include "adios2/operator/OperatorFactory.h" +#include #include #include #include @@ -1013,258 +1014,301 @@ bool BP5Deserializer::QueueGetSingle(core::VariableBase &variable, return true; } -bool BP5Deserializer::NeedWriter(BP5ArrayRequest Req, size_t WriterRank, - size_t &NodeFirst) +static bool IntersectionStartCount(const size_t dimensionsSize, + const size_t *start1, const size_t *count1, + const size_t *start2, const size_t *count2, + size_t *outstart, size_t *outcount) noexcept { - MetaArrayRec *writer_meta_base = - (MetaArrayRec *)GetMetadataBase(Req.VarRec, Req.Step, WriterRank); - - if (!writer_meta_base) - return false; - - if (Req.RequestType == Local) + for (size_t d = 0; d < dimensionsSize; ++d) { - size_t WriterBlockCount = - writer_meta_base->Dims - ? writer_meta_base->DBCount / writer_meta_base->Dims - : 1; - if (m_RandomAccessMode) - { - // Not ideal, but we don't keep this around for every var in random - // access mode, so calc from scratch - NodeFirst = 0; - for (size_t TmpRank = 0; TmpRank < WriterRank; TmpRank++) - { - ControlInfo *TmpCI = - m_ControlArray[Req.Step][TmpRank]; // writer control array - - size_t MetadataFieldOffset = - (*TmpCI->MetaFieldOffset)[Req.VarRec->VarNum]; - MetaArrayRec *tmp_meta_base = - (MetaArrayRec - *)(((char *)(*MetadataBaseArray[Req.Step])[TmpRank]) + - MetadataFieldOffset); - size_t TmpBlockCount = - tmp_meta_base->Dims - ? tmp_meta_base->DBCount / tmp_meta_base->Dims - : 1; - NodeFirst += TmpBlockCount; - } + // Don't intercept + const size_t end1 = start1[d] + count1[d] - 1; + const size_t end2 = start2[d] + count2[d] - 1; + + if ((count1[d] == 0) || (count2[d] == 0)) + { + return false; } - else + if (start2[d] > end1 || end2 < start1[d]) { - NodeFirst = Req.VarRec->PerWriterBlockStart[WriterRank]; + return false; } - size_t NodeLast = WriterBlockCount + NodeFirst - 1; - bool res = (NodeFirst <= Req.BlockID) && (NodeLast >= Req.BlockID); - return res; } - // else Global case - for (size_t i = 0; i < writer_meta_base->BlockCount; i++) + for (size_t d = 0; d < dimensionsSize; d++) { - bool NeedThisBlock = true; - for (size_t j = 0; j < writer_meta_base->Dims; j++) - { - size_t SelOffset = Req.Start[j]; - size_t SelSize = Req.Count[j]; - size_t RankOffset; - size_t RankSize; + const size_t intersectionStart = + (start1[d] < start2[d]) ? start2[d] : start1[d]; - RankOffset = - writer_meta_base->Offsets[i * writer_meta_base->Dims + j]; - RankSize = writer_meta_base->Count[i * writer_meta_base->Dims + j]; - if ((SelSize == 0) || (RankSize == 0)) - { - NeedThisBlock = false; - } - if ((RankOffset < SelOffset && - (RankOffset + RankSize) <= SelOffset) || - (RankOffset >= SelOffset + SelSize)) - { - NeedThisBlock = false; - } - } - if (NeedThisBlock) - return true; + // end, must be inclusive + const size_t end1 = start1[d] + count1[d] - 1; + const size_t end2 = start2[d] + count2[d] - 1; + const size_t intersectionEnd = (end1 > end2) ? end2 : end1; + outstart[d] = intersectionStart; + outcount[d] = intersectionEnd - intersectionStart + 1; + if (outcount[d] == 0) + return false; } - return false; + return true; } -std::vector -BP5Deserializer::GenerateReadRequests() +static size_t LinearIndex(const size_t dimensionsSize, const size_t *count, + const size_t *pos, bool IsRowMajor) noexcept { - std::vector Ret; - // std::vector WriterInfo(m_WriterCohortSize); - typedef std::pair pair; - std::map WriterTSNeeded; - - for (const auto &Req : PendingRequests) + size_t offset = 0; + if (IsRowMajor) { - const size_t writerCohortSize = WriterCohortSize(Req.Step); - size_t NodeFirst = 0; - for (size_t i = 0; i < writerCohortSize; i++) + for (size_t d = 0; d < dimensionsSize; ++d) { - if (!NeedWriter(Req, i, NodeFirst)) - { - continue; - } - if (WriterTSNeeded.count(std::make_pair(Req.Step, i)) == 0) - { - WriterTSNeeded[std::make_pair(Req.Step, i)] = true; - } + offset = offset * count[d] + pos[d]; } } - - for (std::pair element : WriterTSNeeded) + else { - ReadRequest RR; - RR.Timestep = element.first.first; - RR.WriterRank = element.first.second; - RR.StartOffset = 0; - if (m_RandomAccessMode) + for (size_t d = dimensionsSize - 1; d < dimensionsSize; d--) { - RR.ReadLength = - ((struct BP5MetadataInfoStruct *)(( - *MetadataBaseArray[RR.Timestep])[RR.WriterRank])) - ->DataBlockSize; + offset = offset * count[d] + pos[d]; } - else - { - RR.ReadLength = ((struct BP5MetadataInfoStruct - *)(*m_MetadataBaseAddrs)[RR.WriterRank]) - ->DataBlockSize; - } - RR.DestinationAddr = (char *)malloc(RR.ReadLength); - RR.Internal = NULL; - Ret.push_back(RR); } - return Ret; + return offset; } -void BP5Deserializer::FinalizeGets(std::vector Requests) +static size_t CalcBlockLength(const size_t dimensionsSize, const size_t *count) { - for (const auto &Req : PendingRequests) + size_t len = count[0]; + for (size_t d = 1; d < dimensionsSize; ++d) { - // ImplementGapWarning(Reqs); - const size_t writerCohortSize = WriterCohortSize(Req.Step); - for (size_t WriterRank = 0; WriterRank < writerCohortSize; WriterRank++) + len = len * count[d]; + } + return len; +} + +std::vector +BP5Deserializer::GenerateReadRequests() +{ + std::vector Ret; + + for (size_t ReqIndex = 0; ReqIndex < PendingRequests.size(); ReqIndex++) + { + auto Req = &PendingRequests[ReqIndex]; + if (Req->RequestType == Local) { - size_t NodeFirst = 0; - if (NeedWriter(Req, WriterRank, NodeFirst)) + const size_t writerCohortSize = WriterCohortSize(Req->Step); + size_t NodeFirstBlock = 0; + for (size_t WriterRank = 0; WriterRank < writerCohortSize; + WriterRank++) { - /* if needed this writer fill destination with acquired data */ - int ElementSize = Req.VarRec->ElementSize; - MetaArrayRec *writer_meta_base = - (MetaArrayRec *)GetMetadataBase(Req.VarRec, Req.Step, - WriterRank); + MetaArrayRecOperator *writer_meta_base = + (MetaArrayRecOperator *)GetMetadataBase( + Req->VarRec, Req->Step, WriterRank); + if (!writer_meta_base) + { + continue; // Not writen on this step + } + size_t NodeLastBlock = + NodeFirstBlock + writer_meta_base->BlockCount - 1; + if ((NodeFirstBlock <= Req->BlockID) && + (NodeLastBlock >= Req->BlockID)) + { + // block is here + size_t NeededBlock = Req->BlockID - NodeFirstBlock; + size_t StartDim = NeededBlock * Req->VarRec->DimCount; + ReadRequest RR; + RR.Timestep = Req->Step; + RR.WriterRank = WriterRank; + RR.StartOffset = + writer_meta_base->DataLocation[NeededBlock]; + + RR.ReadLength = + helper::GetDataTypeSize(Req->VarRec->Type) * + CalcBlockLength(Req->VarRec->DimCount, + &writer_meta_base->Count[StartDim]); + RR.DestinationAddr = (char *)malloc(RR.ReadLength); + RR.Internal = NULL; + RR.OffsetInBlock = 0; + RR.ReqIndex = ReqIndex; + RR.BlockID = NeededBlock; + Ret.push_back(RR); + break; + } + NodeFirstBlock += writer_meta_base->BlockCount; + } + } + else + { + /* global case */ + const size_t writerCohortSize = WriterCohortSize(Req->Step); + for (size_t WriterRank = 0; WriterRank < writerCohortSize; + WriterRank++) + { + MetaArrayRecOperator *writer_meta_base = + (MetaArrayRecOperator *)GetMetadataBase( + Req->VarRec, Req->Step, WriterRank); if (!writer_meta_base) continue; // Not writen on this step - size_t *GlobalDimensions = writer_meta_base->Shape; - int DimCount = writer_meta_base->Dims; for (size_t Block = 0; Block < writer_meta_base->BlockCount; Block++) { - size_t *RankOffset = - &writer_meta_base - ->Offsets[Block * writer_meta_base->Dims]; - const size_t *RankSize = - &writer_meta_base - ->Count[Block * writer_meta_base->Dims]; - std::vector ZeroSel(DimCount); - std::vector ZeroRankOffset(DimCount); - std::vector ZeroGlobalDimensions(DimCount); - const size_t *SelOffset = NULL; - const size_t *SelSize = NULL; - int ReqIndex = 0; - while (Requests[ReqIndex].WriterRank != - static_cast(WriterRank) || - (Requests[ReqIndex].Timestep != Req.Step)) - ReqIndex++; - if (writer_meta_base->DataLocation == NULL) - { - // No Data from this writer - continue; - } - char *IncomingData = - (char *)Requests[ReqIndex].DestinationAddr + - writer_meta_base->DataLocation[Block]; - std::vector decompressBuffer; - if (Req.VarRec->Operator != NULL) + std::array intersectionstart; + std::array intersectioncount; + + size_t StartDim = Block * Req->VarRec->DimCount; + if (IntersectionStartCount( + Req->VarRec->DimCount, Req->Start.data(), + Req->Count.data(), + &writer_meta_base->Offsets[StartDim], + &writer_meta_base->Count[StartDim], + &intersectionstart[0], &intersectioncount[0])) { - size_t DestSize = Req.VarRec->ElementSize; - for (size_t dim = 0; dim < Req.VarRec->DimCount; dim++) + if (Req->VarRec->Operator != NULL) { - DestSize *= - writer_meta_base - ->Count[dim + - Block * writer_meta_base->Dims]; + // need the whole thing for decompression anyway + ReadRequest RR; + RR.Timestep = Req->Step; + RR.WriterRank = WriterRank; + RR.StartOffset = + writer_meta_base->DataLocation[Block]; + RR.ReadLength = + writer_meta_base->DataLengths[Block]; + RR.DestinationAddr = (char *)malloc(RR.ReadLength); + RR.Internal = NULL; + RR.ReqIndex = ReqIndex; + RR.BlockID = Block; + RR.OffsetInBlock = 0; + Ret.push_back(RR); } - decompressBuffer.resize(DestSize); - core::Decompress( - IncomingData, - ((MetaArrayRecOperator *)writer_meta_base) - ->DataLengths[Block], - decompressBuffer.data()); - IncomingData = decompressBuffer.data(); - } - if (Req.Start.size()) - { - SelOffset = Req.Start.data(); - } - if (Req.Count.size()) - { - SelSize = Req.Count.data(); - } - if (Req.RequestType == Local) - { - int LocalBlockID = Req.BlockID - NodeFirst; - IncomingData = - (char *)Requests[ReqIndex].DestinationAddr + - writer_meta_base->DataLocation[LocalBlockID]; - - RankOffset = ZeroRankOffset.data(); - GlobalDimensions = ZeroGlobalDimensions.data(); - if (SelSize == NULL) - { - SelSize = RankSize; - } - if (SelOffset == NULL) + else { - SelOffset = ZeroSel.data(); + for (size_t Dim = 0; Dim < Req->VarRec->DimCount; + Dim++) + { + intersectionstart[Dim] -= + writer_meta_base->Offsets[StartDim + Dim]; + } + size_t StartOffsetInBlock = + helper::GetDataTypeSize(Req->VarRec->Type) * + LinearIndex(Req->VarRec->DimCount, + &writer_meta_base->Count[StartDim], + &intersectionstart[0], + m_ReaderIsRowMajor); + for (size_t Dim = 0; Dim < Req->VarRec->DimCount; + Dim++) + { + intersectionstart[Dim] += + intersectioncount[Dim] - 1; + } + size_t EndOffsetInBlock = + helper::GetDataTypeSize(Req->VarRec->Type) * + (LinearIndex(Req->VarRec->DimCount, + &writer_meta_base->Count[StartDim], + &intersectionstart[0], + m_ReaderIsRowMajor) + + 1); + ReadRequest RR; + RR.Timestep = Req->Step; + RR.WriterRank = WriterRank; + RR.StartOffset = + writer_meta_base->DataLocation[Block] + + StartOffsetInBlock; + RR.ReadLength = + EndOffsetInBlock - StartOffsetInBlock; + RR.DestinationAddr = (char *)malloc(RR.ReadLength); + RR.Internal = NULL; + RR.OffsetInBlock = StartOffsetInBlock; + RR.ReqIndex = ReqIndex; + RR.BlockID = Block; + Ret.push_back(RR); } - for (int i = 0; i < DimCount; i++) - { - GlobalDimensions[i] = RankSize[i]; - } - } - - auto inStart = - adios2::Dims(RankOffset, RankOffset + DimCount); - auto inCount = adios2::Dims(RankSize, RankSize + DimCount); - auto outStart = - adios2::Dims(SelOffset, SelOffset + DimCount); - auto outCount = adios2::Dims(SelSize, SelSize + DimCount); - - if (!m_ReaderIsRowMajor) - { - std::reverse(inStart.begin(), inStart.end()); - std::reverse(inCount.begin(), inCount.end()); - std::reverse(outStart.begin(), outStart.end()); - std::reverse(outCount.begin(), outCount.end()); } - - helper::NdCopy(IncomingData, inStart, inCount, true, true, - (char *)Req.Data, outStart, outCount, true, - true, ElementSize, Dims(), Dims(), Dims(), - Dims(), false, Req.MemSpace); } } } } - for (const auto &Req : Requests) - { - free((char *)Req.DestinationAddr); + return Ret; +} + +void BP5Deserializer::FinalizeGets(std::vector Reads) +{ + for (const auto &Read : Reads) + { + auto Req = PendingRequests[Read.ReqIndex]; + int ElementSize = Req.VarRec->ElementSize; + MetaArrayRec *writer_meta_base = (MetaArrayRec *)GetMetadataBase( + Req.VarRec, Req.Step, Read.WriterRank); + + size_t *GlobalDimensions = writer_meta_base->Shape; + int DimCount = writer_meta_base->Dims; + std::vector ZeroSel(DimCount); + size_t *RankOffset = + &writer_meta_base->Offsets[DimCount * Read.BlockID]; + size_t *RankSize = &writer_meta_base->Count[DimCount * Read.BlockID]; + std::vector ZeroRankOffset(DimCount); + std::vector ZeroGlobalDimensions(DimCount); + const size_t *SelOffset = NULL; + const size_t *SelSize = NULL; + char *IncomingData = Read.DestinationAddr; + char *VirtualIncomingData = Read.DestinationAddr - Read.OffsetInBlock; + std::vector decompressBuffer; + if (Req.VarRec->Operator != NULL) + { + size_t DestSize = Req.VarRec->ElementSize; + for (size_t dim = 0; dim < Req.VarRec->DimCount; dim++) + { + DestSize *= + writer_meta_base + ->Count[dim + Read.BlockID * writer_meta_base->Dims]; + } + decompressBuffer.resize(DestSize); + core::Decompress(IncomingData, + ((MetaArrayRecOperator *)writer_meta_base) + ->DataLengths[Read.BlockID], + decompressBuffer.data()); + IncomingData = decompressBuffer.data(); + VirtualIncomingData = IncomingData; + } + if (Req.Start.size()) + { + SelOffset = Req.Start.data(); + } + if (Req.Count.size()) + { + SelSize = Req.Count.data(); + } + if (Req.RequestType == Local) + { + RankOffset = ZeroRankOffset.data(); + GlobalDimensions = ZeroGlobalDimensions.data(); + if (SelSize == NULL) + { + SelSize = RankSize; + } + if (SelOffset == NULL) + { + SelOffset = ZeroSel.data(); + } + for (int i = 0; i < DimCount; i++) + { + GlobalDimensions[i] = RankSize[i]; + } + } + + auto inStart = adios2::Dims(RankOffset, RankOffset + DimCount); + auto inCount = adios2::Dims(RankSize, RankSize + DimCount); + auto outStart = adios2::Dims(SelOffset, SelOffset + DimCount); + auto outCount = adios2::Dims(SelSize, SelSize + DimCount); + if (!m_ReaderIsRowMajor) + { + std::reverse(inStart.begin(), inStart.end()); + std::reverse(inCount.begin(), inCount.end()); + std::reverse(outStart.begin(), outStart.end()); + std::reverse(outCount.begin(), outCount.end()); + } + + helper::NdCopy(VirtualIncomingData, inStart, inCount, true, true, + (char *)Req.Data, outStart, outCount, true, true, + ElementSize, Dims(), Dims(), Dims(), Dims(), false, + Req.MemSpace); + free((char *)Read.DestinationAddr); } PendingRequests.clear(); } diff --git a/source/adios2/toolkit/format/bp5/BP5Deserializer.h b/source/adios2/toolkit/format/bp5/BP5Deserializer.h index 4dbb406568..5cd3e0f856 100644 --- a/source/adios2/toolkit/format/bp5/BP5Deserializer.h +++ b/source/adios2/toolkit/format/bp5/BP5Deserializer.h @@ -46,6 +46,9 @@ class BP5Deserializer : virtual public BP5Base size_t ReadLength; char *DestinationAddr; void *Internal; + size_t ReqIndex; + size_t OffsetInBlock; + size_t BlockID; }; void InstallMetaMetaData(MetaMetaInfoBlock &MMList); void InstallMetaData(void *MetadataBlock, size_t BlockLen, @@ -201,7 +204,6 @@ class BP5Deserializer : virtual public BP5Base void *Data; }; std::vector PendingRequests; - bool NeedWriter(BP5ArrayRequest Req, size_t i, size_t &NodeFirst); void *GetMetadataBase(BP5VarRec *VarRec, size_t Step, size_t WriterRank) const; size_t CurTimestep = 0;