MADARA  3.0.5
madara::transport Namespace Reference

Provides the network transport layer for knowledge bases to communicate within and across agents. More...

Classes

class  BandwidthMonitor
 Provides monitoring capability of a transport's bandwidth. More...
 
class  Base
 Base class from which all transports must be derived. More...
 
class  BroadcastTransport
 Multicast-based transport for knowledge. More...
 
class  BroadcastTransportReadThread
 Thread for reading knowledge updates through a Multicast datagram socket. More...
 
class  FragmentMessageHeader
 Defines a fragmentation header which allows for multi-part messages that are only applied once all fragments are received. More...
 
class  MessageHeader
 Defines a robust message header which is the default for KaRL messages. More...
 
class  MulticastTransport
 Multicast-based transport for knowledge. More...
 
class  MulticastTransportReadThread
 Thread for reading knowledge updates through a Multicast datagram socket. More...
 
class  NDDSListener
 Container for NDDS callbacks. More...
 
class  NDDSReadThread
 Thread for reading knowledge updates via a NDDS waitset (deprecated and unused). More...
 
class  NDDSTransport
 This class provides an interface into the NDDS dissemination transport. More...
 
class  PacketScheduler
 Provides scheduler for dropping packets. More...
 
class  QoSTransportSettings
 Container for quality-of-service settings. More...
 
class  ReducedMessageHeader
 Defines a simple, smaller message header of 29 bytes that supports less QoS. More...
 
class  SpliceDataReaderListener
 Container for DDS-related callbacks (deprecated and unused) More...
 
class  SpliceDDSTransport
 This class provides an interface into the Open Splice dissemination transport. More...
 
class  SpliceReadThread
 Thread for reading knowledge updates via waitsets. More...
 
class  SpliceSubscriberListener
 Container for subscriber callbacks (deprecated and unused) More...
 
class  TcpTransport
 TCP-based transport (skeleton code) More...
 
class  TcpTransportReadThread
 Thread for reading knowledge updates through a TCP socket (unused but planned) More...
 
class  TransportContext
 Provides context about the transport. More...
 
class  TransportSettings
 Holds basic transport settings. More...
 
class  UdpRegistryClient
 UDP-based transport for knowledge. More...
 
class  UdpRegistryClientReadThread
 Thread for reading knowledge and registry updates through a UDP socket. More...
 
class  UdpRegistryServer
 UDP-based server that handles a registry of UDP endpoints, which makes it ideal for any NAT-protected agents. More...
 
class  UdpRegistryServerReadThread
 Thread for reading registry updates through a UDP socket. More...
 
class  UdpTransport
 UDP-based transport for knowledge. More...
 
class  UdpTransportReadThread
 Thread for reading knowledge updates through a UDP socket. More...
 
class  ZMQTransport
 ZMQ-based transport for knowledge. More...
 
class  ZMQTransportReadThread
 Thread for reading knowledge updates through a ZMQ datagram socket. More...
 

Typedefs

typedef std::deque< BandwidthRecordBandwidthMessages
 
typedef std::pair< time_t, uint64_t > BandwidthRecord
 
typedef std::map< uint64_t, FragmentMapClockFragmentMap
 Map of clocks to fragments. More...
 
typedef ACE_Condition< ACE_Thread_Mutex > Condition
 
typedef std::map< uint32_t, const char * > FragmentMap
 Map of fragment identifiers to fragments. More...
 
typedef std::map< std::string, ClockFragmentMapOriginatorFragmentMap
 Map of originator to a map of clocks to fragments. More...
 
typedef ACE_High_Res_Timer Timer
 
typedef std::vector< TimerTimers
 
typedef utility::ThreadSafeVector< Base * > Transports
 

Enumerations

enum  Messages {
  ASSIGN = 0, OPERATION = 1, MULTIASSIGN = 2, REGISTER = 3,
  LATENCY = 10, LATENCY_AGGREGATE = 11, LATENCY_SUMMATION = 12, VOTE = 20
}
 
enum  PacketDropType { PACKET_DROP_DETERMINISTIC, PACKET_DROP_PROBABLISTIC }
 Enumeration for packet drop policy types. More...
 
enum  Reliabilities { BEST_EFFORT = 0, RELIABLE = 1 }
 
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
}
 

Functions

MADARA_Export char * add_fragment (const char *originator, uint64_t clock, uint32_t update_number, const char *fragment, uint32_t queue_length, OriginatorFragmentMap &map, bool clear=true)
 Adds a fragment to an originator fragment map and returns the aggregate message if the message is complete. More...
 
