Add proactor timer wheel

core/include/join/proactor.hpp

@@ -31,6 +31,7 @@
 #include <join/function.hpp>
 #ifdef JOIN_HAS_IO_URING
 #include <join/io_policy.hpp>
+#include <join/wheel.hpp>
 #else
 #include <join/reactor.hpp>
 #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 <class Wheel>
+        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 <class Wheel>
+        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 <class Wheel>
+        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 <class Wheel>
+        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<void (), 64>;
 
+#ifdef JOIN_HAS_IO_URING
+    /// timer wheel, able to arm a timer for every queued command.
+    using Wheel = BasicWheel<Monotonic, WheelPolicy, _queueSize, 1'000'000>;
+#else
+    /// timer wheel of the underlying reactor.
+    using Wheel = Reactor::Wheel;
+#endif
+
     /**
      * @brief initialize the proactor and its I/O backend.
      */
@@ -302,15 +408,15 @@ 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.
      */
     int mbind (int numa) const noexcept;
 #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<Command> _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

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

core/include/join/proactor_uring_impl.hpp

@@ -382,6 +382,11 @@ int join::BasicProactor<IoPolicy>::unregisterBufferRing (uint16_t group)
 template <typename IoPolicy>
 int join::BasicProactor<IoPolicy>::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<IoPolicy>::mbind (int numa) const noexcept
 template <typename IoPolicy>
 int join::BasicProactor<IoPolicy>::mlock () const noexcept
 {
+    if (_wheel.mlock () == -1)
+    {
+        return -1;  // LCOV_EXCL_LINE
+    }
+
     return _commands.mlock ();
 }
 
@@ -416,6 +426,16 @@ bool join::BasicProactor<IoPolicy>::isProactorThread () const noexcept
     return _threadId.load (std::memory_order_acquire) == pthread_self ();
 }
 
+// =========================================================================
+//   CLASS     : BasicProactor
+//   METHOD    : wheel
+// =========================================================================
+template <typename IoPolicy>
+typename join::BasicProactor<IoPolicy>::Wheel& join::BasicProactor<IoPolicy>::wheel () noexcept
+{
+    return _wheel;
+}
+
 // =========================================================================
 //   CLASS     : BasicProactor
 //   METHOD    : initWakeup
@@ -525,21 +545,31 @@ void join::BasicProactor<IoPolicy>::initSqThreadCpu (io_uring_params& params, st
 template <typename IoPolicy>
 int join::BasicProactor<IoPolicy>::writeCommand (const Command& cmd) noexcept
 {
-    return writeCommand (cmd, is_default<IoPolicy>{});
+    if (JOIN_UNLIKELY (_commands.push (cmd) == -1))
+    {
+        return -1;  // LCOV_EXCL_LINE
+    }
+
+    return wakeup ();
 }
 
 // =========================================================================
 //   CLASS     : BasicProactor
-//   METHOD    : writeCommand
+//   METHOD    : wakeup
 // =========================================================================
 template <typename IoPolicy>
-int join::BasicProactor<IoPolicy>::writeCommand (const Command& cmd, std::true_type) noexcept
+int join::BasicProactor<IoPolicy>::wakeup () noexcept
 {
-    if (JOIN_UNLIKELY (_commands.push (cmd) == -1))
-    {
-        return -1;  // LCOV_EXCL_LINE
-    }
+    return wakeup (is_default<IoPolicy>{});
+}
 
+// =========================================================================
+//   CLASS     : BasicProactor
+//   METHOD    : wakeup
+// =========================================================================
+template <typename IoPolicy>
+int join::BasicProactor<IoPolicy>::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<IoPolicy>::writeCommand (const Command& cmd, std::true_t
 
 // =========================================================================
 //   CLASS     : BasicProactor
-//   METHOD    : writeCommand
+//   METHOD    : wakeup
 // =========================================================================
 template <typename IoPolicy>
-int join::BasicProactor<IoPolicy>::writeCommand (const Command& cmd, std::false_type) noexcept
+int join::BasicProactor<IoPolicy>::wakeup (std::false_type) noexcept
 {
-    return _commands.push (cmd);
+    return 0;
 }
 
 // =========================================================================
@@ -1091,11 +1121,13 @@ void join::BasicProactor<IoPolicy>::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<IoPolicy>::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<IoPolicy>::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<IoPolicy>::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 ();