Warten auf den Zufall

Man sollte meinen, dass in Betriebssystemen und Programmiersprachen die Einbindung verteilter Ressourcen heute ähnlich selbstverständlich ist wie die Darstellung von Grafik auf dem Bildschirm. Auch hier werden komplexe und von der jeweiligen Umgebung abhängige Arbeitsschritte durchlaufen, von denen der Online-Spieler höchstens das Zucken seines virtuellen Monsters nach dem Schuss eines anderen Monsters bemerkt. Wenn beide Monster sich auf verschiedenen Rechnern befinden, muss man sich fragen, wie sie miteinander kommunizieren. Vielleicht wird, möglicherweise auf einem dritten Rechner, eine Server-Instanz den Weg von Kugel und Monster 2 verfolgen, um festzustellen, ob sich beide zu irgendeinem Zeitpunkt am selben Ort befanden. Dies setzt den Austausch von Nachrichten voraus. Und genau diese Kommunikation erfordert leider oft noch erhebliche Feinarbeit.

Dieser dritte Teil des Boost-Tutorials beschäftigt sich mit der Frage, wie man unter C++ mit Boost die Kommunikation zwischen verteilten Programmen gestaltet. Um dieses Hauptthema ranken sich Informationen zur Serialisierung von Objekten und zur Generierung von Zufallszahlen.

Nicht alles dem Zufall überlassen

Von Computerprogrammen erwartet man voraussagbares Verhalten, wenn sie mit den gleichen Randbedingungen starten. So etwas wie „echte Zufallszahlen“ bekommt man von einem Computer also nur dann, wenn der Algorithmus Zugriff auf eine Quelle nicht vorhersagbarer Werte hat. In der Realität wird man sich meist mit weniger zufriedengeben müssen.

Generatoren liefern ihre Zahlen oft als gleich verteilte Integer-Zahlen im gesamten Wertebereich des Basistyps oder als ebenfalls gleich verteilte Floatingpoint-Zahlen auf einem Intervall (meist [0,1] ). Viele Anwendungen benötigen aber spezielle Verteilungen. So verlangen beispielsweise Evolutionsstrategien zufällige Gleitkommazahlen, die eine Gaußverteilung aufweisen.

Hier hilft Boost.Random (Header: <boost/random.hpp>), eine Sammlung unterschiedlicher Generatoren, die sich mit Klassen zur Berechnung vieler Verteilungen kombinieren lassen. Das Prinzip ist einfach, die Beschreibung auf der Boost-Hompage (siehe „Onlinequellen“ [a]) allerdings sehr technisch.

Listing 1 liefert eine Einführung in diese Thematik. Zunächst erzeugt es jeweils einen Generator für Zufallszahlen nach dem Mersenne-Twister- und dem Linear-Congruential-Algorithmus. Den Startwert bezieht das Programm jeweils aus der aktuellen Zeit.

int main(int argc, char** argv){   
  // Mersenne Twister
  mt19937 mersenne(static_cast<unsigned int> (time(0)));
  // Linear Congruential
  minstd_rand0 linearCongruential(static_cast<unsigned int> (time(0)));
  // Uniform distribution
  uniform_real<double> evenDist(0.,7.);
  // Normal/Gaussian distribution
  normal_distribution<double> normalDist(0.,1.); 
  // Combination of generators with distributions
  variate_generator<mt19937&, uniform_real<double> > 
                          even(mersenne, evenDist);
  variate_generator<minstd_rand0&, normal_distribution<double> > 
                          gauss(linearCongruential, normalDist);
  // Output
  cout.precision(2);
  cout << "Even distribution [0,7[ :" << endl;
  for(unsigned int i=0; i<10; i++) cout << fixed << even() << " ";
  cout << endl;
  cout << "Normal distribution with mean 0 and sigma 1 :" << endl;
  for(unsigned int i=0; i<10; i++) cout << gauss() << " ";
  cout << endl;
  return 0;
}

Even distribution [0,7] :
1.50 1.82 0.94 3.03 5.35 2.61 1.96 4.16 1.71 3.81
Normal distribution with mean 0 and sigma 1 :
0.12 1.98 -0.41 0.48 -1.83 0.28 -0.54 -1.21 0.68 -0.19

Es folgen Objekte für eine Gleich- und eine Normalverteilung. Erstere soll Werte im Bereich [0,7] ausgeben, die Normalverteilung solche mit dem Mittelwert 0 und einer kumulierten Breite von 1.

Der nächste Schritt verbindet Generator und Verteilung, indem er sie als Argumente an ein Objekt des Typs variate_generator übergibt. So wird ein Funktionsobjekt definiert, das beim Aufruf Zufallszahlen der gewählten Verteilung zurückgibt. Die Zeilen in Abbildung 1 demonstrieren dies. Aus Gründen der Darstellung sind die Nachkommastellen auf 2 beschränkt.

Man sollte einige Zeit einplanen, um sich in die Thematik einzuarbeiten. Zum einen erfordert die Beschreibung auf der Webseite einiges an Vorkenntnissen. Zum anderen kommt der Wahl des Algorithmus eine entscheidende Bedeutung zu. Hier muss eine Abwägung zwischen der Geschwindigkeit des Generators und der Qualität der erzeugten Zufallszahlen erfolgen.

