Boost C++ Libraries

As a message passing library, MPI's primary purpose is to routine messages from one process to another, i.e., point-to-point. MPI contains routines that can send messages, receive messages, and query whether messages are available. Each message has a source process, a target process, a tag, and a payload containing arbitrary data. The source and target processes are the ranks of the sender and receiver of the message, respectively. Tags are integers that allow the receiver to distinguish between different messages coming from the same sender.

The following program uses two MPI processes to write "Hello, world!" to the screen (hello_world.cpp):

#include <boost/mpi.hpp>
#include <iostream>
#include <string>
#include <boost/serialization/string.hpp>
namespace mpi = boost::mpi;

int main()
{
  mpi::environment env;
  mpi::communicator world;

  if (world.rank() == 0) {
    world.send(1, 0, std::string("Hello"));
    std::string msg;
    world.recv(1, 1, msg);
    std::cout << msg << "!" << std::endl;
  } else {
    std::string msg;
    world.recv(0, 0, msg);
    std::cout << msg << ", ";
    std::cout.flush();
    world.send(0, 1, std::string("world"));
  }

  return 0;
}

The first processor (rank 0) passes the message "Hello" to the second processor (rank 1) using tag 0. The second processor prints the string it receives, along with a comma, then passes the message "world" back to processor 0 with a different tag. The first processor then writes this message with the "!" and exits. All sends are accomplished with the communicator::send method and all receives use a corresponding communicator::recv call.

The default MPI communication operations--send and recv--may have to wait until the entire transmission is completed before they can return. Sometimes this blocking behavior has a negative impact on performance, because the sender could be performing useful computation while it is waiting for the transmission to occur. More important, however, are the cases where several communication operations must occur simultaneously, e.g., a process will both send and receive at the same time.

Let's revisit our "Hello, world!" program from the previous section. The core of this program transmits two messages:

if (world.rank() == 0) {
  world.send(1, 0, std::string("Hello"));
  std::string msg;
  world.recv(1, 1, msg);
  std::cout << msg << "!" << std::endl;
} else {
  std::string msg;
  world.recv(0, 0, msg);
  std::cout << msg << ", ";
  std::cout.flush();
  world.send(0, 1, std::string("world"));
}

The first process passes a message to the second process, then prepares to receive a message. The second process does the send and receive in the opposite order. However, this sequence of events is just that--a sequence--meaning that there is essentially no parallelism. We can use non-blocking communication to ensure that the two messages are transmitted simultaneously (hello_world_nonblocking.cpp):

#include <boost/mpi.hpp>
#include <iostream>
#include <string>
#include <boost/serialization/string.hpp>
namespace mpi = boost::mpi;

int main()
{
  mpi::environment env;
  mpi::communicator world;

  if (world.rank() == 0) {
    mpi::request reqs[2];
    std::string msg, out_msg = "Hello";
    reqs[0] = world.isend(1, 0, out_msg);
    reqs[1] = world.irecv(1, 1, msg);
    mpi::wait_all(reqs, reqs + 2);
    std::cout << msg << "!" << std::endl;
  } else {
    mpi::request reqs[2];
    std::string msg, out_msg = "world";
    reqs[0] = world.isend(0, 1, out_msg);
    reqs[1] = world.irecv(0, 0, msg);
    mpi::wait_all(reqs, reqs + 2);
    std::cout << msg << ", ";
  }

  return 0;
}

We have replaced calls to the communicator::send and communicator::recv members with similar calls to their non-blocking counterparts, communicator::isend and communicator::irecv. The prefix i indicates that the operations return immediately with a mpi::request object, which allows one to query the status of a communication request (see the test method) or wait until it has completed (see the wait method). Multiple requests can be completed at the same time with the wait_all operation.

Important

