From 72e0331bb57323f653ed280b60d01ac4e085e9e6 Mon Sep 17 00:00:00 2001 From: CAICAIIs <3360776475@qq.com> Date: Sat, 3 Oct 2026 02:33:02 +0800 Subject: [PATCH] perf(store): partition a tenant's object route into 64 stripes A tenant's object route held one map under one shared_mutex: writes to distinct keys serialized on it, and a map that outgrew its buckets rehashed every node it held while that lock was held, a walk proportional to the nodes already in the map. Each stripe is a shared_mutex and a map, and a key's stripe is a function of the key alone, so a per-key operation still takes exactly one lock and one key's growth is charged to its stripe's keys. The aggregates are counted stripe by stripe, which the comments record as the weaker meaning they now have, and WithExclusiveRoute takes the key whose stripe it holds. The count belongs to this index: kStripeCount = 64 is the default and the constructor takes a count, so benchmarks/object_index_contention_bench.cpp and the tests can build the same route at another count. The group directory keeps StripedGroupIndex<64> and the count its own measurement chose; nothing needs the two counts to agree. This is the part of #4370 that stripes the route, taken out of that PR. --- mooncake-store/benchmarks/CMakeLists.txt | 4 + .../object_index_contention_bench.cpp | 262 ++++++++++++++++++ mooncake-store/include/metadata/tenant.h | 12 +- mooncake-store/include/object_index.h | 173 +++++++++--- mooncake-store/tests/object_index_test.cpp | 181 ++++++++++++ 5 files changed, 582 insertions(+), 50 deletions(-) create mode 100644 mooncake-store/benchmarks/object_index_contention_bench.cpp diff --git a/mooncake-store/benchmarks/CMakeLists.txt b/mooncake-store/benchmarks/CMakeLists.txt index c05a90132d..bf3580a3cc 100644 --- a/mooncake-store/benchmarks/CMakeLists.txt +++ b/mooncake-store/benchmarks/CMakeLists.txt @@ -69,3 +69,7 @@ target_link_libraries(uring_batch_read_bench PRIVATE mooncake_store) # Add group index write path benchmark executable add_executable(group_index_contention_bench group_index_contention_bench.cpp) target_link_libraries(group_index_contention_bench PRIVATE mooncake_store) + +# Add object route benchmark executable +add_executable(object_index_contention_bench object_index_contention_bench.cpp) +target_link_libraries(object_index_contention_bench PRIVATE mooncake_store) diff --git a/mooncake-store/benchmarks/object_index_contention_bench.cpp b/mooncake-store/benchmarks/object_index_contention_bench.cpp new file mode 100644 index 0000000000..608c340a67 --- /dev/null +++ b/mooncake-store/benchmarks/object_index_contention_bench.cpp @@ -0,0 +1,262 @@ +// Measures the object route's cost against its stripe count: how many stripes a +// tenant's route needs before its writes stop serializing, what a lookup costs +// beside that, and what the count costs in per-tenant memory. +// +// The route is a map under a lock per stripe. One stripe puts every key behind +// one lock, so writes to distinct keys serialize; more stripes spread them at +// about 120 bytes each. A stripe also bounds the map's own cost, because a map +// that outgrows its buckets rehashes every node it holds under its exclusive +// lock: the insert-stall table reports how far the worst single insert grows as +// that walk gets longer. +// +// Every point builds its own route with the count under test, so the printed +// size is the per-tenant cost of that count. Keys are deterministic strings and +// each thread publishes and reads only its own, so a point shows what the +// stripe count buys rather than what two threads writing one key cost. A point +// fills every worker's keys before its window opens and starts every worker at +// a barrier, and the rate is over the window the workers actually ran in, so +// the count under test is not charged for a slower fill. +// +// Output is CSV on stdout. The throughput table is +// stripes,threads,mode,ops_per_s,route_bytes +// with one of three modes: `get` (a lookup), `publish` (build an entry, publish +// it, drop the slot the key held before, as a put does) and `mixed` (eight +// lookups per publish). `ops_per_s` counts a lookup as one operation and a +// publish, including the drop it does first, as one. The insert-stall table is +// stripes,keys,p50_us,p99_us,max_us +// from a single thread publishing distinct keys into a growing route, where the +// entries stay alive so the map keeps every node it rehashes. + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "object_entry.h" +#include "object_index.h" + +namespace mooncake { +namespace { + +constexpr int kThreadCounts[] = {1, 4, 16, 32}; +constexpr size_t kStripeCounts[] = {1, 4, 16, 64, 256}; +// Keys one thread owns. A publisher keeps all of them published and cycles +// over them, dropping each key's previous slot before it publishes the new one. +constexpr size_t kKeysPerThread = 4096; +// Lookups the mixed mode runs per publish. +constexpr size_t kLookupsPerPublish = 8; +constexpr auto kMeasureFor = std::chrono::seconds(2); +// Keys the single-threaded insert-stall pass publishes into a growing route. +constexpr size_t kStallKeys = 1 << 18; + +// One stripe is a lock plus the map of the keys that land in it, and the +// stripes are the index's own allocation, so sizeof(ObjectIndex) alone does not +// carry their cost. A map's own size does not depend on what it maps to, so the +// mapped type only has to be a string-keyed map with the same hasher and +// comparator as the route's. +using RouteMap = std::unordered_map, + TransparentStringHash, std::equal_to<>>; +constexpr size_t kStripeBytes = sizeof(std::shared_mutex) + sizeof(RouteMap); + +size_t RouteBytes(size_t stripe_count) { + return sizeof(ObjectIndex) + stripe_count * kStripeBytes; +} + +std::string KeyOf(size_t thread, size_t index) { + return "t" + std::to_string(thread) + "_k" + std::to_string(index); +} + +// The minimal 128 B, replica-less envelope the object suites build: the route +// stores the shell, and a benchmark needs no replica validity. +std::shared_ptr MakeEntry(const std::string& key) { + constexpr auto kWriteTime = + std::chrono::system_clock::time_point(std::chrono::seconds(1)); + return std::make_shared(std::make_unique( + UUID{1, 2}, kWriteTime, 128, std::vector{}, std::nullopt, + false, ObjectDataType::UNKNOWN, std::string{}, TenantId(), key)); +} + +// Every thread's keys, published before anything is timed, and the strong +// handle of the entry each key currently holds. Keeping the handles is what +// lets a publisher drop the slot it published before, and what keeps the +// entries alive so the map holds every node it has rehashed. +std::vector> Fill(ObjectIndex& route, + size_t thread) { + std::vector> entries; + entries.reserve(kKeysPerThread); + for (size_t i = 0; i < kKeysPerThread; ++i) { + auto entry = MakeEntry(KeyOf(thread, i)); + const bool published = route.Insert(entry); + assert(published); + (void)published; + entries.push_back(std::move(entry)); + } + return entries; +} + +enum class Mode { kGet, kPublish, kMixed }; + +const char* ModeName(Mode mode) { + switch (mode) { + case Mode::kGet: + return "get"; + case Mode::kPublish: + return "publish"; + case Mode::kMixed: + return "mixed"; + } + return "unknown"; +} + +// What one thread does until it is told to stop: read its own keys, or cycle +// over them publishing each one, with eight lookups before each publish in the +// mixed mode. The worker fills the route and then waits at the start barrier, +// so the window opens for every worker at once and the fill is not charged to +// it: a fill that serializes more (fewer stripes) would otherwise be paid for +// out of the measured window while its inserts are not counted. +void RunWorker(ObjectIndex& route, size_t thread, Mode mode, + std::atomic& total_ops, std::atomic& stop, + std::barrier<>& start_barrier) { + std::vector keys; + keys.reserve(kKeysPerThread); + for (size_t i = 0; i < kKeysPerThread; ++i) { + keys.push_back(KeyOf(thread, i)); + } + std::vector> entries = Fill(route, thread); + start_barrier.arrive_and_wait(); + uint64_t ops = 0; + size_t at = 0; + while (!stop.load(std::memory_order_relaxed)) { + if (mode == Mode::kGet) { + for (const auto& key : keys) { + const auto entry = route.Get(key); + assert(entry != nullptr); + (void)entry; + } + ops += keys.size(); + continue; + } + const size_t slot = at % keys.size(); + if (mode == Mode::kMixed) { + for (size_t i = 0; i < kLookupsPerPublish; ++i) { + const auto entry = route.Get(keys[(slot + i) % keys.size()]); + assert(entry != nullptr); + (void)entry; + } + ops += kLookupsPerPublish; + } + // A publish drops the slot this key held and publishes a fresh entry, + // counting as one operation either way. + const bool dropped = route.EraseIf(keys[slot], entries[slot]); + assert(dropped); + (void)dropped; + auto fresh = MakeEntry(keys[slot]); + const bool published = route.Insert(fresh); + assert(published); + (void)published; + entries[slot] = std::move(fresh); + ++ops; + ++at; + } + total_ops.fetch_add(ops, std::memory_order_relaxed); +} + +void RunPoint(size_t stripe_count, int threads, Mode mode) { + ObjectIndex route{stripe_count}; + std::atomic total_ops{0}; + std::atomic stop{false}; + // Every worker fills the route and reaches the barrier before the window + // opens, so the fill is neither counted nor charged to the measurement. + std::barrier start_barrier(threads + 1); + std::vector workers; + workers.reserve(threads); + for (int t = 0; t < threads; ++t) { + workers.emplace_back([&, t] { + RunWorker(route, static_cast(t), mode, total_ops, stop, + start_barrier); + }); + } + start_barrier.arrive_and_wait(); + const auto start = std::chrono::steady_clock::now(); + while (std::chrono::steady_clock::now() - start < kMeasureFor) { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + stop.store(true, std::memory_order_relaxed); + for (auto& worker : workers) { + worker.join(); + } + // The rate is over the window the workers actually ran in, which is the + // sleeping window plus the time the last worker took to notice the stop. + const double seconds = + std::chrono::duration(std::chrono::steady_clock::now() - start) + .count(); + std::cout << stripe_count << "," << threads << "," << ModeName(mode) << "," + << static_cast(total_ops.load() / seconds) << "," + << RouteBytes(stripe_count) << std::endl; +} + +// The same publish path without the erase in front of it, one thread, into a +// route that keeps growing: the tail of this table is one stripe's rehash walk. +void RunInsertStall(size_t stripe_count) { + ObjectIndex route{stripe_count}; + std::vector> entries; + entries.reserve(kStallKeys); + std::vector latencies_ns; + latencies_ns.reserve(kStallKeys); + for (size_t i = 0; i < kStallKeys; ++i) { + auto entry = MakeEntry(KeyOf(0, i)); + const auto before = std::chrono::steady_clock::now(); + const bool published = route.Insert(entry); + const auto after = std::chrono::steady_clock::now(); + assert(published); + (void)published; + latencies_ns.push_back(static_cast( + std::chrono::duration_cast(after - before) + .count())); + entries.push_back(std::move(entry)); + } + std::sort(latencies_ns.begin(), latencies_ns.end()); + const auto percentile = [&latencies_ns](double share) { + const double rank = + share * static_cast(latencies_ns.size() - 1); + return static_cast(latencies_ns[static_cast(rank)]) / + 1000.0; + }; + std::cout << stripe_count << "," << kStallKeys << "," << std::fixed + << std::setprecision(1) << percentile(0.50) << "," + << percentile(0.99) << "," + << static_cast(latencies_ns.back()) / 1000.0 << std::endl; +} + +} // namespace +} // namespace mooncake + +int main() { + std::cout << "stripes,threads,mode,ops_per_s,route_bytes" << std::endl; + for (size_t stripe_count : mooncake::kStripeCounts) { + for (int threads : mooncake::kThreadCounts) { + for (mooncake::Mode mode : + {mooncake::Mode::kGet, mooncake::Mode::kPublish, + mooncake::Mode::kMixed}) { + mooncake::RunPoint(stripe_count, threads, mode); + } + } + } + std::cout << std::endl; + std::cout << "stripes,keys,p50_us,p99_us,max_us" << std::endl; + for (size_t stripe_count : mooncake::kStripeCounts) { + mooncake::RunInsertStall(stripe_count); + } + return 0; +} diff --git a/mooncake-store/include/metadata/tenant.h b/mooncake-store/include/metadata/tenant.h index a9cbb84593..21fc649057 100644 --- a/mooncake-store/include/metadata/tenant.h +++ b/mooncake-store/include/metadata/tenant.h @@ -10,9 +10,9 @@ // its group membership with the slot. Mutating a published object goes through // `WithPublishedObject`, which re-checks that identity under the entry lock. // -// Every method synchronizes internally. Lock order: entry lock, route lock, -// then the group table. Only `RemoveObject` holds the route lock across the -// group table, and nothing takes the route lock while holding a group stripe. +// Every method synchronizes internally. Lock order: entry lock, route stripe, +// then the group table. Only `RemoveObject` holds a route stripe across the +// group table, and nothing takes a route stripe while holding a group stripe. #include #include @@ -69,8 +69,8 @@ class Tenant { // Removes a torn-down object: its route slot and its group membership, // only while the slot still holds `entry`; false, touching nothing, - // otherwise. Both go under one hold of the route lock, because membership - // is keyed by the object key: once the slot is released, a newer + // otherwise. Both go under one hold of the key's route stripe, because + // membership is keyed by the object key: once the slot is released, a newer // publication of the same key can register the membership this teardown // would drop. Callers hold the entry's own lock, so a publication still // wiring its state cannot be torn down half-registered. @@ -78,7 +78,7 @@ class Tenant { if (entry == nullptr) { return false; } - return object_index_.WithExclusiveRoute([&](auto& route) { + return object_index_.WithExclusiveRoute(entry->key(), [&](auto& route) { const auto it = route.find(entry->key()); if (it == route.end() || it->second != entry) { return false; diff --git a/mooncake-store/include/object_index.h b/mooncake-store/include/object_index.h index 6460560069..85c8090fd7 100644 --- a/mooncake-store/include/object_index.h +++ b/mooncake-store/include/object_index.h @@ -16,20 +16,52 @@ namespace mooncake { -// The object route for one tenant: a flat map from object key to a strong -// ObjectEntry handle. +// The object route for one tenant: a key to a strong ObjectEntry handle. // // Identity is the handle: a lookup that hands back the same handle names the // same publication, and an entry instance stands for exactly one publication, // so comparing handles is the whole of the identity check. +// +// The route is striped: which stripe a key lands in is a function of the key +// alone, so a per-key operation still takes exactly one lock, and a stripe is +// what the route's own cost is charged to. A map that outgrows its buckets +// rehashes every node it holds under its exclusive lock, and that walk is +// proportional to the nodes, so striping charges one key's growth to that +// stripe's keys rather than to every key the tenant holds. A stripe is a lock +// and a map, about 120 bytes, so the count trades a tenant's memory for write +// concurrency and for the length of that walk. Measurements are in +// benchmarks/object_index_contention_bench.cpp. class ObjectIndex { public: + // 64 stripes: the shipped default. What the sweep settles is a ratio rather + // than a count. With 32 writer threads on a 256-core host, one stripe per + // 512 keys still came out ahead of one per 2,048: 256 stripes reached 114M + // lookups and 9.1M publishes per second against 79M and 6.7M in 64, and one + // stripe reaches 8.5M and 250k, with the worst single publish into a + // growing route falling from 20 to 30 ms to 0.8 to 2.6 ms. So 64 stripes + // are that ratio for a tenant of about thirty thousand objects, and a + // tenant holding fewer keys cannot use more stripes than it has while one + // holding far more would want more: the count follows the tenant's size, + // and this constant stands in for one size. It costs such a tenant + // about 7.7 kB, against 30.7 kB at 256, which a tenant of a many-tenant + // service would pay for concurrency it does not have. + static constexpr size_t kStripeCount = 64; + + // The count is an argument so the measurement can build the same route at + // another count; the tests take it too, and every other caller takes the + // default. Nothing reads it from configuration. + explicit ObjectIndex(size_t stripe_count = kStripeCount) + : stripes_(stripe_count) { + assert(stripe_count > 0); + } + // nullptr when the key is absent. The returned handle is strong, so it - // keeps the entry alive after the route lock is released. + // keeps the entry alive after the stripe lock is released. [[nodiscard]] std::shared_ptr Get(std::string_view key) const { - std::shared_lock lock(route_lock_); - auto it = route_.find(key); - return it == route_.end() ? nullptr : it->second; + const Stripe& stripe = StripeOf(key); + std::shared_lock lock(stripe.mutex); + const auto it = stripe.route.find(key); + return it == stripe.route.end() ? nullptr : it->second; } // Publish `entry` under its own key. Returns false when the key is @@ -46,9 +78,10 @@ class ObjectIndex { [[nodiscard]] bool Insert(std::shared_ptr entry) { assert(entry != nullptr); assert(!entry->IsPublished()); - std::unique_lock lock(route_lock_); + Stripe& stripe = StripeOf(entry->key()); + std::unique_lock lock(stripe.mutex); const auto [it, inserted] = - route_.try_emplace(entry->key(), std::move(entry)); + stripe.route.try_emplace(entry->key(), std::move(entry)); if (!inserted) { return false; } @@ -64,12 +97,13 @@ class ObjectIndex { // null one never matches. [[nodiscard]] bool EraseIf(std::string_view key, const std::shared_ptr& expected) { - std::unique_lock lock(route_lock_); - const auto it = route_.find(key); - if (it == route_.end() || it->second != expected) { + Stripe& stripe = StripeOf(key); + std::unique_lock lock(stripe.mutex); + const auto it = stripe.route.find(key); + if (it == stripe.route.end() || it->second != expected) { return false; } - route_.erase(it); + stripe.route.erase(it); return true; } @@ -79,61 +113,112 @@ class ObjectIndex { if (entry == nullptr) { return false; } - std::shared_lock lock(route_lock_); - const auto it = route_.find(key); - return it != route_.end() && it->second == entry; + const Stripe& stripe = StripeOf(key); + std::shared_lock lock(stripe.mutex); + const auto it = stripe.route.find(key); + return it != stripe.route.end() && it->second == entry; } - // Runs `fn(route)` with the route held exclusively. The tenant layer uses - // this to decide which publication owns a key's records and drop those - // records in one section, so a publication that replaced the one being torn - // down cannot have its records dropped in between. + // Runs `fn(route)` with the stripe that holds `key` held exclusively. The + // tenant layer uses this to decide which publication owns a key's records + // and drop those records in one section, so a publication that replaced the + // one being torn down cannot have its records dropped in between. template - decltype(auto) WithExclusiveRoute(Fn&& fn) { - std::unique_lock lock(route_lock_); - return std::forward(fn)(route_); + decltype(auto) WithExclusiveRoute(std::string_view key, Fn&& fn) { + Stripe& stripe = StripeOf(key); + std::unique_lock lock(stripe.mutex); + return std::forward(fn)(stripe.route); } [[nodiscard]] bool Contains(std::string_view key) const { - std::shared_lock lock(route_lock_); - return route_.contains(key); + const Stripe& stripe = StripeOf(key); + std::shared_lock lock(stripe.mutex); + return stripe.route.contains(key); } + // Counted stripe by stripe, so the answer is the route's size at no one + // instant: a key added to another stripe while this walks is counted. [[nodiscard]] size_t ObjectCount() const { - std::shared_lock lock(route_lock_); - return route_.size(); + size_t count = 0; + for (const Stripe& stripe : stripes_) { + std::shared_lock lock(stripe.mutex); + count += stripe.route.size(); + } + return count; } - // True when no object is currently routed. + // True when no object is currently routed. Checked stripe by stripe for the + // same reason. [[nodiscard]] bool Empty() const { - std::shared_lock lock(route_lock_); - return route_.empty(); + for (const Stripe& stripe : stripes_) { + std::shared_lock lock(stripe.mutex); + if (!stripe.route.empty()) { + return false; + } + } + return true; } - // Collect strong handles to every object currently routed. The route lock - // is released before returning, so a caller can take each entry's own - // mutex without holding the route lock. + // Collect strong handles to every object currently routed. Each stripe's + // lock is released before returning, so a caller can take each entry's own + // mutex without holding any of them, and the result is a collection rather + // than the route at one instant. + // + // The handles are sized in one pass before any of them is copied: a + // reservation that grew with each stripe would re-copy every handle + // collected so far, and it would do that inside that stripe's lock. A route + // that grows while this runs can still outgrow the reservation, and then + // one growth happens with the handles already collected. [[nodiscard]] std::vector> SnapshotObjects() const { - std::shared_lock lock(route_lock_); + size_t total = 0; + for (const Stripe& stripe : stripes_) { + std::shared_lock lock(stripe.mutex); + total += stripe.route.size(); + } + std::vector> entries; - entries.reserve(route_.size()); - for (const auto& entry : route_) { - entries.push_back(entry.second); + entries.reserve(total); + for (const Stripe& stripe : stripes_) { + std::shared_lock lock(stripe.mutex); + for (const auto& entry : stripe.route) { + entries.push_back(entry.second); + } } return entries; } + // The number of stripes this route was built with. + [[nodiscard]] size_t StripeCount() const { return stripes_.size(); } + private: - // Object route: the strong entry handles keyed by object key, guarded by a - // single shared_mutex. Mutating one object's state is finer-grained: each + // One stripe of the route: the strong entry handles keyed by object key + // under its own lock. Mutating one object's state is finer-grained: each // entry guards its own. - mutable std::shared_mutex route_lock_; - // Transparent lookup: every accessor takes a view, so a caller that already - // has one does not build a string to find the entry. - std::unordered_map, - TransparentStringHash, std::equal_to<>> - route_; + struct Stripe { + mutable std::shared_mutex mutex; + // Transparent lookup: every accessor takes a view, so a caller that + // already has one does not build a string to find the entry. + std::unordered_map, + TransparentStringHash, std::equal_to<>> + route; + }; + + // The same hash the table uses, masked by the count the route was built + // with, so a key reaches one stripe whether the caller holds a string or a + // view. + [[nodiscard]] size_t StripeIndex(std::string_view key) const { + return TransparentStringHash{}(key) % stripes_.size(); + } + + [[nodiscard]] Stripe& StripeOf(std::string_view key) { + return stripes_[StripeIndex(key)]; + } + [[nodiscard]] const Stripe& StripeOf(std::string_view key) const { + return stripes_[StripeIndex(key)]; + } + + std::vector stripes_; }; } // namespace mooncake diff --git a/mooncake-store/tests/object_index_test.cpp b/mooncake-store/tests/object_index_test.cpp index 602fe45ee5..59306a521b 100644 --- a/mooncake-store/tests/object_index_test.cpp +++ b/mooncake-store/tests/object_index_test.cpp @@ -2,7 +2,11 @@ #include "object_test_helpers.h" #include +#include +#include +#include #include +#include #include #include @@ -10,6 +14,10 @@ namespace mooncake { namespace { +// The stripe counts a route can be built with: one stripe makes every key share +// one lock, the rest spread keys over that many. +constexpr size_t kStripeCounts[] = {1, 4, 16, 64}; + TEST(ObjectIndexTest, InsertPublishesTheEntryUnderItsKey) { ObjectIndex store; EXPECT_EQ(store.ObjectCount(), 0u); @@ -114,5 +122,178 @@ TEST(ObjectIndexTest, SnapshotObjectsEnumeratesEveryEntry) { EXPECT_EQ(keys, (std::vector{"k1", "k2", "k3"})); } +TEST(ObjectIndexTest, EveryStripeCountRoutesTheSameWay) { + for (const size_t count : kStripeCounts) { + ObjectIndex store{count}; + EXPECT_EQ(store.StripeCount(), count) << count; + EXPECT_TRUE(store.Empty()) << count; + EXPECT_EQ(store.ObjectCount(), 0u) << count; + + auto entry = test::MakeObjectEntry("k1"); + ASSERT_TRUE(store.Insert(entry)) << count; + EXPECT_TRUE(store.Contains("k1")) << count; + EXPECT_EQ(store.ObjectCount(), 1u) << count; + EXPECT_FALSE(store.Empty()) << count; + + // Whichever stripe the key landed in, the route hands back the same + // handle. + EXPECT_EQ(store.Get("k1").get(), entry.get()) << count; + EXPECT_EQ(store.Get("missing"), nullptr) << count; + EXPECT_TRUE(store.IsCurrent("k1", entry)) << count; + + // Identity still decides the erase, at any count. + EXPECT_FALSE(store.EraseIf("k1", test::MakeObjectEntry("k1"))) << count; + EXPECT_TRUE(store.Contains("k1")) << count; + EXPECT_TRUE(store.EraseIf("k1", entry)) << count; + EXPECT_FALSE(store.IsCurrent("k1", entry)) << count; + EXPECT_TRUE(store.Empty()) << count; + EXPECT_EQ(store.ObjectCount(), 0u) << count; + } +} + +TEST(ObjectIndexTest, AKeyLandsInOneStripeHoweverOftenItIsReached) { + for (const size_t count : kStripeCounts) { + ObjectIndex store{count}; + auto entry = test::MakeObjectEntry("k1"); + ASSERT_TRUE(store.Insert(entry)) << count; + + // Every accessor hashes the key the same way, so repeated lookups all + // reach the stripe the insert placed the entry in. + for (int i = 0; i < 64; ++i) { + EXPECT_TRUE(store.Contains("k1")) << count; + EXPECT_EQ(store.Get("k1").get(), entry.get()) << count; + EXPECT_TRUE(store.IsCurrent("k1", entry)) << count; + } + + // The snapshot walks every stripe, so the key every lookup found is the + // one handle it hands back. + const auto snapshot = store.SnapshotObjects(); + ASSERT_EQ(snapshot.size(), 1u) << count; + EXPECT_EQ(snapshot[0].get(), entry.get()) << count; + + EXPECT_TRUE(store.EraseIf("k1", entry)) << count; + EXPECT_TRUE(store.Empty()) << count; + } +} + +// The handles a snapshot hands back are strong: nothing but the collection +// itself has to keep the publication alive, so the route slot can go while what +// the snapshot named is still readable. +TEST(ObjectIndexTest, SnapshotObjectsHoldsTheHandlesItCollected) { + ObjectIndex store; + auto entry = test::MakeObjectEntry("k1"); + ASSERT_TRUE(store.Insert(entry)); + + const auto snapshot = store.SnapshotObjects(); + ASSERT_EQ(snapshot.size(), 1u); + // The caller drops its own reference: the route slot and the snapshot are + // what hold the entry now. + entry.reset(); + EXPECT_EQ(snapshot[0]->key(), "k1"); + + ASSERT_TRUE(store.EraseIf("k1", snapshot[0])); + EXPECT_TRUE(store.Empty()); + EXPECT_EQ(snapshot[0]->key(), "k1"); + EXPECT_FALSE(store.IsCurrent("k1", snapshot[0])); + EXPECT_FALSE(store.Contains("k1")); + EXPECT_EQ(store.Get("k1"), nullptr); +} + +// A collection taken while the route is mutated. Each stripe is copied under +// its own lock, so a key that lands in one stripe is named at most once and +// every handle handed back was routed when its stripe was copied; which instant +// each stripe contributes is not fixed, because the stripes are walked one +// after another. This pins what the walk guarantees while it overlaps a mutator +// that keeps replacing keys. +TEST(ObjectIndexTest, SnapshotObjectsOverlappingMutationNamesEachKeyOnce) { + constexpr size_t kKeys = 512; + constexpr int kCollections = 200; + ObjectIndex store{16}; + + std::vector> live; + live.reserve(kKeys); + for (size_t i = 0; i < kKeys; ++i) { + live.push_back(test::MakeObjectEntry("k" + std::to_string(i))); + ASSERT_TRUE(store.Insert(live[i])); + } + + std::atomic stop{false}; + std::atomic failures{0}; + std::thread mutating([&] { + size_t at = 0; + while (!stop.load(std::memory_order_relaxed)) { + const std::string key = live[at]->key(); + auto fresh = test::MakeObjectEntry(key); + if (!store.EraseIf(key, live[at]) || !store.Insert(fresh)) { + failures.fetch_add(1, std::memory_order_relaxed); + } + live[at] = std::move(fresh); + at = (at + 1) % live.size(); + } + }); + + for (int round = 0; round < kCollections; ++round) { + const auto snapshot = store.SnapshotObjects(); + std::vector keys; + keys.reserve(snapshot.size()); + for (const auto& entry : snapshot) { + ASSERT_NE(entry, nullptr); + keys.push_back(entry->key()); + } + std::sort(keys.begin(), keys.end()); + EXPECT_EQ(std::adjacent_find(keys.begin(), keys.end()), keys.end()) + << "a key landed in two stripes, or one was walked twice"; + } + + stop.store(true, std::memory_order_relaxed); + mutating.join(); + EXPECT_EQ(failures.load(), 0); + // The mutator only ever replaces a key, so the route holds every one of + // them however the collections overlapped it. + EXPECT_EQ(store.ObjectCount(), kKeys); +} + +TEST(ObjectIndexTest, ConcurrentInsertOfDistinctKeysPublishesEveryOne) { + constexpr size_t kThreads = 4; + constexpr size_t kKeysPerThread = 500; + + for (const size_t count : kStripeCounts) { + ObjectIndex store{count}; + // Keys are distinct and each thread owns its own, so no two threads + // contend for the same route slot and every insert publishes. + std::vector> entries; + entries.reserve(kThreads * kKeysPerThread); + for (size_t t = 0; t < kThreads; ++t) { + for (size_t i = 0; i < kKeysPerThread; ++i) { + entries.push_back(test::MakeObjectEntry( + "t" + std::to_string(t) + "_k" + std::to_string(i))); + } + } + + std::atomic rejected{0}; + std::vector threads; + threads.reserve(kThreads); + for (size_t t = 0; t < kThreads; ++t) { + threads.emplace_back([&, t] { + for (size_t i = 0; i < kKeysPerThread; ++i) { + if (!store.Insert(entries[t * kKeysPerThread + i])) { + rejected.fetch_add(1, std::memory_order_relaxed); + } + } + }); + } + for (auto& thread : threads) { + thread.join(); + } + + EXPECT_EQ(rejected.load(), 0) << count; + EXPECT_EQ(store.ObjectCount(), entries.size()) << count; + EXPECT_FALSE(store.Empty()) << count; + for (const auto& entry : entries) { + EXPECT_EQ(store.Get(entry->key()).get(), entry.get()) << count; + } + } +} + } // namespace } // namespace mooncake