Skip to content

Refactor to drop TransportMan where we can (not bp3/4) #5048

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
eisenhauer merged 14 commits into from
May 28, 2026
Merged

Refactor to drop TransportMan where we can (not bp3/4) #5048

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
c5c0b74
transport: extract OpenFile free function from TransportMan
eisenhauer May 26, 2026
f22d09b
campaign: use transport::OpenFile directly for remote file reads
eisenhauer May 26, 2026
526d23e
filepool: take a TransportOpener lambda instead of TransportMan*
eisenhauer May 26, 2026
08f2043
BP5Reader: drop unused m_TransportFactory
eisenhauer May 26, 2026
4aa66ba
transport: lift MkDirsBarrier; BP5Writer drops m_TransportFactory
eisenhauer May 26, 2026
e21321b
BP5: reject multi-transport-per-IO configs at Init()
eisenhauer May 26, 2026
d1adb15
BP5Writer: replace m_FileDataManager with explicit substream vector
eisenhauer May 26, 2026
34887c5
TransportMan: mark as legacy in header comment
eisenhauer May 26, 2026
b61d471
clang-format
eisenhauer May 26, 2026
08e6e34
DAOS: drop TransportMan, use shared_ptr<Transport>
eisenhauer May 26, 2026
47a55cc
clang-format
eisenhauer May 26, 2026
efb56c3
TestFilePool: switch to TransportOpener lambda
eisenhauer May 26, 2026
f99b52e
BP5Writer: singularize m_DataSubstream and substream names
eisenhauer May 27, 2026
dc1ed43
BP5Writer: null-guard Flush+FinalizeSegment for non-aggregator ranks
eisenhauer May 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions source/adios2/CMakeLists.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -123,6 +123,7 @@ add_library(adios2_core
toolkit/remote/Remote.cpp

toolkit/transport/Transport.cpp
toolkit/transport/OpenFile.cpp
toolkit/transport/file/FileStdio.cpp
toolkit/transport/file/FileFStream.cpp
toolkit/transport/null/NullTransport.cpp
Expand Down
29 changes: 21 additions & 8 deletions source/adios2/engine/bp5/BP5Reader.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
#include "adios2/toolkit/remote/EVPathRemote.h"
#include "adios2/toolkit/remote/XrootdHttpRemote.h"
#include "adios2/toolkit/remote/XrootdRemote.h"
#include "adios2/toolkit/transport/OpenFile.h"
#include "adios2/toolkit/transport/file/FileFStream.h"
#include "adios2sys/SystemTools.hxx"
#include <adios2-perfstubs-interface.h>
Expand All @@ -39,8 +40,7 @@ namespace engine
{

BP5Reader::BP5Reader(IO &io, const std::string &name, const Mode mode, helper::Comm comm)
: Engine("BP5Reader", io, name, mode, std::move(comm)), m_TransportFactory(io, m_Comm), m_Remote(),
m_JSONProfiler(m_Comm)
: Engine("BP5Reader", io, name, mode, std::move(comm)), m_Remote(), m_JSONProfiler(m_Comm)
{
PERFSTUBS_SCOPED_TIMER("BP5Reader::Open");
Init();
Expand All @@ -49,8 +49,7 @@ BP5Reader::BP5Reader(IO &io, const std::string &name, const Mode mode, helper::C

BP5Reader::BP5Reader(IO &io, const std::string &name, const Mode mode, helper::Comm comm,
const char *md, const size_t mdsize)
: Engine("BP5Reader", io, name, mode, std::move(comm)), m_TransportFactory(io, m_Comm), m_Remote(),
m_JSONProfiler(m_Comm)
: Engine("BP5Reader", io, name, mode, std::move(comm)), m_Remote(), m_JSONProfiler(m_Comm)
{
PERFSTUBS_SCOPED_TIMER("BP5Reader::Open");
m_ReadMetadataFromFile = false;
Expand Down Expand Up @@ -1147,6 +1146,15 @@ void BP5Reader::Init()
m_Name);
}

if (m_IO.m_TransportsParameters.size() > 1)
{
helper::Throw<std::invalid_argument>(
"Engine", "BP5Reader", "Init",
"BP5 engine accepts only one transport per IO; got " +
std::to_string(m_IO.m_TransportsParameters.size()) +
". Multiple parallel transports per IO are no longer supported.");
}

// if IO was involved in reading before this flag may be true now
m_IO.m_ReadStreaming = false;
m_ReaderIsRowMajor = (m_IO.m_ArrayOrder == ArrayOrdering::RowMajor);
Expand Down Expand Up @@ -1536,9 +1544,11 @@ void BP5Reader::InitTransports()
}

// Metadata files always use default local transport
m_MetadataFiles =
std::make_shared<FilePool>(&m_TransportFactory, m_IO.m_TransportsParameters[0],
m_Parameters.MaxOpenFilesAtOnce, &m_TarInfoMap);
auto opener = [this](const std::string &name, const Params &params) {
return transport::OpenFile(m_Comm, name, Mode::Read, params, false);
};
m_MetadataFiles = std::make_shared<FilePool>(opener, m_IO.m_TransportsParameters[0],
m_Parameters.MaxOpenFilesAtOnce, &m_TarInfoMap);

// Auto-detect S3 hybrid storage via s3.json sidecar.
// Only rank 0 reads the file to avoid filesystem metadata storms at scale.
Expand Down Expand Up @@ -1658,7 +1668,10 @@ void BP5Reader::InitTransports()
dataTransportParams["verbose"] = std::to_string(m_Parameters.verbose);
}

m_DataFiles = std::make_shared<FilePool>(&m_TransportFactory, dataTransportParams,
auto dataOpener = [this](const std::string &name, const Params &params) {
return transport::OpenFile(m_Comm, name, Mode::Read, params, false);
};
m_DataFiles = std::make_shared<FilePool>(dataOpener, dataTransportParams,
m_Parameters.MaxOpenFilesAtOnce, &m_TarInfoMap);
}
else
Expand Down
7 changes: 2 additions & 5 deletions source/adios2/engine/bp5/BP5Reader.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,6 @@
#include "adios2/toolkit/format/buffer/heap/BufferMalloc.h"
#include "adios2/toolkit/kvcache/KVCacheCommon.h"
#include "adios2/toolkit/remote/Remote.h"
#include "adios2/toolkit/transportman/TransportMan.h"

#include <chrono>
#include <map>
Expand Down Expand Up @@ -83,13 +82,11 @@ class BP5Reader : public BP5Engine, public Engine
/* m_MDFileAbsolutePos <= m_MDFileProcessedSize <= m_MDFileAlreadyReadSize
*/

/* transport manager for managing data file(s) - may use S3 if DataFileTransport is set */
/* pool of open transports for data file(s) - may use S3 if DataFileTransport is set */
std::shared_ptr<FilePool> m_DataFiles;
/* transport manager for managing metadata file(s) - always local */
/* pool of open transports for metadata file(s) - always local */
std::shared_ptr<FilePool> m_MetadataFiles;

transportman::TransportMan m_TransportFactory;

std::unique_ptr<PoolableFile> m_MDIndexFile;
std::unique_ptr<PoolableFile> m_MDFile;
std::unique_ptr<PoolableFile> m_MetaMetadataFile;
Expand Down
109 changes: 63 additions & 46 deletions source/adios2/engine/bp5/BP5Writer.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
#include "adios2/helper/adiosSystem.h" // CleanupBPDirectory
#include "adios2/toolkit/format/buffer/chunk/ChunkV.h"
#include "adios2/toolkit/format/buffer/malloc/MallocV.h"
#include "adios2/toolkit/transport/OpenFile.h"
#include "adios2/toolkit/transport/file/FileFStream.h"
#include <adios2-perfstubs-interface.h>
#include <adios2sys/SystemTools.hxx>
Expand All @@ -37,8 +38,8 @@ namespace engine
using namespace adios2::format;

BP5Writer::BP5Writer(IO &io, const std::string &name, const Mode mode, helper::Comm comm)
: Engine("BP5Writer", io, name, mode, std::move(comm)), m_BP5Serializer(),
m_TransportFactory(io, m_Comm), m_Profiler(m_Comm), m_AggregatorInitializedThisStep(false)
: Engine("BP5Writer", io, name, mode, std::move(comm)), m_BP5Serializer(), m_Profiler(m_Comm),
m_AggregatorInitializedThisStep(false)
{
m_EngineStart = Now();
PERFSTUBS_SCOPED_TIMER("BP5Writer::Open");
Expand Down Expand Up @@ -482,8 +483,13 @@ void BP5Writer::WriteData(format::BufferV *Data)
"is not supported in BP5");
}
AggTransportData &aggData = m_AggregatorSpecifics.at(GetCacheKey(m_Aggregator));
aggData.m_FileDataManager.FlushFiles();
aggData.m_FileDataManager.FinalizeSegment();
// Non-aggregator ranks in TwoLevelShm-style modes reach here without an
// open substream (only the aggregator rank holds the transport).
if (aggData.m_DataSubstream)
{
aggData.m_DataSubstream->Flush();
aggData.m_DataSubstream->FinalizeSegment();
}
delete Data;
}
}
Expand Down Expand Up @@ -530,7 +536,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));
aggData.m_FileDataManager.WriteFileAt(DataVec.data(), 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)
{
Expand Down Expand Up @@ -1277,6 +1284,15 @@ void BP5Writer::EndStep()
// PRIVATE
void BP5Writer::Init()
{
if (m_IO.m_TransportsParameters.size() > 1)
{
helper::Throw<std::invalid_argument>(
"Engine", "BP5Writer", "Init",
"BP5 engine accepts only one transport per IO; got " +
std::to_string(m_IO.m_TransportsParameters.size()) +
". Multiple parallel transports per IO are no longer supported.");
}

m_BP5Serializer.m_Engine = this;
m_RankMPI = m_Comm.Rank();
InitParameters();
Expand Down Expand Up @@ -1783,8 +1799,8 @@ void BP5Writer::InitMetadataTransports()
m_MetadataIndexFileName = GetBPMetadataIndexFileName(m_Name);
}

m_TransportFactory.MkDirsBarrier({m_MetadataFileName}, m_IO.m_TransportsParameters,
m_Parameters.NodeLocal || m_WriteToBB);
transport::MkDirsBarrier(m_Comm, {m_MetadataFileName}, m_IO.m_TransportsParameters,
m_Parameters.NodeLocal || m_WriteToBB);

// Write s3.json sidecar so readers can auto-detect hybrid S3 storage
if (m_Comm.Rank() == 0 && !m_Parameters.DataFileTransport.empty())
Expand Down Expand Up @@ -1845,16 +1861,14 @@ void BP5Writer::InitMetadataTransports()
{
m_IO.m_TransportsParameters[i]["DirectIO"] = "false";
}
m_MetaMetadataFile = m_TransportFactory.OpenFileTransport(
m_MetaMetadataFileName, m_OpenMode, m_IO.m_TransportsParameters[0], true, false,
m_Comm);
m_MetaMetadataFile = transport::OpenFile(m_Comm, m_MetaMetadataFileName, m_OpenMode,
m_IO.m_TransportsParameters[0], true);

m_MetadataFile = m_TransportFactory.OpenFileTransport(
m_MetadataFileName, m_OpenMode, m_IO.m_TransportsParameters[0], true, false, m_Comm);
m_MetadataFile = transport::OpenFile(m_Comm, m_MetadataFileName, m_OpenMode,
m_IO.m_TransportsParameters[0], true);

m_MetadataIndexFile = m_TransportFactory.OpenFileTransport(
m_MetadataIndexFileName, m_OpenMode, m_IO.m_TransportsParameters[0], true, false,
m_Comm);
m_MetadataIndexFile = transport::OpenFile(m_Comm, m_MetadataIndexFileName, m_OpenMode,
m_IO.m_TransportsParameters[0], true);

if (m_DrainBB)
{
Expand Down Expand Up @@ -1903,8 +1917,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,
Expand All @@ -1917,18 +1931,15 @@ void BP5Writer::InitTransports()
if (m_DrainBB)
{
/* Create the directories on target anyway by main thread */
aggData.m_FileDataManager.MkDirsBarrier(aggData.m_DrainSubStreamNames,
m_DataTransportsParameters, m_Parameters.NodeLocal);
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);
}
}