Es sei noch darauf hingewiesen, dass die uniform_real-Verteilung laut ihrer Beschreibung einen Fehler aufweist. In den Tests war sie jedoch problemlos zu verwenden. Die Entwicklung von Boost ist so dynamisch, dass dieser Fehler in absehbarer Zeit behoben sein sollte.

Man stelle sich nun eine Rechenumgebung mit häufigen Stromausfällen vor. In solchen Fällen wäre es vorteilhaft, den Zustand einzelner Objekte auf der Festplatte zwischenzuspeichern, um sie zu einem späteren Zeitpunkt aus diesen Daten wieder rekonstruieren zu können. Diese Umwandlung zwischen Objekten und Text- oder Binärdarstellungen leistet Boost.Serialization.

Serienproduktion über den Stream-Operator

In Standardsituationen ist die Anwendung denkbar einfach. Listing 2 demonstriert zunächst die Serialisierung eines vector mit drei Zahlen.

int main(int argc, char **argv){ 
  vector<int> simple;
  // Fill vector<int> with values
  for(int i=0; i<3; i++) simple.push_back(i);
  // Serialize
  ostringstream oss;
  {
    xml_oarchive oa(oss);
    oa << make_nvp("simple",simple);
  }
  // Let the audience know
  cout << "XML description of \"simple:\"" << endl << endl
       << oss.str() << endl;
  // Fill XML code back into another vector<int>
  vector<int> simple2;
  istringstream iss(oss.str());
  {
    xml_iarchive ia(iss);
    ia >> make_nvp("simple",simple2);
  }
  // Show the content of the new object
  cout << "Content of \"simple2:\"" << endl;
  for(int i=0; i<simple2.size(); i++) cout << simple2[i] << " ";
  cout << endl;
}

<?xml version="1.0" encoding="UTF-8" standalone="yes" ?>
<!DOCTYPE boost_serialization>
<boost_serialization signature="serialization::archive" version="4">
<simple>
        <count>3</count>
        <item_version>0</item_version>
        <item>0</item>
        <item>1</item>
        <item>2</item>
</simple>
</boost_serialization>

Boost.Serialization umfasst verschiedene Archivklassen, deren Konstruktoren jeweils ein ostream- respektive istream-Argument oder davon abgeleitete Klassen erwarten (je nachdem, ob es sich um die Generierung oder das Lesen eines Archivs handelt). Dies würde es erlauben, als Zielmedium etwa einen ofstream zu verwenden und die Ausgabe direkt in eine Datei auf der Festplatte umzuleiten. Stattdessen soll hier aber ein stringstream einen direkten Zugriff auf die serialisierte Darstellung des vector ermöglichen.

Da das Archiv im XML-Format vorliegen soll, wird ein xml_oarchive-Objekt erzeugt, dem ein ostringstream übergeben wird. Neben der XML-Darstellung von Objekten existiert eine eigene Darstellung ohne die XML-typischen Tags. Sie besitzt damit deutlich weniger Overhead, ist allerdings für manuelle Änderungen wenig geeignet. Die Speicherung der Objektinformationen im Binärformat ist eine weitere Alternative.

Die eigentliche Serialisierung erfolgt über den Stream-Operator: oa << make_nvp(„simple“,simple);. make_nvp steht für „make name/value pair“ und ordnet einem Objekt die XML-typischen Tags zu.

Boost.Serialization kennt die Algorithmen der Standard Template Library. Die Bibliothek besitzt zudem eine umfassende Sammlung weiterer Datenstrukturen insbesondere aus dem Boost-Umfeld. Entwickler müssen sich daher nicht die Mühe machen, einen shared_ptr zu beschreiben. Das darin referenzierte Objekt muss natürlich serialisierbar sein. Auch im vorliegenden Beispiel ist es nicht notwendig, den Serialisierer über die inneren Strukturen eines vector aufzuklären.

Wichtig sind die geschweiften Klammern um die Serialisierung. Sie sorgen für die Zerstörung der xml_oarchive-Objekts nach der Serialisierung. Hierbei wird der Destruktor von xml_oarchive aufgerufen, der die passenden schließenden Tags in das XML-Archiv schreibt.

Der nächste Schritt gibt den serialisierten Vektor auf der Konsole aus. Nach dem Header - er markiert die Ausgabe als Boost.Serialization-Archiv einer gegebenen Version - folgt die XML-Repräsentation des vector. Das abschließende </boost_serialization> würde fehlen, wenn nicht geschweifte Klammern die Serialisierung umschließen würden.

Anschließend wird ein neuer (leerer) vector<int> erzeugt, in den das XML-Archiv über ein istringstream-Objekt hineingeladen wird. Dabei muss der Anwender die Details dieses Prozesses nicht kennen. Er verwendet einfach den Stream-Operator „>“.

Zu guter Letzt erfolgt die Ausgabe des Inhalt des neuen Vektors auf der Konsole. Sie lautet wie erwartet 0 1 2.

Etwas Handarbeit muss auch sein

Das nächste Beispiel (Listing 3) erstellt eine Klasse secretNumber. Sie enthält als private Komponente eine Double-Zahl, die über einen Konstruktor gesetzt und über die Funktion value() abgefragt werden kann.

class secretNumber
{
  ///////////////////////////////////////////////////////////////
  friend class boost::serialization::access;

