#include <cppcoro/multi_producer_sequencer.hpp>

#include <cppcoro/config.hpp>

#include <cppcoro/sequence_barrier.hpp>

#include <cppcoro/sequence_traits.hpp>

#include <cppcoro/on_scope_exit.hpp>

#include <cppcoro/sync_wait.hpp>

#include <cppcoro/when_all.hpp>

#include <cppcoro/task.hpp>

#include <cppcoro/static_thread_pool.hpp>

#include <thread>

#include <chrono>

#include <ostream>

#include "doctest/doctest.h"

DOCTEST_TEST_SUITE_BEGIN("multi_producer_sequencer");

using namespace cppcoro;

{

task<> one_at_a_time_producer(

}

task<> batch_producer(

}

task<std::uint64_t> consumer(

}

}

DOCTEST_TEST_CASE("two producers (batch) / single consumer")

{

static_thread_pool tp{ 3 };

// Allow time for threads to start up.

using namespace std::chrono_literals;

std::this_thread::sleep_for(1ms);

constexpr std::size_t batchSize = 10;

constexpr std::size_t bufferSize = 16384;

sequence_barrier<std::size_t> readBarrier;

multi_producer_sequencer<std::size_t> sequencer(readBarrier, bufferSize);

constexpr std::uint64_t iterationCount = 1'000'000;

std::uint64_t buffer[bufferSize];

auto startTime = std::chrono::high_resolution_clock::now();

constexpr std::uint32_t producerCount = 2;

auto result = std::get<0>(sync_wait(when_all(

consumer(tp, sequencer, readBarrier, buffer, producerCount),

batch_producer(tp, sequencer, buffer, iterationCount, batchSize),

batch_producer(tp, sequencer, buffer, iterationCount, batchSize))));

auto endTime = std::chrono::high_resolution_clock::now();

auto totalTimeInNs = std::chrono::duration_cast<std::chrono::nanoseconds>(endTime - startTime).count();

MESSAGE(

"Producers = " << producerCount

<< ", BatchSize = " << batchSize

<< ", MessagesPerProducer = " << iterationCount

<< ", TotalTime = " << totalTimeInNs/1000 << "us"

<< ", TimePerMessage = " << totalTimeInNs/double(iterationCount * producerCount) << "ns"

<< ", MessagesPerSecond = " << 1'000'000'000 * (producerCount * iterationCount) / totalTimeInNs);

constexpr std::uint64_t expectedResult =

producerCount * std::uint64_t(iterationCount) * std::uint64_t(iterationCount + 1) / 2;

CHECK(result == expectedResult);

}

DOCTEST_TEST_CASE("two producers (single) / single consumer")

{

static_thread_pool tp{ 3 };

// Allow time for threads to start up.

using namespace std::chrono_literals;

std::this_thread::sleep_for(1ms);

constexpr std::size_t bufferSize = 16384;

sequence_barrier<std::size_t> readBarrier;

multi_producer_sequencer<std::size_t> sequencer(readBarrier, bufferSize);

constexpr std::uint64_t iterationCount = 1'000'000;

std::uint64_t buffer[bufferSize];

auto startTime = std::chrono::high_resolution_clock::now();

constexpr std::uint32_t producerCount = 2;

auto result = std::get<0>(sync_wait(when_all(

consumer(tp, sequencer, readBarrier, buffer, producerCount),

one_at_a_time_producer(tp, sequencer, buffer, iterationCount),

one_at_a_time_producer(tp, sequencer, buffer, iterationCount))));

auto endTime = std::chrono::high_resolution_clock::now();

auto totalTimeInNs = std::chrono::duration_cast<std::chrono::nanoseconds>(endTime - startTime).count();

MESSAGE(

"Producers = " << producerCount

<< ", NoBatch"

<< ", MessagesPerProducer = " << iterationCount

<< ", TotalTime = " << totalTimeInNs / 1000 << "us"

<< ", TimePerMessage = " << totalTimeInNs / double(iterationCount * producerCount) << "ns"

<< ", MessagesPerSecond = " << 1'000'000'000 * (producerCount * iterationCount) / totalTimeInNs);

constexpr std::uint64_t expectedResult =

producerCount * std::uint64_t(iterationCount) * std::uint64_t(iterationCount + 1) / 2;

CHECK(result == expectedResult);

}

DOCTEST_TEST_SUITE_END();