BenchmarkSessionListener.h

Go to the documentation of this file.

/*
 * Copyright Onix Solutions Limited [OnixS]. All rights reserved.
 *
 * This software owned by Onix Solutions Limited [OnixS] and is protected by copyright law
 * and international copyright treaties.
 *
 * Access to and use of the software is governed by the terms of the applicable ONIXS Software
 * Services Agreement (the Agreement) and Customer end user license agreements granting
 * a non-assignable, non-transferable and non-exclusive license to use the software
 * for it's own data processing purposes under the terms defined in the Agreement.
 *
 * Except as otherwise granted within the terms of the Agreement, copying or reproduction of any part
 * of this source code or associated reference material to any other location for further reproduction
 * or redistribution, and any amendments to this copyright notice, are expressly prohibited.
 *
 * Any reproduction or redistribution for sale or hiring of the Software not in accordance with
 * the terms of the Agreement is a violation of copyright law.
 */
 
#pragma once
 
#include <OnixS/ICE/BOE/CompilerDetection.h>
 
#include "PerformanceCounter.h"
#include "BenchmarkData.h"
#include "Helpers.h"
 
#ifdef __GNUC__
#  include <sys/mman.h>
#endif
 
#ifndef ONIXS_USE_HUGE_PAGE
#define ONIXS_USE_HUGE_PAGE 1
#endif
 
#if ONIXS_USE_HUGE_PAGE
# ifndef _WIN32
#  ifdef MAP_FAILED
#   undef MAP_FAILED
#  endif
# define MAP_FAILED reinterpret_cast<void*>(-1)
# endif
#endif
 
#ifndef MAP_HUGETLB
#  undef ONIXS_USE_HUGE_PAGE
#  define ONIXS_USE_HUGE_PAGE 0
#  undef MAP_FAILED
#endif
 
#ifdef __GNUC__
# define ONIXS_ICEBOE_PREFETCH(ptr) __builtin_prefetch(ptr, 1, 1);
#else
# define ONIXS_ICEBOE_PREFETCH(ptr) /**/
#endif
 
class Allocator
{
public:
    Allocator()
        : chunk_(init())
        , ptr_(chunk_)
        , available_(HugePageSize)
    {
    }
 
    ~Allocator()
    {
        fini(chunk_);
    }
 
    void* allocate(size_t size)
    {
        size = round(size);
 
        if(size > available_)
            throwAllocatorExhausted();
 
        void* const ptr = ptr_;
        available_ -= size;
        ptr_ = Messaging::advanceByBytes(ptr_, size);
 
#ifdef __GNUC__
       ::madvise(ptr, size, MADV_SEQUENTIAL);
       ::madvise(ptr, size, MADV_WILLNEED);
#endif
 
        return ptr;
    }
 
    template <typename T>
    T* allocate(size_t n)
    {
        return static_cast<T*>(allocate(sizeof(T) * n));
    }
 
private:
    enum { HugePageSize = 1 << 21 };
 
    void* init()
    {
        const std::size_t size = HugePageSize;
        void* ptr = nullptr;
 
#ifdef _WIN32
        ptr = _aligned_malloc(size, 4096);
#else
 
#  if ONIXS_USE_HUGE_PAGE
        const int flags =  MAP_PRIVATE | MAP_ANONYMOUS | MAP_LOCKED | MAP_POPULATE | MAP_HUGETLB;
 
        ptr = mmap(nullptr, size, PROT_READ | PROT_WRITE, flags, -1, 0);
 
        if (ptr == MAP_FAILED)
            throwNoHugePage();
#  else
        if (0 != posix_memalign(&ptr, 4096, size))
            ptr = nullptr;
#  endif
 
#endif
        if (!ptr)
            throw std::bad_alloc();
 
        return ptr;
    }
 
    void fini(void* ptr) noexcept
    {
#ifdef _WIN32
        _aligned_free(ptr);
#else
#  if ONIXS_USE_HUGE_PAGE
        munmap(ptr, HugePageSize);
# else
        free(ptr);
# endif
#endif
    }
 
