Asynchron aufrufbare Wrapper

22 November 2015

Inhaltsverzeichnis[Anzeigen]

std::packaged_task erlaubt es, einen einfachen Wrapper um eine aufrufbare Einheit zu erzeugen, so dass diese später ausgeführt werden kann.

std::packaged_task

Die Arbeit mit std::packaged_task verläuft typischerweise in vier Schritten:

Verpacke die Aufgabe
Erzeuge den Future
Führe die Berechnung aus
Hole das Ergebnis ab

Am einfachsten lassen sich diese Schritte am Beispiel nachvollziehen.

#include <utility>
#include <future>
#include <iostream>
#include <thread>
#include <deque>

class SumUp{
  public:
    int operator()(int beg, int end){
      long long int sum{0};
      for (int i= beg; i < end; ++i ) sum += i;
      return sum;
    }
};

int main(){

  std::cout << std::endl;

  SumUp sumUp1;
  SumUp sumUp2;
  SumUp sumUp3;
  SumUp sumUp4;

  // define the tasks
  std::packaged_task<int(int,int)> sumTask1(sumUp1);
  std::packaged_task<int(int,int)> sumTask2(sumUp2);
  std::packaged_task<int(int,int)> sumTask3(sumUp3);
  std::packaged_task<int(int,int)> sumTask4(sumUp4);

  // get the futures
  std::future<int> sumResult1= sumTask1.get_future();
  std::future<int> sumResult2= sumTask2.get_future();
  std::future<int> sumResult3= sumTask3.get_future();
  std::future<int> sumResult4= sumTask4.get_future();

  // push the tasks on the container
  std::deque< std::packaged_task<int(int,int)> > allTasks;
  allTasks.push_back(std::move(sumTask1));
  allTasks.push_back(std::move(sumTask2));
  allTasks.push_back(std::move(sumTask3));
  allTasks.push_back(std::move(sumTask4));
  
  int begin{1};
  int increment{2500};
  int end= begin + increment;

  // execute each task in a separate thread
  while ( not allTasks.empty() ){
    std::packaged_task<int(int,int)> myTask= std::move(allTasks.front());
    allTasks.pop_front();
    std::thread sumThread(std::move(myTask),begin,end);
    begin= end;
    end += increment;
    sumThread.detach();
  }

  // get the results
  auto sum= sumResult1.get() + sumResult2.get() + sumResult3.get() + sumResult4.get();

  std::cout << "sum of 0 .. 10000 = " << sum << std::endl;

  std::cout << std::endl;

}

Die Aufgabe des Programmes ist relativ einfach. Berechne die Summe von 0 bis 10000 mit der Hilfe der vier Threads und sammle die Ergebnisse mit den assoziierten Futures ein. Das geht natürlich mit der Gaußschen Summenformel deutlich einfacher.

Im ersten Schritt verpacke ich die vier Arbeitspakete in std::packaged_task-Objekte. Die Arbeitspakete sind Instanzen der Klasse SumUp (Zeile 7 - 14). Der Klammeroperator der Klasse (Zeile 8 - 13) addiert alle Zahlen von beg bis end auf und gibt die Summe als Ergebnis zurück. Zum Abschluss werden die packaged_task-Objekte instanziiert. Die Objekte können aufrufbare Einheiten, die zwei ganze Zahlen erwarten und eine ganze Zahl zurückgeben, annehmen.

Nun gilt es im zweiten Schritt, die Future Objekte mit Hilfe der packaged_task-Objekte zu erzeugen. Genau das findet in den Zeilen 32 bis 35 statt. Die packaged_task-Objekte stellen die Promise in dem Kommunikationskanal dar. In diesem Fall definiere ich die Typen der Future explizit: std::future<int> sumResult1= sumTask1.get_future(). Automatische Typableitung mit auto ist da natürlich einfacher.: auto sumResult1= sumTask1.get_future().

