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	BasicASIOTransport

class	BroadcastTransport
	Broadcast-based transport for knowledge. 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	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	SharedMemoryPush
	Custom network transport generated by gpc.pl. 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	TransportContext
	Provides context about the transport. More...

class	TransportSettings
	Holds basic transport settings. More...

class	UdpRegistryClient
	UDP-based transport for knowledge. More...

class	UdpRegistryServer
	UDP-based server that handles a registry of UDP endpoints, which makes it ideal for any NAT-protected agents. More...

class	UdpTransport
	UDP-based transport for knowledge. More...

class	UdpTransportReadThread
	Thread for reading knowledge updates through a UDP socket. More...

class	ZMQContext
	A class that manages a ZMQ context. 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< BandwidthRecord >	BandwidthMessages

typedef std::pair< time_t, uint64_t >	BandwidthRecord

typedef std::map< uint64_t, FragmentMap >	ClockFragmentMap
	Map of clocks to fragments. More...

typedef std::map< uint32_t, utility::ScopedArray< const char > >	FragmentMap
	Map of fragment identifiers to fragments. More...

typedef std::map< std::string, ClockFragmentMap >	OriginatorFragmentMap
	Map of originator to a map of clocks to fragments. More...

typedef utility::ThreadSafeVector< Base * >	Transports

using	udp = boost::asio::ip::udp

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, uint64_t &total_size, 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, uint64_t &total_size)
	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 (const char source, uint64_t total_size, const char originator, const char *domain, uint64_t clock, uint64_t timestamp, uint32_t quality, unsigned char ttl, uint64_t fragment_size, FragmentMap &map)
	Breaks a large packet into smaller packets. More...

std::string MADARA_EXPORT	get_operation_type (int type)
	Return the type of operation (usually from TransportContext) More...

MADARA_EXPORT bool	is_complete (const char *originator, uint64_t clock, OriginatorFragmentMap &map)
	Breaks a large packet into smaller packets. More...

int	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	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...

std::string	type_name (const TransportSettings &settings)

MADARA_EXPORT std::string	types_to_string (int id)
	Converts a transport type enum to a string equivalent. More...

Variables
static const int	MAX_PACKET_SIZE = 512000

MADARA_EXPORT ZMQContext	zmq_context

Detailed Description

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

Typedef Documentation

◆ BandwidthMessages

typedef std::deque<BandwidthRecord> madara::transport::BandwidthMessages

Definition at line 26 of file BandwidthMonitor.h.

◆ BandwidthRecord

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

Definition at line 25 of file BandwidthMonitor.h.

◆ ClockFragmentMap

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

Map of clocks to fragments.

Definition at line 143 of file Fragmentation.h.

◆ FragmentMap

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

Map of fragment identifiers to fragments.

Definition at line 138 of file Fragmentation.h.

◆ OriginatorFragmentMap

typedef std::map<std::string, ClockFragmentMap> madara::transport::OriginatorFragmentMap

Map of originator to a map of clocks to fragments.

Definition at line 148 of file Fragmentation.h.

◆ Transports

typedef utility::ThreadSafeVector<Base*> madara::transport::Transports

Definition at line 251 of file Transport.h.

◆ udp

typedef boost::asio::ip::udp madara::transport::udp

Definition at line 23 of file BasicASIOTransport.h.

Enumeration Type Documentation

◆ Messages

enum madara::transport::Messages

Enumerator
ASSIGN
OPERATION
MULTIASSIGN
REGISTER
LATENCY
LATENCY_AGGREGATE
LATENCY_SUMMATION
VOTE

Definition at line 57 of file TransportSettings.h.

◆ PacketDropType

enum madara::transport::PacketDropType

Enumeration for packet drop policy types.

Enumerator
PACKET_DROP_DETERMINISTIC
PACKET_DROP_PROBABLISTIC

Definition at line 39 of file QoSTransportSettings.h.

◆ Reliabilities

enum madara::transport::Reliabilities

Enumerator
BEST_EFFORT
RELIABLE

Definition at line 51 of file TransportSettings.h.

◆ Types

enum madara::transport::Types

Enumerator
NO_TRANSPORT
SPLICE
NDDS
TCP
UDP
MULTICAST
BROADCAST
REGISTRY_SERVER
REGISTRY_CLIENT
ZMQ

Definition at line 37 of file TransportSettings.h.

Function Documentation

◆ add_fragment()

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

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

Parameters

originator	the originator of the message
clock	the clock of the message
update_number	fragment identifier within clock message
fragment	the fragment to add
queue_length	number of clock entries allowed per originator
map	a map of existing message fragments
total_size	the size of the returned buffer (!0 on success)
clear	if 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 637 of file Fragmentation.cpp.

