Class "Pending" to keep track of pending operations on distinct items

Author: ximinez

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

@@ -23,6 +23,7 @@
 #include <ripple/app/misc/NetworkOPs.h>
 #include <ripple/basics/DecayingSample.h>
 #include <ripple/basics/Log.h>
+#include <ripple/basics/Pending.h>
 #include <ripple/basics/PerfLog.h>
 #include <ripple/beast/container/aged_map.h>
 #include <ripple/beast/core/LexicalCast.h>
@@ -260,29 +261,25 @@ class InboundLedgersImp : public InboundLedgers
         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)
+        if (Pending pending{acquiresMutex_, pendingAcquires_, hash})
         {
-            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;
+            try
+            {
+                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>

src/ripple/app/misc/NetworkOPs.cpp

@@ -42,6 +42,7 @@
 #include <ripple/app/rdb/backend/SQLiteDatabase.h>
 #include <ripple/app/reporting/ReportingETL.h>
 #include <ripple/app/tx/apply.h>
+#include <ripple/basics/Pending.h>
 #include <ripple/basics/PerfLog.h>
 #include <ripple/basics/UptimeClock.h>
 #include <ripple/basics/mulDiv.h>
@@ -2366,34 +2367,27 @@ NetworkOPsImp::recvValidation(
     JLOG(m_journal.trace())
         << "recvValidation " << val->getLedgerHash() << " from " << source;
 
-    std::unique_lock lock(validationsMutex_);
-    BypassAccept bypassAccept = BypassAccept::no;
-    try
-    {
-        if (pendingValidations_.contains(val->getLedgerHash()))
-            bypassAccept = BypassAccept::yes;
-        else
-            pendingValidations_.insert(val->getLedgerHash());
-        lock.unlock();
-        handleNewValidation(app_, val, source, bypassAccept, m_journal);
-    }
-    catch (std::exception const& e)
-    {
-        JLOG(m_journal.warn())
-            << "Exception thrown for handling new validation "
-            << val->getLedgerHash() << ": " << e.what();
-    }
-    catch (...)
-    {
-        JLOG(m_journal.warn())
-            << "Unknown exception thrown for handling new validation "
-            << val->getLedgerHash();
-    }
-    if (bypassAccept == BypassAccept::no)
     {
-        lock.lock();
-        pendingValidations_.erase(val->getLedgerHash());
-        lock.unlock();
+        Pending pending(
+            validationsMutex_, pendingValidations_, val->getLedgerHash());
+        try
+        {
+            BypassAccept bypassAccept =
+                pending.unusable() ? BypassAccept::yes : BypassAccept::no;
+            handleNewValidation(app_, val, source, bypassAccept, m_journal);
+        }
+        catch (std::exception const& e)
+        {
+            JLOG(m_journal.warn())
+                << "Exception thrown for handling new validation "
+                << val->getLedgerHash() << ": " << e.what();
+        }
+        catch (...)
+        {
+            JLOG(m_journal.warn())
+                << "Unknown exception thrown for handling new validation "
+                << val->getLedgerHash();
+        }
     }
 
     pubValidation(val);

src/ripple/basics/Pending.h

@@ -0,0 +1,106 @@
+//------------------------------------------------------------------------------
+/*
+    This file is part of rippled: https://github.com/ripple/rippled
+    Copyright (c) 2024 Ripple Labs Inc.
+
+    Permission to use, copy, modify, and/or distribute this software for any
+    purpose  with  or without fee is hereby granted, provided that the above
+    copyright notice and this permission notice appear in all copies.
+
+    THE  SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+    WITH  REGARD  TO  THIS  SOFTWARE  INCLUDING  ALL  IMPLIED  WARRANTIES  OF
+    MERCHANTABILITY  AND  FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+    ANY  SPECIAL ,  DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+    WHATSOEVER  RESULTING  FROM  LOSS  OF USE, DATA OR PROFITS, WHETHER IN AN
+    ACTION  OF  CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+    OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+*/
+//==============================================================================
+
+#ifndef RIPPLE_BASICS_PENDING_H_INCLUDED
+#define RIPPLE_BASICS_PENDING_H_INCLUDED
+
+/** RAII class to check if an Item is Pending processing in a Collection.
+ *
+ * If the Item is not in the Collection, it will be added under lock in the
+ * ctor, and removed under lock in the dtor.
+ *
+ * If the Item is in the Collection, no changes will be made to the collection,
+ * and the Pending object will be considered "unusable". It's up to the caller
+ * to decide what "unusable" means. (e.g. Skip a block of code, or set a flag.)
+ *
+ * The current use is to avoid lock contention that would be involved in
+ * processing something associated with the Item.
+ *
+ * Example:
+ *
+ * void IncomingLedgers::acquireAsync(LedgerHash const& hash, ...)
+ * {
+ *    Pending pending{acquiresMutex_, pendingAcquires_, hash};
+ *    if (pending.unusable())
+ *      return;
+ *    acquire(hash, ...);
+ * }
+ *
+ * Additionally, a Pending will evaluate to `true` if it is "usable", `false` if
+ * it is "unusable".
+ *
+ * Example:
+ *
+ * void IncomingLedgers::acquireAsync(LedgerHash const& hash, ...)
+ * {
+ *    if (Pending pending{acquiresMutex_, pendingAcquires_, hash})
+ *    {
+ *      acquire(hash, ...);
+ *    }
+ * }
+ *
+
+ */
+template <class Mutex, class Collection, class Item>
+class Pending
+{
+public:
+    Pending(Mutex& mtx, Collection& collection, Item const& item)
+        : mtx_(mtx), collection_(collection), item_(item), unusable_(insert())
+    {
+    }
+
+    ~Pending()
+    {
+        if (!unusable_)
+        {
+            std::unique_lock<Mutex> lock_(mtx_);
+            collection_.erase(item_);
+        }
+    }
+
+    bool
+    unusable() const
+    {
+        return unusable_;
+    }
+
+    operator bool() const
+    {
+        return !unusable_;
+    }
+
+private:
+    bool
+    insert()
+    {
+        std::unique_lock<Mutex> lock_(mtx_);
+        bool exists = collection_.contains(item_);
+        if (!exists)
+            collection_.insert(item_);
+        return exists;
+    }
+
+    Mutex& mtx_;
+    Collection& collection_;
+    Item const item_;
+    bool const unusable_;
+};
+
+#endif