Benchmark.cpp File Reference

Go to the source code of this file.

Functions
int	main (int argc, char *argv[])

Function Documentation

main()

int main	(	int	argc,
		char *	argv[] )

Definition at line 43 of file Benchmark.cpp.

{
    // `--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= makeUnique<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;
}