std::future

std::future ist nicht der Standardbibliothek vorbehalten. Auch in eigenem Code kann man die Klasse nutzen, um Ergebnisse asynchroner Berechnungen zu signalisieren.

Video

Quellcode

#include <iostream>
#include <deque>
#include <memory>
#include <functional>
#include <chrono>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <future>
#include <list>

class ExecutionManager
{
  struct ExecutorBase
  {
    virtual void operator()() = 0;
  };
  
  template <typename Ret> struct Executor : public ExecutorBase
  {
    std::function<Ret ()> fn;
    std::promise<Ret> result;
    
    Executor(std::function<Ret ()> const &f)
      : fn(f)
    {
    }
    
    void operator()() override
    {
      auto res = fn();
      result.set_value(res);
    }
  };
  
  std::deque<std::shared_ptr<ExecutorBase>> mTasks;
  std::list<std::shared_ptr<std::thread>> mThreads;
  std::mutex mMutex;
  std::condition_variable mCond;
  bool mFinish;

  void executor()
  {
    std::cout << "- Thread gestartet\n";
    bool running = true;
    while (running) {
      std::unique_lock<std::mutex> l(mMutex);
      mCond.wait(l, [this]() { 
	    return mFinish || mTasks.size() != 0;
      });
      std::cerr << "Prüfe Warteschlange\n";
      if (!mTasks.empty()) {
	auto task = mTasks.front();
	mTasks.pop_front();
	l.unlock();
	(*task)();  
      }
      else {
	running = !mFinish;
      }
    }
    std::cout << "- Thread beendet\n";
  }
  
public:
  
  explicit ExecutionManager(unsigned int threads)
  {
    mFinish = false;
    for (unsigned int i = 0; i < threads; ++i)
    {
      mThreads.push_back(std::shared_ptr<std::thread>(
	new std::thread(&ExecutionManager::executor, this)));
    }
  }
  
  template <typename F> auto addTask(F task) -> std::future<decltype(task())> {
    typedef decltype(task()) Ret;
    std::unique_lock<std::mutex> l(mMutex);
    auto t = std::shared_ptr<Executor<Ret>>(new Executor<Ret>(task));
    mTasks.push_back(t);
    mCond.notify_one();
    return t->result.get_future();
  }
  
  void joinAll()
  {
    {
      std::unique_lock<std::mutex> l(mMutex);
      mFinish = true;
    }
    mCond.notify_all();
    for (auto t : mThreads) {
      t->join();
    }
    std::cout << "-- Alle Threads beendet.\n";
  }
};

int main()
{
  ExecutionManager e(2);

  std::this_thread::sleep_for(std::chrono::seconds(2));
  
  auto  t1 = e.addTask([]() {     
    std::cout << "lambda aufgerufen\n";
    std::this_thread::sleep_for(std::chrono::seconds(7)); 
    std::cout << "lambda beendet\n"; 
    return 10;
  });
  auto t2 = e.addTask([]() { 
    std::cout << "Zweiter Task!\n"; 
    return std::string("Hallo");    
  });
  
  auto r1 = t1.get();
  auto r2 = t2.get();
  
  std::cout << "r1 (" << typeid(r1).name() << ") = " << r1 << std::endl;
  std::cout << "r2 (" << typeid(r2).name() << ") = " << r2 << std::endl;
  
  e.joinAll();
}

Erklärung

Ziel des ziemlich großen Umbaus unseres ExecutionManager ist die Unterstützung der Rückgabe von Ergebnissen aus den übergebenen Tasks. Das macht einige neue Techniken notwendig.

Einstiegspunkt bleibt die Funktion addTask(). Bisher übernahm die einfach eine fest definierte Variante von std::function. Das geht nun nicht mehr so einfach, denn durch die Unterstützung von Rückgabewerten müssen unterschiedliche std::function-Instanziierungen übernommen werden. Der Rückgabetyp ist einfach Bestandteil des Funktionstyps, lambda : int ist einfach etwas anderes, als lambda : string. Daher wird aus addTask ein Template, welches als Templateparameter den Funktionstyp übernimmt. So haben wir den Vorteil, dass die Aufrufstelle gleich bleibt, da sich der Compiler den Funktionstyp automatisch aus dem übergebenen Parameter ableiten kann.

Auf der anderen Seite haben wir nun das Problem, dass wir an den Rückgabetyp der übergebenen Funktion kommen müssen. Den brauchen wir ja, damit wir das korrekte std::future zurückliefern können. Normalerweise würde die Deklaration von addTask so aussehen:

template <typename F> std::future<?> addTask(F fn)