  template<class Archive>
  void serialize(Archive & ar, const unsigned int version){
    using boost::serialization::make_nvp;
    ar & make_nvp("secret_", secret_);
  }
  ///////////////////////////////////////////////////////////////

public:
  secretNumber(void) : secret_(0) { /* nothing */ }
  secretNumber(double secret) : secret_(secret) { /* nothing */ }
  double value(void) const { return secret_; }
private:
  double secret_;
};

Auffällig sind die Zeilen im Kopfteil. Boost.Serialization erhält hier Zugang zu den geschützten Teilen der Klasse. Ferner wird hier eine Memberfunktion serialize erstellt. Dieser Funktion muss das Programm bekannt geben, dass es ein Datenelement secret_ in der Klasse gibt. Generell beschreibt man hier alle Daten und Objekte, die zur Erstellung eines Objekts des serialisierten Typs notwendig sind. Der Entwickler kann aber auch eigene Aktionen anstoßen. So wäre es möglich, einen Eintrag in ein Logfile zu schreiben, sobald ein Objekt dieses Typs (de-)serialisiert wird.

Als nächste Komponente wird die Klasse base erzeugt (Listing 4). Sie leitet sich von einem vector ab, dessen Serialisierung bereits erläutert wurde. Die Ableitung von einem Container der STL würde keinen Preis für gute Programmierung einbringen, ermöglicht aber eine vereinfachte Darstellung für diesen Artikel.

template <class T>
class base
  :public vector<T>
{
  ///////////////////////////////////////////////////////////////
  friend class boost::serialization::access;
  template<class Archive>
  void serialize(Archive & ar, const unsigned int version){
    using boost::serialization::make_nvp;
    ar & make_nvp("stdvector", boost::serialization::base_object<vector<T> >(*this));
  }
  ///////////////////////////////////////////////////////////////
public:
  virtual void sort(void) = 0;
};
namespace boost {
  namespace serialization {
    template<class T>
    struct is_abstract<base<T> > {
      typedef mpl::bool_<true> type;
      BOOST_STATIC_CONSTANT(bool, value = true);
    };
}}

Boost.Serialization muss darüber Kenntnis erhalten, dass es eine Elternklasse gibt, die es zu serialisieren gilt. Bei deren Umwandlung greift die Bibliothek entweder auf die Serialisierungsspezifikationen der Elternklasse oder auf externe Spezifikationen zurück. Im Falle von vector sind diese in der einzubindenden Header-Datei <boost/serialization/vector.hpp> enthalten. Die Serialisierung erfordert die Einbindung ungewöhnlich vieler Header-Dateien. Informationen über die benötigten Dateien sind dem Gesamtlisting dieses Beispiels zu entnehmen, das über den iX-Listingservice erhältlich ist.

Der Ausdruck base_object<vector<T> >(*this) macht in base<T>::serialize() die Elternklasse bekannt. Gäbe es mehrere Elternklassen, könnte man auch mehrere solcher Ausdrücke angeben.

Da base als Template implementiert ist, weiß man nicht, welche Daten die Klasse speichert. Die Memberfunktion sort() muss damit rein virtuell sein, und Objekte dieser Klasse sind nicht instantiierbar. Bezüglich der Serialisierung ist das Beispiel deshalb alles andere als trivial. Es wäre übrigens durchaus möglich, shared_ptr auf Objekte dieses Typs in einem Vektor zu speichern und diese (Smart-)Pointer auf abgeleitete Klassen zeigen zu lassen. Boost.Serialization ist in der Lage, mit dieser Situation umzugehen und solche Vektoren zu (de-)serialisieren.

Templates und abstrakte Klassen

Den Bibliotheksfunktionen muss mitgeteilt werden, dass es sich bei base um eine abstrakte Klasse handelt. Für „normale“ abstrakte Klassen würde man diesen Hinweis über das Makro BOOST_IS_ABSTRACT(myClass) geben. Leider funktioniert das nicht für Templates. Der Programmierer muss den im Makro enthaltenen Code deshalb gewissermassen von Hand angeben. Dies geschieht am Ende von Listing 4.

Details zu BOOST_IS_ABSTRACT und zu anderen „Type Traits“ sind auf der Boost-Website zu finden [b]. Ein weiteres wichtiges Makro ist BOOST_CLASS_EXPORT_GUID(my_class, "my_class_external_identifier"). Es wird benötigt, wenn man abgeleitete Klassen über einen Zeiger auf ihre Basisklassen (de-)serialisieren will. Mehr zu diesem Thema findet sich ebenfalls im Web [c].

Um in diesem Tutorial möglichst viele Fälle abzudecken, ist die base-Klasse etwas konstruiert. Man hätte beispielsweise auch auf die sort() Funktion der STL zurückgreifen können (eine entsprechende Metrik vorausgesetzt).

Serialisieren hilft beim Kommunizieren

Im nächsten Schritt geht es darum, eine Klasse von base<T> abzuleiten (Listing 5). Sie ist das eigentliche Serialisierungsziel und leistet auch in der Netzkommunikation gute Dienste. Ziel ist es, in der Subklasse Objekte des Typs shared_ptr<secretNumber> zu speichern. Entsprechend wird secretContainer von base<shared_ptr<secretNumber> > abgeleitet.