◆ defrag()

char * madara::transport::defrag	(	FragmentMap &	map,
		uint64_t &	total_size
	)

Pieces together a fragment map into a single buffer.

Parameters

map	map containing fragments
total_size	the total size of the resulting buffer

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

Definition at line 522 of file Fragmentation.cpp.

◆ delete_fragments()

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

Deletes fragments within a fragment map and clears the map.

Parameters

map	map containing fragments

Definition at line 628 of file Fragmentation.cpp.

◆ exists()

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

originator	the originator of the message
clock	the clock of the message
update_number	fragment identifier within clock message
map	a map of existing message fragments

Returns: true if

Definition at line 1039 of file Fragmentation.cpp.

◆ frag()

void madara::transport::frag	(	const char *	source,
		uint64_t	total_size,
		const char *	originator,
		const char *	domain,
		uint64_t	clock,
		uint64_t	timestamp,
		uint32_t	quality,
		unsigned char	ttl,
		uint64_t	fragment_size,
		FragmentMap &	map
	)

Breaks a large packet into smaller packets.

Parameters

source	large packet that needs to be fragmented
total_size	total size of the source buffer
originator	the originator id of the agent
domain	the transport knowledge domain
clock	the lamport clock for fragments to use
timestamp	the ns timestamp of the message
quality	the quality of the sender
ttl	the time-to-live of the message for rebroadcasting
fragment_size	maximum fragment size
map	map containing resulting fragments

Definition at line 840 of file Fragmentation.cpp.

◆ get_operation_type()

std::string MADARA_EXPORT madara::transport::get_operation_type ( int type )

inline

Return the type of operation (usually from TransportContext)

Parameters

type	the enum type of the message operation

Returns: the string-based, human-readable equivalent of the type

Definition at line 227 of file Transport.h.

◆ is_complete()

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

Breaks a large packet into smaller packets.

Parameters

originator	the originator of the message
clock	the clock of the message
map	a map of existing message fragments

Returns: true if the clock entry has all fragments received

Definition at line 955 of file Fragmentation.cpp.

◆ prep_rebroadcast()

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

context	the key/value storage for variables
buffer	the buffer to fill with header and records
buffer_remaining	will contain the buffer remaining, in case your transport must send other info
settings	transport settings
print_prefix	prefix to include before every log message, e.g., "MyTransport::svc"
header	message header from the received message
records	a map of variables to records to send
packet_scheduler	scheduler for mimicking network conditions

Definition at line 791 of file Transport.cpp.

◆ process_received_update()

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

buffer	buffer containing all serialized updates
bytes_read	bytes in the buffer
id	unique identifier for originator strings
context	variable context of the knowledge base
settings	transport settings
send_monitor	monitor of send traffic
receive_monitor	monitor of receive traffice
rebroadcast_records	map of variables to records to be rebroadcasted (will be filled in by this method)
remote_host	ip:port who actually sent this message
on_data_received	compiled expression tree of the settings.on_data_received_logic (you have to provide the compiled tree)
print_prefix	prefix to include before every log message, e.g., "MyTransport::svc"
header	will 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 109 of file Transport.cpp.

◆ type_name()

std::string madara::transport::type_name ( const TransportSettings & settings )

inline

Definition at line 339 of file TransportSettings.h.

◆ types_to_string()

std::string madara::transport::types_to_string ( int id )

Converts a transport type enum to a string equivalent.

Parameters

id	the id of the type to retrieve

Returns: the name of the registered transport

Definition at line 10 of file TransportSettings.cpp.

Variable Documentation

◆ MAX_PACKET_SIZE

const int madara::transport::MAX_PACKET_SIZE = 512000

static

Definition at line 31 of file MessageHeader.h.

◆ zmq_context

madara::transport::ZMQContext madara::transport::zmq_context

extern

Definition at line 5 of file ZMQContext.cpp.

Classes

Typedefs

Enumerations

Functions

Variables

Detailed Description

Typedef Documentation

◆ BandwidthMessages

◆ BandwidthRecord

◆ ClockFragmentMap

◆ FragmentMap

◆ OriginatorFragmentMap

◆ Transports

◆ udp

Enumeration Type Documentation

◆ Messages

◆ PacketDropType

◆ Reliabilities

◆ Types

Function Documentation

◆ add_fragment()

◆ defrag()

◆ delete_fragments()

◆ exists()

◆ frag()

◆ get_operation_type()

◆ is_complete()

◆ prep_rebroadcast()

◆ process_received_update()

◆ type_name()

◆ types_to_string()

Variable Documentation

◆ MAX_PACKET_SIZE

◆ zmq_context