TransportSettings.h

Go to the documentation of this file.

#ifndef _MADARA_TRANSPORT_SETTINGS_H_
#define _MADARA_TRANSPORT_SETTINGS_H_

#include <string>
#include <sstream>
#include <vector>
#include <map>
#include <ostream>
#include "ace/Thread_Mutex.h"
#include "ace/Recursive_Thread_Mutex.h"
#include "ace/Condition_T.h"
#include "ace/Guard_T.h"

#include "ace/High_Res_Timer.h"

#ifdef _USE_CID_

#include "madara/cid/CIDTransportSettings.h"
#include "madara/cid/CIDConvenience.h"
#include "madara/cid/CIDHeuristic.h"
#include "madara/cid/CIDGenetic.h"
#include "madara/utility/Utility.h"

#endif // _USE_CID_

#include "madara/LockType.h"
#include "madara/knowledge/KnowledgeRecord.h"
#include "madara/knowledge/ThreadSafeContext.h"
#include "madara/expression/ExpressionTree.h"
#include "madara/expression/Interpreter.h"
#include "madara/MADARA_export.h"
#include "madara/transport/Fragmentation.h"

namespace madara
{
  namespace transport
  {
    // forward declare for friending
    class Base;

    typedef   ACE_High_Res_Timer                   Timer;
    typedef   std::vector<Timer>                   Timers;

#ifdef _USE_CID_
    typedef   madara::Cid::TransportSettings                LatencyTransportSettings;
#endif

    enum Types {
      NO_TRANSPORT = 0,
      SPLICE       = 1,
      NDDS         = 2,
      TCP          = 3,
      UDP          = 4,
      MULTICAST    = 5,
      BROADCAST    = 6,
      REGISTRY_SERVER = 7,
      REGISTRY_CLIENT = 8,
      ZMQ = 9
    };

    enum Reliabilities {
      BEST_EFFORT = 0,
      RELIABLE = 1
    };

    enum Messages {
      ASSIGN = 0,
      OPERATION = 1,
      MULTIASSIGN = 2,
      REGISTER = 3,
      LATENCY = 10,
      LATENCY_AGGREGATE = 11,
      LATENCY_SUMMATION = 12,
      VOTE = 20
    };

    class MADARA_Export TransportSettings
    {
    public:
      // allow transport::Base to alter private/protected members
      friend class Base;

      // for ease-of-use, typedef the templated guard
      typedef ACE_Guard <MADARA_LOCK_TYPE> ContextGuard;

      #define DEFAULT_DOMAIN      "KaRL"

      typedef std::vector <std::string>    Voters;

      static const uint32_t DEFAULT_QUEUE_LENGTH = 500000;

      static const uint32_t DEFAULT_DEADLINE = 0;

      static const uint32_t DEFAULT_TRANSPORT = NO_TRANSPORT;

      static const uint32_t DEFAULT_RELIABILITY = RELIABLE;

      #define DEFAULT_ID                0

      #define DEFAULT_PROCESSES         1

      #define MAXIMUM_RESEND_ATTEMPTS   10

#ifdef _USE_CID_

      #define DEFAULT_LATENCY_ENABLED   false

      #define DEFAULT_LATENCY_TIMEOUT   10.0

      #define DEFAULT_LATENCY          200000000000

      #define DEFAULT_REDEPLOYMENT_PERCENTAGE     0.10

#endif // _USE_CID_

      TransportSettings ();

      TransportSettings (const TransportSettings & settings);

      virtual ~TransportSettings ();

      void operator= (const TransportSettings & settings);
      
      void add_read_domain (const std::string domain);

      void clear_read_domains (void);

      void get_read_domains (std::vector<std::string> & domains) const;

      bool is_reading_domain (const std::string domain) const;

      size_t num_read_domains (void) const;

#ifdef _USE_CID_

      inline void reset_timers (void)
      {
        ContextGuard guard (mutex);

        for (uint32_t i = 0; i < processes; ++i)
          timers[i].reset ();
        madara::Cid::reset_latencies (latencies, latency_default);
      }

      inline void start_all_timers (void)
      {
        ContextGuard guard (mutex);

        reset_timers ();
        
        for (unsigned int i = 0; i < processes; ++i)
          timers[i].start ();
      }

      inline void stop_timer (int index, bool roundtrip = true)
      {
        ACE_hrtime_t measured;
        ContextGuard guard (mutex);

        timers[index].stop ();
        timers[index].elapsed_time (measured);

        latencies.network_latencies[id][index].second = roundtrip ?
          (uint64_t) measured / 2 : (uint64_t) measured;
      }

      inline void setup (void)
      {
        ContextGuard guard (mutex);

        if (latency_enabled)
        {
          madara::Cid::init (processes, latencies);
          timers.resize (processes);

          madara::Cid::reset_latencies (latencies, latency_default);
        }
      }

