MadaraArchitecture
C++ Guide Series
Architecture | Knowledge Base | Networking | Containers | Threads | Optimizations | KaRL | Encryption | Checkpointing | Knowledge Performance | Logging
The MADARA architecture provides powerful tools for control, timing, and reasoning for distributed networks of devices, services, and applications. The following image outlines the types of interactions available through user configurations or interactions with MADARA functional entities. Each number in the diagram matches a corresponding numbered description below.
- The MADARA Architecture
- 1. Updating/retrieving knowledge
- 2. Sending/receiving knowledge
- 3. Printing to files, loggers, and consoles
- 4. Calling native user functions
- 5. Filtering updates on receive, send, and rebroadcast
- 6. Accessing bandwidth monitoring and enforcing deadlines
- 7. Setting packet drop policies
- 8. Adding new transports
- More Information
The main way that users interact with the MADARA system is by retrieving or mutating state from their C/C++/Java programs. The easiest and most popular way to do this is through Containers, which are C++ classes that point directly to variables in the Knowledge Base.
There are also four functions provided by the KnowledgeBase that allow for accessing state information. We detail these functions below, but you can see a full list of these functions in the developer documentation for the KnowledgeBase class.
Example
// create a knowledge base with no transport
madara::knowledge::KnowledgeBase knowledge;
// check if agents are ready, and if so, set starting_maneuver to 1,
// and call a function called perform_maneuver
knowledge.evaluate ("agents.ready => (starting_maneuver # 1; perform_maneuver ())");
Evaluate is an one-shot evaluation of Knowledge and Reasoning Language (KaRL) logic. KaRL logic allows for various conditionals, predicates, and mutations of variables in the knowledge base.
Evaluate has two function signatures, one of which takes a string KaRL logic and the other which takes a compiled KaRL logic, which is preferred for real-time systems and production code.
The above example checks if someone has set agents.ready to true (non-zero) and if so, we set starting_maneuver to 1 and then call a function called perform_maneuver, which can be mapped to our own C++ function that moves a robotic arm, moves to a location, or some other appropriate action.
Example
// create a knowledge base with no transport
madara::knowledge::KnowledgeBase knowledge;
// check if the value of all_agents_ready is equal to 1
if (knowledge.get ("all_agents_ready").to_integer () ## 1)
{
// all_agents_ready is equal to 1. Do some logic here.
...
}
Developers who want to directly access a single variable from the knowledge base, they can use the get function. Get is an atomic call to the thread-safe context that returns the value of a variable. Values can be strings, 64-bit integers, doubles, arrays of integers or doubles, XML, text files, images, and arbitrary byte arrays or files. The value returned is a knowledge record, which allows for explicit conversions to other types via to_string, to_integer, to_double, and similar functions.
Example
// create a knowledge base with no transport
madara::knowledge::KnowledgeBase knowledge;
// set the variable all_agents_ready to 1.0 (double)
knowledge.set ("all_agents_ready", 1.0);
Developers can use set to change the value of a variable in the knowledge base to a double, integer, string, byte array, image, XML file, or some other supported type. If the variable begins with a ".", like ".i", the variable is considered local and is changes are not sent to the transport layer. If the variable begins with a non-period, like "agents.ready", then the value is sent through the transport layer to all agents who may be interested in the information.
Get and set paired together can be used by developers to emulate a high performance no-SQL database with nanosecond mutation operations and microsecond overhead.
The above example would set the global variable "all_agents_ready" to double value 1.0 and then send that update over the transport.
For a more detailed guide on using the Knowledge Base, see [InteractingWithTheKnowledgeBase].
Example
// create a knowledge base with no transport
madara::knowledge::KnowledgeBase knowledge;
// wait for the value of all_agents_ready to be non-zero
knowledge.wait ("all_agents_ready");
Developers can use the wait function to wait on some important predicate before proceeding to other operations in their C/C++/Java code. Like evaluate, wait can take a user-provided stringified KaRL logic or a compiled version of that logic, which will save lookup/compilation time within the KaRL interpreter. The wait statement can be configured to poll at a certain interval and have a maximum wait time to control how the logic is evaluated and for how long.
The above example waits on the global variable all_agents_ready to be not 0.
For more information about the interaction with the Knowledge Base, check out the following helpful guides and presentations:
- [InteractingWithTheKnowledgeBase] - Part 2 of this guide that specifically focuses on working with the Knowledge Base
- The KaRL Primer - a powerpoint presentation that goes over using KaRL and MADARA
The second interaction that affects users of MADARA is the sending and receiving of knowledge between reasoning agents, generally over a network. Sending/receiving is done by mutating global variables with the evaluate, set, or wait functions outlined in the previous section.
Example
// Create transport settings for a multicast transport
madara::transport::QoSTransportSettings settings;
settings.hosts.push_back ("239.255.0.1:4150");
settings.type = Madara::Transport::MULTICAST;
// create a knowledge base with the multicast transport settings
madara::knowledge::KnowledgeBase knowledge ("agent1", settings);
/*
* send all agents listening on this multicast address the knowledge
* that all agents are currently ready
**/
knowledge.evaluate ("all_agents_ready = 1");
As of version 1.1.9, MADARA currently supports five transports: UDP, IP Multicast, IP Broadcast, Open Splice DDS, and RTI DDS. MADARA transports are not reliable by default because reliability is expensive, and because most real-time systems do not wish to have blocking semantics for reliable transfer of packets. However, even though reliability is not enforced at the transport layer for these three transports, you can wait for acks for important information using evaluate and wait calls to provide reliable delivery of important information that must be received by one or more entities in the network.
Sophisticated example showcasing reliable delivery, despite unreliable transport. This essentially builds a barrier that waits for all agents to acknowledge each other's readiness.
// Create transport settings for a multicast transport
madara::transport::QoSTransportSettings settings;
settings.hosts.push_back ("239.255.0.1:4150");
settings.type = madara::transport::MULTICAST;
// create a knowledge base with the multicast transport settings
madara::knowledge::KnowledgeBase knowledge ("agent1", settings);
// set our unique number id
knowledge.set (".id", id_from_command_line);
// set the number of agents in the network
knowledge.set (".agents", number_of_agents_from_command_line);
// wait for all agents to reply back that they understand we are ready
knowledge.wait (
// keep track of the acks others have sent us for our own readiness
// ";>" indicates the wait should look to the right for the truth condition
"agent{.id}.acks = 0 ;>"
// agents finished is used for our truth condition later.
// Keep looking right in the expression tree for the wait truth condition.
".agents_finished = 0 ;>"
// indicate that this agent is ready. Keep looking right for condition.
"agent{.id}.ready = 1 ;>"
// for loop on variable i from 0 to .agents to check our acks and send acks
".i [0->.agents) ("
// count the number of acks to our ready message
" agent{.id}.{.i}.ack => ++agent{.id}.acks;"
// send an ack to the agent who is ready
" agent{.i}.ready => agent{.i}.{.id}.ack # 1"
// end the for loop and keep looking right for truth condition for wait statement
") ;>"
// for loop on variable i from 0 to .agents to check if everyone has received all necessary information
".i [0->.agents) ("
// check if the agent has received all their acks and mark them as finished
" agent{.i}.acks == .agents => ++.agents_finished"
// end the for loop and keep looking right for truth condition for wait statement
") ;>"
// truth condition for wait statement
".agents_finished == .agents"
// end knowledge.wait statement
);
Two transports (Open Splice DDS and RTI DDS) do provide "reliable" delivery, but each has different semantics for what happens when queue lengths run out (e.g. due to overpublishing, disconnected operations between agents, etc.) You should seek the documentation of these transports' "reliable" transports to get a better grasp of what that means. In general, on a wireless network, reliability is a gradient of support and not an absolute guarantee of arrival. However, both DDS implementations will block your application if you send too many packets during a disconnected operation. For this reason, most real-time systems are built to prioritize the most recent information over stale resends and to only use acks on important information and not every single piece of information sent over the network.
Transports are configured with the QoSTransportSettings class (preferred) or the base transport::Settings class, which does not provide some advanced features like packet filtering, monitoring, etc. These classes are passed to the KnowledgeBase class constructor or through the attach methods, available via the KnowledgeBase class.
Among the more interesting settings available to the transports are the ips and ports to include in the network, how deep of a queue to use (in bytes), and whether to request reliable transport (not checked in the UDP, Multicast, and Broadcast transports). Additionally, these settings classes allow the user to specify whether or not to attempt to rebroadcast packets.
We cover network configuration in more detail in the [InteractingWithTheTransport] guide.
Example
madara::knowledge::KnowledgeBase knowledge;
knowledge.print ("!All_Agents_Ready is set to {all_agents_ready}.\n");
The main entry point for users wanting to log/print knowledge is with the print function. This function takes a string and atomically expands any variables in the string before printing the results to the console, a file, or the system log. Print is thread-safe, with the caveat that by using other printing functions like std::cerr/cout, you may see your own text interrupted by the MADARA logger.
The MADARA logger allows for 10 levels of granularity and prioritization of messages, and the print statement assumes that the text is of highest priority (level 0). To change this to a lower level, simply add a positive integer to the parameter list. See the documentation for print for more information.
Example
// read the current log level
int old_level = madara::logger::global_logger->get_level ();
// set the new logging level to 7
madara::logger::global_logger->set_level (7);
The log level is an integer, generally from 0 to 10, that indicates what level of logging the user wants to see. Passing no parameters to the function will return the current logging level.
Setting log level will affect all MADARA operations after that point until you change the log level again (e.g., if you set log level to 7 that will be the log level for all MADARA operations after that point). These are static functions so they are applied across all knowledge bases in the local process. Changing log level does not send additional information across the network.
Example
// enable logging to a file in the local directory called "my_file.txt"
knowledge.log_to_file ("my_file.txt");
// add logging to the system log with an application name "robotic reasoner"
// and indicate that you want to add to the logging options not reset them
// (true is default and would change logging to only system logging)
knowledge.log_to_system_log ("robotic reasoner", false);
Many systems do not have a std::cerr or such output might simply be piped to /dev/null or discarded in some other way, but developers and users may still want to be able to see the logging information. log_to_file is a function that indicates output should go to a file. The logger is flexible, however, and you can use the companion log_to_stderr and log_to_system_log functions to log to all three locations simultaneously with each print statement.
The first example shows how to reset the logging output to a local file called my_file.txt. The second example indicates that we want to use the system log (e.g. eventvwr in Windows) and our program should be named "robotic reasoner". The second parameter "false" indicates that we do not want to clear the logging flags but want to add the system log as an output. If both examples were executed in the order above, the log would go to my_file.txt and the system log with the program name "robotic reasoner."
Example
/*</b>
* user-defined C function to call from MADARA
* This function calls some other C/C++ function called gps_move ()
**/
madara::knowledge::KnowledgeRecord
my_movement_function (
madara::knowledge::FunctionArguments & args,
madara::knowledge::Variables & variables)
{
gps_move ();
return madara::knowledge::KnowledgeRecord::Integer (0);
}
int main (int argc, char ** argv)
{
// for this example, we're not using a transport
madara::knowledge::KnowledgeBase knowledge;
// attach the callback my_movement_function to a MADARA function called move
knowledge.define_function ("move", my_movement_function);
// call the MADARA move function, which maps to my_movement_function
knowledge.evaluate ("move ()");
return 0;
}
MADARA is intended for real-time systems and especially for autonomous systems, like robots, UAVs, etc. The KaRL engine and transport system is powerful, but it is really only intended to maintain state in a thread-safe, distributed way. For robotics systems, you still need to access native drivers, programs, and code that makes the agent do something interesting to move within or modify its environment. MADARA provides this feature via attaching user callbacks to MADARA function calls. In this manner, predicates can be formed to reason about global or local state and react or interact accordingly.
Example
/**
* filters are like functions, but they are specifically intended to
* modify the provided arguments under certain conditions. This filter
* checks the first argument, and if it is a file type larger than
* 1k bytes, then it removes the file payload entirely.
**/
madara::knowledge::KnowledgeRecord
remove_large_files (
madara::knowledge::FunctionArguments & args,
madara::knowledge::Variables & variables)
{
// by default, we return an empty record, which means remove it
madara::knowledge::KnowledgeRecord result;
if (args.size () > 0)
{
/**
* for simplicity of explanation, we show the condition that
* results in removal (the default).
**/
if (args[0].is_file_type () && args[0].size () > 1000)
{
// we do not have to do anything else
}
else
{
result = args[0];
}
}
return result;
}
int main (int argc, char ** argv)
{
// Create transport settings for a multicast transport
madara::transport::QoSTransportSettings settings;
settings.hosts.push_back ("239.255.0.1:4150");
settings.type = madara::transport::MULTICAST;
// add the above filter for all file types, applied before sending
settings.add_send_filter (madara::knowledge::KnowledgeRecord::ALL_TYPES,
remove_large_files);
// create a knowledge base with the multicast transport settings
madara::knowledge::KnowledgeBase knowledge ("agent1", settings);
/**
* let other agents know that data should be coming. The filter
* will be called, but the type is not a file, and the size is
* going to be less than 1000 bytes, so the information will be
* transported.
**/
knowledge.set ("publisher.ready", madara::knowledge::KnowledgeRecord::Integer (1));
/**
* read some sensor output that has been saved to a file. The filter
* will be applied, and if the size is greater than 1000 bytes, the
* images will be dropped at the transport layer because the above
* filter will have stripped the payload
**/
knowledge.read_file ("current_sensor_readings", "\sensor_output.data");
return 0;
}
User callbacks can also be inserted into the transport layer to modify payloads. The MADARA transport system allows users to insert callbacks into three key operations: receive, send, and rebroadcast. Filter callbacks are given a number of arguments that are relevant to the filtering operation. As of version 1.1.9, these arguments include the following:
- args
[0]: The knowledge record that the filter is acting upon - args
[1]: The name of the knowledge record, if applicable ("" if unnamed, but this should never happen) - args
[2]: The type of operation calling the filter (integer valued). Valid types are:madara::transport::TransportContext::IDLE_OPERATION(should never see),madara::transport::TransportContext::SENDING_OPERATION(transport is trying to send the record),madara::transport::TransportContext::RECEIVING_OPERATION(transport has received the record and is ready to apply the update),madara::transport::TransportContext::REBROADCASTING_OPERATION(transport is trying to rebroadcast the record -- only happens if rebroadcast is enabled in Transport Settings) - args
[3]: Bandwidth used while sending through this transport, measured in bytes per second. - args
[4]: Bandwidth used while receiving from this transport, measured in bytes per second. - args
[5]: Message timestamp (when the message was originally sent, in seconds) - args
[6]: Current timestamp (the result of time (NULL)) - args
[7]: Knowledge domain (sort of like topics in pub/sub architectures in that it is a partition of the knowledge space that you are interested in) - args
[8]: Originator (the name of the originator of the update)
The filter can add data to the payload by pushing a variable name (string) followed by a value, which can be a double, string, integer, byte array, or array of integers or doubles, just as you would do with a set operation. This can be useful if other reasoners in the network are expecting additional meta data for the update (which they are free to strip out or ignore in a receive filter, if they don't need the information).
The filter can also access any variable in the KnowledgeBase through the Variables facade. With the arguments and variables interfaces, developers can respond to transport events in highly dynamic and extensible ways.
The above example creates a filter that strips any file payloads that are larger than 1KB before sending the updates over a multicast network.
Example
/**
* This filter rejects enforces a 100KB/s send/recv limit by dropping
* any knowledge updates on send once the threshold is met.
**/
madara::knowledge::KnowledgeRecord
enforce_bandwidth_limit (
madara::knowledge::FunctionArguments & args,
madara::knowledge::Variables & variables)
{
madara::knowledge::KnowledgeRecord result;
if (args.size () >= 7)
{
if (args[3].to_integer () > 100000 || args[4].to_integer () > 100000)
{
/**
* if either the send or receive bandwidth is greater than 100k,
* we drop the message and try not to make the situation worse
**/
}
else
{
// if the bandwidth limit is not violated, set result to args[0]
result = args[0];
}
}
return result;
}
int main (int argc, char ** argv)
{
// Create transport settings for a multicast transport
madara::transport::QoSTransportSettings settings;
settings.hosts.push_back ("239.255.0.1:4150");
settings.type = madara::transport::MULTICAST;
// add the above filter for all file types, applied before sending
settings.add_send_filter (madara::knowledge::KnowledgeRecord::DOUBLE,
enforce_bandwidth_limit);
// create a knowledge base with the multicast transport settings
madara::knowledge::KnowledgeBase knowledge ("agent1", settings);
/**
* let other agents know that data should be coming. The filter
* will be called, but the type is not a file, and the size is
* going to be less than 1000 bytes, so the information will be
* transported.
**/
knowledge.set ("publisher.ready", madara::knowledge::KnowledgeRecord::Integer (1));
/**
* read some sensor output that has been saved to a file. The filter
* will be applied, and if the size is greater than 1000 bytes, the
* images will be dropped at the transport layer because the above
* filter will have stripped the payload
**/
while (1)
{
knowledge.read_file ("current_sensor_readings", "sensor_output.data");
sleep (1);
}
return 0;
}
MADARA provides developers with bandwidth monitoring and deadline enforcement through the filtering mechanisms outlined in Section 5. MADARA does this by providing key information from the Transport Context (a transport monitor that keeps track of important knowledge about the transport and the current packet). The information specific to bandwidth monitoring and deadline enforcement is provided in argument indexes 3-6 that are passed to the filter.
The above example checks the send and receive bandwidth used over the transport and only sends new updates if neither the send nor receive bandwidth has exceeded 100KB. Depending on how the transport operates, the send bandwidth check may not be necessary as most transports (e.g. DDS and multicast) deliver updates to the person who sent the update (though this is discarded thanks to consistency checks within the transport). The receive bandwidth is updated appropriately, even if a message is rejected (e.g., someone else is using the same multicast IP and sending us packets with the wrong header).
If you'd like to see additional information on filters for bandwidth monitoring and deadline enforcement, see the Bandwidth Filtering tutorial or the Deadline Filtering tutorial.
Example
// Create transport settings for a multicast transport
madara::transport::QoSTransportSettings settings;
settings.hosts.push_back ("239.255.0.1:4150");
settings.type = madara::transport::MULTICAST;
/**
* tell the transport to drop 20% of the packets with a probablistic
* policy. The deterministic policy would send one packet and then
* drop 4 in a set, predictable pattern. Probablistic
* will instead use a uniform random distribution
**/
settings.update_drop_rate (.2, madara::transport::PACKET_DROP_PROBABLISTIC);
// create a knowledge base with the multicast transport settings
madara::knowledge::KnowledgeBase knowledge ("agent1", settings);
/**
* send 100 packets called packet.0, packet.1, etc. valued at 1 each.
* Because of the drop policy, we should expect to only get 20% of
* our packets sent over the network, perhaps to emulate a flaky
* network transport in a simulation.
**/
for (madara::knowledge::KnowledgeRecord::Integer i = 0; i < 100; ++i)
{
knowledge.set (".i", i);
knowledge.evaluate ("packet.{.i} = 1)");
}
For most real-time system developers, the preferred method for dropping messages will always be via targeted filtering with feedback regarding deadlines and bandwidth monitoring (Section 6). However, we also support arbitrary packet dropping decisions based on probablistic or deterministic scheduling policies. This feature can be very useful to developers working with simulations who want to simulate certain network quality-of-service, like the drop rate seen in wireless networks as agents get farther from each other. Unlike the targeted packet filtering discussed earlier, which could take a scalpel to the packet and cut out larger or unwanted components of the packet, packet drop policies discard an entire packet.
Chronologically, packet scheduling drop policies are applied after filtering to emulate the real-world performance of applying filters and having the network layer drop packets.
In the example above, a user has requested a packet drop policy of 20% and then attempted to send 100 packets. Other agents in the network should receive a small portion of the packets (roughly 20%). If we wanted to ensure 20%, we could have used a deterministic drop policy.
Some projects may require custom transports for sending knowledge over serial ports, Bluetooth, or various other protocols. Creating a new transport for MADARA is relatively straight forward, thanks to an extensible transport layer.
There are two major functionalities needed for providing a new transport type to MADARA: sending and receiving. The sender side should extend the transport::Base class. The receiving side should be some type of thread or listener, if such is available via callback with the underlying transport.
The transport::Base class has a virtual method called send_data that should be overridden. This method takes one parameter, a map of variable names to KnowledgeRecord locations, called KnowledgeRecords, within the context that called the send_data method. The contents of this method should be a call to prep_send, which will perform consistency checks and filtering, and then martial the updates to a scoped character array stored within your transport called buffer_.
To show you how easy this is for transport implementers, here is the actual implementation of the MADARA multicast transport send_data function with logging and commenting stripped out.
Example implementation of multicast send_data function
long
madara::transport::MulticastTransport::send_data (
const madara::knowledge::KnowledgeRecords & orig_updates)
{
long result
prep_send (orig_updates, "MulticastTransport::send_data:");
if (addresses_.size () > 0 && result > 0)
{
ssize_t bytes_sent = socket_.send(
buffer_.get_ptr (), (ssize_t)result, addresses_[0]);
send_monitor_.add ((uint32_t)bytes_sent);
result = (long) bytes_sent;
}
return result;
}
The other function you will want to override is the setup () and close () methods in order to create/destroy sockets or do whatever prep or cleanup work that is needed.
There is no set way of creating a receiver for the transport layer. You can create a thread and poll the medium for updates, or you can insert a callback or condition into transport libraries that support that functionality. However, we do have recommendations on how to handle received data. We would recommend usage of the Madara::Transport::process_received_update static method (see Developer documentation) to filter and apply updates from a character buffer according to the user-provided !Transport_Settings.
If your transport can't support buffer martialling/demartialling process we provide, then you won't be able to use the process_received_update method to apply these settings automatically for you. There is no requirement that your transport does honor the transport settings classes. We just ask that you document such functionality for others (or even yourself two years down the road) who might use the transport in your code base.
Anyway, to provide an example of what the service loop of a thread might look like, here's code taken from the MulticastTransportReadThread class:
Example
// read the message
ssize_t bytes_read = read_socket_.recv ((void *)buffer,
settings_.queue_length, remote, 0, &wait_time);
// do other operations specific to your transport
...
if (bytes_read > 0)
{
Message_Header * header = 0;
// use socket's remote address functionality to build a string
// that identifies who the sender was at the socket layer
std::stringstream remote_host;
remote_host << remote.get_host_addr ();
remote_host << ":";
remote_host << remote.get_port_number ();
// helper function provided by MADARA to process a received
// message and prep it for rebroadcast, if necessary
process_received_update (buffer, bytes_read, id_, context_,
*qos_settings_, send_monitor_, receive_monitor_, rebroadcast_records,
on_data_received_, "MulticastTransportReadThread::svc",
remote_host.str ().c_str (), header);
// header will be set if there was a message to handle
if (header)
{
// if we've enabled rebroadcasting,
if (header->ttl > 0 && rebroadcast_records.size () > 0 &&
qos_settings_ && qos_settings_->get_participant_ttl () > 0)
{
--header->ttl;
header->ttl = std::min (
qos_settings_->get_participant_ttl (), header->ttl);
rebroadcast (print_prefix, header, rebroadcast_records);
}
delete header;
}
}
And here's the rebroadcast function for multicast...
Example rebroadcast function for multicast
void
madara::transport::MulticastTransportReadThread::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 = prep_rebroadcast (buffer, buffer_remaining,
*qos_settings_, print_prefix,
header, records,
packet_scheduler_);
if (result > 0)
{
ssize_t bytes_sent = write_socket_.send(
buffer, (ssize_t)result, address_);
send_monitor_.add ((uint32_t)bytes_sent);
}
}
Again, there is no requirement that you do rebroadcasts or honor anything in the settings classes, whether that be related to reliability, queue length, or whatever. A custom transport that you maintain is not something we plan to regulate. However, users of your transport probably would appreciate implementations of the various QoS features we've defined in the QoS_Transport_Settings class.
If you'd like to see a complete tutorial that creates a custom multicast transport, see the GAMS gpc.pl tutorial.
If you are looking for code examples and guides, your best bet would be to start with the tutorials (located in the tutorials directory of the MADARA root directory--see the README.txt file for descriptions). After that, there are many dozens of tests that showcase and profile the many functions, classes, and functionalities available to MADARA users.
Users may also browse the Library Documentation for all MADARA functions, classes, etc. and the Wiki pages on this website.
First Steps in MADARA: Covers installation and a Hello World program.
Intro to Networking: Covers creating a multicast transport and coordination and image sharing between multiple agents.
C++ Guide Series
Architecture | Knowledge Base | Networking | Containers | Threads | Optimizations | KaRL | Encryption | Checkpointing | Knowledge Performance | Logging