diff --git a/src/kv/store.h b/src/kv/store.h index f82df0588f9..46c0fd7cb16 100644 --- a/src/kv/store.h +++ b/src/kv/store.h @@ -121,6 +121,12 @@ namespace ccf::kv ccf::kv::ConsensusHookPtrs& hooks, bool track_deletes_on_missing_keys) override { + std::unique_lock maps_guard(maps_lock, std::defer_lock); + if (!new_maps.empty()) + { + maps_guard.lock(); + } + auto c = apply_changes( changes, [v](bool) { return std::make_tuple(v, v - 1); }, diff --git a/src/kv/test/kv_dynamic_tables.cpp b/src/kv/test/kv_dynamic_tables.cpp index f75e518f0c0..4f22565b277 100644 --- a/src/kv/test/kv_dynamic_tables.cpp +++ b/src/kv/test/kv_dynamic_tables.cpp @@ -4,7 +4,11 @@ #include "kv/test/null_encryptor.h" #include "kv/test/stub_consensus.h" +#include #include +#include +#include +#include struct MapTypes { @@ -387,6 +391,126 @@ TEST_CASE("Dynamic map serialisation" * doctest::test_suite("dynamic")) } } +TEST_CASE( + "Concurrent deserialised dynamic map publication" * + doctest::test_suite("dynamic")) +{ + constexpr size_t map_count = 256; + constexpr size_t reader_count = 4; + + auto consensus = std::make_shared(); + auto encryptor = std::make_shared(); + + ccf::kv::Store source_store; + source_store.set_encryptor(encryptor); + source_store.set_consensus(consensus); + + std::vector> entries; + entries.reserve(map_count); + + // Keep map comparisons in progress long enough for TSAN to observe overlap. + const std::string map_name_prefix(4096, 'm'); + for (size_t i = 0; i < map_count; ++i) + { + auto tx = source_store.create_tx(); + auto handle = + tx.rw(map_name_prefix + std::to_string(i)); + handle->put("key", "value"); + REQUIRE(tx.commit() == ccf::kv::CommitResult::SUCCESS); + + const auto latest_data = consensus->get_latest_data(); + REQUIRE(latest_data.has_value()); + entries.push_back(latest_data.value()); + } + + ccf::kv::Store target_store; + target_store.set_encryptor(encryptor); + + std::atomic ready = 0; + std::atomic start = false; + std::atomic done = false; + std::atomic readers_started = 0; + std::atomic reads = 0; + std::atomic all_entries_applied = true; + std::exception_ptr writer_exception; + + const auto searched_map_name = map_name_prefix + "not_present"; + std::vector readers; + readers.reserve(reader_count); + for (size_t i = 0; i < reader_count; ++i) + { + readers.emplace_back([&]() { + ready.fetch_add(1, std::memory_order_release); + while (!start.load(std::memory_order_acquire)) + { + std::this_thread::yield(); + } + + target_store.get_map(target_store.current_version(), searched_map_name); + reads.fetch_add(1, std::memory_order_relaxed); + readers_started.fetch_add(1, std::memory_order_release); + + while (!done.load(std::memory_order_acquire)) + { + target_store.get_map(target_store.current_version(), searched_map_name); + reads.fetch_add(1, std::memory_order_relaxed); + } + }); + } + + std::thread writer([&]() { + ready.fetch_add(1, std::memory_order_release); + while (!start.load(std::memory_order_acquire)) + { + std::this_thread::yield(); + } + while (readers_started.load(std::memory_order_acquire) != reader_count) + { + std::this_thread::yield(); + } + + try + { + for (const auto& entry : entries) + { + if ( + target_store.deserialize(entry)->apply() != + ccf::kv::ApplyResult::PASS) + { + all_entries_applied.store(false, std::memory_order_relaxed); + break; + } + } + } + catch (...) + { + writer_exception = std::current_exception(); + } + done.store(true, std::memory_order_release); + }); + + while (ready.load(std::memory_order_acquire) != reader_count + 1) + { + std::this_thread::yield(); + } + start.store(true, std::memory_order_release); + + writer.join(); + for (auto& reader : readers) + { + reader.join(); + } + + if (writer_exception) + { + std::rethrow_exception(writer_exception); + } + + REQUIRE(all_entries_applied.load(std::memory_order_relaxed)); + REQUIRE(reads.load(std::memory_order_relaxed) >= reader_count); + REQUIRE(target_store.current_version() == map_count); +} + TEST_CASE("Dynamic map snapshot serialisation" * doctest::test_suite("dynamic")) { ccf::kv::Store store;