class secretContainer
  :public base<shared_ptr<secretNumber> >
{
  ///////////////////////////////////////////////////////////////
  friend class boost::serialization::access;

  template<class Archive>
  void serialize(Archive & ar, const unsigned int version){
    using boost::serialization::make_nvp;
    ar & make_nvp("base", boost::serialization::base_object<
                          base<shared_ptr<secretNumber> > >(*this));
  }
  ///////////////////////////////////////////////////////////////
public:
  // Default constructor not shown. Initializes 
  // random number generator
  // Initialize container with random numbers
  secretContainer(unsigned int size)
    :mersenne(static_cast<unsigned int> (time(0))),
     normalDist(0.,1.), gauss(mersenne,normalDist)
  {
    for(unsigned int i=0; i<size; i++){ 
      shared_ptr<secretNumber> p(new secretNumber(gauss()));
      this->push_back(p);
    }
  }
  string toString(void){
    ostringstream oss;  // serialize
    {
#ifdef XMLARCHIVE
      boost::archive::xml_oarchive oa(oss);
#else
      boost::archive::text_oarchive oa(oss);
#endif
      oa << boost::serialization::make_nvp("secretContainer",*this);
    } // archive and stream closed at end of scope    
    return oss.str();
  }

  void fromString(const string& descr){
  { /* code analog toString() */ }
    
  virtual void sort(void){
    std::sort(this->begin(), this->end(), 
      bind(&secretNumber::value,_1) < bind(&secretNumber::value,_2));
  }
private:
  mt19937 mersenne;
  normal_distribution<double> normalDist;
  variate_generator<mt19937&, normal_distribution<double> > gauss;
};

Die Klasse besitzt durchaus lokale Objekte - einen Zufallszahlengenerator und eine Verteilung - jedoch sollen diese nicht serialisiert werden. Stattdessen initialisiert das Beispiel den Generator für jedes neue Objekt des Typs secretContainer im Konstruktor mit der aktuellen Zeit. In secretContainer::serialize() muss deshalb nur die Elternklasse angegeben werden.

Der Konstruktor füllt das Objekt mit einer Anzahl size von Zufallszahlen. Hier kommt der Mersenne-Twister-Generator zusammen mit einer Normalverteilung zum Einsatz.

Es folgen zwei Funktionen toString() und fromString, die lediglich den Code zur (De-)Serialisierung kapseln. Sie dienen eher der Lesbarkeit der main()-Funktion. Über ein #define lässt sich steuern, ob bei der Serialisierung XML-Code oder das bibliothekseigene Ausgabeformat genutzt werden soll. Im Fall von XML benötigt der Compiler nur den Schalter -DXMLARCHIVE.

Die Funktion secretContainer::sort() spezifiziert nun den Code, der in base<T> fehlt, indem sie einfach auf die STL-eigene sort()-Funktion zurückgreift. Wie schon in den vergangen Artikeln wird hier auf boost::bind zurückgegriffen, um eine Metrik für die Sortierung anzugeben.

Listing 6 zeigt schließlich die main()-Funktion. Für mehr Komfort definiert die Anwendung noch einen eigenen Stream-Operator für secretContainer, der hier nicht gezeigt wird. Die main()-Funktion ist einfach. Das Listing erzeugt ein secretContainer-Objekt, das es mit zehn shared_ptr<secretNumber> füllt und sortiert.

int main(int argc, char **argv){
  secretContainer sc(10);
  sc.sort();
  string serialization = sc.toString();
  // Let the user know
  cout << "Printing serialized secretContainer:" << endl;
  cout << serialization << endl;
  // Load serialized object into new secretContainer
  secretContainer sc2;
  sc2.fromString(serialization);
  cout << "Printing content of de-serialized secretContainer:" << endl
       << sc2 << endl;
  return 0;
}

22 serialization::archive 4 0 0 0 0 0 0 10 1 0 1 4 1 0
0 -1.1783758763357466 4
/* 1-8 gelöscht */
9 1.1338391375353956
Printing content of de-serialized secretContainer:
-1.18 -0.65 -0.38 -0.20 0.32 0.45 0.55 0.63 1.00 1.13

Die Hilfsfunktion toString() erzeugt eine Textrepräsentation des Objekts und gibt sie auf der Konsole aus. Wie das einfachere Vektorbeispiel lädt dieses Programm den String in ein anderes secretContainer-Objekt, dessen Ausgabe erneut auf der Konsole erfolgt. Verwendet man das text_oarchive von Boost.Serialization, sieht die Ausgabe aus wie in Abbildung 3.

Wegen der Rundung des operator<< unterscheiden sich die Ausgaben des zweiten Objekts sc2 leicht von der serialisierten Version von sc. Intern sind sie jedoch identisch.

Spektakulär ist insbesondere, wie unspektakulär die Serialisierung abläuft - das Beispiel ist ja durchaus auf reale Fälle übertragbar. Im Vergleich zur Ausgabe von Listing 2 sollte aber auch ersichtlich sein, dass das gewählte Textformat deutlich weniger Freiraum für manuelle Änderungen lässt.

Noch ein Wort zur Effizienz der Archiverzeugung und zur benötigten Rechenzeit. Ein secretContainer mit einer Million Einträgen erzeugt ein Textarchiv von 28 MByte. Im XML-Modus sind es immerhin schon 122 MByte. Verzichtet man auf das Sortieren des Vektors sowie auf die Ausgabe, dauert der gesamte Programmlauf, einschließlich (De-)Serialisierung auf einem Athlon64 6000+ für das reine Textarchiv sieben Sekunden. Im Falle des XML-Archivs sind es über 16 Sekunden. Das Programm wurde optimiert.

Es ist damit klar, dass sich text_oarchive deutlich besser zum Datenaustausch eignet als xml_oarchive. Es gibt Binärarchive, die noch effizienter, jedoch nicht unbedingt portabel sind, insbesondere, wenn man einen Datenaustausch über Architekturgrenzen hinweg plant.

Einige der Boost-Datenstrukturen sind nicht ohne Weiteres serialisierbar. Konzeptbedingt ist dies etwa bei boost::function<> schwierig.

Das Sortieren netzweit delegieren

Die Erstellung einer Client-Server-Applikation, in der der Server auf Anfrage eines Clients ein secretContainer-Objekt erzeugt, serialisiert und dem Client übermittelt, soll den Abschluss dieses Tutorials bilden. Der Client sortiert die Daten und schickt das Objekt an den Server zurück. Wichtig ist, dass die Kommunikation von den Clients ausgeht. Daher muss lediglich der Server eine öffentliche IP erhalten.

Als Basis dient die Boost.Asio-Bibliothek. Diese ist noch nicht Teil der aktuellen Release 1.34.1, jedoch schon offiziell in die Boost-Familie aufgenommen worden. Ab Boost 1.35.x dürfte sie in der Kernbibliothek zu finden sein. Teil 2 des Tutorials ist bereits darauf eingegangen, wie man diese Bibliothek in Boost 1.34.1 einbaut.

Es geht los mit dem stark vereinfachten Beispiel einer main()-Funktion, die sowohl für Clients als auch den Server verwendet wird (Listing 7). Zahl und Art der Argumente entscheiden über die Verwendung.

int main(int argc, char* argv[]){
  try {
    // Are we a client or a server ?
    if(argv[1] == string("client")){      
      if (argc != 4) {
     std::cerr << "Usage: vectorExample client <host> <port>" <<                                                         endl;
     return 1;
      }

      client Client(argv[2], argv[3]);
      Client.run();
    }
    else{
      if (argc != 3) {
     std::cerr << "Usage: vectorExample server <port>" << endl;
     return 1;
      }

      server Server(argv[2]);
      Server.run();
    }
  }
  catch (std::exception& e) {
    std::cerr << "Exception: " << e.what() << endl;
    return 1;
  }
  return 0;
}

Sowohl die server- als auch die client-Klasse besitzen eine run()-Funktion, deren Aufruf nach der Initialisierung erfolgt. In puncto Fehlerbehandlung beschränkt sich das Programm darauf, pauschal alle von std::exception abgeleiteten Ausnahmen in main() abzufangen und sich gegebenenfalls zu beenden.

Zunächst ein Blick auf die client-Klasse (Listing 8). Als private Datenelemente existieren vier Objekte. io_service ist für das gesamte Low-Level-I/O-Management zuständig. Über das socket-Objekt erfolgt die Kommunikation, und ein resolver- sowie ein query-Objekt dienen der Namensauflösung.

class client
{
public:
  client(string server, string port)
    :socket_(io_service_),
     resolver_(io_service_),
     query_(server, port)     
  { endpoint_iterator0=resolver_.resolve(query_); }
  void run(void) {
    for(unsigned int i=0; i<NPROCESS; i++){
      if(!tryConnect()) {
        cout << "Error: Could not connect to server" << endl;
        break;
      }
      string data=retrieve();

      if(data == "error"){
        cout << "Error: Invalid response from server" << endl;
        break;
      }
      // De-serialize the object
      secretContainer sc;
      sc.fromString(data);
      // Sort it. This is where the actual work is done
      sc.sort();
      if(!tryConnect()) {
        cout << "Error: Could not connect to server" << endl;
        break;
      }
      // And send it off again
      submit(sc.toString());
      cout << "Processed one data set" << endl;
    }
  }
private:
  bool tryConnect(void){ /* See listing 9 for the code */ }
  string retrieve(void){ /* See listing 10 for the code */ }
  void submit(string data){ /* See listing 11 for the code */ }
  asio::io_service io_service_;
  socket socket_;
  resolver resolver_;
  resolver::query query_;
  resolver::iterator endpoint_iterator0;
  resolver::iterator end;
};

Der Konstruktor initialisiert zunächst das socket- und anschließend das resolver- sowie das query-Objekt. Da es zu einer Kombination aus Hostname und Port verschiedene Auflösungen geben kann (man spricht vom „endpoint“), liefert Boost.Asio als Ergebnis eines resolver.resolve()-Aufrufs einen Iterator zurück, der alle möglichen Auflösungen durchlaufen kann. Den Endpunkt dieses Iterators, der in der end-Variable als privates Datenmember gespeichert wird, beschreibt der Default-Konstruktor. Da die Applikation mehrfach Verbindungen auf- und abbauen soll, speichert sie zudem das Ergebnis des resolve()-Aufrufs in endpoint_iterator0.

Statischer Setup vorausgesetzt

Es kann passieren, dass zwei resolve()-Aufrufe unterschiedliche Informationen zurückliefern - in diesem Fall wäre das Zwischenspeichern nicht sinnvoll. Für die vorliegende Spielzeug-Applikation soll jedoch von einem statischen Setup ausgegangen werden.

Als Nächstes ein Blick auf die run()-Funktion. Der Client soll sich NPROCESS mal mit dem Server verbinden, sich einen Vektor mit Zufallszahlen holen und diesen sortiert zurückschicken. Die Funktion tryConnect() baut zunächst eine Verbindung zum Server auf. Schlägt diese fehl, erfolgt eine Fehlermeldung und das Programm verlässt die run()-Schleife. Eine weitere Wrapper-Funktion - retrieve() - holt nun die Beschreibung eines secretContainer-Objekts vom Server. Hat der Server keine gültige Antwort geschickt, liefert retrieve() den String "error" zurück, was zur Beendigung der Schleife führt. War der Transfer erfolgreich, wird ein secretContainer-Objekt erzeugt. Der Aufruf von secretContainer::fromString() lädt die Objektbeschreibung in das secretContainer-Objekt. Am Ende der retrieve()-Funktion wird die Verbindung zum Server unterbrochen - hierzu unten mehr.

secretContainer::sort() erledigt die eigentliche Arbeit - das Sortieren des Vektors. Nun bleibt nur, das sortierte Objekt zum Server zurückzuschicken. Hierzu erfolgt wieder ein Verbindungsaufbau, toString() serialisiert den secretContainer und die Wrapper-Funktion client::submit() schickt die Daten zum Server zurück. Auch hier wird am Ende von submit() die Verbindung unterbrochen. Diese Unterbrechung für die Dauer der Berechnung ist sinnvoll, denn es ist für den Client nicht ersichtlich, wie lange dieser Vorgang dauert, da das nur dem secretContainer-Objekt bekannt ist.

Verbindung wieder aufnehmen

Die Funktion client::tryConnect (Listing 9) iteriert über alle Endpunkte. Das Ende der Iteration zeigt die bereits besprochene end-Variable an. Zunächst wird sicherheitshalber der socket geschlossen, danach erfolgt ein Verbindungsversuch zum Server über einen neuen Endpunkt. Ist dieser erfolgreich, setzt das Programm die Fehlervariable error auf 0, und die while()-Schleife bricht ab. Nach dem Durchlaufen aller Endpunkte des resolve()-Aufrufs, ohne dass eine Verbindung zustande gekommen ist, erfolgt ebenfalls ein Programmabbruch. In diesem Fall ist error noch auf den Fehlercode host_not_found gesetzt. In diesem Fall liefert das Beispiel false als Ergebnis von tryConnect() zurück. Ist diese Hürde überwunden, war der Verbindungsversuch erfolgreich und der Rückgabewert ist true.

bool client::tryConnect(void){
      resolver::iterator endpoint_iterator=endpoint_iterator0;
      boost::system::error_code error = asio::error::host_not_found;
      while (error && endpoint_iterator != end){
           socket_.close();
           socket_.connect(*endpoint_iterator++, error);
      }
      if (error) return false;
      return true;
  }

Nun kann es an den Empfang der Daten gehen (Listing 10). Die Funktion client::retrieve() schickt dem Server mithilfe von asio::write() über den in tryConnect() geöffneten Socket zunächst die Mitteilung getData. Diese teilt dem Server mit, dass sie ein neues Datenpaket benötigt. Die Hilfsfunktion commandString erzeugt dabei einen String der Länge COMMANDLENGTH. Diese Länge ist mit dem Server fest vereinbart, da er keine Möglichkeit hat, diese Größe zur Laufzeit festzustellen.

string client::retrieve(void){
    asio::write(socket_, asio::buffer(commandString("getData", COMMANDLENGTH)));
    char  inboundCommand_[COMMANDLENGTH];
    asio::read(socket_, asio::buffer(inboundCommand_));
    string inboundCommand=boost::algorithm::trim_copy(
          std::string(inboundCommand_, COMMANDLENGTH));
    if(inboundCommand != "compute") {
      socket_.close();
      return string("error"); 
    }
    char inboundHeader_[COMMANDLENGTH];
    asio::read(socket_, asio::buffer(inboundHeader_));
    string inboundHeader=boost::algorithm::trim_copy(
          std::string(inboundHeader_, COMMANDLENGTH));
    unsigned int dataSize = lexical_cast<unsigned int> (inboundHeader);
    vector<char> inboundData(dataSize);
    asio::read(socket_, asio::buffer(inboundData));
    ostringstream oss;
    vector<char>::iterator it;
    for(it=inboundData.begin(); it!=inboundData.end(); ++it) oss << *it;
    socket_.close();
    return oss.str();
  }

Warten auf die richtige Antwort

Das Programm wartet nun auf eine Antwort des Servers. Die einzige zulässige Antwort auf getData ist das Kommando compute. Es liest die Antwort mithilfe von asio::read() über den Socket in ein Char-Array, das ebenfalls die Länge COMMANDLENGTH besitzt. Da die Antwort wahrscheinlich eine Reihe von Leerzeichen beinhaltet, entfernt die Boost-Funktion trim_copy() sie. Ist die erhaltene Antwort ungültig, schließt die Applikation den Socket und liefert den String "error" zurück.

Da bei der Programmierung des Beispiels ein Protokoll für den Datenaustausch zugrunde gelegt wurde, ist klar, dass nach dem compute-Kommando als nächstes Element die Größe der Datensektion - das heißt des serialisierten secretContainer - zu erwarten ist. Diese liest der Client ebenfalls in ein Char-Array passender Größe, entfernt die Leerzeichen und wandelt es mit boost::lexical_cast<unsigned int> in eine Variable um.

Nun wird ein vector<char> passender Größe für das serialisierte Objekt erzeugt, in den asio::read() die Daten des secretContainer einliest. Den in einen String gewandelten Vektor liefert der Client nach dem Schließen der Verbindung als Ergebnis der Operation zur weiteren Verarbeitung zurück.

Nach dieser Vorarbeit sollte der Inhalt der submit()-Funktion (Listing 11) sofort verständlich sein. Ihr Aufruf erfolgt nach getaner Arbeit und dient dazu, das Ergebnis der Berechnung an den Server zurückzuschicken. Hierzu schickt sie dem Server die Mitteilung result, gefolgt von der Größe des Datensegments und den Daten. Letztendlich bricht das Schließen des Socket die Verbindung ab.

void client::submit(string data){
    std::vector<asio::const_buffer> buffers; 
    string result = commandString("result", COMMANDLENGTH);
    buffers.push_back(asio::buffer(result));
    string dataSize = commandString(lexical_cast<string>(data.size()), COMMANDLENGTH);
    buffers.push_back(asio::buffer(dataSize));
    buffers.push_back(asio::buffer(data));
    asio::write(socket_, buffers);
    socket_.close();
  }

Es bleibt noch die Erläuterung des Server-Codes: server ist die Management-Instanz, und session wird für jede neue Verbindung instantiiert. Zunächst zur server-Klasse (Listing 12). Als private Datenmember besitzt die server-Klasse ein io_service- sowie ein acceptor-Objekt. Letzteres ist für das Handling neuer Verbindungen zuständig und wird im Konstruktor mit Port und io_service-Objekt initialisiert. Ferner wird ein threadpool-Objekt mit vier Threads erzeugt. Das ist für das Funktionieren des Beispiels zwar nicht notwendig, kann die Klasse in ihrer Funktionsweise aber deutlich verbessern.

class server{
public:
  server(string port)
    : acceptor_(io_service_, tcp::endpoint(tcp::v4(), lexical_cast<short>(port))),
      tp(4)
  {
    shared_ptr<session> new_session(new session(io_service_));
    acceptor_.async_accept(new_session->socket(),
          boost::bind(&server::handle_accept, this, new_session,
                           asio::placeholders::error));
  }
  void handle_accept(shared_ptr<session> current_session, 
                           const boost::system::error_code& error) {
    if (error) return;
    // You need to comment-out "current_session->processRequest();" below,
    // if you want to use this.
    // tp.schedule(boost::bind(&session::processRequest,current_session));
    // Make sure a new session is started before we handle the current request
    shared_ptr<session> new_session(new session(io_service_));
    acceptor_.async_accept(new_session->socket(),
                           boost::bind(&server::handle_accept, this, new_session,
                           asio::placeholders::error));
    current_session->processRequest();      
  }
  void run(void){
    io_service_.run();
  }
private:
  asio::io_service io_service_;
  tcp::acceptor acceptor_;
  boost::threadpool::pool tp;
};
class server{
public:
  server(string port)
    : acceptor_(io_service_, tcp::endpoint(tcp::v4(), lexical_cast<short>(port))),
      tp(4)
  {
    shared_ptr<session> new_session(new session(io_service_));
    acceptor_.async_accept(new_session->socket(),
          boost::bind(&server::handle_accept, this, new_session,
                           asio::placeholders::error));
  }
  void handle_accept(shared_ptr<session> current_session, 
                           const boost::system::error_code& error) {
    if (error) return;
    // You need to comment-out "current_session->processRequest();" below,
    // if you want to use this.
    // tp.schedule(boost::bind(&session::processRequest,current_session));
    // Make sure a new session is started before we handle the current request
    shared_ptr<session> new_session(new session(io_service_));
    acceptor_.async_accept(new_session->socket(),
                           boost::bind(&server::handle_accept, this, new_session,
                           asio::placeholders::error));
    current_session->processRequest();      
  }
  void run(void){
    io_service_.run();
  }

private:
  asio::io_service io_service_;
  tcp::acceptor acceptor_;
  boost::threadpool::pool tp;
};

Der Konstruktor erzeugt ein neues session-Objekt in einem shared_ptr. Seine Verwendung erspart ein hässliches delete this im session-Objekt.

Das acceptor-Objekt wird angewiesen, eine neue Verbindung auf dem Socket des session-Objekts anzunehmen. async_accept erwartet neben dem Socket einen Handler für eine Funktion, deren Aufruf nach Beendigung der accept-Operation erfolgt. Sie erhält als einzigen Parameter einen Fehlercode. Nun sollen aber genau aus diesem Handler heraus neue accept()-Operationen und die dazugehörige Session starten, was die Übergabe weiterer Parameter erfordert. Hier hilft wieder boost::bind(), das eine shared_ptr<session> an das erste Argument von server::handle_accept() bindet. Der einzige freie Parameter dieser Funktion ist danach der Fehlercode. asio::placeholders::error fungiert einfach als Platzhalter für das boost::bind-typische _1.

Verbindung erfolgt im Hintergrund

Als asynchrone Operation beendet sich async_accept sofort nach dem Aufruf. Das Handling der neuen Verbindung findet im Hintergrund statt. Somit ist auch der Konstruktor beendet. In der main()-Funktion erfolgt nun der Aufruf von server::run(), was das io_service-Objekt veranlasst, die event-loop zu starten. Erst ab diesem Moment ist die Applikation wirklich einsatzfähig und in der Lage, neue Verbindungen anzunehmen.

Diese initiiert jeweils die server::handle_accept()-Funktion, die praktisch als Kopie des Konstruktors fungiert. Der einzige Unterschied zum Konstruktor ist, dass sie die aktuelle Session current_session dazu bewegt, ihre Arbeit zu tun. Im vorliegenden Fall bedeutet das, einem Client einen unsortierten Vektor zu liefern oder sortierte Vektoren anzunehmen.

Die Verlagerung dieser Arbeit in einen Thread würde das Beispiel noch interessanter machen. Die hierfür notwendigen Änderungen sind überraschend klein. Verwendet man die in Teil 2 des Tutorials angesprochene threadpool-Bibliothek (deren Initialisierung bereits im Konstruktor erfolgt ist), muss man die session::processRequest()-Funktion lediglich als neue Task übergeben. Wird einer der vier konfigurierten Threads frei, beginnt er mit der Ausführung dieser Funktion, die bis zu diesem Zeitpunkt in einer Taskqueue gespeichert ist.

Allerdings muss man sich sicher sein, dass processRequest() auch für die Ausführung in einer Multithreaded-Umgebung geeignet ist. Und genau hier gibt es bei Boost.Serialization im Moment Fragezeichen. Eine neue, sicherere Version dieser Bibliothek ist aber unterwegs. Gleichwohl ist die betreffende Zeile der handle_accept()-Funktion deshalb auskommentiert. Die Verwendung von Boost.Serialization mit Boost.Asio allein scheint trotz dessen asynchronen Designs aber sicher zu sein.

Wichtig sind noch zwei Bemerkungen: Beim momentanen Design startet der Server auf dem Umweg über handle_accept in jeder async_accept()-Operation eine neue Aktion. Dies führt zwar zu einer Endlosschleife, überschwemmt den Rechner aber nicht mit async_accept-Operationen. Denn ein weiterer async_accept-Aufruf erfolgt erst, wenn ein Client eine neue Verbindung akzeptiert. Die Zahl der accept-Aufrufe entspricht also immer der benötigten Zahl.

Den Einbau eines Mechanismus zur Beendigung des Servers berücksichtigt das Beispiel nicht. Hierfür müsste man einfach auf die Annahme weiterer Verbindungen verzichten, indem kein weiterer async_accept-Aufruf erfolgt. In dem Fall würde sich io_service.run() beenden.

Ein Blick auf die session-Klasse (Listing 13) soll das Tutorial abschließen. Jedes session-Objekt erhält seinen eigenen Socket - er wird im Konstruktor mithilfe einer Referenz auf den io_service initialisiert. Die eigentliche Arbeit übernimmt session::processRequest().

class session
{
public:
  session(asio::io_service& io_service)
    : socket_(io_service)
  { /* nothing */ }
  tcp::socket& socket() { return socket_; }

  void processRequest() {
    // First check what the client wants from us. 
   // This will remove the command from the stream.
    char  inboundCommand_[COMMANDLENGTH];
    asio::read(socket_, asio::buffer(inboundCommand_));
    string inboundCommand=boost::algorithm::trim_copy(
                             std::string(inboundCommand_, COMMANDLENGTH));
    // We accept only two commands: "getData" and "result"
    if(inboundCommand == "getData"){
      // Create a new secretContainer object, serialize it and send it off
      cout << "Received getData command" << endl;
      secretContainer sc(VECTORSIZE);
      submit(sc.toString());
      cout << "Sent off data" << endl;
    }
    else if(inboundCommand == "result"){
      // De-serialize the data and print the resulting object
      cout << "De-serializing data" << endl;
      string data=retrieve();
      secretContainer sc;
      sc.fromString(data);
      cout << "Received a response" << endl;
    }
    else{
      cout << "Error: unknown command: " << inboundCommand << endl;
      return;
    }
  }

private:
  void submit(string data){ /* see listing service */ }
  string retrieve(void){ /* see listing service */ }

  tcp::socket socket_;
};

Wie beim Client liest die Funktion zunächst die Mitteilung der Gegenseite. Zwei Kommandos sind erlaubt: getData und result. Der erste Fall erzeugt einen unsortierten secretContainer der Größe VECTORSIZE, serialisiert ihn und schickt ihn über die Hilfsfunktion session::submit() zum Client.

Im zweiten Fall empfängt session::retrieve() die Antwort (insbesondere den sortierten Vektor) und lädt die Daten zurück in einen secretContainer.

Die Zerstörung des session-Objekts impliziert das Schließen des Socket. Dies geschieht, wenn die letzte Kopie des shared_ptr<session> für das aktuelle session-Objekt in der server-Klasse seinen Geltungsbereich verlässt. session::submit() und session::retrieve() folgen dem aus der client-Klasse bekannten Schema.

Dr. Rüdiger Berlich
führt am Institut für Wissenschaftliches Rechnen des Karlsruhe Institute of Technology (KIT) eine Ausgründung aus dem Bereich der Parameteroptimierung in verteilten Umgebungen durch.

(ka)