The Multicast/Reduction Network: A User's Guide to MRNet v2.1

MRNet has been developed to be highly portable; we expect it to run properly on all common Unix-based as well as Microsoft Windows platforms. This being said, we have successfully built and tested MRNet on the following systems:

Our build system attempts to use native system compilers where appropriate.

Here we list the third party tools that MRNet uses and needs for proper installation on UNIX/Linux systems. For Windows systems, pre-compiled libraries and binaries are available (upon request).

MRNet uses GNU autoconf to discover the platform specific configuration parameters. The script that does this auto-configuration is called configure.

UNIX>  ./configure --help

shows all possible options of the command. Below, we display the MRNet-specific ones:

  --with-libfldir dir            Directory containing flex library


  --enable-shared                Build shared library versions of MRNet and XPlat


  --enable-debug                 Build MRNet and XPlat with debug information


  --enable-verbosebuild          Show build actions (useful for debugging build problems)

./configure without any options should give reasonable results, but the user may specify certain options. For example,

UNIX> ./configure CXX=g++ --with-libfldir=/usr/local/lib

instructs the configure script to use g++ for the C++ compiler and /usr/local/lib as the location of the flex library.

To build MRNet:

UNIX>  make

After a successful build, the following files will be present:

To build the MRNet tests:

UNIX>  make tests

builds the MRNet test files. In addition to those files above, you will also generate:

To install the MRNet components (i.e., libraries, executables, and headers) to the directories specified during configure (default install locations are /usr/local/{bin,lib,include}):

UNIX>  make install

To install the MRNet tests:

UNIX>  make install-tests

If your system does not provide the C++ Boost headers (normally installed in /usr/include/boost), we provide the subset of Boost header files necessary for building MRNet. To install these headers:

UNIX>  make install-boost

UNIX> mrnet_tests.sh [ -l | -r <hostfile> | -a <hostfile> ] [ -f <sharedobject> ]

MRNet is maintained primarily by the Paradyn Project, University of Wisconsin-Madison. Comments and feedback whether positive or negative are encouraged.

Please report bugs to paradyn@cs.wisc.edu.

The MRNet webpage is https://www.paradyn.org/mrnet/

An MRNet end-point represents a tool or application process. In particular, they represent the back-end processes (i.e., leaf processes) in the tree overlay. The front-end can communicate in a unicast or multicast fashion with one or more of these end-points as described below.

MRNet uses communicators to represent groups of end-points. Like communicators in MPI, MRNet communicators provide a handle that identifies a set of end-points for point-to-point, multicast or broadcast communications. MPI applications typically have a non-hierarchical layout of potentially identical processes. In contrast, MRNet enforces a tree-like layout of all processes, rooted at the front-end. Accordingly, MRNet communicators are created and managed by the front-end, and communication is only allowed between a front-end and its back-ends. As such, back-ends cannot interact with each other directly using the MRNet API.

A stream is a logical channel that connects the front-end to the end-points of a communicator. All MRNet communication uses the stream abstraction. Streams carry data packets downstream, from the front-end toward the back-ends, and upstream, from the back-ends toward the front-end. Streams are expected to carry data of a specific type, allowing data aggregation operations to be associated with a stream. The type is specified using a format string (See Appendix A, MRNet Format Strings) similar to those used in C formatted I/O primitives (e.g., a packet whose data is described by the format string "%d %d %f %s" contains two integers followed by a float then a character string). MRNet expands the standard format string specification to allow for description of arrays.

Data aggregation is the process of merging multiple input data packets and transforming them into one or more output packets. Though it is not necessary for the aggregation to result in less or even different data, aggregations that reduce or modify data values are most common. MRNet uses data filters to aggregate data packets. Filters specify an operation to perform and the type of the data expected on the bound stream. Filter instances are bound to a stream at stream creation. MRNet uses two types of filters: synchronization filters and transformation filters. Synchronization filters organize data packets from downstream nodes into synchronized waves of data packets, while transformation filters operate on the synchronized data packets yielding one or more output packets. A distinction between synchronization and transformation filters is that synchronization filters are independent of the packet data type, but transformation filters operate on packets of a specific type.

