Kollektiver Datenaustausch

Den in diesem Kapitel bisher vorgestellten Funktionen ist gemein, dass es eine Eins-zu-eins-Beziehung zwischen Prozessen gibt: Es gibt einen Prozess, der sendet, und einen Prozess, der empfängt. Die Verknüpfung erfolgt über den Tag. In diesem Abschnitt lernen Sie Funktionen kennen, die mit den gleichen Parametern für mehrere Prozesse aufgerufen werden können und für den einen Prozess Datenversand und für den anderen Prozess Datenempfang bedeuten. Diese Funktionen werden als kollektive Operationen bezeichnet.

#include <boost/mpi.hpp>
#include <boost/serialization/string.hpp>
#include <vector>
#include <string>
#include <iterator>
#include <algorithm>
#include <iostream>

int main(int argc, char *argv[])
{
  boost::mpi::environment env{argc, argv};
  boost::mpi::communicator world;
  if (world.rank() == 0)
  {
    std::vector<std::string> v;
    boost::mpi::gather<std::string>(world, "", v, 0);
    std::ostream_iterator<std::string> out{std::cout, "\n"};
    std::copy(v.begin(), v.end(), out);
  }
  else if (world.rank() == 1)
  {
    boost::mpi::gather(world, std::string{"Hello, world!"}, 0);
  }
  else if (world.rank() == 2)
  {
    boost::mpi::gather(world, std::string{"Hello, moon!"}, 0);
  }
}

Im Beispiel 47.9 wird für mehrere Prozesse die Funktion boost::mpi::gather() aufgerufen. Ob diese Funktion Daten sendet oder empfängt, hängt von den Parametern ab, die ihr übergeben werden.

Die Prozesse mit den Rängen 1 und 2 nutzen boost::mpi::gather() zum Datenversand. Dazu übergeben sie die zu sendenden Daten – hier die Strings „Hello, world!“ und „Hello, moon!“ – und den Rang des Prozesses, an den die Daten gesendet werden sollen. Da es sich bei boost::mpi::gather() um eine Funktion und keine Methode handelt, muss außerdem der Kommunikator world übergeben werden.

Der Prozess mit Rang 0 ruft boost::mpi::gather() zum Datenempfang auf. Da die empfangenen Daten irgendwo gespeichert werden müssen, wird mit v ein Objekt vom Typ std::vector<std::string> übergeben. Beachten Sie, dass Sie für boost::mpi::gather() diesen Typ verwenden müssen. Es werden keine anderen Container oder String-Typen unterstützt.

Es müssen im Prozess mit Rang 0 außerdem die gleichen Parameter übergeben werden, die von den Prozessen mit Rang 1 und 2 verwendet werden. So wird auch im Prozess mit Rang 0 world, ein zu sendender String und 0 an boost::mpi::gather() übergeben.

Wenn Sie Beispiel 47.9 ausführen, wird Hello, world! und Hello, moon! in zwei Zeilen auf die Standardausgabe ausgegeben. Wenn Sie sich die Ausgabe genau ansehen, stellen Sie fest, dass jedoch zuerst eine leere Zeile ausgegeben wird. In dieser ersten Zeile wird der leere String ausgegeben, der vom Prozess mit Rang 0 an boost::mpi::gather() übergeben wird. Im Vektor v befinden sich also drei Strings, die von den Prozessen mit den Rängen 0, 1 und 2 empfangen wurden. Die Daten werden dabei gemäß den Rängen der Prozesse abgelegt. Die Ränge der Prozesse werden als Index verwendet, um die Daten im Vektor zu speichern. Wenn Sie das Beispiel mehrfach ausführen, erhalten Sie immer an erster Stelle im Vektor den leeren String, an zweiter Stelle den String „Hello, world!“ und an den dritter Stelle den String „Hello, moon!“.

Beachten Sie, dass Sie Beispiel 47.9 nicht mit mehr als drei Prozessen ausführen dürfen. Wenn Sie beim Aufruf von mpiexec zum Beispiel -n 4 angeben, gibt das Programm keine Daten aus. Es bleibt außerdem hängen und muss mit CTRL+C abgebrochen werden.