Regarding communication completion/progress: The MPI standard requires users to keep the request handle for a non-blocking communication, and to call the "wait" operation (or successfully test for completion) to complete the send or receive. Unlike most C MPI implementations, which allow the user to discard the request for a non-blocking send, Boost.MPI requires the user to call "wait" or "test", since the request object might contain temporary buffers that have to be kept until the send is completed. Moreover, the MPI standard does not guarantee that the receive makes any progress before a call to "wait" or "test", although most implementations of the C MPI do allow receives to progress before the call to "wait" or "test". Boost.MPI, on the other hand, generally requires "test" or "wait" calls to make progress. More specifically, Boost.MPI guarantee that calling "test" multiple time will eventually complete the communication (this is due to the fact that serialized communication are potentially a multi step operation.).

If you run this program multiple times, you may see some strange results: namely, some runs will produce:

Hello, world!

while others will produce:

world!
Hello,

or even some garbled version of the letters in "Hello" and "world". This indicates that there is some parallelism in the program, because after both messages are (simultaneously) transmitted, both processes will concurrent execute their print statements. For both performance and correctness, non-blocking communication operations are critical to many parallel applications using MPI.

The broadcast algorithm is by far the simplest collective operation. It broadcasts a value from a single process to all other processes within a communicator. For instance, the following program broadcasts "Hello, World!" from process 0 to every other process. (hello_world_broadcast.cpp)

#include <boost/mpi.hpp>
#include <iostream>
#include <string>
#include <boost/serialization/string.hpp>
namespace mpi = boost::mpi;

int main()
{
  mpi::environment env;
  mpi::communicator world;

  std::string value;
  if (world.rank() == 0) {
    value = "Hello, World!";
  }

  broadcast(world, value, 0);

  std::cout << "Process #" << world.rank() << " says " << value
            << std::endl;
  return 0;
}

Running this program with seven processes will produce a result such as:

Process #0 says Hello, World!
Process #2 says Hello, World!
Process #1 says Hello, World!
Process #4 says Hello, World!
Process #3 says Hello, World!
Process #5 says Hello, World!
Process #6 says Hello, World!

The gather collective gathers the values produced by every process in a communicator into a vector of values on the "root" process (specified by an argument to gather). The /i/th element in the vector will correspond to the value gathered from the /i/th process. For instance, in the following program each process computes its own random number. All of these random numbers are gathered at process 0 (the "root" in this case), which prints out the values that correspond to each processor. (random_gather.cpp)

#include <boost/mpi.hpp>
#include <iostream>
#include <vector>
#include <cstdlib>
namespace mpi = boost::mpi;

int main()
{
  mpi::environment env;
  mpi::communicator world;

  std::srand(time(0) + world.rank());
  int my_number = std::rand();
  if (world.rank() == 0) {
    std::vector<int> all_numbers;
    gather(world, my_number, all_numbers, 0);
    for (int proc = 0; proc < world.size(); ++proc)
      std::cout << "Process #" << proc << " thought of "
                << all_numbers[proc] << std::endl;
  } else {
    gather(world, my_number, 0);
  }

  return 0;
}

Executing this program with seven processes will result in output such as the following. Although the random values will change from one run to the next, the order of the processes in the output will remain the same because only process 0 writes to std::cout.

Process #0 thought of 332199874
Process #1 thought of 20145617
Process #2 thought of 1862420122
Process #3 thought of 480422940
Process #4 thought of 1253380219
Process #5 thought of 949458815
Process #6 thought of 650073868

The gather operation collects values from every process into a vector at one process. If instead the values from every process need to be collected into identical vectors on every process, use the all_gather algorithm, which is semantically equivalent to calling gather followed by a broadcast of the resulting vector.

The scatter collective scatters the values from a vector in the "root" process in a communicator into values in all the processes of the communicator. The /i/th element in the vector will correspond to the value received by the /i/th process. For instance, in the following program, the root process produces a vector of random nomber and send one value to each process that will print it. (random_scatter.cpp)

