Allow only 1 job queue slot for acquiring inbound ledger.

* Log when duplicate concurrent inbound ledger are filtered.
* RAII for containers that track concurrent inbound ledger.
* Comment on when to asynchronously acquire inbound ledgers, which
   is possible to be always OK, but should have further review.
* Other small logging changes

Co-authored-by: Ed Hennis <ed@ripple.com>
This commit is contained in:
Mark Travis
2024-08-25 19:10:07 -07:00
committed by Ed Hennis
parent 00ed7c9424
commit 7741483894
7 changed files with 68 additions and 8 deletions

View File

@@ -135,8 +135,12 @@ RCLConsensus::Adaptor::acquireLedger(LedgerHash const& hash)
acquiringLedger_ = hash;
app_.getJobQueue().addJob(
jtADVANCE, "getConsensusLedger", [id = hash, &app = app_]() {
app.getInboundLedgers().acquire(
jtADVANCE,
"getConsensusLedger1",
[id = hash, &app = app_, this]() {
JLOG(j_.debug())
<< "JOB advanceLedger getConsensusLedger1 started";
app.getInboundLedgers().acquireAsync(
id, 0, InboundLedger::Reason::CONSENSUS);
});
}

View File

@@ -142,8 +142,10 @@ RCLValidationsAdaptor::acquire(LedgerHash const& hash)
Application* pApp = &app_;
app_.getJobQueue().addJob(
jtADVANCE, "getConsensusLedger", [pApp, hash]() {
pApp->getInboundLedgers().acquire(
jtADVANCE, "getConsensusLedger2", [pApp, hash, this]() {
JLOG(j_.debug())
<< "JOB advanceLedger getConsensusLedger2 started";
pApp->getInboundLedgers().acquireAsync(
hash, 0, InboundLedger::Reason::CONSENSUS);
});
return std::nullopt;

View File

@@ -23,6 +23,7 @@
#include <ripple/app/ledger/InboundLedger.h>
#include <ripple/protocol/RippleLedgerHash.h>
#include <memory>
#include <set>
namespace ripple {
@@ -37,11 +38,20 @@ public:
virtual ~InboundLedgers() = default;
// VFALCO TODO Should this be called findOrAdd ?
//
// Callers should use this if they possibly need an authoritative
// response immediately.
virtual std::shared_ptr<Ledger const>
acquire(uint256 const& hash, std::uint32_t seq, InboundLedger::Reason) = 0;
// Callers should use this if they are known to be executing on the Job
// 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;
virtual std::shared_ptr<InboundLedger>
find(LedgerHash const& hash) = 0;

View File

@@ -536,7 +536,7 @@ InboundLedger::trigger(std::shared_ptr<Peer> const& peer, TriggerReason reason)
return;
}
if (auto stream = journal_.trace())
if (auto stream = journal_.debug())
{
stream << "Trigger acquiring ledger " << hash_;
if (peer)

View File

@@ -29,6 +29,7 @@
#include <ripple/core/JobQueue.h>
#include <ripple/nodestore/DatabaseShard.h>
#include <ripple/protocol/jss.h>
#include <exception>
#include <memory>
#include <mutex>
#include <vector>
@@ -149,6 +150,37 @@ public:
return ledger;
}
void
acquireAsync(
uint256 const& hash,
std::uint32_t seq,
InboundLedger::Reason reason) override
{
std::unique_lock lock(acquiresMutex_);
try
{
if (pendingAcquires_.contains(hash))
return;
pendingAcquires_.insert(hash);
lock.unlock();
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;
}
lock.lock();
pendingAcquires_.erase(hash);
}
std::shared_ptr<InboundLedger>
find(uint256 const& hash) override
{
@@ -441,6 +473,9 @@ private:
beast::insight::Counter mCounter;
std::unique_ptr<PeerSetBuilder> mPeerSetBuilder;
std::set<uint256> pendingAcquires_;
std::mutex acquiresMutex_;
};
//------------------------------------------------------------------------------

View File

@@ -1730,7 +1730,8 @@ NetworkOPsImp::checkLastClosedLedger(
}
JLOG(m_journal.warn()) << "We are not running on the consensus ledger";
JLOG(m_journal.info()) << "Our LCL: " << getJson({*ourClosed, {}});
JLOG(m_journal.info()) << "Our LCL: " << ourClosed->info().hash
<< getJson({*ourClosed, {}});
JLOG(m_journal.info()) << "Net LCL " << closedLedger;
if ((mMode == OperatingMode::TRACKING) || (mMode == OperatingMode::FULL))

View File

@@ -106,6 +106,14 @@ public:
return {};
}
virtual void
acquireAsync(
uint256 const& hash,
std::uint32_t seq,
InboundLedger::Reason reason) override
{
}
virtual std::shared_ptr<InboundLedger>
find(LedgerHash const& hash) override
{