Kollektive Operationen müssen immer für alle Prozesse ausgeführt werden. Wenn Ihr Programm eine Funktion wie boost::mpi::gather() aufruft, müssen Sie sicherstellen, dass der Aufruf für alle Prozesse erfolgt. Sie verletzen sonst den MPI-Standard. Da eine Funktion wie boost::mpi::gather() für alle Prozesse ausgerufen werden muss, erfolgt der Aufruf üblicherweise nicht wie im Beispiel 47.9 mehrfach für unterschiedliche Prozesse. Sehen Sie sich dazu Beispiel 47.10 an, das grundsätzlich genauso funktioniert wie Beispiel 47.9.

#include <boost/mpi.hpp>
#include <boost/serialization/string.hpp>
#include <vector>
#include <string>
#include <iterator>
#include <algorithm>
#include <iostream>

int main(int argc, char *argv[])
{
  boost::mpi::environment env{argc, argv};
  boost::mpi::communicator world;
  std::string s;
  if (world.rank() == 1)
    s = "Hello, world!";
  else if (world.rank() == 2)
    s = "Hello, moon!";
  std::vector<std::string> v;
  boost::mpi::gather(world, s, v, 0);
  std::ostream_iterator<std::string> out{std::cout, "\n"};
  std::copy(v.begin(), v.end(), out);
}

Sie rufen Funktionen, die kollektive Operationen anstoßen, üblicherweise für alle Prozesse auf. Diese Funktionen sind dabei so definiert, dass klar ist, welche Aufgaben die Prozesse zu erledigen haben – obwohl alle Prozesse die gleichen Parameter beim Aufruf übergeben.

Im Beispiel 47.10 wird boost::mpi::gather() verwendet. Diese Funktion – der Name deutet es an – sammelt Daten. Die Daten werden dabei in dem Prozess gesammelt, dessen Rang als letzter Parameter an boost::mpi::gather() übergeben wird. Dieser Prozess sammelt Daten, die er von allen Prozessen erhält. Der Vektor, in dem die Daten gespeichert werden sollen, wird ausschließlich von dem Prozess verwendet, der die Daten sammeln soll.

Beachten Sie, dass boost::mpi::gather() Daten von allen Prozessen sammelt. Dies schließt den Prozess mit ein, in dem die Daten gesammelt werden sollen. Im Beispiel 47.10 ist dies der Prozess mit Rang 0. Dieser Prozess sendet beim Aufruf von Beispiel 47.10 einen leeren String in der Variablen s an sich selbst. Der leere String wird im Vektor v abgelegt. Wie Sie anhand der folgenden Beispiele sehen werden, schließen kollektive Operationen immer alle Prozesse ein.

Sie können Beispiel 47.10 mit beliebig vielen Prozessen ausführen, da für jeden Prozess ein Aufruf von boost::mpi::gather() stattfindet. Wenn Sie das Beispiel mit drei Prozessen starten, erhalten Sie die gleiche Datenausgabe wie beim vorherigen Beispiel.

#include <boost/mpi.hpp>
#include <boost/serialization/string.hpp>
#include <vector>
#include <string>
#include <iostream>

int main(int argc, char *argv[])
{
  boost::mpi::environment env{argc, argv};
  boost::mpi::communicator world;
  std::vector<std::string> v{"Hello, world!", "Hello, moon!",
    "Hello, sun!"};
  std::string s;
  boost::mpi::scatter(world, v, s, 0);
  std::cout << world.rank() << ": " << s << '\n';
}

Beispiel 47.11 stellt die Funktion boost::mpi::scatter() vor. Es handelt sich dabei um das Gegenstück zu boost::mpi::gather(). Während boost::mpi::gather() Daten von mehreren Prozessen in einem Prozess sammelt, verteilt boost::mpi::scatter() Daten von einem Prozess an mehrere Prozesse.

Im Beispiel 47.11 werden die Daten im Vektor v vom Prozess mit Rang 0 an alle Prozesse verteilt. Dies schließt den eigenen Prozess – also den mit Rang 0 – ein. Der String „Hello, world!“ wird vom Prozess mit Rang 0 in der Variablen s empfangen. Der Prozess mit Rang 1 empfängt „Hello, moon!“ in dieser Variablen. Der Prozess mit Rang 2 erhält „Hello, sun!“ in s.

#include <boost/mpi.hpp>
#include <boost/serialization/string.hpp>
#include <string>
#include <iostream>