Synchronization filters operate on data flowing upstream in the network, receiving packets one at a time and outputting packets only when the specified synchronization criteria has been met. Synchronization filters provide a mechanism to deal with the asynchronous arrival of packets from children nodes. The synchronizer collects packets and typically aligns them into waves, passing an entire wave onward at the same time. Therefore, synchronization filters do no data transformation and can operate on packets in a type-independent fashion. MRNet currently supports two synchronization modes:

Transformation filters can be used on both upstream and downstream data flows. Transformation filters input a group of synchronized packets, and combine data from multiple packets by performing an aggregation that yields one or more new data packets. Since transformation filters are expected to perform computational operations on data packets, there is a type requirement for the data packets to be passed to this type of filter: the data format string of the stream's packets and the filter must be the same. Transformation operations must be synchronous, but are able to maintain state from one execution to the next. MRNet provides several transformation filters that should be of general use:

Chapter 7, Adding New Filters describes facilities for adding new user-defined transformation and synchronization filters.

A complete description of the MRNet API is in Chapter 5, The MRNet C++ API Reference. This section offers a brief overview only. Using libmrnet.a, a tool can leverage a system of internal processes, instances of the mrnet_commnode program, as a communication substrate. After instantiation of the MRNet network (discussed in the section called “MRNet Instantiation”, the front-end and back-end processes are connected by the internal processes. The connection topology and host assignment of these processes is determined by a configuration file, thus the geometry of MRNet's process tree can be customized to suit the physical topology of the underlying hardware resources. While MRNet can generate a variety of standard topologies, users can easily specify their own topologies; see Chapter 6, MRNET Process-tree Topologies for further discussion.

The MRNet API contains Network, EndPoint, Communicator, and Stream objects that a tool's front-end and back-end use for communication. The Network object is used to instantiate the MRNet network and access EndPoint objects that represent available tool back-ends. The Communicator object is a container for groups of end-points, and Stream objects are used to send data to the EndPoints in a Communicator.

   front_end_main(...) {
1.     Network * net;
2.     Communicator * comm;
3.     Stream * stream;
4.     PacketPtr packet;
5.     int tag = FirstApplicationTag;
6.     float result;

7.     net = new Network(topology_file, backend_exe, backend_argv);

8.     comm = net->get_BroadcastCommunicator( );

9.     stream = net->new_Stream(comm, TFILTER_SUM, SFILTER_WAITFORALL);

10.    stream->send(tag, "%d", SUM_INIT);

11.    stream->recv(&tag, packet)

12.    packet->unpack("%f", &result);
   }

A simplified version of code from an example tool front-end is shown in Example 4.1, “MRNet Front-end Sample Code”. In the front-end code, after some variable definitions in lines 1-6, an instance of the MRNet network is created on line 7 using the topology specified in topology_file. In line 8, the newly created Network object is queried for an auto-generated broadcast communicator that contains all available end-points. In line 9, this Communicator is used to establish a Stream that will use a built-in filter that finds the summation of the data sent upstream. The front-end then sends one or more initialization messages to the backends; in our example code on line 10, we broadcast an integer initializer on the new stream. The tag parameter is an application-specific value denoting the nature of the message being transmitted. After the send operation, the front-end performs a blocking stream receive at line 11. This call returns a tag and a packet. Finally, line 12 calls unpack to deserialize the floating point value contained in packet.

   back_end_main(int argc, char** argv) {
1.     Stream * stream;
2.     PacketPtr packet;
3.     int val, tag;
4.     float random_float = (float) random( );

5.     Network * net = new Network(argc,argv);

6.     net->recv(&tag, packet, &stream);

7.     packet->unpack("%d", &val );

8.     if( val == SUM_INIT )
9.         stream->send(tag, "%f", random_float);
   }

Example 4.2, “MRNet Back-end Sample Code” shows the code for the back-end that reciprocates the actions of the front-end. Each tool back-end first connects to the MRNet network in line 5, using the back-end version of the Network constructor that receives its arguments via the program argument vector (argc/argv). While the front-end makes a stream-specific receive call, the back-ends use a stream-anonymous network receive that returns the tag sent by the front-end, the packet containing the actual data sent, and a stream object representing the stream that the front-end has established. Finally, each back-end sends a scalar floating point value upstream toward the front-end.

A complete example of MRNet code can be found below in Appendix C, A Complete MRNet Example: Integer Addition.

While conceptually simple, creating and connecting the internal processes is complicated by interactions with the various job scheduling systems. In the simplest environments, we can launch jobs manually using facilities like rsh or ssh. In more complex environments, it is necessary to submit all requests to a job management system. In this case, we are constrained by the operations provided by the job manager (and these vary from system to system). We currently support two modes of instantiating MRNet-based tools.

In the first mode of process instantiation, MRNet creates the internal and back-end processes, using the specified MRNet topology configuration to determine the hosts on which the components should be located. First, the front-end consults the configuration and uses a remote shell program to create internal processes for the first level of the communication tree on the appropriate hosts. Upon instantiation, the newly created processes establish a network connection to the process that created it. The first activity on this connection is a message from parent to child containing the portion of the configuration relevant to that child. The child then uses this information to begin instantiation of the sub-tree rooted at that child. When a sub-tree has been established, the root of that sub-tree sends a report to its parent containing the end-points accessible via that sub-tree. Each internal node establishes its children processes and their respective connections sequentially. However, since the various processes are expected to run on different compute nodes, sub-trees in different branches of the network are created concurrently, maximizing the efficiency of network instantiation.

In the second mode of process instantiation, MRNet relies on a process management system to create some or all of the MRNet processes. This mode accommodates tools that require their back-ends to create, monitor, and control other processes. For example, IBM's POE uses environment variables to pass information, such as the process' rank within the application's global MPI communicator, to the MPI run-time library in each application process. In cases like this, MRNet cannot provide back-end processes with the environment necessary to start MPI application processes. As a result, MRNet creates its internal processes recursively as in the first instantiation mode, but does not instantiate any back-end processes. MRNet then waits for the tool back-ends to be started by the process management system to ensure they have the environment needed to create application processes successfully. To allow back-ends to connect to the MRNet network, information such as process host names and connection port numbers must be provided to the back-ends. This information can be provided via the environment, using shared filesystems or other information services as available on the target system. To collect the necessary information, the front-end can use the MRNet API methods for discovering the network topology details. We show how to construct a tool that requires back-ends to be separately instantiated in $MRNET_ROOT/Examples/NoBackEndInstantiation.

Instances of NetworkTopology are network specific, so they are created when a Network is instantiated. MRNet API users should not need to create their own NetworkTopology instances.

Instances of Communicator are network specific, so their creation methods are functions of an instantiated Network object.

Instances of Stream are network specific, so their creation methods are functions of an instantiated Network object. There are two common approaches to creating a new stream.

A Packet encapsulates a chunk of formatted data sent on a Stream. Packets are created using a format string (e.g., "%s %d" describes a null-terminated string followed by a 32-bit integer, and the Packet is said to contain 2 data elements). MRNet front-end and back-end processes do not create instances of Packet; instead they are automatically produced from the formatted data passed to Stream::send. Appendix A, MRNet Format Strings contains the full listing of data types that can be sent in a Packet.

When receiving a Packet via Stream::recv or Network::recv, the Packet instance is stored within a PacketPtr object. PacketPtr is a class based on the Boost library shared_ptr class, and helps with memory management of Packets. A PacketPtr can be assumed to be equivalent to "Packet *", and all operations on Packets require use of PacketPtr.

The first parameter to the Network::Network() front-end constructor is the name of an MRNet topology file. This file defines the topological layout of the front-end, internal nodes, and back-end MRNet processes. In the syntax of the topology file, the hostname:id tuple represents a process with a MRNet instance id running on hostname. It is important to note that the id is of symbolic value only and does not reflect a port or process number associated with the system. A line in the topology file is always of the form:

hostname1:0 => hostname1:1 hostname1:2 ;

meaning a process on hostname1 with MRNet id 0 has two children, with MRNet ids 1 and 2, running on the same host. MRNet will parse the topology file without error if the file properly defines a tree in the mathematical sense (i.e. a tree must have a single root, no cycles, full connection, and no node can be its own descendant).

   hostname1:0 => 
                  hostname1:1 
                  hostname1:2 
                  ;

When the MRNet test programs are built, a topology generator program, $MRNET_ROOT/bin/$MRNET_ARCH/mrnet_topgen, will also be created. The usage of this program is:

mrnet_topgen [OPTIONS] TOPOLOGY_SPEC [INFILE] [OUTFILE]

Create a MRNet topology from the machines listed in [INFILE],
or standard input, and writes output to [OUTFILE], or standard output.

The format of the input machine list is one machine specification per line,
where each specification is of the form "host[:num-processors]". Note that
the first machine listed should be the host where the front-end should be run.

         OPTIONS:

         -m max-host-procs, --maxprocs=max-host-procs
                 Specify the maximum number of processes to place on any machine,
                 in which case the number of processes allocated to a machine will be
                 the minimum of its processor count and "max-host-procs".

         TOPOLOGIES:

         -b topology, --balanced=topology
                 Create a balanced tree using "topology" specification. The specification
                 is in the format F^D, where F is the fan-out (or out-degree) and D is the
                 tree depth. The number of tree leaves (or back-ends) will be F^D.
                 An alternative specification is FxFxF, where the fan-out at each level
                 is specified explicitly and can differ between levels.

                 Example: "16^3" is a tree of depth 3 with fan-out 16, with 4096 leaves.
                 Example: "2x4x8" is a tree of depth 3 with 64 leaves.

         -k topology, --knomial=topology
                 Create a k-nomial tree using "topology" specification. The specification
                 is in the format K@N, where K is the k-factor (≥2) and N is the total
                 number of tree nodes. The number of tree leaves (or back-ends) will be
                 (N/K)*(K-1).

                 Example: "2@128" is a binomial tree of 128 nodes, with 64 leaves.
                 Example: "3@27" is a trinomial tree of 27 nodes, with 18 leaves.

         -o topology, --other=topology
                 Create a generic tree using "topology" specification. The specification
                 for this option is (the agreeably complicated) N:N,N,N:... where N is
                 the number of children, ',' distinguishes nodes on the same level,
                 and ':' separates the tree into levels.

                 Example: "2:8,4" is a tree where the root has 2 children,
                          the 1st child has 8 children, and the
                          2nd child has 4 children.

A filter function has the following signature:

packets_in is a reference to a vector of Packets serving as input to the filter function. packets_out is a reference to a vector into which output Packets should be placed. In the rare case where Packets need to be sent in the reverse direction on the Stream, packets_out_reverse should be used instead of packets_out. filter_state may be used to define and maintain state specific to a filter instance. Finally, config_params is a reference to a PacketPtr containing the current configuration settings for the filter instance, as can be set using Stream::set_FilterParameters.

For each filter function defined in a shared object file, there must be a const char * symbol named by the string formed by the concatenation of the filter function name and the suffix "_format_string". For instance, if the filter function is named my_filter_func, the shared object must define a symbol const char *my_filter_func_format_string. The value of this string will be the MRNet format string describing the format of data that the filter can operate on. A value of "" denotes that the filter can operate on data of arbitrary value.

MRNet automatically recovers from failures of internal tree processes (i.e., those processes that are not the front-end (root) or back-ends (leaves)). As part of the recovery, MRNet will extract filter state from the children of a failed process and pass that state as input to each child's newly chosen parent. If you have a filter that maintains persistent state using filter_state, you can provide an additional function within the shared object for your filter that MRNet may use to extract the state. The name of this extraction function should be the same as the filter name with the suffix "_get_state" appended. For instance, if the filter function is named my_filter_func, the extraction function should be named my_filter_func_get_state.

A filter state extraction function has the following signature:

filter_state is a pointer to the state defined by the filter for the given Stream identified by stream_id. This function should extract the necessary state and return a Packet that can be passed as input to the filter function. Since the Packet will be processed as a normal input Packet for the filter, it's format must match that expected by the filter. A fault-tolerant filter example is provided in $MRNET_ROOT/Examples/FaultRecovery.

This topic currently pertains to use with the GNU C++ compiler only.

Since we use the C facility dlopen() to dynamically load new filter functions, all C++ symbols must be exported. That is, the symbol definitions must fall with the statements

extern "C" {

and

}

The file that contains the filter functions and format strings may be compiled with the GNU compiler options "-fPIC -shared -rdynamic" to produce a valid shared object.

Front-end and back-end programs that will dynamically load filters must be built with compiler options that notify the linker to export global symbols (for GNU compilers, you can use "-Wl,-E").