Optimize when to acquire ledgers from the network.

ximinez · Mark Travis · ximinez · commit 88f9f1884f91 · 2024-09-02T19:55:26.000-04:00
Particularly avoid acquiring ledgers likely to be produced locally very soon. Derived from XRPLF#4764 Co-authored-by: Mark Travis <mtravis@ripple.com>
diff --git a/src/ripple/app/ledger/InboundLedger.h b/src/ripple/app/ledger/InboundLedger.h
@@ -59,7 +59,7 @@ class InboundLedger final : public TimeoutCounter,
 
     // Called when another attempt is made to fetch this same ledger
     void
-    update(std::uint32_t seq);
+    update(std::uint32_t seq, bool broadcast);
 
     /** Returns true if we got all the data. */
     bool
@@ -90,7 +90,7 @@ class InboundLedger final : public TimeoutCounter,
     bool
     checkLocal();
     void
-    init(ScopedLockType& collectionLock);
+    init(ScopedLockType& collectionLock, bool broadcast);
 
     bool
     gotData(
@@ -197,6 +197,26 @@ class InboundLedger final : public TimeoutCounter,
     std::unique_ptr<PeerSet> mPeerSet;
 };
 
+inline std::string
+to_string(InboundLedger::Reason reason)
+{
+    using enum InboundLedger::Reason;
+    switch (reason)
+    {
+        case HISTORY:
+            return "HISTORY";
+        case SHARD:
+            return "SHARD";
+        case GENERIC:
+            return "GENERIC";
+        case CONSENSUS:
+            return "CONSENSUS";
+        default:
+            assert(false);
+            return "unknown";
+    }
+}
+
 }  // namespace ripple
 
 #endif
diff --git a/src/ripple/app/ledger/impl/InboundLedger.cpp b/src/ripple/app/ledger/impl/InboundLedger.cpp
@@ -103,7 +103,7 @@ InboundLedger::InboundLedger(
 }
 
 void
-InboundLedger::init(ScopedLockType& collectionLock)
+InboundLedger::init(ScopedLockType& collectionLock, bool broadcast)
 {
     ScopedLockType sl(mtx_);
     collectionLock.unlock();
@@ -148,8 +148,18 @@ InboundLedger::init(ScopedLockType& collectionLock)
     }
     if (!complete_)
     {
-        addPeers();
-        queueJob(sl);
+        if (broadcast)
+        {
+            addPeers();
+            queueJob(sl);
+        }
+        else
+        {
+            // Delay to give time to build the ledger before sending
+            JLOG(journal_.debug()) << "init: Deferring peer requests";
+            deferred_ = true;
+            setTimer(sl);
+        }
         return;
     }
 
@@ -180,7 +190,7 @@ InboundLedger::getPeerCount() const
 }
 
 void
-InboundLedger::update(std::uint32_t seq)
+InboundLedger::update(std::uint32_t seq, bool broadcast)
 {
     ScopedLockType sl(mtx_);
 
@@ -190,6 +200,24 @@ InboundLedger::update(std::uint32_t seq)
 
     // Prevent this from being swept
     touch();
+
+    // If the signal is to broadcast, and this request has never tried to
+    // broadcast before, cancel any waiting timer, then fire off the job to
+    // broadcast. Note that this is calling mPeerSet->getPeerIds(), not
+    // getPeerCount(), because the latter will filter out peers that have been
+    // tried, but are since lost. This wants to check if peers have _ever_ been
+    // tried. If they have, stick with the normal timer flow.
+    if (broadcast && mPeerSet->getPeerIds().empty())
+    {
+        if (cancelTimer(sl))
+        {
+            JLOG(journal_.debug())
+                << "update: cancelling timer to send peer requests";
+            deferred_ = false;
+            addPeers();
+            queueJob(sl);
+        }
+    }
 }
 
 bool
@@ -428,7 +456,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)
+            assert(isDone());
+            JLOG(journal_.info()) << "Finished while waiting " << hash_;
+            return;
+        }
 
         mByHash = true;
 
diff --git a/src/ripple/app/ledger/impl/InboundLedgers.cpp b/src/ripple/app/ledger/impl/InboundLedgers.cpp
@@ -78,18 +78,114 @@ class InboundLedgersImp : public InboundLedgers
                 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))
