#ifndef CPPCORO_MULTI_PRODUCER_SEQUENCER_HPP_INCLUDED

#define CPPCORO_MULTI_PRODUCER_SEQUENCER_HPP_INCLUDED

#include <cppcoro/config.hpp>

#include <cppcoro/sequence_barrier.hpp>

#include <cppcoro/sequence_range.hpp>

#include <cppcoro/sequence_traits.hpp>

#include <cppcoro/detail/manual_lifetime.hpp>

#include <atomic>

#include <cstdint>

#include <cassert>

namespace cppcoro

{

template<typename SEQUENCE, typename TRAITS, typename SCHEDULER>

class multi_producer_sequencer_claim_one_operation;

template<typename SEQUENCE, typename TRAITS, typename SCHEDULER>

class multi_producer_sequencer_claim_operation;

template<typename SEQUENCE, typename TRAITS>

class multi_producer_sequencer_wait_operation_base;

template<typename SEQUENCE, typename TRAITS, typename SCHEDULER>

class multi_producer_sequencer_wait_operation;

/// A multi-producer sequencer is a thread-synchronisation primitive that can be

/// used to synchronise access to a ring-buffer of power-of-two size where you

/// have multiple producers concurrently claiming slots in the ring-buffer and

/// publishing items.

///

/// When a writer wants to write to a slot in the buffer it first atomically

/// increments a counter by the number of slots it wishes to allocate.

/// It then waits until all of those slots have become available and then

/// returns the range of sequence numbers allocated back to the caller.

/// The caller then writes to those slots and when done publishes them by

/// writing the sequence numbers published to each of the slots to the

/// corresponding element of an array of equal size to the ring buffer.

/// When a reader wants to check if the next sequence number is available

/// it then simply needs to read from the corresponding slot in this array

/// to check if the value stored there is equal to the sequence number it

/// is wanting to read.

///

/// This means concurrent writers are wait-free when there is space available

/// in the ring buffer, requiring a single atomic fetch-add operation as the

/// only contended write operation. All other writes are to memory locations

/// owned by a particular writer. Concurrent writers can publish items out of

/// order so that one writer does not hold up other writers until the ring

/// buffer fills up.

template<

typename SEQUENCE = std::size_t,

typename TRAITS = sequence_traits<SEQUENCE>>

class multi_producer_sequencer

};

template<typename SEQUENCE, typename TRAITS, typename SCHEDULER>

class multi_producer_sequencer_claim_awaiter

};

template<typename SEQUENCE, typename TRAITS, typename SCHEDULER>

class multi_producer_sequencer_claim_operation

};

template<typename SEQUENCE, typename TRAITS, typename SCHEDULER>

class multi_producer_sequencer_claim_one_awaiter

};

template<typename SEQUENCE, typename TRAITS, typename SCHEDULER>

class multi_producer_sequencer_claim_one_operation

};

template<typename SEQUENCE, typename TRAITS>

class multi_producer_sequencer_wait_operation_base

};

template<typename SEQUENCE, typename TRAITS, typename SCHEDULER>

class multi_producer_sequencer_wait_operation :

};

template<typename SEQUENCE, typename TRAITS>

multi_producer_sequencer<SEQUENCE, TRAITS>::multi_producer_sequencer(

const sequence_barrier<SEQUENCE, TRAITS>& consumerBarrier,

std::size_t bufferSize,

SEQUENCE initialSequence)

: m_consumerBarrier(consumerBarrier)

, m_sequenceMask(bufferSize - 1)

, m_published(std::make_unique<std::atomic<SEQUENCE>[]>(bufferSize))

, m_nextToClaim(initialSequence + 1)

, m_awaiters(nullptr)

