From 48c2d78dfb4972d6125e053b53890280bdec1255 Mon Sep 17 00:00:00 2001 From: mrabine Date: Thu, 8 Oct 2026 18:18:03 +0200 Subject: [PATCH] Add proactor timer wheel --- core/include/join/proactor.hpp | 142 ++++++++++++++++++++-- core/include/join/proactor_epoll_impl.hpp | 9 ++ core/include/join/proactor_uring_impl.hpp | 66 ++++++++-- core/tests/hybrid_proactor_test.cpp | 94 ++++++++++++++ core/tests/proactor_test.cpp | 94 ++++++++++++++ core/tests/sqpoll_proactor_test.cpp | 94 ++++++++++++++ 6 files changed, 476 insertions(+), 23 deletions(-) diff --git a/core/include/join/proactor.hpp b/core/include/join/proactor.hpp index 2465a7b4..c500dc33 100644 --- a/core/include/join/proactor.hpp +++ b/core/include/join/proactor.hpp @@ -31,6 +31,7 @@ #include #ifdef JOIN_HAS_IO_URING #include +#include #else #include #endif @@ -154,10 +155,115 @@ class join::BasicProactor class join::BasicProactor : public join::EventHandler #endif { +private: + /// command queue size. + static constexpr size_t _queueSize = 1024; + +#ifdef JOIN_HAS_IO_URING + /** + * @brief timer wheel run policy, letting the event loop advance the wheel. + */ + class WheelPolicy + { + public: + /** + * @brief create instance. + * @param proactor proactor advancing the wheel. + */ + explicit WheelPolicy (BasicProactor& proactor) noexcept + : _proactor (proactor) + { + } + + /** + * @brief no-op, the event loop advances the wheel. + * @param wheel wheel to advance. + * @param period ignored, the event loop waits one tick at most while timers are armed. + * @return 0. + */ + template + int start ([[maybe_unused]] Wheel& wheel, [[maybe_unused]] std::chrono::nanoseconds period) noexcept + { + // do nothing, the event loop advances the wheel. + return 0; + } + + /** + * @brief no-op, the event loop advances the wheel. + */ + void stop () noexcept + { + // do nothing, the event loop advances the wheel. + } + + /** + * @brief check if the calling thread is the proactor thread. + * @return true if called from the proactor thread. + */ + bool isRunnerThread () const noexcept + { + return _proactor.isProactorThread (); + } + + /** + * @brief wake the event loop up without waiting for it, so that it advances the wheel. + * @param wheel wheel to advance. + * @return 0 on success, -1 if the event loop is not running. + */ + template + int flush ([[maybe_unused]] Wheel& wheel) noexcept + { + if (JOIN_UNLIKELY (!_proactor.isRunning ())) + { + return -1; + } + + return _proactor.wakeup (); + } + + private: + /** + * @brief advance the wheel, to be called by the event loop. + * @param wheel wheel to advance. + * @return number of ticks processed. + */ + template + static uint64_t advance (Wheel& wheel) noexcept + { + return wheel.advance (); + } + + /** + * @brief check if no timer is armed, to be called by the event loop. + * @param wheel wheel to check. + * @return true if no timer is armed. + */ + template + static bool empty (const Wheel& wheel) noexcept + { + return wheel.empty (); + } + + /// proactor advancing the wheel. + BasicProactor& _proactor; + + /// friendship with proactor. + friend class BasicProactor; + }; +#endif + public: /// function invoked on the proactor thread. using InvokeHandler = Function; +#ifdef JOIN_HAS_IO_URING + /// timer wheel, able to arm a timer for every queued command. + using Wheel = BasicWheel; +#else + /// timer wheel of the underlying reactor. + using Wheel = Reactor::Wheel; +#endif + /** * @brief initialize the proactor and its I/O backend. */ @@ -302,7 +408,7 @@ class join::BasicProactor : public join::EventHandler #ifdef JOIN_HAS_NUMA /** - * @brief bind proactor command queue memory to a NUMA node. + * @brief bind proactor command queue and timer wheel memory to a NUMA node. * @param numa NUMA node ID. * @return 0 on success, -1 on failure. */ @@ -310,7 +416,7 @@ class join::BasicProactor : public join::EventHandler #endif /** - * @brief lock proactor command queue memory in RAM. + * @brief lock proactor command queue and timer wheel memory in RAM. * @return 0 on success, -1 on failure. */ int mlock () const noexcept; @@ -327,6 +433,12 @@ class join::BasicProactor : public join::EventHandler */ bool isProactorThread () const noexcept; + /** + * @brief get the timer wheel advanced by the event loop. + * @return timer wheel. + */ + Wheel& wheel () noexcept; + private: #ifdef JOIN_HAS_IO_URING /** @@ -434,18 +546,22 @@ class join::BasicProactor : public join::EventHandler #ifdef JOIN_HAS_IO_URING /** - * @brief push command to queue and write to the wakeup eventfd to wake the dispatcher. - * @param cmd command to write. + * @brief wake the event loop up if it is asleep. * @return 0 on success, -1 on failure. */ - int writeCommand (const Command& cmd, std::true_type) noexcept; + int wakeup () noexcept; /** - * @brief push command to queue; the polling event loop drains it without a wakeup signal. - * @param cmd command to write. + * @brief write to the wakeup eventfd if the event loop is asleep. * @return 0 on success, -1 on failure. */ - int writeCommand (const Command& cmd, std::false_type) noexcept; + int wakeup (std::true_type) noexcept; + + /** + * @brief no-op, the polling event loop needs no wakeup signal. + * @return 0. + */ + int wakeup (std::false_type) noexcept; #endif /** @@ -632,8 +748,10 @@ class join::BasicProactor : public join::EventHandler static bool isWriteOperation (const IoOperation& op) noexcept; #endif - /// command queue size. - static constexpr size_t _queueSize = 1024; +#ifdef JOIN_HAS_IO_URING + /// timer wheel advanced by the event loop, built first so that nothing leaks if it throws. + Wheel _wheel{*this}; +#endif /// command queue. LocalMem::Mpsc::Queue _commands; @@ -1059,7 +1177,7 @@ class join::BasicProactorThread #ifdef JOIN_HAS_NUMA /** - * @brief bind proactor command queue memory to a NUMA node. + * @brief bind proactor command queue and timer wheel memory to a NUMA node. * @param numa NUMA node ID. * @return 0 on success, -1 on failure. */ @@ -1070,7 +1188,7 @@ class join::BasicProactorThread #endif /** - * @brief lock proactor command queue memory in RAM. + * @brief lock proactor command queue and timer wheel memory in RAM. * @return 0 on success, -1 on failure. */ static int mlock () noexcept diff --git a/core/include/join/proactor_epoll_impl.hpp b/core/include/join/proactor_epoll_impl.hpp index 1ecfea34..94f1d6a3 100644 --- a/core/include/join/proactor_epoll_impl.hpp +++ b/core/include/join/proactor_epoll_impl.hpp @@ -313,6 +313,15 @@ inline bool join::BasicProactor::isProactorThread () const noexcept return _reactor.isReactorThread (); } +// ========================================================================= +// CLASS : BasicProactor +// METHOD : wheel +// ========================================================================= +inline join::BasicProactor::Wheel& join::BasicProactor::wheel () noexcept +{ + return _reactor.wheel (); +} + // ========================================================================= // CLASS : BasicProactor // METHOD : writeCommand diff --git a/core/include/join/proactor_uring_impl.hpp b/core/include/join/proactor_uring_impl.hpp index acd9bacd..eee67253 100644 --- a/core/include/join/proactor_uring_impl.hpp +++ b/core/include/join/proactor_uring_impl.hpp @@ -382,6 +382,11 @@ int join::BasicProactor::unregisterBufferRing (uint16_t group) template int join::BasicProactor::mbind (int numa) const noexcept { + if (_wheel.mbind (numa) == -1) + { + return -1; // LCOV_EXCL_LINE + } + return _commands.mbind (numa); } #endif @@ -393,6 +398,11 @@ int join::BasicProactor::mbind (int numa) const noexcept template int join::BasicProactor::mlock () const noexcept { + if (_wheel.mlock () == -1) + { + return -1; // LCOV_EXCL_LINE + } + return _commands.mlock (); } @@ -416,6 +426,16 @@ bool join::BasicProactor::isProactorThread () const noexcept return _threadId.load (std::memory_order_acquire) == pthread_self (); } +// ========================================================================= +// CLASS : BasicProactor +// METHOD : wheel +// ========================================================================= +template +typename join::BasicProactor::Wheel& join::BasicProactor::wheel () noexcept +{ + return _wheel; +} + // ========================================================================= // CLASS : BasicProactor // METHOD : initWakeup @@ -525,21 +545,31 @@ void join::BasicProactor::initSqThreadCpu (io_uring_params& params, st template int join::BasicProactor::writeCommand (const Command& cmd) noexcept { - return writeCommand (cmd, is_default{}); + if (JOIN_UNLIKELY (_commands.push (cmd) == -1)) + { + return -1; // LCOV_EXCL_LINE + } + + return wakeup (); } // ========================================================================= // CLASS : BasicProactor -// METHOD : writeCommand +// METHOD : wakeup // ========================================================================= template -int join::BasicProactor::writeCommand (const Command& cmd, std::true_type) noexcept +int join::BasicProactor::wakeup () noexcept { - if (JOIN_UNLIKELY (_commands.push (cmd) == -1)) - { - return -1; // LCOV_EXCL_LINE - } + return wakeup (is_default{}); +} +// ========================================================================= +// CLASS : BasicProactor +// METHOD : wakeup +// ========================================================================= +template +int join::BasicProactor::wakeup (std::true_type) noexcept +{ // pairs with the fence in the event loop: an acquire load would let the push above be // reordered after it, the loop would sleep and the wakeup would be lost. std::atomic_thread_fence (std::memory_order_seq_cst); @@ -564,12 +594,12 @@ int join::BasicProactor::writeCommand (const Command& cmd, std::true_t // ========================================================================= // CLASS : BasicProactor -// METHOD : writeCommand +// METHOD : wakeup // ========================================================================= template -int join::BasicProactor::writeCommand (const Command& cmd, std::false_type) noexcept +int join::BasicProactor::wakeup (std::false_type) noexcept { - return _commands.push (cmd); + return 0; } // ========================================================================= @@ -1091,11 +1121,13 @@ void join::BasicProactor::eventLoop (std::false_type, std::false_type) submitOperation (_wakeupOp, true); } + __kernel_timespec tick{0, Wheel::resolution ().count ()}; Backoff backoff; bool running; while ((running = _running.load (std::memory_order_acquire)) || !_pendingOps.empty ()) { + WheelPolicy::advance (_wheel); readCommands (); io_uring_submit (&_ring); @@ -1123,13 +1155,21 @@ void join::BasicProactor::eventLoop (std::false_type, std::false_type) _wakeupState.store (WakeupState::Sleeping, std::memory_order_relaxed); std::atomic_thread_fence (std::memory_order_seq_cst); readCommands (); + WheelPolicy::advance (_wheel); io_uring_submit (&_ring); if (JOIN_LIKELY (_running.load (std::memory_order_acquire))) { if (io_uring_peek_cqe (&_ring, &cqe) != 0) { - io_uring_wait_cqe (&_ring, &cqe); + if (WheelPolicy::empty (_wheel)) + { + io_uring_wait_cqe (&_ring, &cqe); + } + else + { + io_uring_wait_cqe_timeout (&_ring, &cqe, &tick); + } } } @@ -1149,6 +1189,8 @@ void join::BasicProactor::eventLoop (std::true_type, std::false_type) while ((running = _running.load (std::memory_order_acquire)) || !_pendingOps.empty ()) { + WheelPolicy::advance (_wheel); + if (JOIN_LIKELY (running)) { readCommands (); @@ -1188,6 +1230,8 @@ void join::BasicProactor::eventLoop (std::true_type, std::true_type) n while ((running = _running.load (std::memory_order_acquire)) || !_pendingOps.empty ()) { + WheelPolicy::advance (_wheel); + if (JOIN_LIKELY (running)) { readCommands (); diff --git a/core/tests/hybrid_proactor_test.cpp b/core/tests/hybrid_proactor_test.cpp index f97b7fc5..18e2d9dc 100644 --- a/core/tests/hybrid_proactor_test.cpp +++ b/core/tests/hybrid_proactor_test.cpp @@ -31,6 +31,13 @@ // Libraries. #include +// C++. +#include +#include +#include + +using namespace std::chrono_literals; + using join::Errc; using join::Mutex; using join::Condition; @@ -806,6 +813,93 @@ TEST_F (HybridProactorTest, waitStopped) th.join (); } +/** + * @brief Test wheel. + */ +TEST_F (HybridProactorTest, wheel) +{ + HybridProactor proactor; + std::atomic before{0}, after{0}, inside{0}, count{0}; + std::atomic elapsed{0}; + + ASSERT_GT (proactor.wheel ().setOneShot (10ms, + [&before] () { + ++before; + }), + 0); + + ssize_t idle = proactor.wheel ().setOneShot (10ms, [] () { + }); + ASSERT_GT (idle, 0); + ASSERT_EQ (proactor.wheel ().cancel (idle, true), -1); + ASSERT_EQ (join::lastError, Errc::OperationFailed); + + Thread th ([&proactor] () { + proactor.run (); + }); + while (!proactor.isRunning ()) + { + } + + std::this_thread::sleep_for (50ms); + EXPECT_EQ (before, 1); + + const auto origin = std::chrono::steady_clock::now (); + ASSERT_GT (proactor.wheel ().setOneShot (30ms, + [&after, &elapsed, origin] () { + elapsed = std::chrono::duration_cast ( + std::chrono::steady_clock::now () - origin) + .count (); + ++after; + }), + 0); + + HybridProactor::InvokeHandler fn = [&proactor, &inside] () { + proactor.wheel ().setOneShot (10ms, [&inside] () { + ++inside; + }); + }; + ASSERT_EQ (proactor.invoke (&fn), 0) << join::lastError.message (); + + std::this_thread::sleep_for (100ms); + EXPECT_EQ (after, 1); + EXPECT_GE (elapsed, 30000); + EXPECT_EQ (inside, 1); + + // armed from the proactor thread with an empty wheel. + ASSERT_EQ (proactor.invoke (&fn), 0) << join::lastError.message (); + std::this_thread::sleep_for (30ms); + EXPECT_EQ (inside, 2); + + ssize_t id = -1; + HybridProactor::InvokeHandler interval = [&proactor, &count, &id] () { + id = proactor.wheel ().setInterval (5ms, [&count] () { + ++count; + }); + }; + ASSERT_EQ (proactor.invoke (&interval), 0) << join::lastError.message (); + ASSERT_GT (id, 0); + std::this_thread::sleep_for (100ms); + ASSERT_EQ (proactor.wheel ().cancel (id, true), 0) << join::lastError.message (); + const int fired = count; + EXPECT_GT (fired, 10); + std::this_thread::sleep_for (50ms); + EXPECT_EQ (count, fired); + + id = proactor.wheel ().setOneShot (50ms, [&count] () { + ++count; + }); + ASSERT_GT (id, 0); + ASSERT_EQ (proactor.wheel ().cancel (id, true), 0) << join::lastError.message (); + std::this_thread::sleep_for (100ms); + EXPECT_EQ (count, fired); + ASSERT_EQ (proactor.wheel ().cancel (id), -1); + ASSERT_EQ (join::lastError, Errc::NotFound); + + proactor.stop (); + th.join (); +} + #ifdef JOIN_HAS_IO_URING /** * @brief Test registerFixedBuffers and unregisterFixedBuffers. diff --git a/core/tests/proactor_test.cpp b/core/tests/proactor_test.cpp index af30b90b..29a23000 100644 --- a/core/tests/proactor_test.cpp +++ b/core/tests/proactor_test.cpp @@ -31,6 +31,13 @@ // Libraries. #include +// C++. +#include +#include +#include + +using namespace std::chrono_literals; + using join::Errc; using join::Mutex; using join::Condition; @@ -917,6 +924,93 @@ TEST_F (ProactorTest, waitStopped) th.join (); } +/** + * @brief Test wheel. + */ +TEST_F (ProactorTest, wheel) +{ + Proactor proactor; + std::atomic before{0}, after{0}, inside{0}, count{0}; + std::atomic elapsed{0}; + + ASSERT_GT (proactor.wheel ().setOneShot (10ms, + [&before] () { + ++before; + }), + 0); + + ssize_t idle = proactor.wheel ().setOneShot (10ms, [] () { + }); + ASSERT_GT (idle, 0); + ASSERT_EQ (proactor.wheel ().cancel (idle, true), -1); + ASSERT_EQ (join::lastError, Errc::OperationFailed); + + Thread th ([&proactor] () { + proactor.run (); + }); + while (!proactor.isRunning ()) + { + } + + std::this_thread::sleep_for (50ms); + EXPECT_EQ (before, 1); + + const auto origin = std::chrono::steady_clock::now (); + ASSERT_GT (proactor.wheel ().setOneShot (30ms, + [&after, &elapsed, origin] () { + elapsed = std::chrono::duration_cast ( + std::chrono::steady_clock::now () - origin) + .count (); + ++after; + }), + 0); + + Proactor::InvokeHandler fn = [&proactor, &inside] () { + proactor.wheel ().setOneShot (10ms, [&inside] () { + ++inside; + }); + }; + ASSERT_EQ (proactor.invoke (&fn), 0) << join::lastError.message (); + + std::this_thread::sleep_for (100ms); + EXPECT_EQ (after, 1); + EXPECT_GE (elapsed, 30000); + EXPECT_EQ (inside, 1); + + // armed from the proactor thread with an empty wheel. + ASSERT_EQ (proactor.invoke (&fn), 0) << join::lastError.message (); + std::this_thread::sleep_for (30ms); + EXPECT_EQ (inside, 2); + + ssize_t id = -1; + Proactor::InvokeHandler interval = [&proactor, &count, &id] () { + id = proactor.wheel ().setInterval (5ms, [&count] () { + ++count; + }); + }; + ASSERT_EQ (proactor.invoke (&interval), 0) << join::lastError.message (); + ASSERT_GT (id, 0); + std::this_thread::sleep_for (100ms); + ASSERT_EQ (proactor.wheel ().cancel (id, true), 0) << join::lastError.message (); + const int fired = count; + EXPECT_GT (fired, 10); + std::this_thread::sleep_for (50ms); + EXPECT_EQ (count, fired); + + id = proactor.wheel ().setOneShot (50ms, [&count] () { + ++count; + }); + ASSERT_GT (id, 0); + ASSERT_EQ (proactor.wheel ().cancel (id, true), 0) << join::lastError.message (); + std::this_thread::sleep_for (100ms); + EXPECT_EQ (count, fired); + ASSERT_EQ (proactor.wheel ().cancel (id), -1); + ASSERT_EQ (join::lastError, Errc::NotFound); + + proactor.stop (); + th.join (); +} + #ifdef JOIN_HAS_IO_URING /** * @brief Test registerFixedBuffers and unregisterFixedBuffers. diff --git a/core/tests/sqpoll_proactor_test.cpp b/core/tests/sqpoll_proactor_test.cpp index 736b1336..cee13ee3 100644 --- a/core/tests/sqpoll_proactor_test.cpp +++ b/core/tests/sqpoll_proactor_test.cpp @@ -31,6 +31,13 @@ // Libraries. #include +// C++. +#include +#include +#include + +using namespace std::chrono_literals; + using join::Errc; using join::Mutex; using join::Condition; @@ -806,6 +813,93 @@ TEST_F (SqpollProactorTest, waitStopped) th.join (); } +/** + * @brief Test wheel. + */ +TEST_F (SqpollProactorTest, wheel) +{ + SqpollProactor proactor; + std::atomic before{0}, after{0}, inside{0}, count{0}; + std::atomic elapsed{0}; + + ASSERT_GT (proactor.wheel ().setOneShot (10ms, + [&before] () { + ++before; + }), + 0); + + ssize_t idle = proactor.wheel ().setOneShot (10ms, [] () { + }); + ASSERT_GT (idle, 0); + ASSERT_EQ (proactor.wheel ().cancel (idle, true), -1); + ASSERT_EQ (join::lastError, Errc::OperationFailed); + + Thread th ([&proactor] () { + proactor.run (); + }); + while (!proactor.isRunning ()) + { + } + + std::this_thread::sleep_for (50ms); + EXPECT_EQ (before, 1); + + const auto origin = std::chrono::steady_clock::now (); + ASSERT_GT (proactor.wheel ().setOneShot (30ms, + [&after, &elapsed, origin] () { + elapsed = std::chrono::duration_cast ( + std::chrono::steady_clock::now () - origin) + .count (); + ++after; + }), + 0); + + SqpollProactor::InvokeHandler fn = [&proactor, &inside] () { + proactor.wheel ().setOneShot (10ms, [&inside] () { + ++inside; + }); + }; + ASSERT_EQ (proactor.invoke (&fn), 0) << join::lastError.message (); + + std::this_thread::sleep_for (100ms); + EXPECT_EQ (after, 1); + EXPECT_GE (elapsed, 30000); + EXPECT_EQ (inside, 1); + + // armed from the proactor thread with an empty wheel. + ASSERT_EQ (proactor.invoke (&fn), 0) << join::lastError.message (); + std::this_thread::sleep_for (30ms); + EXPECT_EQ (inside, 2); + + ssize_t id = -1; + SqpollProactor::InvokeHandler interval = [&proactor, &count, &id] () { + id = proactor.wheel ().setInterval (5ms, [&count] () { + ++count; + }); + }; + ASSERT_EQ (proactor.invoke (&interval), 0) << join::lastError.message (); + ASSERT_GT (id, 0); + std::this_thread::sleep_for (100ms); + ASSERT_EQ (proactor.wheel ().cancel (id, true), 0) << join::lastError.message (); + const int fired = count; + EXPECT_GT (fired, 10); + std::this_thread::sleep_for (50ms); + EXPECT_EQ (count, fired); + + id = proactor.wheel ().setOneShot (50ms, [&count] () { + ++count; + }); + ASSERT_GT (id, 0); + ASSERT_EQ (proactor.wheel ().cancel (id, true), 0) << join::lastError.message (); + std::this_thread::sleep_for (100ms); + EXPECT_EQ (count, fired); + ASSERT_EQ (proactor.wheel ().cancel (id), -1); + ASSERT_EQ (join::lastError, Errc::NotFound); + + proactor.stop (); + th.join (); +} + #ifdef JOIN_HAS_IO_URING /** * @brief Test registerFixedBuffers and unregisterFixedBuffers.