    static size_t round(size_t n) noexcept
    {
        return (((n - 1) / 4096) + 1) * 4096;
    }
 
    [[noreturn]]
    static void throwNoHugePage()
    {
        struct Exception : public std::bad_alloc
        {
            const char* what() const throw() override
            {
                return
                    "Unable to allocate a huge page. "
                    "Please enable it on your system (sudo sysctl -w vm.nr_hugepages=N), "
                    "or disable it's usage in the application (ONIXS_USE_HUGE_PAGE).";
            }
        };
 
        throw Exception();
    }
 
    [[noreturn]]
    static void throwAllocatorExhausted()
    {
        struct Exception : public std::bad_alloc
        {
            const char* what() const throw() override
            {
                return "The allocator is exhausted.";
            }
        };
 
        throw Exception();
    }
 
    void * const chunk_;
    void * ptr_;
    size_t available_;
};
 
class BenchmarkSessionListener : public SessionListener
{
public:
    explicit BenchmarkSessionListener(size_t numberOfMessages)
        : allocator_()
        , active_(false)
        , packetGroupingDetected_()
        , numberOfMessages_(numberOfMessages)
        , receiveMarks_(allocator_.allocate<BenchmarkData::ReceiveMarks>(numberOfMessages))
        , receiveCounter_(0)
        , sendMarks_(allocator_.allocate<BenchmarkData::SendMarks>(numberOfMessages))
        , sendCounter_(0)
    {
        for(size_t i = 0; i < numberOfMessages; ++i)
            new (sendMarks_ + i) BenchmarkData::SendMarks();
 
        for(size_t i = 0; i < numberOfMessages; ++i)
            new (receiveMarks_ + i) BenchmarkData::ReceiveMarks();
    }
 
    ~BenchmarkSessionListener() override = default;
 
    bool active() const noexcept
    {
        return active_;
    }
 
    bool packetGroupingDetected() const noexcept
    {
        return packetGroupingDetected_;
    }
 
    const BenchmarkData::SendMarks * sendMarks() const noexcept
    {
        return sendMarks_;
    }
 
    const BenchmarkData::ReceiveMarks * receiveMarks() const noexcept
    {
        return receiveMarks_;
    }
 
    const BenchmarkData::Overhead & overhead() const noexcept
    {
        return overhead_;
    }
 
    void measureOverhead()
    {
        Threading::ThisThread::spinWait(100000);
 
        ReceivedDataTimestamp dummy;
        resetBenchmark();
 
        const size_t iterations = numberOfMessages_ - 1;
 
        for(size_t i = 0; i < iterations; i++)
        {
            beforeSending();
            onMessageSending(nullptr, 0, nullptr);
            afterSending();
            onReceivedBytes(nullptr, 0, dummy, nullptr);
            onReceivedApplicationMessage();
        }
 
        const size_t median = iterations / 2;
 
        std::vector<PerformanceCounter::Span> data;
        data.reserve(iterations);
 
        for(size_t i = 0; i < iterations; i++)
            data.push_back(receiveMarks_[i].recvSpanNano());
 
        std::sort(data.begin(), data.end());
        overhead_.receive = data[median];
 
        data.clear();
        for(size_t i = 0; i < iterations; i++)
            data.push_back(sendMarks_[i].sendSpanNano());
 
        std::sort(data.begin(), data.end());
        overhead_.send = data[median];
 
        data.clear();
        for(size_t i = 0; i < iterations; i++)
            data.push_back(sendMarks_[i].overallSendSpanNano());
 
        std::sort(data.begin(), data.end());
        overhead_.overallSend = data[median];
 
        data.clear();
        for(size_t i = 0; i < iterations; i++)
            data.push_back(BenchmarkData::oneWaySpanNano(sendMarks_[i], receiveMarks_[i]));
 
        std::sort(data.begin(), data.end());
        overhead_.oneWay = data[median];
 
        resetBenchmark();
        active_ = false;
    }
 