{

// bufferSize must be a positive power-of-two

assert(bufferSize > 0 && (bufferSize & (bufferSize - 1)) == 0);

// but must be no larger than the max diff value.

using diff_t = typename TRAITS::difference_type;

using unsigned_diff_t = std::make_unsigned_t<diff_t>;

constexpr unsigned_diff_t maxSize = static_cast<unsigned_diff_t>(std::numeric_limits<diff_t>::max());

assert(bufferSize <= maxSize);

SEQUENCE seq = initialSequence - (bufferSize - 1);

do

{

#ifdef __cpp_lib_atomic_value_initialization

m_published[seq & m_sequenceMask].store(seq, std::memory_order_relaxed);

#else // ^^^ __cpp_lib_atomic_value_initialization // !__cpp_lib_atomic_value_initialization vvv

std::atomic_init(&m_published[seq & m_sequenceMask], seq);

#endif // !__cpp_lib_atomic_value_initialization

} while (seq++ != initialSequence);

}

template<typename SEQUENCE, typename TRAITS>

SEQUENCE multi_producer_sequencer<SEQUENCE, TRAITS>::last_published_after(

SEQUENCE lastKnownPublished) const noexcept

{

const auto mask = m_sequenceMask;

SEQUENCE seq = lastKnownPublished + 1;

while (m_published[seq & mask].load(std::memory_order_acquire) == seq)

{

lastKnownPublished = seq++;

}

return lastKnownPublished;

}

template<typename SEQUENCE, typename TRAITS>

template<typename SCHEDULER>

multi_producer_sequencer_wait_operation<SEQUENCE, TRAITS, SCHEDULER>

multi_producer_sequencer<SEQUENCE, TRAITS>::wait_until_published(

SEQUENCE targetSequence,

SEQUENCE lastKnownPublished,

SCHEDULER& scheduler) const noexcept

{

return multi_producer_sequencer_wait_operation<SEQUENCE, TRAITS, SCHEDULER>{

*this, targetSequence, lastKnownPublished, scheduler

};

}

template<typename SEQUENCE, typename TRAITS>

bool multi_producer_sequencer<SEQUENCE, TRAITS>::any_available() const noexcept

{

return TRAITS::precedes(

m_nextToClaim.load(std::memory_order_relaxed),

m_consumerBarrier.last_published() + buffer_size());

}

template<typename SEQUENCE, typename TRAITS>

template<typename SCHEDULER>

multi_producer_sequencer_claim_one_operation<SEQUENCE, TRAITS, SCHEDULER>

multi_producer_sequencer<SEQUENCE, TRAITS>::claim_one(SCHEDULER& scheduler) noexcept

{

return multi_producer_sequencer_claim_one_operation<SEQUENCE, TRAITS, SCHEDULER>{ *this, scheduler };

}

template<typename SEQUENCE, typename TRAITS>

template<typename SCHEDULER>

multi_producer_sequencer_claim_operation<SEQUENCE, TRAITS, SCHEDULER>

multi_producer_sequencer<SEQUENCE, TRAITS>::claim_up_to(std::size_t count, SCHEDULER& scheduler) noexcept

{

return multi_producer_sequencer_claim_operation<SEQUENCE, TRAITS, SCHEDULER>{ *this, count, scheduler };

}

template<typename SEQUENCE, typename TRAITS>

void multi_producer_sequencer<SEQUENCE, TRAITS>::publish(SEQUENCE sequence) noexcept

{

m_published[sequence & m_sequenceMask].store(sequence, std::memory_order_seq_cst);

// Resume any waiters that might have been satisfied by this publish operation.

resume_ready_awaiters();

}

template<typename SEQUENCE, typename TRAITS>

void multi_producer_sequencer<SEQUENCE, TRAITS>::publish(const sequence_range<SEQUENCE, TRAITS>& range) noexcept

{

if (range.empty())

{

return;

}

// Publish all but the first sequence number using relaxed atomics.

// No consumer should be reading those subsequent sequence numbers until they've seen

// that the first sequence number in the range is published.

for (SEQUENCE seq : range.skip(1))

{

m_published[seq & m_sequenceMask].store(seq, std::memory_order_relaxed);

}

// Now publish the first sequence number with seq_cst semantics.

m_published[range.front() & m_sequenceMask].store(range.front(), std::memory_order_seq_cst);

// Resume any waiters that might have been satisfied by this publish operation.

resume_ready_awaiters();

}

template<typename SEQUENCE, typename TRAITS>

