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

MRNet has been developed to be highly portable; there is no reason why it should not run properly on all common Unix-based platforms. This being said, we have successfully built and tested MRNet using GCC version 3 compilers on the following systems:

We are currently upgrading our build system to allow the use of native compilers where appropriate, for instance, xlc and xlC in AIX environments.

Here we list the third party tools that MRNet uses and needs for proper installation:

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-libfl                Link line for flex library
  --with-outputlevel          output level (0=>off, *1=>error*, 2=>info, 3=>lo-debug, 4=>mid-debug, 5=>hi-debug)

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

UNIX> ./configure CXX=g++ CXXFLAGS=-O3 -with-outputlevel=5 --with-libfl=/usr/local/lib/libfl.a

instructs the configure script to use g++ for the C++ compiler with level 3 optimization, very verbose output and /usr/local/lib/libfl.a as the location of the flex library.

To build the MRNet toolkit by type:

UNIX>  make

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

MRNet is maintained primarily by Dorian Arnold, Paradyn Project, University of Wisconsin-Madison. Comments and other feedback whether positive or negative are welcome.

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

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

MRNet uses communicators to represent groups of network 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 tool front-end. Accordingly, MRNet communicators are created and managed by the front-end, and communication is only allowed between a tool's front-end and its back-ends, i.e. back-ends cannot interact with each other directly via MRNet.

A stream is a logical channel that connects the front-end to the end-points of a communicator. All tool-level communication via MRNet must use these streams. Streams carry data packets downstream, from the front-end toward the back-ends, and upstream, from the back-ends toward the front-end. Upward 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 specification to allow for specifiers that describe arrays of integers and floats.

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.

Filters operate on data flowing upstream in the network. Synchronization filters receive packets one at a time and do not output any packets until the specified synchronization criteria has occurred. Transformation filters input the group of synchronized packets, perform some type of data transformation on the data contained in the packets and output one or more 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 provide a mechanism to deal with the asynchronous rrival 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 three synchronization modes:

Transformation filters 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 can carry state from one transformation to the next using static storage structures. MRNet provides several transformation filters that should be of general use:

Chapter 7, Adding New Filters describes facilities a tool developer may use to add new filters to the provided set.

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, provided by libmrnet, contains network, end-point, 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 end-point objects that represent available tool back-ends. The communicator object is a container for groups of end-points, and streams are used to send data to the end-points in a communicator.

   front_end_main(){
1.   MR_Network * net;
2.   MR_Communicator * comm;
3.   MR_Stream * stream;
4.   float result;
5.   net = new MR_Network(config_file);
6.   comm = net->get_broadcast_communicator( );
7.   stream = new MR_Stream(comm, FMAX_FIL);
8.   stream->send("%d", FLOAT_MAX_INIT);
9.   stream->recv("%f", result);
   }

A simplified version of code from an example tool front-end is shown in Figure 4.1. In the front-end code, after some variable definitions in lines 1-4, in line 5, an instance of the MRNet network is created using the topology specification in config_file. In line 6, the newly created network object is queried for an auto-generated broadcast communicator that contains all available end-points. In line 7, this communicator is used to established a stream for which the MRNet internal processes will use a filter that finds the maximum floating point data value of the data sent upstream. The front-end then might send one or more initialization messages to the backends; in our example code on line 9, we broadcast an integer initializer and await the single floating point value result.

   back_end_main(){
1.   MR_Stream * stream;
2.   int val;
3.   MR_Network::init_backend( );
4.   MR_Stream::recv("%d", &val, &stream);
5.   if(val == FLOAT_MAX_INIT){
6.      stream->send("%f", rand_float);
     }
   }

Figure 4.2 shows the code for the back-end that reciprocates the actions of the front-end. Each tool back-end first connects to the appropriate internal process, via the init_backend call in line 3. While the front-end makes a stream-specific recv call, the back-ends make a stream-anonymous recv that returns the integer sent by the front-end along with 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 in Appendix D, A Complete MRNet Example.

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 rsh or ssh 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 in 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 the application 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 instantiates any back-end processes. MRNet then starts the tool back-ends using the process management system to ensure they have the environment needed to create application processes successfully. When starting the back-ends, MRNet must provide them with the information needed to connect to the MRNet internal process tree, such as the leaf processes' host names and connection port numbers. This information is provided via the environment, using shared filesystems or other information services as available on the target system.

Currently, MRNet has a primitive asynchronous error notification based on events. Whenever errors, failures, or other significant events take place at internal or back-end nodes, an event is propagated toward the front-end. This event is placed in a queue which the user can access to determine the systems state. Eventually, MRNet will migrate to error semantics this will give the user more immediate and responsive error notification, including the use of callbacks into the user's space. For now, we describe the portion of the event infrastructure relevant to the user.

The first parameter to the Network::new_Network() function is the name of an MRNet topology file. This file defines the topological layout of the front-end, internal nodes, and back-end MRNet and tool processes. In the syntax of the topology file, the hostname:id tuple represents a process with MRNet id 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 process on hostname1 with MRNet id, 0, has two children, with MRNet ids, 1 and 2, respectively, 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).

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 <infile> <outfile> <befile> <num_backends> <fan-out>

infile is a machine file containing a set of MRNet host/process identifiers in the format, hostname:id, described above. outfile is the name of the MRNet topology file to be created by the generator. befile must be specified, but can be ignored. Finally, num_backends and fan-out define the number of backends and the fan-out to be used at the front-end and all internal nodes, respectively. The specified input machine file must contain enough unique host/process tuples to support the entire process tree. Currently, this program can only build completely balanced trees with the same fan-out at each parent node.

so_file is the path to a shared object file that contains the filter function to be loaded and func is the name of the function to be loaded. The last parameter is_transformation_filter defaults to true and can usually be omitted since the common case is to load transformation, not synchronization, filters. Additionally, the shared object file must contain 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.

On success, Stream::load_FilterFunc returns the id of the newly loaded filter which may be used in subsequent calls to Stream::new_Stream(). A value of -1 is returned on failure.

To properly demonstrate MRNet filters, we must discuss MRNet's packet abstraction in more detail. A packet encapsulates a chunk of formatted data, usually created as the result of a Stream::send() call. Let's say a packet was created using the format string "%s %d" describing a null-terminated string, followed by a 32-bit integer, the packet is said to contain 2 data elements, of those types respectively. In the packet class, the operator[] is overloaded so that packet[i] conveniently returns the i^th data element in the packet. Furthermore, the data element abstraction contains accessor functions to return proper data values. In the above example, packet[0].get_string() returns the value of the string in the packet and packet[1].get_int32_t() returns the value of the integer that represents the second data element. It is the filter developers responsibility to properly define the format of the data expected by and determine how to access respective elements. Appendix B, MRNET Data Elements contains the full listings of data types and accessor functions available to MRNet filter functions.

A filter function has the following signature

packets_in is the reference to a vector of packets serving as input to the filter function. packets_out is the reference to a vector into which output packets should be placed. local_storage may be used to define and maintain filter-instance specific state.

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.

This topic currently pertains to usage with the GNU C++ compiler only. We will update the topic to discuss using other compilers as well.

Since we use the C facility dlopen() to dynamically load new filter functions, all symbols to be exported this way must be "extern C'd". 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.

A front-end that will dynamically load filters must be built with the GNU compiler options "-Wl,-E" to notify the linker export global symbols externally.