BP5Writer: singularize m_DataSubstream and substream names

eisenhauer · eisenhauer · commit f99b52e4a05c · 2026-05-27T12:50:04.000-04:00
AggTransportData's m_DataSubstreams (vector), m_SubStreamNames (vector),
and m_DrainSubStreamNames (vector) become singular shared_ptr&lt;Transport&gt;
and string members. The vectors were size-1 by construction
(GetBPSubStreamNames is called with a single base name, and the user
multi-transport feature is rejected at Init).

Drops the for-each loops over m_DataSubstreams in the write path and
across the async writer, where the size-1 invariant made the loop
vestigial. AsyncWriteInfo carries a shared_ptr&lt;Transport&gt;* instead of a
vector*.

CloseFiles and FlushProfiler iterations now null-check m_DataSubstream:
an AggTransportData entry may exist with no open transport when the
rank hasn't written through that aggregator.
diff --git a/source/adios2/engine/bp5/BP5Writer.cpp b/source/adios2/engine/bp5/BP5Writer.cpp
@@ -483,11 +483,8 @@ void BP5Writer::WriteData(format::BufferV *Data)
                                                      "is not supported in BP5");
         }
         AggTransportData &aggData = m_AggregatorSpecifics.at(GetCacheKey(m_Aggregator));
-        for (auto &t : aggData.m_DataSubstreams)
-        {
-            t->Flush();
-            t->FinalizeSegment();
-        }
+        aggData.m_DataSubstream->Flush();
+        aggData.m_DataSubstream->FinalizeSegment();
         delete Data;
     }
 }
@@ -534,10 +531,8 @@ void BP5Writer::WriteData_EveryoneWrites(format::BufferV *Data, bool SerializedW
     m_DataPos += Data->Size();
     std::vector<core::iovec> DataVec = Data->DataVec();
     AggTransportData &aggData = m_AggregatorSpecifics.at(GetCacheKey(m_Aggregator));
-    for (auto &t : aggData.m_DataSubstreams)
-    {
-        t->WriteV(DataVec.data(), static_cast<int>(DataVec.size()), m_StartDataPos);
-    }
+    aggData.m_DataSubstream->WriteV(DataVec.data(), static_cast<int>(DataVec.size()),
+                                    m_StartDataPos);
 
     if (SerializedWriters && a->m_Comm.Rank() < a->m_Comm.Size() - 1)
     {
@@ -1917,8 +1912,8 @@ void BP5Writer::InitTransports()
         {
             const std::vector<std::string> drainTransportNames =
                 transportman::TransportMan::GetFilesBaseNames(m_Name, m_DataTransportsParameters);
-            aggData.m_DrainSubStreamNames =
-                GetBPSubStreamNames(drainTransportNames, m_Aggregator->m_SubStreamIndex);
+            aggData.m_DrainSubStreamName =
+                GetBPSubStreamName(drainTransportNames[0], m_Aggregator->m_SubStreamIndex);
             /* start up BB thread */
             //            m_FileDrainer.SetVerbose(
             //				     m_Parameters.BurstBufferVerbose,
@@ -1931,18 +1926,15 @@ void BP5Writer::InitTransports()
     if (m_DrainBB)
     {
         /* Create the directories on target anyway by main thread */
-        transport::MkDirsBarrier(m_Comm, aggData.m_DrainSubStreamNames, m_DataTransportsParameters,
+        transport::MkDirsBarrier(m_Comm, {aggData.m_DrainSubStreamName}, m_DataTransportsParameters,
                                  m_Parameters.NodeLocal);
     }
 
     if (m_IAmDraining)
     {
         if (m_DrainBB)
         {
-            for (const auto &name : aggData.m_DrainSubStreamNames)
-            {
-                m_FileDrainer.AddOperationOpen(name, m_OpenMode);
-            }
+            m_FileDrainer.AddOperationOpen(aggData.m_DrainSubStreamName, m_OpenMode);
         }
     }
 
@@ -1978,7 +1970,7 @@ void BP5Writer::OpenSubfile(bool useComm, bool forceAppend)
     AggTransportData &aggData = m_AggregatorSpecifics.at(cacheKey);
 
     // /path/name.bp.dir/name.bp.rank
-    aggData.m_SubStreamNames = GetBPSubStreamNames({m_Name}, m_Aggregator->m_SubStreamIndex);
+    aggData.m_SubStreamName = GetBPSubStreamName(m_Name, m_Aggregator->m_SubStreamIndex);
 
     helper::Comm openSyncComm;
 
@@ -2016,12 +2008,9 @@ void BP5Writer::OpenSubfile(bool useComm, bool forceAppend)
                     std::cout << "Rank " << m_Comm.Rank() << " opening data file with Comm"
                               << std::endl;
                 }
-                for (const auto &name : aggData.m_SubStreamNames)
-                {
-                    aggData.m_DataSubstreams.push_back(transport::OpenFileChained(
-                        m_Comm, name, mode, m_IO.m_TransportsParameters[0], true,
-                        *DataWritingComm));
-                }
+                aggData.m_DataSubstream = transport::OpenFileChained(
+                    m_Comm, aggData.m_SubStreamName, mode, m_IO.m_TransportsParameters[0], true,
+                    *DataWritingComm);
             }
             else
             {
@@ -2030,11 +2019,8 @@ void BP5Writer::OpenSubfile(bool useComm, bool forceAppend)
                     std::cout << "Rank " << m_Comm.Rank() << " opening data file no Comm"
                               << std::endl;
                 }
-                for (const auto &name : aggData.m_SubStreamNames)
-                {
-                    aggData.m_DataSubstreams.push_back(transport::OpenFile(
-                        m_Comm, name, mode, m_IO.m_TransportsParameters[0], true));
-                }
+                aggData.m_DataSubstream = transport::OpenFile(m_Comm, aggData.m_SubStreamName, mode,
+                                                              m_IO.m_TransportsParameters[0], true);
             }
         }
     }
