Test AtomicSharedPtr concurrency

include/tsutil/AtomicSharedPtr.h

@@ -27,24 +27,22 @@
 #include <atomic>
 #include <memory>
 
-// Use the C++20 std::atomic<std::shared_ptr<T>> 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<std::shared_ptr<T>> 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 <class T> using AtomicSharedPtr = std::atomic<std::shared_ptr<T>>;
 
 #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"
 

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

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 <catch2/catch_test_macros.hpp>
+
+#include "tsutil/AtomicSharedPtr.h"
+
+#include <algorithm>
+#include <atomic>
+#include <chrono>
+#include <memory>
+#include <thread>
+#include <vector>
+
+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<bool> should_start{false};
+  std::atomic<bool> should_stop{false};
+  std::atomic<int>  invalid_reads{0};
+  std::atomic<int>  read_count{0};
+};
+
+void
+run_reader(AtomicSharedPtr<Payload> &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<std::thread>
+make_readers(int reader_count, AtomicSharedPtr<Payload> &ptr, ReaderState &state)
+{
+  std::vector<std::thread> 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<std::thread> &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<int> ptr;
+
+  CHECK(ptr.load() == nullptr);
+
+  auto first = std::make_shared<int>(1);
+  ptr.store(first);
+  CHECK(ptr.load() == first);
+  CHECK(*ptr.load() == 1);
+
+  auto second   = std::make_shared<int>(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<Payload> ptr{std::make_shared<Payload>(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<Payload>(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);
+}