Measuring job queue latencies

Builds/levelization/results/loops.txt

@@ -29,7 +29,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

@@ -162,6 +162,7 @@ xrpld.ledger > xrpl.basics
 xrpld.ledger > xrpl.json
 xrpld.ledger > xrpl.protocol
 xrpld.net > xrpl.basics
+xrpld.net > xrpld.perflog
 xrpld.net > xrpl.json
 xrpld.net > xrpl.protocol
 xrpld.net > xrpl.resource

src/xrpld/app/consensus/RCLConsensus.cpp

@@ -37,6 +37,7 @@
 #include <xrpld/consensus/LedgerTiming.h>
 #include <xrpld/overlay/Overlay.h>
 #include <xrpld/overlay/predicates.h>
+#include <xrpld/perflog/PerfLog.h>
 
 #include <xrpl/basics/random.h>
 #include <xrpl/beast/core/LexicalCast.h>
@@ -49,6 +50,8 @@
 #include <iomanip>
 #include <mutex>
 
+using namespace std::chrono_literals;
+
 namespace ripple {
 
 RCLConsensus::RCLConsensus(
@@ -140,11 +143,17 @@ RCLConsensus::Adaptor::acquireLedger(LedgerHash const& hash)
             app_.getJobQueue().addJob(
                 jtADVANCE,
                 "getConsensusLedger1",
-                [id = hash, &app = app_, this]() {
-                    JLOG(j_.debug())
-                        << "JOB advanceLedger getConsensusLedger1 started";
-                    app.getInboundLedgers().acquireAsync(
-                        id, 0, InboundLedger::Reason::CONSENSUS);
+                [id = hash, &app = app_, this, journal = j_]() {
+                    perf::measureDurationAndLog(
+                        [&]() {
+                            JLOG(j_.debug()) << "JOB advanceLedger "
+                                                "getConsensusLedger1 started";
+                            app.getInboundLedgers().acquireAsync(
+                                id, 0, InboundLedger::Reason::CONSENSUS);
+                        },
+                        "getConsensusLedger1",
+                        1s,
+                        journal);
                 });
         }
         return std::nullopt;
@@ -442,21 +451,27 @@ RCLConsensus::Adaptor::onAccept(
     app_.getJobQueue().addJob(
         jtACCEPT,
         "acceptLedger",
-        [=, this, cj = std::move(consensusJson)]() mutable {
-            // Note that no lock is held or acquired during this job.
-            // This is because generic Consensus guarantees that once a ledger
-            // is accepted, the consensus results and capture by reference state
-            // will not change until startRound is called (which happens via
-            // endConsensus).
-            RclConsensusLogger clog("onAccept", validating, j_);
-            this->doAccept(
-                result,
-                prevLedger,
-                closeResolution,
-                rawCloseTimes,
-                mode,
-                std::move(cj));
-            this->app_.getOPs().endConsensus(clog.ss());
+        [=, this, cj = std::move(consensusJson), journal = j_]() mutable {
+            perf::measureDurationAndLog(
+                [&]() {
+                    // Note that no lock is held or acquired during this job.
+                    // This is because generic Consensus guarantees that once a
+                    // ledger is accepted, the consensus results and capture by
+                    // reference state will not change until startRound is
+                    // called (which happens via endConsensus).
+                    RclConsensusLogger clog("onAccept", validating, j_);
+                    this->doAccept(
+                        result,
+                        prevLedger,
+                        closeResolution,
+                        rawCloseTimes,
+                        mode,
+                        std::move(cj));
+                    this->app_.getOPs().endConsensus(clog.ss());
+                },
+                "acceptLedger",
+                1s,
+                journal);
         });
 }
 

src/xrpld/app/consensus/RCLValidations.cpp

@@ -32,6 +32,8 @@
 
 #include <memory>
 
+using namespace std::chrono_literals;
+
 namespace ripple {
 
 RCLValidatedLedger::RCLValidatedLedger(MakeGenesis)
@@ -142,11 +144,19 @@ RCLValidationsAdaptor::acquire(LedgerHash const& hash)
         Application* pApp = &app_;
 
         app_.getJobQueue().addJob(
-            jtADVANCE, "getConsensusLedger2", [pApp, hash, this]() {
-                JLOG(j_.debug())
-                    << "JOB advanceLedger getConsensusLedger2 started";
-                pApp->getInboundLedgers().acquireAsync(
-                    hash, 0, InboundLedger::Reason::CONSENSUS);
+            jtADVANCE,
+            "getConsensusLedger2",
+            [pApp, hash, this, journal = j_]() {
+                perf::measureDurationAndLog(
+                    [&]() {
+                        JLOG(j_.debug())
+                            << "JOB advanceLedger getConsensusLedger2 started";
+                        pApp->getInboundLedgers().acquireAsync(
+                            hash, 0, InboundLedger::Reason::CONSENSUS);
+                    },
+                    "getConsensusLedger2",
+                    1s,
+                    journal);
             });
         return std::nullopt;
     }

src/xrpld/app/ledger/ConsensusTransSetSF.cpp

@@ -23,11 +23,14 @@
 #include <xrpld/app/misc/Transaction.h>
 #include <xrpld/core/JobQueue.h>
 #include <xrpld/nodestore/Database.h>
+#include <xrpld/perflog/PerfLog.h>
 
 #include <xrpl/basics/Log.h>
 #include <xrpl/protocol/HashPrefix.h>
 #include <xrpl/protocol/digest.h>
 
+using namespace std::chrono_literals;
+
 namespace ripple {
 
 ConsensusTransSetSF::ConsensusTransSetSF(Application& app, NodeCache& nodeCache)
@@ -65,9 +68,14 @@ ConsensusTransSetSF::gotNode(
                 "ripple::ConsensusTransSetSF::gotNode : transaction hash "
                 "match");
             auto const pap = &app_;
-            app_.getJobQueue().addJob(jtTRANSACTION, "TXS->TXN", [pap, stx]() {
-                pap->getOPs().submitTransaction(stx);
-            });
+            app_.getJobQueue().addJob(
+                jtTRANSACTION, "TXS->TXN", [pap, stx, journal = j_]() {
+                    perf::measureDurationAndLog(
+                        [&]() { pap->getOPs().submitTransaction(stx); },
+                        "TXS->TXN",
+                        1s,
+                        journal);
+                });
         }
         catch (std::exception const& ex)
         {

src/xrpld/app/ledger/Ledger.cpp

@@ -30,6 +30,7 @@
 #include <xrpld/core/SociDB.h>
 #include <xrpld/nodestore/Database.h>
 #include <xrpld/nodestore/detail/DatabaseNodeImp.h>
+#include <xrpld/perflog/PerfLog.h>
 
 #include <xrpl/basics/Log.h>
 #include <xrpl/basics/contract.h>
@@ -46,6 +47,8 @@
 #include <utility>
 #include <vector>
 
+using namespace std::chrono_literals;
+
 namespace ripple {
 
 create_genesis_t const create_genesis{};
@@ -1028,7 +1031,12 @@ pendSaveValidated(
             isCurrent ? jtPUBLEDGER : jtPUBOLDLEDGER,
             std::to_string(ledger->seq()),
             [&app, ledger, isCurrent]() {
-                saveValidatedLedger(app, ledger, isCurrent);
+                beast::Journal journal = app.journal("Ledger");
+                perf::measureDurationAndLog(
+                    [&]() { saveValidatedLedger(app, ledger, isCurrent); },
+                    "OrderBookDB::update:",
+                    1s,
+                    journal);
             }))
     {
         return true;

src/xrpld/app/ledger/OrderBookDB.cpp

@@ -24,10 +24,13 @@
 #include <xrpld/app/misc/NetworkOPs.h>
 #include <xrpld/core/Config.h>
 #include <xrpld/core/JobQueue.h>
+#include <xrpld/perflog/PerfLog.h>
 
 #include <xrpl/basics/Log.h>
 #include <xrpl/protocol/Indexes.h>
 
+using namespace std::chrono_literals;
+
 namespace ripple {
 
 OrderBookDB::OrderBookDB(Application& app)
@@ -69,7 +72,13 @@ OrderBookDB::setup(std::shared_ptr<ReadView const> const& ledger)
             app_.getJobQueue().addJob(
                 jtUPDATE_PF,
                 "OrderBookDB::update: " + std::to_string(ledger->seq()),
-                [this, ledger]() { update(ledger); });
+                [this, ledger, journal = j_]() {
+                    perf::measureDurationAndLog(
+                        [&]() { update(ledger); },
+                        "OrderBookDB::update:",
+                        1s,
+                        journal);
+                });
     }
 }
 

src/xrpld/app/ledger/detail/InboundLedger.cpp

@@ -25,6 +25,7 @@
 #include <xrpld/app/main/Application.h>
 #include <xrpld/core/JobQueue.h>
 #include <xrpld/overlay/Overlay.h>
+#include <xrpld/perflog/PerfLog.h>
 #include <xrpld/shamap/SHAMapNodeID.h>
 
 #include <xrpl/basics/Log.h>
@@ -37,6 +38,8 @@
 #include <algorithm>
 #include <random>
 
+using namespace std::chrono_literals;
+
 namespace ripple {
 
 using namespace std::chrono_literals;
@@ -473,15 +476,24 @@ InboundLedger::done()
 
     // We hold the PeerSet lock, so must dispatch
     app_.getJobQueue().addJob(
-        jtLEDGER_DATA, "AcquisitionDone", [self = shared_from_this()]() {
-            if (self->complete_ && !self->failed_)
-            {
-                self->app_.getLedgerMaster().checkAccept(self->getLedger());
-                self->app_.getLedgerMaster().tryAdvance();
-            }
-            else
-                self->app_.getInboundLedgers().logFailure(
-                    self->hash_, self->mSeq);
+        jtLEDGER_DATA,
+        "AcquisitionDone",
+        [self = shared_from_this(), journal = journal_]() {
+            perf::measureDurationAndLog(
+                [&]() {
+                    if (self->complete_ && !self->failed_)
+                    {
+                        self->app_.getLedgerMaster().checkAccept(
+                            self->getLedger());
+                        self->app_.getLedgerMaster().tryAdvance();
+                    }
+                    else
+                        self->app_.getInboundLedgers().logFailure(
+                            self->hash_, self->mSeq);
+                },
+                "AcquisitionDone",
+                1s,
+                journal);
         });
 }
 

src/xrpld/app/ledger/detail/InboundLedgers.cpp

@@ -35,6 +35,8 @@
 #include <mutex>
 #include <vector>
 
+using namespace std::chrono_literals;
+
 namespace ripple {
 
 class InboundLedgersImp : public InboundLedgers
@@ -212,8 +214,14 @@ class InboundLedgersImp : public InboundLedgers
             // dispatch
             if (ledger->gotData(std::weak_ptr<Peer>(peer), packet))
                 app_.getJobQueue().addJob(
-                    jtLEDGER_DATA, "processLedgerData", [ledger]() {
-                        ledger->runData();
+                    jtLEDGER_DATA,
+                    "processLedgerData",
+                    [ledger, journal = j_]() {
+                        perf::measureDurationAndLog(
+                            [&]() { ledger->runData(); },
+                            "processLedgerData",
+                            1s,
+                            journal);
                     });
 
             return true;
@@ -227,8 +235,12 @@ class InboundLedgersImp : public InboundLedgers
         if (packet->type() == protocol::liAS_NODE)
         {
             app_.getJobQueue().addJob(
-                jtLEDGER_DATA, "gotStaleData", [this, packet]() {
-                    gotStaleData(packet);
+                jtLEDGER_DATA, "gotStaleData", [this, packet, journal = j_]() {
+                    perf::measureDurationAndLog(
+                        [&]() { gotStaleData(packet); },
+                        "gotStaleData",
+                        1s,
+                        journal);
                 });
         }
 

src/xrpld/app/ledger/detail/LedgerDeltaAcquire.cpp

@@ -25,6 +25,9 @@
 #include <xrpld/app/main/Application.h>
 #include <xrpld/core/JobQueue.h>
 #include <xrpld/overlay/PeerSet.h>
+#include <xrpld/perflog/PerfLog.h>
+
+using namespace std::chrono_literals;
 
 namespace ripple {
 
@@ -244,22 +247,31 @@ LedgerDeltaAcquire::onLedgerBuilt(
     app_.getJobQueue().addJob(
         jtREPLAY_TASK,
         "onLedgerBuilt",
-        [=, ledger = this->fullLedger_, &app = this->app_]() {
-            for (auto reason : reasons)
-            {
-                switch (reason)
-                {
-                    case InboundLedger::Reason::GENERIC:
-                        app.getLedgerMaster().storeLedger(ledger);
-                        break;
-                    default:
-                        // TODO for other use cases
-                        break;
-                }
-            }
+        [=,
+         ledger = this->fullLedger_,
+         &app = this->app_,
+         journal = journal_]() {
+            perf::measureDurationAndLog(
+                [&]() {
+                    for (auto reason : reasons)
+                    {
+                        switch (reason)
+                        {
+                            case InboundLedger::Reason::GENERIC:
+                                app.getLedgerMaster().storeLedger(ledger);
+                                break;
+                            default:
+                                // TODO for other use cases
+                                break;
+                        }
+                    }
 
-            if (firstTime)
-                app.getLedgerMaster().tryAdvance();
+                    if (firstTime)
+                        app.getLedgerMaster().tryAdvance();
+                },
+                "onLedgerBuilt",
+                1s,
+                journal);
         });
 }