Disruptor-cpp/Disruptor.PerfTests/OneToThreePipelineSequencedThroughputTest.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
#include "stdafx.h"
#include "OneToThreePipelineSequencedThroughputTest.h"

#include "PerfTestUtil.h"


namespace Disruptor
{
namespace PerfTests
{

    OneToThreePipelineSequencedThroughputTest::OneToThreePipelineSequencedThroughputTest()
    {
        auto stepOneFunctionHandler = std::make_shared< FunctionEventHandler >(FunctionStep::One);
        auto stepTwoFunctionHandler = std::make_shared< FunctionEventHandler >(FunctionStep::Two);
        m_stepThreeFunctionHandler = std::make_shared< FunctionEventHandler >(FunctionStep::Three);

        auto stepOneSequenceBarrier = m_ringBuffer->newBarrier();
        m_stepOneBatchProcessor = std::make_shared< BatchEventProcessor< FunctionEvent > >(m_ringBuffer, stepOneSequenceBarrier, stepOneFunctionHandler);

        auto stepTwoSequenceBarrier = m_ringBuffer->newBarrier({ m_stepOneBatchProcessor->sequence() });
        m_stepTwoBatchProcessor = std::make_shared< BatchEventProcessor< FunctionEvent > >(m_ringBuffer, stepTwoSequenceBarrier, stepTwoFunctionHandler);

        auto stepThreeSequenceBarrier = m_ringBuffer->newBarrier({ m_stepTwoBatchProcessor->sequence() });
        m_stepThreeBatchProcessor = std::make_shared< BatchEventProcessor< FunctionEvent > >(m_ringBuffer, stepThreeSequenceBarrier, m_stepThreeFunctionHandler);

        std::int64_t temp = 0;
        auto operandTwo = m_operandTwoInitialValue;

        for (std::int64_t i = 0; i < m_iterations; ++i)
        {
            auto stepOneResult = i + operandTwo--;
            auto stepTwoResult = stepOneResult + 3;

            if ((stepTwoResult & 4L) == 4L)
            {
                ++temp;
            }
        }
        m_expectedResult = temp;

        m_ringBuffer->addGatingSequences({ m_stepThreeBatchProcessor->sequence() });
    }

    std::int64_t OneToThreePipelineSequencedThroughputTest::run(Stopwatch& stopwatch)
    {
        auto latch = std::make_shared< Tests::ManualResetEvent >(false);
        m_stepThreeFunctionHandler->reset(latch, m_stepThreeBatchProcessor->sequence()->value() + m_iterations);

        auto processorTask1 = m_executor->submit(m_stepOneBatchProcessor);
        auto processorTask2 = m_executor->submit(m_stepTwoBatchProcessor);
        auto processorTask3 = m_executor->submit(m_stepThreeBatchProcessor);

        auto& rb = *m_ringBuffer;

        stopwatch.start();

        auto operandTwo = m_operandTwoInitialValue;
        for (std::int64_t i = 0; i < m_iterations; ++i)
        {
            auto sequence = rb.next();
            auto& event = rb[sequence];
            event.operandOne = i;
            event.operandTwo = operandTwo--;
            rb.publish(sequence);
        }

        latch->waitOne();
        stopwatch.stop();

        m_stepOneBatchProcessor->halt();
        m_stepTwoBatchProcessor->halt();
        m_stepThreeBatchProcessor->halt();

        processorTask1.wait();
        processorTask2.wait();
        processorTask3.wait();

        PerfTestUtil::failIfNot(m_expectedResult, m_stepThreeFunctionHandler->stepThreeCounter());

        return m_iterations;
    }

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

} // namespace PerfTests
} // namespace Disruptor