coro::semaphore uses coro::mutex during shutdown (#399)

* coro::semaphore uses coro::mutex during shutdown

If there are multiple in-flight tasks acquiring a semaphore while another
task is calling shutdown, it was possible for a race to occur between setting the
shutdown flag and popping all awaiters to resume them and the in-flight
acquire suspending itself and adding itself to the awaiter list.
Consequently the in-progress suspending awaiter would get lost and never
wake.

Since both acquire and release already grab the internal mutex we can solve
the problem by doing the same in shutdown such that flipping the shutdown
switch and popping the awaiters is atomic. It now cannot be the case that
an acquire mid-suspend can add itself to awaiter list after the shutdown
flag is flipped.

As a consequence, shutdown is now also a coroutine that must be awaited.

* No need to grab the lock if we're already shutdown
* Fix semaphore example

Author: wence-

README.md

@@ -689,7 +689,7 @@ int main()
         if (result == coro::semaphore_acquire_result::acquired)
         {
             std::cout << task_num << ", ";
-            semaphore.release();
+            co_await semaphore.release();
         }
         else
         {

examples/coro_semaphore.cpp

@@ -18,7 +18,7 @@ int main()
         if (result == coro::semaphore_acquire_result::acquired)
         {
             std::cout << task_num << ", ";
-            semaphore.release();
+            co_await semaphore.release();
         }
         else
         {

include/coro/semaphore.hpp

@@ -4,6 +4,7 @@
 #include "coro/expected.hpp"
 #include "coro/export.hpp"
 #include "coro/mutex.hpp"
+#include "coro/sync_wait.hpp"
 
 #include <atomic>
 #include <coroutine>
@@ -88,7 +89,7 @@ class semaphore
         : m_counter(starting_value)
     { }
 
-    ~semaphore() { shutdown(); }
+    ~semaphore() { coro::sync_wait(shutdown()); }
 
     semaphore(const semaphore&) = delete;
     semaphore(semaphore&&)      = delete;
@@ -167,12 +168,17 @@ class semaphore
      * Stops the semaphore and will notify all release/acquire waiters to wake up in a failed state.
      * Once this is set it cannot be un-done and all future oprations on the semaphore will fail.
      */
-    auto shutdown() noexcept -> void
+    [[nodiscard]] auto shutdown() noexcept -> coro::task<void>
     {
+        if (is_shutdown()) {
+            co_return;
+        }
+        auto lock = co_await m_mutex.scoped_lock();
         bool expected{false};
         if (m_shutdown.compare_exchange_strong(expected, true, std::memory_order::release, std::memory_order::relaxed))
         {
             auto* waiter = detail::awaiter_list_pop_all(m_acquire_waiters);
+            lock.unlock();
             while (waiter != nullptr)
             {
                 auto* next = waiter->m_next;

test/test_semaphore.cpp

@@ -183,7 +183,7 @@ TEST_CASE("semaphore produce consume", "[semaphore]")
         }
 
         std::cerr << "producer exiting\n";
-        s.shutdown();
+        co_await s.shutdown();
         co_return;
     };
 
@@ -252,7 +252,7 @@ TEST_CASE("semaphore 1 producers and many consumers", "[semaphore]")
         }
 
         std::cerr << "producer " << id << " exiting\n";
-        s.shutdown();
+        co_await s.shutdown();
         co_return;
     };