BasicASIOTransport.cpp

Go to the documentation of this file.

#include "madara/transport/BasicASIOTransport.h"
#include "madara/transport/TransportContext.h"

#include "madara/transport/ReducedMessageHeader.h"
#include "madara/utility/Utility.h"
#include "madara/expression/ExpressionTree.h"
#include "madara/expression/Interpreter.h"
#include "madara/transport/Fragmentation.h"

#include <iostream>
#include "madara/utility/IntTypes.h"

namespace madara { namespace transport {

namespace mcast = ip::multicast;

//const double BasicASIOTransport::default_read_hertz = 10.0;

BasicASIOTransport::BasicASIOTransport (const std::string & id,
        knowledge::ThreadSafeContext & context,
        TransportSettings & config)
: Base (id, config, context)
{
  // create a reference to the knowledge base for threading
  knowledge_.use (context);

  // set the data plane for the read threads
  read_threads_.set_data_plane (knowledge_);
}

BasicASIOTransport::~BasicASIOTransport ()
{
  BasicASIOTransport::close ();
}

void
BasicASIOTransport::close (void)
{
  this->invalidate_transport ();

  read_threads_.terminate ();

  read_threads_.wait ();

  socket_.close ();
}

int
BasicASIOTransport::setup (void)
{
  // call base setup method to initialize certain common variables
  if (Base::setup () < 0) {
    return -1;
  }

  // resize addresses to be the size of the list of hosts
  addresses_.clear ();
  addresses_.reserve (this->settings_.hosts.size ());

  if (settings_.hosts.size () == 0) {
    madara_logger_log (context_.get_logger (), logger::LOG_MINOR,
      "BasicASIOTransport::setup:" \
      " No host addresses. Aborting setup.\n");
    this->invalidate_transport ();
    return -1;
  }

  // convert the string host:port into an asio address
  for (unsigned int i = 0; i < settings_.hosts.size (); ++i)
  {
    try {
      auto addr_parts = utility::parse_address(settings_.hosts[i]);

      auto addr = ip::address::from_string (addr_parts.first);
      addresses_.emplace_back (addr, addr_parts.second);

      madara_logger_log (context_.get_logger (), logger::LOG_MAJOR,
        "BasicASIOTransport::setup:" \
        " settings address[%d] to %s:%d\n", i,
        addresses_.back ().address ().to_string ().c_str (),
        addresses_.back ().port ());
    } catch (const boost::system::system_error &e) {
      madara_logger_log (context_.get_logger (), logger::LOG_MAJOR,
        "BasicASIOTransport::setup:" \
        " Error parsing address %s: %s\n", settings_.hosts[i].c_str (), e.what ());
    }
  }

  if (addresses_.size () == 0) {
    madara_logger_log (context_.get_logger (), logger::LOG_MAJOR,
      "BasicASIOTransport::setup:" \
      " No valid addresses. Aborting setup.\n");
    this->invalidate_transport ();
    return -1;
  }

  try {
    socket_.open(addresses_[0].protocol ());
  } catch (const boost::system::system_error &e) {
    madara_logger_log (context_.get_logger (), logger::LOG_MAJOR,
      "BasicASIOTransport::setup:" \
      " Error opening sockets: %s\n", e.what ());

    this->invalidate_transport ();
    return -1;
  }

  int ret = setup_sockets ();
  if (ret < 0) {
    return ret;
  }

  ret = setup_read_threads ();
  if (ret < 0) {
    return ret;
  }

  return this->validate_transport ();
}

int
BasicASIOTransport::setup_socket (udp::socket &socket)
{
  int tar_buff_size (settings_.queue_length);

  asio::socket_base::send_buffer_size send_buffer_option;
  asio::socket_base::receive_buffer_size receive_buffer_option;

  try {
    socket.get_option(send_buffer_option);
    int send_buff_size = send_buffer_option.value();

    socket.get_option(receive_buffer_option);
    int rcv_buff_size = receive_buffer_option.value();

    madara_logger_log (context_.get_logger (), logger::LOG_MAJOR,
      "BasicASIOTransport::setup:" \
      " default socket buff size is send=%d, rcv=%d\n",
      send_buff_size, rcv_buff_size);

    if (send_buff_size < tar_buff_size)
    {
      madara_logger_log (context_.get_logger (), logger::LOG_MAJOR,
        "BasicASIOTransport::setup:" \
        " setting send buff size to settings.queue_length (%d)\n",
        tar_buff_size);

      send_buffer_option = tar_buff_size;
      socket.set_option (send_buffer_option);

      socket.get_option (send_buffer_option);
      send_buff_size = send_buffer_option.value();

      madara_logger_log (context_.get_logger (), logger::LOG_MAJOR,
        "BasicASIOTransport::setup:" \
        " current socket buff size is send=%d, rcv=%d\n",
        send_buff_size, rcv_buff_size);
    }

    if (rcv_buff_size < tar_buff_size)
    {
      madara_logger_log (context_.get_logger (), logger::LOG_MAJOR,
        "BasicASIOTransport::setup:" \
        " setting rcv buff size to settings.queue_length (%d)\n",
        tar_buff_size);

      receive_buffer_option = tar_buff_size;
      socket.set_option (receive_buffer_option);

      socket.get_option (receive_buffer_option);
      rcv_buff_size = receive_buffer_option.value();

      madara_logger_log (context_.get_logger (), logger::LOG_MAJOR,
        "BasicASIOTransport::setup:" \
        " current socket buff size is send=%d, rcv=%d\n",
        send_buff_size, rcv_buff_size);
    }
  } catch (const boost::system::system_error &e) {
    madara_logger_log (context_.get_logger (), logger::LOG_MAJOR,
      "BasicASIOTransport::setup:" \
      " Error setting up sockets: %s\n", e.what ());

    this->invalidate_transport ();
    return -1;
  }

  return 0;
}

int
BasicASIOTransport::setup_read_socket (void)
{
  return setup_socket (socket_);
}

int
BasicASIOTransport::setup_write_socket (void)
{
  //return setup_socket (write_socket_);
  return 0;
}

int
BasicASIOTransport::setup_sockets (void)
{
  int ret = setup_read_socket ();
  if (ret < 0) {
    return ret;
  }
  ret = setup_write_socket ();
  if (ret < 0) {
    return ret;
  }

  return 0;
}

int
BasicASIOTransport::setup_read_threads ()
{
  if (!settings_.no_receiving)
  {
    double hertz = settings_.read_thread_hertz;
    if (hertz < 0.0)
    {
      // we need to maintain backwards compatibility
      // people should be capable of bursting reads especially
      hertz = 0.0;
    }

    madara_logger_log (context_.get_logger (), logger::LOG_MAJOR,
      "BasicASIOTransport::setup:" \
      " starting %d threads at %f hertz\n", settings_.read_threads,
      hertz);

    for (uint32_t i = 0; i < settings_.read_threads; ++i)
    {
      std::stringstream thread_name;
      thread_name << "read";
      thread_name << i;

      madara_logger_log (context_.get_logger (), logger::LOG_MAJOR,
        "BasicASIOTransport::setup:" \
        " starting thread %s at %f hertz\n",
        thread_name.str ().c_str (), hertz);
      setup_read_thread (hertz, thread_name.str ());
    }
  }

  return 0;
}

} }