Use proactor backend in timers

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;

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<bool> _notified{false};
-
     /// command queue.
     LocalMem::Mpsc::Queue<Command> _commands;
 
-    /// set to true while a sync stop() is in progress.
-    std::atomic<bool> _stopping{false};
+    /// event loop wakeup state.
+#ifdef JOIN_HAS_IO_URING
+    alignas (64) std::atomic<WakeupState> _wakeupState{WakeupState::Polling};
+#else
+    alignas (64) std::atomic<WakeupState> _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 ();
+        }
     }
 
     /**

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<bool> 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);
 }
 
 // =========================================================================