void multi_producer_sequencer<SEQUENCE, TRAITS>::resume_ready_awaiters() noexcept

{

using awaiter_t = multi_producer_sequencer_wait_operation_base<SEQUENCE, TRAITS>;

awaiter_t* awaiters = m_awaiters.load(std::memory_order_seq_cst);

if (awaiters == nullptr)

{

// No awaiters

return;

}

// There were some awaiters. Try to acquire the list of waiters with an

// atomic exchange as we might be racing with other consumers/producers.

awaiters = m_awaiters.exchange(nullptr, std::memory_order_seq_cst);

if (awaiters == nullptr)

{

// Didn't acquire the list

// Some other thread is now responsible for resuming them. Our job is done.

return;

}

SEQUENCE lastKnownPublished;

awaiter_t* awaitersToResume;

awaiter_t** awaitersToResumeTail = &awaitersToResume;

awaiter_t* awaitersToRequeue;

awaiter_t** awaitersToRequeueTail = &awaitersToRequeue;

do

{

using diff_t = typename TRAITS::difference_type;

lastKnownPublished = last_published_after(awaiters->m_lastKnownPublished);

// First scan the list of awaiters and split them into 'requeue' and 'resume' lists.

auto minDiff = std::numeric_limits<diff_t>::max();

do

{

auto diff = TRAITS::difference(awaiters->m_targetSequence, lastKnownPublished);

if (diff > 0)

{

// Not ready yet.

minDiff = diff < minDiff ? diff : minDiff;

*awaitersToRequeueTail = awaiters;

awaitersToRequeueTail = &awaiters->m_next;

}

else

{

*awaitersToResumeTail = awaiters;

awaitersToResumeTail = &awaiters->m_next;

}

awaiters->m_lastKnownPublished = lastKnownPublished;

awaiters = awaiters->m_next;

} while (awaiters != nullptr);

// Null-terinate the requeue list

*awaitersToRequeueTail = nullptr;

if (awaitersToRequeue != nullptr)

{

// Requeue the waiters that are not ready yet.

awaiter_t* oldHead = nullptr;

while (!m_awaiters.compare_exchange_weak(oldHead, awaitersToRequeue, std::memory_order_seq_cst, std::memory_order_relaxed))

{

*awaitersToRequeueTail = oldHead;

}

// Reset the awaitersToRequeue list

awaitersToRequeueTail = &awaitersToRequeue;

const SEQUENCE earliestTargetSequence = lastKnownPublished + minDiff;

// Now we need to check again to see if any of the waiters we just enqueued

// is now satisfied by a concurrent call to publish().

//

// We need to be a bit more careful here since we are no longer holding any

// awaiters and so producers/consumers may advance the sequence number arbitrarily

// far. If the sequence number advances more than buffer_size() ahead of the

// earliestTargetSequence then the m_published[] array may have sequence numbers

// that have advanced beyond earliestTargetSequence, potentially even wrapping

// sequence numbers around to then be preceding where they were before. If this

// happens then we don't need to worry about resuming any awaiters that were waiting

// for 'earliestTargetSequence' since some other thread has already resumed them.

// So the only case we need to worry about here is when all m_published entries for

// sequence numbers in range [lastKnownPublished + 1, earliestTargetSequence] have

// published sequence numbers that match the range.

const auto sequenceMask = m_sequenceMask;

SEQUENCE seq = lastKnownPublished + 1;

while (m_published[seq & sequenceMask].load(std::memory_order_seq_cst) == seq)

{

lastKnownPublished = seq;

if (seq == earliestTargetSequence)

{

// At least one of the awaiters we just published is now satisfied.

// Reacquire the list of awaiters and continue around the outer loop.

awaiters = m_awaiters.exchange(nullptr, std::memory_order_acquire);

break;

}

++seq;

}

}

} while (awaiters != nullptr);

// Null-terminate list of awaiters to resume.

*awaitersToResumeTail = nullptr;

while (awaitersToResume != nullptr)

{

awaiter_t* next = awaitersToResume->m_next;

awaitersToResume->resume(lastKnownPublished);

awaitersToResume = next;

}

}

