diff --git a/Framework/AODNative/CMakeLists.txt b/Framework/AODNative/CMakeLists.txt index 154826aa31690..462ad6ee0d154 100644 --- a/Framework/AODNative/CMakeLists.txt +++ b/Framework/AODNative/CMakeLists.txt @@ -13,6 +13,7 @@ # ROOT, at build or run time: see the check at the end of this file. o2_add_library(AODNative SOURCES src/ByteSource.cxx + src/FailoverByteSource.cxx src/LocalByteSource.cxx src/Container.cxx src/Objects.cxx @@ -54,7 +55,8 @@ add_executable(o2-test-framework-aod-native test/test_Codecs.cxx test/test_Objects.cxx test/test_Decoder.cxx - test/test_ReadPlan.cxx) + test/test_ReadPlan.cxx + test/test_Failover.cxx) target_include_directories(o2-test-framework-aod-native PRIVATE ${CMAKE_CURRENT_LIST_DIR}/src) target_link_libraries(o2-test-framework-aod-native PRIVATE O2::AODNative O2::Catch2 PkgConfig::zstd PkgConfig::lz4 PkgConfig::lzma ZLIB::ZLIB) diff --git a/Framework/AODNative/DEVELOPMENT.md b/Framework/AODNative/DEVELOPMENT.md index 8482f51d218a5..248a467397c62 100644 --- a/Framework/AODNative/DEVELOPMENT.md +++ b/Framework/AODNative/DEVELOPMENT.md @@ -229,10 +229,20 @@ ROOT's 50 MB TTreeCache. With the macOS allocator cache on, max RSS is higher `XRootDByteSource` (root://, roots://, xroot://, xroots://), built when O2 finds XRootD (optional in O2); otherwise those URLs get an IoError, so `auto` falls -back. Links only libXrdCl. `readvInto` sends a whole read plan at once: ranges -up to 2097136 bytes go in vectored reads of at most 1024 chunks, larger ones -in plain reads of at most 64 MiB, all asynchronous, then waits for all of them; -a short or failed reply is an IoError. `readCalls()` counts requests sent. +back. Links only libXrdCl. After opening it asks the data server for its +kXR_readv limits (`readv_ior_max`, `readv_iov_max`), as TNetXNGFile does, +keeping XrdCl's defaults (2097136 bytes, 1024 chunks) when it does not answer. +`readvInto` sends ranges up to the chunk limit in vectored reads and larger ones +in plain reads of at most 8 MiB, asynchronously, in rounds of at most +`Options::remoteInFlightBytes` (32 MiB), waiting for each round; a short or +failed reply is an IoError. `readCalls()` counts requests sent. + +Until 2026-11-05 a whole plan went out at once, with plain reads of up to +64 MiB. On the Grid, at ORNL reading Poznan EOS (slow transatlantic link), the +native reader then failed 2-6x more often than ROOT, mostly with "Operation +expired": requests sharing the link took longer than XRD_REQUESTTIMEOUT (150 s +in the ALICE environment). Smaller requests and the in-flight limit keep each +request well within it. Tested against a local server (Homebrew xrootd 6.2.0, `all.export ~/src/data r/o`): byte-identical with both readers over root:// on hy_3993289/0065, @@ -319,9 +329,16 @@ in DPL (analysis-histograms, LHC24ar) 380 vs 319, with identical results. sends `access read ` through `gGrid`, as `TJAlienFile::Open` does, and turns each replica into `root:////?&authz=`. Files inside an archive (`#member`, on the LFN or the replica) are refused, so `auto` falls -back to ROOT. `openNativeAOD()` tries the replicas in the order AliEn proposes, -moving on after an IoError only (a format problem is the same on every -replica); DPL's reader and the comparison tool both open files through it. +back to ROOT. `openNativeAOD()` reads the replicas, in the order AliEn +proposes, through `ByteSource::openReplicas()` (`FailoverByteSource`): the +first replica which opens is read; a read which fails with an IoError, at any +point of the file, is retried on that replica, reopened, up to +`Options::readRetries` (2) times per replica with a doubling delay from +`retryDelay` (2 s), and then continues on the next replica of the same size. +Only IoErrors are retried: a format problem is the same on every replica. Each +retry and change of replica is logged. ROOT only moves to another replica when +opening fails, so this makes the native reader more robust than ROOT on flaky +storage. DPL's reader and the comparison tool both open files through it. The file keeps its LFN as name: what DataInputDirector compares, what goes with derived output, and what monitoring reports, with `se=` and `open_time=` as for TJAlienFile. `XRootDByteSource` uses the full URL only to diff --git a/Framework/AODNative/include/AODNative/ByteSource.h b/Framework/AODNative/include/AODNative/ByteSource.h index 26376923328c2..9ba49c284faa2 100644 --- a/Framework/AODNative/include/AODNative/ByteSource.h +++ b/Framework/AODNative/include/AODNative/ByteSource.h @@ -14,9 +14,11 @@ #include #include +#include #include #include #include +#include namespace o2::aod::native { @@ -46,6 +48,23 @@ struct Options { size_t streamWindowBytes = 2 * 1024 * 1024; HttpVersion httpVersion = HttpVersion::Auto; int httpConnections = 4; + /// For a source opened with ByteSource::openReplicas: how many times, at + /// most, failed reads are retried on a replica (reopened) during the life of + /// the file before moving on to the next replica, and the seconds waited + /// before the first retry (doubled for each further one). + int readRetries = 2; + double retryDelay = 2.0; + /// Most bytes of one read plan in flight at once on a remote file, so that + /// each request finishes within the server timeout on a slow link (XRootD + /// expires requests after XRD_REQUESTTIMEOUT, 150 s in the ALICE setup). + size_t remoteInFlightBytes = 32 * 1024 * 1024; +}; + +/// One copy of a file, for ByteSource::openReplicas: where it is, and a label +/// (e.g. the storage element) to report which one is being read. +struct Replica { + std::string url; + std::string label; }; struct ReadRequest { @@ -77,7 +96,15 @@ class ByteSource virtual uint64_t bytesRead() const = 0; virtual uint64_t readCalls() const = 0; virtual std::string url() const = 0; + /// The label of the replica being read, for a source opened with openReplicas(). + virtual std::string replicaLabel() const { return {}; } static std::unique_ptr open(std::string url, const Options&); + /// Opens the first replica which opens, and on a read error retries it + /// (Options::readRetries times, reopened) and then moves on to the next + /// replica of the same size. @a log, if given, is told about every retry + /// and every change of replica. Throws IoError when no replica is left. + static std::unique_ptr openReplicas(std::vector replicas, const Options&, + std::function log = {}); }; } // namespace o2::aod::native diff --git a/Framework/AODNative/src/ByteSource.cxx b/Framework/AODNative/src/ByteSource.cxx index adbf54d8d8954..c70dae511507e 100644 --- a/Framework/AODNative/src/ByteSource.cxx +++ b/Framework/AODNative/src/ByteSource.cxx @@ -11,6 +11,7 @@ #include "AODNative/ByteSource.h" #include "AODNative/Errors.h" +#include "FailoverByteSource.h" #include "LocalByteSource.h" #ifdef AODNATIVE_WITH_XROOTD #include "XRootDByteSource.h" @@ -59,4 +60,10 @@ std::unique_ptr ByteSource::open(std::string url, const Options& opt return std::make_unique(std::move(url), options); } +std::unique_ptr ByteSource::openReplicas(std::vector replicas, const Options& options, + std::function log) +{ + return std::make_unique(std::move(replicas), options, &ByteSource::open, std::move(log)); +} + } // namespace o2::aod::native diff --git a/Framework/AODNative/src/FailoverByteSource.cxx b/Framework/AODNative/src/FailoverByteSource.cxx new file mode 100644 index 0000000000000..c44c1640fe80d --- /dev/null +++ b/Framework/AODNative/src/FailoverByteSource.cxx @@ -0,0 +1,151 @@ +// Copyright 2019-2026 CERN and copyright holders of ALICE O2. +// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders. +// All rights not expressly granted are reserved. +// +// This software is distributed under the terms of the GNU General Public +// License v3 (GPL Version 3), copied verbatim in the file "COPYING". +// +// In applying this license CERN does not waive the privileges and immunities +// granted to it by virtue of its status as an Intergovernmental Organization +// or submit itself to any jurisdiction. + +#include "FailoverByteSource.h" +#include "AODNative/Errors.h" +#include +#include +#include + +namespace o2::aod::native +{ + +namespace +{ +std::string labelOf(Replica const& r) +{ + return r.label.empty() ? r.url.substr(0, r.url.find('?')) : r.label; +} +} // namespace + +FailoverByteSource::FailoverByteSource(std::vector replicas, Options const& options, Opener opener, Log log) + : mReplicas(std::move(replicas)), mOptions(options), mOpener(std::move(opener)), mLog(std::move(log)) +{ + if (mReplicas.empty()) { + throw IoError("(no replica)", "no replica to open"); + } + for (mIndex = 0; mIndex < mReplicas.size(); ++mIndex) { + try { + mSource = openReplica(mIndex); + mSize = mSource->size(); + mRemote = mSource->isRemote(); + return; + } catch (IoError const& e) { + report("Could not open replica on " + labelOf(mReplicas[mIndex]) + ": " + e.what()); + mErrors += " " + labelOf(mReplicas[mIndex]) + ": " + e.what() + ";"; + } + } + throw IoError(labelOf(mReplicas.front()), "no replica could be opened:" + mErrors); +} + +std::shared_ptr FailoverByteSource::openReplica(size_t index) +{ + std::shared_ptr source = mOpener(mReplicas[index].url, mOptions); + if (mSource && source->size() != mSize) { + throw IoError(source->url(), "size " + std::to_string(source->size()) + " differs from the " + std::to_string(mSize) + " bytes of the replica read so far"); + } + return source; +} + +std::shared_ptr FailoverByteSource::current() const +{ + std::lock_guard lock{mMutex}; + return mSource; +} + +std::string FailoverByteSource::replicaLabel() const +{ + std::lock_guard lock{mMutex}; + return mReplicas[mIndex].label; +} + +uint64_t FailoverByteSource::bytesRead() const +{ + return mRetiredBytes.load(std::memory_order_relaxed) + current()->bytesRead(); +} + +uint64_t FailoverByteSource::readCalls() const +{ + return mRetiredCalls.load(std::memory_order_relaxed) + current()->readCalls(); +} + +void FailoverByteSource::report(std::string const& message) const +{ + if (mLog) { + mLog(message); + } +} + +void FailoverByteSource::readInto(uint64_t off, size_t len, void* dst) +{ + withFailover([&](ByteSource& s) { s.readInto(off, len, dst); }); +} + +void FailoverByteSource::readvInto(std::span requests) +{ + withFailover([&](ByteSource& s) { s.readvInto(requests); }); +} + +void FailoverByteSource::withFailover(std::function const& read) +{ + while (true) { + std::shared_ptr source; + uint64_t generation; + { + std::lock_guard lock{mMutex}; + source = mSource; + generation = mGeneration; + } + try { + read(*source); + return; + } catch (IoError const& e) { + std::lock_guard lock{mMutex}; + if (generation != mGeneration) { + continue; // another read already replaced the replica: try the new one + } + mRetiredBytes.fetch_add(source->bytesRead(), std::memory_order_relaxed); + mRetiredCalls.fetch_add(source->readCalls(), std::memory_order_relaxed); + std::string error = e.what(); + // Reopen the same replica, after a growing delay, or move on to the next + // one, until a replica opens. Other readers wait on mMutex meanwhile. + while (true) { + if (mAttempts < mOptions.readRetries) { + double delay = mOptions.retryDelay * std::pow(2., mAttempts); + ++mAttempts; + report("Read error on " + labelOf(mReplicas[mIndex]) + ", retry " + std::to_string(mAttempts) + " of " + + std::to_string(mOptions.readRetries) + " in " + std::to_string((int)std::lround(delay)) + " s: " + error); + std::this_thread::sleep_for(std::chrono::duration(delay)); + } else { + mErrors += " " + labelOf(mReplicas[mIndex]) + ": " + error + ";"; + if (mIndex + 1 >= mReplicas.size()) { + throw IoError(labelOf(mReplicas.front()), "no replica left to read from:" + mErrors); + } + ++mIndex; + mAttempts = 0; + report("Moving to replica on " + labelOf(mReplicas[mIndex]) + " after: " + error); + } + try { + mSource = openReplica(mIndex); + ++mGeneration; + break; + } catch (IoError const& reopen) { + error = reopen.what(); + if (mAttempts == 0) { + mAttempts = mOptions.readRetries; // a replica that does not open is not retried + } + } + } + } + } +} + +} // namespace o2::aod::native diff --git a/Framework/AODNative/src/FailoverByteSource.h b/Framework/AODNative/src/FailoverByteSource.h new file mode 100644 index 0000000000000..15e0aee055f75 --- /dev/null +++ b/Framework/AODNative/src/FailoverByteSource.h @@ -0,0 +1,78 @@ +// Copyright 2019-2026 CERN and copyright holders of ALICE O2. +// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders. +// All rights not expressly granted are reserved. +// +// This software is distributed under the terms of the GNU General Public +// License v3 (GPL Version 3), copied verbatim in the file "COPYING". +// +// In applying this license CERN does not waive the privileges and immunities +// granted to it by virtue of its status as an Intergovernmental Organization +// or submit itself to any jurisdiction. + +#ifndef O2_FRAMEWORK_AODNATIVE_FAILOVERBYTESOURCE_H_ +#define O2_FRAMEWORK_AODNATIVE_FAILOVERBYTESOURCE_H_ + +#include "AODNative/ByteSource.h" +#include +#include +#include + +namespace o2::aod::native +{ + +/// The replicas of one file, read as one ByteSource. A read which fails with +/// an IoError is retried on the same replica, reopened, Options::readRetries +/// times with a growing delay, and then on the next replica of the same size. +/// Opening goes through the replicas in order too. Reads already in flight on +/// a replica that is being replaced finish there, or fail and are retried. +class FailoverByteSource : public ByteSource +{ + public: + using Opener = std::function(std::string const& url, Options const&)>; + using Log = std::function; + + /// Throws IoError when no replica opens. + FailoverByteSource(std::vector replicas, Options const& options, Opener opener, Log log); + + uint64_t size() const override { return mSize; } + void readInto(uint64_t off, size_t len, void* dst) override; + void readvInto(std::span requests) override; + std::span tryMap(uint64_t off, size_t len) override { return current()->tryMap(off, len); } + bool canMap() const override { return current()->canMap(); } + bool isRemote() const override { return mRemote; } + void willNeed(uint64_t off, size_t len) override { current()->willNeed(off, len); } + void dontNeed(uint64_t off, size_t len) override { current()->dontNeed(off, len); } + uint64_t bytesRead() const override; + uint64_t readCalls() const override; + std::string url() const override { return current()->url(); } + std::string replicaLabel() const override; + + private: + std::shared_ptr current() const; + /// Runs @a read on the current replica, retrying and failing over on IoError. + void withFailover(std::function const& read); + /// Opens replica @a index; throws IoError, also when its size differs. + std::shared_ptr openReplica(size_t index); + void report(std::string const& message) const; + + std::vector mReplicas; + Options mOptions; + Opener mOpener; + Log mLog; + + mutable std::mutex mMutex; + std::shared_ptr mSource; // guarded by mMutex + size_t mIndex = 0; // guarded by mMutex + uint64_t mGeneration = 0; // guarded by mMutex, bumped at every reopen + int mAttempts = 0; // retries done on the current replica, guarded by mMutex + std::string mErrors; // what went wrong on the replicas left behind + + uint64_t mSize = 0; + bool mRemote = false; + std::atomic mRetiredBytes = 0; + std::atomic mRetiredCalls = 0; +}; + +} // namespace o2::aod::native + +#endif // O2_FRAMEWORK_AODNATIVE_FAILOVERBYTESOURCE_H_ diff --git a/Framework/AODNative/src/XRootDByteSource.cxx b/Framework/AODNative/src/XRootDByteSource.cxx index d2025b6960043..2393401cf5417 100644 --- a/Framework/AODNative/src/XRootDByteSource.cxx +++ b/Framework/AODNative/src/XRootDByteSource.cxx @@ -12,9 +12,12 @@ #include "XRootDByteSource.h" #include "AODNative/Errors.h" #include +#include +#include #include #include #include +#include #include namespace o2::aod::native @@ -22,12 +25,14 @@ namespace o2::aod::native namespace { -// XrdCl's defaults for kXR_readv: chunks of at most 2097136 bytes, at most 1024 -// of them per request. Larger ranges are read with plain reads, in pieces small -// enough for the uint32_t size of XrdCl::File::Read. -constexpr size_t kMaxVectorChunk = 2097136; -constexpr size_t kMaxVectorChunks = 1024; -constexpr size_t kMaxPlainRead = 64 * 1024 * 1024; +// XrdCl's defaults for kXR_readv, used when the server does not tell its own +// limits: chunks of at most 2097136 bytes, at most 1024 of them per request. +// Larger ranges are read with plain reads, in pieces of at most kMaxPlainRead: +// small enough that one request completes well within the request timeout on +// a slow link, as ROOT's basket-sized reads do. +constexpr size_t kDefaultVectorChunk = 2097136; +constexpr size_t kDefaultVectorChunks = 1024; +constexpr size_t kMaxPlainRead = 8 * 1024 * 1024; /// Counts down the replies of the requests sent by one readvInto(), keeping /// the first error. @@ -103,20 +108,55 @@ class Reply : public XrdCl::ResponseHandler size_t mExpected; }; -struct PlainRead { - uint64_t offset; - uint32_t length; - void* dst; -}; - -struct VectorRead { +/// One request to send: a plain read when it has one chunk, a vectored one otherwise. +struct Request { XrdCl::ChunkList chunks; size_t bytes = 0; }; + +/// Asks the data server for its kXR_readv limits ("readv_ior_max" and +/// "readv_iov_max"), as ROOT's TNetXNGFile does. Leaves the defaults when the +/// server does not answer. +void queryVectorLimits(XrdCl::File& file, size_t& chunkMax, size_t& chunksMax) +{ + std::string server; + if (!file.GetProperty("DataServer", server) || server.empty()) { + return; + } + XrdCl::FileSystem fs{XrdCl::URL{server}}; + XrdCl::Buffer arg; + arg.FromString("readv_ior_max readv_iov_max"); + XrdCl::Buffer* response = nullptr; + auto status = fs.Query(XrdCl::QueryCode::Config, arg, response); + if (!status.IsOK() || !response) { + delete response; + return; + } + std::istringstream lines{response->ToString()}; + delete response; + std::string ior, iov; + if (!std::getline(lines, ior) || !std::getline(lines, iov)) { + return; + } + // An unknown setting is echoed back by name instead of a number. + try { + if (auto v = std::stoull(ior); v > 0) { + chunkMax = std::min(v, kDefaultVectorChunk); + } + if (auto v = std::stoull(iov); v > 0) { + chunksMax = std::min(v, kDefaultVectorChunks); + } + } catch (std::exception const&) { + } +} } // namespace XRootDByteSource::XRootDByteSource(std::string url, Options const& options) - : mName(url.substr(0, url.find('?'))), mFile(std::make_unique()) + : mName(url.substr(0, url.find('?'))), + mFile(std::make_unique()), + mVectorChunkMax(kDefaultVectorChunk), + mVectorChunksMax(kDefaultVectorChunks), + mInFlightMax(options.remoteInFlightBytes) { auto status = mFile->Open(url, XrdCl::OpenFlags::Read); if (!status.IsOK()) { @@ -130,6 +170,7 @@ XRootDByteSource::XRootDByteSource(std::string url, Options const& options) } mSize = info->GetSize(); delete info; + queryVectorLimits(*mFile, mVectorChunkMax, mVectorChunksMax); } XRootDByteSource::~XRootDByteSource() @@ -154,63 +195,76 @@ void XRootDByteSource::readInto(uint64_t off, size_t len, void* dst) void XRootDByteSource::readvInto(std::span requests) { - std::vector plain; - std::vector vectors; + // Ranges larger than a vector chunk are read with plain reads of at most + // kMaxPlainRead; the others are gathered into vectored reads within the + // server's limits. No request carries more than the in-flight limit. + size_t window = mInFlightMax > 0 ? mInFlightMax : SIZE_MAX; + std::vector out; + Request vector; size_t total = 0; + auto flushVector = [&] { + if (!vector.chunks.empty()) { + out.push_back(std::move(vector)); + vector = Request{}; + } + }; for (auto const& r : requests) { checkRange(r.offset, r.length); total += r.length; - if (r.length > kMaxVectorChunk) { + if (r.length == 0) { + continue; + } + if (r.length > mVectorChunkMax) { for (size_t done = 0; done < r.length; done += kMaxPlainRead) { auto n = std::min(kMaxPlainRead, r.length - done); - plain.push_back({r.offset + done, (uint32_t)n, (uint8_t*)r.dst + done}); + Request plain; + plain.chunks.emplace_back(r.offset + done, (uint32_t)n, (uint8_t*)r.dst + done); + plain.bytes = n; + out.push_back(std::move(plain)); } continue; } - if (r.length == 0) { - continue; - } - if (vectors.empty() || vectors.back().chunks.size() == kMaxVectorChunks) { - vectors.emplace_back(); - } - vectors.back().chunks.emplace_back(r.offset, (uint32_t)r.length, r.dst); - vectors.back().bytes += r.length; - } - // A vector of one chunk is a plain read. - for (auto& v : vectors) { - if (v.chunks.size() == 1) { - auto& c = v.chunks.front(); - plain.push_back({c.offset, c.length, c.buffer}); - v.chunks.clear(); + if (vector.chunks.size() == mVectorChunksMax || (!vector.chunks.empty() && vector.bytes + r.length > window)) { + flushVector(); } + vector.chunks.emplace_back(r.offset, (uint32_t)r.length, r.dst); + vector.bytes += r.length; } - std::erase_if(vectors, [](VectorRead const& v) { return v.chunks.empty(); }); - if (plain.empty() && vectors.empty()) { - return; - } + flushVector(); - Completion completion{plain.size() + vectors.size()}; - for (auto const& p : plain) { - auto* reply = new Reply(completion, p.length); - auto status = mFile->Read(p.offset, p.length, p.dst, reply); - if (!status.IsOK()) { - delete reply; - completion.done(status.ToString()); + // Send the requests in rounds of at most `window` bytes, waiting for each + // round: on a slow link every request then gets its share of the bandwidth, + // instead of all of them competing until some expire. + size_t calls = 0; + for (size_t first = 0; first < out.size();) { + size_t last = first, bytes = 0; + while (last < out.size() && (last == first || bytes + out[last].bytes <= window)) { + bytes += out[last++].bytes; } - } - for (auto const& v : vectors) { - auto* reply = new Reply(completion, v.bytes); - auto status = mFile->VectorRead(v.chunks, nullptr, reply); - if (!status.IsOK()) { - delete reply; - completion.done(status.ToString()); + Completion completion{last - first}; + for (size_t i = first; i < last; ++i) { + auto& request = out[i]; + auto* reply = new Reply(completion, request.bytes); + XrdCl::XRootDStatus status; + if (request.chunks.size() == 1) { + auto& c = request.chunks.front(); + status = mFile->Read(c.offset, c.length, c.buffer, reply); + } else { + status = mFile->VectorRead(request.chunks, nullptr, reply); + } + if (!status.IsOK()) { + delete reply; + completion.done(status.ToString()); + } } + auto error = completion.wait(); + if (!error.empty()) { + throw IoError(mName, error); + } + calls += last - first; + first = last; } - auto error = completion.wait(); - if (!error.empty()) { - throw IoError(mName, error); - } - mReadCalls.fetch_add(plain.size() + vectors.size(), std::memory_order_relaxed); + mReadCalls.fetch_add(calls, std::memory_order_relaxed); mBytesRead.fetch_add(total, std::memory_order_relaxed); } diff --git a/Framework/AODNative/src/XRootDByteSource.h b/Framework/AODNative/src/XRootDByteSource.h index d0169fac157ea..a92afb752d4f5 100644 --- a/Framework/AODNative/src/XRootDByteSource.h +++ b/Framework/AODNative/src/XRootDByteSource.h @@ -25,9 +25,11 @@ namespace o2::aod::native { /// A file read with the XRootD client (root://, roots://, xroot://, xroots://). -/// readvInto() sends all its ranges at once, as vectored reads for the small -/// ones and plain reads for the large ones, and waits for all of them: one -/// round trip for a whole read plan. +/// readvInto() sends its ranges as vectored reads for the small ones, within +/// the server's kXR_readv limits, and plain reads of at most 8 MB for the +/// large ones, in rounds of at most Options::remoteInFlightBytes: usually one +/// round trip for a whole read plan, and no request so large that it expires +/// on a slow link. Errors are not retried here: see FailoverByteSource. class XRootDByteSource : public ByteSource { public: @@ -49,6 +51,9 @@ class XRootDByteSource : public ByteSource std::string mName; std::unique_ptr mFile; uint64_t mSize = 0; + size_t mVectorChunkMax; + size_t mVectorChunksMax; + size_t mInFlightMax; std::atomic mBytesRead = 0; std::atomic mReadCalls = 0; }; diff --git a/Framework/AODNative/test/test_Failover.cxx b/Framework/AODNative/test/test_Failover.cxx new file mode 100644 index 0000000000000..ecb056b99042f --- /dev/null +++ b/Framework/AODNative/test/test_Failover.cxx @@ -0,0 +1,228 @@ +// Copyright 2019-2026 CERN and copyright holders of ALICE O2. +// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders. +// All rights not expressly granted are reserved. +// +// This software is distributed under the terms of the GNU General Public +// License v3 (GPL Version 3), copied verbatim in the file "COPYING". +// +// In applying this license CERN does not waive the privileges and immunities +// granted to it by virtue of its status as an Intergovernmental Organization +// or submit itself to any jurisdiction. + +#include "FailoverByteSource.h" +#include "AODNative/Errors.h" +#include +#include +#include +#include + +using namespace o2::aod::native; + +namespace +{ +// A replica whose byte i is i % 251. Its behaviour is shared between the +// instances opened for the same URL, so that a reopen sees the same state. +struct Behaviour { + size_t size = 1000; + bool opens = true; + std::atomic failingReads = 0; // the next reads which fail with IoError + std::atomic opened = 0; + std::atomic reads = 0; +}; + +struct FakeSource : ByteSource { + FakeSource(std::string u, Behaviour& b) : name(std::move(u)), behaviour(b) {} + uint64_t size() const override { return behaviour.size; } + void readInto(uint64_t off, size_t len, void* dst) override + { + ReadRequest r{off, len, dst}; + readvInto({&r, 1}); + } + void readvInto(std::span requests) override + { + ++behaviour.reads; + int failing = behaviour.failingReads.load(); + while (failing > 0 && !behaviour.failingReads.compare_exchange_weak(failing, failing - 1)) { + } + if (failing > 0) { + throw IoError(name, "Operation expired"); + } + for (auto& r : requests) { + for (size_t i = 0; i < r.length; ++i) { + ((uint8_t*)r.dst)[i] = (r.offset + i) % 251; + } + bytes += r.length; + } + ++calls; + } + bool isRemote() const override { return true; } + uint64_t bytesRead() const override { return bytes; } + uint64_t readCalls() const override { return calls; } + std::string url() const override { return name; } + std::string name; + Behaviour& behaviour; + std::atomic bytes = 0; + std::atomic calls = 0; +}; + +struct Setup { + std::map replicas; + std::vector log; + Options options; + Setup() { options.retryDelay = 0; } + + std::unique_ptr open() + { + std::vector list; + for (auto& [url, _] : replicas) { + list.push_back({url, "SE-" + url}); + } + auto opener = [this](std::string const& url, Options const&) -> std::unique_ptr { + auto& b = replicas.at(url); + ++b.opened; + if (!b.opens) { + throw IoError(url, "no such file"); + } + return std::make_unique(url, b); + }; + return std::make_unique(list, options, opener, [this](std::string const& m) { log.push_back(m); }); + } +}; + +bool readsCorrectly(ByteSource& s, uint64_t off = 100, size_t len = 50) +{ + std::vector buf(len); + s.readInto(off, len, buf.data()); + for (size_t i = 0; i < len; ++i) { + if (buf[i] != (off + i) % 251) { + return false; + } + } + return true; +} +} // namespace + +TEST_CASE("FailoverByteSource reads the first replica") +{ + Setup s; + s.replicas["a"]; + s.replicas["b"]; + auto f = s.open(); + REQUIRE(readsCorrectly(*f)); + CHECK(f->replicaLabel() == "SE-a"); + CHECK(s.replicas["b"].opened == 0); + CHECK(f->bytesRead() == 50); + CHECK(f->readCalls() == 1); + CHECK(s.log.empty()); +} + +TEST_CASE("FailoverByteSource skips replicas which do not open") +{ + Setup s; + s.replicas["a"].opens = false; + s.replicas["b"]; + auto f = s.open(); + CHECK(f->replicaLabel() == "SE-b"); + REQUIRE(readsCorrectly(*f)); + CHECK(s.log.size() == 1); +} + +TEST_CASE("FailoverByteSource throws when no replica opens") +{ + Setup s; + s.replicas["a"].opens = false; + s.replicas["b"].opens = false; + CHECK_THROWS_AS(s.open(), IoError); +} + +TEST_CASE("FailoverByteSource retries a transient read error on the same replica") +{ + Setup s; + s.replicas["a"].failingReads = 1; + s.replicas["b"]; + auto f = s.open(); + REQUIRE(readsCorrectly(*f)); + CHECK(f->replicaLabel() == "SE-a"); + CHECK(s.replicas["a"].opened == 2); // reopened for the retry + CHECK(s.replicas["b"].opened == 0); + REQUIRE(s.log.size() == 1); + CHECK(s.log[0].find("retry 1 of 2") != std::string::npos); +} + +TEST_CASE("FailoverByteSource moves to the next replica after the retries") +{ + Setup s; + s.replicas["a"].failingReads = 100; + s.replicas["b"]; + auto f = s.open(); + REQUIRE(readsCorrectly(*f)); + CHECK(f->replicaLabel() == "SE-b"); + CHECK(s.replicas["a"].reads == 3); // first attempt + 2 retries + CHECK(s.log.back().find("Moving to replica on SE-b") != std::string::npos); + // Later reads stay on the new replica. + REQUIRE(readsCorrectly(*f, 900, 100)); + CHECK(s.replicas["a"].reads == 3); +} + +TEST_CASE("FailoverByteSource does not use a replica of a different size") +{ + Setup s; + s.replicas["a"].failingReads = 100; + s.replicas["b"].size = 999; + s.replicas["c"]; + auto f = s.open(); + REQUIRE(readsCorrectly(*f)); + CHECK(f->replicaLabel() == "SE-c"); +} + +TEST_CASE("FailoverByteSource throws when every replica fails") +{ + Setup s; + s.replicas["a"].failingReads = 100; + s.replicas["b"].failingReads = 100; + auto f = s.open(); + std::vector buf(10); + try { + f->readInto(0, 10, buf.data()); + FAIL("no exception"); + } catch (IoError const& e) { + CHECK(std::string(e.what()).find("no replica left") != std::string::npos); + } + CHECK(s.replicas["a"].reads == 3); + CHECK(s.replicas["b"].reads == 3); +} + +TEST_CASE("FailoverByteSource counts the bytes of retired replicas") +{ + Setup s; + s.replicas["a"]; + s.replicas["b"]; + auto f = s.open(); + REQUIRE(readsCorrectly(*f, 0, 100)); + s.replicas["a"].failingReads = 100; + REQUIRE(readsCorrectly(*f, 0, 30)); + CHECK(f->bytesRead() == 130); + CHECK(f->readCalls() == 2); +} + +TEST_CASE("FailoverByteSource with concurrent readers") +{ + Setup s; + s.replicas["a"].failingReads = 100; + s.replicas["b"]; + auto f = s.open(); + std::vector threads; + std::atomic good = 0; + for (int t = 0; t < 4; ++t) { + threads.emplace_back([&, t] { + for (int i = 0; i < 20; ++i) { + good += readsCorrectly(*f, 10 * t + i, 40); + } + }); + } + for (auto& t : threads) { + t.join(); + } + CHECK(good == 80); + CHECK(f->replicaLabel() == "SE-b"); +} diff --git a/Framework/AnalysisSupport/src/AODInputFile.cxx b/Framework/AnalysisSupport/src/AODInputFile.cxx index e33c2c73f843a..d73df2924587a 100644 --- a/Framework/AnalysisSupport/src/AODInputFile.cxx +++ b/Framework/AnalysisSupport/src/AODInputFile.cxx @@ -152,18 +152,19 @@ std::shared_ptr openNativeAOD(std::string const& filename, RootObj if (replicas.empty()) { throw o2::aod::native::IoError(filename, "no replica the native reader can read"); } - std::string errors; + // The file is read from the first replica which opens; a read error later on + // is retried and then continues on the next replica (FailoverByteSource). + std::vector list; for (auto const& replica : replicas) { - try { - auto file = NativeAODInputFile::open(replica.url, factory, options, filename); - file->setGridInfo(replica.se, std::chrono::duration(std::chrono::steady_clock::now() - start).count()); - return file; - } catch (o2::aod::native::IoError const& e) { - LOGP(info, "Could not open replica of \"{}\" on {}: {}", filename, replica.se, e.what()); - errors += fmt::format(" {}: {};", replica.se, e.what()); - } + list.push_back({replica.url, replica.se}); } - throw o2::aod::native::IoError(filename, "no replica could be opened:" + errors); + auto source = o2::aod::native::ByteSource::openReplicas(std::move(list), options, [filename](std::string const& message) { + LOGP(info, "\"{}\": {}", filename, message); + }); + auto se = source->replicaLabel(); + auto file = NativeAODInputFile::open(std::move(source), factory, options, filename); + file->setGridInfo(se, std::chrono::duration(std::chrono::steady_clock::now() - start).count()); + return file; } } // namespace o2::framework diff --git a/Framework/AnalysisSupport/src/NativeAOD.cxx b/Framework/AnalysisSupport/src/NativeAOD.cxx index 3d88439c158df..4c4fb6ef31cb8 100644 --- a/Framework/AnalysisSupport/src/NativeAOD.cxx +++ b/Framework/AnalysisSupport/src/NativeAOD.cxx @@ -326,10 +326,15 @@ arrow::Result NativeAODFileSystem::GetFileInfo(const std::s } std::shared_ptr NativeAODInputFile::open(std::string const& url, RootObjectReadingFactory& factory, Options const& options, std::string const& name) +{ + return open(ByteSource::open(url, options), factory, options, name.empty() ? url : name); +} + +std::shared_ptr NativeAODInputFile::open(std::unique_ptr source, RootObjectReadingFactory& factory, Options const& options, std::string const& name) { auto result = std::shared_ptr(new NativeAODInputFile); - result->mName = name.empty() ? url : name; - result->mFile = std::make_shared(ByteSource::open(url, options)); + result->mName = name; + result->mFile = std::make_shared(std::move(source)); result->mFilesystem = std::make_shared(result->mFile, options, factory); result->mFilesystem->setName(result->mName); result->mParentFiles = result->mFile->readStringMap("parentFiles"); diff --git a/Framework/AnalysisSupport/src/NativeAOD.h b/Framework/AnalysisSupport/src/NativeAOD.h index ec62e6347cff3..70e1b515ff547 100644 --- a/Framework/AnalysisSupport/src/NativeAOD.h +++ b/Framework/AnalysisSupport/src/NativeAOD.h @@ -131,6 +131,8 @@ class NativeAODInputFile : public AODInputFile /// file can still go to ROOT. Throws o2::aod::native::FormatError. /// @a name is what the file is called (an LFN for a grid file); @a url by default. static std::shared_ptr open(std::string const& url, RootObjectReadingFactory& factory, o2::aod::native::Options const& options, std::string const& name = {}); + /// The same, reading from @a source (e.g. the replicas of a grid file). + static std::shared_ptr open(std::unique_ptr source, RootObjectReadingFactory& factory, o2::aod::native::Options const& options, std::string const& name); /// The storage element and the seconds the grid lookup and open took, for monitoring. void setGridInfo(std::string se, double openTime); std::string monitoringDetails() const override;