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