Solarflare Specific Features

Solarflare Onload

Solarflare Onload is a sockets acceleration technology for the transparent acceleration of sockets-based applications.

On Linux, the FIX Engine uses Solarflare Onload Extensions API to support additional features of Solarflare network cards. If you use such network cards, in your system, you can get an additional benefit from advanced features of such network cards.

FIX Engine automatically detects if the corresponding Onload library installed in your system and if so, then loads it in run-time and provides an ability to use additional features. In this case, there should be the record, in the FIX Engine log, that the corresponding version of the Onload extension library is used.

Note

If you want to test Onload features on the loopback network interface, then EF_TCP_CLIENT_LOOPBACK and EF_TCP_SERVER_LOOPBACK parameters should be used, e.g. :

EF_TCP_CLIENT_LOOPBACK=1 EF_TCP_SERVER_LOOPBACK=1 onload ..

Please see the "Onload User Guide".

Currently, the FIX Engine supports the following Onload features:

• Onload message warmup

When you use the OnixS::FIX::Session::warmUp method, the FIX Engine automatically enables this feature if possible. As a result, you can get the maximal warmup effect including the complete warm-up of the sending path of the TCP stack.

The onload_stackdump utility could be used to monitor the usage of this feature.

For example:

onload_stackdump lots | grep warm

• Zero-Copy TCP send

In order to use this feature you need to use the overloaded OnixS::FIX::Session::send(FlatMessage * msg, SessionSendMode::Enum mode) method with the OnixS::FIX::SessionSendMode::OnloadZeroCopy mode. As a result, you can avoid the additional data copying from an application allocated buffer to the network adapter buffer and can decrease the sending latency.

This feature also is supported when one is sending messages in a batch. In this case, you need to use the overloaded OnixS::FIX::Session::send(FlatMessageBatch & msgs, SessionSendMode::Enum mode, size_t maxPacketSize) method.

Note

If a Solarflare network card is not present in the system and you use this feature then the usual socket send call will be performed.

Note

The Zero-Copy feature requires the allocation of specific buffers from NIC. After each successful send, the corresponding buffers become invalid, so it needs to reallocate them. This reallocation is performed right after the send method is completed. This adds additional latency to the OnixS::FIX::Session::send overall latency.

Solarflare TCPDirect

Solarflare's TCPDirect is a highly accelerated network middleware. It is similar to Onload but provides lower latency.

Stack

Before a TCPDirect-enabled session is created, the application must create an OnixS::FIX::TCPDirect::Stack instance.

 
TCPDirect::Attributes attributes;
// The default interface name could be provided by environment variable
//     ZF_ATTR="interface=<interface-name>", see TCPDirect specs,
//
// or could be set manually:
//     attr.set("interface", "interface-name");
 
TCPDirect::Stack stack(attributes);
 

Attributes (OnixS::FIX::TCPDirect::Attributes) control the configuration of the stack and its behavior. Please see the "Solarflare TCPDirect User Guide" for a complete list of available attributes and their description.

Note

The application should create TCPDirect-enabled sessions in the same thread where it has created the corresponding OnixS::FIX::TCPDirect::Stack instance.

The pointer to the stack is provided in Session's constructor, for example:

 
SessionListener listener;
 
Session session(&stack, SenderCompId, TargetCompId, FixProtocolVersion, &listener);
 

By default, each TCPDirect stack can handle up to 64 TCP endpoints, so this is the maximum number of sessions that could be created using the same stack instance.

The max_tcp_endpoints attribute can set the maximum number of TCP endpoints. Please refer to Solarflare TCPDirect documentation.

See also max_tcp_listen_endpoints, max_tcp_syn_backlog, max_udp_rx_endpoints, max_udp_tx_endpoints attributes.

These attributes could be set to a minimum value to save hardware and software resources.

Note

The TCPDirect does not support loopback connections, so it needs to test it between different machines or NICs.

Note

The TCPDirect requires the allocation of huge pages. Please see the "Solarflare TCPDirect User Guide" for details. Huge pages are generally enabled by default. One can use, e.g., the following command to check the current state of huge pages:

cat /sys/kernel/mm/transparent_hugepage/enabled

Asynchronous Logon and Logout

Applications must call asynchronous Session's methods to logon (OnixS::FIX::Session::logonAsInitiatorAsync) and logout (OnixS::FIX::Session::logoutAsync).

session.logonAsInitiatorAsync(CounterpartyHost, CounterpartyPort, HeartBtInt, nullptr, true);
 
// Dispatching events..
 
session.logoutAsync("End of connection");

Dispatching Network Events

Applications must call OnixS::FIX::TCPDirect::Stack::dispatchEvents frequently for each stack that is in use.

bool finished = false;
 
while (!finished)
    stack.dispatchEvents();
 

