struct Message
{
    int id;
    std::chrono::steady_clock::time_point timestamp;
    float data;
};

constexpr std::size_t bufferSize = 1024; // Must be power-of-two
Message buffer[bufferSize];

awaitable<void> producer(MultiProducerSequencer<std::size_t>& sequencer)
{
    const std::size_t indexMask = sequencer.buffer_size() - 1;
    
    for (int i = 0; i < 1'000'000; ++i)
    {
        // Wait until a slot is free in the buffer.
        auto sequence = co_await sequencer.claim_one();
        
        // Populate the message.
        Message& msg = buffer[sequence & indexMask];
        msg.id = i;
        msg.timestamp = std::chrono::steady_clock::now();
        msg.data = 123;
        
        // Publish message.
        sequencer.publish(sequence);
    }
    
    // Publish a sentinel.
    auto sequence = co_await sequencer.claim_one();
    
    Message& msg = buffer[sequence & indexMask];
    msg.id = -1;
    
    sequencer.publish(sequence);
    
    co_return;
}

awaitable<void> consumer(const MultiProducerSequencer<std::size_t>& sequencer,
                         SequenceBarrier<std::size_t>& consumerBarrier,
                         unsigned producerCount)
{
    const std::size_t indexMask = sequencer.buffer_size() - 1;
    
    unsigned endCount = 0;
    std::size_t nextToRead = 0;
    do
    {
        // Wait until next message is available.
        // There may be more than one available.
        const std::size_t available = 
            co_await sequencer.wait_until_publish(nextToRead, nextToRead - 1);
        do
        {
            Message& msg = buffer[nextToRead & indexMask];
            
            if (msg.id == -1) {
                // Sentry message.
                endCount++;
            } else {
                processMessage(msg);
            }
        } while (nextToRead++ != available);
        // Notify that we've finished processing up to 'available'.
        consumerBarrier.publish(available);
    } while (endConunt < producerCount);
    
    co_return;
}

SequenceBarrier<std::size_t> consumerBarrier;
MultiProducerSequencer<std::size_t> sequencer{consumerBarrier, bufferSize};

thread_pool tp{3};
co_spawn(tp, producer(sequencer), detached);
co_spawn(tp, producer(sequencer), detached);
co_spawn(tp, consumer(sequencer, consumerBarrier, 2), detached);
tp.join();

template<std::unsigned_integral TSequence, typename Traits, IsSequenceBarrier<TSequence> ConsumerBarrier, typename Awaiter>
awaitable<TSequence> MultiProducerSequencer<TSequence, Traits, ConsumerBarrier, Awaiter>::claim_one()
{
    const TSequence claimedSequence = _nextToClaim.fetch_add(1, std::memory_order_relaxed);
    co_await _consumerBarrier.wait_until_published(claimedSequence - buffer_size());
    co_return claimedSequence;
}

template<std::unsigned_integral TSequence, typename Traits, IsSequenceBarrier<TSequence> ConsumerBarrier, typename Awaiter>
awaitable<SequenceRange<TSequence, Traits>> MultiProducerSequencer<TSequence, Traits, ConsumerBarrier, Awaiter>::claim_up_to(std::size_t count)
{
    count = std::min(count, buffer_size());
    const TSequence first = _nextToClaim.fetch_add(count, std::memory_order_relaxed);
    auto claimedRange = SequenceRange<TSequence, Traits>{first, first + count};
    co_await _consumerBarrier.wait_until_published(claimedRange.back() - buffer_size());
    co_return claimedRange;
}

template<std::unsigned_integral TSequence, typename Traits, IsSequenceBarrier<TSequence> ConsumerBarrier, typename Awaiter>
void MultiProducerSequencer<TSequence, Traits, ConsumerBarrier, Awaiter>::publish(TSequence sequence)
{
    _published[sequence & _indexMask].store(sequence, std::memory_order_seq_cst);
    resume_ready_awaiters();
}

template<std::unsigned_integral TSequence, typename Traits, IsSequenceBarrier<TSequence> ConsumerBarrier, typename Awaiter>
void MultiProducerSequencer<TSequence, Traits, ConsumerBarrier, Awaiter>::publish(const SequenceRange<TSequence, Traits>& range)
{
    assert(!range.empty());

    // 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 (TSequence seq : range.skip(1))
    {
        _published[seq & _indexMask].store(seq, std::memory_order_relaxed);
    }
    // Now publish the first sequence number with seq_cst semantics.
    _published[range.front() & _indexMask].store(range.front(), std::memory_order_seq_cst);

    resume_ready_awaiters();
}

