fix_engine.cpp

#include "fix_engine.h"

#include <netinet/in.h>
#include <netinet/tcp.h>
#include <sys/socket.h>

#include <cerrno>
#include <chrono>
#include <cstdio>
#include <iostream>
#include <stdexcept>
#include <string>
#include <thread>

#include "fix.h"
#include "msg_logon.h"
#include "msg_logout.h"
#include "socketbuf.h"

template <class SessionConfig>
void Acceptor<SessionConfig>::listen() {
    if ((serverSocket = socket(AF_INET, SOCK_STREAM, 0)) == 0) {
        perror("socket failed");
        return;
    }
    struct sockaddr_in serverAddress;

    serverAddress.sin_family = AF_INET;
    serverAddress.sin_addr.s_addr = INADDR_ANY;
    serverAddress.sin_port = htons(port);

    int opt = 1;
    if (setsockopt(serverSocket, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt))) {
        perror("setsockopt SO_REUSEADDR failed");
        return;
    }
    if (setsockopt(serverSocket, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt))) {
        perror("setsockopt SO_REUSEPORT failed");
        return;
    }

    if (bind(serverSocket, (struct sockaddr *)&serverAddress, sizeof(serverAddress)) < 0) {
        std::cerr << "Error binding socket." << std::endl;
        return;
    }

    typedef boost::fibers::buffered_channel<Session<SessionConfig> *> channel_t;
    channel_t chan{2};

    workerThreads = std::max(workerThreads,1);

    std::cout << "starting " << workerThreads << " worker threads\n";
    std::vector<std::thread> workers;

    boost::fibers::barrier b(workerThreads+1);

    for (int i = 0; i < workerThreads; i++) {
        workers.push_back(std::thread(
            [&chan, &b] {
                boost::fibers::use_scheduling_algorithm<boost::fibers::algo::shared_work>(true);
                // wait till all threads joined the shared pool
                b.wait();
                while (true) {
                    Session<SessionConfig> *session;
                    if (chan.pop(session) != boost::fibers::channel_op_status::success) {
                        return;
                    }
                    auto fiber = boost::fibers::fiber(&Session<SessionConfig>::handle, session);
                    fiber.detach();
                }
            }));
    }
    // wait till all threads joined the shared pool
    b.wait();

    std::cout << "starting poller\n";
    std::thread poller_thread([this]() {
        while (true) {
            try {
                poller.poll();
            } catch (std::runtime_error &err) {
                std::cerr << "poller error: " << err.what() << "\n";
                return;
            }
        }
    });

    std::cout << "listening for connections on port " << port << "\n";
    if (::listen(serverSocket, 5) < 0) {  // 5 is the backlog size
        std::cerr << "error listening for connections" << std::endl;
        return;
    }

    while (true) {
        sockaddr_in clientAddr;
        socklen_t clientAddrLen = sizeof(clientAddr);
        char ip_str[INET_ADDRSTRLEN];

        int clientSocket = accept(serverSocket, (struct sockaddr *)&clientAddr, &clientAddrLen);
        if (clientSocket < 0) {
            if(errno==EBADF) break;
            perror("error accepting connection");
            continue;
        }

        int flag = 1;
        if (setsockopt(clientSocket, IPPROTO_TCP, TCP_NODELAY, &flag, sizeof(flag)) < 0) {
            perror("unable to set TCP_NODELAY");
        }

        int flags = fcntl(clientSocket, F_GETFL, 0);
        if (fcntl(clientSocket, F_SETFL, flags | O_NONBLOCK) < 0) {
            perror("unable to set O_NONBLOCK");
            close(clientSocket);
            continue;
        }

        inet_ntop(AF_INET, &(clientAddr.sin_addr), ip_str, INET_ADDRSTRLEN);
        std::cout << "connection from " << ip_str << " port " << ntohs(clientAddr.sin_port) << " on thread " << std::this_thread::get_id() << "\n";
        try {
            auto session = new Session(clientSocket, *this, config);
            chan.push(session);

            onConnected(clientAddr);
            poller.add_socket(clientSocket, session, [](struct kevent &event, void *data) {
                auto session = static_cast<Session<SessionConfig> *>(data);
                if (event.flags & EV_EOF) {
                    close(session->socket);
                }
                session->unpark();
            });
        } catch (std::runtime_error &err) {
            std::cerr << "acceptor refused connection: " << err.what() << "\n";
        }
    }
    // close the sockets for any active sessions
    {
        std::unique_lock<std::shared_mutex> lu(sessionLock);
        for(auto entry : sessionMap) {
            close(entry.second->socket);
            entry.second->unpark();
        }
    }

    // wait for sessions to terminate
    while(true) {
        {
            std::unique_lock<std::shared_mutex> lu(sessionLock);
            if(sessionMap.empty()) break;
        }
        std::this_thread::sleep_for(std::chrono::milliseconds(1000));
    }

    poller.close();
    poller_thread.join();

    chan.close();

    for (auto &worker : workers) worker.join();
}