MADARA_Export char * defrag (FragmentMap &map)
 Pieces together a fragment map into a single buffer. More...
 
MADARA_Export void delete_fragments (FragmentMap &map)
 Deletes fragments within a fragment map and clears the map. More...
 
MADARA_Export bool exists (const char *originator, uint64_t clock, uint32_t update_number, OriginatorFragmentMap &map)
 Checks if a fragment already exists within a fragment map. More...
 
MADARA_Export void frag (char *source, uint32_t fragment_size, FragmentMap &map)
 Breaks a large packet into smaller packets. More...
 
MADARA_Export bool is_complete (const char *originator, uint64_t clock, OriginatorFragmentMap &map)
 Breaks a large packet into smaller packets. More...
 
int MADARA_Export prep_rebroadcast (knowledge::ThreadSafeContext &context, char *buffer, int64_t &buffer_remaining, const QoSTransportSettings &settings, const char *print_prefix, MessageHeader *header, const knowledge::KnowledgeMap &records, PacketScheduler &packet_scheduler)
 Preps a buffer for rebroadcasting records to other agents on the network. More...
 
int MADARA_Export process_received_update (const char *buffer, uint32_t bytes_read, const std::string &id, knowledge::ThreadSafeContext &context, const QoSTransportSettings &settings, BandwidthMonitor &send_monitor, BandwidthMonitor &receive_monitor, knowledge::KnowledgeMap &rebroadcast_records, knowledge::CompiledExpression &on_data_received, const char *print_prefix, const char *remote_host, MessageHeader *&header)
 Processes a received update, updates monitors, fills rebroadcast records according to settings filters, and performs the bulk of the logic required for a receiver thread. More...
 

Variables

static const int MAX_PACKET_SIZE = 512000
 

Detailed Description

Provides the network transport layer for knowledge bases to communicate within and across agents.

Typedef Documentation

Definition at line 28 of file BandwidthMonitor.h.

typedef std::pair<time_t, uint64_t> madara::transport::BandwidthRecord

Definition at line 27 of file BandwidthMonitor.h.

typedef std::map<uint64_t, FragmentMap> madara::transport::ClockFragmentMap

Map of clocks to fragments.

Definition at line 134 of file Fragmentation.h.

typedef ACE_Condition<ACE_Thread_Mutex> madara::transport::Condition

Definition at line 53 of file Transport.h.

typedef std::map<uint32_t, const char *> madara::transport::FragmentMap

Map of fragment identifiers to fragments.

Definition at line 129 of file Fragmentation.h.

Map of originator to a map of clocks to fragments.

Definition at line 140 of file Fragmentation.h.

typedef ACE_High_Res_Timer madara::transport::Timer

Definition at line 51 of file TransportSettings.h.

typedef std::vector<Timer> madara::transport::Timers

Definition at line 54 of file TransportSettings.h.

Enumeration Type Documentation

Enumerator
ASSIGN 
OPERATION 
MULTIASSIGN 
REGISTER 
LATENCY 
LATENCY_AGGREGATE 
LATENCY_SUMMATION 
VOTE 

Definition at line 78 of file TransportSettings.h.

Enumeration for packet drop policy types.

Enumerator
PACKET_DROP_DETERMINISTIC 
PACKET_DROP_PROBABLISTIC 

Definition at line 40 of file QoSTransportSettings.h.

Enumerator
BEST_EFFORT 
RELIABLE 

Definition at line 73 of file TransportSettings.h.

Enumerator
NO_TRANSPORT 
SPLICE 
NDDS 
TCP 
UDP 
MULTICAST 
BROADCAST 
REGISTRY_SERVER 
REGISTRY_CLIENT 
ZMQ 

Definition at line 60 of file TransportSettings.h.

Function Documentation

char * madara::transport::add_fragment ( const char *  originator,
uint64_t  clock,
uint32_t  update_number,
const char *  fragment,
uint32_t  queue_length,
OriginatorFragmentMap map,
bool  clear = true 
)

Adds a fragment to an originator fragment map and returns the aggregate message if the message is complete.

Parameters
originatorthe originator of the message
clockthe clock of the message
update_numberfragment identifier within clock message
fragmentthe fragment to add
queue_lengthnumber of clock entries allowed per originator
mapa map of existing message fragments
clearif true, clears a fragment entry if message is complete
Returns
unmanaged buffer that contains completed message. If not zero, you must clean this up with delete []

