ZMQTransportReadThread.cpp

Go to the documentation of this file.

#include "madara/transport/zmq/ZMQTransportReadThread.h"

#include "madara/transport/ReducedMessageHeader.h"
#include "madara/transport/Fragmentation.h"
#include "ace/Time_Value.h"
#include "madara/utility/Utility.h"
#include "ZMQContext.h"


#include <iostream>
#include <algorithm>

madara::transport::ZMQTransportReadThread::ZMQTransportReadThread (
  const TransportSettings & settings,
  const std::string & id,
  void * write_socket,
  BandwidthMonitor & send_monitor,
  BandwidthMonitor & receive_monitor,
  PacketScheduler & packet_scheduler)
  : settings_ (settings), id_ (id), context_ (0),
    write_socket_ (write_socket),
    read_socket_ (0),
    send_monitor_ (send_monitor),
    receive_monitor_ (receive_monitor),
    packet_scheduler_ (packet_scheduler)
{
}

void
madara::transport::ZMQTransportReadThread::init (
  knowledge::KnowledgeBase & knowledge)
{
  context_ = &(knowledge.get_context ());

  if (!settings_.no_receiving)
  {
    int send_buff_size = 0;
    int rcv_buff_size = 0;
    int timeout = 1000;
    int buff_size = settings_.queue_length;
    size_t opt_len = sizeof (int);

    // setup the receive buffer
    if (settings_.queue_length > 0)
      buffer_ = new char[settings_.queue_length];

    madara_logger_log (context_->get_logger (), logger::LOG_MAJOR,
      "ZMQTransportReadThread::init:" \
      " setting up read socket\n");

    read_socket_ = zmq_socket (zmq_context.get_context (), ZMQ_SUB);

    // subscribe to all messages
    zmq_setsockopt (read_socket_, ZMQ_SUBSCRIBE, 0, 0);

    madara_logger_log (context_->get_logger (), logger::LOG_MAJOR,
      "ZMQTransportReadThread::init:" \
      " setting rcv buff size to settings.queue_length (%d)\n",
      buff_size);

    int result = zmq_setsockopt (
      read_socket_, ZMQ_RCVBUF, (void *)&buff_size, opt_len);

    if (result == 0)
    {
      int result = zmq_getsockopt (
        read_socket_, ZMQ_RCVBUF, (void *)&rcv_buff_size, &opt_len);

      madara_logger_log (context_->get_logger (), logger::LOG_MAJOR,
        "ZMQTransportReadThread::init:" \
        " successfully set sockopt rcvbuf size to %d. Actual %d allocated\n",
        buff_size, rcv_buff_size);
    }
    else
    {
      madara_logger_log (context_->get_logger (), logger::LOG_MAJOR,
        "ZMQTransportReadThread::init:" \
        " ERROR: errno = %s\n",
        zmq_strerror (zmq_errno ()));
    }


    result = zmq_setsockopt (
      read_socket_, ZMQ_RCVTIMEO, (void *)&timeout, opt_len);

    if (result == 0)
    {
      int result = zmq_getsockopt (
        read_socket_, ZMQ_RCVTIMEO, (void *)&timeout, &opt_len);

      madara_logger_log (context_->get_logger (), logger::LOG_MAJOR,
        "ZMQTransportReadThread::init:" \
        " successfully set rcv timeout to %d\n",
        timeout);
    }
    else
    {
      madara_logger_log (context_->get_logger (), logger::LOG_MAJOR,
        "ZMQTransportReadThread::init:" \
        " ERROR: When setting timeout on rcv, errno = %s\n",
        zmq_strerror (zmq_errno ()));
    }


    if (settings_.hosts.size () >= 1)
    {
      // if the first host was a reliable multicast, we need to connect to it
      if (utility::begins_with (settings_.hosts[0], "pgm") ||
        utility::begins_with (settings_.hosts[0], "epgm"))
      {

      }
    }

    // connect the reader to the host sockets
    size_t i = 0;

    if (settings_.hosts.size () >= 1)
    {
      // we ignore first host unless it is pgm because writer is on first host
      if (!utility::begins_with (settings_.hosts[0], "pgm") &&
        !utility::begins_with (settings_.hosts[0], "epgm"))
      {
        ++i;
      }
    }

    for (; i < settings_.hosts.size (); ++i)
    {
      int connect_result = zmq_connect (
        read_socket_, settings_.hosts[i].c_str ());

      if (connect_result == 0)
      {
        madara_logger_log (context_->get_logger (), logger::LOG_MAJOR,
          "ZMQTransportReadThread::init:" \
          " successfully connected to %s\n",
          settings_.hosts[i].c_str ());
      }
      else
      {
        madara_logger_log (context_->get_logger (), logger::LOG_ERROR,
          "ZMQTransportReadThread::init:" \
          " ERROR: could not connect to %s\n",
          settings_.hosts[i].c_str ());
        madara_logger_log (context_->get_logger (), logger::LOG_MAJOR,
          "ZMQTransportReadThread::init:" \
          " ERROR: errno = %s\n",
          zmq_strerror (zmq_errno ()));
      }
    }
  }

  if (context_)
  {
    // check for an on_data_received ruleset
    if (settings_.on_data_received_logic.length () != 0)
    {
      madara_logger_log (this->context_->get_logger (), logger::LOG_MAJOR,
        "ZMQTransportReadThread::init:" \
        " setting rules to %s\n",
        settings_.on_data_received_logic.c_str ());


#ifndef _MADARA_NO_KARL_
      madara::expression::Interpreter interpreter;
      on_data_received_ = context_->compile (settings_.on_data_received_logic);
#endif // _MADARA_NO_KARL_
    }
    else
    {
      madara_logger_log (this->context_->get_logger (), logger::LOG_MAJOR,
        "ZMQTransportReadThread::init:" \
        " no permanent rules were set\n");
    }
  }
}