    virtual void resetBenchmark() noexcept
    {
        for(size_t i = 0; i < numberOfMessages_; ++i)
        {
            sendMarks_[i] = BenchmarkData::SendMarks();
            receiveMarks_[i] = BenchmarkData::ReceiveMarks();
        }
 
        sendCounter_ = 0;
        receiveCounter_ = 0;
        packetGroupingDetected_ = false;
        active_ = true;
 
        ONIXS_ICEBOE_PREFETCH(&sendMarks_[sendCounter_]);
    }
 
    ONIXS_ICEBOE_HOTPATH ONIXS_ICEBOE_FORCEINLINE
    void beforeSending() noexcept
    {
        assert(sendCounter_ < numberOfMessages_);
        PerformanceCounter::current(&sendMarks_[sendCounter_].sendStart);
    }
 
    ONIXS_ICEBOE_HOTPATH ONIXS_ICEBOE_FORCEINLINE
    void afterSending() noexcept
    {
        assert(sendCounter_ < numberOfMessages_);
        PerformanceCounter::current(&sendMarks_[sendCounter_++].overallSendFinish);
 
        ONIXS_ICEBOE_PREFETCH(&sendMarks_[sendCounter_]);
    }
 
    void onStateChange(SessionStateId::Enum newState, SessionStateId::Enum, Session *) override
    {
        switch(newState)
        {
            case SessionStateId::Established:
                active_ = true;
                ONIXS_ICEBOE_PREFETCH(&receiveMarks_[receiveCounter_]);
                break;
 
            case SessionStateId::Disconnected:
                active_ = false;
                break;
 
            default:
                break;
        }
    }
 
    ONIXS_ICEBOE_HOTPATH
    void onExecutionReport_New(const Messaging::ExecutionReport_New, Session*) override
    {
        onReceivedApplicationMessage();
    }
 
    ONIXS_ICEBOE_HOTPATH
    void onExecutionReport_Modify(const Messaging::ExecutionReport_Modify, Session*) override
    {
        onReceivedApplicationMessage();
    }
 
    ONIXS_ICEBOE_HOTPATH
    void onExecutionReport_Cancel(const Messaging::ExecutionReport_Cancel, Session*) override
    {
        onReceivedApplicationMessage();
    }
 
    ONIXS_ICEBOE_HOTPATH
    void onExecutionReport_Reject(const Messaging::ExecutionReport_Reject, Session*) override
    {
        onReceivedApplicationMessage();
    }
 
    ONIXS_ICEBOE_HOTPATH
    void onReceivedBytes(const char * /*bytes*/, size_t /*size*/, const ReceivedDataTimestamp&, Session * /* sn*/) noexcept override
    {
        if ONIXS_ICEBOE_UNLIKELY(!active_)
            return;
 
        // More than one SBE message could be received in one IP packet.
        assert(receiveCounter_ < numberOfMessages_);
        PerformanceCounter::current(&receiveMarks_[receiveCounter_].recvStart);
    }
 
    ONIXS_ICEBOE_HOTPATH
    void onMessageSending(char *, size_t, Session *) noexcept override
    {
        if ONIXS_ICEBOE_UNLIKELY(!active_)
            return;
 
        assert(sendCounter_ < numberOfMessages_);
        PerformanceCounter::current(&sendMarks_[sendCounter_].sendFinish);
    }
 
    using SessionListener::onError;
 
    void onError(
        SessionErrorReason::Enum /* reason */, const std::string & description, Session * /* sn*/, Messaging::SbeMessage) override
    {
        std::cerr << "\nSession-level error: " << description << std::endl;
        active_ = false;
    }
 
