|
| 1 | +// Copyright 2019-2026 CERN and copyright holders of ALICE O2. |
| 2 | +// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders. |
| 3 | +// All rights not expressly granted are reserved. |
| 4 | +// |
| 5 | +// This software is distributed under the terms of the GNU General Public |
| 6 | +// License v3 (GPL Version 3), copied verbatim in the file "COPYING". |
| 7 | +// |
| 8 | +// In applying this license CERN does not waive the privileges and immunities |
| 9 | +// granted to it by virtue of its status as an Intergovernmental Organization |
| 10 | +// or submit itself to any jurisdiction. |
| 11 | + |
| 12 | +#include "FailoverByteSource.h" |
| 13 | +#include "AODNative/Errors.h" |
| 14 | +#include <chrono> |
| 15 | +#include <cmath> |
| 16 | +#include <thread> |
| 17 | + |
| 18 | +namespace o2::aod::native |
| 19 | +{ |
| 20 | + |
| 21 | +namespace |
| 22 | +{ |
| 23 | +std::string labelOf(Replica const& r) |
| 24 | +{ |
| 25 | + return r.label.empty() ? r.url.substr(0, r.url.find('?')) : r.label; |
| 26 | +} |
| 27 | +} // namespace |
| 28 | + |
| 29 | +FailoverByteSource::FailoverByteSource(std::vector<Replica> replicas, Options const& options, Opener opener, Log log) |
| 30 | + : mReplicas(std::move(replicas)), mOptions(options), mOpener(std::move(opener)), mLog(std::move(log)) |
| 31 | +{ |
| 32 | + if (mReplicas.empty()) { |
| 33 | + throw IoError("(no replica)", "no replica to open"); |
| 34 | + } |
| 35 | + for (mIndex = 0; mIndex < mReplicas.size(); ++mIndex) { |
| 36 | + try { |
| 37 | + mSource = openReplica(mIndex); |
| 38 | + mSize = mSource->size(); |
| 39 | + mRemote = mSource->isRemote(); |
| 40 | + return; |
| 41 | + } catch (IoError const& e) { |
| 42 | + report("Could not open replica on " + labelOf(mReplicas[mIndex]) + ": " + e.what()); |
| 43 | + mErrors += " " + labelOf(mReplicas[mIndex]) + ": " + e.what() + ";"; |
| 44 | + } |
| 45 | + } |
| 46 | + throw IoError(labelOf(mReplicas.front()), "no replica could be opened:" + mErrors); |
| 47 | +} |
| 48 | + |
| 49 | +std::shared_ptr<ByteSource> FailoverByteSource::openReplica(size_t index) |
| 50 | +{ |
| 51 | + std::shared_ptr<ByteSource> source = mOpener(mReplicas[index].url, mOptions); |
| 52 | + if (mSource && source->size() != mSize) { |
| 53 | + throw IoError(source->url(), "size " + std::to_string(source->size()) + " differs from the " + std::to_string(mSize) + " bytes of the replica read so far"); |
| 54 | + } |
| 55 | + return source; |
| 56 | +} |
| 57 | + |
| 58 | +std::shared_ptr<ByteSource> FailoverByteSource::current() const |
| 59 | +{ |
| 60 | + std::lock_guard lock{mMutex}; |
| 61 | + return mSource; |
| 62 | +} |
| 63 | + |
| 64 | +std::string FailoverByteSource::replicaLabel() const |
| 65 | +{ |
| 66 | + std::lock_guard lock{mMutex}; |
| 67 | + return mReplicas[mIndex].label; |
| 68 | +} |
| 69 | + |
| 70 | +uint64_t FailoverByteSource::bytesRead() const |
| 71 | +{ |
| 72 | + return mRetiredBytes.load(std::memory_order_relaxed) + current()->bytesRead(); |
| 73 | +} |
| 74 | + |
| 75 | +uint64_t FailoverByteSource::readCalls() const |
| 76 | +{ |
| 77 | + return mRetiredCalls.load(std::memory_order_relaxed) + current()->readCalls(); |
| 78 | +} |
| 79 | + |
| 80 | +void FailoverByteSource::report(std::string const& message) const |
| 81 | +{ |
| 82 | + if (mLog) { |
| 83 | + mLog(message); |
| 84 | + } |
| 85 | +} |
| 86 | + |
| 87 | +void FailoverByteSource::readInto(uint64_t off, size_t len, void* dst) |
| 88 | +{ |
| 89 | + withFailover([&](ByteSource& s) { s.readInto(off, len, dst); }); |
| 90 | +} |
| 91 | + |
| 92 | +void FailoverByteSource::readvInto(std::span<ReadRequest> requests) |
| 93 | +{ |
| 94 | + withFailover([&](ByteSource& s) { s.readvInto(requests); }); |
| 95 | +} |
| 96 | + |
| 97 | +void FailoverByteSource::withFailover(std::function<void(ByteSource&)> const& read) |
| 98 | +{ |
| 99 | + while (true) { |
| 100 | + std::shared_ptr<ByteSource> source; |
| 101 | + uint64_t generation; |
| 102 | + { |
| 103 | + std::lock_guard lock{mMutex}; |
| 104 | + source = mSource; |
| 105 | + generation = mGeneration; |
| 106 | + } |
| 107 | + try { |
| 108 | + read(*source); |
| 109 | + return; |
| 110 | + } catch (IoError const& e) { |
| 111 | + std::lock_guard lock{mMutex}; |
| 112 | + if (generation != mGeneration) { |
| 113 | + continue; // another read already replaced the replica: try the new one |
| 114 | + } |
| 115 | + mRetiredBytes.fetch_add(source->bytesRead(), std::memory_order_relaxed); |
| 116 | + mRetiredCalls.fetch_add(source->readCalls(), std::memory_order_relaxed); |
| 117 | + std::string error = e.what(); |
| 118 | + // Reopen the same replica, after a growing delay, or move on to the next |
| 119 | + // one, until a replica opens. Other readers wait on mMutex meanwhile. |
| 120 | + while (true) { |
| 121 | + if (mAttempts < mOptions.readRetries) { |
| 122 | + double delay = mOptions.retryDelay * std::pow(2., mAttempts); |
| 123 | + ++mAttempts; |
| 124 | + report("Read error on " + labelOf(mReplicas[mIndex]) + ", retry " + std::to_string(mAttempts) + " of " + |
| 125 | + std::to_string(mOptions.readRetries) + " in " + std::to_string((int)std::lround(delay)) + " s: " + error); |
| 126 | + std::this_thread::sleep_for(std::chrono::duration<double>(delay)); |
| 127 | + } else { |
| 128 | + mErrors += " " + labelOf(mReplicas[mIndex]) + ": " + error + ";"; |
| 129 | + if (mIndex + 1 >= mReplicas.size()) { |
| 130 | + throw IoError(labelOf(mReplicas.front()), "no replica left to read from:" + mErrors); |
| 131 | + } |
| 132 | + ++mIndex; |
| 133 | + mAttempts = 0; |
| 134 | + report("Moving to replica on " + labelOf(mReplicas[mIndex]) + " after: " + error); |
| 135 | + } |
| 136 | + try { |
| 137 | + mSource = openReplica(mIndex); |
| 138 | + ++mGeneration; |
| 139 | + break; |
| 140 | + } catch (IoError const& reopen) { |
| 141 | + error = reopen.what(); |
| 142 | + if (mAttempts == 0) { |
| 143 | + mAttempts = mOptions.readRetries; // a replica that does not open is not retried |
| 144 | + } |
| 145 | + } |
| 146 | + } |
| 147 | + } |
| 148 | + } |
| 149 | +} |
| 150 | + |
| 151 | +} // namespace o2::aod::native |
0 commit comments