From f5900b96e8984e69aedaef93111c3d8f80a5bc58 Mon Sep 17 00:00:00 2001 From: Pratik Mankawde <3397372+pratikmankawde@users.noreply.github.com> Date: Thu, 17 Sep 2026 22:04:59 +0100 Subject: [PATCH] perf: Cap the number of threads TaggedCache::sweep() starts sweep() started one thread per cache partition, and the partition count defaults to the host's core count, so a sweep of one cache started one thread per core. Several caches are swept per timer tick, so the churn multiplied by the number of caches. Cap the workers instead of the partitions. sweep() now starts min(partitions, kMaxSweepThreads) of them, and worker w takes partitions w, w + workerCount, w + 2 * workerCount and so on, which covers every partition exactly once. sweepHelper decided what one partition's sweep does and also wrapped it in a std::thread. Split those: sweepPartition returns void and runs in the calling thread, and sweep() alone owns thread creation. Both overloads change the same way; the sweeping logic is untouched. Fewer workers means the exclusive cache lock is held for longer, since sweep() joins inside it. That trade is unmeasured. Add gtest cases for the cap, and for eviction reaching every partition on both the key/value and key-only caches. --- include/xrpl/basics/TaggedCache.h | 42 ++++- include/xrpl/basics/TaggedCache.ipp | 176 +++++++++--------- .../xrpl/basics/partitioned_unordered_map.h | 33 +--- src/tests/libxrpl/basics/TaggedCache.cpp | 66 +++++++ .../basics/partitioned_unordered_map.cpp | 69 ------- 5 files changed, 205 insertions(+), 181 deletions(-) delete mode 100644 src/tests/libxrpl/basics/partitioned_unordered_map.cpp diff --git a/include/xrpl/basics/TaggedCache.h b/include/xrpl/basics/TaggedCache.h index 7bb2cb552b..366c411104 100644 --- a/include/xrpl/basics/TaggedCache.h +++ b/include/xrpl/basics/TaggedCache.h @@ -19,7 +19,6 @@ #include #include #include -#include #include #include @@ -73,6 +72,16 @@ public: using shared_weak_combo_pointer_type = SharedWeakUnionPointerType; using shared_pointer_type = SharedPointerType; + /** + * Most worker threads a single sweep() may start. + * + * The cache is split into partitions and sweep() shares them out over its + * workers. A worker takes several partitions when there are more of them + * than this, so the thread count is fixed however many partitions the + * cache holds and however many cores the host has. + */ + static constexpr std::size_t kMaxSweepThreads = 4; + public: TaggedCache( std::string const& name, @@ -357,8 +366,19 @@ private: using cache_type = hardened_partitioned_hash_map; - [[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 timestamp set in the future. + * @param partition The one partition to walk. + * @param stuffToSweep Collects the evicted pointers so the caller can + * destroy them once it has released the cache lock. + * @param allRemovals Accumulates this partition's removal count across all + * workers. + */ + void + sweepPartition( clock_type::time_point const& whenExpire, [[maybe_unused]] clock_type::time_point const& now, KeyValueCacheType::map_type& partition, @@ -366,8 +386,20 @@ private: std::atomic& allRemovals, std::scoped_lock 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 nothing is collected for later + * destruction and the stuffToSweep parameter is unused. + * + * @param whenExpire Entries last accessed at or before this point expire. + * @param now Current time, used to pull back a timestamp set in the future. + * @param partition The one partition to walk. + * @param allRemovals Accumulates this partition's removal count across all + * workers. + */ + void + sweepPartition( clock_type::time_point const& whenExpire, clock_type::time_point const& now, KeyOnlyCacheType::map_type& partition, diff --git a/include/xrpl/basics/TaggedCache.ipp b/include/xrpl/basics/TaggedCache.ipp index 447743a7b7..7489fe051e 100644 --- a/include/xrpl/basics/TaggedCache.ipp +++ b/include/xrpl/basics/TaggedCache.ipp @@ -6,6 +6,7 @@ #include #include +#include namespace xrpl { @@ -261,14 +262,25 @@ TaggedCache workers; - workers.reserve(cache_.partitions()); + workers.reserve(workerCount); std::atomic 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( - whenExpire, now, cache_.map()[p], allStuffToSweep[p], allRemovals, lock)); + workers.emplace_back([&, this, w]() { + 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 +785,9 @@ template < class Hash, class KeyEqual, class Mutex> -inline std::thread +inline void TaggedCache:: - sweepHelper( + sweepPartition( clock_type::time_point const& whenExpire, [[maybe_unused]] clock_type::time_point const& now, KeyValueCacheType::map_type& partition, @@ -783,65 +795,63 @@ TaggedCache& allRemovals, std::scoped_lock const&) { - return std::thread([&, this]() { - int cacheRemovals = 0; - int mapRemovals = 0; + int cacheRemovals = 0; + int mapRemovals = 0; - // Keep references to all the stuff we sweep - // so that we can destroy them outside the lock. - stuffToSweep.reserve(partition.size()); + // Keep references to all the stuff we sweep + // so that we can destroy them outside the lock. + stuffToSweep.reserve(partition.size()); + { + auto cit = partition.begin(); + while (cit != partition.end()) { - auto cit = partition.begin(); - while (cit != partition.end()) + if (cit->second.isWeak()) { - if (cit->second.isWeak()) + // weak + if (cit->second.isExpired()) { - // weak - if (cit->second.isExpired()) - { - 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; - } + stuffToSweep.emplace_back(std::move(cit->second.ptr)); + ++mapRemovals; + cit = partition.erase(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) - { - JLOG(journal_.debug()) - << "TaggedCache partition sweep " << name_ << ": cache = " << partition.size() - << "-" << cacheRemovals << ", map-=" << mapRemovals; - } + if (mapRemovals > 0 || cacheRemovals > 0) + { + JLOG(journal_.debug()) << "TaggedCache partition sweep " << name_ + << ": cache = " << partition.size() << "-" << cacheRemovals + << ", map-=" << mapRemovals; + } - allRemovals += cacheRemovals; - }); + allRemovals += cacheRemovals; } template < @@ -853,9 +863,9 @@ template < class Hash, class KeyEqual, class Mutex> -inline std::thread +inline void TaggedCache:: - sweepHelper( + sweepPartition( clock_type::time_point const& whenExpire, clock_type::time_point const& now, KeyOnlyCacheType::map_type& partition, @@ -863,43 +873,41 @@ TaggedCache& allRemovals, std::scoped_lock const&) { - return std::thread([&, this]() { - // NOLINTBEGIN https://github.com/XRPLF/rippled/issues/7056 - int cacheRemovals = 0; - int mapRemovals = 0; - // NOLINTEND + // NOLINTBEGIN https://github.com/XRPLF/rippled/issues/7056 + int cacheRemovals = 0; + int mapRemovals = 0; + // NOLINTEND - // Keep references to all the stuff we sweep - // so that we can destroy them outside the lock. + // Keep references to all the stuff we sweep + // so that we can destroy them outside the lock. + { + auto cit = partition.begin(); + while (cit != partition.end()) { - auto cit = partition.begin(); - while (cit != partition.end()) + if (cit->second.lastAccess > now) { - if (cit->second.lastAccess > now) - { - cit->second.lastAccess = now; - ++cit; - } - else if (cit->second.lastAccess <= whenExpire) - { - cit = partition.erase(cit); - } - else - { - ++cit; - } + cit->second.lastAccess = now; + ++cit; + } + else if (cit->second.lastAccess <= whenExpire) + { + cit = partition.erase(cit); + } + else + { + ++cit; } } + } - if (mapRemovals > 0 || cacheRemovals > 0) - { - JLOG(journal_.debug()) - << "TaggedCache partition sweep " << name_ << ": cache = " << partition.size() - << "-" << cacheRemovals << ", map-=" << mapRemovals; - } + if (mapRemovals > 0 || cacheRemovals > 0) + { + JLOG(journal_.debug()) << "TaggedCache partition sweep " << name_ + << ": cache = " << partition.size() << "-" << cacheRemovals + << ", map-=" << mapRemovals; + } - allRemovals += cacheRemovals; - }); + allRemovals += cacheRemovals; } } // namespace xrpl diff --git a/include/xrpl/basics/partitioned_unordered_map.h b/include/xrpl/basics/partitioned_unordered_map.h index 3f3d0e4d19..c6b0107b93 100644 --- a/include/xrpl/basics/partitioned_unordered_map.h +++ b/include/xrpl/basics/partitioned_unordered_map.h @@ -1,6 +1,7 @@ #pragma once #include +#include #include #include @@ -8,6 +9,7 @@ #include #include #include +#include #include #include #include @@ -36,22 +38,9 @@ template < typename Alloc = std::allocator>> class PartitionedUnorderedMap { - /** - * How many sub-maps the keys are spread over. Never zero. - */ std::size_t partitions_; public: - /** - * Partition count used when the caller does not ask for one. - * - * A partition is the unit of parallel work for callers that walk the whole - * map: TaggedCache::sweep() runs one thread per partition. Holding this - * small and fixed keeps that thread count independent of how many cores - * the host has. - */ - static constexpr std::size_t kDefaultPartitions = 2; - using key_type = Key; using mapped_type = Value; using value_type = std::pair; @@ -226,21 +215,19 @@ private: } public: - /** - * Builds an empty map spread over a fixed number of partitions. - * - * @param partitions How many partitions to use. An empty optional, or 0, - * selects kDefaultPartitions. - */ PartitionedUnorderedMap(std::optional partitions = std::nullopt) - : partitions_(partitions && (*partitions != 0u) ? *partitions : kDefaultPartitions) + // Set partitions to the number of hardware threads if the parameter + // is either empty or set to 0. + : partitions_( + partitions && (*partitions != 0u) ? *partitions : std::thread::hardware_concurrency()) { map_.resize(partitions_); + XRPL_ASSERT( + partitions_, + "xrpl::PartitionedUnorderedMap::PartitionedUnorderedMap : " + "nonzero partitions"); } - /** - * Returns how many partitions the keys are spread over. - */ std::size_t partitions() const { diff --git a/src/tests/libxrpl/basics/TaggedCache.cpp b/src/tests/libxrpl/basics/TaggedCache.cpp index c8ccc415ad..639fc84144 100644 --- a/src/tests/libxrpl/basics/TaggedCache.cpp +++ b/src/tests/libxrpl/basics/TaggedCache.cpp @@ -10,6 +10,7 @@ #include #include +#include #include #include #include @@ -243,4 +244,69 @@ TEST(TaggedCacheTest, tagged_cache) } } +TEST(TaggedCacheTest, sweep_thread_count_is_capped) +{ + using Cache = TaggedCache; + + // sweep() runs one worker per partition unless it is capped, and the + // partition count follows the host's core count. + EXPECT_EQ(Cache::kMaxSweepThreads, 4u); +} + +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; + + // 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(key), std::to_string(key))); + + ASSERT_EQ(c.size(), count); + ASSERT_EQ(c.getCacheSize(), static_cast(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; + + 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(key))); + + ASSERT_EQ(c.size(), count); + + ++clock; + c.sweep(); + + EXPECT_EQ(c.size(), 0); +} + } // namespace xrpl diff --git a/src/tests/libxrpl/basics/partitioned_unordered_map.cpp b/src/tests/libxrpl/basics/partitioned_unordered_map.cpp deleted file mode 100644 index 7c82bfcf42..0000000000 --- a/src/tests/libxrpl/basics/partitioned_unordered_map.cpp +++ /dev/null @@ -1,69 +0,0 @@ -#include - -#include - -#include -#include -#include - -namespace xrpl { - -using TestMap = PartitionedUnorderedMap>; - -TEST(PartitionedUnorderedMapTest, default_partition_count_is_two) -{ - // Callers that walk the whole map run one thread per partition, so this - // count bounds that thread count. It must not follow the core count. - EXPECT_EQ(TestMap::kDefaultPartitions, 2u); - - TestMap const map; - EXPECT_EQ(map.partitions(), TestMap::kDefaultPartitions); -} - -TEST(PartitionedUnorderedMapTest, zero_partitions_falls_back_to_the_default) -{ - TestMap const map{0}; - EXPECT_EQ(map.partitions(), TestMap::kDefaultPartitions); -} - -TEST(PartitionedUnorderedMapTest, explicit_partition_count_is_honoured) -{ - TestMap const map{TestMap::kDefaultPartitions + 5}; - EXPECT_EQ(map.partitions(), TestMap::kDefaultPartitions + 5); -} - -TEST(PartitionedUnorderedMapTest, every_key_is_reachable_across_partitions) -{ - // The partitioner is key % partitions, so with two partitions the even - // keys land in one sub-map and the odd keys in the other. - std::size_t const count = 10; - TestMap map; - - for (std::size_t key = 0; key < count; ++key) - map.emplace(key, std::to_string(key)); - - EXPECT_EQ(map.size(), count); - - // Pin the split itself, not just the total: a partitioner that sent every - // key to one sub-map would still satisfy size() and the lookups below. - ASSERT_EQ(map.map().size(), 2u); - EXPECT_EQ(map.map()[0].size(), count / 2); - EXPECT_EQ(map.map()[1].size(), count / 2); - - for (std::size_t key = 0; key < count; ++key) - { - auto const it = map.find(key); - ASSERT_NE(it, map.end()); - EXPECT_EQ(it->second, std::to_string(key)); - } - - std::size_t visited = 0; - for (auto const& entry : map) - { - EXPECT_EQ(entry.second, std::to_string(entry.first)); - ++visited; - } - EXPECT_EQ(visited, count); -} - -} // namespace xrpl