socket_server_impl.hpp

Go to the documentation of this file.

/******************************************************************************
** Copyright © 2015 by J.M.McGuiness, coder@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
*/
 
namespace jmmcg { namespace LIBJMMCG_VER_NAMESPACE { namespace socket {
 
template<class ProcessingRules, class SvrMgr>
template<class LatencyTimestamps, class>
inline
svr<ProcessingRules, SvrMgr>::svr(boost::asio::ip::address const &addr, unsigned short port_num, unsigned short timeout, typename svr_mgr_t::socket_priority priority, thread_t::thread_traits::api_params_type::processor_mask_type mask, const thread_t::thread_traits::api_params_type::priority_type cpu_priority, proc_rules_t const &proc_ops, LatencyTimestamps &ts, char const *svr_name, server_to_client_flow_t &&server_to_client_flow) noexcept(false)
:  base_t(),
   address(addr),
   port_number(port_num),
   processor(proc_ops),
   manager(
      address,
      port_number,
      proc_rules_t::src_msg_details_t::min_msg_size,
      proc_rules_t::src_msg_details_t::max_msg_size,
      timeout,
      priority,
      mask.get_cpu(),
      std::move(server_to_client_flow)
   ),
   io_thread(
      [this, &ts]() {
         manager.start_accept(
            [this, &ts](auto &src_cxn, auto &dest_skt) {
               return this->read_and_process_msgs(src_cxn, dest_skt, ts);
            }
         );
         // This must come after the accept otherwise starting subsequent servers will fail.
         return std::bind(&svr::run, this);
      }()
   ) {
   io_thread.kernel_priority(cpu_priority);
   io_thread.kernel_affinity(mask);
   io_thread.set_name(svr_name);
}
 
template<class ProcessingRules, class SvrMgr>
template<class LatencyTimestamps, class>
inline
svr<ProcessingRules, SvrMgr>::svr(boost::asio::ip::address const &addr, unsigned short port_num, socket_priority priority, thread_t::thread_traits::api_params_type::processor_mask_type mask, const thread_t::thread_traits::api_params_type::priority_type cpu_priority, proc_rules_t const &proc_ops, LatencyTimestamps &ts, char const *svr_name, server_to_client_flow_t &&server_to_client_flow) noexcept(false)
:  svr(addr, port_num, (svr_mgr_t::heartbeats_t::max_missed_heartbeats*svr_mgr_t::heartbeats_t::heartbeat_interval).count(), priority, mask, cpu_priority, proc_ops, ts, svr_name, std::move(server_to_client_flow)) {
}
 
template<class ProcessingRules, class SvrMgr>
template<class LatencyTimestamps, class>
inline
svr<ProcessingRules, SvrMgr>::svr(boost::asio::ip::address const &addr, unsigned short port_num, unsigned short timeout, socket_priority priority, thread_t::thread_traits::api_params_type::processor_mask_type mask, const thread_t::thread_traits::api_params_type::priority_type cpu_priority, proc_rules_t const &proc_ops, socket_t &dest_skt, LatencyTimestamps &ts, char const *svr_name, server_to_client_flow_t &&server_to_client_flow) noexcept(false)
:  base_t(),
   address(addr),
   port_number(port_num),
   processor(proc_ops),
   manager(
      address,
      port_number,
      proc_rules_t::src_msg_details_t::min_msg_size,
      proc_rules_t::src_msg_details_t::max_msg_size,
      timeout,
      priority,
      mask.get_cpu(),
      std::move(server_to_client_flow),
      dest_skt
   ),
   io_thread(
      [this, &ts]() {
         manager.start_accept(
            [this, &ts](auto &src_cxn, auto &dest_skt) {
               return this->read_and_process_msgs(src_cxn, dest_skt, ts);
            }
         );
         // This must come after the accept otherwise starting subsequent servers will fail.
         return std::bind(&svr::run, this);
      }()
   ) {
   io_thread.kernel_priority(cpu_priority);
   io_thread.kernel_affinity(mask);
   io_thread.set_name(svr_name);
}
 
template<class ProcessingRules, class SvrMgr>
template<class LatencyTimestamps, class>
inline
svr<ProcessingRules, SvrMgr>::svr(boost::asio::ip::address const &addr, unsigned short port_num, socket_priority priority, thread_t::thread_traits::api_params_type::processor_mask_type mask, const thread_t::thread_traits::api_params_type::priority_type cpu_priority, proc_rules_t const &proc_ops, socket_t &dest_skt, LatencyTimestamps &ts, server_to_client_flow_t &&server_to_client_flow) noexcept(false)
:  svr(addr, port_num, (svr_mgr_t::heartbeats_t::max_missed_heartbeats*svr_mgr_t::heartbeats_t::heartbeat_interval).count(), priority, mask, cpu_priority, proc_ops, dest_skt, ts, std::move(server_to_client_flow)) {
}
 
template<class ProcessingRules, class SvrMgr>
template<class LatencyTimestamps>
inline
svr<ProcessingRules, SvrMgr>::svr(ctor_args const &args, socket_priority priority, thread_t::thread_traits::api_params_type::processor_mask_type mask, const thread_t::thread_traits::api_params_type::priority_type cpu_priority, LatencyTimestamps &ts, server_to_client_flow_t &&server_to_client_flow) noexcept(false)
: svr(args.addr, args.port, priority, mask, cpu_priority, args.proc_ops, ts, std::move(server_to_client_flow)) {
}
 
template<class ProcessingRules, class SvrMgr>
template<class LatencyTimestamps>
inline
svr<ProcessingRules, SvrMgr>::svr(ctor_args const &args, socket_t &dest_skt, unsigned short timeout, socket_priority priority, thread_t::thread_traits::api_params_type::processor_mask_type mask, const thread_t::thread_traits::api_params_type::priority_type cpu_priority, LatencyTimestamps &ts, char const *svr_name, server_to_client_flow_t &&server_to_client_flow) noexcept(false)
: svr(args.addr, args.port, timeout, priority, mask, cpu_priority, args.proc_ops, dest_skt, ts, svr_name, std::move(server_to_client_flow)) {
}
 
template<class ProcessingRules, class SvrMgr> inline
svr<ProcessingRules, SvrMgr>::~svr() noexcept(true) {
   stop();
}
 
template<class ProcessingRules, class SvrMgr> inline void
svr<ProcessingRules, SvrMgr>::stop() noexcept(true) {
   exit_=true;
   manager.stop();
}
 
template<class ProcessingRules, class SvrMgr> inline int
svr<ProcessingRules, SvrMgr>::main(int argc, char const * const *argv) noexcept(true) {
   using thread_traits=thread_t::thread_traits;
 
   try {
      boost::program_options::options_description desc(
         "A simple exchange-simulator that listens to an IPADDR:PORT combination. The executable name indicates the message version implemented. For details regarding the properties of the simulator see the documentation that came with the distribution. Copyright © J.M.McGuiness, coder@hussar.me.uk. http://libjmmcg.sf.net/ Distributed under the terms of the GPL v2.1.\n"+exit_codes::to_string()+"Arguments"
      );
      desc.add_options()
         ("help", "Print this help message.")
         ("version", "Print the build number of this program.")
         ("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 bound) to which the server should listen.")
         ("port", boost::program_options::value<unsigned short>()->required(), "An unused port to which the server should listen.")
         ("processor", boost::program_options::value<unsigned short>()->required()->default_value(libisimud::exchanges::common::thread_traits::exchange_simulator_thread.core), "The core to which the server should be bound.")
      ;
      boost::program_options::variables_map vm;
      boost::program_options::store(boost::program_options::parse_command_line(argc, argv, desc), vm);
      if (vm.count("help")) {
         std::cout<<desc<<std::endl;
         return exit_codes::exit_print_help;
      }
      if (vm.count("version")) {
         std::cout<<"Build: " LIBJMMCG_VERSION_NUMBER ", " LIBJMMCG_BUILD_TYPE<<std::endl;
         return exit_codes::exit_print_version;
      }
      boost::program_options::notify(vm);
      thread_traits::set_backtrace_on_signal();
      proc_rules_t proc_rules;
      no_latency_timestamps ts(0U);
      const std::string svr_name(std::string(argv[0]).substr(0, 15));
      svr sim(
         vm["address"].as<boost::asio::ip::address>(),
         vm["port"].as<unsigned short>(),
         svr::socket_t::socket_priority::low,
         thread_traits::api_params_type::processor_mask_type(vm["processor"].as<unsigned short>()),
         thread_traits::api_params_type::priority_type::normal,
         proc_rules,
         ts,
         svr_name.c_str()
      );
      std::clog<<sim<<std::endl;
      thread_traits::set_kernel_affinity(
         thread_traits::get_current_thread(),
         thread_traits::api_params_type::processor_mask_type(0)
      );
      thread_traits::set_kernel_priority(
         thread_traits::get_current_thread(),
         thread_traits::api_params_type::priority_type::idle
      );
      while (!sim.signal_status()) {
         thread_traits::sleep(0);
      }
      std::clog<<sim<<std::endl;
      return exit_codes::exit_success;
   } catch (exception_type const &ex) {
      std::cerr<<"CRT exception. Detail: "<<ex.what()<<std::endl;
      return exit_codes::exit_crt_exception;
   } 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;
}
 
template<class ProcessingRules, class SvrMgr> inline std::string
svr<ProcessingRules, SvrMgr>::to_string() const noexcept(false) {
   std::ostringstream ss;
   ss
      <<thread_t::thread_traits::demangle_name(typeid(*this))
      <<",\naddress="<<address
      <<", port_number: "<<port_number
      <<", exit="<<static_cast<bool>(exit_)
      <<", client processing-error: '"<<client_ex<<"'"
      <<", processor: "<<processor
      <<", manager: "<<manager;
   return ss.str();
}
 
template<class ProcessingRules, class SvrMgr>
template<class LatencyTimestamps>
inline bool
svr<ProcessingRules, SvrMgr>::read_and_process_msgs(typename svr_mgr_t::session &src_cxn, socket_t &dest_skt, LatencyTimestamps &ts) noexcept(false) {
   if (LIKELY(!static_cast<bool>(exit_))) {
      src_cxn.start(); // Should really only do this once a logon message received...
      while (LIKELY(!static_cast<bool>(exit_))) {
         assert(src_cxn.socket().is_open());
         if (UNLIKELY(processor.read_and_process_a_msg(src_cxn.socket(), dest_skt, ts))) {
            // TODO the glibc wrapper prefers this to always return "true"...
            return static_cast<bool>(exit_);
         }
      }
   }
   return static_cast<bool>(exit_);
}
 
template<class ProcessingRules, class SvrMgr>
template<class LatencyTimestamps>
inline bool
svr<ProcessingRules, SvrMgr>::read_and_process_msgs(typename svr_mgr_t::session &src_cxn, typename svr_mgr_t::session &dest_cxn, LatencyTimestamps &ts) noexcept(false) {
   return read_and_process_msgs(src_cxn, dest_cxn.socket(), ts);
}
 
template<class ProcessingRules, class SvrMgr> inline void
svr<ProcessingRules, SvrMgr>::run() noexcept(true) {
   try {
      manager.run();
   } catch (...) {
      client_ex=boost::current_exception();
   }
}
 
template<class ProcessingRules, class SvrMgr> inline std::ostream &
operator<<(std::ostream &os, svr<ProcessingRules, SvrMgr> const &ec) noexcept(false) {
   os<<ec.to_string();
   return os;
}
 
} } }