Problem hier: was muss an Stelle des Fragezeichens stehen? Das können wir feststellen: es ist der Rückgabetyp des Ausdrucks fn(), also des Aufrufs der übergebenen Funktion. Diesen können wir problemlos ermitteln: decltype(fn()) tut genau das: der Compiler stellt fest, welcher Typ das Ergebnis des Ausdrucks ist (das ist kein echter Aufruf von fn, sondern nur der Ausdruck, wie ein Aufruf aussehen würde, damit der Compiler weiß, was genau er ermitteln soll). decltype macht darauf dann eine Typ-Deklaration, so, als hätten wir die direkt hingeschrieben. Ist fn also so deklariert:

int fn()

dann liefert decltype(fn()) das gleiche Ergebnis, als hätten wir direkt int hingeschrieben. Soweit, so gut. Wir könnten also schreiben (Achtung: falsch! Nicht so übernehmen!):

template <typename F> std::future<decltype(fn())> addTask(F fn)

Leider beißt uns hier ein Problem: das Symbol fn ist erst ab dem Moment bekannt, wo es deklariert wurde. Wir können es also in der Zeile nicht links davon schonmal verwenden, da der Compiler es schlicht nicht kennt. Für genau dieses Problem bringt C++11 einen alternativen Syntax für die Deklaration von Funktionen, der die Deklaration des Rückgabetyps nach hinten zieht (siehe Zeile 77). Statt eines Rückgabetyps vor dem Funktionsnamen steht da nur auto. Der eigentliche Typ kommt dann mit der Syntax -> typ hinter die Parameterliste. Damit stehen uns alle Parameter für die Ermittlung des Typs zur Verfügung und die automatische Ableitung mittels decltype() funktioniert.

Herzstück der Rückgabe ist das Paar std::promise und std::future. Beide Klassen sind Templates und werden mit ihrem zu übermittelnden Wert parametrisisert. Ein std::promise<int> kann also immer nur mit einem std::future<int> reden. Wir müssenen nun ein geeignetes promise erzeugen, welches wir später als Einstiegspunkt für die Rückgabe verwenden können. Das ist mit dem bekannten Rückgabetyp einfach. Dieses promise müssen wir nun noch mit dem Task passend verbinden und uns in unserer Task-Liste merken (die beiden müssen verbunden sein, damit der ExecutionManager nach dem Ausführen des Tasks auch das richtige promise mit dem Ergebnis beschreibt, siehe Zeile 32). Das wiederum geht aber nur, wenn wir das Objekt, in dem unser promise gespeichert ist, ebenfalls parametrisierbar machen, weil wir nur dann in diesem Objekt das richtige promise als Feld anlegen können. Diese Problematik lösen wir mit einem kleinen Template namens Executor, welches Task und Rückgabetyp speichert (Zeile 19-34).

Soweit, so gut. Dummerweise machen uns nun die Datenstrukturen der Standardbibliothek einen Strich durch die Rechnung: die können nämlich nur gleiche Objekte speichern. Da unsere unterschiedlichen Instanziierungen des Executor-Templates als unterschiedliche Typen gelten, können wir sie leider nicht in der gleichen std::deque speichern. An dieser Stelle muss uns die Laufzeitpolymorphie mit virtuellen Funktionen zu Hilfe eilen: wir definieren die gemeinsame Basisklasse ExecutorBase, die wiederum eine virtuelle Funktion execute definiert. Diese wird dann in jeder Instanz des Executor-Templates unterschiedlich überschrieben (auch wenn sich nur der Rückgabetyp des promise ändert). Statt die Objekte nun direkt in der std::deque abzulegen, speichern wir nun Pointer auf die einzelnen Executors. Das ist notwendig, weil wir uns sonst das Slicing der Objekte einen Strich durch die schöne polymorphe Rechnung machen würde (muss ich mal ein Video dazu machen).

Damit haben wir alle Zutaten zusammen: die Executor-Objekte übernehmen das Ausführen des Tasks (Zeile 31), holen das Ergebnis ab und schreiben es in das std::promise. Die Methode addTask() liefert dem Aufrufer das passende std::future zurück (mittels std::promise::get_future() ermittelt, siehe Zeile 83). Das kann (und sollte) der sich merken und dann ganz normal mittel get() das Ergebnis des Tasks abholen. Der eigentliche Thread im ExecutionManager ruft jetzt nicht mehr den Task direkt auf, sondern den passenden Executor, damit die ganze Problematik mit den passenden Rückgabetypen dort versteckt ist. Das Umleiten auf die richtige Variante des operator() im Executor übernimmt die Laufzeitpolymorphie, da dieser virtual ist.