@@ -2211,19 +2197,16 @@ void BP5Writer::InitBPBuffer()
             const size_t off = m_AppendDataPos[m_Aggregator->m_SubStreamIndex];
             if (off < MaxSizeT)
             {
-                for (auto &t : aggData.m_DataSubstreams)
-                {
-                    t->Truncate(off);
-                    // Seek is needed since truncate does not seek.
-                    // Seek instead of SeekToEnd in case a transport
-                    // does not support actual truncate.
-                    t->Seek(off);
-                }
+                aggData.m_DataSubstream->Truncate(off);
+                // Seek is needed since truncate does not seek.
+                // Seek instead of SeekToEnd in case a transport
+                // does not support actual truncate.
+                aggData.m_DataSubstream->Seek(off);
                 m_DataPos = off;
             }
             else
             {
-                m_DataPos = aggData.m_DataSubstreams[0]->GetSize();
+                m_DataPos = aggData.m_DataSubstream->GetSize();
             }
         }
 
@@ -2287,10 +2270,7 @@ void BP5Writer::InitBPBuffer()
         // data existed but index was missing
         if (m_Aggregator->m_IsAggregator)
         {
-            for (auto &t : aggData.m_DataSubstreams)
-            {
-                t->Seek(0);
-            }
+            aggData.m_DataSubstream->Seek(0);
         }
     }
 
@@ -2446,9 +2426,9 @@ void BP5Writer::DoClose(const int transportIndex)
         (void)transportIndex;
         for (auto &aggPair : m_AggregatorSpecifics)
         {
-            for (auto &t : aggPair.second.m_DataSubstreams)
+            if (aggPair.second.m_DataSubstream)
             {
-                t->Close();
+                aggPair.second.m_DataSubstream->Close();
             }
         }
 
@@ -2496,7 +2476,7 @@ void BP5Writer::DoClose(const int transportIndex)
     // cleaned up while those resources are still available.
     for (auto &aggPair : m_AggregatorSpecifics)
     {
-        aggPair.second.m_DataSubstreams.clear();
+        aggPair.second.m_DataSubstream.reset();
     }
     m_MetadataFile.reset();
     m_MetaMetadataFile.reset();
