diff --git a/include/tsutil/AtomicSharedPtr.h b/include/tsutil/AtomicSharedPtr.h index c7138a8e575..c6838a7d74b 100644 --- a/include/tsutil/AtomicSharedPtr.h +++ b/include/tsutil/AtomicSharedPtr.h @@ -27,24 +27,22 @@ #include #include -// Use the C++20 std::atomic> specialization when the -// standard library provides it, otherwise fall back to the pre-C++20 -// std::atomic_*_explicit free-function overloads on shared_ptr. The -// fallback exists for libstdc++ < 12 and libc++ < 14, which predate the -// specialization. When those toolchains are no longer supported, delete -// the #else branch and the surrounding #if; call sites do not change. +// Use the C++20 std::atomic> specialization when its +// feature-test macro reports support, otherwise fall back to the pre-C++20 +// std::atomic_*_explicit free-function overloads on shared_ptr. When all +// supported toolchains provide the specialization, delete the #else branch +// and the surrounding #if; call sites do not change. #if defined(__cpp_lib_atomic_shared_ptr) && __cpp_lib_atomic_shared_ptr >= 201711L template using AtomicSharedPtr = std::atomic>; #else -// Belt-and-suspenders: on the toolchains that take this branch (libstdc++ -// < 12, libc++ < 16) the free-function overloads are not yet marked -// [[deprecated]], so the suppression below is usually a no-op. It -// matters only if someone forces the fallback on a modern library (e.g. -// -D__cpp_lib_atomic_shared_ptr=0) or compiles against a library that -// ships the deprecation markers ahead of the specialization. +// Belt-and-suspenders: on toolchains that take this branch, the free-function +// overloads are normally not marked [[deprecated]], so the suppression below is +// usually a no-op. It matters only if someone forces the fallback on a modern +// library (e.g. -D__cpp_lib_atomic_shared_ptr=0) or compiles against a library +// that ships the deprecation markers ahead of the specialization. #pragma GCC diagnostic push #pragma GCC diagnostic ignored "-Wdeprecated-declarations" diff --git a/src/tsutil/CMakeLists.txt b/src/tsutil/CMakeLists.txt index 7d0e9a651ca..0526007351d 100644 --- a/src/tsutil/CMakeLists.txt +++ b/src/tsutil/CMakeLists.txt @@ -74,6 +74,7 @@ endif() if(BUILD_TESTING) add_executable( test_tsutil + unit_tests/test_AtomicSharedPtr.cc unit_tests/test_Bravo.cc unit_tests/test_LocalBuffer.cc unit_tests/test_Metrics.cc diff --git a/src/tsutil/unit_tests/test_AtomicSharedPtr.cc b/src/tsutil/unit_tests/test_AtomicSharedPtr.cc new file mode 100644 index 00000000000..4954221c442 --- /dev/null +++ b/src/tsutil/unit_tests/test_AtomicSharedPtr.cc @@ -0,0 +1,162 @@ +/** @file + + Unit tests for AtomicSharedPtr + + @section license License + + Licensed to the Apache Software Foundation (ASF) under one + or more contributor license agreements. See the NOTICE file + distributed with this work for additional information + regarding copyright ownership. The ASF licenses this file + to you under the Apache License, Version 2.0 (the + "License"); you may not use this file except in compliance + with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. + */ + +#include + +#include "tsutil/AtomicSharedPtr.h" + +#include +#include +#include +#include +#include +#include + +namespace +{ +struct Payload { + static constexpr size_t VALUE_COUNT = 8; + + explicit Payload(int generation) : generation_(generation) { std::fill(std::begin(values_), std::end(values_), generation); } + + bool + is_valid() const + { + return std::all_of(std::begin(values_), std::end(values_), [this](int value) { return value == generation_; }); + } + + int generation_ = 0; + int values_[VALUE_COUNT]; +}; + +struct ReaderState { + std::atomic should_start{false}; + std::atomic should_stop{false}; + std::atomic invalid_reads{0}; + std::atomic read_count{0}; +}; + +void +run_reader(AtomicSharedPtr &ptr, ReaderState &state) +{ + while (!state.should_start.load(std::memory_order_acquire)) { + std::this_thread::yield(); + } + + while (!state.should_stop.load(std::memory_order_acquire)) { + auto current = ptr.load(std::memory_order_acquire); + if (current == nullptr || !current->is_valid()) { + state.invalid_reads.fetch_add(1, std::memory_order_relaxed); + } + auto const reads = state.read_count.fetch_add(1, std::memory_order_release) + 1; + if (reads % 64 == 0) { + std::this_thread::yield(); + } + } +} + +std::vector +make_readers(int reader_count, AtomicSharedPtr &ptr, ReaderState &state) +{ + std::vector readers; + + readers.reserve(reader_count); + for (int i = 0; i < reader_count; ++i) { + readers.emplace_back([&ptr, &state] { run_reader(ptr, state); }); + } + return readers; +} + +bool +wait_for_reader(const ReaderState &state, std::chrono::steady_clock::duration timeout) +{ + auto const deadline = std::chrono::steady_clock::now() + timeout; + + while (std::chrono::steady_clock::now() < deadline) { + if (state.read_count.load(std::memory_order_acquire) > 0) { + return true; + } + std::this_thread::yield(); + } + return false; +} + +void +stop_readers(ReaderState &state, std::vector &readers) +{ + state.should_stop.store(true, std::memory_order_release); + for (auto &reader : readers) { + if (reader.joinable()) { + reader.join(); + } + } +} +} // end anonymous namespace + +TEST_CASE("AtomicSharedPtr load store exchange", "[libts][AtomicSharedPtr]") +{ + AtomicSharedPtr ptr; + + CHECK(ptr.load() == nullptr); + + auto first = std::make_shared(1); + ptr.store(first); + CHECK(ptr.load() == first); + CHECK(*ptr.load() == 1); + + auto second = std::make_shared(2); + auto previous = ptr.exchange(second); + CHECK(previous == first); + CHECK(ptr.load() == second); + CHECK(*ptr.load() == 2); +} + +TEST_CASE("AtomicSharedPtr supports concurrent readers during writer swaps", "[libts][AtomicSharedPtr]") +{ + static constexpr int READER_COUNT = 8; + static constexpr int WRITE_COUNT = 5000; + + AtomicSharedPtr ptr{std::make_shared(0)}; + ReaderState state; + auto readers = make_readers(READER_COUNT, ptr, state); + + state.should_start.store(true, std::memory_order_release); + auto const reader_started = wait_for_reader(state, std::chrono::seconds(5)); + if (!reader_started) { + stop_readers(state, readers); + } + REQUIRE(reader_started); + + for (int generation = 1; generation <= WRITE_COUNT; ++generation) { + ptr.store(std::make_shared(generation), std::memory_order_release); + if (generation % 64 == 0) { + std::this_thread::yield(); + } + } + stop_readers(state, readers); + + CHECK(state.invalid_reads.load() == 0); + CHECK(state.read_count.load() > 0); + REQUIRE(ptr.load() != nullptr); + CHECK(ptr.load()->generation_ == WRITE_COUNT); +}