include/cppcoro/single_consumer_async_auto_reset_event.hpp

0001 ///////////////////////////////////////////////////////////////////////////////
0002 // Copyright (c) Lewis Baker
0003 // Licenced under MIT license. See LICENSE.txt for details.
0004 ///////////////////////////////////////////////////////////////////////////////
0005 #ifndef CPPCORO_SINGLE_CONSUMER_ASYNC_AUTO_RESET_EVENT_HPP_INCLUDED
0006 #define CPPCORO_SINGLE_CONSUMER_ASYNC_AUTO_RESET_EVENT_HPP_INCLUDED
0007
0008 #include <cppcoro/coroutine.hpp>
0009 #include <atomic>
0010 #include <cstdint>
0011 #include <cassert>
0012
0013 namespace cppcoro
0014 {
0015     class single_consumer_async_auto_reset_event
0016     {
0017     public:
0018
0019         single_consumer_async_auto_reset_event(bool initiallySet = false) noexcept
0020             : m_state(initiallySet ? this : nullptr)
0021         {}
0022
0023         void set() noexcept
0024         {
0025             void* oldValue = m_state.exchange(this, std::memory_order_release);
0026             if (oldValue != nullptr && oldValue != this)
0027             {
0028                 // There was a waiting coroutine that we now need to resume.
0029                 auto handle = *static_cast<cppcoro::coroutine_handle<>*>(oldValue);
0030
0031                 // We also need to transition the state back to 'not set' before
0032                 // resuming the coroutine. This operation needs to be 'acquire'
0033                 // so that it synchronises with other calls to .set() that execute
0034                 // concurrently with this call and execute the above m_state.exchange(this)
0035                 // operation with 'release' semantics.
0036                 // This needs to be an exchange() instead of a store() so that it can have
0037                 // 'acquire' semantics.
0038                 (void)m_state.exchange(nullptr, std::memory_order_acquire);
0039
0040                 // Finally, resume the waiting coroutine.
0041                 handle.resume();
0042             }
0043         }
0044
0045         auto operator co_await() const noexcept
0046         {
0047             class awaiter
0048             {
0049             public:
0050
0051                 awaiter(const single_consumer_async_auto_reset_event& event) noexcept
0052                     : m_event(event)
0053                 {}
0054
0055                 bool await_ready() const noexcept { return false; }
0056
0057                 bool await_suspend(cppcoro::coroutine_handle<> awaitingCoroutine) noexcept
0058                 {
0059                     m_awaitingCoroutine = awaitingCoroutine;
0060
0061                     void* oldValue = nullptr;
0062                     if (!m_event.m_state.compare_exchange_strong(
0063                         oldValue,
0064                         &m_awaitingCoroutine,
0065                         std::memory_order_release,
0066                         std::memory_order_relaxed))
0067                     {
0068                         // This will only fail if the event was already 'set'
0069                         // In which case we can just reset back to 'not set'
0070                         // Need to use exchange() rather than store() here so we can make this
0071                         // operation an 'acquire' operation so that we get visibility of all
0072                         // writes prior to all preceding calls to .set().
0073                         assert(oldValue == &m_event);
0074                         (void)m_event.m_state.exchange(nullptr, std::memory_order_acquire);
0075                         return false;
0076                     }
0077
0078                     return true;
0079                 }
0080
0081                 void await_resume() noexcept {}
0082
0083             private:
0084                 const single_consumer_async_auto_reset_event& m_event;
0085                 cppcoro::coroutine_handle<> m_awaitingCoroutine;
0086             };
0087
0088             return awaiter{ *this };
0089         }
0090
0091     private:
0092
0093         // nullptr - not set, no waiter
0094         // this    - set
0095         // other   - not set, pointer is address of a coroutine_handle<> to resume.
0096         mutable std::atomic<void*> m_state;
0097
0098     };
0099 }
0100
0101 #endif