Merge branch 'develop'

Resolve conflicts with develop's typedef/using rename (#8177) in the
sweepPartition declarations and definitions.

include/xrpl/basics/TaggedCache.h

@@ -19,7 +19,6 @@
 #include <memory>
 #include <mutex>
 #include <string>
-#include <thread>
 #include <type_traits>
 #include <vector>
 
@@ -73,6 +72,14 @@ public:
     using SharedWeakComboPointerType = SharedWeakUnionPointerType;
     using SharedPointerType = SharedPointer;
 
+    /**
+     * Most worker threads a single sweep() may start.
+     *
+     * A worker takes several partitions when there are more than this, so the
+     * thread count does not follow the host's core count.
+     */
+    static constexpr std::size_t kMaxSweepThreads = 8;
+
 public:
     TaggedCache(
         std::string const& name,
@@ -357,8 +364,18 @@ private:
 
     using CacheType = HardenedPartitionedHashMap<key_type, Entry, Hash, KeyEqual>;
 
-    [[nodiscard]] std::thread
+    /**
-    sweepHelper(
+     * Sweeps one partition of a key/value cache, in the calling thread.
+     *
+     * @param whenExpire Entries last accessed at or before this point expire.
+     * @param now Current time, used to pull back a future timestamp.
+     * @param partition The one partition to walk.
+     * @param stuffToSweep Collects evicted pointers, destroyed once the caller
+     * releases the cache lock.
+     * @param allRemovals Accumulates removals across all workers.
+     */
+    void
+    sweepPartition(
         ClockType::time_point const& whenExpire,
         [[maybe_unused]] ClockType::time_point const& now,
         KeyValueCacheType::MapType& partition,
@@ -366,8 +383,18 @@ private:
         std::atomic<int>& allRemovals,
         std::scoped_lock<std::recursive_mutex> const&);
 
-    [[nodiscard]] std::thread
+    /**
-    sweepHelper(
+     * Sweeps one partition of a key-only cache, in the calling thread.
+     *
+     * A key-only cache owns no pointers, so stuffToSweep is unused.
+     *
+     * @param whenExpire Entries last accessed at or before this point expire.
+     * @param now Current time, used to pull back a future timestamp.
+     * @param partition The one partition to walk.
+     * @param allRemovals Accumulates removals across all workers.
+     */
+    void
+    sweepPartition(
         ClockType::time_point const& whenExpire,
         ClockType::time_point const& now,
         KeyOnlyCacheType::MapType& partition,

include/xrpl/basics/TaggedCache.ipp

@@ -6,6 +6,7 @@
 #include <xrpl/basics/scope.h>
 
 #include <algorithm>
+#include <thread>
 
 namespace xrpl {
 
@@ -261,14 +262,23 @@ TaggedCache<Key, T, IsKeyCache, SharedWeakUnionPointer, SharedPointerType, Hash,
                 << " aging at " << (now - whenExpire).count() << " of " << targetAge_.count();
         }
 
+        // Worker w takes partitions w, w + workerCount, w + 2 * workerCount
+        // and so on, covering every partition exactly once.
+        std::size_t const workerCount = std::min(cache_.partitions(), kMaxSweepThreads);
+
         std::vector<std::thread> workers;
-        workers.reserve(cache_.partitions());
+        workers.reserve(workerCount);
         std::atomic<int> allRemovals = 0;
 
-        for (std::size_t p = 0; p < cache_.partitions(); ++p)
+        for (std::size_t w = 0; w < workerCount; ++w)
         {
-            workers.push_back(sweepHelper(
+            workers.emplace_back([&, this, w]() {
-                whenExpire, now, cache_.map()[p], allStuffToSweep[p], allRemovals, lock));
+                for (std::size_t p = w; p < cache_.partitions(); p += workerCount)
+                {
+                    sweepPartition(
+                        whenExpire, now, cache_.map()[p], allStuffToSweep[p], allRemovals, lock);
+                }
+            });
         }
         for (std::thread& worker : workers)
             worker.join();
@@ -773,9 +783,9 @@ template <
     class Hash,
     class KeyEqual,
     class Mutex>
-inline std::thread
+inline void
 TaggedCache<Key, T, IsKeyCache, SharedWeakUnionPointer, SharedPointerType, Hash, KeyEqual, Mutex>::
-    sweepHelper(
+    sweepPartition(
         ClockType::time_point const& whenExpire,
         [[maybe_unused]] ClockType::time_point const& now,
         KeyValueCacheType::MapType& partition,
@@ -783,65 +793,63 @@ TaggedCache<Key, T, IsKeyCache, SharedWeakUnionPointer, SharedPointerType, Hash,
         std::atomic<int>& allRemovals,
         std::scoped_lock<std::recursive_mutex> const&)
 {
-    return std::thread([&, this]() {
+    int cacheRemovals = 0;
-        int cacheRemovals = 0;
+    int mapRemovals = 0;
-        int mapRemovals = 0;
 
-        // Keep references to all the stuff we sweep
+    // Keep references to all the stuff we sweep
-        // so that we can destroy them outside the lock.
+    // so that we can destroy them outside the lock.
-        stuffToSweep.reserve(partition.size());
+    stuffToSweep.reserve(partition.size());
+    {
+        auto cit = partition.begin();
+        while (cit != partition.end())
         {
-            auto cit = partition.begin();
+            if (cit->second.isWeak())
-            while (cit != partition.end())
             {
-                if (cit->second.isWeak())
+                // weak
+                if (cit->second.isExpired())
                 {
-                    // weak
+                    stuffToSweep.emplace_back(std::move(cit->second.ptr));
-                    if (cit->second.isExpired())
+                    ++mapRemovals;
-                    {
+                    cit = partition.erase(cit);
-                        stuffToSweep.emplace_back(std::move(cit->second.ptr));
-                        ++mapRemovals;
-                        cit = partition.erase(cit);
-                    }
-                    else
-                    {
-                        ++cit;
-                    }
-                }
-                else if (cit->second.lastAccess <= whenExpire)
-                {
-                    // strong, expired
-                    ++cacheRemovals;
-                    if (cit->second.ptr.useCount() == 1)
-                    {
-                        stuffToSweep.emplace_back(std::move(cit->second.ptr));
-                        ++mapRemovals;
-                        cit = partition.erase(cit);
-                    }
-                    else
-                    {
-                        // remains weakly cached
-                        cit->second.ptr.convertToWeak();
-                        ++cit;
-                    }
                 }
                 else
                 {
-                    // strong, not expired
                     ++cit;
                 }
             }
+            else if (cit->second.lastAccess <= whenExpire)
+            {
+                // strong, expired
+                ++cacheRemovals;
+                if (cit->second.ptr.useCount() == 1)
+                {
+                    stuffToSweep.emplace_back(std::move(cit->second.ptr));
+                    ++mapRemovals;
+                    cit = partition.erase(cit);
+                }
+                else
+                {
+                    // remains weakly cached
+                    cit->second.ptr.convertToWeak();
+                    ++cit;
+                }
+            }
+            else
+            {
+                // strong, not expired
+                ++cit;
+            }
         }
+    }
 
-        if (mapRemovals || cacheRemovals)
+    if (mapRemovals > 0 || cacheRemovals > 0)
-        {
+    {
-            JLOG(journal_.debug())
+        JLOG(journal_.debug()) << "TaggedCache partition sweep " << name_
-                << "TaggedCache partition sweep " << name_ << ": cache = " << partition.size()
+                               << ": cache = " << partition.size() << "-" << cacheRemovals
-                << "-" << cacheRemovals << ", map-=" << mapRemovals;
+                               << ", map-=" << mapRemovals;
-        }
+    }
 
-        allRemovals += cacheRemovals;
+    allRemovals += cacheRemovals;
-    });
 }
 
 template <
@@ -853,9 +861,9 @@ template <
     class Hash,
     class KeyEqual,
     class Mutex>
-inline std::thread
+inline void
 TaggedCache<Key, T, IsKeyCache, SharedWeakUnionPointer, SharedPointerType, Hash, KeyEqual, Mutex>::
-    sweepHelper(
+    sweepPartition(
         ClockType::time_point const& whenExpire,
         ClockType::time_point const& now,
         KeyOnlyCacheType::MapType& partition,
@@ -863,43 +871,41 @@ TaggedCache<Key, T, IsKeyCache, SharedWeakUnionPointer, SharedPointerType, Hash,
         std::atomic<int>& allRemovals,
         std::scoped_lock<std::recursive_mutex> const&)
 {
-    return std::thread([&, this]() {
+    // NOLINTBEGIN https://github.com/XRPLF/rippled/issues/7056
-        // NOLINTBEGIN https://github.com/XRPLF/rippled/issues/7056
+    int cacheRemovals = 0;
-        int cacheRemovals = 0;
+    int mapRemovals = 0;
-        int mapRemovals = 0;
+    // NOLINTEND
-        // NOLINTEND
 
-        // Keep references to all the stuff we sweep
+    // Keep references to all the stuff we sweep
-        // so that we can destroy them outside the lock.
+    // so that we can destroy them outside the lock.
+    {
+        auto cit = partition.begin();
+        while (cit != partition.end())
         {
-            auto cit = partition.begin();
+            if (cit->second.lastAccess > now)
-            while (cit != partition.end())
             {
-                if (cit->second.lastAccess > now)
+                cit->second.lastAccess = now;
-                {
+                ++cit;
-                    cit->second.lastAccess = now;
+            }
-                    ++cit;
+            else if (cit->second.lastAccess <= whenExpire)
-                }
+            {
-                else if (cit->second.lastAccess <= whenExpire)
+                cit = partition.erase(cit);
-                {
+            }
-                    cit = partition.erase(cit);
+            else
-                }
+            {
-                else
+                ++cit;
-                {
-                    ++cit;
-                }
             }
         }
+    }
 
-        if (mapRemovals > 0 || cacheRemovals > 0)
+    if (mapRemovals > 0 || cacheRemovals > 0)
-        {
+    {
-            JLOG(journal_.debug())
+        JLOG(journal_.debug()) << "TaggedCache partition sweep " << name_
-                << "TaggedCache partition sweep " << name_ << ": cache = " << partition.size()
+                               << ": cache = " << partition.size() << "-" << cacheRemovals
-                << "-" << cacheRemovals << ", map-=" << mapRemovals;
+                               << ", map-=" << mapRemovals;
-        }
+    }
 
-        allRemovals += cacheRemovals;
+    allRemovals += cacheRemovals;
-    });
 }
 
 }  // namespace xrpl

src/tests/libxrpl/basics/TaggedCache.cpp

@@ -10,6 +10,7 @@
 #include <gtest/gtest.h>
 #include <helpers/TestSink.h>
 
+#include <cstddef>
 #include <memory>
 #include <string>
 #include <utility>
@@ -243,4 +244,69 @@ TEST(TaggedCacheTest, tagged_cache)
     }
 }
 
+TEST(TaggedCacheTest, sweep_thread_count_is_capped)
+{
+    using Cache = TaggedCache<LedgerIndex, std::string>;
+
+    // Without the cap, sweep() runs one worker per partition, and the
+    // partition count follows the host's core count.
+    EXPECT_EQ(Cache::kMaxSweepThreads, 8u);
+}
+
+TEST(TaggedCacheTest, sweep_evicts_from_every_partition)
+{
+    using namespace std::chrono_literals;
+    beast::Journal const journal{TestSink::instance()};
+
+    TestStopwatch clock;
+    clock.set(0);
+
+    using Cache = TaggedCache<LedgerIndex, std::string>;
+
+    // targetSize 0 keeps whenExpire at now - targetAge_, so every entry below
+    // expires on the first tick rather than being aged proportionally.
+    Cache c("sweep-all-partitions", 0, 1s, clock, journal);
+
+    // More keys than any plausible partition count, so each partition holds
+    // several. A worker that skipped a partition would leave its keys behind.
+    std::size_t const count = 512;
+    for (std::size_t key = 0; key < count; ++key)
+        EXPECT_FALSE(c.insert(static_cast<LedgerIndex>(key), std::to_string(key)));
+
+    ASSERT_EQ(c.size(), count);
+    ASSERT_EQ(c.getCacheSize(), static_cast<int>(count));
+
+    ++clock;
+    c.sweep();
+
+    EXPECT_EQ(c.size(), 0);
+    EXPECT_EQ(c.getCacheSize(), 0);
+}
+
+TEST(TaggedCacheTest, key_only_sweep_evicts_from_every_partition)
+{
+    using namespace std::chrono_literals;
+    beast::Journal const journal{TestSink::instance()};
+
+    TestStopwatch clock;
+    clock.set(0);
+
+    // The key-only cache is a separate sweepPartition overload, so it needs
+    // its own coverage.
+    using Cache = TaggedCache<LedgerIndex, int, true>;
+
+    Cache c("key-only-sweep-all-partitions", 0, 1s, clock, journal);
+
+    std::size_t const count = 512;
+    for (std::size_t key = 0; key < count; ++key)
+        EXPECT_TRUE(c.insert(static_cast<LedgerIndex>(key)));
+
+    ASSERT_EQ(c.size(), count);
+
+    ++clock;
+    c.sweep();
+
+    EXPECT_EQ(c.size(), 0);
+}
+
 }  // namespace xrpl