      inline void setup (madara::Cid::AlgorithmConfigs & configs)
      {
        ContextGuard guard (mutex);

        latencies.algorithm_configs = configs;
        latencies.results.resize (configs.size ());
      }

      void print_my_latencies (std::ostream & output)
      {
        // we do not use a guard here because we want to do I/O operations
        // outside of the mutex.
        std::stringstream buffer;

        mutex.acquire ();
        madara::Cid::Identifiers & ids = latencies.ids;
        madara::Cid::LatencyVector & current = latencies.network_latencies[id];

        buffer << "Latencies for id = " << id << std::endl;

        // print each id -> latency
        for (uint32_t i = 0; i < processes; ++i)
        {
          buffer << ids[i] << " = " << current[i].second << std::endl;
        }

        mutex.release ();

        output << buffer.str ();
      }

      void print_all_latencies (std::ostream & output)
      {
        // we do not use a guard here because we want to do I/O operations
        // outside of the mutex.
        std::stringstream buffer;

        mutex.acquire ();
        madara::Cid::Identifiers & ids = latencies.ids;

        buffer << "\nAll latencies in the context:\n\n";

        // print each id -> latency
        for (uint32_t i = 0; i < processes; ++i)
        {
          madara::Cid::LatencyVector & current = latencies.network_latencies[i];
          for (uint32_t j = 0; j < processes; ++j)
          {
            buffer << ids[i] << " to " << ids[j] << 
                      " = " << current[j].second << std::endl;
          }
        }

        mutex.release ();

        // print the buffer
        output << buffer.str ();
      }

      void print_all_summations (std::ostream & output)
      {
        // we do not use a guard here because we want to do I/O operations
        // outside of the mutex.
        std::stringstream buffer;

        mutex.acquire ();
        madara::Cid::Identifiers & ids = latencies.ids;
        madara::Cid::SummationsMap & summation_map =
          latencies.network_summations;

        buffer << "\nAll summations in the context:\n\n";

        // print each id -> latency
        for (madara::Cid::SummationsMap::iterator i = summation_map.begin ();
          i != summation_map.end (); ++i)
        {
          buffer << "Degree = " << i->first << std::endl;
          madara::Cid::LatencyVector & current = i->second;

          // temporary, since we can't figure out why the summations aren't
          // sorting in the read thread.
          std::sort (current.begin (), current.end (),
            madara::Cid::IncreasingLatency);

          for (madara::Cid::LatencyVector::size_type j = 0;
            j < current.size (); ++j)
          {
            buffer << "  " << current[j].first << " = " << 
                      current[j].second << "\n";
          }
        }

        mutex.release ();

        // print the buffer
        output << buffer.str ();
      }

      void print_all_results (std::ostream & output)
      {
        // we do not use a guard here because we want to do I/O operations
        // outside of the mutex.
        std::stringstream buffer;

        mutex.acquire ();
        madara::Cid::Identifiers & ids = latencies.ids;
        madara::Cid::AlgorithmConfigs & configs = latencies.algorithm_configs;
        madara::Cid::AlgorithmResults & results = latencies.results;

        buffer << "\nAll redeployment algorithm results in the context:\n\n";

        // print each id -> latency
        for (madara::Cid::AlgorithmConfigs::size_type i = 0;
             i < configs.size (); ++i)
        {
          if (     results[i].algorithm == madara::Cid::CID)
            buffer << "CID,";
          else if (results[i].algorithm == madara::Cid::BCID)
            buffer << "BCID,";
          else if (results[i].algorithm == madara::Cid::BCID_GGA)
            buffer << "BCID-GGA (" << configs[i].time << "),";
          else if (results[i].algorithm == madara::Cid::BCID_BGA)
            buffer << "BCID-BGA (" << configs[i].time << "),";
          else if (results[i].algorithm == madara::Cid::CID_BGA)
            buffer << "CID-BGA (" << configs[i].time << "),";
          else if (results[i].algorithm == madara::Cid::CID_GGA)
            buffer << "CID-GGA (" << configs[i].time << "),";

          buffer << results[i].latency << "\n  ";
          buffer << results[i].deployment << "\n";
        }

        mutex.release ();

        buffer << "\n";

        // print the buffer
        output << buffer.str ();
      }

      std::string aggregate_latencies (void)
      {
        std::stringstream buffer;
        ContextGuard guard (mutex);
        madara::Cid::LatencyVector & current = latencies.network_latencies[id];

        // print each id -> latency
        for (uint32_t i = 0; i < processes; ++i)
        {
          buffer << i << "=" << current[i].second << ";";
        }

        return buffer.str ();
      }