Im dritten Schritt folgt die eigentliche Arbeit. Die packaged_task werden auf ein std::deque geschoben (Zeile 38 - 42), die in der while-Schleife (Zeile 49 - 56) abgearbeitet wird. Dazu verschiebe ich den Kopf der std::deque in eine std::packaged_task (Zeile 50) und verschiebe diese weiter in einen neuen Thread (Zeile 52) und lasse diesen im Hintergrund arbeiten (Zeile 55). std::packaged_task-Objekte können natürlich nicht kopiert (Copy-Semantik) werden. Dies ist der Grund für die Move-Semantik in Zeile 50 und 52. Dies gilt natürlich auch für Promise im allgemein, aber auch Futures und Threads.

Im vierten und letzten Schritt verwende ich die get-Methoden der Futures, um die Ergebnisse der vier Futures zusammen zu addieren (Zeile 61).

Zugegeben, std::packaged_task ist nicht für den einfachen Anwendungsfall wie std:.async konzipiert. Dafür ist das Ergebnis um so schlichter.

packagedTask

Optimierungspotential

C++11 besitzt die Funktion std::thread::hardware_concurrency. Diese Funktion gibt ein Hinweis auf die Anzahl der Prozessoren eines Systems zurück. Kann die C++-Laufzeit diesen Wert nicht ermitteln, ist es standardkonform, den Wert 0 zurückzugeben. Mit dem aktuellen gcc, clang oder Microsoft Compiler erhalte ich immer die richtige Antwort 4. Genau diese Information über die Anzahl meiner Prozessoren nützte ich in der optimierten Version des Programms aus, denn in ihm entspricht die Anzahl der Threads der Anzahl meiner Prozessoren. Damit ist mein System optimal ausgelastet.

#include <algorithm>
#include <future>
#include <iostream>
#include <thread>
#include <deque>
#include <vector>

class SumUp{
public:
  SumUp(int b, int e): beg(b),end(e){}
  int operator()(){
    long long int sum{0};
    for (int i= beg; i < end; ++i ) sum += i;
    return sum;
  }
private:
    int beg;
    int end;
};

static const unsigned int hwGuess= 4;
static const unsigned int numbers= 10001;

int main(){

  std::cout << std::endl;

  unsigned int hw= std::thread::hardware_concurrency();
  unsigned int hwConcurr= (hw != 0)? hw : hwGuess;

  // define the functors
  std::vector<SumUp> sumUp;
  for ( unsigned int i= 0; i < hwConcurr; ++i){
    int begin= (i*numbers)/hwConcurr;
    int end= (i+1)*numbers/hwConcurr;
    sumUp.push_back(SumUp(begin ,end));
  }

  // define the tasks
  std::deque<std::packaged_task<int()>> sumTask;
  for ( unsigned int i= 0; i < hwConcurr; ++i){
    std::packaged_task<int()> SumTask(sumUp[i]);
    sumTask.push_back(std::move(SumTask));
  }

  // get the futures
  std::vector< std::future<int>> sumResult;
  for ( unsigned int i= 0; i < hwConcurr; ++i){
    sumResult.push_back(sumTask[i].get_future());
  }

  // execute each task in a separate thread
  while ( not sumTask.empty() ){
    std::packaged_task<int()> myTask= std::move(sumTask.front());
    sumTask.pop_front();
    std::thread sumThread(std::move(myTask));
    sumThread.detach();
  }

  // get the results
  int sum= 0;
  for ( unsigned int i= 0; i < hwConcurr; ++i){
    sum += sumResult[i].get();
  }

  std::cout << "sum of 0 .. 100000 = " << sum << std::endl;

  std::cout << std::endl;

}

Hintergrundinformationen

Automatische Typableitung

Die automatische Typableitung mit auto und decltype stelle ich in dem Artikel Neue Ausdruckskraft 02/2014 vor.

Copy- versus Move-Semantik

In dem Artikel Rasch verschoben 02/2015 stelle ich Rvalue Referenzen, Copy- und Move-Semantik genauer vor.

Wie geht's weiter?

Im nächsten Artikel gehe ich tiefer auf Future und Promise ein.

title page small Go to Leanpub/cpplibrary "What every professional C++ programmer should know about the C++ standard library". Hole dir dein E-Book. Unterstütze meinen Blog.