@@ -2511,8 +2491,9 @@ void BP5Writer::FlushProfiler()
 
     for (auto &specs : m_AggregatorSpecifics)
     {
-        for (auto &t : specs.second.m_DataSubstreams)
+        if (specs.second.m_DataSubstream)
         {
+            auto &t = specs.second.m_DataSubstream;
             transportTypes.push_back(t->m_Type);
             transportNames.push_back(t->m_Name);
             transportProfilers.push_back(&t->m_Profiler);
diff --git a/source/adios2/engine/bp5/BP5Writer.h b/source/adios2/engine/bp5/BP5Writer.h
@@ -56,16 +56,17 @@ class BP5Writer : public BP5Engine, public core::Engine
 private:
     struct AggTransportData
     {
-        /** Open data-file transports for this aggregator's substreams. One
-         *  entry per name in m_SubStreamNames. */
-        std::vector<std::shared_ptr<Transport>> m_DataSubstreams;
+        /** Open data-file transport for this aggregator's substream. Null
+         *  if not yet opened (no write done by this rank in this aggregator
+         *  context). */
+        std::shared_ptr<Transport> m_DataSubstream;
 
-        /** Name of subfiles to directly write to. Either original target or
-         *  burst buffer if used. */
-        std::vector<std::string> m_SubStreamNames;
+        /** Name of the subfile to directly write to. Either original target
+         *  or burst buffer if used. */
+        std::string m_SubStreamName;
 
-        /** Name of subfiles on target if burst buffer is used. */
-        std::vector<std::string> m_DrainSubStreamNames;
+        /** Name of the subfile on target if burst buffer is used. */
+        std::string m_DrainSubStreamName;
     };
 
     std::string GetCacheKey(aggregator::MPIAggregator *aggregator);
@@ -359,7 +360,7 @@ class BP5Writer : public BP5Engine, public core::Engine
         int nproc_chain;
         TimePoint tstart;
         adios2::shm::TokenChain<uint64_t> *tokenChain;
-        std::vector<std::shared_ptr<Transport>> *DataSubstreams;
+        std::shared_ptr<Transport> *DataSubstream;
         adios2::format::BufferV *Data;
         uint64_t startPos;
         uint64_t totalSize;
diff --git a/source/adios2/engine/bp5/BP5Writer_EveryoneWrites_Async.cpp b/source/adios2/engine/bp5/BP5Writer_EveryoneWrites_Async.cpp
@@ -141,10 +141,7 @@ void BP5Writer::AsyncWriteOwnData(AsyncWriteInfo *info, std::vector<core::iovec>
                       << " write the rest of  " << totalsize - wrote
                       << " bytes at pos " << pos << std::endl;*/
 
-            for (auto &t : *info->DataSubstreams)
-            {
-                t->WriteV(vec.data(), static_cast<int>(vec.size()), pos);
-            }
+            (*info->DataSubstream)->WriteV(vec.data(), static_cast<int>(vec.size()), pos);
 
             break; /* Exit loop after this final write */
         }
@@ -167,18 +164,12 @@ void BP5Writer::AsyncWriteOwnData(AsyncWriteInfo *info, std::vector<core::iovec>
         const char *buf = (const char *)DataVec[block].iov_base + temp_offset;
         if (firstWrite)
         {
-            for (auto &t : *info->DataSubstreams)
-            {
-                t->Write(buf, n, info->startPos);
-            }
+            (*info->DataSubstream)->Write(buf, n, info->startPos);
             firstWrite = false;
         }
         else
         {
-            for (auto &t : *info->DataSubstreams)
-            {
-                t->Write(buf, n);
-            }
+            (*info->DataSubstream)->Write(buf, n);
         }
 
         /* Have we processed the entire block or staying with it? */
@@ -218,10 +209,7 @@ int BP5Writer::AsyncWriteThread_EveryoneWrites(AsyncWriteInfo *info)
             info->tokenChain->RecvToken();
         }
     }
-    for (auto &t : *info->DataSubstreams)
-    {
-        t->FinalizeSegment();
-    }
+    (*info->DataSubstream)->FinalizeSegment();
     delete info->Data;
     return 1;
 };