add a fragment to the map in the following conditions 1) originator does not exist yet in the map 2) clock is newer than existing fragments 3) fragment update does not exist yet in an existing clock entry

clear a fragment map and return a buffer in the following situations 4) if we have all fragments

Definition at line 369 of file Fragmentation.cpp.

char * madara::transport::defrag ( FragmentMap map)

Pieces together a fragment map into a single buffer.

Parameters
mapmap containing fragments
Returns
unmanaged buffer that contains completed message. If not zero, you must clean this up with delete []

Definition at line 267 of file Fragmentation.cpp.

void madara::transport::delete_fragments ( FragmentMap map)

Deletes fragments within a fragment map and clears the map.

Parameters
mapmap containing fragments

Definition at line 359 of file Fragmentation.cpp.

bool madara::transport::exists ( const char *  originator,
uint64_t  clock,
uint32_t  update_number,
OriginatorFragmentMap map 
)

Checks if a fragment already exists within a fragment map.

Parameters
originatorthe originator of the message
clockthe clock of the message
update_numberfragment identifier within clock message
mapa map of existing message fragments
Returns
true if

Definition at line 678 of file Fragmentation.cpp.

void madara::transport::frag ( char *  source,
uint32_t  fragment_size,
FragmentMap map 
)

Breaks a large packet into smaller packets.

Parameters
sourcelarge packet that needs to be fragmented
fragment_sizemaximum fragment size
mapmap containing resulting fragments

Definition at line 556 of file Fragmentation.cpp.

bool madara::transport::is_complete ( const char *  originator,
uint64_t  clock,
OriginatorFragmentMap map 
)

Breaks a large packet into smaller packets.

Parameters
originatorthe originator of the message
clockthe clock of the message
mapa map of existing message fragments
Returns
true if the clock entry has all fragments received

Definition at line 643 of file Fragmentation.cpp.

int madara::transport::prep_rebroadcast ( knowledge::ThreadSafeContext context,
char *  buffer,
int64_t &  buffer_remaining,
const QoSTransportSettings settings,
const char *  print_prefix,
MessageHeader header,
const knowledge::KnowledgeMap records,
PacketScheduler packet_scheduler 
)

Preps a buffer for rebroadcasting records to other agents on the network.

Parameters
contextthe key/value storage for variables
bufferthe buffer to fill with header and records
buffer_remainingwill contain the buffer remaining, in case your transport must send other info
settingstransport settings
print_prefixprefix to include before every log message, e.g., "MyTransport::svc"
headermessage header from the received message
recordsa map of variables to records to send
packet_schedulerscheduler for mimicking network conditions

Definition at line 736 of file Transport.cpp.

int madara::transport::process_received_update ( const char *  buffer,
uint32_t  bytes_read,
const std::string &  id,
knowledge::ThreadSafeContext context,
const QoSTransportSettings settings,
BandwidthMonitor send_monitor,
BandwidthMonitor receive_monitor,
knowledge::KnowledgeMap rebroadcast_records,
knowledge::CompiledExpression on_data_received,
const char *  print_prefix,
const char *  remote_host,
MessageHeader *&  header 
)

Processes a received update, updates monitors, fills rebroadcast records according to settings filters, and performs the bulk of the logic required for a receiver thread.

Parameters
bufferbuffer containing all serialized updates
bytes_readbytes in the buffer
idunique identifier for originator strings
contextvariable context of the knowledge base
settingstransport settings
send_monitormonitor of send traffic
receive_monitormonitor of receive traffice
rebroadcast_recordsmap of variables to records to be rebroadcasted (will be filled in by this method)
remote_hostip:port who actually sent this message
on_data_receivedcompiled expression tree of the settings.on_data_received_logic (you have to provide the compiled tree)
print_prefixprefix to include before every log message, e.g., "MyTransport::svc"
headerwill contain the message header object from the message received (you have to clean this up delete–e.g., "delete header").
Returns
-1 Rejected: Non-MADARA Message
-2 Rejected: Message from Self
-3 Rejected: Untrusted Peer
-4 Rejected: Untrusted Originator
-5 Rejected: Wrong domain
>= Number of accepted updates

if we defragged the message, then we need to process the message. In order to do that, we need to overwrite buffer with message so it can be processed normally.

Definition at line 111 of file Transport.cpp.

Variable Documentation

const int madara::transport::MAX_PACKET_SIZE = 512000
static

Definition at line 30 of file MessageHeader.h.