#include <boost/mpi.hpp>
#include <boost/mpi/collectives.hpp>
#include <iostream>
#include <cstdlib>
#include <vector>

namespace mpi = boost::mpi;

int main(int argc, char* argv[])
{
  mpi::environment env(argc, argv);
  mpi::communicator world;

  std::srand(time(0) + world.rank());
  std::vector<int> all;
  int mine = -1;
  if (world.rank() == 0) {
    all.resize(world.size());
    std::generate(all.begin(), all.end(), std::rand);
  }
  mpi::scatter(world, all, mine, 0);
  for (int r = 0; r < world.size(); ++r) {
    world.barrier();
    if (r == world.rank()) {
      std::cout << "Rank " << r << " got " << mine << '\n';
    }
  }
  return 0;
}

Executing this program with seven processes will result in output such as the following. Although the random values will change from one run to the next, the order of the processes in the output will remain the same because of the barrier.

Rank 0 got 1409381269
Rank 1 got 17045268
Rank 2 got 440120016
Rank 3 got 936998224
Rank 4 got 1827129182
Rank 5 got 1951746047
Rank 6 got 2117359639

The reduce collective summarizes the values from each process into a single value at the user-specified "root" process. The Boost.MPI reduce operation is similar in spirit to the STL accumulate operation, because it takes a sequence of values (one per process) and combines them via a function object. For instance, we can randomly generate values in each process and the compute the minimum value over all processes via a call to reduce (random_min.cpp):

#include <boost/mpi.hpp>
#include <iostream>
#include <cstdlib>
namespace mpi = boost::mpi;

int main()
{
  mpi::environment env;
  mpi::communicator world;

  std::srand(time(0) + world.rank());
  int my_number = std::rand();

  if (world.rank() == 0) {
    int minimum;
    reduce(world, my_number, minimum, mpi::minimum<int>(), 0);
    std::cout << "The minimum value is " << minimum << std::endl;
  } else {
    reduce(world, my_number, mpi::minimum<int>(), 0);
  }

  return 0;
}

The use of mpi::minimum<int> indicates that the minimum value should be computed. mpi::minimum<int> is a binary function object that compares its two parameters via < and returns the smaller value. Any associative binary function or function object will work provided it's stateless. For instance, to concatenate strings with reduce one could use the function object std::plus<std::string> (string_cat.cpp):

#include <boost/mpi.hpp>
#include <iostream>
#include <string>
#include <functional>
#include <boost/serialization/string.hpp>
namespace mpi = boost::mpi;

int main()
{
  mpi::environment env;
  mpi::communicator world;

  std::string names[10] = { "zero ", "one ", "two ", "three ",
                            "four ", "five ", "six ", "seven ",
                            "eight ", "nine " };

  std::string result;
  reduce(world,
         world.rank() < 10? names[world.rank()]
                          : std::string("many "),
         result, std::plus<std::string>(), 0);

  if (world.rank() == 0)
    std::cout << "The result is " << result << std::endl;

  return 0;
}

In this example, we compute a string for each process and then perform a reduction that concatenates all of the strings together into one, long string. Executing this program with seven processors yields the following output:

The result is zero one two three four five six

Binary operations for reduce

Any kind of binary function objects can be used with reduce. For instance, and there are many such function objects in the C++ standard <functional> header and the Boost.MPI header <boost/mpi/operations.hpp>. Or, you can create your own function object. Function objects used with reduce must be associative, i.e. f(x, f(y, z)) must be equivalent to f(f(x, y), z). If they are also commutative (i..e, f(x, y) == f(y, x)), Boost.MPI can use a more efficient implementation of reduce. To state that a function object is commutative, you will need to specialize the class is_commutative. For instance, we could modify the previous example by telling Boost.MPI that string concatenation is commutative:

namespace boost { namespace mpi {

  template<>
  struct is_commutative<std::plus<std::string>, std::string>
    : mpl::true_ { };

} } // end namespace boost::mpi