+            bool const shouldAcquire = [&]() {
+                if (!app_.getOPs().isNeedNetworkLedger())
+                    return true;
+                if (reason == InboundLedger::Reason::GENERIC)
+                    return true;
+                if (reason == InboundLedger::Reason::CONSENSUS)
+                    return true;
+                return false;
+            }();
+            assert(
+                shouldAcquire ==
+                !(app_.getOPs().isNeedNetworkLedger() &&
+                  (reason != InboundLedger::Reason::GENERIC) &&
+                  (reason != InboundLedger::Reason::CONSENSUS)));
+
+            std::stringstream ss;
+            ss << "InboundLedger::acquire: "
+               << "Request: " << to_string(hash) << ", " << seq
+               << " NeedNetworkLedger: "
+               << (app_.getOPs().isNeedNetworkLedger() ? "yes" : "no")
+               << " Reason: " << to_string(reason)
+               << " Should acquire: " << (shouldAcquire ? "true." : "false.");
+
+            /*  Acquiring ledgers is somewhat expensive. It requires lots of
+             *  computation and network communication. Avoid it when it's not
+             *  appropriate. Every validation from a peer for a ledger that
+             *  we do not have locally results in a call to this function: even
+             *  if we are moments away from validating the same ledger.
+             */
+            bool const shouldBroadcast = [&]() {
+                // If the node is not in "full" state, it needs to sync to
+                // the network, and doesn't have the necessary tx's and
+                // ledger entries to build the ledger.
+                bool const isFull = app_.getOPs().isFull();
+                // fallingBehind means the last closed ledger is at least 2
+                // behind the validated ledger. If the node is falling
+                // behind the network, it probably needs information from
+                // the network to catch up.
+                //
+                // The reason this should not simply be only at least 1
+                // behind the validated ledger is that a slight lag is
+                // normal case because some nodes get there slightly later
+                // than others. A difference of 2 means that at least a full
+                // ledger interval has passed, so the node is beginning to
+                // fall behind.
+                bool const fallingBehind = app_.getOPs().isFallingBehind();
+                // If everything else is ok, don't try to acquire the ledger
+                // if the requested seq is in the near future relative to
+                // the validated ledger. If the requested ledger is between
+                // 1 and 19 inclusive ledgers ahead of the valid ledger this
+                // node has not built it yet, but it's possible/likely it
+                // has the tx's necessary to build it and get caught up.
+                // Plus it might not become validated. On the other hand, if
+                // it's more than 20 in the future, this node should request
+                // it so that it can jump ahead and get caught up.
+                LedgerIndex const validSeq =
+                    app_.getLedgerMaster().getValidLedgerIndex();
+                constexpr std::size_t lagLeeway = 20;
+                bool const nearFuture =
+                    (seq > validSeq) && (seq < validSeq + lagLeeway);
+                // If everything else is ok, don't try to acquire the ledger
+                // if the request is related to consensus. (Note that
+                // consensus calls usually pass a seq of 0, so nearFuture
+                // will be false other than on a brand new network.)
+                bool const consensus =
+                    reason == InboundLedger::Reason::CONSENSUS;
+                ss << " Evaluating whether to broadcast requests to peers"
+                   << ". full: " << (isFull ? "true" : "false")
+                   << ". falling behind: " << (fallingBehind ? "true" : "false")
+                   << ". ledger sequence " << seq
+                   << ". Valid sequence: " << validSeq
+                   << ". Lag leeway: " << lagLeeway
+                   << ". request for near future ledger: "
+                   << (nearFuture ? "true" : "false")
+                   << ". Consensus: " << (consensus ? "true" : "false");
+
+                // If the node is not synced, send requests.
+                if (!isFull)
+                    return true;
+                // If the node is falling behind, send requests.
+                if (fallingBehind)
+                    return true;
+                // If the ledger is in the near future, do NOT send requests.
+                // This node is probably about to build it.
+                if (nearFuture)
+                    return false;
+                // If the request is because of consensus, do NOT send requests.
+                // This node is probably about to build it.
+                if (consensus)
+                    return false;
+                return true;
+            }();
+            ss << ". Broadcast to peers? "
+               << (shouldBroadcast ? "true." : "false.");
+
+            if (!shouldAcquire)
+            {
+                JLOG(j_.debug()) << "Abort(rule): " << ss.str();
                 return {};
+            }
 
             bool isNew = true;
             std::shared_ptr<InboundLedger> inbound;
             {
                 ScopedLockType sl(mLock);
                 if (stopping_)
                 {
+                    JLOG(j_.debug()) << "Abort(stopping): " << ss.str();
                     return {};
                 }
 
@@ -109,19 +205,26 @@ class InboundLedgersImp : public InboundLedgers
                         std::ref(m_clock),
                         mPeerSetBuilder->build());
                     mLedgers.emplace(hash, inbound);
-                    inbound->init(sl);
+                    inbound->init(sl, shouldBroadcast);
                     ++mCounter;
                 }
             }
+            ss << " IsNew: " << (isNew ? "true" : "false");
 
             if (inbound->isFailed())
+            {
+                JLOG(j_.debug()) << "Abort(failed): " << ss.str();
                 return {};
+            }
 
             if (!isNew)