@@ -286,7 +274,7 @@ void BP5Writer::WriteData_EveryoneWrites_Async(format::BufferV *Data, bool Seria
     }
     m_AsyncWriteInfo->tstart = m_EngineStart;
     AggTransportData *aggData = &(m_AggregatorSpecifics.at(GetCacheKey(m_Aggregator)));
-    m_AsyncWriteInfo->DataSubstreams = &aggData->m_DataSubstreams;
+    m_AsyncWriteInfo->DataSubstream = &aggData->m_DataSubstream;
     m_AsyncWriteInfo->Data = Data;
     m_AsyncWriteInfo->startPos = m_StartDataPos;
     m_AsyncWriteInfo->totalSize = Data->Size();
diff --git a/source/adios2/engine/bp5/BP5Writer_TwoLevelShm.cpp b/source/adios2/engine/bp5/BP5Writer_TwoLevelShm.cpp
@@ -176,10 +176,8 @@ void BP5Writer::WriteMyOwnData(format::BufferV *Data)
     std::vector<core::iovec> DataVec = Data->DataVec();
     m_StartDataPos = m_DataPos;
     AggTransportData *aggData = &(m_AggregatorSpecifics.at(GetCacheKey(m_Aggregator)));
-    for (auto &t : aggData->m_DataSubstreams)
-    {
-        t->WriteV(DataVec.data(), static_cast<int>(DataVec.size()), m_StartDataPos);
-    }
+    aggData->m_DataSubstream->WriteV(DataVec.data(), static_cast<int>(DataVec.size()),
+                                     m_StartDataPos);
     m_DataPos += Data->Size();
 }
 
@@ -292,10 +290,7 @@ void BP5Writer::WriteOthersData(size_t TotalSize)
         aggregator::MPIShmChain::ShmDataBuffer *b = a->LockConsumerBuffer();
 
         // b->actual_size: how much we need to write
-        for (auto &t : aggData->m_DataSubstreams)
-        {
-            t->Write(b->buf, b->actual_size);
-        }
+        aggData->m_DataSubstream->Write(b->buf, b->actual_size);
 
         wrote += b->actual_size;
 
