feat: Added timers to asynchronous contexts.

Author: kcexn

.clang-tidy

@@ -6,6 +6,7 @@ Checks: >
   -cppcoreguidelines-narrowing-conversions,
   modernize-*,
   performance-*,
+  -performance-unnecessary-value-param,
   readability-*,
   -readability-braces-around-statements,
   -readability-magic-numbers,

CMakePresets.json

@@ -47,10 +47,7 @@
             "name": "debug",
             "description": "tests",
             "displayName": "Debug",
-            "configurePreset": "debug",
-            "execution": {
-                "jobs": 1
-            }
+            "configurePreset": "debug"
         }
     ],
     "packagePresets": [

include/net/detail/concepts.hpp

@@ -31,10 +31,22 @@ struct async_context;
  * @brief The root namespace for all cppnet components.
  */
 namespace net {
+/** @brief ServiceLike describes types that behave like an application or
+ * service. */
 template <typename S>
 concept ServiceLike = requires(S service, service::async_context &ctx) {
   { service.signal_handler(1) } noexcept -> std::same_as<void>;
   { service.start(ctx) } noexcept -> std::same_as<void>;
 };
+
+/** @brief This namespace is for timers and interrupts. */
+namespace timers {
+/** @brief A concept for constraining interrupt sources. */
+template <typename Tag>
+concept InterruptSource = requires(const Tag tag) {
+  { tag.interrupt() } noexcept -> std::same_as<void>;
+};
+} // namespace timers.
+
 } // namespace net
 #endif // CPPNET_CONCEPT_HPP

include/net/service/context_thread.hpp

@@ -22,13 +22,12 @@
 #define CPPNET_CONTEXT_THREAD_HPP
 #include "net/detail/concepts.hpp"
 #include "net/detail/immovable.hpp"
-#include "net/timers/interrupt.hpp"
+#include "net/timers/timers.hpp"
 
 #include <exec/async_scope.hpp>
 #include <io/io.hpp>
 
 #include <atomic>
-#include <condition_variable>
 #include <cstdint>
 #include <mutex>
 #include <thread>
@@ -48,8 +47,8 @@ struct async_context : detail::immovable {
   using signal_mask = std::uint64_t;
   /** @brief Interrupt source type. */
   using interrupt_source = timers::socketpair_interrupt_source_t;
-  /** @brief The interrupt type. */
-  using interrupt_type = timers::interrupt<interrupt_source>;
+  /** @brief The timers type. */
+  using timers_type = timers::timers<interrupt_source>;
 
   /** @brief An enum of all valid async context signals. */
   enum signals : std::uint8_t { terminate = 0, user1, END };
@@ -64,15 +63,18 @@ struct async_context : detail::immovable {
   std::atomic<context_states> state{PENDING};
   /** @brief The active signal mask. */
   std::atomic<signal_mask> sigmask;
-  /** @brief The event loop interrupt. */
-  interrupt_type interrupt;
+  /** @brief The event loop timers. */
+  timers_type timers;
 
   /**
    * @brief Sets the signal mask, then interrupts the service.
    * @param signum The signal to send. Must be in range of
    *               enum signals.
    */
-  auto signal(int signum) -> void;
+  inline auto signal(int signum) -> void;
+
+  /** @brief Calls the timers interrupt. */
+  inline auto interrupt() const noexcept -> void;
 };
 
 /**
@@ -92,10 +94,6 @@ template <ServiceLike Service> class context_thread : public async_context {
   using clock = std::chrono::steady_clock;
   /** @brief The duration type. */
   using duration = std::chrono::milliseconds;
-
-  /** @brief Internal context loop interval.*/
-  static constexpr int INTERVAL_MS = 2000;
-
   /**
    * @brief An interrupt service routine.
    *
@@ -137,9 +135,7 @@ template <ServiceLike Service> class context_thread : public async_context {
    *             thread.
    * @param args The arguments to forward to the Service constructor.
    */
-  template <typename... Args>
-  auto start(std::mutex &mtx, std::condition_variable &cvar,
-             Args &&...args) -> void;
+  template <typename... Args> auto start(Args &&...args) -> void;
 
   /** @brief The destructor signals the thread before joining it. */
   ~context_thread();

include/net/service/impl/async_context_impl.hpp

@@ -31,5 +31,11 @@ inline auto async_context::signal(int signum) -> void
   interrupt();
 }
 
+/** @brief Calls the timers interrupt. */
+inline auto async_context::interrupt() const noexcept -> void
+{
+  static_cast<const timers_type::interrupt_source_t &>(timers).interrupt();
+}
+
 } // namespace net::service
 #endif // CPPNET_ASYNC_CONTEXT_IMPL_HPP

include/net/service/impl/context_thread_impl.hpp

@@ -21,7 +21,6 @@
 #pragma once
 #ifndef CPPNET_CONTEXT_THREAD_IMPL_HPP
 #define CPPNET_CONTEXT_THREAD_IMPL_HPP
