Benchmark Sample

Measures the latency from the network socket read to the user callback. Logging is turned off to avoid affecting the results, a FeedListener timestamps each multicast message block, and the collected latencies are reported (min / median) and saved to a CSV file. It also shows where to apply CPU affinity for the market data processing threads.

Source code

#include "../Common/Defaults.h"
#include "../Common/ServiceEventTracer.h"
#include "../Common/Utils.h"
 
#include <OnixS/ICE/iMpact/MarketData.h>
 
#include <algorithm>
#include <cstdlib>
#include <fstream>
#include <stdexcept>
 
using namespace OnixS::ICE::iMpact::MarketData;
 
namespace {
 
class MyListener : public FeedListener
{
    static const unsigned MAX_SAMPLES_COUNT = 100000;
 
public:
    MyListener(Handler& handler)
      : handler_(handler)
    {
        subscribe(this);
 
        latencies_.reserve(MAX_SAMPLES_COUNT);
    }

    virtual ~MyListener()
    {
        try
        {
            if (handler_.active())
            {
                handler_.stop();
            }
 
            // Unsubscribe from subscribed events to do not
            // continue receive events after disposing.
            subscribe(0);
        }
        catch (...)
        {
            std::cerr << "Exception while unsubscribing." << std::endl;
        }
    }
 
    void onFeedStarted(FeedId /*feedId*/, const MarketIds& /*ids*/) /* override */ {}
 
    void onFeedStopped(FeedId /*feedId*/) /* override */ {}
 
    void onSessionNumberChange(FeedId /*feedId*/, short /*oldSessionNumber*/, short /*newSessionNumber*/) {}
 
    void onMulticastMessageBlockEnd(FeedId /*feedId*/, const MarketIds& /*marketIds*/) /* override */ {}

    void onMulticastMessageBlockBegin(FeedId /*feedId*/, const MessageInfo& msgInfo) /* override */
    {
        if (msgInfo.isMessageReceivedDuringSnapshot || latencies_.size() >= MAX_SAMPLES_COUNT)
        {
            return;
        }
 
        latencies_.push_back(Timestamp::utcNow() - msgInfo.receiveTime);
    }

    void processLatencies()
    {
        if (latencies_.empty())
        {
            std::cout << "Nothing to process (latencies list is empty)." << std::endl;
            return;
        }
 
        std::size_t medianPosition = latencies_.size() / 2;
        std::nth_element(latencies_.begin(), latencies_.begin() + medianPosition, latencies_.end());
        TimeSpan medianLatency = latencies_.at(medianPosition);
        TimeSpan minLatency = *std::min_element(latencies_.begin(), latencies_.end());
 
        std::cout << "Latency from network socket read to user callback (nanoseconds):\n"
                     "Min: "
                  << minLatency.nanoseconds()
                  << " ns\n"
                     "Median: "
                  << medianLatency.nanoseconds()
                  << " ns\n"
                     "Count: "
                  << latencies_.size() << " multicast message blocks" << std::endl;
    }

    void saveLatencies(const std::string& path)
    {
        std::ofstream csvFile;
 
        csvFile.open(path.c_str());
 
        if (csvFile.good())
        {
            for (Latencies::const_iterator it = latencies_.begin(); it != latencies_.end(); ++it)
            {
                csvFile << it->nanoseconds() << ";\n";
            }
        }
        else
        {
            std::string errorReason;
 
            errorReason += "Cannot open file to output benchmarking statistics [filename=";
            errorReason += path;
            errorReason += "]. ";
 
            throw std::domain_error(errorReason);
        }
 
        csvFile.close();
    }
 
private:
    void subscribe(MyListener* listener)
    {
        handler_.registerFeedListener(listener);
    }
 
private:
    typedef std::vector<TimeSpan> Latencies;
 
    Handler& handler_;
    Latencies latencies_;
};
 
} // namespace
 
int main(int /*argc*/, char** /*argv*/)
{
    try
    {
        introduceMyself("Benchmark");
 
        FeedEngineSettings feedEngineSettings;
        FeedEngine feedEngine(feedEngineSettings);
 
        HandlerSettings handlerSettings;
 
        // For most of Linux operating systems it would be
        // better to define following parameter to be able to
        // successfully receive market data from multicast feeds.
        handlerSettings.networkInterface = NETWORK_INTERFACE;
 
        // Turns off logging to speed-up processing.
        handlerSettings.logLevel = LogLevels::Error;
        handlerSettings.advancedLogOptions = AdvancedLogOptions::LogNothing;
 
        // By default, license is stored in lib folder of the distribute
        // package. Instead of copying, let's use original file.
        handlerSettings.licenseDirectory = LICENSE_DIRECTORY;
 
        // Primary settings - connectivity configuration.
        handlerSettings.connectivityConfiguration = PF1_C10Y_FILE;
 
        // Defining CPU affinity for market data processing
        // threads may greatly speed-up overall processing.
        // handlerSettings.processingThreadAffinity.insert(2);
 
        // Constructs handler with custom license root and enabled logging.
        Handler handler(handlerSettings);
 
        // Binds the handler to the feed engine.
        handler.bindFeedEngine(feedEngine);
 
        // Traces all service events coming from handlers.
        ServiceEventTracer serviceEventTracer;
 
        serviceEventTracer.attachTo(handler);
 
        // Attaching event listeners to the Handler.
        MyListener myListener(handler);
 
        // Making subscription.
        MarketSubscriptions subscriptions;
 
        MarketSubscription brentFutures(
            KnownMarketTypes::IpeBrentFutures,
            MarketSubType::NonImplied,
            SecurityType::Futures,
            BookDepth::FOD
        );
 
        subscriptions.insert(brentFutures);
 
        // Now starts the subscription.
        handler.start(subscriptions);
 
        std::cout << "\nPress ENTER to stop Handler, analyze statistics and exit Sample." << std::endl;
 
        readUserResponse();
 
        std::cout << "Stopping Handler..." << std::endl;
 
        // Stops multicast data processing.
        handler.stop();
 
        std::cout << "Handler is stopped." << std::endl;

        myListener.processLatencies();

        myListener.saveLatencies("results.csv");
 
        std::cout << "Done." << std::endl;
    }
    catch (const std::exception& ex)
    {
        std::cout << "Error: " << ex.what() << std::endl;
        return EXIT_FAILURE;
    }
    catch (...)
    {
        std::cout << "Unknown exception." << std::endl;
        return EXIT_FAILURE;
    }
 
    return EXIT_SUCCESS;
}