Thread Safety

Warning

Access to the stack instance at each moment is possible from only one thread. The TCPDirect middleware does not use inter-threaded synchronization, does not serialize calls, and does not check the correctness of access to its API by the user. Therefore, the event dispatching and message sending should be performed from the same thread.

Shutdown

The TCPDirect stack requires finishing all outstanding work and handling all outstanding events. Before destroying the OnixS::FIX::TCPDirect::Stack instance, the following code should be executed:

 
while (!stack.isQuiescent())
    stack.dispatchEvents();
 

Method OnixS::FIX::TCPDirect::Stack::isQuiescent returns a boolean value indicating whether a stack is quiescent.

This can be used to ensure that all connections have been closed gracefully before destroying a stack (or exiting the application). Destroying a stack while it is not quiescent is permitted by the API, but when doing so there is no guarantee that sent data has been acknowledged by the peer or even transmitted, and there is the possibility that peers' connections will be reset.

TCPDirect for Acceptor Sessions

There is an ability to use the TCPDirect mode for acceptor sessions too. However, to dispatch listening events and accept incoming TCP connections, one must pass the pointer(s) to the OnixS::FIX::TCPDirect::Stack object(s) to the OnixS::FIX::Engine::init() or OnixS::FIX::Engine::addListenPort() methods. The pointer to the same stack (if a few stack pointers are passed, any of them) should also be passed to the acceptor session's constructor. In this case, the FIX Engine does not create an additional thread to listen for incoming connections and uses the corresponding stack(s) to dispatch these events:

LocalNetworkInterfaces ifaces;
ifaces.push_back(ListenAddress);
 
EngineSettings engineSettings;
 
// Set listen local network interface, which corresponds to Solarflare's NIC.
engineSettings.localNetworkInterface(ifaces);
// Set the listen port.
engineSettings.listenPort(ListenPort);
 
TCPDirect::Attributes attributes;
// Set the interface name, which corresponds to Solarflare's NIC.
attributes.networkInterface(NetworkInterfaceName);
 
TCPDirect::Stack stack(attributes);
 
// Invoke the Engine::init with the given stack class instance.
Engine::init(engineSettings, &stack);
 
SessionListener listener;
 
Session acceptor(&stack, SenderCompId, TargetCompId, FixProtocolVersion, &listener);
 
acceptor.logonAsAcceptor();
 
bool finished = false;
 
// Dispatching events..
while (!finished)
    stack.dispatchEvents();
 
acceptor.shutdown();
Engine::shutdown();
 
while (!stack.isQuiescent())
    stack.dispatchEvents();

The TDPDirect limits the number of listen ports by the max_tcp_listen_endpoints attribute and acceptor sessions by the max_tcp_endpoints attribute per one stack. To overcome this limit, one can use more OnixS::FIX::TCPDirect::Stack objects. However, different reactors cannot process the same listen port(s). Therefore, each reactor and corresponding acceptor session(s) should use different listen port(s):

LocalNetworkInterfaces ifaces;
ifaces.push_back(ListenAddress);
 
EngineSettings engineSettings;
 
// Set listen local network interface, which corresponds to Solarflare's NIC.
engineSettings.localNetworkInterface(ifaces);
// Set listen ports.
ListenPorts listenPorts;
listenPorts.push_back(4200);
listenPorts.push_back(4201);
engineSettings.listenPorts(listenPorts);
 
TCPDirect::Attributes attributes;
// Set the interface name, which corresponds to Solarflare's NIC.
attributes.networkInterface(NetworkInterfaceName);
 
TCPDirect::Stack stack1(attributes);
TCPDirect::Stack stack2(attributes);
 
ISessionReactors reactors;
reactors.push_back(&stack1);
reactors.push_back(&stack2);
 
// Invoke the Engine::init with given stack class instances.
 
Engine::init(engineSettings, reactors);
 
SessionListener listener;
 
// Pass any of current stacks to the acceptor session's constructor.
Session acceptor(&stack1, SenderCompId, TargetCompId, FixProtocolVersion, &listener);
 
acceptor.logonAsAcceptor();
 
bool finished = false;
 
// Dispatching events..
while (!finished)
{
    stack1.dispatchEvents();
    stack2.dispatchEvents();
}
 
acceptor.shutdown();
Engine::shutdown();
 
while (!stack1.isQuiescent() || !stack2.isQuiescent())
{
    stack1.dispatchEvents();
    stack2.dispatchEvents();
}

Split the Sessions Processing by Different Stacks in Separate Threads

When the amount of high-loaded sessions is large, splitting sessions into groups and performing the event dispatching in separate threads makes sense and can improve performance. In this case, considering that the external thread reactors are not thread safe, each stack with corresponding sessions should be created in its thread.