@@ -327,10 +322,7 @@ void BP5Writer::WriteOthersData_DrainOnError(const size_t TotalSize, std::except
         {
             try
             {
-                for (auto &t : aggData->m_DataSubstreams)
-                {
-                    t->Write(b->buf, b->actual_size);
-                }
+                aggData->m_DataSubstream->Write(b->buf, b->actual_size);
             }
             catch (...)
             {
diff --git a/source/adios2/engine/bp5/BP5Writer_TwoLevelShm_Async.cpp b/source/adios2/engine/bp5/BP5Writer_TwoLevelShm_Async.cpp
@@ -41,10 +41,7 @@ void BP5Writer::AsyncWriteThread_TwoLevelShm_Aggregator(AsyncWriteInfo *info)
     {
         std::vector<core::iovec> DataVec = info->Data->DataVec();
         const uint64_t mysize = info->Data->Size();
-        for (auto &t : *info->DataSubstreams)
-        {
-            t->Seek(info->startPos);
-        }
+        (*info->DataSubstream)->Seek(info->startPos);
         AsyncWriteOwnData(info, DataVec, mysize, false);
         totalSize -= mysize;
     }
@@ -148,10 +145,7 @@ int BP5Writer::AsyncWriteThread_TwoLevelShm(AsyncWriteInfo *info)
         info->tokenChain->SendToken(nextWriterPos);
         AsyncWriteThread_TwoLevelShm_Aggregator(info);
         info->tokenChain->RecvToken();
-        for (auto &t : *info->DataSubstreams)
-        {
-            t->FinalizeSegment();
-        }
+        (*info->DataSubstream)->FinalizeSegment();
     }
     else
     {
@@ -263,7 +257,7 @@ void BP5Writer::WriteData_TwoLevelShm_Async(format::BufferV *Data)
     m_AsyncWriteInfo->tstart = m_EngineStart;
     m_AsyncWriteInfo->tokenChain = new shm::TokenChain<uint64_t>(&a->m_Comm);
     AggTransportData *aggData = &(m_AggregatorSpecifics.at(GetCacheKey(m_Aggregator)));
-    m_AsyncWriteInfo->DataSubstreams = &aggData->m_DataSubstreams;
+    m_AsyncWriteInfo->DataSubstream = &aggData->m_DataSubstream;
     m_AsyncWriteInfo->Data = Data;
     m_AsyncWriteInfo->flagRush = &m_flagRush;
     m_AsyncWriteInfo->lock = &m_AsyncWriteLock;
diff --git a/source/adios2/engine/bp5/BP5Writer_WithRerouting.cpp b/source/adios2/engine/bp5/BP5Writer_WithRerouting.cpp
@@ -1003,10 +1003,8 @@ void BP5Writer::WriteData_WithRerouting(format::BufferV *Data)
     std::vector<core::iovec> DataVec = Data->DataVec();
 
     AggTransportData &aggData = m_AggregatorSpecifics.at(GetCacheKey(m_Aggregator));
-    for (auto &t : aggData.m_DataSubstreams)
-    {
-        t->WriteV(DataVec.data(), static_cast<int>(DataVec.size()), m_StartDataPos);
-    }
+    aggData.m_DataSubstream->WriteV(DataVec.data(), static_cast<int>(DataVec.size()),
+                                    m_StartDataPos);
 
     m_DataPos += Data->Size();
 

Original file line number	Diff line number	Diff line change
`@@ -483,11 +483,8 @@ void BP5Writer::WriteData(format::BufferV *Data)`
`483`	`483`	`"is not supported in BP5");`
`484`	`484`	`}`
`485`	`485`	`AggTransportData &aggData = m_AggregatorSpecifics.at(GetCacheKey(m_Aggregator));`
`486`		`- for (auto &t : aggData.m_DataSubstreams)`
`487`		`- {`
`488`		`- t->Flush();`
`489`		`- t->FinalizeSegment();`
`490`		`- }`
	`486`	`+ aggData.m_DataSubstream->Flush();`
	`487`	`+ aggData.m_DataSubstream->FinalizeSegment();`
`491`	`488`	`delete Data;`
`492`	`489`	`}`
`493`	`490`	`}`
`@@ -534,10 +531,8 @@ void BP5Writer::WriteData_EveryoneWrites(format::BufferV *Data, bool SerializedW`
`534`	`531`	`m_DataPos += Data->Size();`
`535`	`532`	`std::vector<core::iovec> DataVec = Data->DataVec();`
`536`	`533`	`AggTransportData &aggData = m_AggregatorSpecifics.at(GetCacheKey(m_Aggregator));`
`537`		`- for (auto &t : aggData.m_DataSubstreams)`
`538`		`- {`
`539`		`- t->WriteV(DataVec.data(), static_cast<int>(DataVec.size()), m_StartDataPos);`
`540`		`- }`
	`534`	`+ aggData.m_DataSubstream->WriteV(DataVec.data(), static_cast<int>(DataVec.size()),`
	`535`	`+ m_StartDataPos);`
`541`	`536`
`542`	`537`	`if (SerializedWriters && a->m_Comm.Rank() < a->m_Comm.Size() - 1)`
`543`	`538`	`{`
`@@ -1917,8 +1912,8 @@ void BP5Writer::InitTransports()`
`1917`	`1912`	`{`
`1918`	`1913`	`const std::vector<std::string> drainTransportNames =`
`1919`	`1914`	`transportman::TransportMan::GetFilesBaseNames(m_Name, m_DataTransportsParameters);`
`1920`		`- aggData.m_DrainSubStreamNames =`
`1921`		`- GetBPSubStreamNames(drainTransportNames, m_Aggregator->m_SubStreamIndex);`
	`1915`	`+ aggData.m_DrainSubStreamName =`
	`1916`	`+ GetBPSubStreamName(drainTransportNames[0], m_Aggregator->m_SubStreamIndex);`
`1922`	`1917`	`/* start up BB thread */`
`1923`	`1918`	`// m_FileDrainer.SetVerbose(`
`1924`	`1919`	`// m_Parameters.BurstBufferVerbose,`
`@@ -1931,18 +1926,15 @@ void BP5Writer::InitTransports()`
`1931`	`1926`	`if (m_DrainBB)`
`1932`	`1927`	`{`
`1933`	`1928`	`/* Create the directories on target anyway by main thread */`
`1934`		`- transport::MkDirsBarrier(m_Comm, aggData.m_DrainSubStreamNames, m_DataTransportsParameters,`
	`1929`	`+ transport::MkDirsBarrier(m_Comm, {aggData.m_DrainSubStreamName}, m_DataTransportsParameters,`
`1935`	`1930`	`m_Parameters.NodeLocal);`
`1936`	`1931`	`}`
`1937`	`1932`
`1938`	`1933`	`if (m_IAmDraining)`
`1939`	`1934`	`{`
`1940`	`1935`	`if (m_DrainBB)`
`1941`	`1936`	`{`
`1942`		`- for (const auto &name : aggData.m_DrainSubStreamNames)`
`1943`		`- {`
`1944`		`- m_FileDrainer.AddOperationOpen(name, m_OpenMode);`
`1945`		`- }`
	`1937`	`+ m_FileDrainer.AddOperationOpen(aggData.m_DrainSubStreamName, m_OpenMode);`
`1946`	`1938`	`}`
`1947`	`1939`	`}`
`1948`	`1940`
`@@ -1978,7 +1970,7 @@ void BP5Writer::OpenSubfile(bool useComm, bool forceAppend)`
`1978`	`1970`	`AggTransportData &aggData = m_AggregatorSpecifics.at(cacheKey);`
`1979`	`1971`
`1980`	`1972`	`// /path/name.bp.dir/name.bp.rank`
`1981`		`- aggData.m_SubStreamNames = GetBPSubStreamNames({m_Name}, m_Aggregator->m_SubStreamIndex);`
	`1973`	`+ aggData.m_SubStreamName = GetBPSubStreamName(m_Name, m_Aggregator->m_SubStreamIndex);`
`1982`	`1974`
`1983`	`1975`	`helper::Comm openSyncComm;`
`1984`	`1976`
`@@ -2016,12 +2008,9 @@ void BP5Writer::OpenSubfile(bool useComm, bool forceAppend)`
`2016`	`2008`	`std::cout << "Rank " << m_Comm.Rank() << " opening data file with Comm"`
`2017`	`2009`	`<< std::endl;`
`2018`	`2010`	`}`
`2019`		`- for (const auto &name : aggData.m_SubStreamNames)`
`2020`		`- {`
`2021`		`- aggData.m_DataSubstreams.push_back(transport::OpenFileChained(`
`2022`		`- m_Comm, name, mode, m_IO.m_TransportsParameters[0], true,`
`2023`		`- *DataWritingComm));`
`2024`		`- }`
	`2011`	`+ aggData.m_DataSubstream = transport::OpenFileChained(`
	`2012`	`+ m_Comm, aggData.m_SubStreamName, mode, m_IO.m_TransportsParameters[0], true,`
	`2013`	`+ *DataWritingComm);`
`2025`	`2014`	`}`
`2026`	`2015`	`else`
`2027`	`2016`	`{`
`@@ -2030,11 +2019,8 @@ void BP5Writer::OpenSubfile(bool useComm, bool forceAppend)`
`2030`	`2019`	`std::cout << "Rank " << m_Comm.Rank() << " opening data file no Comm"`
`2031`	`2020`	`<< std::endl;`
`2032`	`2021`	`}`
`2033`		`- for (const auto &name : aggData.m_SubStreamNames)`
`2034`		`- {`
`2035`		`- aggData.m_DataSubstreams.push_back(transport::OpenFile(`
`2036`		`- m_Comm, name, mode, m_IO.m_TransportsParameters[0], true));`
`2037`		`- }`
	`2022`	`+ aggData.m_DataSubstream = transport::OpenFile(m_Comm, aggData.m_SubStreamName, mode,`
	`2023`	`+ m_IO.m_TransportsParameters[0], true);`
`2038`	`2024`	`}`
`2039`	`2025`	`}`
`2040`	`2026`	`}`
`@@ -2211,19 +2197,16 @@ void BP5Writer::InitBPBuffer()`
`2211`	`2197`	`const size_t off = m_AppendDataPos[m_Aggregator->m_SubStreamIndex];`
`2212`	`2198`	`if (off < MaxSizeT)`
`2213`	`2199`	`{`
`2214`		`- for (auto &t : aggData.m_DataSubstreams)`
`2215`		`- {`
`2216`		`- t->Truncate(off);`
`2217`		`- // Seek is needed since truncate does not seek.`
`2218`		`- // Seek instead of SeekToEnd in case a transport`
`2219`		`- // does not support actual truncate.`
`2220`		`- t->Seek(off);`
`2221`		`- }`
	`2200`	`+ aggData.m_DataSubstream->Truncate(off);`
	`2201`	`+ // Seek is needed since truncate does not seek.`
	`2202`	`+ // Seek instead of SeekToEnd in case a transport`
	`2203`	`+ // does not support actual truncate.`
	`2204`	`+ aggData.m_DataSubstream->Seek(off);`
`2222`	`2205`	`m_DataPos = off;`
`2223`	`2206`	`}`
`2224`	`2207`	`else`
`2225`	`2208`	`{`
`2226`		`- m_DataPos = aggData.m_DataSubstreams[0]->GetSize();`
	`2209`	`+ m_DataPos = aggData.m_DataSubstream->GetSize();`
`2227`	`2210`	`}`
`2228`	`2211`	`}`
`2229`	`2212`
`@@ -2287,10 +2270,7 @@ void BP5Writer::InitBPBuffer()`
`2287`	`2270`	`// data existed but index was missing`
`2288`	`2271`	`if (m_Aggregator->m_IsAggregator)`
`2289`	`2272`	`{`
`2290`		`- for (auto &t : aggData.m_DataSubstreams)`
`2291`		`- {`
`2292`		`- t->Seek(0);`
`2293`		`- }`
	`2273`	`+ aggData.m_DataSubstream->Seek(0);`
`2294`	`2274`	`}`
`2295`	`2275`	`}`
`2296`	`2276`
`@@ -2446,9 +2426,9 @@ void BP5Writer::DoClose(const int transportIndex)`
`2446`	`2426`	`(void)transportIndex;`
`2447`	`2427`	`for (auto &aggPair : m_AggregatorSpecifics)`
`2448`	`2428`	`{`
`2449`		`- for (auto &t : aggPair.second.m_DataSubstreams)`
	`2429`	`+ if (aggPair.second.m_DataSubstream)`
`2450`	`2430`	`{`
`2451`		`- t->Close();`
	`2431`	`+ aggPair.second.m_DataSubstream->Close();`
`2452`	`2432`	`}`
`2453`	`2433`	`}`
`2454`	`2434`
`@@ -2496,7 +2476,7 @@ void BP5Writer::DoClose(const int transportIndex)`
`2496`	`2476`	`// cleaned up while those resources are still available.`
`2497`	`2477`	`for (auto &aggPair : m_AggregatorSpecifics)`
`2498`	`2478`	`{`
`2499`		`- aggPair.second.m_DataSubstreams.clear();`
	`2479`	`+ aggPair.second.m_DataSubstream.reset();`
`2500`	`2480`	`}`
`2501`	`2481`	`m_MetadataFile.reset();`
`2502`	`2482`	`m_MetaMetadataFile.reset();`
`@@ -2511,8 +2491,9 @@ void BP5Writer::FlushProfiler()`
`2511`	`2491`
`2512`	`2492`	`for (auto &specs : m_AggregatorSpecifics)`
`2513`	`2493`	`{`
`2514`		`- for (auto &t : specs.second.m_DataSubstreams)`
	`2494`	`+ if (specs.second.m_DataSubstream)`
`2515`	`2495`	`{`
	`2496`	`+ auto &t = specs.second.m_DataSubstream;`
`2516`	`2497`	`transportTypes.push_back(t->m_Type);`
`2517`	`2498`	`transportNames.push_back(t->m_Name);`
`2518`	`2499`	`transportProfilers.push_back(&t->m_Profiler);`