template<std::unsigned_integral TSequence, typename Traits, IsSequenceBarrier<TSequence> ConsumerBarrier, typename Awaiter>
awaitable<TSequence> MultiProducerSequencer<TSequence, Traits, ConsumerBarrier, Awaiter>::wait_until_published(TSequence targetSequence,
                                                                                                               TSequence lastKnownPublished) const
{
    auto awaiter = Awaiter{targetSequence, lastKnownPublished};
    add_awaiter(&awaiter);
    TSequence available = co_await awaiter.wait(co_await this_coro::executor);
    co_return available;
}

template<std::unsigned_integral TSequence, typename Traits, IsSequenceBarrier<TSequence> ConsumerBarrier, typename Awaiter>
TSequence MultiProducerSequencer<TSequence, Traits, ConsumerBarrier, Awaiter>::last_published_after(TSequence lastKnownPublished) const
{
    TSequence seq = lastKnownPublished + 1;
    while (_published[seq & _indexMask].load(std::memory_order_acquire) == seq) {
        lastKnownPublished = seq++;
    }
    return lastKnownPublished;
}

// Loop
do
{
    // 1
    lastKnownPublished = last_published_after(awaiters->lastKnownPublished);

    // 2
    // First scan the list of awaiters and split them into 'requeue' and 'resume' lists.
    auto minDiff = std::numeric_limits<typename Traits::difference_type>::max();
        do
        {
            auto diff = Traits::difference(awaiters->targetSequence, lastKnownPublished);
            if (diff > 0)
            {
                // Not ready yet.
                minDiff = std::min(diff, minDiff);                 // 2.1
                awaiters->lastKnownPublished = lastKnownPublished; // 2.2
                *awaitersToRequeueTail = awaiters;
                awaitersToRequeueTail = &(awaiters->next);
            }
            else
            {
                *awaitersToResumeTail = awaiters;
                awaitersToResumeTail = &(awaiters->next);
            }
            awaiters = awaiters->next;
        } while (awaiters != nullptr);
        
    // 3
    if (awaitersToRequeue != nullptr)
    {
        // Requeue awaiters
        // Check publsihed
    }
} while (awaiters != nullptr);

// Loop
do
{
    // 1
    lastKnownPublished = last_published_after(awaiters->lastKnownPublished);
    
    // 2
    // First scan the list of awaiters and split them into 'requeue' and 'resume' lists.
    
    // 3
    if (awaitersToRequeue != nullptr)
    {
        // 3.1
        // Requeue the waiters that are not ready yet.
        Awaiter* oldHead = nullptr;
        while (!_awaiters.compare_exchange_weak(oldHead,
                                                    awaitersToRequeue,
                                                    std::memory_order_seq_cst,
                                                    std::memory_order_relaxed))
        {
            *awaitersToRequeueTail = oldHead;
        }

        // Reset the awaitersToRequeue list
        awaitersToRequeueTail = &awaitersToRequeue
           
        // 3.2
        // Check published
        const TSequence earliestTargetSequence = lastKnownPublished + minDiff;
        TSequence seq = lastKnownPublished + 1;
        while (_published[seq & _indexMask].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 = _awaiters.exchange(nullptr, std::memory_order_acquire);
                break;
            }
            ++seq;
        }
    }
} while (awaiters != nullptr);

// Loop
do
{
    // 1
    // Enqueue awaiters
    
    // 2
    while (_published[(lastKnownPublished + 1) & _indexMask]
           .load(std::memory_order_seq_cst) == (lastKnownPublished + 1))
    {
        ++lastKnownPublished;
    }
    
    if (!Traits::precedes(lastKnownPublished, targetSequence))
    {
        // 3
        Awaiter* awaiters = _awaiters.exchange(nullptr, std::memory_order_acquire);
         
        // Split list of awaiters
    }   
} while (awaitersToEnqueue != nullptr);