Split the Initiator Sessions Processing by Different Stacks in Separate Threads

class InitiatorSessionReactorThread : public OnixS::Threading::Thread
{
    using Sessions = std::vector<std::unique_ptr<Session>>;
 
    size_t sessionNumber_;
    size_t firstSessionIndex_;
 
public:
 
    InitiatorSessionReactorThread(size_t sessionNumber, size_t firstSessionIndex) :
        OnixS::Threading::Thread("SessionReactorThread"),
        sessionNumber_(sessionNumber),
        firstSessionIndex_(firstSessionIndex)
    {}
 
    ~InitiatorSessionReactorThread() override
    {
        join();
    }
 
protected:
 
    void run() override
    {
        // Initialize stack.
        TCPDirect::Attributes attributes;
        // Set the interface name, which corresponds to Solarflare's NIC.
        attributes.networkInterface(NetworkInterfaceName);
 
        TCPDirect::Stack stack(attributes);
 
        // Initialize sessions.
        SessionListener listener;
        Sessions sessions;
 
        for (size_t sessionCounter = 0; sessionCounter < sessionNumber_ && !stopRequested(); ++sessionCounter)
        {
            const std::string sessionId = std::to_string(firstSessionIndex_ + sessionCounter);
            sessions.push_back(std::unique_ptr<Session>(new Session(
                &stack,
                SessionSenderCompIdPrefix + sessionId,
                SessionTargetCompIdPrefix + sessionId,
                ProtocolVersion::FIX_40,
                &listener)));
 
            OnixS::Threading::SharedFuture<void> logonTask = sessions.back()->logonAsInitiatorAsync(CounterpartyHost, CounterpartyPort, HeartBtInt, nullptr, true);
            while (!logonTask.isReady() && !stopRequested())
                stack.dispatchEvents();
        }
 
        // Useful work and dispatching events...
        while (!stopRequested())
            stack.dispatchEvents();
 
        // Shutdown sessions.
        for (const std::unique_ptr<Session> & session : sessions)
        {
            OnixS::Threading::SharedFuture<void> logoutTask = session->logoutAsync("The session is disconnected");
 
            while (!logoutTask.isReady())
                stack.dispatchEvents();
 
            session->shutdown();
        }
 
        while (!stack.isQuiescent())
            stack.dispatchEvents();
    }
};

LocalNetworkInterfaces ifaces;
ifaces.push_back(ListenAddress);
 
// Initialize engine in the main thread.
EngineSettings engineSettings;
// Set listen local network interface, which corresponds to Solarflare's NIC.
engineSettings.localNetworkInterface(ifaces);
Engine::init(engineSettings);
 
// Initialize and start the required amount of threads.
constexpr size_t SessionNumber = 25;
constexpr size_t SessionPerThread = 5;
constexpr size_t ThreadNumber = SessionNumber / SessionPerThread;
 
using ReactorThreads = std::vector<std::unique_ptr<InitiatorSessionReactorThread>>;
 
// Destroy and join all reactor threads before shutting down the Engine.
{
    ReactorThreads reactorThreads;
 
    for(size_t reactorThreadCounter = 0; reactorThreadCounter < ThreadNumber; ++reactorThreadCounter)
    {
        const size_t firstSessionIndex = reactorThreadCounter * SessionPerThread;
 
        reactorThreads.push_back(std::unique_ptr<InitiatorSessionReactorThread>(
            new InitiatorSessionReactorThread(SessionPerThread, firstSessionIndex)));
        reactorThreads.back()->start();
    }
 
    // Perform useful work, then request worker completion.
    for(const std::unique_ptr<InitiatorSessionReactorThread> & reactorThread : reactorThreads)
        reactorThread->stopAsync();
}
 
// Shutdown the FIX Engine.
Engine::shutdown();

Split the Acceptor Sessions Processing by Different Stacks in Separate Threads

To split acceptor Sessions across separate threads, register a listen endpoint and its reactor for each thread using OnixS::FIX::Engine::addListenPort() or OnixS::FIX::Engine::addListenNetworkInterfaceAndPort().

Note

Each reactor must own a distinct listen endpoint, because the same port on the same interface cannot be processed by more than one reactor.

Use OnixS::FIX::Engine::addListenPort() to listen on a port using the default local network interface (ANY) or the one set by OnixS::FIX::EngineSettings::localNetworkInterface(), giving each thread a different port. Use OnixS::FIX::Engine::addListenNetworkInterfaceAndPort() to listen on an explicit interface-and-port endpoint (for example, 127.0.0.1:5000); this is required when several threads must listen on the same port across different local interfaces. For TCPDirect, the listen interface must be a local IPv4 address assigned to the Solarflare NIC; loopback or ANY is not applicable.

Note