-#include "net/detail/with_lock.hpp"
 #include "net/service/context_thread.hpp"
 
 #include <stdexec/execution.hpp>
@@ -42,10 +41,7 @@ auto context_thread<Service>::isr(async_scope &scope,
   static auto msg = socket_message{.buffers = buffer};
 
   if (!handle())
-  {
-    scope.request_stop();
     return;
-  }
 
   auto recvmsg =
       io::recvmsg(socket, msg, 0) |
@@ -57,55 +53,67 @@ auto context_thread<Service>::isr(async_scope &scope,
 template <ServiceLike Service>
 auto context_thread<Service>::stop() noexcept -> void
 {
-  auto socket = interrupt.sockets[1];
-  interrupt.sockets[1] = interrupt.INVALID_SOCKET;
-  if (socket != interrupt.INVALID_SOCKET)
+  auto socket = timers.sockets[1];
+  timers.sockets[1] = timers.INVALID_SOCKET;
+  if (socket != timers.INVALID_SOCKET)
     io::socket::close(socket);
   state = STOPPED;
 }
 
 template <ServiceLike Service>
 template <typename... Args>
-auto context_thread<Service>::start(std::mutex &mtx,
-                                    std::condition_variable &cvar,
-                                    Args &&...args) -> void
+auto context_thread<Service>::start(Args &&...args) -> void
 {
-  auto lock = std::lock_guard{mtx};
+  auto lock = std::lock_guard{mtx_};
   if (started_)
     throw std::invalid_argument("context_thread can't be started twice.");
 
   server_ = std::thread([&]() noexcept {
     using namespace detail;
     using namespace io::socket;
+    using namespace std::chrono;
 
     auto service = Service{std::forward<Args>(args)...};
-    auto &sockets = interrupt.sockets;
+    auto &sockets = timers.sockets;
     if (!socketpair(AF_UNIX, SOCK_STREAM, 0, sockets.data()))
     {
+      const auto token = scope.get_stop_token();
+
       isr(scope, poller.emplace(sockets[0]), [&]() noexcept {
         auto sigmask_ = sigmask.exchange(0);
         for (int signum = 0; auto mask = (sigmask_ >> signum); ++signum)
         {
           if (mask & (1 << 0))
             service.signal_handler(signum);
         }
-        return !(sigmask_ & (1 << terminate));
-      });
 
-      state = STARTED;
-      cvar.notify_all();
+        if (sigmask_ & (1 << terminate))
+        {
+          scope.request_stop();
+          timers.add(
+              seconds(1),
+              [&](timers::timer_id) { service.signal_handler(terminate); },
+              seconds(1));
+        }
+
+        return !token.stop_requested();
+      });
 
       service.start(static_cast<async_context &>(*this));
+      state = STARTED;
 
-      const auto token = scope.get_stop_token();
       if (token.stop_requested())
+      {
+        state = STOPPED;
         signal(terminate);
+      }
 
+      state.notify_all();
       run(service, token);
     }
 
-    with_lock(std::unique_lock{mtx}, [&]() noexcept { stop(); });
-    cvar.notify_all();
+    stop();
+    state.notify_all();
   });
 
   started_ = true;
@@ -126,27 +134,21 @@ auto context_thread<Service>::run(Service &service,
                                   const StopToken &token) -> void
 {
   using namespace stdexec;
-  using std::chrono::duration_cast;
+  using namespace std::chrono;
 
-  int next = 0;
-  auto start = clock::now();
-  auto is_empty = false;
-  scope.spawn(poller.on_empty() | then([&]() noexcept { is_empty = true; }));
+  auto next = timers.resolve();
+  int wait_ms =
+      (next.count() < 0) ? next.count() : duration_cast<duration>(next).count();
 
-  while (poller.wait_for(next) || !is_empty)
-  {
-    const auto now = clock::now();
+  auto is_empty = std::atomic_flag();
+  scope.spawn(poller.on_empty() |
+              then([&]() noexcept { is_empty.test_and_set(); }));
 
-    next -= duration_cast<duration>(now - start).count();
-    if (next <= 0)
-    {
-      if (token.stop_requested())
-        service.signal_handler(terminate);
-
-      next = INTERVAL_MS;
-    }
-
-    start = now;
+  while (poller.wait_for(wait_ms) || !is_empty.test())
+  {
+    next = timers.resolve();
+    wait_ms = (next.count() < 0) ? next.count()
+                                 : duration_cast<duration>(next).count();
   }
 }
 } // namespace net::service

include/net/timers/impl/interrupt_impl.hpp

@@ -30,8 +30,7 @@ inline auto socketpair_interrupt_source_t::interrupt() const noexcept -> void
   static constexpr auto buf = std::array<char, 1>{'x'};
   static const auto msg = socket_message<sockaddr_in>{.buffers = buf};
 
-  if (sockets[1] != INVALID_SOCKET)
-    ::io::sendmsg(sockets[1], msg, 0);
+  ::io::sendmsg(sockets[1], msg, MSG_NOSIGNAL);
 }
 
 template <InterruptSource Interrupt>