Expand Down Expand Up @@ -1958,13 +1969,13 @@ void BP5Writer::OpenSubfile(bool useComm, bool forceAppend)
std::cout << "Rank " << m_Comm.Rank() << " cache miss for aggregator key " << cacheKey
<< std::endl;
}
m_AggregatorSpecifics.emplace(std::make_pair(cacheKey, AggTransportData(m_IO, m_Comm)));
m_AggregatorSpecifics.emplace(cacheKey, AggTransportData{});
}

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;

Expand Down Expand Up @@ -2002,9 +2013,9 @@ void BP5Writer::OpenSubfile(bool useComm, bool forceAppend)
std::cout << "Rank " << m_Comm.Rank() << " opening data file with Comm"
<< std::endl;
}
aggData.m_FileDataManager.OpenFiles(aggData.m_SubStreamNames, mode,
m_IO.m_TransportsParameters, true,
*DataWritingComm);
aggData.m_DataSubstream = transport::OpenFileChained(
m_Comm, aggData.m_SubStreamName, mode, m_IO.m_TransportsParameters[0], true,
*DataWritingComm);
}
else
{
Expand All @@ -2013,8 +2024,8 @@ void BP5Writer::OpenSubfile(bool useComm, bool forceAppend)
std::cout << "Rank " << m_Comm.Rank() << " opening data file no Comm"
<< std::endl;
}
aggData.m_FileDataManager.OpenFiles(aggData.m_SubStreamNames, mode,
m_IO.m_TransportsParameters, true);
aggData.m_DataSubstream = transport::OpenFile(m_Comm, aggData.m_SubStreamName, mode,
m_IO.m_TransportsParameters[0], true);
}
}
}
Expand Down Expand Up @@ -2191,16 +2202,16 @@ void BP5Writer::InitBPBuffer()
const size_t off = m_AppendDataPos[m_Aggregator->m_SubStreamIndex];
if (off < MaxSizeT)
{
aggData.m_FileDataManager.Truncate(off);
aggData.m_DataSubstream->Truncate(off);
// Seek is needed since truncate does not seek.
// SeekTo instead of SeetToFileEnd in case a transport
// Seek instead of SeekToEnd in case a transport
// does not support actual truncate.
aggData.m_FileDataManager.SeekTo(off);
aggData.m_DataSubstream->Seek(off);
m_DataPos = off;
}
else
{
m_DataPos = aggData.m_FileDataManager.GetFileSize(0);
m_DataPos = aggData.m_DataSubstream->GetSize();
}
}