// Loop
do
{
    // 1
    // Enqueue awaiters
    
    // 2
    // Calc lastKnownPublished
    
    if (!Traits::precedes(lastKnownPublished, targetSequence))
    {
        // 3
        Awaiter* awaiters = _awaiters.exchange(nullptr, std::memory_order_acquire);

        // 3.1
        auto minDiff = std::numeric_limits<typename Traits::difference_type>::max();
        while (awaiters != nullptr)
        {
            auto diff = Traits::difference(awaiters->targetSequence, lastKnownPublished);
            if (diff > 0)
            {
                // Not yet ready.
                minDiff = std::min(diff, minDiff); // 3.1.1
                awaiters->lastKnownPublished = lastKnownPublished; // 3.1.2
                *awaitersToEnqueueTail = awaiters;
                awaitersToEnqueueTail = &(awaiters->next);
            }
            else
            {
                // Now ready.
                *awaitersToResumeTail = awaiters;
                awaitersToResumeTail = &(awaiters->next);
            }
            awaiters = awaiters->next;
        }

        // 3.2
        // Calculate earliest target
        targetSequence = static_cast<TSequence>(lastKnownPublished + minDiff);    
    }   
} while (awaitersToEnqueue != nullptr);

#include "multi_producer_sequencer.h"

#include <boost/asio/thread_pool.hpp>
#include <boost/test/unit_test.hpp>

#include <disruptorplus/spin_wait_strategy.hpp>
#include <disruptorplus/sequence_barrier.hpp>
#include <disruptorplus/sequence_barrier_group.hpp>

namespace boost::asio::awaitable_ext::test {

namespace {
using TSequence = disruptorplus::sequence_t;
using Traits = SequenceTraits<TSequence>;

struct MockAwaiter
{
    const TSequence targetSequence;
    TSequence lastKnownPublished;
    MockAwaiter* next;

    MockAwaiter(TSequence targetSequence, TSequence lastKnownPublished)
        :
        targetSequence{targetSequence}, lastKnownPublished{lastKnownPublished},
        next{nullptr}, _published{lastKnownPublished}
    {
        assert(Traits::precedes(lastKnownPublished, targetSequence));
    }

    void resume(disruptorplus::sequence_t published) {
        assert(!Traits::precedes(published, targetSequence));
        lastKnownPublished = published;
        _published.store(published, std::memory_order_release);
        _waitStrategy.signal_all_when_blocking();
    }