template<typename SEQUENCE, typename TRAITS>

void multi_producer_sequencer<SEQUENCE, TRAITS>::add_awaiter(

multi_producer_sequencer_wait_operation_base<SEQUENCE, TRAITS>* awaiter) const noexcept

{

using awaiter_t = multi_producer_sequencer_wait_operation_base<SEQUENCE, TRAITS>;

SEQUENCE targetSequence = awaiter->m_targetSequence;

SEQUENCE lastKnownPublished = awaiter->m_lastKnownPublished;

awaiter_t* awaitersToEnqueue = awaiter;

awaiter_t** awaitersToEnqueueTail = &awaiter->m_next;

awaiter_t* awaitersToResume;

awaiter_t** awaitersToResumeTail = &awaitersToResume;

const SEQUENCE sequenceMask = m_sequenceMask;

do

{

// Enqueue the awaiters.

{

awaiter_t* oldHead = m_awaiters.load(std::memory_order_relaxed);

do

{

*awaitersToEnqueueTail = oldHead;

} while (!m_awaiters.compare_exchange_weak(

oldHead,

awaitersToEnqueue,

std::memory_order_seq_cst,

std::memory_order_relaxed));

}

// Reset list of waiters

awaitersToEnqueueTail = &awaitersToEnqueue;

// Check to see if the last-known published sequence number has advanced

// while we were enqueuing the awaiters. Need to use seq_cst memory order

// here to ensure that if there are concurrent calls to publish() that would

// wake up any of the awaiters we just enqueued that either we will see their

// write to m_published slots or they will see our write to m_awaiters.

//

// Note also, that we are assuming that the last-known published sequence is

// not going to advance more than buffer_size() ahead of targetSequence since

// there is at least one consumer that won't be resumed and so thus can't

// publish the sequence number it's waiting for to its sequence_barrier and so

// producers won't be able to claim its slot in the buffer.

//

// TODO: Check whether we can weaken the memory order here to just use 'seq_cst' on the

// first .load() and then use 'acquire' on subsequent .load().

while (m_published[(lastKnownPublished + 1) & sequenceMask].load(std::memory_order_seq_cst) == (lastKnownPublished + 1))

{

++lastKnownPublished;

}

if (!TRAITS::precedes(lastKnownPublished, targetSequence))

{

// At least one awaiter we just enqueued has now been satisified.

// To ensure it is woken up we need to reacquire the list of awaiters and resume

awaiter_t* awaiters = m_awaiters.exchange(nullptr, std::memory_order_acquire);

using diff_t = typename TRAITS::difference_type;

diff_t minDiff = std::numeric_limits<diff_t>::max();

while (awaiters != nullptr)

{

diff_t diff = TRAITS::difference(targetSequence, lastKnownPublished);

if (diff > 0)

{

// Not yet ready.

minDiff = diff < minDiff ? diff : minDiff;

*awaitersToEnqueueTail = awaiters;

awaitersToEnqueueTail = &awaiters->m_next;

awaiters->m_lastKnownPublished = lastKnownPublished;

}

else

{

// Now ready.

*awaitersToResumeTail = awaiters;

awaitersToResumeTail = &awaiters->m_next;

}

awaiters = awaiters->m_next;

}

// Calculate the earliest sequence number that any awaiters in the

// awaitersToEnqueue list are waiting for. We'll use this next time

// around the loop.

targetSequence = static_cast<SEQUENCE>(lastKnownPublished + minDiff);

}

// Null-terminate list of awaiters to enqueue.

*awaitersToEnqueueTail = nullptr;

} while (awaitersToEnqueue != nullptr);

// Null-terminate awaiters to resume.

*awaitersToResumeTail = nullptr;

// Finally, resume any awaiters we've found that are ready to go.

while (awaitersToResume != nullptr)

{

// Read m_next before calling .resume() as resuming could destroy the awaiter.

awaiter_t* next = awaitersToResume->m_next;

awaitersToResume->resume(lastKnownPublished);

awaitersToResume = next;

}

}

}

#endif