send logs on crash example

Author: Dmitry Malakhov

.clang-tidy

@@ -3,6 +3,7 @@ Checks: >
     -performance-enum-size,
     bugprone-*,
     -bugprone-easily-swappable-parameters,
+    -bugprone-suspicious-stringview-data-usage,
     modernize-*,
     -modernize-use-trailing-return-type,
     portability-*,

.github/workflows/Codestyle.yaml

@@ -26,4 +26,4 @@ jobs:
         run: devenv shell "xmake project -k compile_commands -y"
 
       - name: clang-tidy checks
-        run: devenv shell "find include src -type f \( -name \*.cpp -o -name \*.hpp \) -print0 | xargs -0 -n1 -P$(nproc) clang-tidy -p build"
+        run: devenv shell "find include src example -type f \( -name \*.cpp -o -name \*.hpp \) -print0 | xargs -0 -n1 -P$(nproc) clang-tidy -p build"

example/SendLogsOnCrash.cpp

@@ -0,0 +1,135 @@
+/// 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
+/// 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 "boost/outcome/success_failure.hpp"
+#include "corosig/reactor/Default.hpp"
+
+#include <algorithm>
+#include <boost/outcome/try.hpp>
+#include <corosig/Coro.hpp>
+#include <corosig/ErrorTypes.hpp>
+#include <corosig/Parallel.hpp>
+#include <corosig/Result.hpp>
+#include <corosig/Sighandler.hpp>
+#include <corosig/io/File.hpp>
+#include <corosig/io/TcpSocket.hpp>
+#include <csignal>
+#include <fstream>
+#include <netinet/in.h>
+#include <thread>
+
+namespace {
+
+sockaddr_storage const SERVER_ADDR = [] {
+  ::sockaddr_storage addr;
+  std::memset(&addr, 0, sizeof(addr));
+
+  auto *addr4 = (::sockaddr_in *)&addr;
+  addr4->sin_family = AF_INET;
+  addr4->sin_port = htons(8080);
+  addr4->sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+  return addr;
+}();
+
+constexpr std::string_view FILE1 = "file1.log";
+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;
+
+  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();
+  };
+
+  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();
+  };
+
+  BOOST_OUTCOME_CO_TRY(co_await when_all_succeed(write_to_file(FILE1.data()),
+                                                 write_to_file(FILE2.data()), send_via_tcp()));
+
+  co_return success();
+}
+
+} // namespace
+
+int main() {
+  constexpr auto REACTOR_MEMORY = 8 * 1024;
+  for (auto signal : {SIGILL, SIGFPE, SIGTERM, SIGABRT}) {
+    corosig::set_sighandler<REACTOR_MEMORY, sighandler>(signal);
+  }
+
+  std::string remote_server_data;
+  auto remote_server_thread = std::jthread([&] {
+    int srv_fd = ::socket(AF_INET, SOCK_STREAM, 0);
+    assert(srv_fd >= 0);
+
+    sockaddr_in addr{};
+    addr.sin_family = AF_INET;
+    addr.sin_port = ::htons(8080);
+    addr.sin_addr.s_addr = ::htonl(INADDR_LOOPBACK);
+
+    int opt = 1;
+    setsockopt(srv_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
+    ::bind(srv_fd, (sockaddr *)&addr, sizeof(addr)); // NOLINT
+    ::listen(srv_fd, 1);
+
+    int client = ::accept(srv_fd, nullptr, nullptr);
+    char buf[1024]; // NOLINT
+    while (true) {
+      ssize_t n = ::read(client, buf, sizeof(buf));
+      if (n <= 0) {
+        break;
+      }
+      remote_server_data += std::string_view{buf, size_t(n)};
+    }
+    ::close(client);
+    ::close(srv_fd);
+  });
+
+  logs_buffer.emplace_back("Log message 1\n");
+  logs_buffer.emplace_back("Log message 2\n");
+  logs_buffer.emplace_back("Log message 3\n");
+  logs_buffer.emplace_back("Log message 4\n");
+
+  ::raise(SIGFPE);
+
+  {
+    std::ifstream file{FILE1.data()};
+    if (!file.is_open()) {
+      std::cerr << "Failed to open file 1\n";
+    } else {
+      std::cout << "File 1 contains\n" << file.rdbuf() << '\n';
+    }
+  }
+  {
+    std::ifstream file{FILE2.data()};
+    if (!file.is_open()) {
+      std::cerr << "Failed to open file 2\n";
+    } else {
+      std::cout << "File 2 contains\n" << file.rdbuf() << '\n';
+    }
+  }
+
+  remote_server_thread.join();
+  std::cout << "Remote server received\n" << remote_server_data << '\n';
+
+  return 0;
+}

include/corosig/Coro.hpp

@@ -37,23 +37,19 @@ struct CoroutinePromiseType : CoroListNode {
 
   template <typename... ARGS>
   static void *operator new(size_t n, ARGS &&...) noexcept {
-    return reactor().allocate_frame(n);
+    return REACTOR::instance().allocate_frame(n);
   }
 
   static void operator delete(void *frame) noexcept {
-    return reactor().free_frame(frame);
-  }
-
-  static REACTOR &reactor() noexcept {
-    return reactor_provider<REACTOR>::engine();
+    return REACTOR::instance().free_frame(frame);
   }
 
   void yield_to_reactor() noexcept {
-    reactor().yield(*this);
+    REACTOR::instance().yield(*this);
   }
 
   void poll_to_reactor(PollListNode &node) noexcept {
-    reactor().poll(node);
+    REACTOR::instance().poll(node);
   }
 
   [[noreturn]] static void unhandled_exception() noexcept {
@@ -134,7 +130,7 @@ struct [[nodiscard("forgot to await?")]] Fut {
 
   Result<T, extend_error<E, SyscallError>> block_on() && noexcept {
     while (!m_value.has_value()) {
-      Result res = promise_type::reactor().do_event_loop_iteration();
+      Result res = REACTOR::instance().do_event_loop_iteration();
       if (!res) {
         return failure(std::move(res.assume_error()));
       }

include/corosig/Parallel.hpp

@@ -3,7 +3,11 @@
 #include "corosig/Result.hpp"
 
 #include <boost/mp11/algorithm.hpp>
+#include <concepts>
 #include <optional>
+#include <type_traits>
+#include <utility>
+#include <variant>
 
 namespace corosig {
 
@@ -31,13 +35,17 @@ std::optional<ERROR> first_error(std::tuple<RESULTS...> &t) noexcept {
   return error_opt;
 }
 
+template <typename T>
+using WrapVoid = std::conditional_t<std::same_as<void, T>, std::monostate, T>;
+
 } // namespace detail
 
 template <typename... REACTOR, typename... R, typename... E>
-Fut<std::tuple<R...>, extend_error<E...>> when_all_succeed(Fut<R, E> &&...futs) noexcept {
+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)...);
   if (results_res.has_error()) {
-    co_return failure(results_res.assume_error());
+    co_return failure(std::move(results_res.assume_error()));
   }
   std::tuple<Result<R, E>...> &results = results_res.assume_value();
 
@@ -47,7 +55,13 @@ Fut<std::tuple<R...>, extend_error<E...>> when_all_succeed(Fut<R, E> &&...futs)
 
   co_return success(std::apply(
       [](Result<R, E> &&...current_result) {
-        return std::tuple<R...>{std::move(current_result.assume_value())...};
+        return std::tuple{[](Result<R, E> &&r) {
+          if constexpr (std::same_as<void, R>) {
+            return std::monostate{};
+          } else {
+            return std::move(r.assume_value());
+          }
+        }(std::move(current_result))...};
       },
       std::move(results)));
 }

include/corosig/Result.hpp

@@ -6,7 +6,8 @@
 namespace corosig {
 
 template <typename R, typename E>
-struct Result : boost::outcome_v2::result<R, E, boost::outcome_v2::policy::terminate> {
+struct [[nodiscard]] Result
+    : boost::outcome_v2::result<R, E, boost::outcome_v2::policy::terminate> {
   using boost::outcome_v2::result<R, E, boost::outcome_v2::policy::terminate>::result;
 };
 

include/corosig/Sighandler.hpp

@@ -0,0 +1,37 @@
+#pragma once
+
+#include "corosig/reactor/Default.hpp"
+
+#include <csignal>
+#include <stdexcept>
+
+namespace corosig {
+
+namespace detail {
+
+template <size_t MEMORY, auto F>
+void sighandler(int sig) noexcept {
+  std::signal(sig, SIG_DFL);
+
+  Alloc::Memory<MEMORY> mem;
+  auto &reactor = Reactor::instance();
+  reactor = Reactor{mem};
+
+  (void)F(sig).block_on();
+}
+
+} // namespace detail
+
+template <size_t MEMORY, auto F>
+void set_sighandler(int sig) {
+  // initialize underlying TLS with default reactor
+  // important to do it here since it may cause allocation
+  // which is not safe to do inside sighandler
+  (void)Reactor::instance();
+
+  if (std::signal(sig, detail::sighandler<MEMORY, F>) == SIG_ERR) {
+    throw std::runtime_error{"std::signal failed"};
+  }
+}
+
+} // namespace corosig

include/corosig/reactor/Custom.hpp

@@ -8,17 +8,14 @@
 
 namespace corosig {
 
-template <typename REACTOR>
-struct reactor_provider;
-
 template <typename REACTOR>
 concept AReactor = requires(REACTOR &reactor) {
   { reactor.allocate_frame(std::declval<size_t>()) } noexcept -> std::same_as<void *>;
   { reactor.free_frame(std::declval<void *>()) } noexcept -> std::same_as<void>;
   { reactor.yield(std::declval<CoroListNode &>()) } noexcept -> std::same_as<void>;
   { reactor.poll(std::declval<PollListNode &>()) } noexcept -> std::same_as<void>;
   { reactor.do_event_loop_iteration() } noexcept;
-  { reactor_provider<REACTOR>::engine() } noexcept -> std::same_as<REACTOR &>;
+  { REACTOR::instance() } noexcept -> std::same_as<REACTOR &>;
 };
 
 } // namespace corosig

include/corosig/reactor/Default.hpp

@@ -15,6 +15,8 @@
 namespace corosig {
 
 struct Reactor {
+  static Reactor &instance() noexcept;
+
   Reactor() noexcept = default;
 
   template <size_t SIZE>
@@ -47,11 +49,6 @@ struct Reactor {
   Alloc m_alloc;
 };
 
-template <>
-struct reactor_provider<Reactor> {
-  static Reactor &engine() noexcept;
-};
-
 static_assert(AReactor<Reactor>);
 
 } // namespace corosig

src/reactor/Default.cpp

@@ -73,7 +73,7 @@ void resume(CoroList &ready) noexcept {
 
 namespace corosig {
 
-Reactor &reactor_provider<Reactor>::engine() noexcept {
+Reactor &Reactor::instance() noexcept {
   thread_local Reactor thread_local_reactor;
   return thread_local_reactor;
 }