Proposed version 2.2.2

src/ripple/app/consensus/RCLConsensus.cpp

@@ -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);
                 });
         }

src/ripple/app/consensus/RCLValidations.cpp

@@ -25,6 +25,7 @@
 #include <ripple/app/misc/NetworkOPs.h>
 #include <ripple/app/misc/ValidatorList.h>
 #include <ripple/basics/Log.h>
+#include <ripple/basics/PerfLog.h>
 #include <ripple/basics/StringUtilities.h>
 #include <ripple/basics/chrono.h>
 #include <ripple/consensus/LedgerTiming.h>
@@ -126,7 +127,13 @@ RCLValidationsAdaptor::now() const
 std::optional<RCLValidatedLedger>
 RCLValidationsAdaptor::acquire(LedgerHash const& hash)
 {
-    auto ledger = app_.getLedgerMaster().getLedgerByHash(hash);
+    using namespace std::chrono_literals;
+    auto ledger = perf::measureDurationAndLog(
+        [&]() { return app_.getLedgerMaster().getLedgerByHash(hash); },
+        "getLedgerByHash",
+        10ms,
+        j_);
+
     if (!ledger)
     {
         JLOG(j_.debug())
@@ -135,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;
@@ -152,7 +161,9 @@ void
 handleNewValidation(
     Application& app,
     std::shared_ptr<STValidation> const& val,
-    std::string const& source)
+    std::string const& source,
+    BypassAccept const bypassAccept,
+    std::optional<beast::Journal> j)
 {
     auto const& signingKey = val->getSignerPublic();
     auto const& hash = val->getLedgerHash();
@@ -177,7 +188,21 @@ handleNewValidation(
     if (outcome == ValStatus::current)
     {
         if (val->isTrusted())
-            app.getLedgerMaster().checkAccept(hash, seq);
+        {
+            if (bypassAccept == BypassAccept::yes)
+            {
+                assert(j.has_value());
+                if (j.has_value())
+                {
+                    JLOG(j->trace()) << "Bypassing checkAccept for validation "
+                                     << val->getLedgerHash();
+                }
+            }
+            else
+            {
+                app.getLedgerMaster().checkAccept(hash, seq);
+            }
+        }
         return;
     }
 

src/ripple/app/consensus/RCLValidations.h

@@ -25,12 +25,16 @@
 #include <ripple/protocol/Protocol.h>
 #include <ripple/protocol/RippleLedgerHash.h>
 #include <ripple/protocol/STValidation.h>
+#include <optional>
+#include <set>
 #include <vector>
 
 namespace ripple {
 
 class Application;
 
+enum class BypassAccept : bool { no = false, yes };
+
 /** Wrapper over STValidation for generic Validation code
 
     Wraps an STValidation for compatibility with the generic validation code.
@@ -248,7 +252,9 @@ void
 handleNewValidation(
     Application& app,
     std::shared_ptr<STValidation> const& val,
-    std::string const& source);
+    std::string const& source,
+    BypassAccept const bypassAccept = BypassAccept::no,
+    std::optional<beast::Journal> j = std::nullopt);
 
 }  // namespace ripple
 

src/ripple/app/ledger/InboundLedgers.h

@@ -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 @@ class InboundLedgers
 
     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;
 

src/ripple/app/ledger/impl/InboundLedger.cpp

@@ -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)

src/ripple/app/ledger/impl/InboundLedgers.cpp

@@ -23,11 +23,13 @@
 #include <ripple/app/misc/NetworkOPs.h>
 #include <ripple/basics/DecayingSample.h>
 #include <ripple/basics/Log.h>
+#include <ripple/basics/PerfLog.h>
 #include <ripple/beast/container/aged_map.h>
 #include <ripple/beast/core/LexicalCast.h>
 #include <ripple/core/JobQueue.h>
 #include <ripple/nodestore/DatabaseShard.h>
 #include <ripple/protocol/jss.h>
+#include <exception>
 #include <memory>
 #include <mutex>
 #include <vector>
@@ -69,76 +71,114 @@ class InboundLedgersImp : public InboundLedgers
         std::uint32_t seq,
         InboundLedger::Reason reason) override
     {
-        assert(hash.isNonZero());
-        assert(
-            reason != InboundLedger::Reason::SHARD ||
-            (seq != 0 && app_.getShardStore()));
-
-        // probably not the right rule
-        if (app_.getOPs().isNeedNetworkLedger() &&
-            (reason != InboundLedger::Reason::GENERIC) &&
-            (reason != InboundLedger::Reason::CONSENSUS))
-            return {};
-
-        bool isNew = true;
-        std::shared_ptr<InboundLedger> inbound;
-        {
-            ScopedLockType sl(mLock);
-            if (stopping_)
-            {
+        auto doAcquire = [&, seq, reason]() -> std::shared_ptr<Ledger const> {
+            assert(hash.isNonZero());
+            assert(
+                reason != InboundLedger::Reason::SHARD ||
+                (seq != 0 && app_.getShardStore()));
+
+            // probably not the right rule
+            if (app_.getOPs().isNeedNetworkLedger() &&
+                (reason != InboundLedger::Reason::GENERIC) &&
+                (reason != InboundLedger::Reason::CONSENSUS))
                 return {};
+
+            bool isNew = true;
+            std::shared_ptr<InboundLedger> inbound;
+            {
+                ScopedLockType sl(mLock);
+                if (stopping_)
+                {
+                    return {};
+                }
+
+                auto it = mLedgers.find(hash);
+                if (it != mLedgers.end())
+                {
+                    isNew = false;
+                    inbound = it->second;
+                }
+                else
+                {
+                    inbound = std::make_shared<InboundLedger>(
+                        app_,
+                        hash,
+                        seq,
+                        reason,
+                        std::ref(m_clock),
+                        mPeerSetBuilder->build());
+                    mLedgers.emplace(hash, inbound);
+                    inbound->init(sl);
+                    ++mCounter;
+                }
             }
 
-            auto it = mLedgers.find(hash);
-            if (it != mLedgers.end())
+            if (inbound->isFailed())
+                return {};
+
+            if (!isNew)
+                inbound->update(seq);
+
+            if (!inbound->isComplete())
+                return {};
+
+            if (reason == InboundLedger::Reason::HISTORY)
             {
-                isNew = false;
-                inbound = it->second;
+                if (inbound->getLedger()->stateMap().family().isShardBacked())
+                    app_.getNodeStore().storeLedger(inbound->getLedger());
             }
-            else
+            else if (reason == InboundLedger::Reason::SHARD)
             {
-                inbound = std::make_shared<InboundLedger>(
-                    app_,
-                    hash,
-                    seq,
-                    reason,
-                    std::ref(m_clock),
-                    mPeerSetBuilder->build());
-                mLedgers.emplace(hash, inbound);
-                inbound->init(sl);
-                ++mCounter;
+                auto shardStore = app_.getShardStore();
+                if (!shardStore)
+                {
+                    JLOG(j_.error())
+                        << "Acquiring shard with no shard store available";
+                    return {};
+                }
+                if (inbound->getLedger()->stateMap().family().isShardBacked())
+                    shardStore->setStored(inbound->getLedger());
+                else
+                    shardStore->storeLedger(inbound->getLedger());
             }
-        }
-
-        if (inbound->isFailed())
-            return {};
+            return inbound->getLedger();
+        };
+        using namespace std::chrono_literals;
+        std::shared_ptr<Ledger const> ledger = perf::measureDurationAndLog(
+            doAcquire, "InboundLedgersImp::acquire", 500ms, j_);
 
-        if (!isNew)
-            inbound->update(seq);
-
-        if (!inbound->isComplete())
-            return {};
+        return ledger;
+    }
 
-        if (reason == InboundLedger::Reason::HISTORY)
+    void
+    acquireAsync(
+        uint256 const& hash,
+        std::uint32_t seq,
+        InboundLedger::Reason reason) override
+    {
+        std::unique_lock lock(acquiresMutex_);
+        try
         {
-            if (inbound->getLedger()->stateMap().family().isShardBacked())
-                app_.getNodeStore().storeLedger(inbound->getLedger());
+            if (pendingAcquires_.contains(hash))
+                return;
+            pendingAcquires_.insert(hash);
+            lock.unlock();
+            acquire(hash, seq, reason);
         }
-        else if (reason == InboundLedger::Reason::SHARD)
+        catch (std::exception const& e)
         {
-            auto shardStore = app_.getShardStore();
-            if (!shardStore)
-            {
-                JLOG(j_.error())
-                    << "Acquiring shard with no shard store available";
-                return {};
-            }
-            if (inbound->getLedger()->stateMap().family().isShardBacked())
-                shardStore->setStored(inbound->getLedger());
-            else
-                shardStore->storeLedger(inbound->getLedger());
+            JLOG(j_.warn())
+                << "Exception thrown for acquiring new inbound ledger " << hash
+                << ": " << e.what();
         }
-        return inbound->getLedger();
+        catch (...)
+        {
+            JLOG(j_.warn())
+                << "Unknown exception thrown for acquiring new inbound ledger "
+                << hash;
+        }
+        lock.lock();
+        pendingAcquires_.erase(hash);
     }
 
     std::shared_ptr<InboundLedger>
@@ -433,6 +473,9 @@ class InboundLedgersImp : public InboundLedgers
     beast::insight::Counter mCounter;
 
     std::unique_ptr<PeerSetBuilder> mPeerSetBuilder;
+
+    std::set<uint256> pendingAcquires_;
+    std::mutex acquiresMutex_;
 };
 
 //------------------------------------------------------------------------------

src/ripple/app/main/Application.cpp

@@ -942,7 +942,7 @@ class ApplicationImp : public Application, public BasicApp
             auto setup = setup_DatabaseCon(*config_, m_journal);
             setup.useGlobalPragma = false;
 
-            mWalletDB = makeWalletDB(setup);
+            mWalletDB = makeWalletDB(setup, m_journal);
         }
         catch (std::exception const& e)
         {

src/ripple/app/main/Main.cpp

@@ -595,7 +595,7 @@ run(int argc, char** argv)
         try
         {
             auto setup = setup_DatabaseCon(*config);
-            if (!doVacuumDB(setup))
+            if (!doVacuumDB(setup, config->journal()))
                 return -1;
         }
         catch (std::exception const& e)