<commit_before>#include "Common/MPassPch.h"
#define BOOST_TEST_NO_MAIN MPassPerformanceTest
#include <boost/test/unit_test.hpp>

#include <InfiniteVector/IvProducer.h>
#include <InfiniteVector/IvConsumer.h>
#include <Common/Stopwatch.h>
#include <MPassPerformance/TestMessage.h>

using namespace MPass;
using namespace InfiniteVector;

namespace
{
    volatile std::atomic<uint32_t> producersWaiting = 0;
    volatile bool producersGo = false;

    void producerFunction(IvConnection & connection, uint32_t producerNumber, uint64_t messageCount)
    {
        IvProducer producer(connection);
        Buffers::Buffer producerBuffer;
        connection.allocate(producerBuffer);
        //{
        //    std::stringstream msg;
        //    msg << "Start producer " <<producerNumber << " in thread " << std::this_thread::get_id() << std::endl;
        //    std::cerr << msg.str() << std::flush;
        //}
        ++producersWaiting;
        while(!producersGo)
        {
            std::this_thread::yield();
        }

        for(uint64_t messageNumber = 0; messageNumber < messageCount; ++messageNumber)
        {
            auto testMessage = producerBuffer.construct<TestMessage>(producerNumber, messageNumber);
            testMessage->touch();
            producer.publish(producerBuffer);
        }

        //{
        //    std::stringstream msg;
        //    msg << "ready to Exit " << producerNumber << " T#: " << std::this_thread::get_id() << std::endl;
        //    std::cerr << msg.str() << std::flush;
        //}

    }
}

BOOST_AUTO_TEST_CASE(testMultithreadMessagePassingPerformance)
{
    IvConsumerWaitStrategy strategy;
    static const size_t entryCount = 100000;
    static const size_t bufferSize = sizeof(TestMessage);
    static const size_t bufferCount = entryCount + 10;

    static const uint64_t perProducer = 1000000 * 10; // runs about 5 seconds in release/optimized build
    static const size_t producerCount = 7; // TODO: This fails if producerCount >= 10.  Not sure why.
    
    static const uint64_t messageCount = perProducer * producerCount;

    std::cerr << "Start "<< producerCount << " producer/1 consumer thread test." << std::endl;

    IvCreationParameters parameters(strategy, entryCount, bufferSize, bufferCount);
    IvConnection connection;
    connection.createLocal("LocalIv", parameters);

    IvConsumer consumer(connection);
    Buffers::Buffer consumerBuffer;
    connection.allocate(consumerBuffer);

    std::thread producerThreads[producerCount + 1];
    uint64_t nextMessage[producerCount + 1];<commit_msg>Fixed the 'tenth thread fails' issue -- d'oh<commit_after>#include "Common/MPassPch.h"
#define BOOST_TEST_NO_MAIN MPassPerformanceTest
#include <boost/test/unit_test.hpp>

#include <InfiniteVector/IvProducer.h>
#include <InfiniteVector/IvConsumer.h>
#include <Common/Stopwatch.h>
#include <MPassPerformance/TestMessage.h>

using namespace MPass;
using namespace InfiniteVector;

namespace
{
    volatile std::atomic<uint32_t> producersWaiting = 0;
    volatile bool producersGo = false;

    void producerFunction(IvConnection & connection, uint32_t producerNumber, uint64_t messageCount)
    {
        IvProducer producer(connection);
        Buffers::Buffer producerBuffer;
        if(!connection.allocate(producerBuffer))
        {
            std::cerr << "Failed to allocate buffer for producer Number " << producerNumber << std::endl;
            return;
        }
        //{
        //    std::stringstream msg;
        //    msg << "Start producer " <<producerNumber << " in thread " << std::this_thread::get_id() << std::endl;
        //    std::cerr << msg.str() << std::flush;
        //}
        ++producersWaiting;
        while(!producersGo)
        {
            std::this_thread::yield();
        }

        for(uint64_t messageNumber = 0; messageNumber < messageCount; ++messageNumber)
        {
            auto testMessage = producerBuffer.construct<TestMessage>(producerNumber, messageNumber);
            testMessage->touch();
            producer.publish(producerBuffer);
        }

        //{
        //    std::stringstream msg;
        //    msg << "ready to Exit " << producerNumber << " T#: " << std::this_thread::get_id() << std::endl;
        //    std::cerr << msg.str() << std::flush;
        //}

    }
}

BOOST_AUTO_TEST_CASE(testMultithreadMessagePassingPerformance)
{
    IvConsumerWaitStrategy strategy;
    static const size_t entryCount = 100000;
    static const size_t bufferSize = sizeof(TestMessage);

    static const uint64_t perProducer = 1000000 * 10; // runs about 5 seconds in release/optimized build
    static const size_t producerCount = 20; 
    static const size_t bufferCount = entryCount + producerCount + 1;

    static const uint64_t messageCount = perProducer * producerCount;

    std::cerr << "Start "<< producerCount << " producer/1 consumer thread test." << std::endl;

    IvCreationParameters parameters(strategy, entryCount, bufferSize, bufferCount);
    IvConnection connection;
    connection.createLocal("LocalIv", parameters);

    IvConsumer consumer(connection);
    Buffers::Buffer consumerBuffer;
    BOOST_REQUIRE(connection.allocate(consumerBuffer));


    std::thread producerThreads[producerCount + 1];
    uint64_t nextMessage[producerCount + 1];
{'fraction_non_alphanumeric': 0.11375212224108659, 'fraction_numerical': 0.014714204867006225, 'mean_word_length': 3.2112787926926134, 'pattern_counts': {'":': 0, '<': 97, '<?xml version=': 0, '>': 20, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4620387', 'n_tokens_mistral': 1585, 'n_tokens_neox': 1489, 'n_words': 334}