      void unaggregate_latencies (uint32_t source, 
        const std::string & aggregation)
      {
        std::stringstream stream (aggregation);
        ContextGuard guard (mutex);
        madara::Cid::LatencyVector & current = latencies.network_latencies[source];

        // key symbol value symbol
        // 0 = 15     or 24 = 13847169741, for instance
        char symbol;
        unsigned int key;
        uint64_t value;

        for (unsigned int i = 0; !stream.eof (); ++i)
        {
          stream >> key >> symbol >> value >> symbol;

          // make a quick check to see if these values are indeed useful
          if (i < processes)
          {
            current[i].first = key;
            current[i].second = value;
          }
        }
      }

      std::string pack_summations (void)
      {
        ContextGuard guard (mutex);

        return madara::Cid::prepare_summations (id, latencies);
      }

      void unpack_summations (uint32_t source, 
        const std::string & summations)
      {
        std::stringstream stream (summations);
        ContextGuard guard (mutex);
        madara::Cid::LatencyVector & current =
          latencies.network_latencies[source];
        madara::Cid::SummationsMap & summations_map =
          latencies.network_summations;

        // key symbol value symbol
        // 0 = 15     or 24 = 13847169741, for instance
        char symbol;
        unsigned int key;
        uint64_t value;

        while (!stream.eof ())
        {
          stream >> key >> symbol >> value >> symbol;

          madara::Cid::LatencyVector & cur_summation = summations_map[key];

          if (cur_summation.size () == processes)
          {
            // if the array had already been created, sort by Id
            std::sort (cur_summation.begin (), cur_summation.end (),
              madara::Cid::IncreasingId);
          }
          else
          {
            // if the size of this entry is 0, create a new processes-length
            // summation for this degree, and set each element to the
            // corresponding id of that process (essentially sorting by Id)
            cur_summation.resize (processes);
            for (unsigned int i = 0; i < processes; ++i)
              cur_summation[i].first = i;
          }

          // make a quick check to see if these values are indeed useful
          if (source < processes && cur_summation[source].second != value)
          {
            cur_summation[source].second = value;
          }
        }
      }

      void read_dataflow (const std::string & filename)
      {
        ContextGuard guard (mutex);
        latencies.network_summations.clear ();
        madara::Cid::read_deployment (latencies,
          madara::utility::clean_dir_name (filename));
      }

      void read_solution (const std::string & source,
        const std::string & my_host_port)
      {
        ContextGuard guard (mutex);
        unsigned int id;
        std::string host_port;

        std::vector <std::string> tokens;
        std::vector <std::string> splitters;
        std::vector <std::string> pivot_list;

        tokens.resize (2);
        tokens[0] = ";";
        tokens[1] = "=";

        madara::utility::tokenizer (source, splitters, tokens, pivot_list);

        for (std::vector <std::string>::size_type i = 0;
          i + 1 < tokens.size (); i += 2)
        {
          std::stringstream buffer (tokens[i]);
          buffer >> id;

          latencies.ids[id] = tokens[i + 1];
        }
      }

      void run_cid (unsigned int index)
      {
        ContextGuard guard (mutex);

        latencies.solution_lookup.clear ();
        madara::Cid::reset_solution (latencies);

        madara::Cid::approximate (latencies);
        
        latencies.results[index].algorithm = madara::Cid::CID;
        latencies.results[index].deployment = madara::Cid::stringify_solution (latencies);
        latencies.results[index].latency =
          madara::Cid::calculate_latency (latencies);
      }

      void run_bcid (unsigned int index)
      {
        ContextGuard guard (mutex);

        latencies.solution_lookup.clear ();
        madara::Cid::reset_solution (latencies);

        madara::Cid::fill_by_highest_degree (latencies);
        
        latencies.results[index].algorithm = madara::Cid::BCID;
        latencies.results[index].deployment =
          madara::Cid::stringify_solution (latencies);
        latencies.results[index].latency =
          madara::Cid::calculate_latency (latencies);
      }

      void run_bcid_bga (unsigned int index)
      {
        ContextGuard guard (mutex);

        latencies.solution_lookup.clear ();
        madara::Cid::reset_solution (latencies);

        madara::Cid::fill_by_highest_degree (latencies);
        madara::Cid::ga_naive (latencies, 
          latencies.algorithm_configs[index].time);

        latencies.results[index].algorithm = madara::Cid::BCID_BGA;
        latencies.results[index].deployment =
          madara::Cid::stringify_solution (latencies);
        latencies.results[index].latency =
          madara::Cid::calculate_latency (latencies);
      }

