Disruptor-cpp/Disruptor.PerfTests/OneToThreeReleasingWorkerPoolThroughputTest.cpp at master · githubCode3a/Disruptor-cpp · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
#include "stdafx.h"
#include "OneToThreeReleasingWorkerPoolThroughputTest.h"

#include "Disruptor/BasicExecutor.h"
#include "Disruptor/FatalExceptionHandler.h"
#include "Disruptor/RoundRobinThreadAffinedTaskScheduler.h"

#include "Disruptor.TestTools/ScopeExitFunctor.h"

#include "PerfTestUtil.h"


namespace Disruptor
{
namespace PerfTests
{

    OneToThreeReleasingWorkerPoolThroughputTest::OneToThreeReleasingWorkerPoolThroughputTest()
    {
        for (auto i = 0; i < m_numWorkers; ++i)
        {
            m_counters[i] = std::make_shared< PaddedLong >();
        }

        for (auto i = 0; i < m_numWorkers; ++i)
        {
            m_handlers[i] = std::make_shared< EventCountingAndReleasingWorkHandler >(m_counters, i);
        }

        m_workerPool = std::make_shared< WorkerPool< ValueEvent > >(m_ringBuffer, m_ringBuffer->newBarrier(), std::make_shared< FatalExceptionHandler< ValueEvent > >(), m_handlers);

        m_ringBuffer->addGatingSequences({ m_workerPool->getWorkerSequences() });
    }

    std::int64_t OneToThreeReleasingWorkerPoolThroughputTest::run(Stopwatch& stopwatch)
    {
        auto scheduler = std::make_shared< RoundRobinThreadAffinedTaskScheduler >();
        scheduler->start(requiredProcessorCount());
        TestTools::ScopeExitFunctor atScopeExit([scheduler] { scheduler->stop(); });

        resetCounters();
        auto& ringBuffer = *m_workerPool->start(std::make_shared< BasicExecutor >(scheduler));
        stopwatch.start();

        for (std::int64_t i = 0; i < m_iterations; ++i)
        {
            std::int64_t sequence = ringBuffer.next();
            ringBuffer[sequence].value = i;
            ringBuffer.publish(sequence);
        }

        m_workerPool->drainAndHalt();

        // Workaround to ensure that the last worker(s) have completed after releasing their events
        std::this_thread::sleep_for(std::chrono::milliseconds(1));
        stopwatch.stop();

        PerfTestUtil::failIfNot(m_iterations, sumCounters());

        return m_iterations;
    }

    std::int32_t OneToThreeReleasingWorkerPoolThroughputTest::requiredProcessorCount() const
    {
        return 4;
    }

    void OneToThreeReleasingWorkerPoolThroughputTest::resetCounters()
    {
        for (auto i = 0; i < m_numWorkers; ++i)
        {
            m_counters[i]->value = 0L;
        }
    }

    std::int64_t OneToThreeReleasingWorkerPoolThroughputTest::sumCounters()
    {
        std::int64_t sumJobs = 0L;
        for (auto i = 0; i < m_numWorkers; ++i)
        {
            sumJobs += m_counters[i]->value;
        }

        return sumJobs;
    }

    OneToThreeReleasingWorkerPoolThroughputTest::EventCountingAndReleasingWorkHandler::EventCountingAndReleasingWorkHandler(const std::vector< std::shared_ptr< PaddedLong > >& counters, std::int32_t index)
        : m_counters(counters)
        , m_index(index)
    {
    }

    void OneToThreeReleasingWorkerPoolThroughputTest::EventCountingAndReleasingWorkHandler::onEvent(ValueEvent& /*evt*/)
    {
        m_eventReleaser->release();
        m_counters[m_index]->value = m_counters[m_index]->value + 1L;
    }

    void OneToThreeReleasingWorkerPoolThroughputTest::EventCountingAndReleasingWorkHandler::setEventReleaser(const std::shared_ptr< IEventReleaser >& eventReleaser)
    {
        m_eventReleaser = eventReleaser;
    }

} // namespace PerfTests
} // namespace Disruptor