    TSequence wait() const {
        const std::atomic<disruptorplus::sequence_t>* const sequences[] = { &_published };
        return _waitStrategy.wait_until_published(targetSequence, 1, sequences);
    }

private:
    std::atomic<TSequence> _published;
    mutable disruptorplus::spin_wait_strategy _waitStrategy;
};

using DummyBarrier = SequenceBarrier<TSequence>;
using BaseSequencer = MultiProducerSequencer<TSequence, Traits, DummyBarrier, MockAwaiter>;
struct Sequencer : BaseSequencer
{
    using BaseSequencer::BaseSequencer;
    using BaseSequencer::add_awaiter;
};
} // Anonymous namespace

BOOST_AUTO_TEST_CASE(_MultiProducerSequencer)
{
    constexpr std::size_t bufferSize = 1024;
    constexpr std::size_t indexMask = bufferSize - 1;
    std::uint64_t buffer[bufferSize];

    DummyBarrier dummyBarrier;
    Sequencer sequencer{dummyBarrier, bufferSize};

    using WaitStrategy = disruptorplus::spin_wait_strategy;
    WaitStrategy waitStrategy;
    disruptorplus::sequence_barrier_group<WaitStrategy> readBarrierGroup{waitStrategy};

    std::atomic<TSequence> nextToClaim = 0;

    auto claim_one = [&]() -> TSequence
    {
        const TSequence claimedSequence = nextToClaim.fetch_add(1, std::memory_order_relaxed);
        readBarrierGroup.wait_until_published(claimedSequence - bufferSize);
        return claimedSequence;
    };

    auto one_at_time_producer = [&](std::uint64_t iterationCount)
    {
        std::uint64_t i = 0;
        while (i < iterationCount)
        {
            auto seq = claim_one();
            buffer[seq & indexMask] = ++i;
            sequencer.publish(seq);
        }

        auto finalSeq = claim_one();
        buffer[finalSeq & indexMask] = 0;
        sequencer.publish(finalSeq);
    };

    auto claim_up_to = [&](std::size_t count) -> SequenceRange<TSequence, Traits>
    {
        count = std::min(count, bufferSize);
        const TSequence first = nextToClaim.fetch_add(count, std::memory_order_relaxed);
        auto claimedRange = SequenceRange<TSequence, Traits>{first, first + count};
        readBarrierGroup.wait_until_published(claimedRange.back() - bufferSize);
        return claimedRange;
    };

    auto batch_producer = [&](std::uint64_t iterationCount, std::size_t maxBatchSize)
    {
        std::uint64_t i = 0;
        while (i < iterationCount)
        {
            const std::size_t batchSize = static_cast<std::size_t>(
                std::min<std::uint64_t>(maxBatchSize, iterationCount - i));
            auto sequences = claim_up_to(batchSize);
            for (auto seq : sequences)
            {
                buffer[seq & indexMask] = ++i;
            }
            sequencer.publish(sequences);
        }

        auto finalSeq = claim_one();
        buffer[finalSeq & indexMask] = 0;
        sequencer.publish(finalSeq);
    };

    using ReadBarrier = disruptorplus::sequence_barrier<WaitStrategy>;
    auto consumer = [&](unsigned producerCount, ReadBarrier& readBarrier, std::uint64_t& sum)
    {
        TSequence nextToRead = 0;
        unsigned endCount = 0;
        do
        {
            auto awaiter = MockAwaiter{nextToRead, nextToRead - 1};
            sequencer.add_awaiter(&awaiter);
            const TSequence available = awaiter.wait();
            do
            {
                const auto& value = buffer[nextToRead & indexMask];
                sum += value;
                const bool isEndOfStream = value == 0;
                endCount += isEndOfStream ? 1 : 0;
            } while (nextToRead++ != available);

            readBarrier.publish(available);
        } while (endCount < producerCount);
    };

    constexpr std::uint64_t iterationCount = 1'000'000;
    constexpr unsigned producerCount = 2;

    ReadBarrier readBarrier1{waitStrategy}, readBarrier2{waitStrategy};
    readBarrierGroup.add(readBarrier1); readBarrierGroup.add(readBarrier2);
    std::uint64_t result1 = 0, result2 = 0;

    thread_pool tp{4};
    any_io_executor executorA = tp.get_executor();
    any_io_executor executorB = tp.get_executor();
    any_io_executor executorC = tp.get_executor();
    any_io_executor executorD = tp.get_executor();
    post(executorA, [&](){ consumer(producerCount, readBarrier1, result1); });
    post(executorB, [&](){ consumer(producerCount, readBarrier2, result2); });
    post(executorC, [&](){ one_at_time_producer(iterationCount); });
    post(executorD, [&](){ batch_producer(iterationCount, 17); });
    tp.join();

    constexpr std::uint64_t expectedResult =
        producerCount * static_cast<std::uint64_t>(iterationCount) * static_cast<std::uint64_t>(1 + iterationCount) / 2;
    BOOST_TEST(result1 == expectedResult);
    BOOST_TEST(result2 == expectedResult);
}

} // namespace boost::asio::awaitable_ext::test

// Thread producer 1

// A
// Requeue the waiters that are not ready yet.
Awaiter* oldHead = nullptr;
while (!_awaiters.compare_exchange_weak(oldHead,
                                        awaitersToRequeue,
                                        std::memory_order_seq_cst,
                                        std::memory_order_relaxed))
{
    *awaitersToRequeueTail = oldHead;
}

// B
// Check published
const TSequence earliestTargetSequence = lastKnownPublished + minDiff;
TSequence seq = lastKnownPublished + 1;
while (_published[seq & _indexMask].load(std::memory_order_seq_cst) == seq)
{
    lastKnownPublished = seq;
    if (seq == earliestTargetSequence)
    {
        awaiters = _awaiters.exchange(nullptr, std::memory_order_acquire);
        break;
    }
    ++seq;
}


// Thread producer 2

// С
// publish, write sequence
_published[sequence & _indexMask].store(sequence, std::memory_order_seq_cst);

// В
// resume_ready_awaiters, acquire awaiters
Awaiter* awaiters = _awaiters.exchange(nullptr, std::memory_order_seq_cst);