include/cppcoro/single_consumer_event.hpp

0001 ///////////////////////////////////////////////////////////////////////////////
0002 // Copyright (c) Lewis Baker
0003 // Licenced under MIT license. See LICENSE.txt for details.
0004 ///////////////////////////////////////////////////////////////////////////////
0005 #ifndef CPPCORO_SINGLE_CONSUMER_EVENT_HPP_INCLUDED
0006 #define CPPCORO_SINGLE_CONSUMER_EVENT_HPP_INCLUDED
0007
0008 #include <atomic>
0009 #include <cppcoro/coroutine.hpp>
0010
0011 namespace cppcoro
0012 {
0013     /// \brief
0014     /// A manual-reset event that supports only a single awaiting
0015     /// coroutine at a time.
0016     ///
0017     /// You can co_await the event to suspend the current coroutine until
0018     /// some thread calls set(). If the event is already set then the
0019     /// coroutine will not be suspended and will continue execution.
0020     /// If the event was not yet set then the coroutine will be resumed
0021     /// on the thread that calls set() within the call to set().
0022     ///
0023     /// Callers must ensure that only one coroutine is executing a
0024     /// co_await statement at any point in time.
0025     class single_consumer_event
0026     {
0027     public:
0028
0029         /// \brief
0030         /// Construct a new event, initialising to either 'set' or 'not set' state.
0031         ///
0032         /// \param initiallySet
0033         /// If true then initialises the event to the 'set' state.
0034         /// Otherwise, initialised the event to the 'not set' state.
0035         single_consumer_event(bool initiallySet = false) noexcept
0036             : m_state(initiallySet ? state::set : state::not_set)
0037         {}
0038
0039         /// Query if this event has been set.
0040         bool is_set() const noexcept
0041         {
0042             return m_state.load(std::memory_order_acquire) == state::set;
0043         }
0044
0045         /// \brief
0046         /// Transition this event to the 'set' state if it is not already set.
0047         ///
0048         /// If there was a coroutine awaiting the event then it will be resumed
0049         /// inside this call.
0050         void set()
0051         {
0052             const state oldState = m_state.exchange(state::set, std::memory_order_acq_rel);
0053             if (oldState == state::not_set_consumer_waiting)
0054             {
0055                 m_awaiter.resume();
0056             }
0057         }
0058
0059         /// \brief
0060         /// Transition this event to the 'non set' state if it was in the set state.
0061         void reset() noexcept
0062         {
0063             state oldState = state::set;
0064             m_state.compare_exchange_strong(oldState, state::not_set, std::memory_order_relaxed);
0065         }
0066
0067         /// \brief
0068         /// Wait until the event becomes set.
0069         ///
0070         /// If the event is already set then the awaiting coroutine will not be suspended
0071         /// and will continue execution. If the event was not yet set then the coroutine
0072         /// will be suspended and will be later resumed inside a subsequent call to set()
0073         /// on the thread that calls set().
0074         auto operator co_await() noexcept
0075         {
0076             class awaiter
0077             {
0078             public:
0079
0080                 awaiter(single_consumer_event& event) : m_event(event) {}
0081
0082                 bool await_ready() const noexcept
0083                 {
0084                     return m_event.is_set();
0085                 }
0086
0087                 bool await_suspend(cppcoro::coroutine_handle<> awaiter)
0088                 {
0089                     m_event.m_awaiter = awaiter;
0090
0091                     state oldState = state::not_set;
0092                     return m_event.m_state.compare_exchange_strong(
0093                         oldState,
0094                         state::not_set_consumer_waiting,
0095                         std::memory_order_release,
0096                         std::memory_order_acquire);
0097                 }
0098
0099                 void await_resume() noexcept {}
0100
0101             private:
0102
0103                 single_consumer_event& m_event;
0104
0105             };
0106
0107             return awaiter{ *this };
0108         }
0109
0110     private:
0111
0112         enum class state
0113         {
0114             not_set,
0115             not_set_consumer_waiting,
0116             set
0117         };
0118
0119         // TODO: Merge these two fields into a single std::atomic<std::uintptr_t>
0120         // by encoding 'not_set' as 0 (nullptr), 'set' as 1 and
0121         // 'not_set_consumer_waiting' as a coroutine handle pointer.
0122         std::atomic<state> m_state;
0123         cppcoro::coroutine_handle<> m_awaiter;
0124
0125     };
0126 }
0127
0128 #endif