int main(int argc, char *argv[])
{
  boost::mpi::environment env{argc, argv};
  boost::mpi::communicator world;
  std::string s;
  if (world.rank() == 0)
    s = "Hello, world!";
  boost::mpi::broadcast(world, s, 0);
  std::cout << s << '\n';
}

boost::mpi::broadcast() sendet Daten von einem Prozess an alle Prozesse. Der Unterschied zu boost::mpi::scatter() ist, dass die gleichen Daten an alle Prozesse gesendet werden. So erhalten im Beispiel 47.12 alle Prozesse den String „Hello, world!“ in der Variablen s. Alle Prozesse geben Hello, world! auf die Standardausgabe aus.

#include <boost/mpi.hpp>
#include <boost/serialization/string.hpp>
#include <string>
#include <iostream>

std::string min(const std::string &lhs, const std::string &rhs)
{
  return lhs.size() < rhs.size() ? lhs : rhs;
}

int main(int argc, char *argv[])
{
  boost::mpi::environment env{argc, argv};
  boost::mpi::communicator world;
  std::string s;
  if (world.rank() == 0)
    s = "Hello, world!";
  else if (world.rank() == 1)
    s = "Hello, moon!";
  else if (world.rank() == 2)
    s = "Hello, sun!";
  std::string result;
  boost::mpi::reduce(world, s, result, min, 0);
  if (world.rank() == 0)
    std::cout << result << '\n';
}

boost::mpi::reduce() sammelt so wie boost::mpi::gather() Daten von mehreren Prozessen. Diese werden jedoch nicht in einem Vektor abgelegt. boost::mpi::reduce() erwartet, dass eine Funktion oder ein Funktionsobjekt übergeben wird, das zur Auswertung der Daten verwendet wird.

Im Beispiel 47.13 erhält der Prozess mit Rang 0 als Ergebnis in der Variablen result den String „Hello, sun!“. Der Aufruf von boost::mpi::reduce() führt dazu, dass die Strings, die die verschiedenen Prozesse in s an die Funktion übergeben, gesammelt und ausgewertet werden. Die Auswertung findet mit Hilfe der Funktion min() statt, die boost::mpi::reduce() als vierter Parameter übergeben wird. Diese Funktion vergleicht jeweils zwei Strings und gibt den kürzeren der beiden als Ergebnis zurück.

Wenn Sie Beispiel 47.13 mit drei Prozessen ausführen, wird Hello, sun! auf die Standardausgabe ausgegeben. Starten Sie mehr als drei Prozesse, wird eine leere Zeile ausgegeben, da alle Prozesse mit einem Rang größer als 2 in der Variablen s einen leeren String an boost::mpi::reduce() übergeben, der kürzer ist als alle anderen Strings.

#include <boost/mpi.hpp>
#include <boost/serialization/string.hpp>
#include <string>
#include <iostream>

std::string min(const std::string &lhs, const std::string &rhs)
{
  return lhs.size() < rhs.size() ? lhs : rhs;
}

int main(int argc, char *argv[])
{
  boost::mpi::environment env{argc, argv};
  boost::mpi::communicator world;
  std::string s;
  if (world.rank() == 0)
    s = "Hello, world!";
  else if (world.rank() == 1)
    s = "Hello, moon!";
  else if (world.rank() == 2)
    s = "Hello, sun!";
  std::string result;
  boost::mpi::all_reduce(world, s, result, min);
  std::cout << world.rank() << ": " << result << '\n';
}

Im Beispiel 47.14 kommt die Funktion boost::mpi::all_reduce() zum Einsatz. Diese Funktion sammelt und wertet wie boost::mpi::reduce() Daten aus. Der Unterschied zwischen den beiden Funktionen ist, dass boost::mpi::all_reduce() das Ergebnis an alle Prozesse verteilt, während boost::mpi::reduce() das Ergebnis nur an den Prozess weitergibt, dessen Rang als letzter Parameter übergeben wurde. Wie Sie anhand des Funktionsaufrufs im Beispiel 47.14 sehen, wird an boost::mpi::all_reduce() kein Rang übergeben.

Wenn Sie Beispiel 47.14 mit drei Prozessen ausführen, gibt jeder Prozess Hello, sun! auf die Standardausgabe aus.