From 76a184a3d391d9f4bd6f1bdcd18aea19491e1428 Mon Sep 17 00:00:00 2001 From: Jie Yao Date: Thu, 10 Sep 2026 17:26:03 +0800 Subject: [PATCH 1/7] move some common headers from homestore into sisl --- include/sisl/async/coro.hpp | 89 ++++++++++--- include/sisl/fds/bounded_mpmc_queue.hpp | 64 ++++++++++ src/fds/CMakeLists.txt | 7 + src/fds/tests/test_bounded_mpmc_queue.cpp | 149 ++++++++++++++++++++++ 4 files changed, 293 insertions(+), 16 deletions(-) create mode 100644 include/sisl/fds/bounded_mpmc_queue.hpp create mode 100644 src/fds/tests/test_bounded_mpmc_queue.cpp diff --git a/include/sisl/async/coro.hpp b/include/sisl/async/coro.hpp index 1a0590c8..98204556 100644 --- a/include/sisl/async/coro.hpp +++ b/include/sisl/async/coro.hpp @@ -22,11 +22,43 @@ namespace sisl::async { -// Block the calling thread until the task completes and return its value (void for task). Do NOT call on -// an event-loop / reactor thread. +// Return a task that awaits a heap-held completion. Used by the "do a synchronous side effect, then return a +// task that awaits the eventual completion" pattern (e.g. CP trigger / open_log_store: the switchover or map +// insert must run at call time, only the wait is deferred). The shared_ptr is copied into the task frame so the +// completion outlives the await. +template < typename T > +inline task< T > await_shared(std::shared_ptr< shared_awaitable< T > > aw) { + co_return co_await *aw; +} +template < typename T > +inline task< T > await_value(std::shared_ptr< value_awaitable< T > > aw) { + co_return co_await *aw; +} + +// Await a value_awaitable held by reference (e.g. a long-lived member of a heap object that outlives the await). +// The reference is bound into the returned task's frame; the awaitable must stay alive until this task completes. +template < typename T > +inline task< T > await_value_ref(value_awaitable< T >& aw) { + co_return co_await aw; +} + +// write_env injects an inline scheduler so the sticky-affinity exec::task can be started without an enclosing +// scheduler context (it resumes inline on whatever thread completes its awaited sender); start_detached owns the +// operation-state on the heap and frees it on completion. Same pattern as iomgr's io_launch.hpp and sisl's +// when_all. +template < typename Task > +inline void start_coro(Task&& t) { + stdexec::start_detached( + stdexec::write_env(std::forward< Task >(t), stdexec::prop{stdexec::get_scheduler, exec::inline_scheduler{}})); +} + +// Block the calling thread until the task completes and return its value. For the infrequent control-plane and +// shutdown paths that are synchronous today (e.g. a forced CP flush awaited before proceeding). The task is +// fulfilled by other threads, so this drains its run_loop here without self-deadlocking a data path. template < typename Task > inline auto sync_get(Task&& task) { auto result = stdexec::sync_wait(std::forward< Task >(task)).value(); + // result is a tuple of the task's completion values; for task it is empty (nothing to return). if constexpr (std::tuple_size_v< decltype(result) > == 0) { return; } else { @@ -34,22 +66,47 @@ inline auto sync_get(Task&& task) { } } -// Fire-and-forget a coroutine whose result we don't need. The task is taken by value (copied into the -// self-owning wrapper frame); the wrapper swallows exceptions so a throwing body can't reach start_detached's +// Block the calling thread until the task completes OR `timeout` elapses; returns true iff it completed in time. +// The task is started detached and signals a std::promise on completion, which we time-wait on via its future. +// On timeout the detached task remains pending -- it must keep alive whatever it awaits (e.g. by holding a strong +// ref into its frame); we simply stop waiting and leave it to complete (or leak) later. Used by the data-receive +// timeout path, which then inspects per-item readiness and remediates the stragglers. +template < typename Task > +inline bool sync_wait_for(Task&& task, std::chrono::milliseconds timeout) { + auto done = std::make_shared< std::promise< void > >(); + auto fut = done->get_future(); + start_coro([](std::decay_t< Task > t, std::shared_ptr< std::promise< void > > d) -> task< void > { + try { + co_await std::move(t); + } catch (...) {} + d->set_value(); + }(std::forward< Task >(task), std::move(done))); + return fut.wait_for(timeout) == std::future_status::ready; +} + +// Take the task BY VALUE so it is copied into the self-owning coroutine frame; a captured-by-reference task +// would dangle once start_detached returns. Swallows exceptions so a throwing body cannot reach start_detached's // receiver (which would std::terminate) -- tasks normally complete errors-as-values, so this is a backstop. -// write_env injects an inline scheduler so the sticky-affinity exec::task can start without an enclosing -// scheduler (it resumes inline on whatever thread completes its awaited work) -- the same idiom as when_all. template < typename T > -inline void detach(task< T > t) { - auto wrapper = [](task< T > inner) -> task< void > { - try { - co_await std::move(inner); - } catch (const std::exception& e) { LOGERROR("Detached task threw, swallowing: {}", e.what()); } catch (...) { - LOGERROR("Detached task threw an unknown exception, swallowing"); - } - }(std::move(t)); - stdexec::start_detached( - stdexec::write_env(std::move(wrapper), stdexec::prop{stdexec::get_scheduler, exec::inline_scheduler{}})); +inline task< void > detach_wrapper(task< T > t) { + try { + co_await std::move(t); + } catch (const std::exception& e) { LOGERROR("Detached task threw, swallowing: {}", e.what()); } catch (...) { + LOGERROR("Detached task threw an unknown exception, swallowing"); + } +} + +// Fire-and-forget a task whose result is not needed (e.g. a non-forced CP trigger). Starts it detached. +template < typename T > +inline void detach(task< T > task) { + start_coro(detach_wrapper< T >(std::move(task))); +} + +// Fire-and-forget a task but invoke fn(result) when it completes (the non-blocking ".thenValue(cb)" shape). +// fn runs on whatever thread completes the task. Both task and fn are copied into the self-owning frame. +template < typename T, typename Fn > +inline void detach_then(task< T > task, Fn fn) { + start_coro([](task< T > t, Fn f) -> task< void > { f(co_await std::move(t)); }(std::move(task), std::move(fn))); } } // namespace sisl::async diff --git a/include/sisl/fds/bounded_mpmc_queue.hpp b/include/sisl/fds/bounded_mpmc_queue.hpp new file mode 100644 index 00000000..daa9d9a5 --- /dev/null +++ b/include/sisl/fds/bounded_mpmc_queue.hpp @@ -0,0 +1,64 @@ +/********************************************************************************* + * Modifications Copyright 2017-2019 eBay Inc. + * + * Licensed 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 + * https://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. + * + *********************************************************************************/ +#pragma once + +#include +#include + +#include + +namespace sisl { + +// Folly-free replacement for folly::MPMCQueue: a capacity-bounded boost::lockfree MPMC queue plus an +// approximate size counter (folly exposed this as sizeGuess()). The queue is bounded -- write() returns false +// when full -- which blkalloc relies on (the slab cache spills to the next level on a full level, and bounds +// total cached free-blocks). We use the default (pointer-freelist) boost::lockfree::queue pre-reserved to +// `capacity` and push only via bounded_push() (never grows past the reserved nodes): this gives bounded +// behavior WITHOUT boost::lockfree::fixed_sized's hard 65535-element cap (16-bit freelist indices), +// which the blkalloc free-block / slab-cache capacities exceed. boost::lockfree::queue requires a +// trivially-copyable element type (blk_num_t, blk_cache_entry both satisfy this). The size counter is the +// central accounting a bounded+queryable queue inherently needs; boost::lockfree itself keeps no size. +template < typename T > +class BoundedMPMCQueue { +public: + explicit BoundedMPMCQueue(const size_t capacity) : m_q{capacity} {} + + // Non-blocking enqueue; returns false if the queue is full. (folly::MPMCQueue::write) + bool write(const T& value) { + if (m_q.bounded_push(value)) { + m_size.fetch_add(1, std::memory_order_relaxed); + return true; + } + return false; + } + + // Non-blocking dequeue; returns false if the queue is empty. (folly::MPMCQueue::read) + bool read(T& out_value) { + if (m_q.pop(out_value)) { + m_size.fetch_sub(1, std::memory_order_relaxed); + return true; + } + return false; + } + + // Approximate number of elements (racy under concurrency, like folly's). (folly::MPMCQueue::sizeGuess) + size_t sizeGuess() const { return m_size.load(std::memory_order_relaxed); } + +private: + boost::lockfree::queue< T > m_q; + std::atomic< size_t > m_size{0}; +}; + +} // namespace sisl diff --git a/src/fds/CMakeLists.txt b/src/fds/CMakeLists.txt index 84e2a0ac..d0f9940e 100644 --- a/src/fds/CMakeLists.txt +++ b/src/fds/CMakeLists.txt @@ -88,6 +88,13 @@ if(BUILD_TESTING) target_link_libraries(test_idreserver sisl_buffer GTest::gtest) add_test(NAME IdReserver COMMAND test_idreserver) + add_executable(test_bounded_mpmc_queue) + target_sources(test_bounded_mpmc_queue PRIVATE + tests/test_bounded_mpmc_queue.cpp + ) + target_link_libraries(test_bounded_mpmc_queue sisl_buffer GTest::gtest) + add_test(NAME BoundedMPMCQueue COMMAND test_bounded_mpmc_queue) + if (DEFINED MALLOC_IMPL) if (${MALLOC_IMPL} STREQUAL "jemalloc") diff --git a/src/fds/tests/test_bounded_mpmc_queue.cpp b/src/fds/tests/test_bounded_mpmc_queue.cpp new file mode 100644 index 00000000..67aca467 --- /dev/null +++ b/src/fds/tests/test_bounded_mpmc_queue.cpp @@ -0,0 +1,149 @@ +/********************************************************************************* + * Modifications Copyright 2017-2019 eBay Inc. + * + * Licensed 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 + * https://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 +#include +#include + +#include +#include + +#include + +#include + +using namespace sisl; + +SISL_OPTIONS_ENABLE(logging, test_bounded_mpmc_queue) +SISL_OPTION_GROUP(test_bounded_mpmc_queue, + (num_threads, "", "num_threads", "number of producer/consumer threads", + ::cxxopts::value< uint32_t >()->default_value("4"), "number"), + (num_entries, "", "num_entries", "number of entries per producer thread", + ::cxxopts::value< uint32_t >()->default_value("5000"), "number")) + +TEST(BoundedMPMCQueueTest, WriteAndReadSingleValue) { + BoundedMPMCQueue< int > q{4}; + EXPECT_EQ(q.sizeGuess(), 0u); + + EXPECT_TRUE(q.write(42)); + EXPECT_EQ(q.sizeGuess(), 1u); + + int out{0}; + EXPECT_TRUE(q.read(out)); + EXPECT_EQ(out, 42); + EXPECT_EQ(q.sizeGuess(), 0u); +} + +TEST(BoundedMPMCQueueTest, ReadFailsWhenEmpty) { + BoundedMPMCQueue< int > q{4}; + int out{0}; + EXPECT_FALSE(q.read(out)); +} + +TEST(BoundedMPMCQueueTest, WriteFailsWhenFull) { + constexpr size_t capacity{4}; + BoundedMPMCQueue< int > q{capacity}; + + for (size_t i{0}; i < capacity; ++i) { + EXPECT_TRUE(q.write(static_cast< int >(i))); + } + EXPECT_EQ(q.sizeGuess(), capacity); + EXPECT_FALSE(q.write(999)); + EXPECT_EQ(q.sizeGuess(), capacity); + + int out{0}; + EXPECT_TRUE(q.read(out)); + EXPECT_EQ(out, 0); + EXPECT_TRUE(q.write(999)); + EXPECT_EQ(q.sizeGuess(), capacity); +} + +TEST(BoundedMPMCQueueTest, PreservesFifoOrder) { + BoundedMPMCQueue< int > q{8}; + for (int i{0}; i < 8; ++i) { + EXPECT_TRUE(q.write(i)); + } + + for (int i{0}; i < 8; ++i) { + int out{-1}; + EXPECT_TRUE(q.read(out)); + EXPECT_EQ(out, i); + } +} + +TEST(BoundedMPMCQueueTest, ConcurrentMultiProducerMultiConsumer) { + auto const num_threads = SISL_OPTIONS["num_threads"].as< uint32_t >(); + auto const num_entries = SISL_OPTIONS["num_entries"].as< uint32_t >(); + auto const total_entries = num_threads * num_entries; + + BoundedMPMCQueue< uint64_t > q{16}; + std::vector< std::atomic< bool > > received(total_entries); + for (auto& r : received) { + r.store(false); + } + std::atomic< uint32_t > consumed_count{0}; + + std::vector< std::thread > producers; + for (uint32_t t{0}; t < num_threads; ++t) { + producers.emplace_back([&q, t, num_entries]() { + for (uint32_t i{0}; i < num_entries; ++i) { + const uint64_t value{static_cast< uint64_t >(t) * num_entries + i}; + while (!q.write(value)) { + std::this_thread::yield(); + } + } + }); + } + + std::vector< std::thread > consumers; + for (uint32_t t{0}; t < num_threads; ++t) { + consumers.emplace_back([&q, &received, &consumed_count, total_entries]() { + uint64_t value{0}; + while (consumed_count.load(std::memory_order_relaxed) < total_entries) { + if (q.read(value)) { + ASSERT_LT(value, total_entries); + ASSERT_FALSE(received[value].exchange(true)); + consumed_count.fetch_add(1, std::memory_order_relaxed); + } else { + std::this_thread::yield(); + } + } + }); + } + + for (auto& thr : producers) { + thr.join(); + } + for (auto& thr : consumers) { + thr.join(); + } + + EXPECT_EQ(consumed_count.load(), total_entries); + EXPECT_EQ(q.sizeGuess(), 0u); + for (const auto& r : received) { + EXPECT_TRUE(r.load()); + } +} + +int main(int argc, char* argv[]) { + int parsed_argc{argc}; + ::testing::InitGoogleTest(&parsed_argc, argv); + SISL_OPTIONS_LOAD(parsed_argc, argv, logging, test_bounded_mpmc_queue); + + sisl::logging::SetLogger("test_bounded_mpmc_queue"); + spdlog::set_pattern("[%D %T%z] [%^%l%$] [%n] [%t] %v"); + + return RUN_ALL_TESTS(); +} From 8bcc33a62d60d3f014e990735002775c5e5b6575 Mon Sep 17 00:00:00 2001 From: Jie Yao Date: Thu, 10 Sep 2026 17:31:25 +0800 Subject: [PATCH 2/7] bump conan version --- conanfile.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/conanfile.py b/conanfile.py index 86e3b76c..23b5af5a 100644 --- a/conanfile.py +++ b/conanfile.py @@ -10,7 +10,7 @@ class SISLConan(ConanFile): name = "sisl" - version = "14.8.1" + version = "14.8.2" homepage = "https://github.com/eBay/sisl" description = "Library for fast data structures, utilities" From dfc1f662e02cb0afe59709f92ce8c2ced079dbfd Mon Sep 17 00:00:00 2001 From: Jie Yao Date: Thu, 10 Sep 2026 18:17:48 +0800 Subject: [PATCH 3/7] disable GccThreadSanitize for now --- .github/workflows/merge_build.yml | 1 + 1 file changed, 1 insertion(+) diff --git a/.github/workflows/merge_build.yml b/.github/workflows/merge_build.yml index 7a4442e9..81453791 100644 --- a/.github/workflows/merge_build.yml +++ b/.github/workflows/merge_build.yml @@ -11,6 +11,7 @@ on: jobs: GccThreadSanitize: + if: ${{ false }} uses: ./.github/workflows/build_dependencies.yml with: platform: "ubuntu-24.04" From e83c73e2e6d375f45d66d8500375724dcdb348a5 Mon Sep 17 00:00:00 2001 From: Jie Yao Date: Fri, 11 Sep 2026 13:12:20 +0800 Subject: [PATCH 4/7] add UT for coro.hpp --- conanfile.py | 2 +- include/sisl/async/coro.hpp | 26 +++--- src/async/CMakeLists.txt | 13 +++ src/async/tests/test_coro.cpp | 149 ++++++++++++++++++++++++++++++++++ 4 files changed, 178 insertions(+), 12 deletions(-) create mode 100644 src/async/tests/test_coro.cpp diff --git a/conanfile.py b/conanfile.py index 23b5af5a..6e89f640 100644 --- a/conanfile.py +++ b/conanfile.py @@ -10,7 +10,7 @@ class SISLConan(ConanFile): name = "sisl" - version = "14.8.2" + version = "14.9.0" homepage = "https://github.com/eBay/sisl" description = "Library for fast data structures, utilities" diff --git a/include/sisl/async/coro.hpp b/include/sisl/async/coro.hpp index 98204556..19a8961d 100644 --- a/include/sisl/async/coro.hpp +++ b/include/sisl/async/coro.hpp @@ -10,14 +10,17 @@ // // Requires stdexec on the include path (same opt-in as ). +#include #include +#include #include #include -#include #include +#include #include +#include #include namespace sisl::async { @@ -48,8 +51,8 @@ inline task< T > await_value_ref(value_awaitable< T >& aw) { // when_all. template < typename Task > inline void start_coro(Task&& t) { - stdexec::start_detached( - stdexec::write_env(std::forward< Task >(t), stdexec::prop{stdexec::get_scheduler, exec::inline_scheduler{}})); + stdexec::start_detached(stdexec::write_env(std::forward< Task >(t), + stdexec::prop{stdexec::get_scheduler, stdexec::inline_scheduler{}})); } // Block the calling thread until the task completes and return its value. For the infrequent control-plane and @@ -72,15 +75,15 @@ inline auto sync_get(Task&& task) { // ref into its frame); we simply stop waiting and leave it to complete (or leak) later. Used by the data-receive // timeout path, which then inspects per-item readiness and remediates the stragglers. template < typename Task > -inline bool sync_wait_for(Task&& task, std::chrono::milliseconds timeout) { +inline bool sync_wait_for(Task&& t, std::chrono::milliseconds timeout) { auto done = std::make_shared< std::promise< void > >(); auto fut = done->get_future(); - start_coro([](std::decay_t< Task > t, std::shared_ptr< std::promise< void > > d) -> task< void > { + start_coro([](std::decay_t< Task > inner, std::shared_ptr< std::promise< void > > d) -> task< void > { try { - co_await std::move(t); + co_await std::move(inner); } catch (...) {} d->set_value(); - }(std::forward< Task >(task), std::move(done))); + }(std::forward< Task >(t), std::move(done))); return fut.wait_for(timeout) == std::future_status::ready; } @@ -98,15 +101,16 @@ inline task< void > detach_wrapper(task< T > t) { // Fire-and-forget a task whose result is not needed (e.g. a non-forced CP trigger). Starts it detached. template < typename T > -inline void detach(task< T > task) { - start_coro(detach_wrapper< T >(std::move(task))); +inline void detach(task< T > t) { + start_coro(detach_wrapper< T >(std::move(t))); } // Fire-and-forget a task but invoke fn(result) when it completes (the non-blocking ".thenValue(cb)" shape). // fn runs on whatever thread completes the task. Both task and fn are copied into the self-owning frame. template < typename T, typename Fn > -inline void detach_then(task< T > task, Fn fn) { - start_coro([](task< T > t, Fn f) -> task< void > { f(co_await std::move(t)); }(std::move(task), std::move(fn))); +inline void detach_then(task< T > t, Fn fn) { + start_coro( + [](task< T > inner, Fn f) -> task< void > { f(co_await std::move(inner)); }(std::move(t), std::move(fn))); } } // namespace sisl::async diff --git a/src/async/CMakeLists.txt b/src/async/CMakeLists.txt index 0bf5cb07..f5e6804d 100644 --- a/src/async/CMakeLists.txt +++ b/src/async/CMakeLists.txt @@ -55,5 +55,18 @@ if(BUILD_TESTING) GTest::gtest_main ) add_test(NAME ManualScheduler COMMAND test_manual_scheduler) + + # coro.hpp is not included anywhere else in this repo; this test's only job -- besides covering + # its documented behavior -- is to force the compiler to actually parse and instantiate it. + add_executable(test_coro) + target_sources(test_coro PRIVATE + tests/test_coro.cpp + ) + target_link_libraries(test_coro PRIVATE + stdexec::stdexec + sisl_logging + GTest::gtest + ) + add_test(NAME Coro COMMAND test_coro) endif() endif() diff --git a/src/async/tests/test_coro.cpp b/src/async/tests/test_coro.cpp new file mode 100644 index 00000000..a23c2b33 --- /dev/null +++ b/src/async/tests/test_coro.cpp @@ -0,0 +1,149 @@ +// Unit tests for sisl::async::coro.hpp -- the sync/detach bridges between non-coroutine code and +// sisl::async::task (exec::task). This header is not included anywhere else in the sisl repo, so nothing +// else forces the compiler to actually parse and instantiate it; these tests exist primarily to make sure +// it compiles cleanly, in addition to covering its documented behavior. + +#include +#include +#include +#include +#include + +#include + +#include + +namespace { + +using sisl::async::shared_awaitable; +using sisl::async::task; +using sisl::async::value_awaitable; + +// ============================================================================ +// sync_get: blocking bridge, same-thread and cross-thread, value and void +// ============================================================================ + +TEST(coro, SyncGetReturnsImmediateValue) { + auto v = sisl::async::sync_get([]() -> task< int > { co_return 5; }()); + EXPECT_EQ(v, 5); +} + +TEST(coro, SyncGetHandlesVoidTask) { + bool ran{false}; + sisl::async::sync_get([](bool* r) -> task< void > { + *r = true; + co_return; + }(&ran)); + EXPECT_TRUE(ran); +} + +TEST(coro, SyncGetRethrows) { + auto thrower = []() -> task< int > { + throw std::runtime_error("sync boom"); + co_return 0; + }; + EXPECT_THROW((void)sisl::async::sync_get(thrower()), std::runtime_error); +} + +TEST(coro, SyncGetCrossThreadCompletion) { + value_awaitable< int > ev{}; + std::thread producer{[&ev] { ev.complete(123); }}; + auto v = sisl::async::sync_get([](value_awaitable< int >& e) -> task< int > { co_return co_await e; }(ev)); + producer.join(); + EXPECT_EQ(v, 123); +} + +// ============================================================================ +// await_shared / await_value / await_value_ref: wrap an awaitable into a task +// ============================================================================ + +TEST(coro, AwaitSharedResolvesToProducerValue) { + auto aw = std::make_shared< shared_awaitable< int > >(); + std::thread producer{[aw] { aw->complete(42); }}; + auto v = sisl::async::sync_get(sisl::async::await_shared(aw)); + producer.join(); + EXPECT_EQ(v, 42); +} + +TEST(coro, AwaitValueResolvesToProducerValue) { + auto aw = std::make_shared< value_awaitable< int > >(); + std::thread producer{[aw] { aw->complete(43); }}; + auto v = sisl::async::sync_get(sisl::async::await_value(aw)); + producer.join(); + EXPECT_EQ(v, 43); +} + +TEST(coro, AwaitValueRefResolvesToProducerValue) { + value_awaitable< int > ev{}; + std::thread producer{[&ev] { ev.complete(44); }}; + auto v = sisl::async::sync_get(sisl::async::await_value_ref(ev)); + producer.join(); + EXPECT_EQ(v, 44); +} + +// ============================================================================ +// detach() / detach_then(): fire-and-forget, exception-swallowing +// ============================================================================ + +TEST(coro, DetachRunsTaskToCompletion) { + value_awaitable< int > ev{}; + bool got{false}; + auto t = [](value_awaitable< int >& e, bool* g) -> task< int > { + auto const v = co_await e; + *g = true; + co_return v; + }(ev, &got); + + sisl::async::detach(std::move(t)); + EXPECT_FALSE(got); // suspended on ev; detach() only runs inline up to the first suspension + ev.complete(9); + EXPECT_TRUE(got); +} + +TEST(coro, DetachSwallowsException) { + auto t = []() -> task< int > { + throw std::runtime_error("detached boom"); + co_return 0; + }(); + EXPECT_NO_THROW(sisl::async::detach(std::move(t))); +} + +TEST(coro, DetachThenInvokesCallbackWithResult) { + value_awaitable< int > ev{}; + std::optional< int > seen{}; + auto t = [](value_awaitable< int >& e) -> task< int > { co_return co_await e; }(ev); + + sisl::async::detach_then(std::move(t), [&seen](int v) { seen = v; }); + EXPECT_FALSE(seen.has_value()); + ev.complete(21); + ASSERT_TRUE(seen.has_value()); + EXPECT_EQ(*seen, 21); +} + +// ============================================================================ +// sync_wait_for: bounded blocking wait +// ============================================================================ + +TEST(coro, SyncWaitForReturnsTrueWhenCompletedInTime) { + value_awaitable< int > ev{}; + std::thread producer{[&ev] { ev.complete(1); }}; + auto const completed = sisl::async::sync_wait_for(sisl::async::await_value_ref(ev), std::chrono::seconds(5)); + producer.join(); + EXPECT_TRUE(completed); +} + +TEST(coro, SyncWaitForReturnsFalseOnTimeout) { + value_awaitable< int > ev{}; + auto const completed = sisl::async::sync_wait_for(sisl::async::await_value_ref(ev), std::chrono::milliseconds(10)); + EXPECT_FALSE(completed); + // The detached task started by sync_wait_for is still suspended on ev; ev is a stack local that the task + // holds a reference to, so it must be completed before ev goes out of scope (documented caller obligation). + ev.complete(0); +} + +} // namespace + +int main(int argc, char* argv[]) { + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} From e3d25c960dd785f682165b7f963aa7999d6273cd Mon Sep 17 00:00:00 2001 From: Jie Yao Date: Mon, 14 Sep 2026 16:11:17 +0800 Subject: [PATCH 5/7] add lru_map --- include/sisl/fds/lru_map.hpp | 73 +++++++++++++++++++++++++++++++ src/fds/CMakeLists.txt | 6 +++ src/fds/tests/test_lru_map.cpp | 80 ++++++++++++++++++++++++++++++++++ 3 files changed, 159 insertions(+) create mode 100644 include/sisl/fds/lru_map.hpp create mode 100644 src/fds/tests/test_lru_map.cpp diff --git a/include/sisl/fds/lru_map.hpp b/include/sisl/fds/lru_map.hpp new file mode 100644 index 00000000..3d3baea2 --- /dev/null +++ b/include/sisl/fds/lru_map.hpp @@ -0,0 +1,73 @@ +/********************************************************************************* + * Modifications Copyright 2017-2019 eBay Inc. + * + * Licensed 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 + * https://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. + * + *********************************************************************************/ +#pragma once + +#include +#include +#include +#include + +namespace sisl { + +// Tiny bounded cache used by simple in-memory LRU scenarios. +// +// The implementation keeps the most recently touched item at the front of `order_` and +// uses `index_` to map a key to the corresponding list iterator. `std::size_t` is needed +// for the cache capacity, while `std::pair` and `std::move` are required to represent each +// cached element and move replacement values into the list without copying. +template < typename Key, typename Value > +class LruMap { +public: + using value_type = std::pair< Key, Value >; + using list_type = std::list< value_type >; + using iterator = typename list_type::iterator; + using const_iterator = typename list_type::const_iterator; + + explicit LruMap(std::size_t capacity) : capacity_{capacity} {} + + // Inserts or updates a key. Updating an existing key refreshes the value and moves it to + // the front to preserve the recency ordering. When the cache exceeds its capacity, the + // least recently used item is evicted from both `order_` and `index_`. + void set(Key const& key, Value value) { + if (auto it = index_.find(key); it != index_.end()) { + it->second->second = std::move(value); + order_.splice(order_.begin(), order_, it->second); + return; + } + order_.emplace_front(key, std::move(value)); + index_[order_.front().first] = order_.begin(); + while (order_.size() > capacity_) { + index_.erase(order_.back().first); + order_.pop_back(); + } + } + + // Missing keys return a default-constructed Value (shared_ptr -> nullptr). + Value get(Key const& key) const { + auto it = index_.find(key); + if (it == index_.end()) { return Value{}; } + return it->second->second; + } + + const_iterator begin() const { return order_.begin(); } + const_iterator end() const { return order_.end(); } + +private: + std::size_t capacity_; + list_type order_; + std::unordered_map< Key, iterator > index_; +}; + +} // namespace sisl diff --git a/src/fds/CMakeLists.txt b/src/fds/CMakeLists.txt index d0f9940e..3df57757 100644 --- a/src/fds/CMakeLists.txt +++ b/src/fds/CMakeLists.txt @@ -95,6 +95,12 @@ if(BUILD_TESTING) target_link_libraries(test_bounded_mpmc_queue sisl_buffer GTest::gtest) add_test(NAME BoundedMPMCQueue COMMAND test_bounded_mpmc_queue) + add_executable(test_lru_map) + target_sources(test_lru_map PRIVATE + tests/test_lru_map.cpp + ) + target_link_libraries(test_lru_map sisl_buffer GTest::gtest) + add_test(NAME LruMap COMMAND test_lru_map) if (DEFINED MALLOC_IMPL) if (${MALLOC_IMPL} STREQUAL "jemalloc") diff --git a/src/fds/tests/test_lru_map.cpp b/src/fds/tests/test_lru_map.cpp new file mode 100644 index 00000000..e1749c13 --- /dev/null +++ b/src/fds/tests/test_lru_map.cpp @@ -0,0 +1,80 @@ +/********************************************************************************* + * Modifications Copyright 2017-2019 eBay Inc. + * + * Licensed 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 + * https://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 "sisl/fds/lru_map.hpp" + +namespace { + +TEST(LruMapTest, AccessesMostRecentlyUsedEntriesFirst) { + sisl::LruMap< int, int > cache(2); + + cache.set(1, 10); + cache.set(2, 20); + + EXPECT_EQ(cache.get(1), 10); + EXPECT_EQ(cache.get(2), 20); + + cache.set(3, 30); + EXPECT_EQ(cache.get(1), 0); + EXPECT_EQ(cache.get(2), 20); + EXPECT_EQ(cache.get(3), 30); +} + +TEST(LruMapTest, UpdatesExistingKeyAndRefreshesRecency) { + sisl::LruMap< int, std::string > cache(2); + + cache.set(1, "old"); + cache.set(2, "new"); + cache.set(1, "fresh"); + cache.set(3, "latest"); + + EXPECT_EQ(cache.get(1), "fresh"); + EXPECT_EQ(cache.get(2), ""); + EXPECT_EQ(cache.get(3), "latest"); +} + +TEST(LruMapTest, IteratorVisitsMostRecentlyUsedFirst) { + sisl::LruMap< int, int > cache(3); + cache.set(1, 11); + cache.set(2, 22); + cache.set(3, 33); + + auto it = cache.begin(); + ASSERT_NE(it, cache.end()); + EXPECT_EQ(it->first, 3); + EXPECT_EQ(it->second, 33); + + ++it; + ASSERT_NE(it, cache.end()); + EXPECT_EQ(it->first, 2); + EXPECT_EQ(it->second, 22); + + ++it; + ASSERT_NE(it, cache.end()); + EXPECT_EQ(it->first, 1); + EXPECT_EQ(it->second, 11); + + ++it; + EXPECT_EQ(it, cache.end()); +} + +} // namespace + +int main(int argc, char* argv[]) { + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} From 2ef99543a36ff630b7757cb04abb15e0f70cd6c1 Mon Sep 17 00:00:00 2001 From: Jie Yao Date: Mon, 14 Sep 2026 18:06:15 +0800 Subject: [PATCH 6/7] add mcmp_priority_queue --- include/sisl/fds/mcmp_priority_queue.hpp | 220 +++++++++ src/fds/CMakeLists.txt | 7 + src/fds/tests/test_lru_map.cpp | 129 ++++++ src/fds/tests/test_mcmp_priority_queue.cpp | 496 +++++++++++++++++++++ 4 files changed, 852 insertions(+) create mode 100644 include/sisl/fds/mcmp_priority_queue.hpp create mode 100644 src/fds/tests/test_mcmp_priority_queue.cpp diff --git a/include/sisl/fds/mcmp_priority_queue.hpp b/include/sisl/fds/mcmp_priority_queue.hpp new file mode 100644 index 00000000..fff1de96 --- /dev/null +++ b/include/sisl/fds/mcmp_priority_queue.hpp @@ -0,0 +1,220 @@ +/********************************************************************************* + * Modifications Copyright 2017-2019 eBay Inc. + * + * Licensed 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 + * https://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. + * + *********************************************************************************/ +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace sisl { + +/** + * @brief Multi-Producer Multi-Consumer Priority Queue (C++20) + * + * Thread-safe priority queue that supports: + * - Concurrent push operations from multiple producers + * - Concurrent pop operations from multiple consumers + * - Blocking pop when queue is empty + * - Graceful shutdown via close() method + * + * @tparam T Element type (must be comparable) + * @tparam Compare Comparison function (default: std::less for max-heap) + */ +template < typename T, typename Compare = std::less< T > > + requires std::movable< T > && std::predicate< Compare, T, T > +class MPMCPriorityQueue { +public: + using value_type = T; + using size_type = std::size_t; + using comparator_type = Compare; + + /** + * @brief Status codes returned by pop operations + */ + enum class Status : uint8_t { + Ok, ///< Successfully popped an element + Empty, ///< Queue is empty but still open (try_pop only) + Closed ///< Queue is closed, no more elements available + }; + + /** + * @brief Result of a pop operation + */ + struct PopResult { + Status status; + std::optional< T > value; ///< Has value only if status == Ok + + // Convenience methods + [[nodiscard]] constexpr bool is_ok() const noexcept { return status == Status::Ok; } + [[nodiscard]] constexpr bool is_empty() const noexcept { return status == Status::Empty; } + [[nodiscard]] constexpr bool is_closed() const noexcept { return status == Status::Closed; } + }; + + /** + * @brief Construct an empty priority queue + */ + constexpr MPMCPriorityQueue() noexcept(std::is_nothrow_default_constructible_v< Compare >) = default; + + /** + * @brief Destructor - automatically closes the queue + */ + ~MPMCPriorityQueue() { close(); } + + // Disable copy and move to prevent issues with condition variables + MPMCPriorityQueue(const MPMCPriorityQueue&) = delete; + MPMCPriorityQueue& operator=(const MPMCPriorityQueue&) = delete; + MPMCPriorityQueue(MPMCPriorityQueue&&) = delete; + MPMCPriorityQueue& operator=(MPMCPriorityQueue&&) = delete; + + /** + * @brief Thread-safe push operation (copy) + * + * @param value Element to insert + * @return true if pushed successfully, false if queue is closed + */ + bool push(const T& value) + requires std::copy_constructible< T > + { + { + std::scoped_lock lock(mutex_); + if (closed_) [[unlikely]] { + return false; // Queue is closed, cannot push + } + pq_.push(value); + } + cv_.notify_one(); // Wake one waiting consumer + return true; + } + + /** + * @brief Thread-safe push operation (move) + * + * @param value Element to insert (will be moved) + * @return true if pushed successfully, false if queue is closed + */ + bool push(T&& value) { + { + std::scoped_lock lock(mutex_); + if (closed_) [[unlikely]] { return false; } + pq_.push(std::move(value)); + } + cv_.notify_one(); + return true; + } + + /** + * @brief Thread-safe pop operation + * + * Blocks if queue is empty and not closed. + * Returns immediately if queue is closed. + * + * @return PopResult containing status and optional value + * @note Thread-safe for multiple concurrent consumers + */ + [[nodiscard]] PopResult pop() { + std::unique_lock lock(mutex_); + + // Wait until queue has elements or is closed + cv_.wait(lock, [this] { return closed_ || !pq_.empty(); }); + + // Try to pop an element + if (!pq_.empty()) { + T top = std::move(const_cast< T& >(pq_.top())); + pq_.pop(); + return PopResult{.status = Status::Ok, .value = std::move(top)}; + } + + // Queue is empty and closed + return PopResult{.status = Status::Closed, .value = std::nullopt}; + } + + /// Non-blocking pop. Safe to call from an iomgr reactor (never waits). + [[nodiscard]] PopResult try_pop() { + std::scoped_lock lock(mutex_); + if (!pq_.empty()) { + T top = std::move(const_cast< T& >(pq_.top())); + pq_.pop(); + return PopResult{.status = Status::Ok, .value = std::move(top)}; + } + if (closed_) { return PopResult{.status = Status::Closed, .value = std::nullopt}; } + return PopResult{.status = Status::Empty, .value = std::nullopt}; + } + + /** + * @brief Close the queue + * + * After calling close(): + * - All blocked pop() calls will wake up + * - Existing elements can still be popped + * - New push() calls will be ignored + * - pop() returns Status::Closed when queue becomes empty + * + * @note Thread-safe and idempotent + */ + void close() noexcept { + { + std::scoped_lock lock(mutex_); + closed_ = true; + } + cv_.notify_all(); // Wake all waiting consumers + } + + /** + * @brief Get current number of elements + * + * @return Number of elements in the queue + * @note Thread-safe + */ + [[nodiscard]] size_type size() const { + std::scoped_lock lock(mutex_); + return pq_.size(); + } + + /** + * @brief Check if queue is empty + * + * @return true if queue has no elements + * @note Thread-safe + */ + [[nodiscard]] bool empty() const { + std::scoped_lock lock(mutex_); + return pq_.empty(); + } + + /** + * @brief Check if queue is closed + * + * @return true if close() has been called + * @note Thread-safe + */ + [[nodiscard]] bool is_closed() const { + std::scoped_lock lock(mutex_); + return closed_; + } + +private: + mutable std::mutex mutex_; + std::condition_variable cv_; + bool closed_{false}; + std::priority_queue< T, std::vector< T >, Compare > pq_; +}; + +} // namespace sisl diff --git a/src/fds/CMakeLists.txt b/src/fds/CMakeLists.txt index 3df57757..0889d8cb 100644 --- a/src/fds/CMakeLists.txt +++ b/src/fds/CMakeLists.txt @@ -102,6 +102,13 @@ if(BUILD_TESTING) target_link_libraries(test_lru_map sisl_buffer GTest::gtest) add_test(NAME LruMap COMMAND test_lru_map) + add_executable(test_mcmp_priority_queue) + target_sources(test_mcmp_priority_queue PRIVATE + tests/test_mcmp_priority_queue.cpp + ) + target_link_libraries(test_mcmp_priority_queue sisl_buffer GTest::gtest) + add_test(NAME MPMCPriorityQueue COMMAND test_mcmp_priority_queue) + if (DEFINED MALLOC_IMPL) if (${MALLOC_IMPL} STREQUAL "jemalloc") add_executable(test_jemalloc) diff --git a/src/fds/tests/test_lru_map.cpp b/src/fds/tests/test_lru_map.cpp index e1749c13..aa8f8962 100644 --- a/src/fds/tests/test_lru_map.cpp +++ b/src/fds/tests/test_lru_map.cpp @@ -12,6 +12,8 @@ * specific language governing permissions and limitations under the License. * *********************************************************************************/ +#include +#include #include @@ -19,6 +21,133 @@ namespace { +TEST(LruMapTest, EmptyCacheGetReturnsDefaultValue) { + sisl::LruMap< int, int > cache(2); + EXPECT_EQ(cache.get(42), 0); +} + +TEST(LruMapTest, EmptyCacheIteratorEqualsEnd) { + sisl::LruMap< int, int > const cache(2); + EXPECT_EQ(cache.begin(), cache.end()); +} + +TEST(LruMapTest, MissingKeyReturnsDefaultConstructedValue) { + sisl::LruMap< int, std::shared_ptr< int > > cache(2); + cache.set(1, std::make_shared< int >(10)); + + EXPECT_EQ(cache.get(2), nullptr); +} + +TEST(LruMapTest, ZeroCapacityImmediatelyEvictsInsertedEntry) { + sisl::LruMap< int, int > cache(0); + + cache.set(1, 10); + + EXPECT_EQ(cache.get(1), 0); + EXPECT_EQ(cache.begin(), cache.end()); +} + +TEST(LruMapTest, CapacityOfOneKeepsOnlyMostRecentEntry) { + sisl::LruMap< int, int > cache(1); + + cache.set(1, 10); + EXPECT_EQ(cache.get(1), 10); + + cache.set(2, 20); + EXPECT_EQ(cache.get(1), 0); + EXPECT_EQ(cache.get(2), 20); +} + +TEST(LruMapTest, UpdatingExistingKeyDoesNotGrowCacheOrEvict) { + sisl::LruMap< int, int > cache(2); + + cache.set(1, 10); + cache.set(2, 20); + // Key 1 already present -- this must update in place, not append, so nothing gets evicted. + cache.set(1, 100); + + EXPECT_EQ(cache.get(1), 100); + EXPECT_EQ(cache.get(2), 20); + + std::size_t count = 0; + for (auto it = cache.begin(); it != cache.end(); ++it) { + ++count; + } + EXPECT_EQ(count, 2u); +} + +TEST(LruMapTest, ReinsertingSameKeyRepeatedlyKeepsSingleEntry) { + sisl::LruMap< int, int > cache(3); + + for (int i = 0; i < 5; ++i) { + cache.set(1, i); + } + + EXPECT_EQ(cache.get(1), 4); + std::size_t count = 0; + for (auto it = cache.begin(); it != cache.end(); ++it) { + ++count; + } + EXPECT_EQ(count, 1u); +} + +// get() is a const lookup only -- it does NOT splice the entry to the front, so an entry that was +// merely read (not re-set) is still the least-recently-*set* item and can be evicted ahead of one +// that was set more recently, even though it was read after that. +TEST(LruMapTest, GetDoesNotRefreshRecency) { + sisl::LruMap< int, int > cache(2); + + cache.set(1, 10); + cache.set(2, 20); + EXPECT_EQ(cache.get(1), 10); // touch key 1 via get() only + + cache.set(3, 30); // eviction is driven by set() order, not get() order + EXPECT_EQ(cache.get(1), 0); // key 1 still evicted despite the read above + EXPECT_EQ(cache.get(2), 20); + EXPECT_EQ(cache.get(3), 30); +} + +TEST(LruMapTest, SettingFrontEntryAgainIsANoOpForOrdering) { + sisl::LruMap< int, int > cache(2); + + cache.set(1, 10); + cache.set(2, 20); // key 2 is now at the front + cache.set(2, 200); // re-set the entry already at the front + + auto it = cache.begin(); + ASSERT_NE(it, cache.end()); + EXPECT_EQ(it->first, 2); + EXPECT_EQ(it->second, 200); +} + +TEST(LruMapTest, EvictsInStrictLruOrderAcrossMultipleInsertions) { + sisl::LruMap< int, int > cache(3); + + cache.set(1, 1); + cache.set(2, 2); + cache.set(3, 3); + cache.set(4, 4); // evicts 1 + cache.set(5, 5); // evicts 2 + + EXPECT_EQ(cache.get(1), 0); + EXPECT_EQ(cache.get(2), 0); + EXPECT_EQ(cache.get(3), 3); + EXPECT_EQ(cache.get(4), 4); + EXPECT_EQ(cache.get(5), 5); +} + +TEST(LruMapTest, WorksWithStringKeys) { + sisl::LruMap< std::string, int > cache(2); + + cache.set("alpha", 1); + cache.set("beta", 2); + cache.set("gamma", 3); // evicts "alpha" + + EXPECT_EQ(cache.get("alpha"), 0); + EXPECT_EQ(cache.get("beta"), 2); + EXPECT_EQ(cache.get("gamma"), 3); +} + TEST(LruMapTest, AccessesMostRecentlyUsedEntriesFirst) { sisl::LruMap< int, int > cache(2); diff --git a/src/fds/tests/test_mcmp_priority_queue.cpp b/src/fds/tests/test_mcmp_priority_queue.cpp new file mode 100644 index 00000000..3db4aa7c --- /dev/null +++ b/src/fds/tests/test_mcmp_priority_queue.cpp @@ -0,0 +1,496 @@ +/********************************************************************************* + * Modifications Copyright 2017-2019 eBay Inc. + * + * Licensed 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 + * https://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 +#include +#include + +#include + +#include "sisl/fds/mcmp_priority_queue.hpp" + +TEST(MPMCPriorityQueueTest, BasicPushPop) { + sisl::MPMCPriorityQueue< int > queue; + + // Push elements + queue.push(5); + queue.push(2); + queue.push(8); + queue.push(1); + + EXPECT_EQ(queue.size(), 4); + EXPECT_FALSE(queue.empty()); + + // Pop in priority order (max heap by default) + auto r1 = queue.pop(); + EXPECT_TRUE(r1.is_ok()); + EXPECT_EQ(r1.value.value(), 8); + + auto r2 = queue.pop(); + EXPECT_TRUE(r2.is_ok()); + EXPECT_EQ(r2.value.value(), 5); + + auto r3 = queue.pop(); + EXPECT_TRUE(r3.is_ok()); + EXPECT_EQ(r3.value.value(), 2); + + auto r4 = queue.pop(); + EXPECT_TRUE(r4.is_ok()); + EXPECT_EQ(r4.value.value(), 1); + + EXPECT_EQ(queue.size(), 0); + EXPECT_TRUE(queue.empty()); +} + +TEST(MPMCPriorityQueueTest, CustomComparator) { + // Min-heap using std::greater + sisl::MPMCPriorityQueue< int, std::greater< int > > queue; + + queue.push(5); + queue.push(2); + queue.push(8); + queue.push(1); + + // Pop in ascending order + EXPECT_EQ(queue.pop().value.value(), 1); + EXPECT_EQ(queue.pop().value.value(), 2); + EXPECT_EQ(queue.pop().value.value(), 5); + EXPECT_EQ(queue.pop().value.value(), 8); +} + +TEST(MPMCPriorityQueueTest, MoveSemantics) { + struct MoveOnly { + int value; + + explicit MoveOnly(int v) : value(v) {} + MoveOnly(const MoveOnly&) = delete; + MoveOnly& operator=(const MoveOnly&) = delete; + MoveOnly(MoveOnly&&) = default; + MoveOnly& operator=(MoveOnly&&) = default; + + bool operator<(const MoveOnly& other) const { return value < other.value; } + }; + + sisl::MPMCPriorityQueue< MoveOnly > queue; + + queue.push(MoveOnly(5)); + queue.push(MoveOnly(2)); + queue.push(MoveOnly(8)); + + EXPECT_EQ(queue.pop().value.value().value, 8); + EXPECT_EQ(queue.pop().value.value().value, 5); + EXPECT_EQ(queue.pop().value.value().value, 2); +} + +// ============================================================================ +// Close Operation Tests +// ============================================================================ + +TEST(MPMCPriorityQueueTest, Close) { + sisl::MPMCPriorityQueue< int > queue; + + queue.push(1); + queue.push(2); + queue.push(3); + + EXPECT_FALSE(queue.is_closed()); + queue.close(); + EXPECT_TRUE(queue.is_closed()); + + // Can still pop existing elements + EXPECT_EQ(queue.pop().value.value(), 3); + EXPECT_EQ(queue.pop().value.value(), 2); + EXPECT_EQ(queue.pop().value.value(), 1); + + // Now should return Closed status + auto result = queue.pop(); + EXPECT_TRUE(result.is_closed()); + EXPECT_FALSE(result.value.has_value()); +} + +TEST(MPMCPriorityQueueTest, PushAfterClose) { + sisl::MPMCPriorityQueue< int > queue; + + queue.push(1); + queue.close(); + + // Pushes after close are ignored + queue.push(2); + queue.push(3); + + EXPECT_EQ(queue.size(), 1); + + auto r1 = queue.pop(); + EXPECT_TRUE(r1.is_ok()); + EXPECT_EQ(r1.value.value(), 1); + + auto r2 = queue.pop(); + EXPECT_TRUE(r2.is_closed()); +} + +TEST(MPMCPriorityQueueTest, CloseIdempotent) { + sisl::MPMCPriorityQueue< int > queue; + + queue.push(1); + queue.close(); + queue.close(); // Should be safe + queue.close(); + + EXPECT_TRUE(queue.is_closed()); + EXPECT_EQ(queue.size(), 1); +} + +// ============================================================================ +// Blocking Behavior Tests +// ============================================================================ + +TEST(MPMCPriorityQueueTest, BlockingPop) { + sisl::MPMCPriorityQueue< int > queue; + std::atomic< bool > pop_started{false}; + std::atomic< bool > pop_completed{false}; + + // Consumer thread that will block + std::thread consumer([&]() { + pop_started = true; + auto result = queue.pop(); + pop_completed = true; + + EXPECT_TRUE(result.is_ok()); + EXPECT_EQ(result.value.value(), 42); + }); + + // Wait for consumer to start + while (!pop_started) { + std::this_thread::yield(); + } + + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + EXPECT_FALSE(pop_completed); + + // Unblock consumer by pushing + queue.push(42); + + consumer.join(); + EXPECT_TRUE(pop_completed); +} + +TEST(MPMCPriorityQueueTest, CloseUnblocksWaiters) { + sisl::MPMCPriorityQueue< int > queue; + std::atomic< int > closed_count{0}; + + // Start multiple waiting consumers + std::vector< std::thread > consumers; + for (int i = 0; i < 5; ++i) { + consumers.emplace_back([&]() { + auto result = queue.pop(); + if (result.is_closed()) { closed_count.fetch_add(1, std::memory_order_relaxed); } + }); + } + + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + + // Close should wake all waiters + queue.close(); + + for (auto& t : consumers) { + t.join(); + } + + EXPECT_EQ(closed_count.load(), 5); +} + +// ============================================================================ +// Multi-threaded Producer Tests +// ============================================================================ + +TEST(MPMCPriorityQueueTest, MultipleProducers) { + sisl::MPMCPriorityQueue< int > queue; + constexpr int num_producers = 4; + constexpr int items_per_producer = 250; + + std::vector< std::thread > producers; + for (int i = 0; i < num_producers; ++i) { + producers.emplace_back([&, i]() { + for (int j = 0; j < items_per_producer; ++j) { + queue.push(i * items_per_producer + j); + } + }); + } + + for (auto& t : producers) { + t.join(); + } + + EXPECT_EQ(queue.size(), num_producers * items_per_producer); + + // Verify all elements come out in descending order + std::vector< int > popped; + for (int i = 0; i < num_producers * items_per_producer; ++i) { + auto result = queue.pop(); + ASSERT_TRUE(result.is_ok()); + popped.push_back(result.value.value()); + } + + EXPECT_TRUE(std::is_sorted(popped.rbegin(), popped.rend())); +} + +// ============================================================================ +// Multi-threaded Consumer Tests +// ============================================================================ + +TEST(MPMCPriorityQueueTest, MultipleConsumers) { + sisl::MPMCPriorityQueue< int > queue; + constexpr int num_items = 1000; + + // Fill queue + for (int i = 0; i < num_items; ++i) { + queue.push(i); + } + + constexpr int num_consumers = 4; + std::vector< std::thread > consumers; + std::atomic< int > total_consumed{0}; + + for (int i = 0; i < num_consumers; ++i) { + consumers.emplace_back([&]() { + int count = 0; + while (true) { + auto result = queue.pop(); + if (result.is_closed()) { break; } + ++count; + } + total_consumed.fetch_add(count, std::memory_order_relaxed); + }); + } + + // Give consumers time to start + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + + // Close to signal completion + queue.close(); + + for (auto& t : consumers) { + t.join(); + } + + EXPECT_EQ(total_consumed.load(), num_items); +} + +// ============================================================================ +// Concurrent Producers and Consumers +// ============================================================================ + +TEST(MPMCPriorityQueueTest, ConcurrentProducersConsumers) { + sisl::MPMCPriorityQueue< int > queue; + constexpr int num_producers = 3; + constexpr int num_consumers = 3; + constexpr int items_per_producer = 200; + + std::atomic< int > total_consumed{0}; + std::vector< std::thread > threads; + + // Start consumers + for (int i = 0; i < num_consumers; ++i) { + threads.emplace_back([&]() { + int count = 0; + while (true) { + auto result = queue.pop(); + if (result.is_closed()) { break; } + ++count; + } + total_consumed.fetch_add(count, std::memory_order_relaxed); + }); + } + + // Start producers + for (int i = 0; i < num_producers; ++i) { + threads.emplace_back([&, i]() { + for (int j = 0; j < items_per_producer; ++j) { + queue.push(i * items_per_producer + j); + std::this_thread::sleep_for(std::chrono::microseconds(10)); // Simulate work + } + }); + } + + // Wait for producers + for (int i = num_consumers; i < num_consumers + num_producers; ++i) { + threads[i].join(); + } + + // Close and wait for consumers + queue.close(); + for (int i = 0; i < num_consumers; ++i) { + threads[i].join(); + } + + EXPECT_EQ(total_consumed.load(), num_producers * items_per_producer); +} + +// ============================================================================ +// Stress Test +// ============================================================================ + +TEST(MPMCPriorityQueueTest, StressTest) { + sisl::MPMCPriorityQueue< int > queue; + constexpr int num_threads = 8; + constexpr int operations_per_thread = 1000; + + std::atomic< int > push_count{0}; + std::atomic< int > pop_count{0}; + std::vector< std::thread > threads; + + // Half producers, half consumers + for (int i = 0; i < num_threads / 2; ++i) { + threads.emplace_back([&]() { + for (int j = 0; j < operations_per_thread; ++j) { + queue.push(j); + push_count.fetch_add(1, std::memory_order_relaxed); + } + }); + } + + for (int i = 0; i < num_threads / 2; ++i) { + threads.emplace_back([&]() { + for (int j = 0; j < operations_per_thread; ++j) { + auto result = queue.pop(); + if (result.is_ok()) { pop_count.fetch_add(1, std::memory_order_relaxed); } + } + }); + } + + for (auto& t : threads) { + t.join(); + } + + EXPECT_EQ(push_count.load(), (num_threads / 2) * operations_per_thread); + + // Pop remaining elements + while (!queue.empty()) { + auto result = queue.pop(); + if (result.is_ok()) { pop_count.fetch_add(1, std::memory_order_relaxed); } + } + + EXPECT_EQ(pop_count.load(), push_count.load()); +} + +// ============================================================================ +// Destructor Test +// ============================================================================ + +TEST(MPMCPriorityQueueTest, DestructorClosesQueue) { + std::atomic< bool > consumer_unblocked{false}; + + std::thread consumer([&]() { + auto queue = std::make_unique< sisl::MPMCPriorityQueue< int > >(); + queue->push(1); + + std::thread waiter([&, q = queue.get()]() { + auto first_result = q->pop(); // Pop the 1 + (void)first_result; // Explicitly ignore the result + auto result = q->pop(); // This will block until destructor closes queue + if (result.is_closed()) { consumer_unblocked = true; } + }); + + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + // Destructor will be called here + queue.reset(); + + waiter.join(); + }); + + consumer.join(); + EXPECT_TRUE(consumer_unblocked); +} + +TEST(MPMCPriorityQueueTest, PopsHighestPriorityElementFirst) { + sisl::MPMCPriorityQueue< int > queue; + + EXPECT_TRUE(queue.push(5)); + EXPECT_TRUE(queue.push(2)); + EXPECT_TRUE(queue.push(9)); + EXPECT_TRUE(queue.push(7)); + + auto first = queue.pop(); + ASSERT_TRUE(first.is_ok()); + ASSERT_TRUE(first.value.has_value()); + EXPECT_EQ(first.value.value(), 9); + + auto second = queue.pop(); + ASSERT_TRUE(second.is_ok()); + ASSERT_TRUE(second.value.has_value()); + EXPECT_EQ(second.value.value(), 7); + + auto third = queue.pop(); + ASSERT_TRUE(third.is_ok()); + ASSERT_TRUE(third.value.has_value()); + EXPECT_EQ(third.value.value(), 5); + + EXPECT_EQ(queue.size(), 1u); +} + +TEST(MPMCPriorityQueueTest, TryPopReportsEmptyUntilClosed) { + sisl::MPMCPriorityQueue< int > queue; + + auto empty = queue.try_pop(); + EXPECT_TRUE(empty.is_empty()); + EXPECT_FALSE(empty.value.has_value()); + + EXPECT_TRUE(queue.push(42)); + + auto item = queue.try_pop(); + ASSERT_TRUE(item.is_ok()); + ASSERT_TRUE(item.value.has_value()); + EXPECT_EQ(item.value.value(), 42); + + auto empty_again = queue.try_pop(); + EXPECT_TRUE(empty_again.is_empty()); + EXPECT_FALSE(empty_again.value.has_value()); + + queue.close(); + auto closed = queue.try_pop(); + EXPECT_TRUE(closed.is_closed()); + EXPECT_FALSE(closed.value.has_value()); +} + +TEST(MPMCPriorityQueueTest, CloseRejectsNewPushesAndDrainsExistingItems) { + sisl::MPMCPriorityQueue< int > queue; + + EXPECT_TRUE(queue.push(10)); + EXPECT_TRUE(queue.push(20)); + + queue.close(); + EXPECT_FALSE(queue.push(30)); + EXPECT_TRUE(queue.is_closed()); + + auto first = queue.pop(); + ASSERT_TRUE(first.is_ok()); + ASSERT_TRUE(first.value.has_value()); + EXPECT_EQ(first.value.value(), 20); + + auto second = queue.pop(); + ASSERT_TRUE(second.is_ok()); + ASSERT_TRUE(second.value.has_value()); + EXPECT_EQ(second.value.value(), 10); + + auto done = queue.pop(); + EXPECT_TRUE(done.is_closed()); + EXPECT_FALSE(done.value.has_value()); +} + +int main(int argc, char* argv[]) { + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} From cf08343894ee7c7eb682a2a6ccfd48c435870fc8 Mon Sep 17 00:00:00 2001 From: Jie Yao Date: Tue, 15 Sep 2026 08:46:29 +0800 Subject: [PATCH 7/7] fix --- .github/workflows/merge_build.yml | 1 - include/sisl/fds/mcmp_priority_queue.hpp | 15 ++++++++++++++- src/fds/CMakeLists.txt | 2 ++ tsan.supp | 6 ++++++ 4 files changed, 22 insertions(+), 2 deletions(-) diff --git a/.github/workflows/merge_build.yml b/.github/workflows/merge_build.yml index 81453791..7a4442e9 100644 --- a/.github/workflows/merge_build.yml +++ b/.github/workflows/merge_build.yml @@ -11,7 +11,6 @@ on: jobs: GccThreadSanitize: - if: ${{ false }} uses: ./.github/workflows/build_dependencies.yml with: platform: "ubuntu-24.04" diff --git a/include/sisl/fds/mcmp_priority_queue.hpp b/include/sisl/fds/mcmp_priority_queue.hpp index fff1de96..0aeefade 100644 --- a/include/sisl/fds/mcmp_priority_queue.hpp +++ b/include/sisl/fds/mcmp_priority_queue.hpp @@ -75,8 +75,17 @@ class MPMCPriorityQueue { /** * @brief Destructor - automatically closes the queue + * + * @note Blocks until every thread currently blocked in pop() has woken up and returned from + * std::condition_variable::wait(). Destroying mutex_/cv_ while a thread is still inside wait() is undefined + * behavior even after notify_all() has been called, since the woken thread may still be re-acquiring the + * lock internally. */ - ~MPMCPriorityQueue() { close(); } + ~MPMCPriorityQueue() { + close(); + std::unique_lock lock(mutex_); + no_waiters_cv_.wait(lock, [this] { return waiters_ == 0; }); + } // Disable copy and move to prevent issues with condition variables MPMCPriorityQueue(const MPMCPriorityQueue&) = delete; @@ -133,7 +142,9 @@ class MPMCPriorityQueue { std::unique_lock lock(mutex_); // Wait until queue has elements or is closed + ++waiters_; cv_.wait(lock, [this] { return closed_ || !pq_.empty(); }); + if (--waiters_ == 0) { no_waiters_cv_.notify_all(); } // Try to pop an element if (!pq_.empty()) { @@ -213,6 +224,8 @@ class MPMCPriorityQueue { private: mutable std::mutex mutex_; std::condition_variable cv_; + std::condition_variable no_waiters_cv_; + size_type waiters_{0}; bool closed_{false}; std::priority_queue< T, std::vector< T >, Compare > pq_; }; diff --git a/src/fds/CMakeLists.txt b/src/fds/CMakeLists.txt index 0889d8cb..ecfec05d 100644 --- a/src/fds/CMakeLists.txt +++ b/src/fds/CMakeLists.txt @@ -94,6 +94,8 @@ if(BUILD_TESTING) ) target_link_libraries(test_bounded_mpmc_queue sisl_buffer GTest::gtest) add_test(NAME BoundedMPMCQueue COMMAND test_bounded_mpmc_queue) + set_tests_properties(BoundedMPMCQueue PROPERTIES + ENVIRONMENT "TSAN_OPTIONS=suppressions=${CMAKE_SOURCE_DIR}/tsan.supp") add_executable(test_lru_map) target_sources(test_lru_map PRIVATE diff --git a/tsan.supp b/tsan.supp index e2d68091..48846c89 100644 --- a/tsan.supp +++ b/tsan.supp @@ -50,3 +50,9 @@ race:default_free race:sisl::wisr_framework race:sisl::urcu_scoped_ptr +# test_bounded_mpmc_queue: boost::lockfree::queue's freelist/tagged-pointer CAS reclamation is +# itself lock-free and provides a happens-before that TSAN cannot model, same class as the URCU +# suppressions above. Every reported race is internal to do_push/pop, not to sisl::BoundedMPMCQueue +# (its own m_size counter is a plain fetch_add/fetch_sub, already correctly ordered). +race:boost::lockfree +