By adding this code prior to main(), Boost.MPI will assume that string concatenation is commutative and employ a different parallel algorithm for the reduce operation. Using this algorithm, the program outputs the following when run with seven processes:

The result is zero one four five six two three

Note how the numbers in the resulting string are in a different order: this is a direct result of Boost.MPI reordering operations. The result in this case differed from the non-commutative result because string concatenation is not commutative: f("x", "y") is not the same as f("y", "x"), because argument order matters. For truly commutative operations (e.g., integer addition), the more efficient commutative algorithm will produce the same result as the non-commutative algorithm. Boost.MPI also performs direct mappings from function objects in <functional> to MPI_Op values predefined by MPI (e.g., MPI_SUM, MPI_MAX); if you have your own function objects that can take advantage of this mapping, see the class template is_mpi_op.

	Warning
	Due to the underlying MPI limitations, it is important to note that the operation must be stateless.

All process variant

Like gather, reduce has an "all" variant called all_reduce that performs the reduction operation and broadcasts the result to all processes. This variant is useful, for instance, in establishing global minimum or maximum values.

The following code (global_min.cpp) shows a broadcasting version of the random_min.cpp example:

#include <boost/mpi.hpp>
#include <iostream>
#include <cstdlib>
namespace mpi = boost::mpi;

int main(int argc, char* argv[])
{
  mpi::environment env(argc, argv);
  mpi::communicator world;

  std::srand(world.rank());
  int my_number = std::rand();
  int minimum;

  mpi::all_reduce(world, my_number, minimum, mpi::minimum<int>());

  if (world.rank() == 0) {
      std::cout << "The minimum value is " << minimum << std::endl;
  }

  return 0;
}

In that example we provide both input and output values, requiring twice as much space, which can be a problem depending on the size of the transmitted data. If there is no need to preserve the input value, the output value can be omitted. In that case the input value will be overridden with the output value and Boost.MPI is able, in some situation, to implement the operation with a more space efficient solution (using the MPI_IN_PLACE flag of the MPI C mapping), as in the following example (in_place_global_min.cpp):

#include <boost/mpi.hpp>
#include <iostream>
#include <cstdlib>
namespace mpi = boost::mpi;

int main(int argc, char* argv[])
{
  mpi::environment env(argc, argv);
  mpi::communicator world;

  std::srand(world.rank());
  int my_number = std::rand();

  mpi::all_reduce(world, my_number, mpi::minimum<int>());

  if (world.rank() == 0) {
      std::cout << "The minimum value is " << my_number << std::endl;
  }

  return 0;
}

Communication with Boost.MPI always occurs over a communicator. A communicator contains a set of processes that can send messages among themselves and perform collective operations. There can be many communicators within a single program, each of which contains its own isolated communication space that acts independently of the other communicators.

When the MPI environment is initialized, only the "world" communicator (called MPI_COMM_WORLD in the MPI C and Fortran bindings) is available. The "world" communicator, accessed by default-constructing a mpi::communicator object, contains all of the MPI processes present when the program begins execution. Other communicators can then be constructed by duplicating or building subsets of the "world" communicator. For instance, in the following program we split the processes into two groups: one for processes generating data and the other for processes that will collect the data. (generate_collect.cpp)

#include <boost/mpi.hpp>
#include <iostream>
#include <cstdlib>
#include <boost/serialization/vector.hpp>
namespace mpi = boost::mpi;

enum message_tags {msg_data_packet, msg_broadcast_data, msg_finished};

void generate_data(mpi::communicator local, mpi::communicator world);
void collect_data(mpi::communicator local, mpi::communicator world);

int main()
{
  mpi::environment env;
  mpi::communicator world;

  bool is_generator = world.rank() < 2 * world.size() / 3;
  mpi::communicator local = world.split(is_generator? 0 : 1);
  if (is_generator) generate_data(local, world);
  else collect_data(local, world);

  return 0;
}