In ExternalThread mode, a caller-owned listen endpoint must be added and removed on its owning thread. When a thread finishes, remove its listen endpoint and the associated reactor using OnixS::FIX::Engine::removeListenPort() or OnixS::FIX::Engine::removeListenNetworkInterfaceAndPort().

The following example uses a distinct listen port for each reactor. Applications that instead use explicit interfaces must supply actual local interface-and-port endpoints from their configuration.

class AcceptorSessionReactorThread : public OnixS::Threading::Thread
{
    using Sessions = std::vector<std::unique_ptr<Session>>;
 
    size_t sessionNumber_;
    size_t firstSessionIndex_;
    int listenPort_;
 
public:
 
    AcceptorSessionReactorThread(size_t sessionNumber, size_t firstSessionIndex, int listenPort):
        OnixS::Threading::Thread("SessionReactorThread"),
        sessionNumber_(sessionNumber),
        firstSessionIndex_(firstSessionIndex),
        listenPort_(listenPort)
    {}
 
    ~AcceptorSessionReactorThread() override
    {
        join();
    }
 
protected:
 
    void run() override
    {
        // Initialize stack.
        TCPDirect::Attributes attributes;
        // Set the interface name, which corresponds to Solarflare's NIC.
        attributes.networkInterface(NetworkInterfaceName);
 
        TCPDirect::Stack stack(attributes);
 
        // Assign every reactor a distinct listen port on the configured local interface.
        // The stack processes listening and session IO only for its own port.
        Engine::instance()->addListenPort(listenPort_, &stack);
 
        // Initialize sessions.
        SessionListener listener;
        Sessions sessions;
 
        for (size_t sessionCounter = 0; sessionCounter < sessionNumber_ && !stopRequested(); ++sessionCounter)
        {
            const std::string sessionId = std::to_string(firstSessionIndex_ + sessionCounter);
            sessions.push_back(std::unique_ptr<Session>(new Session(
                &stack,
                SessionSenderCompIdPrefix + sessionId,
                SessionTargetCompIdPrefix + sessionId,
                ProtocolVersion::FIX_40,
                &listener)));
            sessions.back()->logonAsAcceptor();
        }
 
        // Useful work and dispatching events...
        while (!stopRequested())
            stack.dispatchEvents();
 
        // Disconnect sessions.
        for (const std::unique_ptr<Session> & session : sessions)
        {
            OnixS::Threading::SharedFuture<void> logoutTask = session->logoutAsync("The session is disconnected");
 
            while (!logoutTask.isReady())
                stack.dispatchEvents();
        }
 
        // Remove the listen port while its reactor is still owned by this thread.
        Engine::instance()->removeListenPort(listenPort_);
 
        // Shutdown sessions.
        for (const std::unique_ptr<Session> & session : sessions)
            session->shutdown();
 
        while (!stack.isQuiescent())
            stack.dispatchEvents();
    }
};

LocalNetworkInterfaces ifaces;
ifaces.push_back(ListenAddress);
 
// Initialize engine in the main thread.
EngineSettings engineSettings;
// Set listen local network interface, which corresponds to Solarflare's NIC.
engineSettings.localNetworkInterface(ifaces);
Engine::init(engineSettings);
 
// Initialize and start the required amount of threads.
constexpr size_t SessionNumber = 25;
constexpr size_t SessionPerThread = 5;
constexpr size_t ThreadNumber = SessionNumber / SessionPerThread;
using ReactorThreads = std::vector<std::unique_ptr<AcceptorSessionReactorThread>>;
 
// Destroy and join all reactor threads before shutting down the Engine.
{
    ReactorThreads reactorThreads;
 
    for(size_t reactorThreadCounter = 0; reactorThreadCounter < ThreadNumber; ++reactorThreadCounter)
    {
        const size_t firstSessionIndex = reactorThreadCounter * SessionPerThread;
        const int listenPort = FirstThreadListenPort + static_cast<int>(reactorThreadCounter);
 
        reactorThreads.push_back(std::unique_ptr<AcceptorSessionReactorThread>(
            new AcceptorSessionReactorThread(SessionPerThread, firstSessionIndex, listenPort)));
        reactorThreads.back()->start();
    }
 
    // Perform useful work, then request worker completion.
    for(const std::unique_ptr<AcceptorSessionReactorThread> & reactorThread : reactorThreads)
        reactorThread->stopAsync();
}
 
// Shutdown the FIX Engine.
Engine::shutdown();

Note

It needs explicitly configure the listen local network interface, which corresponds to Solarflare's NIC. This can be done by the OnixS::FIX::EngineSettings::localNetworkInterface() setting.

See also

The TCPDirect Buy Side Sample and TCPDirect Sell Side Sample from the FIX Engine distribution package

"Solarflare TCPDirect User Guide"