From f44dfc89cca024ff823dee56892c8c9f555f8b4f Mon Sep 17 00:00:00 2001 From: Ed Hennis Date: Fri, 10 Apr 2026 10:31:21 -0400 Subject: [PATCH] refactor: acquireAsync will dispatch a job, not the other way around - Improve job queue collision checks and logging - Improve logging related to ledger acquisition and operating mode changes - Class "CanProcess" to keep track of processing of distinct items --- include/xrpl/basics/CanProcess.h | 139 +++++++++++++++ include/xrpl/core/HashRouter.h | 2 +- include/xrpl/protocol/LedgerHeader.h | 2 + include/xrpl/server/NetworkOPs.h | 2 +- src/test/app/LedgerReplay_test.cpp | 7 +- src/test/basics/CanProcess_test.cpp | 165 ++++++++++++++++++ src/xrpld/app/consensus/RCLConsensus.cpp | 8 +- src/xrpld/app/consensus/RCLValidations.cpp | 8 +- src/xrpld/app/ledger/InboundLedgers.h | 7 +- src/xrpld/app/ledger/detail/InboundLedger.cpp | 9 +- .../app/ledger/detail/InboundLedgers.cpp | 62 ++++--- src/xrpld/app/ledger/detail/LedgerMaster.cpp | 5 +- .../app/ledger/detail/TimeoutCounter.cpp | 8 +- src/xrpld/app/ledger/detail/TimeoutCounter.h | 2 + src/xrpld/app/misc/NetworkOPs.cpp | 71 ++++---- 15 files changed, 415 insertions(+), 82 deletions(-) create mode 100644 include/xrpl/basics/CanProcess.h create mode 100644 src/test/basics/CanProcess_test.cpp diff --git a/include/xrpl/basics/CanProcess.h b/include/xrpl/basics/CanProcess.h new file mode 100644 index 0000000000..ab14523b47 --- /dev/null +++ b/include/xrpl/basics/CanProcess.h @@ -0,0 +1,139 @@ +//------------------------------------------------------------------------------ +/* + This file is part of rippled: https://github.com/ripple/rippled + Copyright (c) 2024 Ripple Labs Inc. + + Permission to use, copy, modify, and/or distribute this software for any + purpose with or without fee is hereby granted, provided that the above + copyright notice and this permission notice appear in all copies. + + THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES + WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF + MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR + ANY SPECIAL , DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES + WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN + ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF + OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. +*/ +//============================================================================== + +#ifndef RIPPLE_BASICS_CANPROCESS_H_INCLUDED +#define RIPPLE_BASICS_CANPROCESS_H_INCLUDED + +#include +#include +#include + +/** RAII class to check if an Item is already being processed on another thread, + * as indicated by it's presence in a Collection. + * + * If the Item is not in the Collection, it will be added under lock in the + * ctor, and removed under lock in the dtor. The object will be considered + * "usable" and evaluate to `true`. + * + * If the Item is in the Collection, no changes will be made to the collection, + * and the CanProcess object will be considered "unusable". + * + * It's up to the caller to decide what "usable" and "unusable" mean. (e.g. + * Process or skip a block of code, or set a flag.) + * + * The current use is to avoid lock contention that would be involved in + * processing something associated with the Item. + * + * Examples: + * + * void IncomingLedgers::acquireAsync(LedgerHash const& hash, ...) + * { + * if (CanProcess check{acquiresMutex_, pendingAcquires_, hash}) + * { + * acquire(hash, ...); + * } + * } + * + * bool + * NetworkOPsImp::recvValidation( + * std::shared_ptr const& val, + * std::string const& source) + * { + * CanProcess check( + * validationsMutex_, pendingValidations_, val->getLedgerHash()); + * BypassAccept bypassAccept = + * check ? BypassAccept::no : BypassAccept::yes; + * handleNewValidation(app_, val, source, bypassAccept, m_journal); + * } + * + */ +class CanProcess +{ +public: + template + CanProcess(Mutex& mtx, Collection& collection, Item const& item) + : cleanup_(insert(mtx, collection, item)) + { + } + + ~CanProcess() + { + if (cleanup_) + cleanup_(); + } + + CanProcess(CanProcess const&) = delete; + + CanProcess& + operator=(CanProcess const&) = delete; + + explicit + operator bool() const + { + return static_cast(cleanup_); + } + +private: + template + std::function + doInsert(Mutex& mtx, Collection& collection, Item const& item) + { + std::unique_lock lock(mtx); + // TODO: Use structured binding once LLVM 16 is the minimum supported + // version. See also: https://github.com/llvm/llvm-project/issues/48582 + // https://github.com/llvm/llvm-project/commit/127bf44385424891eb04cff8e52d3f157fc2cb7c + auto const insertResult = collection.insert(item); + auto const it = insertResult.first; + if (!insertResult.second) + return {}; + if constexpr (useIterator) + return [&, it]() { + std::unique_lock lock(mtx); + collection.erase(it); + }; + else + return [&]() { + std::unique_lock lock(mtx); + collection.erase(item); + }; + } + + // Generic insert() function doesn't use iterators because they may get + // invalidated + template + std::function + insert(Mutex& mtx, Collection& collection, Item const& item) + { + return doInsert(mtx, collection, item); + } + + // Specialize insert() for std::set, which does not invalidate iterators for + // insert and erase + template + std::function + insert(Mutex& mtx, std::set& collection, Item const& item) + { + return doInsert(mtx, collection, item); + } + + // If set, then the item is "usable" + std::function cleanup_; +}; + +#endif diff --git a/include/xrpl/core/HashRouter.h b/include/xrpl/core/HashRouter.h index b4f07f6dc0..9ede7e26ea 100644 --- a/include/xrpl/core/HashRouter.h +++ b/include/xrpl/core/HashRouter.h @@ -199,7 +199,7 @@ public: /** Add a suppression peer and get message's relay status. * Return pair: - * element 1: true if the peer is added. + * element 1: true if the key is added. * element 2: optional is seated to the relay time point or * is unseated if has not relayed yet. */ std::pair> diff --git a/include/xrpl/protocol/LedgerHeader.h b/include/xrpl/protocol/LedgerHeader.h index 6e22ad268d..62050f83fa 100644 --- a/include/xrpl/protocol/LedgerHeader.h +++ b/include/xrpl/protocol/LedgerHeader.h @@ -35,6 +35,8 @@ struct LedgerHeader // If validated is false, it means "not yet validated." // Once validated is true, it will never be set false at a later time. + // NOTE: If you are accessing this directly, you are probably doing it + // wrong. Use LedgerMaster::isValidated(). // VFALCO TODO Make this not mutable bool mutable validated = false; bool accepted = false; diff --git a/include/xrpl/server/NetworkOPs.h b/include/xrpl/server/NetworkOPs.h index 16ec4a4ec0..cf5f6081a3 100644 --- a/include/xrpl/server/NetworkOPs.h +++ b/include/xrpl/server/NetworkOPs.h @@ -185,7 +185,7 @@ public: virtual bool isFull() = 0; virtual void - setMode(OperatingMode om) = 0; + setMode(OperatingMode om, char const* reason) = 0; virtual bool isBlocked() = 0; virtual bool diff --git a/src/test/app/LedgerReplay_test.cpp b/src/test/app/LedgerReplay_test.cpp index f9ab08e900..61ac2a9404 100644 --- a/src/test/app/LedgerReplay_test.cpp +++ b/src/test/app/LedgerReplay_test.cpp @@ -85,7 +85,12 @@ public: } virtual void - acquireAsync(uint256 const& hash, std::uint32_t seq, InboundLedger::Reason reason) override + acquireAsync( + JobType type, + std::string const& name, + uint256 const& hash, + std::uint32_t seq, + InboundLedger::Reason reason) override { } diff --git a/src/test/basics/CanProcess_test.cpp b/src/test/basics/CanProcess_test.cpp new file mode 100644 index 0000000000..0c13fb24ce --- /dev/null +++ b/src/test/basics/CanProcess_test.cpp @@ -0,0 +1,165 @@ +//------------------------------------------------------------------------------ +/* + This file is part of rippled: https://github.com/ripple/rippled + Copyright (c) 2012-2016 Ripple Labs Inc. + + Permission to use, copy, modify, and/or distribute this software for any + purpose with or without fee is hereby granted, provided that the above + copyright notice and this permission notice appear in all copies. + + THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES + WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF + MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR + ANY SPECIAL , DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES + WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN + ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF + OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. +*/ +//============================================================================== + +#include +#include + +#include + +namespace ripple { +namespace test { + +struct CanProcess_test : beast::unit_test::suite +{ + template + void + test( + std::string const& name, + Mutex& mtx, + Collection& collection, + std::vector const& items) + { + testcase(name); + + if (!BEAST_EXPECT(!items.empty())) + return; + if (!BEAST_EXPECT(collection.empty())) + return; + + // CanProcess objects can't be copied or moved. To make that easier, + // store shared_ptrs + std::vector> trackers; + // Fill up the vector with two CanProcess for each Item. The first + // inserts the item into the collection and is "good". The second does + // not and is "bad". + for (int i = 0; i < items.size(); ++i) + { + { + auto const& good = + trackers.emplace_back(std::make_shared(mtx, collection, items[i])); + BEAST_EXPECT(*good); + } + BEAST_EXPECT(trackers.size() == (2 * i) + 1); + BEAST_EXPECT(collection.size() == i + 1); + { + auto const& bad = + trackers.emplace_back(std::make_shared(mtx, collection, items[i])); + BEAST_EXPECT(!*bad); + } + BEAST_EXPECT(trackers.size() == 2 * (i + 1)); + BEAST_EXPECT(collection.size() == i + 1); + } + BEAST_EXPECT(collection.size() == items.size()); + // Now remove the items from the vector two at a time, and + // try to get another CanProcess for that item. + for (int i = 0; i < items.size(); ++i) + { + // Remove the "bad" one in the second position + // This will have no effect on the collection + { + auto const iter = trackers.begin() + 1; + BEAST_EXPECT(!**iter); + trackers.erase(iter); + } + BEAST_EXPECT(trackers.size() == (2 * items.size()) - 1); + BEAST_EXPECT(collection.size() == items.size()); + { + // Append a new "bad" one + auto const& bad = + trackers.emplace_back(std::make_shared(mtx, collection, items[i])); + BEAST_EXPECT(!*bad); + } + BEAST_EXPECT(trackers.size() == 2 * items.size()); + BEAST_EXPECT(collection.size() == items.size()); + + // Remove the "good" one from the front + { + auto const iter = trackers.begin(); + BEAST_EXPECT(**iter); + trackers.erase(iter); + } + BEAST_EXPECT(trackers.size() == (2 * items.size()) - 1); + BEAST_EXPECT(collection.size() == items.size() - 1); + { + // Append a new "good" one + auto const& good = + trackers.emplace_back(std::make_shared(mtx, collection, items[i])); + BEAST_EXPECT(*good); + } + BEAST_EXPECT(trackers.size() == 2 * items.size()); + BEAST_EXPECT(collection.size() == items.size()); + } + // Now remove them all two at a time + for (int i = items.size() - 1; i >= 0; --i) + { + // Remove the "bad" one from the front + { + auto const iter = trackers.begin(); + BEAST_EXPECT(!**iter); + trackers.erase(iter); + } + BEAST_EXPECT(trackers.size() == (2 * i) + 1); + BEAST_EXPECT(collection.size() == i + 1); + // Remove the "good" one now in front + { + auto const iter = trackers.begin(); + BEAST_EXPECT(**iter); + trackers.erase(iter); + } + BEAST_EXPECT(trackers.size() == 2 * i); + BEAST_EXPECT(collection.size() == i); + } + BEAST_EXPECT(trackers.empty()); + BEAST_EXPECT(collection.empty()); + } + + void + run() override + { + { + std::mutex m; + std::set collection; + std::vector const items{1, 2, 3, 4, 5}; + test("set of int", m, collection, items); + } + { + std::mutex m; + std::set collection; + std::vector const items{"one", "two", "three", "four", "five"}; + test("set of string", m, collection, items); + } + { + std::mutex m; + std::unordered_set collection; + std::vector const items{'1', '2', '3', '4', '5'}; + test("unorderd_set of char", m, collection, items); + } + { + std::mutex m; + std::unordered_set collection; + std::vector const items{100u, 1000u, 150u, 4u, 0u}; + test("unordered_set of uint64_t", m, collection, items); + } + } +}; + +BEAST_DEFINE_TESTSUITE(CanProcess, ripple_basics, ripple); + +} // namespace test +} // namespace ripple diff --git a/src/xrpld/app/consensus/RCLConsensus.cpp b/src/xrpld/app/consensus/RCLConsensus.cpp index b7b0919aad..0ae23aa81e 100644 --- a/src/xrpld/app/consensus/RCLConsensus.cpp +++ b/src/xrpld/app/consensus/RCLConsensus.cpp @@ -105,10 +105,8 @@ RCLConsensus::Adaptor::acquireLedger(LedgerHash const& hash) // Tell the ledger acquire system that we need the consensus ledger acquiringLedger_ = hash; - app_.getJobQueue().addJob(jtADVANCE, "GetConsL1", [id = hash, &app = app_, this]() { - JLOG(j_.debug()) << "JOB advanceLedger getConsensusLedger1 started"; - app.getInboundLedgers().acquireAsync(id, 0, InboundLedger::Reason::CONSENSUS); - }); + app_.getInboundLedgers().acquireAsync( + jtADVANCE, "GetConsL1", hash, 0, InboundLedger::Reason::CONSENSUS); } return std::nullopt; } @@ -998,7 +996,7 @@ void RCLConsensus::Adaptor::updateOperatingMode(std::size_t const positions) const { if ((positions == 0u) && app_.getOPs().isFull()) - app_.getOPs().setMode(OperatingMode::CONNECTED); + app_.getOPs().setMode(OperatingMode::CONNECTED, "updateOperatingMode: no positions"); } void diff --git a/src/xrpld/app/consensus/RCLValidations.cpp b/src/xrpld/app/consensus/RCLValidations.cpp index 7bc16f194e..89183b079f 100644 --- a/src/xrpld/app/consensus/RCLValidations.cpp +++ b/src/xrpld/app/consensus/RCLValidations.cpp @@ -116,12 +116,8 @@ RCLValidationsAdaptor::acquire(LedgerHash const& hash) { JLOG(j_.warn()) << "Need validated ledger for preferred ledger analysis " << hash; - Application* pApp = &app_; - - app_.getJobQueue().addJob(jtADVANCE, "GetConsL2", [pApp, hash, this]() { - JLOG(j_.debug()) << "JOB advanceLedger getConsensusLedger2 started"; - pApp->getInboundLedgers().acquireAsync(hash, 0, InboundLedger::Reason::CONSENSUS); - }); + app_.getInboundLedgers().acquireAsync( + jtADVANCE, "GetConsL2", hash, 0, InboundLedger::Reason::CONSENSUS); return std::nullopt; } diff --git a/src/xrpld/app/ledger/InboundLedgers.h b/src/xrpld/app/ledger/InboundLedgers.h index ecac6e07e4..20d3f9fa27 100644 --- a/src/xrpld/app/ledger/InboundLedgers.h +++ b/src/xrpld/app/ledger/InboundLedgers.h @@ -26,7 +26,12 @@ public: // Queue. TODO review whether all callers of acquire() can use this // instead. Inbound ledger acquisition is asynchronous anyway. virtual void - acquireAsync(uint256 const& hash, std::uint32_t seq, InboundLedger::Reason reason) = 0; + acquireAsync( + JobType type, + std::string const& name, + uint256 const& hash, + std::uint32_t seq, + InboundLedger::Reason reason) = 0; virtual std::shared_ptr find(LedgerHash const& hash) = 0; diff --git a/src/xrpld/app/ledger/detail/InboundLedger.cpp b/src/xrpld/app/ledger/detail/InboundLedger.cpp index 2402b5b561..36c41e1d27 100644 --- a/src/xrpld/app/ledger/detail/InboundLedger.cpp +++ b/src/xrpld/app/ledger/detail/InboundLedger.cpp @@ -353,7 +353,14 @@ InboundLedger::onTimer(bool wasProgress, ScopedLockType&) if (!wasProgress) { - checkLocal(); + if (checkLocal()) + { + // Done. Something else (probably consensus) built the ledger + // locally while waiting for data (or possibly before requesting) + XRPL_ASSERT(isDone(), "ripple::InboundLedger::onTimer : done"); + JLOG(journal_.info()) << "Finished while waiting " << hash_; + return; + } mByHash = true; diff --git a/src/xrpld/app/ledger/detail/InboundLedgers.cpp b/src/xrpld/app/ledger/detail/InboundLedgers.cpp index f147a35ca4..bbed8f42b8 100644 --- a/src/xrpld/app/ledger/detail/InboundLedgers.cpp +++ b/src/xrpld/app/ledger/detail/InboundLedgers.cpp @@ -2,6 +2,7 @@ #include #include +#include #include #include #include @@ -58,12 +59,15 @@ public: (reason != InboundLedger::Reason::CONSENSUS)) return {}; + std::stringstream ss; + bool isNew = true; std::shared_ptr inbound; { ScopedLockType sl(mLock); if (stopping_) { + JLOG(j_.debug()) << "Abort(stopping): " << ss.str(); return {}; } @@ -82,47 +86,61 @@ public: ++mCounter; } } + ss << " IsNew: " << (isNew ? "true" : "false"); if (inbound->isFailed()) + { + JLOG(j_.debug()) << "Abort(failed): " << ss.str(); return {}; + } if (!isNew) inbound->update(seq); if (!inbound->isComplete()) + { + JLOG(j_.debug()) << "InProgress: " << ss.str(); return {}; + } + JLOG(j_.debug()) << "Complete: " << ss.str(); return inbound->getLedger(); }; using namespace std::chrono_literals; - std::shared_ptr ledger = - perf::measureDurationAndLog(doAcquire, "InboundLedgersImp::acquire", 500ms, j_); - - return ledger; + return perf::measureDurationAndLog(doAcquire, "InboundLedgersImp::acquire", 500ms, j_); } void - acquireAsync(uint256 const& hash, std::uint32_t seq, InboundLedger::Reason reason) override + acquireAsync( + JobType type, + std::string const& name, + uint256 const& hash, + std::uint32_t seq, + InboundLedger::Reason reason) override { - std::unique_lock lock(acquiresMutex_); - try + if (auto check = std::make_shared(acquiresMutex_, pendingAcquires_, hash); + *check) { - if (pendingAcquires_.contains(hash)) - return; - pendingAcquires_.insert(hash); - scope_unlock const unlock(lock); - acquire(hash, seq, reason); + app_.getJobQueue().addJob(type, name, [check, name, hash, seq, reason, this]() { + JLOG(j_.debug()) << "JOB acquireAsync " << name << " started "; + try + { + acquire(hash, seq, reason); + } + catch (std::exception const& e) + { + JLOG(j_.warn()) << "Exception thrown for acquiring new " + "inbound ledger " + << hash << ": " << e.what(); + } + catch (...) + { + JLOG(j_.warn()) << "Unknown exception thrown for acquiring new " + "inbound ledger " + << hash; + } + }); } - catch (std::exception const& e) - { - JLOG(j_.warn()) << "Exception thrown for acquiring new inbound ledger " << hash << ": " - << e.what(); - } - catch (...) - { - JLOG(j_.warn()) << "Unknown exception thrown for acquiring new inbound ledger " << hash; - } - pendingAcquires_.erase(hash); } std::shared_ptr diff --git a/src/xrpld/app/ledger/detail/LedgerMaster.cpp b/src/xrpld/app/ledger/detail/LedgerMaster.cpp index b9305b743f..d9718b004b 100644 --- a/src/xrpld/app/ledger/detail/LedgerMaster.cpp +++ b/src/xrpld/app/ledger/detail/LedgerMaster.cpp @@ -924,8 +924,9 @@ LedgerMaster::checkAccept(std::shared_ptr const& ledger) return; } - JLOG(m_journal.info()) << "Advancing accepted ledger to " << ledger->header().seq - << " with >= " << minVal << " validations"; + JLOG(m_journal.info()) << "Advancing accepted ledger to " << ledger->header().seq << " (" + << to_short_string(ledger->header().hash) << ") with >= " << minVal + << " validations"; ledger->setValidated(); ledger->setFull(); diff --git a/src/xrpld/app/ledger/detail/TimeoutCounter.cpp b/src/xrpld/app/ledger/detail/TimeoutCounter.cpp index 216771e60d..29d9a228a7 100644 --- a/src/xrpld/app/ledger/detail/TimeoutCounter.cpp +++ b/src/xrpld/app/ledger/detail/TimeoutCounter.cpp @@ -13,7 +13,8 @@ TimeoutCounter::TimeoutCounter( QueueJobParameter&& jobParameter, beast::Journal journal) : app_(app) - , journal_(journal) + , sink_(journal, to_short_string(hash) + " ") + , journal_(sink_) , hash_(hash) , timerInterval_(interval) , queueJobParameter_(std::move(jobParameter)) @@ -29,6 +30,7 @@ TimeoutCounter::setTimer(ScopedLockType& sl) { if (isDone()) return; + JLOG(journal_.debug()) << "Setting timer for " << timerInterval_.count() << "ms"; timer_.expires_after(timerInterval_); timer_.async_wait([wptr = pmDowncast()](boost::system::error_code const& ec) { if (ec == boost::asio::error::operation_aborted) @@ -36,6 +38,10 @@ TimeoutCounter::setTimer(ScopedLockType& sl) if (auto ptr = wptr.lock()) { + JLOG(ptr->journal_.debug()) + << "timer: ec: " << ec + << " (operation_aborted: " << boost::asio::error::operation_aborted << " - " + << (ec == boost::asio::error::operation_aborted ? "aborted" : "other") << ")"; ScopedLockType sl(ptr->mtx_); ptr->queueJob(sl); } diff --git a/src/xrpld/app/ledger/detail/TimeoutCounter.h b/src/xrpld/app/ledger/detail/TimeoutCounter.h index a7e4c043be..055932cbee 100644 --- a/src/xrpld/app/ledger/detail/TimeoutCounter.h +++ b/src/xrpld/app/ledger/detail/TimeoutCounter.h @@ -3,6 +3,7 @@ #include #include +#include #include #include @@ -103,6 +104,7 @@ protected: // Used in this class for access to boost::asio::io_context and // xrpl::Overlay. Used in subtypes for the kitchen sink. Application& app_; + beast::WrappedSink sink_; beast::Journal journal_; mutable std::recursive_mutex mtx_; diff --git a/src/xrpld/app/misc/NetworkOPs.cpp b/src/xrpld/app/misc/NetworkOPs.cpp index e39230efdb..705d08f13b 100644 --- a/src/xrpld/app/misc/NetworkOPs.cpp +++ b/src/xrpld/app/misc/NetworkOPs.cpp @@ -30,10 +30,10 @@ #include #include +#include #include #include #include -#include #include #include #include @@ -401,7 +401,7 @@ public: isFull() override; void - setMode(OperatingMode om) override; + setMode(OperatingMode om, char const* reason) override; bool isBlocked() override; @@ -839,7 +839,7 @@ NetworkOPsImp::strOperatingMode(bool const admin /* = false */) const inline void NetworkOPsImp::setStandAlone() { - setMode(OperatingMode::FULL); + setMode(OperatingMode::FULL, "setStandAlone"); } inline void @@ -982,7 +982,7 @@ NetworkOPsImp::processHeartbeatTimer() { if (mMode != OperatingMode::DISCONNECTED) { - setMode(OperatingMode::DISCONNECTED); + setMode(OperatingMode::DISCONNECTED, "Heartbeat: insufficient peers"); std::stringstream ss; ss << "Node count (" << numPeers << ") has fallen " << "below required minimum (" << minPeerCount_ << ")."; @@ -1006,7 +1006,7 @@ NetworkOPsImp::processHeartbeatTimer() if (mMode == OperatingMode::DISCONNECTED) { - setMode(OperatingMode::CONNECTED); + setMode(OperatingMode::CONNECTED, "Heartbeat: sufficient peers"); JLOG(m_journal.info()) << "Node count (" << numPeers << ") is sufficient."; CLOG(clog.ss()) << "setting mode to CONNECTED based on " << numPeers << " peers. "; } @@ -1017,11 +1017,11 @@ NetworkOPsImp::processHeartbeatTimer() CLOG(clog.ss()) << "mode: " << strOperatingMode(origMode, true); if (mMode == OperatingMode::SYNCING) { - setMode(OperatingMode::SYNCING); + setMode(OperatingMode::SYNCING, "Heartbeat: check syncing"); } else if (mMode == OperatingMode::CONNECTED) { - setMode(OperatingMode::CONNECTED); + setMode(OperatingMode::CONNECTED, "Heartbeat: check connected"); } auto newMode = mMode.load(); if (origMode != newMode) @@ -1726,7 +1726,7 @@ void NetworkOPsImp::setAmendmentBlocked() { amendmentBlocked_ = true; - setMode(OperatingMode::CONNECTED); + setMode(OperatingMode::CONNECTED, "setAmendmentBlocked"); } inline bool @@ -1757,7 +1757,7 @@ void NetworkOPsImp::setUNLBlocked() { unlBlocked_ = true; - setMode(OperatingMode::CONNECTED); + setMode(OperatingMode::CONNECTED, "setUNLBlocked"); } inline void @@ -1857,7 +1857,7 @@ NetworkOPsImp::checkLastClosedLedger(Overlay::PeerSequence const& peerList, uint if ((mMode == OperatingMode::TRACKING) || (mMode == OperatingMode::FULL)) { - setMode(OperatingMode::CONNECTED); + setMode(OperatingMode::CONNECTED, "check LCL: not on consensus ledger"); } if (consensus) @@ -1945,8 +1945,8 @@ NetworkOPsImp::beginConsensus( // this shouldn't happen unless we jump ledgers if (mMode == OperatingMode::FULL) { - JLOG(m_journal.warn()) << "Don't have LCL, going to tracking"; - setMode(OperatingMode::TRACKING); + JLOG(m_journal.warn()) << "beginConsensus Don't have LCL, going to tracking"; + setMode(OperatingMode::TRACKING, "beginConsensus: No LCL"); CLOG(clog) << "beginConsensus Don't have LCL, going to tracking. "; } @@ -2074,7 +2074,7 @@ NetworkOPsImp::endConsensus(std::unique_ptr const& clog) // validations we have for LCL. If the ledger is good enough, go to // TRACKING - TODO if (!needNetworkLedger_) - setMode(OperatingMode::TRACKING); + setMode(OperatingMode::TRACKING, "endConsensus: check tracking"); } if (((mMode == OperatingMode::CONNECTED) || (mMode == OperatingMode::TRACKING)) && @@ -2087,7 +2087,7 @@ NetworkOPsImp::endConsensus(std::unique_ptr const& clog) if (registry_.get().getTimeKeeper().now() < (current->header().parentCloseTime + 2 * current->header().closeTimeResolution)) { - setMode(OperatingMode::FULL); + setMode(OperatingMode::FULL, "endConsensus: check full"); } } @@ -2099,7 +2099,7 @@ NetworkOPsImp::consensusViewChange() { if ((mMode == OperatingMode::FULL) || (mMode == OperatingMode::TRACKING)) { - setMode(OperatingMode::CONNECTED); + setMode(OperatingMode::CONNECTED, "consensusViewChange"); } } @@ -2403,7 +2403,7 @@ NetworkOPsImp::pubPeerStatus(std::function const& func) } void -NetworkOPsImp::setMode(OperatingMode om) +NetworkOPsImp::setMode(OperatingMode om, char const* reason) { using namespace std::chrono_literals; if (om == OperatingMode::CONNECTED) @@ -2423,11 +2423,12 @@ NetworkOPsImp::setMode(OperatingMode om) if (mMode == om) return; + auto const sink = om < mMode ? m_journal.warn() : m_journal.info(); mMode = om; accounting_.mode(om); - JLOG(m_journal.info()) << "STATE->" << strOperatingMode(); + JLOG(sink) << "STATE->" << strOperatingMode() << " - " << reason; pubServer(); } @@ -2436,36 +2437,24 @@ NetworkOPsImp::recvValidation(std::shared_ptr const& val, std::str { JLOG(m_journal.trace()) << "recvValidation " << val->getLedgerHash() << " from " << source; - std::unique_lock lock(validationsMutex_); - BypassAccept bypassAccept = BypassAccept::no; - try { - if (pendingValidations_.contains(val->getLedgerHash())) + CanProcess const check(validationsMutex_, pendingValidations_, val->getLedgerHash()); + try { - bypassAccept = BypassAccept::yes; + BypassAccept bypassAccept = check ? BypassAccept::no : BypassAccept::yes; + handleNewValidation(registry_.app(), val, source, bypassAccept, m_journal); } - else + catch (std::exception const& e) { - pendingValidations_.insert(val->getLedgerHash()); + JLOG(m_journal.warn()) << "Exception thrown for handling new validation " + << val->getLedgerHash() << ": " << e.what(); + } + catch (...) + { + JLOG(m_journal.warn()) + << "Unknown exception thrown for handling new validation " << val->getLedgerHash(); } - scope_unlock const unlock(lock); - handleNewValidation(registry_.get().getApp(), val, source, bypassAccept, m_journal); } - catch (std::exception const& e) - { - JLOG(m_journal.warn()) << "Exception thrown for handling new validation " - << val->getLedgerHash() << ": " << e.what(); - } - catch (...) - { - JLOG(m_journal.warn()) << "Unknown exception thrown for handling new validation " - << val->getLedgerHash(); - } - if (bypassAccept == BypassAccept::no) - { - pendingValidations_.erase(val->getLedgerHash()); - } - lock.unlock(); pubValidation(val);