Remove thread_local Reactor and pass it around everywhere instead

Author: Dmitry Malakhov

.clang-tidy

@@ -10,6 +10,7 @@ Checks: >
     portability-*,
     readability-*,
     -readability-identifier-length,
+    -readability-uppercase-literal-suffix,
     -readability-static-accessed-through-instance,
     -readability-make-member-function-const,
     -readability-avoid-return-with-void-value,
@@ -21,6 +22,7 @@ Checks: >
 WarningsAsErrors: "*"
 
 CheckOptions:
+  - { key: readability-function-cognitive-complexity.IgnoreMacros, value: true       }
   - { key: readability-identifier-naming.GlobalConstantCase,       value: UPPER_CASE }
   - { key: readability-identifier-naming.ConstexprVariableCase,    value: UPPER_CASE }
   - { key: readability-identifier-naming.LocalConstantCase,        value: lower_case }

example/SendLogsOnCrash.cpp

@@ -1,11 +1,13 @@
 /// Use case: there is an application, producing logs. Logs are initially stored
 /// in some in-memory buffer. Periodically, a flush operation is triggered and all
 /// logs are written to various outputs: to remote udp server and to file. But what if
-/// an app receives malicious signal (such as SIGTERM when std::terminate is called or
+/// an app receives malicious signal (such as SIGABRT when std::terminate is called or
 /// SIGFPE if current c++ implementation raises it on zero division) then in-buffer
 /// logs are lost by default. To prevent this we have to write custom signal handler.
 /// And to speed this sighandler up we will use corosig-powered async io
 
+#include "corosig/reactor/Reactor.hpp"
+
 #include <boost/outcome/try.hpp>
 #include <corosig/Coro.hpp>
 #include <corosig/ErrorTypes.hpp>
@@ -37,33 +39,37 @@ constexpr std::string_view FILE2 = "file2.log";
 
 std::vector<std::string> logs_buffer;
 
