Merge remote-tracking branch 'upstream/master' into upstream--develop

* upstream/master:
  Set version to 2.2.2
  Allow only 1 job queue slot for each validation ledger check
  Allow only 1 job queue slot for acquiring inbound ledger.
  Track latencies of certain code blocks, and log if they take too long

Builds/levelization/results/loops.txt

@@ -32,7 +32,7 @@ Loop: xrpld.core xrpld.net
   xrpld.net > xrpld.core
 
 Loop: xrpld.core xrpld.perflog
-  xrpld.perflog ~= xrpld.core
+  xrpld.perflog == xrpld.core
 
 Loop: xrpld.net xrpld.rpc
   xrpld.rpc ~= xrpld.net

Builds/levelization/results/ordering.txt

@@ -171,6 +171,7 @@ xrpld.nodestore > xrpl.protocol
 xrpld.overlay > xrpl.basics
 xrpld.overlay > xrpld.core
 xrpld.overlay > xrpld.peerfinder
+xrpld.overlay > xrpld.perflog
 xrpld.overlay > xrpl.json
 xrpld.overlay > xrpl.protocol
 xrpld.overlay > xrpl.resource

src/test/app/LedgerReplay_test.cpp

@@ -106,6 +106,14 @@ class MagicInboundLedgers : public InboundLedgers
         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
     {

src/test/app/Manifest_test.cpp

@@ -255,7 +255,7 @@ class Manifest_test : public beast::unit_test::suite
             setup.dataDir = getDatabasePath();
             assert(!setup.useGlobalPragma);
 
-            auto dbCon = makeTestWalletDB(setup, dbName);
+            auto dbCon = makeTestWalletDB(setup, dbName, env.journal);
 
             auto getPopulatedManifests =
                 [](ManifestCache const& cache) -> std::vector<Manifest const*> {

src/xrpld/app/consensus/RCLConsensus.cpp

@@ -134,8 +134,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/xrpld/app/consensus/RCLValidations.cpp

@@ -27,6 +27,7 @@
 #include <xrpld/consensus/LedgerTiming.h>
 #include <xrpld/core/JobQueue.h>
 #include <xrpld/core/TimeKeeper.h>
+#include <xrpld/perflog/PerfLog.h>
 #include <xrpl/basics/Log.h>
 #include <xrpl/basics/StringUtilities.h>
 #include <xrpl/basics/chrono.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/xrpld/app/consensus/RCLValidations.h

@@ -25,12 +25,16 @@
 #include <xrpl/protocol/Protocol.h>
 #include <xrpl/protocol/RippleLedgerHash.h>
 #include <xrpl/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/xrpld/app/ledger/InboundLedgers.h

@@ -23,6 +23,7 @@
 #include <xrpld/app/ledger/InboundLedger.h>
 #include <xrpl/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/xrpld/app/ledger/detail/InboundLedger.cpp

@@ -495,7 +495,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/xrpld/app/ledger/detail/InboundLedgers.cpp

@@ -22,11 +22,13 @@
 #include <xrpld/app/main/Application.h>
 #include <xrpld/app/misc/NetworkOPs.h>
 #include <xrpld/core/JobQueue.h>
+#include <xrpld/perflog/PerfLog.h>
 #include <xrpl/basics/DecayingSample.h>
 #include <xrpl/basics/Log.h>
 #include <xrpl/beast/container/aged_map.h>
 #include <xrpl/beast/core/LexicalCast.h>
 #include <xrpl/protocol/jss.h>
+#include <exception>
 #include <memory>
 #include <mutex>
 #include <vector>
@@ -68,54 +70,92 @@ class InboundLedgersImp : public InboundLedgers
         std::uint32_t seq,
         InboundLedger::Reason reason) override
     {
-        assert(hash.isNonZero());
-
-        // probably not the right rule
-        if (app_.getOPs().isNeedNetworkLedger() &&
-            (reason != InboundLedger::Reason::GENERIC) &&
-            (reason != InboundLedger::Reason::CONSENSUS))
-            return {};
+        auto doAcquire = [&, seq, reason]() -> std::shared_ptr<Ledger const> {
+            assert(hash.isNonZero());
 
-        bool isNew = true;
-        std::shared_ptr<InboundLedger> inbound;
-        {
-            ScopedLockType sl(mLock);
-            if (stopping_)
-            {
+            // probably not the right rule
+            if (app_.getOPs().isNeedNetworkLedger() &&
+                (reason != InboundLedger::Reason::GENERIC) &&
+                (reason != InboundLedger::Reason::CONSENSUS))
                 return {};
-            }
 
-            auto it = mLedgers.find(hash);
-            if (it != mLedgers.end())
+            bool isNew = true;
+            std::shared_ptr<InboundLedger> inbound;
             {
-                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;
+                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;
+                }
             }
-        }
 
-        if (inbound->isFailed())
-            return {};
+            if (inbound->isFailed())
+                return {};
 
-        if (!isNew)
-            inbound->update(seq);
+            if (!isNew)
+                inbound->update(seq);
 
-        if (!inbound->isComplete())
-            return {};
+            if (!inbound->isComplete())
+                return {};
 
-        return inbound->getLedger();
+            return inbound->getLedger();
+        };
+        using namespace std::chrono_literals;
+        std::shared_ptr<Ledger const> ledger = perf::measureDurationAndLog(
+            doAcquire, "InboundLedgersImp::acquire", 500ms, j_);
+
+        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>
@@ -410,6 +450,9 @@ class InboundLedgersImp : public InboundLedgers
     beast::insight::Counter mCounter;
 
     std::unique_ptr<PeerSetBuilder> mPeerSetBuilder;
+
+    std::set<uint256> pendingAcquires_;
+    std::mutex acquiresMutex_;
 };
 
 //------------------------------------------------------------------------------

src/xrpld/app/main/Application.cpp

@@ -839,7 +839,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/xrpld/app/main/Main.cpp

@@ -592,7 +592,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)