OnixS C++ B3 BOE Binary Order Entry  1.2.0
API Documentation
Benchmark Sample

This sample shows how to measure latency and use the warm-up feature.

Using OpenOnload® Network Stack

OpenOnload is part of Solarflare’s suite of network acceleration technologies.

To achieve better results, use the latency profile. For example:

./run-under-onload.sh 54 192.168.28.1 20054 1 2

TCP Loopback Acceleration

The TCP loopback acceleration is turned off by default. It is configured via the environment variables EF_TCP_CLIENT_LOOPBACK and EF_TCP_SERVER_LOOPBACK.

Activate

export EF_TCP_CLIENT_LOOPBACK=1
export EF_TCP_SERVER_LOOPBACK=1

Verify

echo $EF_TCP_CLIENT_LOOPBACK
echo $EF_TCP_SERVER_LOOPBACK

Source code

#if defined(ONIXS_B3_BOE_HAS_GATEWAY_EMULATOR)
#include "../Common/BenchmarkGatewayListener.h"
#endif
#include "../Common/BenchmarkSessionListener.h"
#include "../Common/Helpers.h"
#include "../Common/PerformanceCounter.h"
#include "../Settings/Defaults.h"
#ifndef _WIN32
#include <sched.h> /* header file for POSIX scheduling */
#endif
#include <algorithm>
// #define DEBUGGING
using namespace Samples;
void usage()
{
clog << "Usage: Benchmark [MarketSegmentId] [Host] [Port] (MainThreadAffinity) (ReceivingThreadAffinity) (AuxiliarySendingThreadAffinity)"
" (NumberOfMessages) (IntervalBetweenSendingUsec) (WarmupIntervalUsec)" << endl;
}
int main(int argc, char* argv[])
{
clog << "B3 BOE Benchmark Sample, version " << Session::version() << "." << endl << endl;
#if !defined(NDEBUG) && !defined(DEBUGGING)
cerr << "Please use the RELEASE build to measure latency." << endl;
return 1;
#endif
MarketSegmentID marketSegmentId = DefaultMarketSegmentId;
string host = "127.0.0.1";
Port port = 49152;
bool useEmulator = false;
if (argc < 4)
{
#if defined(ONIXS_B3_BOE_HAS_GATEWAY_EMULATOR)
useEmulator = true;
clog << "WARNING: gateway emulator is used!" << endl;
#else
usage();
return 1;
#endif
}
else
{
marketSegmentId = atoi(argv[1]);
host = argv[2];
port = atoi(argv[3]);
}
const Threading::CpuIndex MainThreadAffinity = argc > 4 ? atoi(argv[4]) : 1u;
const Threading::CpuIndex ReceivingThreadAffinity = argc > 5 ? atoi(argv[5]) : 2u;
const Threading::CpuIndex AuxiliarySendingThreadAffinity = argc > 6 ? atoi(argv[6]) : 3u;
const unsigned int NumberOfMessages = argc > 7 ? atoi(argv[7]) : 300u;
const unsigned int WarmupIntervalUsec = argc > 9 ? atoi(argv[9]) : 10u;
const unsigned int IntervalBetweenSendingUsec = std::max(argc > 8 ? atoi(argv[8]) : 10000u, WarmupIntervalUsec);
#if defined(ONIXS_B3_BOE_HAS_GATEWAY_EMULATOR)
BenchmarkGatewayListener gatewayListener;
std::unique_ptr<GatewayEmulatorThread> gateway;
#endif
int result = 0;
try
{
SignalHelper::manageLinuxSignals();
SessionSettings settings = fillSettings(useEmulator);
settings
.receiveSpinningTimeout(1000*1000)
.useSpinLock(true)
.logBeforeSending(false);
#if defined(ONIXS_B3_BOE_HAS_GATEWAY_EMULATOR)
if(useEmulator)
gateway.reset(new GatewayEmulatorThread(settings.licenseStores(), host, port, &gatewayListener));
#endif
class BenchmarkListener ONIXS_B3_BOE_FINAL : public BenchmarkSessionListener
{
public:
explicit BenchmarkListener(size_t numberOfMessages)
: BenchmarkSessionListener(numberOfMessages)
, receivedAllMessages_(0)
{}
void onError(SessionErrorReason::Enum reason, const std::string & description, Session * session, Messaging::SbeMessage msg) ONIXS_B3_BOE_OVERRIDE {
BenchmarkSessionListener::onError(reason, description, session, msg);
receivedAllMessages_.release();
}
void collectSessionTimeMarks(Session & session, Samples::NewOrderSingle & order,
unsigned int numberOfMessages, unsigned int sendPeriodUsec, unsigned int warmupPeriodUsec) {
resetBenchmark();
// When the session is used with the OnixS C++ B3 Market Data Handler,
// this value can be taken from the packet receiving time
Messaging::Timestamp sendingTime = UtcWatch::now();
for(UInt32 messageCounter = 0; messageCounter < numberOfMessages; ++messageCounter) {
order->setOrderQty(messageCounter + 1);
order->setClOrdId(messageCounter);
beforeSending();
session.send(order, sendingTime);
afterSending();
sendingTime = UtcWatch::now();
if(warmupPeriodUsec) {
const PerformanceCounter::Count start = PerformanceCounter::current();
do {
ThisThread::spinWait(warmupPeriodUsec);
session.warmUp(order, sendingTime);
} while(PerformanceCounter::elapsedMicroseconds(start) < sendPeriodUsec);
}
else if(sendPeriodUsec) {
ThisThread::spinWait(sendPeriodUsec);
}
}
waitUntilReceivedAllMessages();
}
private:
void onReceivedAllMessages() ONIXS_B3_BOE_OVERRIDE {
BenchmarkSessionListener::onReceivedAllMessages();
receivedAllMessages_.release();
}
void waitUntilReceivedAllMessages() {
receivedAllMessages_.acquire();
}
Semaphore receivedAllMessages_;
}
listener(NumberOfMessages);
#if defined(DEBUGGING)
const SessionStorageType::Enum storageType = SessionStorageType::FileBased;
#else
const SessionStorageType::Enum storageType = SessionStorageType::MemoryBased;
#endif
clog << "Parameters: NumberOfMessages=" << NumberOfMessages << "; IntervalBetweenSendingUsec=" << IntervalBetweenSendingUsec
<< "; warmupIntervalUsec=" << WarmupIntervalUsec
<< "; StorageType=" << SessionStorageType::toString(storageType) << '.' << endl;
Session session(settings, &listener, storageType);
ThisThread::affinity(MainThreadAffinity);
setPriorityAndPolicy(&session);
listener.measureOverhead();
session
.receivingThreadAffinity(ReceivingThreadAffinity)
.sendingThreadAffinity(AuxiliarySendingThreadAffinity)
.connect(host, port);
NewOrderSingle order;
Helper::setOrderFields(order, marketSegmentId, DefaultSecurityId, DefaultAccount, DefaultPriceMantissa);
clog << "\nWarm-up phase to make first calls faster..." << endl;
listener.collectSessionTimeMarks(session, order, NumberOfMessages, IntervalBetweenSendingUsec, WarmupIntervalUsec);
clog << "\nMeasurement phase..." << endl;
listener.collectSessionTimeMarks(session, order, NumberOfMessages, IntervalBetweenSendingUsec, WarmupIntervalUsec);
TcpInfo tcpInfo;
const bool hasTcpInfo = session.getTcpInfo(tcpInfo);
session.disconnect();
if(listener.packetGroupingDetected())
clog << "Attention: packet grouping detected! Increase the interval between sending!" << endl;
BenchmarkData::reportResults("Latency", listener.sendMarks(), listener.receiveMarks(), NumberOfMessages, listener.overhead());
if(hasTcpInfo)
clog << "\n\n" << tcpInfo.toString();
#ifndef _WIN32
ThisThread::policy(SCHED_OTHER);
#endif
}
catch (const exception& ex)
{
cerr << "\nEXCEPTION: " << ex.what() << endl;
result = 1;
}
#if defined(ONIXS_B3_BOE_HAS_GATEWAY_EMULATOR)
if(gateway) {
try {
gateway->wait();
}
catch(...) {
// Emulator exception (if any) is already reported from Emulator's thread
}
}
#endif
return result;
}
#include "../Common/PerformanceCounter.inc.h"