template <class SessionConfig>
void Session<SessionConfig>::handle() {
    DisconnectHandler disconnectHandler(*this, handler);
    Socketbuf sbuf(socket, *this);
    std::istream is(&sbuf);
    FixMessage msg;
    FixBuilder out;
    try {
        // std::cout << "handling session " << config << " on thread " << std::this_thread::get_id()<<"\n";
        while (true) {
            FixMessage::parse(is, msg, GroupDefs());
            if (is.eof()) {
                return;
            }

            if (!loggedIn && msg.msgType() != Logon::msgType) {
                std::cerr << "rejecting connection, " << msg.msgType() << " is not a Logon\n";
                Logout::build(out, "not logged in");
                sendMessage(Logout::msgType, out);
                return;
            }
            if (msg.seqNum() != config.expectedSeqNum) {
                std::cerr << "rejecting connection, " << msg.seqNum() << " != expected " << config.expectedSeqNum << "\n";
                Logout::build(out, "invalid sequence number");
                sendMessage(Logout::msgType, out);
                return;
            }
            auto targetCompId = msg.getString(Tag::TARGET_COMP_ID);
            if (targetCompId != config.senderCompId) {
                std::cerr << "rejecting connection, invalid target comp id " << targetCompId << ", expected " << config.senderCompId << "\n";
                Logout::build(out, "invalid target comp id");
                sendMessage(Logout::msgType, out);
                return;
            }

            auto senderCompId = msg.getString(Tag::SENDER_COMP_ID);
            if (senderCompId != config.targetCompId) {
                if (config.targetCompId == "*") {
                    config.targetCompId = senderCompId;
                } else {
                    std::cerr << "rejecting connection, invalid sender comp id " << senderCompId << ", expected " << config.targetCompId << "\n";
                    Logout::build(out, "invalid sender comp id");
                    sendMessage(Logout::msgType, out);
                    return;
                }
            }

            if (!loggedIn) {
                if (!handler.validateLogon((msg))) {
                    std::cerr << "logon rejected\n";
                    Logout::build(out, "invalid logon");
                    sendMessage(Logout::msgType, out);
                    return;
                }
                config.initialize(msg);
                Logon::build(out);
                sendMessage(Logon::msgType, out);
                loggedIn = true;
                handler.onLoggedOn(*this);
            }
            handler.onMessage(*this, msg);
            config.expectedSeqNum++;
        }
    } catch (const std::runtime_error &err) {
        std::cerr << "exception processing session: " << config << ", "
                  << err.what() << "\n";
        return;
    }
}

template <class SessionConfig>
void Initiator<SessionConfig>::connect() {
    if ((socket = ::socket(AF_INET, SOCK_STREAM, 0)) == 0) {
        perror("socket failed");
        return;
    }
    std::cout << "connecting...\n";
    if (::connect(socket, (struct sockaddr *)&server, sizeof(server)) < 0) {
        perror("socket connect");
        return;
    }

    int flag = 1;
    if (setsockopt(socket, IPPROTO_TCP, TCP_NODELAY, &flag, sizeof(flag)) < 0) {
        perror("unable to set TCP_NODELAY");
    }

    session = new Session(socket, *this, config);

    if (poller) {
        int flags = fcntl(socket, F_GETFL, 0);
        if (fcntl(socket, F_SETFL, flags | O_NONBLOCK) < 0) {
            perror("unable to set O_NONBLOCK");
            close(socket);
            return;
        }
        poller->add_socket(socket, session, [](struct kevent &event, void *data) {
            auto session = static_cast<Session<SessionConfig> *>(data);
            if (event.flags & EV_EOF) {
                close(session->socket);
            }
            session->unpark();
        });
    }
    connected = true;
    onConnected();
}

template void Acceptor<DefaultSessionConfig>::listen();
template void Initiator<DefaultSessionConfig>::connect();