feat: context_thread wait() methods.

- Added a wait_until() method to basic_context_thread as a
convenience method for blocking the current thread until a service
has reached a certain state.

- Added a wait() method to basic_context_thread as a convenience
method for blocking the current thread until a service has stopped.

Author: kcexn

include/net/service/async_context.hpp

@@ -56,7 +56,7 @@ struct async_context : detail::immovable {
   /** @brief An enum of all valid async context signals. */
   enum signals : std::uint8_t { terminate = 0, user1, END };
   /** @brief An enum of valid context states. */
-  enum context_states : std::uint8_t { PENDING = 0, STARTED, STOPPED };
+  enum context_state : std::uint8_t { PENDING = 0, STARTED, STOPPED };
 
   /** @brief The event loop timers. */
   timers_type timers;
@@ -67,7 +67,7 @@ struct async_context : detail::immovable {
   /** @brief The active signal mask. */
   std::atomic<signal_mask> sigmask;
   /** @brief A counter that tracks the context state. */
-  std::atomic<context_states> state{PENDING};
+  std::atomic<context_state> state{PENDING};
 
   /**
    * @brief Sets the signal mask, then interrupts the service.

include/net/service/context_thread.hpp

@@ -77,6 +77,20 @@ class basic_context_thread : public async_context {
    */
   template <typename... Args> auto start(Args &&...args) -> void;
 
+  /**
+   * @brief Waits for a condition.
+   * @details Blocks the current thread until the service state condition
+   * is satisfied.
+   * @param cond The state that the current thread should wait for.
+   */
+  auto wait_until(context_state cond) const noexcept -> void;
+
+  /**
+   * @brief Waits for the service to stop.
+   * @details Blocks the current thread until the service stops.
+   */
+  auto wait() const noexcept -> void;
+
   /** @brief The destructor signals the thread before joining it. */
   ~basic_context_thread();
 

include/net/service/impl/context_thread_impl.hpp

@@ -91,6 +91,24 @@ auto basic_context_thread<Service>::start(Args &&...args) -> void
     throw std::system_error(error, "service failed to start");
 }
 
+template <ServiceLike Service>
+auto basic_context_thread<Service>::wait_until(
+    context_state cond) const noexcept -> void
+{
+  auto curr_state = state.load();
+  while (curr_state < cond)
+  {
+    state.wait(curr_state);
+    curr_state = state.load();
+  }
+}
+
+template <ServiceLike Service>
+auto basic_context_thread<Service>::wait() const noexcept -> void
+{
+  wait_until(STOPPED);
+}
+
 template <ServiceLike Service>
 basic_context_thread<Service>::~basic_context_thread()
 {

tests/test_async_context.cpp

@@ -61,7 +61,7 @@ struct test_service {
 
 TEST_F(AsyncContextTest, AsyncServiceTest)
 {
-  using enum async_context::context_states;
+  using enum async_context::context_state;
 
   auto service = basic_context_thread<test_service>();
   service.start();
@@ -74,7 +74,7 @@ TEST_F(AsyncContextTest, AsyncServiceTest)
 
 TEST_F(AsyncContextTest, StartTwiceTest)
 {
-  using enum async_context::context_states;
+  using enum async_context::context_state;
 
   auto service = basic_context_thread<test_service>{};
 
@@ -89,7 +89,7 @@ TEST_F(AsyncContextTest, StartTwiceTest)
 
 TEST_F(AsyncContextTest, TestUser1Signal)
 {
-  using enum async_context::context_states;
+  using enum async_context::context_state;
 
   auto service = basic_context_thread<test_service>();
 
@@ -103,4 +103,205 @@ TEST_F(AsyncContextTest, TestUser1Signal)
   }
   EXPECT_EQ(test_signal, service.user1);
 }
+
+TEST_F(AsyncContextTest, WaitUntilStarted)
+{
+  using enum async_context::context_state;
+
+  auto service = basic_context_thread<test_service>();
+  ASSERT_EQ(service.state, PENDING);
+
+  auto wait_thread = std::thread([&] { service.wait_until(STARTED); });
+
+  service.start();
+  wait_thread.join();
+
+  ASSERT_EQ(service.state, STARTED);
+}
+
+TEST_F(AsyncContextTest, WaitUntilStopped)
+{
+  using enum async_context::context_state;
+
+  auto service = basic_context_thread<test_service>();
+  service.start();
+  ASSERT_EQ(service.state, STARTED);
+
+  auto wait_thread = std::thread([&] { service.wait_until(STOPPED); });
+
+  service.signal(service.terminate);
+  wait_thread.join();
+
+  ASSERT_EQ(service.state, STOPPED);
+}
+
+TEST_F(AsyncContextTest, WaitUntilAlreadySatisfied)
+{
+  using enum async_context::context_state;
+
+  auto service = basic_context_thread<test_service>();
+  service.start();
+  ASSERT_EQ(service.state, STARTED);
+
+  service.wait_until(PENDING);
+  EXPECT_EQ(service.state, STARTED);
+
+  service.wait_until(STARTED);
+  EXPECT_EQ(service.state, STARTED);
+}
+
+TEST_F(AsyncContextTest, WaitUntilMultipleWaiters)
+{
+  using enum async_context::context_state;
+
+  auto service = basic_context_thread<test_service>();
+  ASSERT_EQ(service.state, PENDING);
+
+  std::atomic<int> waiters_unblocked{0};
+
+  auto wait_thread1 = std::thread([&] {
+    service.wait_until(STARTED);
+    waiters_unblocked.fetch_add(1);
+  });
+
+  auto wait_thread2 = std::thread([&] {
+    service.wait_until(STARTED);
+    waiters_unblocked.fetch_add(1);
+  });
+
+  auto wait_thread3 = std::thread([&] {
+    service.wait_until(STARTED);
+    waiters_unblocked.fetch_add(1);
+  });
+
+  std::this_thread::sleep_for(std::chrono::milliseconds(10));
+  EXPECT_EQ(waiters_unblocked.load(), 0);
+
+  service.start();
+
+  wait_thread1.join();
+  wait_thread2.join();
+  wait_thread3.join();
+
+  EXPECT_EQ(waiters_unblocked.load(), 3);
+  ASSERT_EQ(service.state, STARTED);
+}
+
+TEST_F(AsyncContextTest, WaitUntilPendingToStopped)
+{
+  using enum async_context::context_state;
+
+  auto service = basic_context_thread<test_service>();
+  ASSERT_EQ(service.state, PENDING);
+
+  auto wait_thread = std::thread([&] { service.wait_until(STOPPED); });
+
+  service.start();
+  std::this_thread::sleep_for(std::chrono::milliseconds(10));
+
+  service.signal(service.terminate);
+  wait_thread.join();
+
+  ASSERT_EQ(service.state, STOPPED);
+}
+
+TEST_F(AsyncContextTest, WaitFromPendingToStopped)
+{
+  using enum async_context::context_state;
+
+  auto service = basic_context_thread<test_service>();
+  ASSERT_EQ(service.state, PENDING);
+
+  auto wait_thread = std::thread([&] { service.wait(); });
+
+  service.start();
+  std::this_thread::sleep_for(std::chrono::milliseconds(10));
+
+  service.signal(service.terminate);
+  wait_thread.join();
+
+  ASSERT_EQ(service.state, STOPPED);
+}
+
+TEST_F(AsyncContextTest, WaitFromStartedToStopped)
+{
+  using enum async_context::context_state;
+
+  auto service = basic_context_thread<test_service>();
+  service.start();
+  ASSERT_EQ(service.state, STARTED);
+
+  auto wait_thread = std::thread([&] { service.wait(); });
+
+  service.signal(service.terminate);
+  wait_thread.join();
+
+  ASSERT_EQ(service.state, STOPPED);
+}
+
+TEST_F(AsyncContextTest, WaitWhenAlreadyStopped)
+{
+  using enum async_context::context_state;
+
+  auto service = basic_context_thread<test_service>();
+  service.start();
+  ASSERT_EQ(service.state, STARTED);
+
+  service.signal(service.terminate);
+  service.state.wait(STARTED);
+  ASSERT_EQ(service.state, STOPPED);
+
+  service.wait();
+
+  EXPECT_EQ(service.state, STOPPED);
+}
+
+TEST_F(AsyncContextTest, WaitMultipleWaiters)
+{
+  using enum async_context::context_state;
+
+  auto service = basic_context_thread<test_service>();
+  service.start();
+  ASSERT_EQ(service.state, STARTED);
+
+  std::atomic<int> waiters_unblocked{0};
+
+  auto wait_thread1 = std::thread([&] {
+    service.wait();
+    waiters_unblocked.fetch_add(1);
+  });
+
+  auto wait_thread2 = std::thread([&] {
+    service.wait();
+    waiters_unblocked.fetch_add(1);
+  });
+
+  auto wait_thread3 = std::thread([&] {
+    service.wait();
+    waiters_unblocked.fetch_add(1);
+  });
+
+  std::this_thread::sleep_for(std::chrono::milliseconds(10));
+  EXPECT_EQ(waiters_unblocked.load(), 0);
+
+  service.signal(service.terminate);
+
+  wait_thread1.join();
+  wait_thread2.join();
+  wait_thread3.join();
+
+  EXPECT_EQ(waiters_unblocked.load(), 3);
+  ASSERT_EQ(service.state, STOPPED);
+}
+
+TEST_F(AsyncContextTest, WaitNoexceptGuarantee)
+{
+  auto service = basic_context_thread<test_service>();
+  service.start();
+
+  EXPECT_TRUE(noexcept(service.wait()));
+
+  service.signal(service.terminate);
+  service.wait();
+}
 // NOLINTEND

tests/test_async_tcp_service.cpp

@@ -109,7 +109,7 @@ TEST_F(AsyncTcpServiceTest, AsyncServerTest)
   using namespace io;
   using namespace io::socket;
   using enum async_context::signals;
-  using enum async_context::context_states;
+  using enum async_context::context_state;
 
   server_v4->start(addr_v4);
   server_v6->start(addr_v6);
@@ -155,7 +155,7 @@ TEST_F(AsyncTcpServiceTest, ServerDrainTest)
   using namespace std::chrono;
   using namespace net::timers;
   using enum async_context::signals;
-  using enum async_context::context_states;
+  using enum async_context::context_state;
 
   server_v4->start(addr_v4);
   ASSERT_EQ(server_v4->state, STARTED);

tests/test_async_udp_service.cpp

@@ -96,7 +96,7 @@ TEST_F(AsyncUDPServiceTest, AsyncServerTest)
   using namespace io;
   using namespace io::socket;
   using enum async_context::signals;
-  using enum async_context::context_states;
+  using enum async_context::context_state;
 
   server_v4->start(addr_v4);
   server_v6->start(addr_v6);

tests/test_mock_socketpair.cpp

@@ -51,7 +51,7 @@ struct test_service {
 
 TEST_F(AsyncServiceTest, StartTest)
 {
-  using enum async_context::context_states;
+  using enum async_context::context_state;
 
   auto service = basic_context_thread<test_service>();