-corosig::Fut<void, corosig::Error<corosig::AllocationError, corosig::SyscallError>>
-sighandler(int) noexcept {
-  using namespace corosig;
+namespace sighandling {
 
-  auto write_to_file = [](char const *path) -> Fut<void, Error<AllocationError, SyscallError>> {
-    using enum File::OpenFlags;
-    BOOST_OUTCOME_CO_TRY(auto file, co_await File::open(path, CREATE | TRUNCATE | WRONLY));
-    for (auto &log : logs_buffer) {
-      BOOST_OUTCOME_CO_TRY(co_await file.write(log));
-    }
-    co_return success();
-  };
+using namespace corosig;
 
-  auto send_via_tcp = []() -> Fut<void, Error<AllocationError, SyscallError>> {
-    BOOST_OUTCOME_CO_TRY(auto socket, co_await TcpSocket::connect(SERVER_ADDR));
-    for (auto &log : logs_buffer) {
-      BOOST_OUTCOME_CO_TRY(co_await socket.write(log));
-    }
-    co_return success();
-  };
+Fut<void, Error<AllocationError, SyscallError>> write_to_file(Reactor &r,
+                                                              char const *path) noexcept {
+  using enum File::OpenFlags;
+  BOOST_OUTCOME_CO_TRY(auto file, co_await File::open(r, path, CREATE | TRUNCATE | WRONLY));
+  for (auto &log : logs_buffer) {
+    BOOST_OUTCOME_CO_TRY(co_await file.write(r, log));
+  }
+  co_return success();
+};
 
-  BOOST_OUTCOME_CO_TRY(co_await when_all_succeed(write_to_file(FILE1.data()),
-                                                 write_to_file(FILE2.data()), send_via_tcp()));
+Fut<void, Error<AllocationError, SyscallError>> send_via_tcp(Reactor &r) {
+  BOOST_OUTCOME_CO_TRY(auto socket, co_await TcpSocket::connect(r, SERVER_ADDR));
+  for (auto &log : logs_buffer) {
+    BOOST_OUTCOME_CO_TRY(co_await socket.write(r, log));
+  }
+  co_return success();
+};
 
+Fut<void, Error<AllocationError, SyscallError>> sighandler(Reactor &r, int) noexcept {
+
+  BOOST_OUTCOME_CO_TRY(co_await when_all_succeed(r, write_to_file(r, FILE1.data()),
+                                                 write_to_file(r, FILE2.data()), send_via_tcp(r)));
   co_return success();
 }
 
+} // namespace sighandling
+
 void run_tcp_server(std::string &out) {
   int srv_fd = ::socket(AF_INET, SOCK_STREAM, 0);
   assert(srv_fd >= 0);
@@ -79,13 +85,13 @@ void run_tcp_server(std::string &out) {
   ::listen(srv_fd, 1);
 
   int client = ::accept(srv_fd, nullptr, nullptr);
-  char buf[1024]; // NOLINT
+  std::array<char, 1024> buf;
   while (true) {
-    ssize_t n = ::read(client, buf, sizeof(buf));
+    ssize_t n = ::read(client, buf.begin(), buf.size());
     if (n <= 0) {
       break;
     }
-    out += std::string_view{buf, size_t(n)};
+    out += std::string_view{buf.begin(), size_t(n)};
   }
   ::close(client);
   ::close(srv_fd);
@@ -97,7 +103,7 @@ int main() {
   try {
     constexpr auto REACTOR_MEMORY = 8 * 1024;
     for (auto signal : {SIGILL, SIGFPE, SIGTERM, SIGABRT}) {
-      corosig::set_sighandler<REACTOR_MEMORY, sighandler>(signal);
+      corosig::set_sighandler<REACTOR_MEMORY, sighandling::sighandler>(signal);
     }
 
     std::string remote_server_data;

include/corosig/Alloc.hpp include/corosig/Allocator.hppinclude/corosig/Alloc.hpp renamed to include/corosig/Allocator.hpp

@@ -7,7 +7,7 @@
 
 namespace corosig {
 
-struct Alloc {
+struct Allocator {
 private:
   struct FreeHeader {
     size_t block_size = 0;
@@ -46,21 +46,21 @@ struct Alloc {
 public:
   constexpr static size_t MIN_ALIGNMENT = 8;
 
-  Alloc() noexcept = default;
+  Allocator() noexcept = default;
 
   template <size_t SIZE>
-  Alloc(Memory<SIZE> &mem) noexcept : m_mem{mem.begin()}, m_mem_size{mem.size()} {
+  Allocator(Memory<SIZE> &mem) noexcept : m_mem{mem.begin()}, m_mem_size{mem.size()} {
     Node *first_node = new (m_mem) Node{};
     first_node->data.block_size = SIZE - sizeof(Node);
     first_node->next = nullptr;
     m_free_list.insert(nullptr, first_node);
   }
 
-  Alloc(const Alloc &) = delete;
-  Alloc(Alloc &&) noexcept;
-  Alloc &operator=(const Alloc &) = delete;
-  Alloc &operator=(Alloc &&) noexcept;
-  ~Alloc();
+  Allocator(const Allocator &) = delete;
+  Allocator(Allocator &&) noexcept;
+  Allocator &operator=(const Allocator &) = delete;
+  Allocator &operator=(Allocator &&) noexcept;
+  ~Allocator();
 
   [[nodiscard]] size_t peak_memory() const noexcept {
     return m_peak;

include/corosig/Coro.hpp

@@ -10,7 +10,6 @@
 #include <concepts>
 #include <coroutine>
 #include <cstddef>
-#include <exception>
 #include <optional>
 #include <utility>
 
@@ -21,39 +20,53 @@ struct Fut;
 
 namespace detail {
 
+template <typename T>
+concept NotReactor = !std::same_as<Reactor, T>;
+
 template <typename T, typename E>
 struct CoroutinePromiseType : CoroListNode {
-  CoroutinePromiseType() noexcept = default;
+
+  CoroutinePromiseType(Reactor &reactor, NotReactor auto const &...) noexcept
+      // reuse m_out to pass reactor to future to avoid creating additional buffer
+      : m_reactor{reactor} {
+  }
+
+  CoroutinePromiseType(NotReactor auto const &, Reactor &reactor,
+                       NotReactor auto const &...) noexcept
+      : CoroutinePromiseType{reactor} {
+  }
 
   CoroutinePromiseType(const CoroutinePromiseType &) = delete;
   CoroutinePromiseType(CoroutinePromiseType &&) = delete;
   CoroutinePromiseType &operator=(const CoroutinePromiseType &) = delete;
   CoroutinePromiseType &operator=(CoroutinePromiseType &&) = delete;
 
-  ~CoroutinePromiseType() {
-    assert(m_out);
-    m_out->m_promise = nullptr;
+  ~CoroutinePromiseType() override = default;
+
+  static void *operator new(size_t n, Reactor &reactor, NotReactor auto const &...) noexcept {
+    return reactor.allocate(n);
   }
 
-  template <typename... ARGS>
-  static void *operator new(size_t n, ARGS &&...) noexcept {
-    return Reactor::instance().allocate_frame(n);
+  static void *operator new(size_t n, NotReactor auto const &, Reactor &reactor,
+                            NotReactor auto const &...) noexcept {
+    return reactor.allocate(n);
   }
 
-  static void operator delete(void *frame) noexcept {
-    return Reactor::instance().free_frame(frame);
+  static void operator delete(void *) noexcept {
+    // nothing to do in here since reactor is not accessible. instead, a coro frame is released when
+    // future is destroyed
   }
 
   void yield_to_reactor() noexcept {
-    Reactor::instance().yield(*this);
+    m_reactor.yield(*this);
   }
 
   void poll_to_reactor(PollListNode &node) noexcept {
-    Reactor::instance().poll(node);
+    m_reactor.poll(node);
   }
 
   [[noreturn]] static void unhandled_exception() noexcept {
-    std::terminate();
+    std::abort();
   }
 
   Fut<T, E> get_return_object() noexcept;
@@ -64,21 +77,37 @@ struct CoroutinePromiseType : CoroListNode {
   }
 
   auto final_suspend() noexcept {
-    return std::suspend_never{};
+    struct ReturnControlToCaller {
+      static bool await_ready() noexcept {
+        return false;
+      }
+
+      static auto await_suspend(std::coroutine_handle<CoroutinePromiseType> self) noexcept {
+        return self.promise().m_waiting_coro;
+      }
+
+      static void await_resume() noexcept {
+      }
+    };
+
+    // coro frame must be destroyed in Fut dtor so we don't let it fall off the end here
+    return ReturnControlToCaller{};
   }
 
   template <std::convertible_to<Result<T, E>> U>
   void return_value(U &&value) noexcept {
     // NOLINTBEGIN false positives about m_out being uninitialized
     assert(m_out);
+    assert(!m_out->m_value);
+    assert(!m_waiting_coro.done());
     m_out->m_value.emplace(std::forward<U>(value));
-    m_waiting_coro.resume();
     // NOLINTEND
   }
 
   template <std::convertible_to<T> T2, std::convertible_to<E> E2>
   void return_value(Result<T2, E2> &&result) noexcept {
     assert(m_out);
+    assert(!m_out->m_value);
     if (result.has_value()) {
       if constexpr (std::same_as<void, T>) {
         m_out->m_value.emplace(success());
@@ -88,7 +117,8 @@ struct CoroutinePromiseType : CoroListNode {
     } else {
       m_out->m_value.emplace(std::move(result.assume_error()));
     }
-    m_waiting_coro.resume();
+
+    assert(!m_waiting_coro.done() && "Waiting coro was destroyed before child has finished");
   }
 
 private:
@@ -99,6 +129,7 @@ struct CoroutinePromiseType : CoroListNode {
   friend struct Fut<T, E>;
 
   std::coroutine_handle<> m_waiting_coro = std::noop_coroutine();
+  Reactor &m_reactor;
   Fut<T, E> *m_out = nullptr;
 };
 
@@ -110,10 +141,10 @@ struct [[nodiscard("forgot to await?")]] Fut {
 
   Fut(const Fut &) = delete;
   Fut(Fut &&rhs) noexcept
-      : m_promise{std::exchange(rhs.m_promise, nullptr)},
+      : m_handle{std::exchange(rhs.m_handle, nullptr)},
         m_value{std::exchange(rhs.m_value, std::nullopt)} {
-    if (m_promise) {
-      m_promise->m_out = this;
+    if (m_handle) {
+      m_handle.promise().m_out = this;
     }
   }
 
@@ -124,13 +155,21 @@ struct [[nodiscard("forgot to await?")]] Fut {
     return *this;
   };
 
+  ~Fut() {
+    if (m_handle != nullptr) {
+      Reactor &reactor = m_handle.promise().m_reactor;
+      m_handle.destroy();
+      reactor.free(m_handle.address());
+    }
+  }
+
   [[nodiscard]] bool has_value() const noexcept {
     return m_value.has_value();
   }
 
   Result<T, extend_error<E, SyscallError>> block_on() && noexcept {
     while (!m_value.has_value()) {
-      Result res = Reactor::instance().do_event_loop_iteration();
+      Result res = m_handle.promise().m_reactor.do_event_loop_iteration();
       if (!res) {
         return failure(res.assume_error());
       }
@@ -157,10 +196,11 @@ struct [[nodiscard("forgot to await?")]] Fut {
     }
 
     void await_suspend(std::coroutine_handle<> h) const noexcept {
-      m_future.m_promise->m_waiting_coro = h;
+      m_future.m_handle.promise().m_waiting_coro = h;
     }
 
     Result<T, E> await_resume() const noexcept {
+      assert(m_future.m_value.has_value());
       return *std::move(m_future.m_value);
     }
 
@@ -172,31 +212,26 @@ struct [[nodiscard("forgot to await?")]] Fut {
     Fut &m_future;
   };
 
-  Awaiter operator co_await() noexcept {
+  Awaiter operator co_await() && noexcept {
     return Awaiter{*this};
   }
 
 private:
-  Fut(promise_type &prom) noexcept : m_promise{&prom} {
+  Fut(std::coroutine_handle<promise_type> handle) noexcept : m_handle{handle} {
   }
 
   Fut(AllocationError e) noexcept : m_value{failure(e)} {
   }
 
   friend promise_type;
 
-  promise_type *m_promise = nullptr;
-
-  struct CoroListNode : bi::slist_base_hook<bi::link_mode<bi::link_mode_type::auto_unlink>> {
-    virtual std::coroutine_handle<> coro_from_this() noexcept = 0;
-  };
-
-  std::optional<Result<T, E>> m_value = std::nullopt;
+  std::coroutine_handle<promise_type> m_handle = nullptr;
+  [[no_unique_address]] std::optional<Result<T, E>> m_value = std::nullopt;
 };
 
 template <typename T, typename E>
 Fut<T, E> detail::CoroutinePromiseType<T, E>::get_return_object() noexcept {
-  Fut<T, E> fut{*this};
+  Fut<T, E> fut{std::coroutine_handle<CoroutinePromiseType>::from_promise(*this)};
   m_out = &fut;
   return fut;
 }

include/corosig/Parallel.hpp

@@ -4,6 +4,7 @@
 #include "corosig/Coro.hpp"
 #include "corosig/ErrorTypes.hpp"
 #include "corosig/Result.hpp"
+#include "corosig/reactor/Reactor.hpp"
 
 #include <boost/mp11/algorithm.hpp>
 #include <concepts>
@@ -15,7 +16,7 @@
 namespace corosig {
 
 template <typename... R, typename... E>
-Fut<std::tuple<Result<R, E>...>> when_all(Fut<R, E> &&...futs) noexcept {
+Fut<std::tuple<Result<R, E>...>> when_all(Reactor &, Fut<R, E> &&...futs) noexcept {
   co_return std::tuple{co_await std::move(futs)...};
 }
 
@@ -45,8 +46,8 @@ using WrapVoid = std::conditional_t<std::same_as<void, T>, std::monostate, T>;
 
 template <typename... R, typename... E>
 Fut<std::tuple<detail::WrapVoid<R>...>, extend_error<E...>>
-when_all_succeed(Fut<R, E> &&...futs) noexcept {
-  Result results_res = co_await when_all(std::move(futs)...);
+when_all_succeed(Reactor &r, Fut<R, E> &&...futs) noexcept {
+  Result results_res = co_await when_all(r, std::move(futs)...);
   if (results_res.has_error()) {
     co_return failure(std::move(results_res.assume_error()));
   }