Implement ability to sleep

Author: Dmitry Malakhov

include/corosig/Coro.hpp

@@ -5,6 +5,7 @@
 #include "corosig/Result.hpp"
 #include "corosig/reactor/CoroList.hpp"
 #include "corosig/reactor/Reactor.hpp"
+#include "corosig/reactor/SleepList.hpp"
 #include "corosig/util/SetDefaultOnMove.hpp"
 
 #include <cassert>
@@ -80,6 +81,11 @@ struct CoroutinePromiseType : CoroListNode {
     m_reactor.schedule(*this);
   }
 
+  /// @brief Add this SleepListNode into reactor to be executed later, when time comes
+  void queue_to_reactor(SleepListNode &node) noexcept {
+    m_reactor.schedule_when_time_passes(node);
+  }
+
   /// @brief Add this PollListNode into reactor to be executed later, when event becomes awailable
   void poll_to_reactor(PollListNode &node) noexcept {
     m_reactor.schedule_when_ready(node);

include/corosig/Sleep.hpp

@@ -0,0 +1,39 @@
+#ifndef COROSIG_SLEEP_HPP
+#define COROSIG_SLEEP_HPP
+
+#include "corosig/Clock.hpp"
+#include "corosig/reactor/SleepList.hpp"
+
+#include <csignal>
+#include <cstdlib>
+
+namespace corosig {
+
+/// @brief  Break the execution of a coroutine until specified amount of time passes
+struct Sleep : SleepListNode {
+  Sleep(Clock::time_point until) {
+    this->awake_time = until;
+  }
+
+  template <typename REP, typename PERIOD>
+  Sleep(std::chrono::duration<REP, PERIOD> sleep_for)
+      : Sleep{Clock::now() + sleep_for} {
+  }
+
+  [[nodiscard]] static bool await_ready() noexcept {
+    return false;
+  }
+
+  template <typename PROMISE>
+  void await_suspend(std::coroutine_handle<PROMISE> h) noexcept {
+    this->waiting_coro = h;
+    h.promise().queue_to_reactor(*this);
+  }
+
+  void await_resume() const noexcept {
+  }
+};
+
+} // namespace corosig
+
+#endif

include/corosig/reactor/Reactor.hpp

@@ -6,6 +6,7 @@
 #include "corosig/container/Allocator.hpp"
 #include "corosig/reactor/CoroList.hpp"
 #include "corosig/reactor/PollList.hpp"
+#include "corosig/reactor/SleepList.hpp"
 
 #include <cstddef>
 
@@ -34,6 +35,10 @@ struct Reactor {
   /// @brief Schedule a coroutine to be executed when handle recieves specified event
   void schedule_when_ready(PollListNode &) noexcept;
 
+  /// @brief Schedule a coroutine to be executed when specified amount of time passes
+  /// @warning UB if given node is ready
+  void schedule_when_time_passes(SleepListNode &) noexcept;
+
   /// @brief Tell if there are any tasks scheduled
   [[nodiscard]] bool has_active_tasks() const noexcept;
 
@@ -49,6 +54,7 @@ struct Reactor {
 private:
   PollList m_polled;
   CoroList m_ready;
+  SleepList m_sleeping;
   Allocator m_alloc;
 };
 

include/corosig/reactor/SleepList.hpp

@@ -0,0 +1,36 @@
+#ifndef COROSIG_REACTOR_SLEEP_LIST_HPP
+#define COROSIG_REACTOR_SLEEP_LIST_HPP
+
+#include "boost/intrusive/intrusive_fwd.hpp"
+#include "boost/intrusive/link_mode.hpp"
+#include "corosig/Clock.hpp"
+
+#include <boost/intrusive/avl_set.hpp>
+#include <boost/intrusive/avl_set_hook.hpp>
+#include <boost/intrusive/options.hpp>
+#include <coroutine>
+
+namespace corosig {
+
+/// @brief A node type for sleep task which may be pending or ready
+struct SleepListNode : boost::intrusive::avl_set_base_hook<
+                           boost::intrusive::link_mode<boost::intrusive::link_mode_type::safe_link>,
+                           boost::intrusive::optimize_size<true>> {
+  auto operator<=>(SleepListNode const &rhs) const noexcept {
+    return awake_time <=> rhs.awake_time;
+  }
+
+  /// @brief Coro to be resumed
+  std::coroutine_handle<> waiting_coro = nullptr;
+
+  /// @brief When to resume a coro
+  Clock::time_point awake_time;
+};
+
+/// @brief A list of objects which are castable to coroutines via coro_from_this virtual method
+using SleepList =
+    boost::intrusive::avl_multiset<SleepListNode, boost::intrusive::constant_time_size<false>>;
+
+} // namespace corosig
+
+#endif

src/reactor/Reactor.cpp

@@ -1,9 +1,11 @@
 #include "corosig/reactor/Reactor.hpp"
 
+#include "corosig/Clock.hpp"
 #include "corosig/ErrorTypes.hpp"
 #include "corosig/Result.hpp"
 #include "corosig/reactor/CoroList.hpp"
 #include "corosig/reactor/PollList.hpp"
+#include "corosig/reactor/SleepList.hpp"
 
 #include <chrono>
 #include <coroutine>
@@ -15,6 +17,21 @@ namespace {
 
 using namespace corosig;
 
+void resume_ready_sleepers(SleepList &sleeping) noexcept {
+  auto now = Clock::now();
+
+  while (!sleeping.empty()) {
+    SleepListNode &node = *sleeping.begin();
+    if (node.awake_time > now) {
+      break;
+    }
+    sleeping.erase(node);
+    assert(node.waiting_coro != nullptr);
+    assert(!node.waiting_coro.done());
+    node.waiting_coro.resume();
+  }
+}
+
 Result<void, SyscallError>
 poll_and_resume(PollList &polled, std::chrono::duration<int, std::milli> timeout) noexcept {
   if (polled.empty()) {
@@ -59,11 +76,16 @@ void resume(CoroList &ready) noexcept {
     auto &node = ready.front();
     ready.pop_front();
     auto coro = node.coro_from_this();
+    assert(coro != nullptr);
     assert(!coro.done());
     coro.resume();
   }
 }
 
+std::chrono::milliseconds ceil_to_millis(std::chrono::nanoseconds nanos) noexcept {
+  return std::chrono::milliseconds{nanos.count() / 1000 + nanos.count() % 1000};
+}
+
 } // namespace
 
 namespace corosig {
@@ -80,6 +102,10 @@ void Reactor::schedule_when_ready(PollListNode &node) noexcept {
   m_polled.push_back(node);
 }
 
+void Reactor::schedule_when_time_passes(SleepListNode &node) noexcept {
+  m_sleeping.insert(node);
+}
+
 bool Reactor::has_active_tasks() const noexcept {
   return !m_polled.empty() && !m_ready.empty();
 }
@@ -93,13 +119,17 @@ size_t Reactor::current_memory() const noexcept {
 }
 
 Result<void, SyscallError> Reactor::do_event_loop_iteration() noexcept {
-  assert((!m_ready.empty() || !m_polled.empty()) && "Nothing to process. Deadlock will happen");
+  assert((!m_sleeping.empty() || !m_ready.empty() || !m_polled.empty()) &&
+         "Nothing to process. Deadlock will happen");
+  resume_ready_sleepers(m_sleeping);
   resume(m_ready);
 
   using namespace std::chrono_literals;
   auto poll_timeout = -1ms;
   if (!m_ready.empty()) {
     poll_timeout = 0ms;
+  } else if (!m_sleeping.empty()) {
+    poll_timeout = std::max(0ms, ceil_to_millis(m_sleeping.begin()->awake_time - Clock::now()));
   }
 
   return poll_and_resume(m_polled, poll_timeout);

test/cases/Sleep.cpp

@@ -0,0 +1,42 @@
+#include "corosig/Sleep.hpp"
+
+#include "corosig/Clock.hpp"
+#include "corosig/Coro.hpp"
+#include "corosig/Parallel.hpp"
+#include "corosig/Result.hpp"
+#include "corosig/reactor/Reactor.hpp"
+#include "corosig/testing/Signals.hpp"
+
+namespace {
+
+using namespace corosig;
+using namespace std::chrono_literals;
+
+} // namespace
+
+COROSIG_SIGHANDLER_TEST_CASE("Sleep is ok") {
+  auto start = Clock::now();
+  auto foo = [](Reactor &) -> Fut<int> {
+    co_await Sleep{10ms};
+    co_return 20;
+  };
+  auto res = foo(reactor).block_on();
+  COROSIG_REQUIRE(res);
+  COROSIG_REQUIRE(res.value() == 20);
+  COROSIG_REQUIRE(Clock::now() - start >= 10ms);
+}
+
+COROSIG_SIGHANDLER_TEST_CASE("Parallel sleep is ok") {
+  constexpr static auto FOO = [](Reactor &) -> Fut<void> {
+    co_await Sleep{10ms};
+    co_return Ok{};
+  };
+
+  constexpr static auto FOO2 = [](Reactor &r) -> Fut<void> {
+    COROSIG_CO_TRYV(co_await when_all_succeed(r, FOO(r), FOO(r), FOO(r), FOO(r)));
+    co_return Ok{};
+  };
+
+  auto res = FOO2(reactor).block_on();
+  COROSIG_REQUIRE(res);
+}