void
madara::transport::ZMQTransportReadThread::cleanup (void)
{
  madara_logger_log (context_->get_logger (), logger::LOG_MAJOR,
    "ZMQTransportReadThread::cleanup:" \
    " starting cleanup\n");

  if (read_socket_ != 0)
  {
    madara_logger_log (context_->get_logger (), logger::LOG_MAJOR,
      "ZMQTransportReadThread::cleanup:" \
      " closing read socket\n");

    int option = 0;
    // if you don't do this, ZMQ waits forever for no reason. Super smart.
    zmq_setsockopt (read_socket_, ZMQ_LINGER, (void *)&option, sizeof (int));

    madara::utility::sleep (0.100);

    zmq_close (read_socket_);
  }

  madara_logger_log (context_->get_logger (), logger::LOG_MAJOR,
    "ZMQTransportReadThread::cleanup:" \
    " finished cleanup\n");

}

void
madara::transport::ZMQTransportReadThread::rebroadcast (
  const char * print_prefix,
  MessageHeader * header,
  const knowledge::KnowledgeMap & records)
{
  int64_t buffer_remaining = (int64_t)settings_.queue_length;
  char * buffer = buffer_.get_ptr ();
  int result (0);

  if (!settings_.no_sending)
  {
    result = prep_rebroadcast (*context_, buffer, buffer_remaining,
      settings_, print_prefix,
      header, records,
      packet_scheduler_);

    if (result > 0)
    {
      if (settings_.hosts.size () > 0 && result > 0)
      {
        madara_logger_log (context_->get_logger (), logger::LOG_MAJOR,
          "ZMQTransportReadThread::send:" \
          " sending %d bytes on socket\n", result);

        //send the prepped buffer over ZeroMQ
        result = zmq_send (
          write_socket_, (void *)buffer_.get_ptr (), (size_t)result, ZMQ_DONTWAIT);

        madara_logger_log (context_->get_logger (), logger::LOG_MAJOR,
          "ZMQTransportReadThread::send:" \
          " sent %d bytes on socket\n", result);
      }
    }
  }
}

void
madara::transport::ZMQTransportReadThread::run (void)
{
  if (!settings_.no_receiving)
  {
    // allocate a buffer to send
    char * buffer = buffer_.get_ptr ();
    const char * print_prefix = "ZMQTransportReadThread::run";
    int64_t buffer_remaining = settings_.queue_length;
    size_t zmq_buffer_size = buffer_remaining;

    madara_logger_log (context_->get_logger (), logger::LOG_MAJOR,
      "%s:" \
      " entering main service loop.\n",
      print_prefix);

    knowledge::KnowledgeMap rebroadcast_records;

    if (buffer == 0)
    {
      madara_logger_log (context_->get_logger (), logger::LOG_MINOR,
        "%s:" \
        " Unable to allocate buffer of size " PRIu32 ". Exiting thread.\n",
        print_prefix,
        settings_.queue_length);

      return;
    }

    madara_logger_log (context_->get_logger (), logger::LOG_MINOR,
      "%s:" \
      " entering a recv on the socket.\n",
      print_prefix);

    // blocking receive up to rcv timeout (1 second)
    buffer_remaining = (int64_t)zmq_recv (
      read_socket_, (void *)buffer, zmq_buffer_size, 0);

    madara_logger_log (context_->get_logger (), logger::LOG_MINOR,
      "%s:" \
      " past recv on the socket.\n",
      print_prefix);

    if (buffer_remaining > 0)
    {
      MessageHeader * header = (MessageHeader *)buffer;

      madara_logger_log (context_->get_logger (), logger::LOG_MINOR,
        "%s:" \
        " processing %d byte update from %s.\n",
        print_prefix, (int)buffer_remaining, header->originator);

      process_received_update (buffer, (uint32_t)buffer_remaining, id_, *context_,
        settings_, send_monitor_, receive_monitor_, rebroadcast_records,
#ifndef _MADARA_NO_KARL_
        on_data_received_,
#endif // _MADARA_NO_KARL_
        print_prefix,
        header->originator, header);

      madara_logger_log (context_->get_logger (), logger::LOG_MINOR,
        "%s:" \
        " done processing %d byte update from %s.\n",
        print_prefix, (int)buffer_remaining, header->originator);

      if (header)
      {
        // delete header
        delete header;
      }
    }
    else
    {
      madara_logger_log (context_->get_logger (), logger::LOG_MAJOR,
        "%s:" \
        " wait timeout on new messages. Proceeding to next wait\n",
        print_prefix);
    }
  }
}