From fee0d3eb7feee5796384432f844fd3891f40604a Mon Sep 17 00:00:00 2001 From: mrabine Date: Sat, 29 Aug 2026 17:44:09 +0200 Subject: [PATCH] Use proactor backend in timers --- core/include/join/backoff.hpp | 9 + core/include/join/proactor.hpp | 38 +++- core/include/join/proactor_epoll_impl.hpp | 79 +++++---- core/include/join/proactor_uring_impl.hpp | 103 ++++++----- core/include/join/reactor.hpp | 9 +- core/include/join/timer.hpp | 206 ++++++++++++++++++---- core/src/reactor.cpp | 40 +++-- core/tests/CMakeLists.txt | 5 + core/tests/backoff_test.cpp | 64 +++++++ core/tests/hybrid_proactor_test.cpp | 71 ++++++++ core/tests/monotonic_timer_test.cpp | 74 +++++--- core/tests/proactor_test.cpp | 71 ++++++++ core/tests/reactor_test.cpp | 36 ++++ core/tests/realtime_timer_test.cpp | 74 +++++--- core/tests/sqpoll_proactor_test.cpp | 71 ++++++++ 15 files changed, 763 insertions(+), 187 deletions(-) create mode 100644 core/tests/backoff_test.cpp diff --git a/core/include/join/backoff.hpp b/core/include/join/backoff.hpp index 35547d9f..13756a59 100644 --- a/core/include/join/backoff.hpp +++ b/core/include/join/backoff.hpp @@ -80,6 +80,15 @@ namespace join _count = 0; } + /** + * @brief check if initial spin phase is over. + * @return true if initial spin phase is over. + */ + bool spinExhausted () const noexcept + { + return _count >= _spin; + } + private: /// number of spin iterations before yielding. size_t _spin; diff --git a/core/include/join/proactor.hpp b/core/include/join/proactor.hpp index 44fc2d37..1e26133a 100644 --- a/core/include/join/proactor.hpp +++ b/core/include/join/proactor.hpp @@ -220,11 +220,16 @@ class join::BasicProactor : public join::EventHandler void run (); /** - * @brief stop the event loop. - * @param sync wait for loop termination if true. + * @brief stop the event loop, ignored if no event loop is running. + * @param sync wait for the event loop thread to terminate. */ void stop (bool sync = true) noexcept; + /** + * @brief wait for the event loop thread to terminate. + */ + void waitStopped () const noexcept; + #ifdef JOIN_HAS_IO_URING /** * @brief register fixed buffers with the io_uring instance. @@ -328,6 +333,16 @@ class join::BasicProactor : public join::EventHandler void initSqThreadCpu (io_uring_params& params, std::true_type) noexcept; #endif + /** + * @brief event loop wakeup state. + */ + enum class WakeupState + { + Pending, /**< the event loop is asleep or about to sleep, a wakeup write is required. */ + Notified, /**< a wakeup has already been posted and not yet consumed. */ + Polling, /**< the event loop is busy-polling the command queue, no wakeup write is required. */ + }; + /** * @brief command type for proactor dispatcher. */ @@ -523,21 +538,22 @@ class join::BasicProactor : public join::EventHandler /// command queue size. static constexpr size_t _queueSize = 1024; - /// coalesce eventfd writes. - alignas (64) std::atomic _notified{false}; - /// command queue. LocalMem::Mpsc::Queue _commands; - /// set to true while a sync stop() is in progress. - std::atomic _stopping{false}; + /// event loop wakeup state. +#ifdef JOIN_HAS_IO_URING + alignas (64) std::atomic _wakeupState{WakeupState::Polling}; +#else + alignas (64) std::atomic _wakeupState{WakeupState::Pending}; +#endif /// eventfd descriptor. int _wakeup = -1; #ifdef JOIN_HAS_IO_URING /// buffer for the wakeup eventfd read. - uint64_t _wakeupBuf = 0; + alignas (64) uint64_t _wakeupBuf = 0; /// internal operation used to watch the wakeup eventfd. IoOperation _wakeupOp = {}; @@ -812,6 +828,12 @@ class join::BasicProactorThread _dispatcher = Thread ([this] () { _proactor.run (); }); + + Backoff backoff; + while (!_proactor.isRunning ()) + { + backoff (); + } } /** diff --git a/core/include/join/proactor_epoll_impl.hpp b/core/include/join/proactor_epoll_impl.hpp index 26bac1e5..773b0a8f 100644 --- a/core/include/join/proactor_epoll_impl.hpp +++ b/core/include/join/proactor_epoll_impl.hpp @@ -57,6 +57,7 @@ inline join::BasicProactor::~BasicProactor () noexcept // ========================================================================= inline void join::BasicProactor::run () { + _wakeupState.store (WakeupState::Pending, std::memory_order_seq_cst); _reactor.run (); } @@ -73,40 +74,26 @@ inline void join::BasicProactor::stop (bool sync) noexcept return; } - if (!_reactor.isRunning ()) + if (JOIN_UNLIKELY (!isRunning ())) { return; } - std::atomic done{false}; + writeCommand ({CommandType::Stop, nullptr, sync, nullptr, nullptr}); if (JOIN_LIKELY (sync)) { - bool expected = false; - if (!_stopping.compare_exchange_strong (expected, true, std::memory_order_acq_rel)) - { - Backoff backoff; - while (isRunning ()) - { - backoff (); - } - return; - } - } - - writeCommand ({CommandType::Stop, nullptr, sync, sync ? &done : nullptr, nullptr}); - - if (JOIN_LIKELY (sync)) - { - Backoff backoff; - while (!done.load (std::memory_order_acquire)) - { - backoff (); - } - _stopping.store (false, std::memory_order_release); + waitStopped (); } +} - _reactor.stop (sync); +// ========================================================================= +// CLASS : BasicProactor +// METHOD : waitStopped +// ========================================================================= +inline void join::BasicProactor::waitStopped () const noexcept +{ + _reactor.waitStopped (); } #ifdef JOIN_HAS_NUMA @@ -168,15 +155,23 @@ inline int join::BasicProactor::writeCommand (const Command& cmd) noexcept return -1; // LCOV_EXCL_LINE } - if (_notified.exchange (true, std::memory_order_acq_rel)) + // 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); + + WakeupState state = _wakeupState.load (std::memory_order_relaxed); + if (state != WakeupState::Pending) { return 0; } - uint64_t value = 1; - if (JOIN_UNLIKELY (::write (_wakeup, &value, sizeof (uint64_t)) == -1)) + if (_wakeupState.compare_exchange_strong (state, WakeupState::Notified, std::memory_order_seq_cst)) { - _notified.store (false); // LCOV_EXCL_LINE + uint64_t value = 1; + if (JOIN_UNLIKELY (::write (_wakeup, &value, sizeof (uint64_t)) == -1)) + { + _wakeupState.store (WakeupState::Pending, std::memory_order_seq_cst); // LCOV_EXCL_LINE + } } return 0; @@ -189,13 +184,9 @@ inline int join::BasicProactor::writeCommand (const Command& cmd) noexcept inline void join::BasicProactor::readCommands () noexcept { uint64_t count; - ssize_t nread = ::read (_wakeup, &count, sizeof (count)); - _notified.store (false); - - if (JOIN_UNLIKELY (nread == -1)) - { - return; // LCOV_EXCL_LINE - } + [[maybe_unused]] ssize_t nread = ::read (_wakeup, &count, sizeof (count)); + _wakeupState.store (WakeupState::Pending, std::memory_order_relaxed); + std::atomic_thread_fence (std::memory_order_seq_cst); Command cmd; while (_commands.tryPop (cmd) == 0) @@ -229,6 +220,7 @@ inline void join::BasicProactor::processCommand (const Command& cmd) noexcept case CommandType::Stop: cancelAllOperations (); + _reactor.stop (false); break; default: @@ -558,8 +550,12 @@ inline void join::BasicProactor::onReadable (int fd) noexcept readCommands (); return; } - - endOperation (_readOps[fd], executeOp (_readOps[fd]), false); + IoOperation* op = _readOps[fd]; + if (JOIN_UNLIKELY (op == nullptr)) + { + return; + } + endOperation (op, executeOp (op), false); } // ========================================================================= @@ -568,7 +564,12 @@ inline void join::BasicProactor::onReadable (int fd) noexcept // ========================================================================= inline void join::BasicProactor::onWriteable (int fd) noexcept { - endOperation (_writeOps[fd], executeOp (_writeOps[fd]), false); + IoOperation* op = _writeOps[fd]; + if (JOIN_UNLIKELY (op == nullptr)) + { + return; + } + endOperation (op, executeOp (op), false); } // ========================================================================= diff --git a/core/include/join/proactor_uring_impl.hpp b/core/include/join/proactor_uring_impl.hpp index b5383b63..3ac74a92 100644 --- a/core/include/join/proactor_uring_impl.hpp +++ b/core/include/join/proactor_uring_impl.hpp @@ -149,35 +149,31 @@ void join::BasicProactor::stop (bool sync) noexcept return; } - if (!isRunning ()) + if (JOIN_UNLIKELY (!isRunning ())) { return; } + writeCommand ({CommandType::Stop, nullptr, sync, nullptr, nullptr}); + if (JOIN_LIKELY (sync)) { - bool expected = false; - if (!_stopping.compare_exchange_strong (expected, true, std::memory_order_acq_rel)) - { - Backoff backoff; - while (_threadId.load (std::memory_order_acquire) != _invalidThreadId) - { - backoff (); - } - return; - } + waitStopped (); } +} - writeCommand ({CommandType::Stop, nullptr, sync, nullptr, nullptr}); +// ========================================================================= +// CLASS : BasicProactor +// METHOD : waitStopped +// ========================================================================= +template +void join::BasicProactor::waitStopped () const noexcept +{ + Backoff backoff; - if (JOIN_LIKELY (sync)) + while (_threadId.load (std::memory_order_acquire) != _invalidThreadId) { - Backoff backoff; - while (_threadId.load (std::memory_order_acquire) != _invalidThreadId) - { - backoff (); - } - _stopping.store (false, std::memory_order_release); + backoff (); } } @@ -244,7 +240,7 @@ int join::BasicProactor::mlock () const noexcept template bool join::BasicProactor::isRunning () const noexcept { - return _running.load (std::memory_order_acquire); + return _threadId.load (std::memory_order_acquire) != _invalidThreadId; } // ========================================================================= @@ -380,15 +376,23 @@ int join::BasicProactor::writeCommand (const Command& cmd, std::true_typ return -1; // LCOV_EXCL_LINE } - if (_notified.exchange (true, std::memory_order_acq_rel)) + // 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); + + WakeupState state = _wakeupState.load (std::memory_order_relaxed); + if (state != WakeupState::Pending) { return 0; } - uint64_t value = 1; - if (JOIN_UNLIKELY (::write (_wakeup, &value, sizeof (uint64_t)) == -1)) + if (_wakeupState.compare_exchange_strong (state, WakeupState::Notified, std::memory_order_seq_cst)) { - _notified.store (false); // LCOV_EXCL_LINE + uint64_t value = 1; + if (JOIN_UNLIKELY (::write (_wakeup, &value, sizeof (uint64_t)) == -1)) + { + _wakeupState.store (WakeupState::Pending, std::memory_order_seq_cst); // LCOV_EXCL_LINE + } } return 0; @@ -734,9 +738,7 @@ void join::BasicProactor::dispatchCqe (io_uring_cqe* cqe, std::true_type if (JOIN_UNLIKELY (op == &_wakeupOp)) { _wakeupOp.state = IoOperation::State::Idle; - _notified.store (false); - readCommands (); if (JOIN_LIKELY (_running.load (std::memory_order_acquire))) { rearmWakeup (); @@ -799,34 +801,49 @@ void join::BasicProactor::eventLoop (std::false_type, std::false_type) n rearmWakeup (); } - while (_running.load (std::memory_order_acquire) || !_pendingOps.empty ()) + Backoff backoff; + bool running; + + while ((running = _running.load (std::memory_order_acquire)) || !_pendingOps.empty ()) { - io_uring_cqe* cqe = nullptr; + readCommands (); + io_uring_submit (&_ring); - if (JOIN_LIKELY (_running.load (std::memory_order_acquire))) + io_uring_cqe* cqe = nullptr; + if (io_uring_peek_cqe (&_ring, &cqe) == 0) { - if (JOIN_UNLIKELY (io_uring_wait_cqe (&_ring, &cqe) < 0)) + backoff.reset (); + + do { - continue; // LCOV_EXCL_LINE + dispatchCqe (cqe); + io_uring_cqe_seen (&_ring, cqe); } + while (io_uring_peek_cqe (&_ring, &cqe) == 0); + + continue; } - else + + if (!running || !backoff.spinExhausted ()) { - // LCOV_EXCL_START - io_uring_submit (&_ring); - if (JOIN_UNLIKELY (io_uring_peek_cqe (&_ring, &cqe) != 0)) - { - continue; - } - // LCOV_EXCL_STOP + backoff (); + continue; } - do + _wakeupState.store (WakeupState::Pending, std::memory_order_relaxed); + std::atomic_thread_fence (std::memory_order_seq_cst); + readCommands (); + + io_uring_submit (&_ring); + if (JOIN_LIKELY (_running.load (std::memory_order_acquire))) { - dispatchCqe (cqe); - io_uring_cqe_seen (&_ring, cqe); + if (io_uring_peek_cqe (&_ring, &cqe) != 0) + { + io_uring_wait_cqe (&_ring, &cqe); + } } - while (io_uring_peek_cqe (&_ring, &cqe) == 0); + + _wakeupState.store (WakeupState::Polling, std::memory_order_seq_cst); } } diff --git a/core/include/join/reactor.hpp b/core/include/join/reactor.hpp index ab9b0d88..2ea0b1cb 100644 --- a/core/include/join/reactor.hpp +++ b/core/include/join/reactor.hpp @@ -190,11 +190,16 @@ namespace join void run (); /** - * @brief stop the event loop. - * @param sync wait for loop termination if true. + * @brief stop the event loop, ignored if no event loop is running. + * @param sync wait for the event loop thread to terminate. */ void stop (bool sync = true) noexcept; + /** + * @brief wait for the event loop thread to terminate. + */ + void waitStopped () const noexcept; + #ifdef JOIN_HAS_NUMA /** * @brief bind command queue memory to a NUMA node. diff --git a/core/include/join/timer.hpp b/core/include/join/timer.hpp index af1b3298..09d5a828 100644 --- a/core/include/join/timer.hpp +++ b/core/include/join/timer.hpp @@ -26,16 +26,21 @@ #define JOIN_CORE_TIMER_HPP // libjoin. -#include +#include +#include +#include #include // C++. -#include #include +#include +#include +#include // C. #include #include +#include namespace join { @@ -43,23 +48,43 @@ namespace join * @brief base timer class. */ template - class BasicTimer : protected EventHandler + class BasicTimer : protected CompletionHandler { public: + /** + * @brief timer state. + */ + enum class State + { + Idle, /**< no callback armed. */ + Armed, /**< a callback is armed. */ + Invoking, /**< the callback is being invoked. */ + Closed, /**< the timerfd read operation is no longer in flight. */ + }; + /** * @brief create instance. - * @param reactor event loop reactor. + * @param proactor completion dispatcher. */ - explicit BasicTimer (Reactor& reactor = ReactorThread::reactor ()) + explicit BasicTimer (Proactor& proactor = ProactorThread::proactor ()) : _handle (timerfd_create (ClockPolicy::type (), TFD_NONBLOCK | TFD_CLOEXEC)) - , _reactor (reactor) + , _proactor (proactor) { if (_handle == -1) { - throw std::system_error (errno, std::system_category (), "timerfd_create failed"); + throw std::system_error (errno, std::system_category (), "timerfd_create failed"); // LCOV_EXCL_LINE } - _reactor.addHandler (_handle, this); + _ops->op = IoOperation::makeRead (_handle, &_ops->expirations, + static_cast (sizeof (_ops->expirations)), this); + + if (_proactor.submit (&_ops->op, true, true) == -1) + { + // LCOV_EXCL_START + close (_handle); + throw std::system_error (lastError, "timerfd submit failed"); + // LCOV_EXCL_STOP + } } /** @@ -93,34 +118,42 @@ namespace join */ ~BasicTimer () noexcept { - _reactor.delHandler (_handle); - if (_handle != -1) + cancel (); + + Backoff backoff; + while (_state.load (std::memory_order_acquire) != State::Closed) { - close (_handle); + _proactor.cancel (&_ops->op, true, true); + backoff (); } + + close (_handle); } /** * @brief arm the timer as a one-shot timer. * @param duration timeout duration before timer expires. - * @param callback function to call when timer expires. + * @param callback function to call when timer expires, captures limited to 32 bytes. */ template void setOneShot (std::chrono::duration duration, Func&& callback) { + cancel (); + auto ns = std::chrono::duration_cast (duration); _callback = std::forward (callback); _oneShot = true; _ns = std::chrono::nanoseconds::zero (); + _state.store (State::Armed, std::memory_order_release); auto ts = toTimerSpec (ns); - timerfd_settime (handle (), 0, &ts, nullptr); + timerfd_settime (_handle, 0, &ts, nullptr); } /** * @brief arm the timer as a one-shot timer with absolute time. * @param timePoint absolute time when timer should expire. - * @param callback function to call when timer expires. + * @param callback function to call when timer expires, captures limited to 32 bytes. */ template void setOneShot (std::chrono::time_point timePoint, Func&& callback) @@ -131,31 +164,37 @@ namespace join std::is_same::value), "Clock type mismatch timer policy"); + cancel (); + auto elapsed = timePoint.time_since_epoch (); auto ns = std::chrono::duration_cast (elapsed); _callback = std::forward (callback); _oneShot = true; _ns = std::chrono::nanoseconds::zero (); + _state.store (State::Armed, std::memory_order_release); auto ts = toTimerSpec (ns); - timerfd_settime (handle (), TFD_TIMER_ABSTIME, &ts, nullptr); + timerfd_settime (_handle, TFD_TIMER_ABSTIME, &ts, nullptr); } /** * @brief arm the timer as a periodic timer. * @param duration interval duration between timer expirations. - * @param callback function to call on each timer expiration. + * @param callback function to call on each timer expiration, captures limited to 32 bytes. */ template void setInterval (std::chrono::duration duration, Func&& callback) { + cancel (); + auto ns = std::chrono::duration_cast (duration); _callback = std::forward (callback); _oneShot = false; _ns = ns; + _state.store (State::Armed, std::memory_order_release); auto ts = toTimerSpec (ns, true); - timerfd_settime (handle (), 0, &ts, nullptr); + timerfd_settime (_handle, 0, &ts, nullptr); } /** @@ -163,12 +202,34 @@ namespace join */ void cancel () noexcept { - _callback = nullptr; _oneShot = true; _ns = std::chrono::nanoseconds::zero (); struct itimerspec ts = {}; - timerfd_settime (handle (), 0, &ts, nullptr); + timerfd_settime (_handle, 0, &ts, nullptr); + + if (JOIN_UNLIKELY (_proactor.isProactorThread ())) + { + _state.store (State::Idle, std::memory_order_release); + return; + } + + Backoff backoff; + State expected = State::Armed; + + while (!_state.compare_exchange_strong (expected, State::Idle, std::memory_order_acq_rel, + std::memory_order_acquire)) + { + if (expected != State::Invoking) + { + return; + } + + backoff (); + expected = State::Armed; + } + + _callback = nullptr; } /** @@ -178,7 +239,7 @@ namespace join bool active () const noexcept { struct itimerspec ts = {}; - timerfd_gettime (handle (), &ts); + timerfd_gettime (_handle, &ts); const bool hasValue = (ts.it_value.tv_sec != 0 || ts.it_value.tv_nsec != 0); const bool hasInterval = (ts.it_interval.tv_sec != 0 || ts.it_interval.tv_nsec != 0); return hasValue || hasInterval; @@ -191,7 +252,7 @@ namespace join std::chrono::nanoseconds remaining () const noexcept { struct itimerspec ts = {}; - timerfd_gettime (handle (), &ts); + timerfd_gettime (_handle, &ts); return std::chrono::seconds (ts.it_value.tv_sec) + std::chrono::nanoseconds (ts.it_value.tv_nsec); } @@ -224,22 +285,62 @@ namespace join private: /** - * @brief method called when data are ready to be read on handle. - * @param fd file descriptor. + * @brief method called when the timerfd read completes. + * @param op completed operation. + * @param result bytes read, or negative errno. */ - virtual void onReadable ([[maybe_unused]] int fd) override + void onComplete ([[maybe_unused]] IoOperation* op, int result) override { - uint64_t expirations; - ssize_t result = read (handle (), &expirations, sizeof (expirations)); - if (result == sizeof (expirations) && _callback) + uint64_t expirations = _ops->expirations; + + if (JOIN_UNLIKELY (result != static_cast (sizeof (_ops->expirations)) && result != -EAGAIN && + result != -EINTR)) + { + // LCOV_EXCL_START + _state.store (State::Closed, std::memory_order_release); + return; + // LCOV_EXCL_STOP + } + + if (JOIN_UNLIKELY (_proactor.submit (&_ops->op) == -1)) + { + // LCOV_EXCL_START + _state.store (State::Closed, std::memory_order_release); + return; + // LCOV_EXCL_STOP + } + + if (JOIN_UNLIKELY (result != static_cast (sizeof (_ops->expirations)))) + { + return; // LCOV_EXCL_LINE + } + + State expected = State::Armed; + + if (JOIN_LIKELY (_state.compare_exchange_strong (expected, State::Invoking, std::memory_order_acq_rel, + std::memory_order_acquire))) { for (uint64_t i = 0; i < expirations; ++i) { _callback (); } + + expected = State::Invoking; + _state.compare_exchange_strong (expected, State::Armed, std::memory_order_acq_rel, + std::memory_order_acquire); } } + /** + * @brief method called when the timerfd read is cancelled. + * @param op cancelled operation. + * @param result negative errno. + */ + void onCancel ([[maybe_unused]] IoOperation* op, [[maybe_unused]] int result) override + { + _state.store (State::Closed, std::memory_order_release); + } + /** * @brief convert nsec to itimerspec. * @param ns value to convert. @@ -264,19 +365,54 @@ namespace join } /** - * @brief get native handle. - * @return native handle. + * @brief operation block, kept at its extended alignment. */ - int handle () const noexcept + struct Ops { - return _handle; - } + /** + * @brief allocate a block honouring its extended alignment. + * @param size allocation size in bytes. + * @return pointer to the allocated storage. + */ + static void* operator new (size_t size) + { + void* mem = ::aligned_alloc (alignof (Ops), size); + + if (mem == nullptr) + { + throw std::bad_alloc (); // LCOV_EXCL_LINE + } + + return mem; + } + + /** + * @brief release storage allocated by operator new. + * @param mem storage to release. + */ + static void operator delete (void* mem) noexcept + { + ::free (mem); + } + + /// timerfd read operation. + IoOperation op = {}; + + /// expiration count, written by the kernel until the read completes. + uint64_t expirations = 0; + }; /// ns per sec. static constexpr uint64_t _nsPerSec = 1000000000ULL; + /// operation block. + const std::unique_ptr _ops{new Ops ()}; + + /// timer state. + std::atomic _state{State::Idle}; + /// callback function - std::function _callback; + Function _callback; /// interval. std::chrono::nanoseconds _ns{}; @@ -287,8 +423,8 @@ namespace join /// timer handle. int _handle = -1; - /// event loop reactor. - Reactor& _reactor; + /// completion dispatcher. + Proactor& _proactor; }; } diff --git a/core/src/reactor.cpp b/core/src/reactor.cpp index 1fd42048..239991a3 100644 --- a/core/src/reactor.cpp +++ b/core/src/reactor.cpp @@ -217,15 +217,30 @@ void Reactor::stop (bool sync) noexcept return; } + if (JOIN_UNLIKELY (!isRunning ())) + { + return; + } + writeCommand ({CommandType::Stop, -1, 0, nullptr, nullptr, nullptr}); if (JOIN_LIKELY (sync)) { - Backoff backoff; - while (_threadId.load (std::memory_order_acquire) != _invalidThreadId) - { - backoff (); - } + waitStopped (); + } +} + +// ========================================================================= +// CLASS : Reactor +// METHOD : waitStopped +// ========================================================================= +void Reactor::waitStopped () const noexcept +{ + Backoff backoff; + + while (_threadId.load (std::memory_order_acquire) != _invalidThreadId) + { + backoff (); } } @@ -255,7 +270,7 @@ int Reactor::mlock () const noexcept // ========================================================================= bool Reactor::isRunning () const noexcept { - return _running.load (std::memory_order_acquire); + return _threadId.load (std::memory_order_acquire) != _invalidThreadId; } // ========================================================================= @@ -417,14 +432,9 @@ void Reactor::processCommand (const Command& cmd) noexcept void Reactor::readCommands () noexcept { uint64_t count; - ssize_t nread = ::read (_wakeup, &count, sizeof (count)); + [[maybe_unused]] ssize_t nread = ::read (_wakeup, &count, sizeof (count)); _notified.store (false); - if (JOIN_UNLIKELY (nread == -1)) - { - return; // LCOV_EXCL_LINE - } - Command cmd; while (_commands.tryPop (cmd) == 0) { @@ -611,6 +621,12 @@ ReactorThread::ReactorThread () _dispatcher = Thread ([this] () { _reactor.run (); }); + + Backoff backoff; + while (!_reactor.isRunning ()) + { + backoff (); + } } // ========================================================================= diff --git a/core/tests/CMakeLists.txt b/core/tests/CMakeLists.txt index fa606e7f..1c00dcc0 100644 --- a/core/tests/CMakeLists.txt +++ b/core/tests/CMakeLists.txt @@ -27,6 +27,11 @@ target_link_libraries(variant.gtest ${JOIN_CORE} GTest::gtest_main) add_test(NAME variant.gtest COMMAND variant.gtest) install(TARGETS variant.gtest RUNTIME DESTINATION ${CMAKE_INSTALL_DATADIR}/${PROJECT_NAME}/test) +add_executable(backoff.gtest backoff_test.cpp) +target_link_libraries(backoff.gtest ${JOIN_CORE} GTest::gtest_main) +add_test(NAME backoff.gtest COMMAND backoff.gtest) +install(TARGETS backoff.gtest RUNTIME DESTINATION ${CMAKE_INSTALL_DATADIR}/${PROJECT_NAME}/test) + add_executable(reactor.gtest reactor_test.cpp) target_link_libraries(reactor.gtest ${JOIN_CORE} GTest::gtest_main) add_test(NAME reactor.gtest COMMAND reactor.gtest) diff --git a/core/tests/backoff_test.cpp b/core/tests/backoff_test.cpp new file mode 100644 index 00000000..ee7aa19d --- /dev/null +++ b/core/tests/backoff_test.cpp @@ -0,0 +1,64 @@ +/** + * MIT License + * + * Copyright (c) 2026 Mathieu Rabine + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +// libjoin. +#include + +// Libraries. +#include + +using join::Backoff; + +/** + * @brief test spinExhausted. + */ +TEST (Backoff, spinExhausted) +{ + Backoff backoff (4); + ASSERT_FALSE (backoff.spinExhausted ()); + + for (size_t i = 0; i < 4; ++i) + { + backoff (); + } + ASSERT_TRUE (backoff.spinExhausted ()); + + backoff (); + ASSERT_TRUE (backoff.spinExhausted ()); + + backoff.reset (); + ASSERT_FALSE (backoff.spinExhausted ()); + + Backoff immediate (0); + ASSERT_TRUE (immediate.spinExhausted ()); +} + +/** + * @brief main function. + */ +int main (int argc, char** argv) +{ + testing::InitGoogleTest (&argc, argv); + return RUN_ALL_TESTS (); +} diff --git a/core/tests/hybrid_proactor_test.cpp b/core/tests/hybrid_proactor_test.cpp index 30b1201f..1df8981a 100644 --- a/core/tests/hybrid_proactor_test.cpp +++ b/core/tests/hybrid_proactor_test.cpp @@ -203,6 +203,53 @@ TEST_F (HybridProactorTest, stop) _op = nullptr; _result = 0; } + + for (int i = 0; i < 32; ++i) + { + HybridProactor concurrent; + Thread loop ([&concurrent] () { + concurrent.run (); + }); + while (!concurrent.isRunning ()) + { + } + + std::atomic go{false}; + auto stopper = [&concurrent, &go] () { + while (!go.load (std::memory_order_acquire)) + { + } + concurrent.stop (); + }; + + Thread first (stopper); + Thread second (stopper); + Thread third (stopper); + Thread fourth (stopper); + + go.store (true, std::memory_order_release); + + first.join (); + second.join (); + third.join (); + fourth.join (); + + ASSERT_FALSE (concurrent.isRunning ()); + + loop.join (); + } + + Thread orphan; + { + HybridProactor dying; + orphan = Thread ([&dying] () { + dying.run (); + }); + while (!dying.isRunning ()) + { + } + } + orphan.join (); } /** @@ -446,6 +493,30 @@ TEST_F (HybridProactorTest, isRunning) ASSERT_FALSE (proactor.isRunning ()); } +/** + * @brief Test waitStopped. + */ +TEST_F (HybridProactorTest, waitStopped) +{ + HybridProactor proactor; + + proactor.waitStopped (); + + Thread th ([&proactor] () { + proactor.run (); + }); + while (!proactor.isRunning ()) + { + } + + proactor.stop (false); + proactor.waitStopped (); + + ASSERT_FALSE (proactor.isRunning ()); + + th.join (); +} + #ifdef JOIN_HAS_IO_URING /** * @brief Test registerBuffers and unregisterBuffers. diff --git a/core/tests/monotonic_timer_test.cpp b/core/tests/monotonic_timer_test.cpp index 08f52ac3..5ff0ffc6 100644 --- a/core/tests/monotonic_timer_test.cpp +++ b/core/tests/monotonic_timer_test.cpp @@ -31,6 +31,7 @@ // C++. #include #include +#include // C. #include @@ -47,19 +48,19 @@ TEST (MonotonicTimer, setOneShot) Monotonic::Timer timer; int count = 0; - timer.setOneShot (10ms, [&] { + timer.setOneShot (50ms, [&] { ++count; }); - std::this_thread::sleep_for (35ms); + std::this_thread::sleep_for (250ms); EXPECT_EQ (count, 1); EXPECT_FALSE (timer.active ()); EXPECT_TRUE (timer.oneShot ()); EXPECT_EQ (timer.interval (), 0ms); - timer.setOneShot (std::chrono::steady_clock::now () + 10ms, [&] { + timer.setOneShot (std::chrono::steady_clock::now () + 50ms, [&] { ++count; }); - std::this_thread::sleep_for (35ms); + std::this_thread::sleep_for (250ms); EXPECT_EQ (count, 2); EXPECT_FALSE (timer.active ()); EXPECT_TRUE (timer.oneShot ()); @@ -68,7 +69,7 @@ TEST (MonotonicTimer, setOneShot) timer.setOneShot (0ms, [&] { ++count; }); - std::this_thread::sleep_for (35ms); + std::this_thread::sleep_for (250ms); EXPECT_EQ (count, 3); EXPECT_FALSE (timer.active ()); EXPECT_TRUE (timer.oneShot ()); @@ -83,25 +84,26 @@ TEST (MonotonicTimer, setInterval) Monotonic::Timer timer; int count = 0; - timer.setInterval (10ms, [&] { + timer.setInterval (50ms, [&] { ++count; }); - std::this_thread::sleep_for (35ms); + std::this_thread::sleep_for (250ms); EXPECT_GT (count, 1); EXPECT_TRUE (timer.active ()); EXPECT_FALSE (timer.oneShot ()); - EXPECT_EQ (timer.interval (), 10ms); + EXPECT_EQ (timer.interval (), 50ms); timer.cancel (); + Monotonic::Timer once; int fired = 0; - timer.setInterval (0ms, [&] { + once.setInterval (0ms, [&] { ++fired; }); - std::this_thread::sleep_for (35ms); + std::this_thread::sleep_for (250ms); EXPECT_EQ (fired, 1); - EXPECT_FALSE (timer.active ()); + EXPECT_FALSE (once.active ()); } /** @@ -112,15 +114,39 @@ TEST (MonotonicTimer, cancel) Monotonic::Timer timer; int count1 = 0, count2 = 0; - timer.setInterval (10ms, [&] { + timer.setInterval (50ms, [&] { count1++; }); - std::this_thread::sleep_for (35ms); + std::this_thread::sleep_for (250ms); timer.cancel (); count2 = count1; EXPECT_GT (count2, 1); - std::this_thread::sleep_for (35ms); + std::this_thread::sleep_for (250ms); EXPECT_EQ (count1, count2); + + Monotonic::Timer slow; + std::atomic ran{0}; + + slow.setInterval (50ms, [&ran] { + ran++; + std::this_thread::sleep_for (300ms); + }); + while (ran.load () == 0) + { + } + slow.cancel (); + EXPECT_GE (ran.load (), 1); + + Monotonic::Timer suicidal; + std::atomic fired{0}; + + suicidal.setInterval (50ms, [&suicidal, &fired] { + fired++; + suicidal.cancel (); + }); + std::this_thread::sleep_for (250ms); + EXPECT_GE (fired.load (), 1); + EXPECT_FALSE (suicidal.active ()); } /** @@ -132,7 +158,7 @@ TEST (MonotonicTimer, active) int count = 0; ASSERT_FALSE (timer.active ()); - timer.setInterval (10ms, [&] { + timer.setInterval (50ms, [&] { ++count; }); ASSERT_TRUE (timer.active ()); @@ -147,25 +173,25 @@ TEST (MonotonicTimer, remaining) { Monotonic::Timer timer; - timer.setOneShot (20ms, [] { + timer.setOneShot (200ms, [] { }); auto t1 = timer.remaining (); - std::this_thread::sleep_for (15ms); + std::this_thread::sleep_for (60ms); auto t2 = timer.remaining (); EXPECT_GT (t2.count (), 0); EXPECT_LT (t2.count (), t1.count ()); - std::this_thread::sleep_for (15ms); + std::this_thread::sleep_for (200ms); auto t3 = timer.remaining (); EXPECT_EQ (t3.count (), 0); // remaining time is zero - timer.setInterval (20ms, [] { + timer.setInterval (200ms, [] { }); t1 = timer.remaining (); - std::this_thread::sleep_for (15ms); + std::this_thread::sleep_for (60ms); t2 = timer.remaining (); EXPECT_GT (t2.count (), 0); EXPECT_LT (t2.count (), t1.count ()); - std::this_thread::sleep_for (15ms); + std::this_thread::sleep_for (200ms); t3 = timer.remaining (); EXPECT_GT (t3.count (), 0); // next interval has started } @@ -179,10 +205,10 @@ TEST (MonotonicTimer, interval) int count = 0; ASSERT_EQ (timer.interval (), 0ms); - timer.setInterval (10ms, [&] { + timer.setInterval (50ms, [&] { ++count; }); - ASSERT_EQ (timer.interval (), 10ms); + ASSERT_EQ (timer.interval (), 50ms); timer.cancel (); ASSERT_EQ (timer.interval (), 0ms); } @@ -196,7 +222,7 @@ TEST (MonotonicTimer, oneShot) int count = 0; ASSERT_TRUE (timer.oneShot ()); - timer.setInterval (10ms, [&] { + timer.setInterval (50ms, [&] { ++count; }); ASSERT_FALSE (timer.oneShot ()); diff --git a/core/tests/proactor_test.cpp b/core/tests/proactor_test.cpp index 2435338d..5f6610e5 100644 --- a/core/tests/proactor_test.cpp +++ b/core/tests/proactor_test.cpp @@ -207,6 +207,53 @@ TEST_F (ProactorTest, stop) _op = nullptr; _result = 0; } + + for (int i = 0; i < 32; ++i) + { + Proactor concurrent; + Thread loop ([&concurrent] () { + concurrent.run (); + }); + while (!concurrent.isRunning ()) + { + } + + std::atomic go{false}; + auto stopper = [&concurrent, &go] () { + while (!go.load (std::memory_order_acquire)) + { + } + concurrent.stop (); + }; + + Thread first (stopper); + Thread second (stopper); + Thread third (stopper); + Thread fourth (stopper); + + go.store (true, std::memory_order_release); + + first.join (); + second.join (); + third.join (); + fourth.join (); + + ASSERT_FALSE (concurrent.isRunning ()); + + loop.join (); + } + + Thread orphan; + { + Proactor dying; + orphan = Thread ([&dying] () { + dying.run (); + }); + while (!dying.isRunning ()) + { + } + } + orphan.join (); } /** @@ -452,6 +499,30 @@ TEST_F (ProactorTest, isRunning) ASSERT_FALSE (proactor.isRunning ()); } +/** + * @brief Test waitStopped. + */ +TEST_F (ProactorTest, waitStopped) +{ + Proactor proactor; + + proactor.waitStopped (); + + Thread th ([&proactor] () { + proactor.run (); + }); + while (!proactor.isRunning ()) + { + } + + proactor.stop (false); + proactor.waitStopped (); + + ASSERT_FALSE (proactor.isRunning ()); + + th.join (); +} + #ifdef JOIN_HAS_IO_URING /** * @brief Test registerBuffers and unregisterBuffers. diff --git a/core/tests/reactor_test.cpp b/core/tests/reactor_test.cpp index 477b0ba3..768a254b 100644 --- a/core/tests/reactor_test.cpp +++ b/core/tests/reactor_test.cpp @@ -337,6 +337,42 @@ TEST_F (ReactorTest, isRunning) ASSERT_FALSE (reactor.isRunning ()); } +/** + * @brief Test waitStopped. + */ +TEST_F (ReactorTest, waitStopped) +{ + Reactor reactor; + + reactor.waitStopped (); + + Thread th ([&reactor] () { + reactor.run (); + }); + while (!reactor.isRunning ()) + { + } + + reactor.stop (false); + reactor.waitStopped (); + + ASSERT_FALSE (reactor.isRunning ()); + + th.join (); + + Thread orphan; + { + Reactor dying; + orphan = Thread ([&dying] () { + dying.run (); + }); + while (!dying.isRunning ()) + { + } + } + orphan.join (); +} + /** * @brief Test onReadable. */ diff --git a/core/tests/realtime_timer_test.cpp b/core/tests/realtime_timer_test.cpp index 87804528..827e6359 100644 --- a/core/tests/realtime_timer_test.cpp +++ b/core/tests/realtime_timer_test.cpp @@ -31,6 +31,7 @@ // C++. #include #include +#include // C. #include @@ -47,19 +48,19 @@ TEST (RealTimer, setOneShot) RealTime::Timer timer; int count = 0; - timer.setOneShot (10ms, [&] { + timer.setOneShot (50ms, [&] { ++count; }); - std::this_thread::sleep_for (35ms); + std::this_thread::sleep_for (250ms); EXPECT_EQ (count, 1); EXPECT_FALSE (timer.active ()); EXPECT_TRUE (timer.oneShot ()); EXPECT_EQ (timer.interval (), 0ms); - timer.setOneShot (std::chrono::system_clock::now () + 10ms, [&] { + timer.setOneShot (std::chrono::system_clock::now () + 50ms, [&] { ++count; }); - std::this_thread::sleep_for (35ms); + std::this_thread::sleep_for (250ms); EXPECT_EQ (count, 2); EXPECT_FALSE (timer.active ()); EXPECT_TRUE (timer.oneShot ()); @@ -68,7 +69,7 @@ TEST (RealTimer, setOneShot) timer.setOneShot (0ms, [&] { ++count; }); - std::this_thread::sleep_for (35ms); + std::this_thread::sleep_for (250ms); EXPECT_EQ (count, 3); EXPECT_FALSE (timer.active ()); EXPECT_TRUE (timer.oneShot ()); @@ -83,25 +84,26 @@ TEST (RealTimer, setInterval) RealTime::Timer timer; int count = 0; - timer.setInterval (10ms, [&] { + timer.setInterval (50ms, [&] { ++count; }); - std::this_thread::sleep_for (35ms); + std::this_thread::sleep_for (250ms); EXPECT_GT (count, 1); EXPECT_TRUE (timer.active ()); EXPECT_FALSE (timer.oneShot ()); - EXPECT_EQ (timer.interval (), 10ms); + EXPECT_EQ (timer.interval (), 50ms); timer.cancel (); + RealTime::Timer once; int fired = 0; - timer.setInterval (0ms, [&] { + once.setInterval (0ms, [&] { ++fired; }); - std::this_thread::sleep_for (35ms); + std::this_thread::sleep_for (250ms); EXPECT_EQ (fired, 1); - EXPECT_FALSE (timer.active ()); + EXPECT_FALSE (once.active ()); } /** @@ -112,15 +114,39 @@ TEST (RealTimer, cancel) RealTime::Timer timer; int count1 = 0, count2 = 0; - timer.setInterval (10ms, [&] { + timer.setInterval (50ms, [&] { count1++; }); - std::this_thread::sleep_for (35ms); + std::this_thread::sleep_for (250ms); timer.cancel (); count2 = count1; EXPECT_GT (count2, 1); - std::this_thread::sleep_for (35ms); + std::this_thread::sleep_for (250ms); EXPECT_EQ (count1, count2); + + RealTime::Timer slow; + std::atomic ran{0}; + + slow.setInterval (50ms, [&ran] { + ran++; + std::this_thread::sleep_for (300ms); + }); + while (ran.load () == 0) + { + } + slow.cancel (); + EXPECT_GE (ran.load (), 1); + + RealTime::Timer suicidal; + std::atomic fired{0}; + + suicidal.setInterval (50ms, [&suicidal, &fired] { + fired++; + suicidal.cancel (); + }); + std::this_thread::sleep_for (250ms); + EXPECT_GE (fired.load (), 1); + EXPECT_FALSE (suicidal.active ()); } /** @@ -132,7 +158,7 @@ TEST (RealTimer, active) int count = 0; ASSERT_FALSE (timer.active ()); - timer.setInterval (10ms, [&] { + timer.setInterval (50ms, [&] { ++count; }); ASSERT_TRUE (timer.active ()); @@ -147,25 +173,25 @@ TEST (RealTimer, remaining) { RealTime::Timer timer; - timer.setOneShot (20ms, [] { + timer.setOneShot (200ms, [] { }); auto t1 = timer.remaining (); - std::this_thread::sleep_for (15ms); + std::this_thread::sleep_for (60ms); auto t2 = timer.remaining (); EXPECT_GT (t2.count (), 0); EXPECT_LT (t2.count (), t1.count ()); - std::this_thread::sleep_for (15ms); + std::this_thread::sleep_for (200ms); auto t3 = timer.remaining (); EXPECT_EQ (t3.count (), 0); // remaining time is zero - timer.setInterval (20ms, [] { + timer.setInterval (200ms, [] { }); t1 = timer.remaining (); - std::this_thread::sleep_for (15ms); + std::this_thread::sleep_for (60ms); t2 = timer.remaining (); EXPECT_GT (t2.count (), 0); EXPECT_LT (t2.count (), t1.count ()); - std::this_thread::sleep_for (15ms); + std::this_thread::sleep_for (200ms); t3 = timer.remaining (); EXPECT_GT (t3.count (), 0); // next interval has started } @@ -179,10 +205,10 @@ TEST (RealTimer, interval) int count = 0; ASSERT_EQ (timer.interval (), 0ms); - timer.setInterval (10ms, [&] { + timer.setInterval (50ms, [&] { ++count; }); - ASSERT_EQ (timer.interval (), 10ms); + ASSERT_EQ (timer.interval (), 50ms); timer.cancel (); ASSERT_EQ (timer.interval (), 0ms); } @@ -196,7 +222,7 @@ TEST (RealTimer, oneShot) int count = 0; ASSERT_TRUE (timer.oneShot ()); - timer.setInterval (10ms, [&] { + timer.setInterval (50ms, [&] { ++count; }); ASSERT_FALSE (timer.oneShot ()); diff --git a/core/tests/sqpoll_proactor_test.cpp b/core/tests/sqpoll_proactor_test.cpp index 64bbbbb8..4e5c98c0 100644 --- a/core/tests/sqpoll_proactor_test.cpp +++ b/core/tests/sqpoll_proactor_test.cpp @@ -203,6 +203,53 @@ TEST_F (SqpollProactorTest, stop) _op = nullptr; _result = 0; } + + for (int i = 0; i < 32; ++i) + { + SqpollProactor concurrent; + Thread loop ([&concurrent] () { + concurrent.run (); + }); + while (!concurrent.isRunning ()) + { + } + + std::atomic go{false}; + auto stopper = [&concurrent, &go] () { + while (!go.load (std::memory_order_acquire)) + { + } + concurrent.stop (); + }; + + Thread first (stopper); + Thread second (stopper); + Thread third (stopper); + Thread fourth (stopper); + + go.store (true, std::memory_order_release); + + first.join (); + second.join (); + third.join (); + fourth.join (); + + ASSERT_FALSE (concurrent.isRunning ()); + + loop.join (); + } + + Thread orphan; + { + SqpollProactor dying; + orphan = Thread ([&dying] () { + dying.run (); + }); + while (!dying.isRunning ()) + { + } + } + orphan.join (); } /** @@ -446,6 +493,30 @@ TEST_F (SqpollProactorTest, isRunning) ASSERT_FALSE (proactor.isRunning ()); } +/** + * @brief Test waitStopped. + */ +TEST_F (SqpollProactorTest, waitStopped) +{ + SqpollProactor proactor; + + proactor.waitStopped (); + + Thread th ([&proactor] () { + proactor.run (); + }); + while (!proactor.isRunning ()) + { + } + + proactor.stop (false); + proactor.waitStopped (); + + ASSERT_FALSE (proactor.isRunning ()); + + th.join (); +} + #ifdef JOIN_HAS_IO_URING /** * @brief Test registerBuffers and unregisterBuffers.