      void run_bcid_gga (unsigned int index)
      {
        ContextGuard guard (mutex);

        latencies.solution_lookup.clear ();
        madara::Cid::reset_solution (latencies);

        madara::Cid::fill_by_highest_degree (latencies);
        madara::Cid::ga_degree (latencies, 
          latencies.algorithm_configs[index].time);

        latencies.results[index].algorithm = madara::Cid::BCID_GGA;
        latencies.results[index].deployment =
          madara::Cid::stringify_solution (latencies);
        latencies.results[index].latency =
          madara::Cid::calculate_latency (latencies);
      }

      void run_cid_bga (unsigned int index)
      {
        ContextGuard guard (mutex);

        latencies.solution_lookup.clear ();
        madara::Cid::reset_solution (latencies);

        madara::Cid::approximate (latencies);
        madara::Cid::ga_naive(latencies, 
          latencies.algorithm_configs[index].time);

        latencies.results[index].algorithm = madara::Cid::CID_BGA;
        latencies.results[index].deployment =
          madara::Cid::stringify_solution (latencies);
        latencies.results[index].latency =
          madara::Cid::calculate_latency (latencies);
      }

      void run_cid_gga (unsigned int index)
      {
        ContextGuard guard (mutex);

        latencies.solution_lookup.clear ();
        madara::Cid::reset_solution (latencies);

        madara::Cid::approximate (latencies);
        madara::Cid::ga_degree(latencies, 
          latencies.algorithm_configs[index].time);

        latencies.results[index].algorithm = madara::Cid::CID_GGA;
        latencies.results[index].deployment =
          madara::Cid::stringify_solution (latencies);
        latencies.results[index].latency =
          madara::Cid::calculate_latency (latencies);
      }

      void run (unsigned int index)
      {
        int algorithm = latencies.algorithm_configs[index].algorithm;
        if (     algorithm == madara::Cid::CID)
          run_cid (index);
        else if (algorithm == madara::Cid::BCID)
          run_bcid (index);
        else if (algorithm == madara::Cid::BCID_GGA)
          run_bcid_gga (index);
        else if (algorithm == madara::Cid::BCID_BGA)
          run_bcid_bga (index);
        else if (algorithm == madara::Cid::CID_BGA)
          run_cid_bga (index);
        else if (algorithm == madara::Cid::CID_GGA)
          run_cid_gga (index);
      }

      void run_all (void)
      {
        for (unsigned int i = 0; i < latencies.algorithm_configs.size (); ++i)
          run (i);

        std::sort (latencies.results.begin (), latencies.results.end (),
          madara::Cid::IncreasingAlgorithmLatency);
      }

#endif //  _USE_CID_

       virtual void load (const std::string & filename,
         const std::string & prefix = "transport");

       virtual void load_text (const std::string & filename,
         const std::string & prefix = "transport");

       virtual void save (const std::string & filename,
         const std::string & prefix = "transport") const;

       virtual void save_text (const std::string & filename,
         const std::string & prefix = "transport") const;


      std::string write_domain;

      uint32_t read_threads;

      uint32_t queue_length;

      uint32_t type;

      uint32_t max_fragment_size;

      int resend_attempts;

      uint32_t fragment_queue_length;

      uint32_t reliability;

      uint32_t id;

      uint32_t processes;

      std::string on_data_received_logic;
      
      bool delay_launch;

      bool never_exit;

      bool send_reduced_message_header;

      mutable OriginatorFragmentMap fragment_map;

      double slack_time;

      double read_thread_hertz;

#ifdef _USE_CID_
      bool latency_enabled;

      double latency_timeout;

      uint64_t latency_default;

      MADARA_LOCK_TYPE mutex;

      LatencyTransportSettings latencies;

      Timers timers;

      uint32_t num_responses;

      uint32_t num_summations;

      uint32_t num_voters;

      uint32_t num_votes_received;

      double  redeployment_percentage_allowed;

      Voters voters;
#endif  // _USE_CID_

      std::vector <std::string> hosts;

      bool no_sending;
      
      bool no_receiving;

    private:

      std::map <std::string, int> read_domains_;
    };

    inline std::string
    type_name (const TransportSettings & settings)
    {
      std::string name = "none";

      switch (settings.type)
      {
      case 0:
        name = "None";
        break;
      case 1:
        name = "PrismTech DDS";
        break;
      case 2:
        name = "RTI DDS";
        break;
      case 3:
        name = "TCP (unsupported)";
        break;
      case 4:
        name = "UDP Unicast";
        break;
      case 5:
        name = "UDP Multicast";
        break;
      case 6:
        name = "UDP Broadcast";
        break;
      case 7:
        name = "UDP P2P Registry Server";
        break;
      case 8:
        name = "UDP P2P Registry Client";
        break;
      case 9:
        name = "ZeroMQ Pub/Sub";
        break;
      }

      return name;
    }
  }
}

#include "TransportSettings.inl"

#endif // _MADARA_TRANSPORT_SETTINGS_H_