coro::semaphore uses coro::mutex internally (#389)

There has been a semaphore test using a thread_pool that constantly
hangs and causes test failures. After reviewing the code this is likely
due to how the m_counter on the semaphore is handled. To remove the
ambiguity of incrementing and decrementing in a coroutine this class now
uses a coro::mutex to correctly acquire and release without going over
or under 0/max_value bounds.

Consequences of this change:
* coro::semaphore::release() is now a task and must be awaited

Closes #388

Author: jbaldwin

include/coro/semaphore.hpp

@@ -3,10 +3,10 @@
 #include "coro/detail/awaiter_list.hpp"
 #include "coro/expected.hpp"
 #include "coro/export.hpp"
+#include "coro/mutex.hpp"
 
 #include <atomic>
 #include <coroutine>
-#include <mutex>
 #include <string>
 
 namespace coro
@@ -38,29 +38,33 @@ class acquire_operation
 public:
     explicit acquire_operation(semaphore<max_value>& s) : m_semaphore(s) { }
 
-    auto await_ready() const noexcept -> bool
+    [[nodiscard]] auto await_ready() const noexcept -> bool
     {
-        if (m_semaphore.m_shutdown.load(std::memory_order::acquire))
+        // If the semaphore is shutdown or a resources can be acquired without suspending release the lock and resume execution.
+        if (m_semaphore.m_shutdown.load(std::memory_order::acquire) || m_semaphore.try_acquire())
         {
+            m_semaphore.m_mutex.unlock();
             return true;
         }
-        return m_semaphore.try_acquire();
+
+        return false;
     }
 
-    auto await_suspend(std::coroutine_handle<> awaiting_coroutine) noexcept -> bool
+    auto await_suspend(const std::coroutine_handle<> awaiting_coroutine) noexcept -> bool
     {
-        // Check again now that we've setup the coroutine frame, the state could have changed.
+        // Check again now that we've set up the coroutine frame, the state could have changed.
         if (await_ready())
         {
             return false;
         }
 
         m_awaiting_coroutine = awaiting_coroutine;
         detail::awaiter_list_push(m_semaphore.m_acquire_waiters, this);
+        m_semaphore.m_mutex.unlock();
         return true;
     }
 
-    auto await_resume() const -> semaphore_acquire_result
+    [[nodiscard]] auto await_resume() const -> semaphore_acquire_result
     {
         if (m_semaphore.m_shutdown.load(std::memory_order::acquire))
         {
@@ -69,9 +73,9 @@ class acquire_operation
         return semaphore_acquire_result::acquired;
     }
 
-    acquire_operation<max_value>*      m_next{nullptr};
-    semaphore<max_value>&              m_semaphore;
-    std::coroutine_handle<> m_awaiting_coroutine;
+    acquire_operation<max_value>* m_next{nullptr};
+    semaphore<max_value>&         m_semaphore;
+    std::coroutine_handle<>       m_awaiting_coroutine;
 };
 
 } // namespace detail
@@ -80,7 +84,7 @@ template<std::ptrdiff_t max_value>
 class semaphore
 {
 public:
-    explicit semaphore(std::ptrdiff_t starting_value)
+    explicit semaphore(const std::ptrdiff_t starting_value)
         : m_counter(starting_value)
     { }
 
@@ -92,36 +96,47 @@ class semaphore
     auto operator=(const semaphore&) noexcept -> semaphore& = delete;
     auto operator=(semaphore&&) noexcept -> semaphore&      = delete;
 
-    auto release() -> void
+    /**
+     * @brief Acquires a resource from the semaphore, if the semaphore has no resources available then
+     * this will suspend and wait until a resource becomes available.
+     */
+    [[nodiscard]] auto acquire() -> coro::task<semaphore_acquire_result>
+    {
+        co_await m_mutex.lock();
+        co_return co_await detail::acquire_operation<max_value>{*this};
+    }
+
+    /**
+     * @brief Releases a resources back to the semaphore, if the semaphore is already at value() == max() this does nothing.
+     * @return
+     */
+    [[nodiscard]] auto release() -> coro::task<void>
     {
-        // If there are any waiters just transfer ownership to the waiter.
+        co_await m_mutex.lock();
+        // Do not resume or increment resources past the max_value.
+        if (value() == max())
+        {
+            m_mutex.unlock();
+            co_return;
+        }
+
+        // If there are any waiters just transfer resource ownership to the waiter.
         auto* waiter = detail::awaiter_list_pop(m_acquire_waiters);
         if (waiter != nullptr)
         {
+            m_mutex.unlock();
             waiter->m_awaiting_coroutine.resume();
         }
         else
         {
-            // Attempt to increment the counter only up to max_value.
-            auto current = m_counter.load(std::memory_order::acquire);
-            do
-            {
-                if (current >= max_value)
-                {
-                    return;
-                }
-            } while (!m_counter.compare_exchange_weak(current, current + 1, std::memory_order::acq_rel, std::memory_order::acquire));
+            // Release the resource.
+            m_counter.fetch_add(1, std::memory_order::release);
+            m_mutex.unlock();
         }
     }
 
     /**
-     * Acquires a resource from the semaphore, if the semaphore has no resources available then
-     * this will wait until a resource becomes available.
-     */
-    [[nodiscard]] auto acquire() -> detail::acquire_operation<max_value> { return detail::acquire_operation<max_value>{*this}; }
-
-    /**
-     * Attemtps to acquire a resource if there is any resources available.
+     * @brief Attempts to acquire a resource if there are any resources available.
      * @return True if the acquire operation was able to acquire a resource.
      */
     auto try_acquire() -> bool
@@ -144,7 +159,7 @@ class semaphore
     [[nodiscard]] static constexpr auto max() noexcept -> std::ptrdiff_t { return max_value; }
 
     /**
-     * The current number of resources available in this semaphore.
+     * @return The current number of resources available to acquire for this semaphore.
      */
     [[nodiscard]] auto value() const noexcept -> std::ptrdiff_t { return m_counter.load(std::memory_order::acquire); }
 
@@ -167,14 +182,20 @@ class semaphore
         }
     }
 
