Sometimes, it is more effective to perform the event dispatching in a loop in the main application thread. For this purpose, one can use the OnixS::FIX::ThreadingModel::ExternalThread mode.

Stack

Before an externally-managed session is created, the application must create an OnixS::FIX::ISessionReactor instance. There are two reactor classes, which can be used:

TCPStandard::Stack stack;

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

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

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::ISessionReactor::dispatchEvents frequently for each stack that is in use.

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

The OnixS::FIX::ISessionReactor::dispatchEvents method polls for network I/O events, checks FIX send and receive timeouts, and invokes the corresponding callbacks sequentially as events occur. Callbacks for dispatched network events run before dispatch returns. See ExternalThread Event Loop for recording a request in a callback and acting on it afterwards, on the same thread.

Note: The FIX Engine attempts to send a FIX message as soon as OnixS::FIX::Session::send is called, without waiting for OnixS::FIX::ISessionReactor::dispatchEvents. If that attempt fails, the FIX Engine stores the message in the sending queue for the subsequent transmission. The pending data in the sending queue is processed during the OnixS::FIX::ISessionReactor::dispatchEvents call.

Thread Safety

Warning: Access to the stack instance at each moment is possible from only one thread. The TCPStandard/TCPDirect middlewares 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 TCPStandard/TCPDirect stack requires finishing all outstanding work and handling all outstanding events. When the FIX Engine is used with an externally-managed reactor, calling OnixS::FIX::Engine::shutdown() may leave reactor work pending. After shutting down the FIX Engine, the application should continue calling OnixS::FIX::ISessionReactor::dispatchEvents() on the reactor-owning thread until OnixS::FIX::ISessionReactor::isQuiescent() returns true.

This step is required for OnixS::FIX::TCPDirect::Stack and is recommended for the ExternalThread mode in general:

Engine::shutdown();
 
while (!stack.isQuiescent())
    stack.dispatchEvents();

Method OnixS::FIX::ISessionReactor::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.

Using Onload with the ExternalThread Mode

The OnixS::FIX::TCPStandard::Stack calls epoll_wait() with a zero timeout when dispatching events. The "Solarflare Onload User Guide" states that Onload uses interrupt-driven behavior by default:

NOTE: If neither of the spinning options; EF_POLL_USEC and EF_SPIN_USEC are set, Onload will resort to default interrupt driven behavior because the EF_INT_DRIVEN environment variable is enabled by default. NOTE: When EF_POLL_USEC or EF_SPIN_USEC are greater than zero, EF_INT_DRIVEN will be zero.

To turn off interrupt-driven behavior, set either EF_POLL_USEC or EF_SPIN_USEC to a positive value. Also, the EF_UL_EPOLL variable affects the epoll acceleration. We recommend trying different values of the mentioned variables to achieve the best latency results. Please see the "Solarflare Onload User Guide" for details.

ExternalThread Mode for Acceptor Sessions

There is an ability to use the ExternalThread mode for acceptor sessions too. However, to dispatch listening events and accept incoming TCP connections, one must pass the pointer(s) to the stack(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:

TCPStandard::Stack stack;
 
EngineSettings engineSettings;
 
// 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();

Split the Sessions Processing by Different Stacks in Separate Threads

To increase capacity, divide sessions into groups. Each thread creates and owns a separate stack and its sessions, runs its own event loop, and performs bounded application work between dispatch calls. Add cross-thread communication only when application work requires coordination between these groups.

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.
        TCPStandard::Stack 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)));
 
            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();
    }
};

// Initialize engine in the main thread.
EngineSettings engineSettings;
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.

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.
        TCPStandard::Stack stack;
 
        // 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();
    }
};

// Initialize engine in the main thread.
EngineSettings engineSettings;
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();

Using Different Sessions with Different Threading Models

It is possible to use initiator sessions with different threading models. For example, if the Engine is configured to use the default threading model (OnixS::FIX::ThreadingModel::DedicatedThreads), and the reactor instance is passed to the session's constructor, then this particular session will use the OnixS::FIX::ThreadingModel::ExternalThread threading model. All other sessions will use the threading model configured for the whole FIX Engine. However, acceptor sessions can be used in the same threading mode only. This is because an acceptor connection is created when no corresponding acceptor session is identified. Therefore, the threading model is determined by whether the reactor instance(s) is passed or not to the OnixS::FIX::Engine::init() or OnixS::FIX::Engine::addListenPort() methods. If you pass the reactor instance(s) to the OnixS::FIX::Engine::init() or OnixS::FIX::Engine::addListenPort() methods, all acceptor connections and sessions will use the OnixS::FIX::ThreadingModel::ExternalThread threading model. If you do not pass the reactor instance(s) to the OnixS::FIX::Engine::init()or OnixS::FIX::Engine::addListenPort() methods, all acceptor connections and sessions will use the configured for the whole FIX Engine threading model.

See also: The External Thread Sample from the FIX Engine distribution package; Scheduler Threading Models