Expand Down Expand Up @@ -2264,7 +2275,7 @@ void BP5Writer::InitBPBuffer()
// data existed but index was missing
if (m_Aggregator->m_IsAggregator)
{
aggData.m_FileDataManager.SeekTo(0);
aggData.m_DataSubstream->Seek(0);
}
}

Expand Down Expand Up @@ -2414,10 +2425,16 @@ void BP5Writer::DoClose(const int transportIndex)
wait += Now() - wait_start;
}

// However many AggTransportData we created, we need to close them all
for (auto it = m_AggregatorSpecifics.begin(); it != m_AggregatorSpecifics.end(); ++it)
// However many AggTransportData we created, we need to close them all.
// transportIndex is ignored: per-engine multi-transport is no longer
// supported (see Init's check), so we always close every substream.
(void)transportIndex;
for (auto &aggPair : m_AggregatorSpecifics)
{
it->second.m_FileDataManager.CloseFiles(transportIndex);
if (aggPair.second.m_DataSubstream)
{
aggPair.second.m_DataSubstream->Close();
}
}

// Delete files from temporary storage if draining was on
Expand Down Expand Up @@ -2464,7 +2481,7 @@ void BP5Writer::DoClose(const int transportIndex)
// cleaned up while those resources are still available.
for (auto &aggPair : m_AggregatorSpecifics)
{
aggPair.second.m_FileDataManager.m_Transports.clear();
aggPair.second.m_DataSubstream.reset();
}
m_MetadataFile.reset();
m_MetaMetadataFile.reset();
Expand All @@ -2479,13 +2496,13 @@ void BP5Writer::FlushProfiler()

for (auto &specs : m_AggregatorSpecifics)
{
std::vector<std::string> types = specs.second.m_FileDataManager.GetTransportsTypes();
std::vector<std::string> names = specs.second.m_FileDataManager.GetTransportsNames();
std::vector<adios2::profiling::IOChrono *> profs =
specs.second.m_FileDataManager.GetTransportsProfilers();
transportTypes.insert(transportTypes.end(), types.begin(), types.end());
transportNames.insert(transportNames.end(), names.begin(), names.end());
transportProfilers.insert(transportProfilers.end(), profs.begin(), profs.end());
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);
}
}

// find first File type output, where we can write the profile
Expand Down
Loading
Loading