Benchmark.cpp

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.
 */
// Uncomment to disable using of huge pages.
//#define ONIXS_USE_HUGE_PAGE 0
 
// Uncomment to enable debug mode.
//#define ONIXS_DEBUGGING
 
#include <iostream>
 
#include <OnixS/ICE/BOE/IceBoe.h>
 
using namespace ONIXS_ICEBOE_NAMESPACE;
using namespace ONIXS_ICEBOE_MESSAGING_NAMESPACE;
 
#include <Settings/Defaults.h>
#include <Common/Helpers.h>
#include <Common/Listener.h>
#include <Common/Settings.h>
#include <Common/Signal.h>
#include <Common/Emulator.h>
#include <Common/BenchmarkSessionListener.h>
 
using namespace Samples;
 
int main(int argc, char* argv[])
{
    // `--help` to show options.
    const AppConfiguration<
            SettingsConfiguration
            , ConnectivityConfiguration
            , LogonConfiguration
            , StorageConfiguration<SessionStorageType::MemoryBased>
            , MainThreadAffinityConfiguration
            , ReceivingThreadAffinityConfiguration
            , BenchmarkConfiguration
        > cfg{"Benchmark", argc, argv};
 
    try
    {
        const auto numberOfMessages = cfg.numberOfMessages();
        const auto warmupInterval = cfg.warmupInterval();
        const auto intervalBetweenSending = (std::max)(cfg.intervalBetweenSending(), warmupInterval);
        const auto storageType = cfg.storage();
        const auto receivingThreadAffinity = cfg.receivingThreadCpu();
 
        ThisThread::affinity(cfg.mainThreadCpu());
        SignalHelper::manageSignals();
 
        auto settings = fillSettings(cfg);
        const auto emulator = createEmulator(settings, cfg);
 
        // Request Binary Order Gateway credentials from Binary Utility Service Gateway
        const auto bgwCredentials = receiveBgwCredentials(settings, cfg.host(), cfg.port());
 
        settings
            .receiveSpinningTimeout(1000 * 1000)
            .useSpinLock(true)
            .logBeforeSending(false);
 
        class BenchmarkListener final : public BenchmarkSessionListener
        {
        public:
            explicit BenchmarkListener(size_t numberOfMessages)
                : BenchmarkSessionListener(numberOfMessages)
            {}
 
            using SessionListener::onError;
 
            void onError(
                    SessionErrorReason::Enum reason, const std::string & description, Session * session, Messaging::SbeMessage msg) override
            {
                BenchmarkSessionListener::onError(reason, description, session, msg);
                receivedAllMessages_.set_exception(std::make_exception_ptr(std::runtime_error(description)));
            }
 
            ONIXS_ICEBOE_HOTPATH
            void collectSessionTimeMarks(Session & session, MessageHolder<NewOrderRequest> & order,
                size_t numberOfMessages, size_t sendPeriodUsec, size_t warmupPeriodUsec)
            {
                receivedAllMessages_ = std::promise<void>{};
 
                resetBenchmark();
 
                auto sendingTime = UtcWatch::now();
 
                for(uint64_t messageCounter = 0; messageCounter < numberOfMessages; ++messageCounter)
                {
                    order->orderQty(messageCounter + 1);
                    order->clOrdId(messageCounter);
 
                    beforeSending();
                    session.send(order, sendingTime);
                    afterSending();
 
                    sendingTime = UtcWatch::now();
 
                    if(warmupPeriodUsec)
                    {
                        const auto start = PerformanceCounter::current();
 
                        do
                        {
                            ThisThread::spinWait(warmupPeriodUsec);
                            session.warmUp(order, sendingTime);
                        }
                        while(PerformanceCounter::elapsedMicroseconds(start) < static_cast<PerformanceCounter::Span>(sendPeriodUsec));
                    }
                    else if(sendPeriodUsec)
                    {
                        ThisThread::spinWait(sendPeriodUsec);
                    }
                }
 
                waitUntilReceivedAllMessages();
            }
 
        private:
            void onReceivedAllMessages() override
            {
                BenchmarkSessionListener::onReceivedAllMessages();
                receivedAllMessages_.set_value();
            }
 
            void waitUntilReceivedAllMessages()
            {
                auto future = receivedAllMessages_.get_future();
 
                if(future.wait_for(std::chrono::seconds{120}) != std::future_status::ready)
                    receivedAllMessages_.set_exception(std::make_exception_ptr(std::runtime_error("Operation timed out.")));
 
                future.get();
            }
 
            std::promise<void> receivedAllMessages_;
        }
        listener{numberOfMessages};
 
        printBenchmarkSettings(std::clog, numberOfMessages, intervalBetweenSending, warmupInterval, storageType);
 
        const auto bgwSession= std::make_unique<BgwSession>(settings, &listener, cfg.storage());
 
        const auto guard = setPriorityAndPolicy(bgwSession.get());
 
        listener.measureOverhead();
 
        bgwSession->
            receivingThreadAffinity(receivingThreadAffinity)
            .sendingThreadAffinity(Default::AuxiliarySendingThreadAffinity)
            .connect(bgwCredentials.host(), bgwCredentials.port(), bgwCredentials.ipSessionToken());
 
        auto order = Helper::createOrder(cfg.traderId());
 
        std::clog << "\nWarm-up phase to make first calls faster..." << std::endl;
        listener.collectSessionTimeMarks(*bgwSession, order, numberOfMessages, intervalBetweenSending, warmupInterval);
 
        std::clog << "\nMeasurement phase..." << std::endl;
        listener.collectSessionTimeMarks(*bgwSession, order, numberOfMessages, intervalBetweenSending, warmupInterval);
 
        TcpInfo tcpInfo;
        const bool hasTcpInfo = bgwSession->getTcpInfo(tcpInfo);
 
        bgwSession->disconnect();
 
        if(listener.packetGroupingDetected())
            std::clog << "Attention: packet grouping detected! Increase the interval between sending!" << std::endl;
 
        BenchmarkData::reportResults("Latency", listener.sendMarks(), listener.receiveMarks(), numberOfMessages, listener.overhead());
 
        if(hasTcpInfo)
            std::clog << "\n\n" << tcpInfo.toString();
    }
    catch (const std::exception& ex)
    {
        std::cerr << "\nEXCEPTION: " << ex.what() << std::endl;
        return 1;
    }
 
    return 0;
}
 
#include <Common/PerformanceCounter.inc.h>