ghs-demo-comms.h

Go to the documentation of this file.

 
#ifndef GHS_DEMO_COMMS
#define GHS_DEMO_COMMS 
#include "ghs-demo-config.h"
#include "ghs-demo-msgutils.h"
#include "ghs-demo-edgemetrics.h"
#include "seque/static_queue.h"
#include <nng/nng.h>//req_s, rep_s, msg
#include <cstring> //memcpy, memset
#include <unordered_map>
#include <vector>
#include <thread>
#include <mutex>
 
#ifndef COMMS_DEMO_MAX_N
#define COMMS_DEMO_MAX_N 8
#endif
 
#ifndef PAYLOAD_MAX_SZ
#define PAYLOAD_MAX_SZ 1024
#endif
 
namespace demo{
 
  enum Errno{
    OK = 0,                 
    ERR_BAD_PAYLOAD_SZ,     
    ERR_NO_PAYLOAD_TYPE,    
    ERR_UNRECOGNIZED_PAYLOAD_TYPE, 
    ERR_NULL_SRC,           
    ERR_DEST_UNSET,         
    ERR_HANGUP,             
    ERR_NNG,                
  };
 
  enum PayloadType
  {
    PAYLOAD_TYPE_NOT_SET =0,   
    PAYLOAD_TYPE_CONTROL,      
    PAYLOAD_TYPE_METRICS,      
    PAYLOAD_TYPE_PING,         
    PAYLOAD_TYPE_GHS,          
  };
 
//typedef uint8_t ControlCommand;
//#define CONTROL_TEST_MSG      1
//#define CONTROL_LATENCY       2
//#define CONTROL_START_GHS     12
//#define CONTROL_RESET_GHS     13
 
  typedef uint64_t OptMask;
 
  typedef uint16_t Destination;
#define MESSAGE_DEST_UNSET -1
 
 
  typedef size_t SequenceCounter;
 
  struct 
    //__attribute__((__packed__,aligned(1))) <-- messes with doxygen
    Header
    {
      Destination agent_to=MESSAGE_DEST_UNSET;
      uint16_t agent_from;
      PayloadType type=PAYLOAD_TYPE_NOT_SET; 
      uint16_t payload_size=0;
    };
 
  struct 
    //__attribute__((__packed__,aligned(1))) <-- messes with doxygen
    Control
    {
      SequenceCounter sequence=0;
    };
 
  struct 
    //__attribute__((__packed__,aligned(1))) <-- messes with doxygen
    WireMessage
    {
      Control control;
      Header header;
      uint8_t bytes[PAYLOAD_MAX_SZ];
 
      size_t size()  const
      { 
        return sizeof(demo::Header)
          +sizeof(demo::Control)
          +header.payload_size; 
      }
    };
 
  class Comms
  {
    public:
      Comms();
      ~Comms();
      bool ok();
 
      static Comms& inst();
 
      Comms& with_config(Config &);
 
      Errno send(demo::WireMessage&, demo::OptMask=0);
 
      void start_receiver();
      void stop_receiver();
 
      bool has_msg();
 
      bool get_next(demo::WireMessage&);
 
      void little_iperf();
      void exchange_iperf();
 
      void print_iperf();
 
      Kbps kbps_to(const uint16_t agent_id) const;
 
      CommsEdgeMetric unique_link_metric_to(const uint16_t agent_id) const;
 
    private:
 
      void validate_sockets();
      void read_loop();
      int recv(demo::WireMessage& buf);
      demo::Errno internal_send(demo::WireMessage &m, const char* endpoint, long &us_rt);
 
      seque::StaticQueue<demo::WireMessage,1024> in_q;
      bool read_continues;
      Config ghs_cfg;
      nng_listener ctr_listener = NNG_LISTENER_INITIALIZER;
      nng_listener ghs_listener = NNG_LISTENER_INITIALIZER;
      nng_socket incoming=NNG_SOCKET_INITIALIZER;
      nng_socket outgoing=NNG_SOCKET_INITIALIZER;
      size_t outgoing_seq;
      std::array<size_t,COMMS_DEMO_MAX_N> sequence_counters;
      std::array<Kbps,COMMS_DEMO_MAX_N> kbps;
      std::mutex q_mut;
      std::mutex send_mut;
      std::thread reader_thread;
 
  };
}
 
 
#endif