{

static_assert(

std::is_integral_v<SEQUENCE>,

"sequence_barrier requires an integral sequence type");

using awaiter_t = sequence_barrier_wait_operation_base<SEQUENCE, TRAITS>;

public:

/// Construct a sequence barrier with the specified initial sequence number

/// as the initial value 'last_published()'.

sequence_barrier(SEQUENCE initialSequence = TRAITS::initial_sequence) noexcept

: m_lastPublished(initialSequence)

, m_awaiters(nullptr)

{}

~sequence_barrier()

{

// Shouldn't be destructing a sequence barrier if there are still waiters.

assert(m_awaiters.load(std::memory_order_relaxed) == nullptr);

}

/// Query the sequence number that was most recently published by the producer.

///

/// You can assume that all sequence numbers prior to the returned sequence number

/// have also been published. This means you can safely access all elements with

/// sequence numbers up to and including the returned sequence number without any

/// further synchronisation.

SEQUENCE last_published() const noexcept

{

return m_lastPublished.load(std::memory_order_acquire);

}

/// Wait until a particular sequence number has been published.

///

/// If the specified sequence number is not yet published then the awaiting coroutine

/// will be suspended and later resumed inside the call to publish() that publishes

/// the specified sequence number.

///

/// \param targetSequence

/// The sequence number to wait for.

///

/// \return

/// An awaitable that when co_await'ed will suspend the awaiting coroutine until

/// the specified target sequence number has been published.

/// The result of the co_await expression will be the last-known published sequence

/// number. This is guaranteed not to precede \p targetSequence but may be a sequence

/// number after \p targetSequence, which indicates that more elements have been

/// published than you were waiting for.

template<typename SCHEDULER>

[[nodiscard]]

sequence_barrier_wait_operation<SEQUENCE, TRAITS, SCHEDULER> wait_until_published(

SEQUENCE targetSequence,

SCHEDULER& scheduler) const noexcept;

/// Publish the specified sequence number to consumers.

///

/// This publishes all sequence numbers up to and including the specified sequence

/// number. This will resume any coroutine that was suspended waiting for a sequence

/// number that was published by this operation.

///

/// \param sequence

/// The sequence number to publish. This number must not precede the current

/// last_published() value. ie. the published sequence numbers must be monotonically

/// increasing.

void publish(SEQUENCE sequence) noexcept;

private:

friend class sequence_barrier_wait_operation_base<SEQUENCE, TRAITS>;

void add_awaiter(awaiter_t* awaiter) const noexcept;

#if CPPCORO_COMPILER_MSVC

# pragma warning(push)

# pragma warning(disable : 4324) // C4324: structure was padded due to alignment specifier

#endif

// First cache-line is written to by the producer only

alignas(CPPCORO_CPU_CACHE_LINE)

std::atomic<SEQUENCE> m_lastPublished;

// Second cache-line is written to by both the producer and consumers

alignas(CPPCORO_CPU_CACHE_LINE)

mutable std::atomic<awaiter_t*> m_awaiters;

#if CPPCORO_COMPILER_MSVC

# pragma warning(pop)

#endif

{

public:

explicit sequence_barrier_wait_operation_base(

const sequence_barrier<SEQUENCE, TRAITS>& barrier,

SEQUENCE targetSequence) noexcept

: m_barrier(barrier)

, m_targetSequence(targetSequence)

, m_lastKnownPublished(barrier.last_published())

, m_readyToResume(false)

{}

sequence_barrier_wait_operation_base(

const sequence_barrier_wait_operation_base& other) noexcept

: m_barrier(other.m_barrier)

, m_targetSequence(other.m_targetSequence)

, m_lastKnownPublished(other.m_lastKnownPublished)

, m_readyToResume(false)

{}

bool await_ready() const noexcept

{

return !TRAITS::precedes(m_lastKnownPublished, m_targetSequence);

}

bool await_suspend(cppcoro::coroutine_handle<> awaitingCoroutine) noexcept

{

m_awaitingCoroutine = awaitingCoroutine;

m_barrier.add_awaiter(this);

return !m_readyToResume.exchange(true, std::memory_order_acquire);

}

SEQUENCE await_resume() noexcept

{

return m_lastKnownPublished;

}

protected:

friend class sequence_barrier<SEQUENCE, TRAITS>;

void resume() noexcept

{

// This synchronises with the exchange(true, std::memory_order_acquire) in await_suspend().

if (m_readyToResume.exchange(true, std::memory_order_release))

{

resume_impl();

}

}

virtual void resume_impl() noexcept = 0;

const sequence_barrier<SEQUENCE, TRAITS>& m_barrier;

const SEQUENCE m_targetSequence;

SEQUENCE m_lastKnownPublished;

sequence_barrier_wait_operation_base* m_next;

cppcoro::coroutine_handle<> m_awaitingCoroutine;

std::atomic<bool> m_readyToResume;

{

using schedule_operation = decltype(std::declval<SCHEDULER&>().schedule());

public:

sequence_barrier_wait_operation(

const sequence_barrier<SEQUENCE, TRAITS>& barrier,

SEQUENCE targetSequence,

SCHEDULER& scheduler) noexcept

: sequence_barrier_wait_operation_base<SEQUENCE, TRAITS>(barrier, targetSequence)

, m_scheduler(scheduler)

{}

sequence_barrier_wait_operation(

const sequence_barrier_wait_operation& other) noexcept

: sequence_barrier_wait_operation_base<SEQUENCE, TRAITS>(other)

, m_scheduler(other.m_scheduler)

{}

~sequence_barrier_wait_operation()

{

if (m_isScheduleAwaiterCreated)

{

m_scheduleAwaiter.destruct();

}

if (m_isScheduleOperationCreated)

{

m_scheduleOperation.destruct();

}

}

decltype(auto) await_resume() noexcept(noexcept(m_scheduleAwaiter->await_resume()))

{

if (m_isScheduleAwaiterCreated)

{

m_scheduleAwaiter->await_resume();

}

return sequence_barrier_wait_operation_base<SEQUENCE, TRAITS>::await_resume();

}

private:

void resume_impl() noexcept override

{

try

{

m_scheduleOperation.construct(m_scheduler.schedule());

m_isScheduleOperationCreated = true;

m_scheduleAwaiter.construct(detail::get_awaiter(

static_cast<schedule_operation&&>(*m_scheduleOperation)));

m_isScheduleAwaiterCreated = true;

if (!m_scheduleAwaiter->await_ready())

{

using await_suspend_result_t = decltype(m_scheduleAwaiter->await_suspend(this->m_awaitingCoroutine));

if constexpr (std::is_void_v<await_suspend_result_t>)

{

m_scheduleAwaiter->await_suspend(this->m_awaitingCoroutine);

return;

}

else if constexpr (std::is_same_v<await_suspend_result_t, bool>)

{

if (m_scheduleAwaiter->await_suspend(this->m_awaitingCoroutine))

{

return;

}

}

else

{

// Assume it returns a coroutine_handle.

m_scheduleAwaiter->await_suspend(this->m_awaitingCoroutine).resume();

return;

}

}

}

catch (...)

{

// Ignore failure to reschedule and resume inline?

// Should we catch the exception and rethrow from await_resume()?

// Or should we require that 'co_await scheduler.schedule()' is noexcept?

}

// Resume outside the catch-block.

this->m_awaitingCoroutine.resume();

}

SCHEDULER& m_scheduler;

// Can't use std::optional<T> here since T could be a reference.

detail::manual_lifetime<schedule_operation> m_scheduleOperation;

detail::manual_lifetime<typename awaitable_traits<schedule_operation>::awaiter_t> m_scheduleAwaiter;

bool m_isScheduleOperationCreated = false;

bool m_isScheduleAwaiterCreated = false;