    void onWarning(
        SessionWarningReason::Enum, const std::string & description, Session *, Messaging::SbeMessage) override
    {
        std::cerr << "\nSession-level warning: " << description << std::endl;
    }
 
protected:
    virtual void onReceivedAllMessages()
    {
        active_ = false;
    }
 
private:
    ONIXS_ICEBOE_HOTPATH
    void onReceivedApplicationMessage() noexcept
    {
        if ONIXS_ICEBOE_UNLIKELY(!active_)
            return;
 
        BenchmarkData::ReceiveMarks & rm = receiveMarks_[receiveCounter_];
        PerformanceCounter::current(&rm.recvFinish);
 
        if ONIXS_ICEBOE_UNLIKELY(PerformanceCounter::isUndefinedValue(rm.recvStart))
        {
            // The received IP packet contained more than one message.
            rm.recvStart = receiveMarks_[receiveCounter_ - 1].recvFinish;
            packetGroupingDetected_ = true;
        }
 
        if ONIXS_ICEBOE_UNLIKELY(numberOfMessages_ == ++receiveCounter_)
            onReceivedAllMessages();
 
        ONIXS_ICEBOE_PREFETCH(&receiveMarks_[receiveCounter_]);
    }
 
    Allocator allocator_;
    bool active_;
    bool packetGroupingDetected_;
 
    const size_t numberOfMessages_;
 
    BenchmarkData::Overhead overhead_;
 
    alignas(ONIXS_ICEBOE_HARDWARE_DESTRUCTIVE_INTERFACE_SIZE) BenchmarkData::ReceiveMarks * const receiveMarks_;
    size_t receiveCounter_;
 
    alignas(ONIXS_ICEBOE_HARDWARE_DESTRUCTIVE_INTERFACE_SIZE) BenchmarkData::SendMarks * const sendMarks_;
    size_t sendCounter_;
};
 
class ReactorBenchmarkListener final : public BenchmarkSessionListener
{
public:
    explicit ReactorBenchmarkListener(size_t numberOfMessages)
        : BenchmarkSessionListener(numberOfMessages)
        , receivedAllMessages_()
    {}
 
    bool receivedAllMessages() const noexcept
    {
        return receivedAllMessages_;
    }
 
    void resetBenchmark() noexcept override
    {
        BenchmarkSessionListener::resetBenchmark();
        receivedAllMessages_ = false;
    }
 
    template <typename Stack>
    void collectSessionTimeMarks(
        Stack& stack, Session & session, MessageHolder<NewOrderRequest>& order,
        size_t numberOfMessages, size_t sendPeriod, size_t warmupPeriod);
 
protected:
    void onReceivedAllMessages() noexcept override
    {
        BenchmarkSessionListener::onReceivedAllMessages();
        receivedAllMessages_ = true;
    }
 
private:
    bool receivedAllMessages_;
};
 
 
template <typename Stack>
ONIXS_ICEBOE_HOTPATH
void ReactorBenchmarkListener::collectSessionTimeMarks(
    Stack& stack, Session & session, MessageHolder<NewOrderRequest>& order,
    size_t numberOfMessages, size_t sendPeriod, size_t warmupPeriod)
{
    resetBenchmark();
 
    // When the session is used with the OnixS C++ CME Market Data Handler,
    // this value can be taken from the packet receiving time
    Messaging::Timestamp sendingTime = UtcWatch::now();
 
    for(uint64_t messageCounter = 0; messageCounter < numberOfMessages; ++messageCounter)
    {
        order->orderQty(messageCounter + 1);
        order->clOrdId(messageCounter);
 
        beforeSending();
        session.send(order, sendingTime);
        afterSending();
 
        stack.dispatchEvents();
 
        sendingTime = UtcWatch::now();
 
        if(warmupPeriod || sendPeriod)
        {
            const auto start = PerformanceCounter::current();
 
            PerformanceCounter::Span elapsedMicroseconds = 0;
            do
            {
                do
                {
                    stack.dispatchEvents();
                    session.warmUp(order, sendingTime);
                    elapsedMicroseconds = PerformanceCounter::elapsedMicroseconds(start);
                }
                while(elapsedMicroseconds < static_cast<PerformanceCounter::Span>(warmupPeriod));
 
                stack.dispatchEvents();
                elapsedMicroseconds = PerformanceCounter::elapsedMicroseconds(start);
            }
            while(elapsedMicroseconds < static_cast<PerformanceCounter::Span>(sendPeriod));
        }
    }
 
    while(active())
        stack.dispatchEvents();
}