When communicators are split in this way, their processes retain membership in both the original communicator (which is not altered by the split) and the new communicator. However, the ranks of the processes may be different from one communicator to the next, because the rank values within a communicator are always contiguous values starting at zero. In the example above, the first two thirds of the processes become "generators" and the remaining processes become "collectors". The ranks of the "collectors" in the world communicator will be 2/3 world.size() and greater, whereas the ranks of the same collector processes in the local communicator will start at zero. The following excerpt from collect_data() (in generate_collect.cpp) illustrates how to manage multiple communicators:

mpi::status msg = world.probe();
if (msg.tag() == msg_data_packet) {
  // Receive the packet of data
  std::vector<int> data;
  world.recv(msg.source(), msg.tag(), data);

  // Tell each of the collectors that we'll be broadcasting some data
  for (int dest = 1; dest < local.size(); ++dest)
    local.send(dest, msg_broadcast_data, msg.source());

  // Broadcast the actual data.
  broadcast(local, data, 0);
}

The code in this except is executed by the "master" collector, e.g., the node with rank 2/3 world.size() in the world communicator and rank 0 in the local (collector) communicator. It receives a message from a generator via the world communicator, then broadcasts the message to each of the collectors via the local communicator.

For more control in the creation of communicators for subgroups of processes, the Boost.MPI group provides facilities to compute the union (|), intersection (&), and difference (-) of two groups, generate arbitrary subgroups, etc.

A communicator can be organised as a cartesian grid, here a basic example:

#include <vector>
#include <iostream>

#include <boost/mpi/communicator.hpp>
#include <boost/mpi/collectives.hpp>
#include <boost/mpi/environment.hpp>
#include <boost/mpi/cartesian_communicator.hpp>

#include <boost/test/minimal.hpp>

namespace mpi = boost::mpi;
int test_main(int argc, char* argv[])
{
  mpi::environment  env;
  mpi::communicator world;

  if (world.size() != 24)  return -1;
  mpi::cartesian_dimension dims[] = {{2, true}, {3,true}, {4,true}};
  mpi::cartesian_communicator cart(world, mpi::cartesian_topology(dims));
  for (int r = 0; r < cart.size(); ++r) {
    cart.barrier();
    if (r == cart.rank()) {
      std::vector<int> c = cart.coordinates(r);
      std::cout << "rk :" << r << " coords: "
                << c[0] << ' ' << c[1] << ' ' << c[2] << '\n';
    }
  }
  return 0;
}

To obtain optimal performance for small fixed-length data types not containing any pointers it is very important to mark them using the type traits of Boost.MPI and Boost.Serialization.

It was already discussed that fixed length types containing no pointers can be using as is_mpi_datatype, e.g.:

namespace boost { namespace mpi {
  template <>
  struct is_mpi_datatype<gps_position> : mpl::true_ { };
} }

or the equivalent macro

BOOST_IS_MPI_DATATYPE(gps_position)

In addition it can give a substantial performance gain to turn off tracking and versioning for these types, if no pointers to these types are used, by using the traits classes or helper macros of Boost.Serialization:

BOOST_CLASS_TRACKING(gps_position,track_never)
BOOST_CLASS_IMPLEMENTATION(gps_position,object_serializable)

More optimizations are possible on homogeneous machines, by avoiding MPI_Pack/MPI_Unpack calls but using direct bitwise copy. This feature is enabled by default by defining the macro BOOST_MPI_HOMOGENEOUS in the include file boost/mpi/config.hpp. That definition must be consistent when building Boost.MPI and when building the application.

In addition all classes need to be marked both as is_mpi_datatype and as is_bitwise_serializable, by using the helper macro of Boost.Serialization:

BOOST_IS_BITWISE_SERIALIZABLE(gps_position)

Usually it is safe to serialize a class for which is_mpi_datatype is true by using binary copy of the bits. The exception are classes for which some members should be skipped for serialization.