+    /**
+     * @return True if this semaphore has been shutdown.
+     */
     [[nodiscard]] auto is_shutdown() const -> bool { return m_shutdown.load(std::memory_order::acquire); }
 
 private:
     friend class detail::acquire_operation<max_value>;
 
+    /// @brief The current number of resources that are available to acquire.
     std::atomic<std::ptrdiff_t> m_counter;
     /// @brief The current list of awaiters attempting to acquire the semaphore.
     std::atomic<detail::acquire_operation<max_value>*> m_acquire_waiters{nullptr};
+    /// @brief mutex used to do acquire and release operations
+    coro::mutex m_mutex;
     /// @brief Flag to denote that all waiters should be woken up with the shutdown result.
     std::atomic<bool> m_shutdown{false};
 };

test/test_semaphore.cpp

@@ -28,12 +28,12 @@ TEST_CASE("semaphore binary", "[semaphore]")
         output.emplace_back(1);
         std::cerr << "coroutine done with resource, releasing\n";
         REQUIRE(s.value() == 0);
-        s.release();
+        co_await s.release();
 
         REQUIRE(s.value() == 1);
 
         REQUIRE(s.try_acquire());
-        s.release();
+        co_await s.release();
 
         co_return;
     };
@@ -43,7 +43,7 @@ TEST_CASE("semaphore binary", "[semaphore]")
     REQUIRE(s.value() == 1);
     REQUIRE(s.try_acquire());
     REQUIRE(s.value() == 0);
-    s.release();
+    coro::sync_wait(s.release());
     REQUIRE(s.value() == 1);
 
     REQUIRE(output.size() == 1);
@@ -71,7 +71,7 @@ TEST_CASE("semaphore binary many waiters until event", "[semaphore]")
         std::cerr << "id = " << id << " semaphore acquired\n";
         value.fetch_add(1, std::memory_order::relaxed);
         std::cerr << "id = " << id << " semaphore release\n";
-        s.release();
+        co_await s.release();
         co_return;
     };
 
@@ -83,7 +83,7 @@ TEST_CASE("semaphore binary many waiters until event", "[semaphore]")
         std::cerr << "block task acquired semaphore, waiting on event\n";
         co_await e;
         std::cerr << "block task releasing semaphore\n";
-        s.release();
+        co_await s.release();
         co_return;
     };
 
@@ -113,11 +113,11 @@ TEST_CASE("semaphore binary many waiters until event", "[semaphore]")
 TEST_CASE("semaphore release over max", "[semaphore]")
 {
     coro::semaphore<2> s{0};
-    s.release();
+    coro::sync_wait(s.release());
     REQUIRE(s.value() == 1);
-    s.release();
+    coro::sync_wait(s.release());
     REQUIRE(s.value() == 2);
-    s.release();
+    coro::sync_wait(s.release());
     REQUIRE(s.value() == 2);
 }
 
@@ -132,10 +132,9 @@ TEST_CASE("semaphore try_acquire", "[semaphore]")
 TEST_CASE("semaphore produce consume", "[semaphore]")
 {
     std::cerr << "BEGIN semaphore produce consume\n";
-    const std::size_t iterations = 10;
+    constexpr std::size_t iterations = 10;
 
-    // This test is run in the context of a thread pool so the producer task can yield.  Otherwise
-    // the producer will just run wild!
+    // This test is run in the context of a thread pool so the producer task can yield. Otherwise, the producer will just run wild!
     auto tp = coro::thread_pool::make_shared(coro::thread_pool::options{.thread_count = 1});
     std::atomic<uint64_t>         value{0};
     std::vector<coro::task<void>> tasks;
@@ -156,7 +155,6 @@ TEST_CASE("semaphore produce consume", "[semaphore]")
                 std::cerr << "id = " << id << " semaphore acquired, consuming value\n";
 
                 value.fetch_add(1, std::memory_order::release);
-                // In the ringbfuffer acquire is 'consuming', we never release back into the buffer
             }
             else
             {
@@ -180,7 +178,7 @@ TEST_CASE("semaphore produce consume", "[semaphore]")
             co_await tp->yield();
 
             std::cerr << "producer: releasing\n";
-            s.release();
+            co_await s.release();
             std::cerr << "producer: produced\n";
         }
 
@@ -201,9 +199,9 @@ TEST_CASE("semaphore produce consume", "[semaphore]")
 TEST_CASE("semaphore 1 producers and many consumers", "[semaphore]")
 {
     std::cerr << "BEGIN semaphore 1 producers and many consumers\n";
-    const std::size_t consumers  = 16;
-    const std::size_t producers  = 1;
-    const std::size_t iterations = 100'000;
+    constexpr std::size_t consumers  = 16;
+    constexpr std::size_t producers  = 1;
+    constexpr std::size_t iterations = 100'000;
 
     std::atomic<uint64_t> value{0};
 
@@ -243,8 +241,7 @@ TEST_CASE("semaphore 1 producers and many consumers", "[semaphore]")
 
         for (size_t i = 0; i < iterations; ++i)
         {
-            // std::cerr << "producer " << id << " s.release()\n";
-            s.release();
+            co_await s.release();
             co_await tp->yield();
         }