Asynchrone Programmierung – Teil 4: Qt6 mit QPromise und QFuture

Das Framework Qt unterstützt mit seinem tief in der Architektur verankerten Event-System und dem Signal-Slot-Mechanismus seit jeher Entwicklerinnen und Entwickler bei der asynchronen Programmierung. Diese Möglichkeiten hat der vorangegangene Teil unserer Serie vorgestellt.

QFuture ist ein neueres Qt-Konstrukt, das das Konzept des „thennables“ – der .then()-Methoden – aus anderen Sprachen aufnimmt (z. B. Promise in JavaScript, CompletableFuture in Java oder Task in C#). Es erlaubt mit QFuture::then() das Ausführen und Hintereinanderschalten von asynchronen Funktionen, unabhängig von Threads, beispielsweise um Benutzeroberflächen asynchron zu aktualisieren.

In Qt tritt ein lesendes, konsumierendes QFuture immer als die eine Seite der gleichen Medaille auf. Auf der schreibenden, produzierenden Seite steht das QPromise. Entwicklerinnen und Entwickler können den Zustand des Promise-Future-Paares detailliert steuern: starten (QPromise::start()), aussetzen (QPromise::suspend()), beenden (QPromise::finish()) oder abbrechen (QPromise::cancel()). Schließlich können Developer nicht nur einen Rückgabewert, sondern beliebig viele setzen (QPromise::addResult()). Mit der QPromise-/QFuture-API haben sie eine praktische Schnittstelle, um Benutzer- sowie Logik-Schichten effektiv zu trennen und Fortschritte an die Bedienoberfläche zu melden.

Die Klasse QFutureWatcher liefert einen Mechanismus, um die QFuture-Funktionalität mit dem im vorangegangenen Artikel beschriebenen Signal-/Slot-Mechanismus zu verbinden. Das folgende Listing 1 zeigt ein Beispiel für die grundlegenden Funktionsweisen.

int main(int argc, char* argv[])
{
    QCoreApplication app(argc, argv);

    QFutureWatcher watcher;
    QObject::connect(&watcher, &QFutureWatcher::started, []()
    {
        qInfo() << "future started ";
    });

    QObject::connect(&watcher, &QFutureWatcher::resultReadyAt, [&watcher](int i)
    {
        qInfo() << "result ready" << i << "=" << watcher.future().resultAt(i);
    });

    QObject::connect(&watcher, &QFutureWatcher::finished, [&watcher, &app]()
    {
        qInfo() << "future finished ";
        for (int i = 0; i < watcher.future().resultCount(); ++i)
        {
            qInfo() << "Final result" << i << "=" << watcher.future().resultAt(i);
        }
        app.quit();
    });

    QPromise promise;
    QFuture future = promise.future();
    watcher.setFuture(future);

    QThreadPool::globalInstance()->start([&promise]()
    {
        promise.start();
        QThread::sleep(1);
        promise.addResult(10);
        QThread::sleep(1);
        promise.addResult(20);
        QThread::sleep(1);
        promise.finish();
    });

    return app.exec();
}

Listing 1: Einfaches Beispiel für die Verwendung von QPromise, QFuture und QFutureWatcher.

Neben QFutureWatcher bietet die Klasse QFuture eine Fluent-API an, mit der man Futures hintereinander schaltet (das angesprochene „thennable“ bzw. auch als „chaining“ bezeichnet): Entwickler legen mit then() einen Nachfolger fest, wobei eine zweite Verwendung von then() den bisherigen Nachfolger überschreibt, sodass es immer nur einen Nachfolger geben kann. Außerdem löst die erste Verwendung von addResult des zugrunde liegenden Promise then() aus, und nicht etwa promise.finish(). Ein Abbruch des Promise mit cancel() aktiviert den Callback onCancelled (siehe Listing 2 unten).

int main(int argc, char* argv[])
{
    QCoreApplication app(argc, argv);

    QPromise promise;
    QFuture future = promise.future();

    QThreadPool::globalInstance()->start([&promise]()
    {
        promise.start();
        qInfo() << QDateTime::currentDateTime().toString(Qt::ISODateWithMs) << QThread::currentThreadId() << "started";
        QThread::sleep(1);
        promise.addResult(1);  // triggert then()
        QThread::sleep(1);
        // promise.cancel(); // triggert onCanceled()
        promise.addResult(2); // kein Effekt auf then()
        promise.finish();// kein Effekt auf then()
    });

    future
    .onCanceled([]{/*...*/ return -1;})
    .then([](int result1)
    {
        qInfo() << QDateTime::currentDateTime().toString(Qt::ISODateWithMs) << QThread::currentThreadId() << "result" << result1;
    });


    return app.exec();
}

Listing 2: Beispiel für das Auslösen von then() und onCancelled().

Die Methode then() ist sowohl beim Parameter als auch dem Rückgabewert generisch. Das bedeutet konkret: Bei QFuture hat das then-Callable einen Parameter vom Typ T. Der Rückgabetyp, den der Entwickler innerhalb des then-Callable verwendet, ist der Rückgabetyp des Futures, das wiederum mit then() in die verkettete Abfolge gehängt werden kann, siehe Listing 3:

void myFunc(QString val){/*...*/ return;}


QFuture future = // siehe Listing 12
QFuture future2 = future.then([](int result1)
{
   return QString("fortytwo");
}).then(myFunc);

Listing 3: Beispiel für Future-Chaining mit then().

Falls innerhalb eines Future oder eines then()-Blocks eine Exception auftritt, wird diese an den Callback onFailed() geleitet. Dabei gilt die Regel, dass die Exception in der Kette weiter wandert, bis sie ein passender onFailed-Callback erreicht. Wenn man Programmteile in Arbeiter-Threads auslagert, muss man dort Exceptions korrekt auffangen und an promise.setException() übergeben. Nur dann funktioniert die Methode onFailed wie erwartet, siehe Listing 4:

QThreadPool::globalInstance()->start([&promise]()
    {
        promise.start();

        try
        {
            throw std::runtime_error("error from f1");
            promise.addResult(42); // das wird nicht erreicht
        }
        catch (const std::exception& ex)
        {
            promise.setException(std::current_exception());
        }
    });
    future.then([](int i)
    {
        // ...
    }).onFailed([](const std::exception& ex)
    {
        qInfo() << ex.what();
    });

    QTimer::singleShot(200, &app, &QCoreApplication::quit);

Listing 4: Korrekte Verarbeitung von Exceptions in Arbeits-Threads mit Promise.

Entwickler können auch mehrere onFailed mit einem Future verwenden. In diesem Fall wird das erste passende aufgerufen. Dabei ist die Reihenfolge relevant: Beispielsweise wird in Listing 5 das zweite onFailed nie erreicht, da im vorherigen bereits die darüber liegende Klasse abgefangen wurde.

future.then([](int res) {
    // ...
    throw std::runtime_error("Exception");
}).onFailed([](const std::exception &e) {
    // dieses Callback wird ausgeführt
}).onFailed([](const std::runtime_error &e) {
});

Listing 5: Verwendung mehrerer onFailed-Callbacks

Innerhalb der Funktion, die onFailed verarbeitet, können Entwickler auch einen Rückgabewert angeben. Die Kette wird dann mit diesem Wert fortgeführt, siehe Listing 6:

QFuture future = ...
future
.onFailed([](const QException& ex)
{
    return -1;
})
.then([](int i)
{
    // im Fehlerfall ist i=-1
})

Listing 6: Fortführung der then()-Kette im Fehlerfall.

Wenn kein passender Callback onFailed vorhanden ist, wird die Exception von demjenigen Thread, in dem sie auftrat, an den aufrufenden Thread weitergeleitet und muss dort entsprechend abgefangen werden (Error Propagation), siehe Listing 7:

auto resultFuture = future.then([](int res) {
    ...
    throw Error("message");
    ...
}).onFailed([](const std::exception &e) {
    // wird nicht aufgerufen
}).onFailed([](const QException &e) {
    // wird nicht aufgerufen
});

try {
    auto result = resultFuture.result();
} catch(Error er) {
    // dieser Teil wird aufgerufen
}

Listing 7: Propagation einer Exception.

Die bisher gezeigte Verwendung von then() führt Folgefunktionen standardmäßig in dem Thread aus, in dem das ursprüngliche QFuture lief. Entwickler können aber auch mit einem Parameter den Thread für die Folgefunktion bestimmen. Übergeben sie als Parameter ein QObject, wird dessen zugeordneter QThread verwendet. Bei einen QThreadPoolkommt dieser zum Einsatz.

Schließlich können Developer auch ein Argument vom Typ QtFuture::Launch übergeben: Die weitere Bestimmung Sync entspricht dem beschriebenen Standardfall (gleicher Thread wie Aufrufer), während Async automatisch den globalen Threadpool verwendet und Inherit die Einstellung des Vorgängers, siehe Listing 8:

// Standardfall, kein Parameter, entspricht Sync)
auto f1 = base.then([](int i) { ... });
auto f2 = f1.then(QtFuture::Launch::Sync, [](int i) { ... });

// QObject => Ausführung im Thread des QObjects
auto f3 = f1.then(&obj, [](int i) { ... });

// QThreadPool => Ausführung im angegebenen Pool
auto f4 = f2.then(&customPool, [](int i) { ... });

// Launch::Async => globaler Threadpool
auto f5 = f4.then(QtFuture::Launch::Async, [](int i) { ... });

// Launch::Inherit: erbt Thread des Vorgängers
auto f6 = f5.then(QtFuture::Launch::Inherit, [](int i) { … });

Listing 8: Bestimmung des Threads mit then(…)

Der Namespace QtFuture enthält darüber hinaus einige nützliche Hilfsfunktionen für Futures. Mit den Funktionen QtFuture::makeReady… erstellen Entwicklerinnen und Entwickler ein Future, das bereits im Status beendet ist (QFuture.isFinished() liefert true). makeReadyVoidFuture erzeugt ein beendetes Future ohne Wert (QFuture), makeReadyValueFuture(T) entsprechend ein QFuture und makeReadyRangeFuture ein beendetes QFuture, das mehrere Ergebnisse besitzt. makeExceptionalFuture schließlich erstellt ein beendetes QFuture, das eine Exception beinhaltet.

Diese Funktionen sind nützlich, wenn man bei einer Funktion anhand von Parametern überprüfen möchte, ob eine Bearbeitung als QFuture überhaupt möglich oder sinnvoll ist. Sofern dies nicht der Fall ist, gibt die Funktion direkt ein beendetes QFuture zurück, siehe Listing 9:

QFuture createIntFuture(int a)
{
    if (a <= 0)
    {
        return QtFuture::makeReadyValueFuture(0);
    }

    QPromise promise;
    QFuture result = promise.future();
    QThreadPool::globalInstance()->start([promise = std::move(promise), a]() mutable
    {
        promise.start();
        QThread::sleep(500);
        promise.addResult(a + 1);
        promise.finish();
    });
}

Listing 9: Verwenden eines bei der Erstellung beendeten Futures mit makeReadyValueFuture.

Eine weitere nützliche Funktion ist QtFuture::whenAll(), die ein QFuture erzeugt, das beendet wird, wenn alle als Parameter übergebenen QFuture beendet sind. Die QFuture können verschiedene Template-Typen haben, dementsprechend ist der Rückgabetyp von whenAll ein QFuture mit einer Liste vom Typ std::variant.

Developer müssen hier noch auf die Besonderheit achten, dass beim Chaining mit then() die Nachfolgefunktion auf dem Thread läuft, dessen Future als Letztes beendet wurde. Dies kann, wie bereits beschrieben, ein Parameter für then() explizit festlegen, siehe Listing 10:

QFuture f_int = createIntFuture();
QFuture f_qstring = createStringFuture();
QFuture f_void = createVoidFuture();

using MyFuturesVariant = std::variant, QFuture, QFuture>;

QFuture> f_whenAll = QtFuture::whenAll(f_int, f_qstring, f_void);
f_whenAll.then(/*QtFuture::Launch::Async*/, [](const
    QList& results)
{
    ...
});

Listing 10: Beispiel für QtFuture::whenAll(…)

QtFuture::whenAny() erzeugt ein QFuture, das beendet wird, sobald eines der übergebenen Futures fertig ist. Beim Verketten gilt, dass die Nachfolgefunktion im Thread des zuerst beendeten Futures ausgeführt wird, sofern kein expliziter Launch-Parameter ein anderes bestimmt, siehe Listing 11:

QFuture f_whenAny = QtFuture::whenAny(f_int, f_qstring, f_void);
f_whenAny.then([](const MyFuturesVariant& f)
{

});

Listing 11: Beispiel für QtFuture::whenAny(…)

Schließlich bietet die Funktion QtFuture::connect() die Möglichkeit, aus einem beliebigen Signal ein QFuture zu erzeugen, siehe Listing 12:

QTimer timer(&app);
timer.setInterval(1000);
timer.setSingleShot(true);
timer.start();

QFuture timerTimeoutFuture = QtFuture::connect(&timer, &QTimer::timeout);
timerTimeoutFuture.then([]
{
    qDebug() << "QTimer timeout captured via QtFuture::connect";
});

Listing 12: Beispielhafte Verwendung von QtFuture::connect()