-                inbound->update(seq);
+                inbound->update(seq, shouldBroadcast);
 
             if (!inbound->isComplete())
+            {
+                JLOG(j_.debug()) << "Incomplete: " << ss.str();
                 return {};
+            }
 
             if (reason == InboundLedger::Reason::HISTORY)
             {
@@ -134,21 +237,22 @@ class InboundLedgersImp : public InboundLedgers
                 if (!shardStore)
                 {
                     JLOG(j_.error())
-                        << "Acquiring shard with no shard store available";
+                        << "Acquiring shard with no shard store available"
+                        << ss.str();
                     return {};
                 }
                 if (inbound->getLedger()->stateMap().family().isShardBacked())
                     shardStore->setStored(inbound->getLedger());
                 else
                     shardStore->storeLedger(inbound->getLedger());
             }
+
+            JLOG(j_.debug()) << "Complete: " << ss.str();
             return inbound->getLedger();
         };
         using namespace std::chrono_literals;
-        std::shared_ptr<Ledger const> ledger = perf::measureDurationAndLog(
+        return perf::measureDurationAndLog(
             doAcquire, "InboundLedgersImp::acquire", 500ms, j_);
-
-        return ledger;
     }
 
     void
diff --git a/src/ripple/app/ledger/impl/LedgerMaster.cpp b/src/ripple/app/ledger/impl/LedgerMaster.cpp
@@ -1114,8 +1114,9 @@ LedgerMaster::checkAccept(std::shared_ptr<Ledger const> const& ledger)
     }
 
     JLOG(m_journal.info()) << "Advancing accepted ledger to "
-                           << ledger->info().seq << " with >= " << minVal
-                           << " validations";
+                           << ledger->info().seq << " ("
+                           << to_short_string(ledger->info().hash)
+                           << ") with >= " << minVal << " validations";
 
     ledger->setValidated();
     ledger->setFull();
diff --git a/src/ripple/app/ledger/impl/TimeoutCounter.cpp b/src/ripple/app/ledger/impl/TimeoutCounter.cpp
@@ -52,6 +52,7 @@ TimeoutCounter::setTimer(ScopedLockType& sl)
 {
     if (isDone())
         return;
+    JLOG(journal_.debug()) << "Setting timer for " << timerInterval_;
     timer_.expires_after(timerInterval_);
     timer_.async_wait(
         [wptr = pmDowncast()](boost::system::error_code const& ec) {
@@ -61,11 +62,25 @@ TimeoutCounter::setTimer(ScopedLockType& sl)
             if (auto ptr = wptr.lock())
             {
                 ScopedLockType sl(ptr->mtx_);
+                JLOG(ptr->journal_.debug())
+                    << "timer: ec: " << ec << " (operation_aborted: "
+                    << boost::asio::error::operation_aborted << " - "
+                    << (ec == boost::asio::error::operation_aborted ? "aborted"
+                                                                    : "other")
+                    << ")";
                 ptr->queueJob(sl);
             }
         });
 }
 
+std::size_t
+TimeoutCounter::cancelTimer(ScopedLockType& sl)
+{
+    auto const ret = timer_.cancel();
+    JLOG(journal_.debug()) << "Cancelled " << ret << " timer(s)";
+    return ret;
+}
+
 void
 TimeoutCounter::queueJob(ScopedLockType& sl)
 {
@@ -100,7 +115,10 @@ TimeoutCounter::invokeOnTimer()
 
     if (!progress_)
     {
-        ++timeouts_;
+        if (deferred_)
+            deferred_ = false;
+        else
+            ++timeouts_;
         JLOG(journal_.debug()) << "Timeout(" << timeouts_ << ") "
                                << " acquiring " << hash_;
         onTimer(false, sl);
diff --git a/src/ripple/app/ledger/impl/TimeoutCounter.h b/src/ripple/app/ledger/impl/TimeoutCounter.h
@@ -101,6 +101,10 @@ class TimeoutCounter
     void
     setTimer(ScopedLockType&);
 
+    /** Cancel any waiting timer */
+    std::size_t
+    cancelTimer(ScopedLockType&);
+
     /** Queue a job to call invokeOnTimer(). */
     void
     queueJob(ScopedLockType&);
@@ -132,6 +136,9 @@ class TimeoutCounter
     int timeouts_;
     bool complete_;
     bool failed_;
+    /** Whether the initialization deferred doing any work until the first
+     * timeout. */
+    bool deferred_ = false;
     /** Whether forward progress has been made. */
     bool progress_;
     /** The minimum time to wait between calls to execute(). */
diff --git a/src/ripple/app/misc/NetworkOPs.cpp b/src/ripple/app/misc/NetworkOPs.cpp
diff --git a/src/ripple/app/misc/NetworkOPs.h b/src/ripple/app/misc/NetworkOPs.h
