fix_client_main.cpp

Go to the documentation of this file.

/******************************************************************************
** Copyright © 2019 by J.M.McGuiness, isimud@hussar.me.uk
**
** This library is free software; you can redistribute it and/or
** modify it under the terms of the GNU Lesser General Public
** License as published by the Free Software Foundation; either
** version 2.1 of the License, or (at your option) any later version.
**
** This library is distributed in the hope that it will be useful,
** but WITHOUT ANY WARRANTY; without even the implied warranty of
** MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
** Lesser General Public License for more details.
**
** You should have received a copy of the GNU Lesser General Public
** License along with this library; if not, write to the Free Software
** Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
*/
 
#include "fix_client.hpp"
 
#include "core/jthread.hpp"
#include "core/latency_timestamps.hpp"
 
#include <boost/asio.hpp>
#include <boost/exception/diagnostic_information.hpp>
#include <boost/program_options.hpp>
 
#include <iostream>
 
using namespace libisimud;
using namespace libjmmcg;
 
int main(int argc, char const * const *argv) {
   try {
      using thread_t=libjmmcg::ppd::jthread;
      using fix_client_t=exchanges::FIX::v5_0sp2::fix_client;
 
      boost::program_options::options_description general(
         "A simple client that repeatedly sends a fixed FIX-message to a link listening at a IPADDR:PORT combination. The executable name indicates the message version implemented. For details regarding the properties of the client see the documentation that came with the distribution. Note that Wireshark has a built-in FIX protocol: see the \"Analyze/Enabled Protocols...\" dialog in Wireshark or https://www.wireshark.org/docs/dfref/f/fix.html. Copyright © J.M.McGuiness, isimud@hussar.me.uk. http://libjmmcg.sf.net/ Distributed under the terms of the GPL v2.1.\n"+exit_codes::to_string()+"Arguments"
      );
      general.add_options()
         ("help", "Print this help message.")
         ("version", "Print the build number of this program.")
      ;
      boost::program_options::options_description client("FIX-translator options.");
      client.add_options()
         ("address", boost::program_options::value<boost::asio::ip::address>()->default_value(boost::asio::ip::address_v4::loopback()), "IP address (in v4 format; thus the network interface to which it is boundmo) to which the messages should be sent.")
         ("port", boost::program_options::value<unsigned short>()->required(), "The port to which the messages should be sent.")
         ("processor", boost::program_options::value<unsigned short>()->required()->default_value(exchanges::common::thread_traits::client_to_exchange_thread.core), "The core to which the client should be bound.")
      ;
      boost::program_options::options_description performance("Performance-monitoring options.");
      performance.add_options()
         ("num_loops", boost::program_options::value<unsigned long>()->required()->default_value(20000), "The number of times the fix message should be sent (and therefore received) in each batch.")
         ("loops_for_conv", boost::program_options::value<unsigned short>()->required()->default_value(500), "The maximum number of times the batch should be run.")
         ("perc_conv_estimate", boost::program_options::value<double>()->required()->default_value(2.0), "The maximum value of mean average-deviation to which the repeated runs of the batches should converge.")
      ;
      performance.add_options()
      (
         "num_timings",
         boost::program_options::value<unsigned long>()->required()->default_value(0UL),
         "The number (positive-integer) of latency timings to reserve. This value must be greater than the total number of messages to be processed, otherwise the behaviour is undefined. Upon exit, these values will be written in CSV-format to a file named 'argv[0]+\"-%%%%%.PID.csv\"' (where each '%' is a random hexadecimal digit and PID is the current process id). This file shall be created in the current directory. If the value is zero, then the file shall not be created, no values will be stored, unlimited messages may be processed."
      );
      boost::program_options::options_description all("All options.");
      all.add(general).add(client).add(performance);
      boost::program_options::variables_map vm;
      boost::program_options::store(boost::program_options::parse_command_line(argc, argv, all), vm);
      if (vm.count("help")) {
         std::cout<<all<<std::endl;
         return exit_codes::exit_print_help;
      }
      if (vm.count("version")) {
         std::cout<<LIBJMMCG_DETAILED_VERSION_INFO " Built with FIX: " ISIMUD_FIX_EXCHANGE_VERSION <<std::endl;
         return exit_codes::exit_print_version;
      }
      boost::program_options::notify(vm);
      ppd::api_threading_traits<ppd::platform_api, ppd::sequential_mode>::set_backtrace_on_signal();
      const std::size_t num_timestamps=vm["num_timings"].as<std::size_t>();
      latency_timestamps timestamps(num_timestamps);
      {
         fix_client_t sender(
            vm["address"].as<boost::asio::ip::address>(),
            vm["port"].as<unsigned short>(),
            socket::socket_priority::high,
            vm["processor"].as<unsigned short>(),
            timestamps
         );
         std::clog<<sender<<std::endl;
         std::pair<fix_client_t::timed_results_t, bool> results;
         thread_t collect_stats(
            [&sender, &vm, &results]() {
               results=sender.in_order_tx_rx_stats(
                  vm["num_loops"].as<unsigned long>(),
                  vm["loops_for_conv"].as<unsigned short>(),
                  vm["perc_conv_estimate"].as<double>()
               );
            }
         );
         thread_t::thread_traits::set_kernel_affinity(
            thread_t::thread_traits::get_current_thread(),
            thread_t::thread_traits::api_params_type::processor_mask_type(0)
         );
         thread_t::thread_traits::set_kernel_priority(
            thread_t::thread_traits::get_current_thread(),
            thread_t::thread_traits::api_params_type::priority_type::idle
         );
         while (!sender.signal_status() && collect_stats.is_running()) {
            thread_t::thread_traits::sleep(0);
         }
         std::cout<<"Exchange round-trip in-order time (microseconds)="<<results.first<<(results.second ? " converged." : " NOT converged.")<<std::endl;
         std::clog<<sender<<std::endl;
         thread_t::thread_traits::raise(SIGQUIT);
      }
      timestamps.write_to_csv_file(std::clog, argv[0]);
      return exit_codes::exit_success;
   } catch (std::exception const &ex) {
      std::cerr<<"STL-derived exception. Details: "<<boost::diagnostic_information(ex)<<std::endl;
      return exit_codes::exit_stl_exception;
   } catch (...) {
      std::cerr<<"Unknown exception."<<std::endl;
      return exit_codes::exit_unknown_exception;
   }
   return exit_codes::exit_unknown_failure;
}