Ein unendlicher Datenstrom dank Coroutinen in C++20

Inhaltsverzeichnis

In diesem Artikel wird das Schlüsselwort co_yield genauer unter die Lupe genommen. Dank ihm ist es möglich, einen unendlichen Datenstrom zu erzeugen.

Die folgenden Artikel habe ich bereits auf meiner pragmatischen Reise durch die drei neuen Schlüsselwörter co_return, co_yield und co_await verfasst:

• Einfache Futures mit Coroutinen implementieren
• Lazy Futures mit Coroutinen in C++20
• Mit Coroutinen einen Future in einem eigenen Thread ausführen
  

Ein Generator

Als Startpunkt meiner Variationen stelle ich einen Generator vor, der nur nach drei Werten fragt. Diese Vereinfachung und seine Visualisierung sollen helfen, den Arbeitsablauf des Generators besser zu verstehen:

// infiniteDataStreamComments.cpp

#include <coroutine>
#include <memory>
#include <iostream>

template<typename T>
struct Generator {
    
    struct promise_type;
    using handle_type = std::coroutine_handle<promise_type>;
    
    Generator(handle_type h): coro(h) {
        std::cout << "        Generator::Generator" << '\n';
    }                

    handle_type coro;
    
    ~Generator() {
        std::cout << "        Generator::~Generator" << '\n';
        if ( coro ) coro.destroy();
    }
    Generator(const Generator&) = delete;
    Generator& operator = (const Generator&) = delete;
    Generator(Generator&& oth): coro(oth.coro) {
        oth.coro = nullptr;
    }
    Generator& operator = (Generator&& oth) {
        coro = oth.coro;
        oth.coro = nullptr;
        return *this;
    }
    T getNextValue() {
        std::cout << "        Generator::getNextValue" << '\n';
        coro.resume();                                             // (13) 
        return coro.promise().current_value;
    }
    struct promise_type {
        promise_type() {                                           // (2)
            std::cout << "            promise_type::promise_type" << '\n';
        }                              
          
        ~promise_type() {
            std::cout << "            promise_type::~promise_type" << '\n';
        }
        
        std::suspend_always initial_suspend() {                    // (5)
            std::cout << "            promise_type::initial_suspend" << '\n';  
            return {};                                             // (6)
        }
        std::suspend_always final_suspend() noexcept {
            std::cout << "            promise_type::final_suspend" << '\n';
            return {};
        }
        auto get_return_object() {                                  // (3)
            std::cout << "            promise_type::get_return_object" << '\n';
            return Generator{handle_type::from_promise(*this)};     // (4)
        }
      
        std::suspend_always yield_value(int value) {                // (8)
            std::cout << "            promise_type::yield_value" << '\n';
            current_value = value;                                  // (9)
            return {};                                              // (10)
        }
        void return_void() {}
        void unhandled_exception() {
            std::exit(1);
        }

        T current_value;
    };

};

Generator<int> getNext(int start = 10, int step = 10) {
    std::cout << "    getNext: start" << '\n';
    auto value = start;
    for (true   ) {                                                 // (11)
        std::cout << "    getNext: before co_yield" << '\n';
        co_yield value;                                             // (7)
        std::cout << "    getNext: after co_yield" << '\n';
        value += step;
    }
}

int main() {
  
    auto gen = getNext();                                           // (1)
    for (int i = 0; i <= 2; ++i) {
        auto val = gen.getNextValue();                              // (12)
        std::cout << "main: " << val << '\n';                       // (14)
    }
    
}

Wird das Programm auf dem Compiler Explorer ausgeführt, legt es seinen Arbeitsablauf offen.

Nun widme ich mich der Analyse des Arbeitsablaufs.

Der Aufruf getNext() (Zeile 1) stößt das Erzeugen des Generator<int> an. Zuerst wird der promise_type (Zeile 2) generiert. Dann kreiert der Aufruf get_return_object (Zeile 3) den Generator (Zeile 4) und speichert ihn in einer lokalen Variable. Das Ergebnis dieses Ausrufs wird dem Aufrufer zurückgegeben, wenn diese Coroutine das erste Mal pausiert. Das initiale Pausieren findet sofort statt (Zeile 5). Da die Methode initial_suspend einen Awaitable std::suspend_always (Zeile 6) als Rückgabewert besitzt, geht der Kontrollfluss der Coroutine getNext weiter, bis der Aufruf co_yield value in Zeile 7 ausgeführt wird. Dieser Aufruf wird auf die Funktion yield_value(int value) (Zeile 8) abgebildet, und der aktuelle Wert wird hinterlegt: current_value = value (Zeile 9).

Die Methode yield_value(int value) gibt den Awaitable std::suspend_always (Zeile 10) zurück. Konsequenterweise wird die Ausführung der Coroutine pausiert, der Kontrollfluss geht an die main-Funktion zurück, und die for-Schleife beginnt (Zeile 11). Der Aufruf gen.getNextValue() (Zeile 12) stößt die Ausführung der Coroutine an, indem er diese dank coro.resume() (Zeile 13) wieder aufweckt.

Darüber hinaus gibt die Funktion getNextValue() den aktuellen Wert zurück, den der bereits erfolgte Aufruf der Methode yield_value(int value) (Zeile 8) vorbereitet hat. Zuletzt werden die erzeugte Zahl in Zeile 14 dargestellt und die for-Schleife weiter ausgeführt. Zum Abschluss werden Generator und Promise zerstört.

Modifikationen

Nach dieser detaillierten Analyse des Arbeitsablaufs möchte ich ihn leicht modifizieren. Mein Code-Schnipsel und die Zeilennummer basieren alle auf dem vorherigen Programm infiniteDataStreamComments.cpp. Der Einfachheit halber stelle ich nur die Veränderungen dar.

Die Coroutine wird nicht aufgeweckt

Wenn ich das Aufwecken der Coroutine (Zeile 12) und das Darstellen des Werts (Zeile 14) auskommentiere, pausiert die Coroutine sofort:

int main() {
  
    auto gen = getNext();
    for (int i = 0; i <= 2; ++i) {
        // auto val = gen.getNextValue();
        // std::cout << "main: " << val << '\n';                   
    }
    
}

Die Coroutine wird in diesem Fall nie ausgeführt, und der Generator und sein Promise werden erzeugt und gleich wieder zerstört.

initial_suspend pausiert nicht

Im Programm gibt die Methode initial_suspend den Awaitable std::suspend_always (Zeile 5) zurück. Wie es der Name andeutet, führt das Awaitable std::suspends_always dazu, dass die Coroutine sofort pausiert. Nun ersetze ich std::suspend_always mit std::suspend_never:

std::suspend_never initial_suspend() {  
    std::cout << "            promise_type::initial_suspend" << '\n';
    return {};
}

Jetzt wird die Coroutine sofort ausgeführt und pausiert erst, wenn die Funktion yield_value (Zeile 8) prozessiert wird. Der darauffolgende Aufruf gen.getNextValue() (Zeile 12) weckt die Coroutine wieder auf und stößt die Ausführung der Funktion yield_value nochmals an. Das Ergebnis ist, dass der Startwert 10 ignoriert wird und die Coroutine die Werte 20, 30 und 40 zurückgibt.

yield_value pausiert nicht

Die Methode yield_value (Zeile 8) wird durch den Aufruf co_yield value aufgerufen und bereitet den current_value (Zeile 9) vor. Die Funktion gibt den Awaitable std::suspend_always (Zeile 10) zurück und pausiert damit konsequenterweise. Ein anschließender Aufruf von gen.getNextValue (Zeile 12) muss daher die Corutine wieder aufwecken. Was passiert nun, wenn ich den Rückgabewert der Methode yield_value auf std::suspend_never ändere?

std::suspend_never yield_value(int value) {    
    std::cout << "            promise_type::yield_value" << '\n';
    current_value = value;
    return {};
}

Die while-Schleife (Zeile 1) läuft für immer und die Coroutine liefert keinen Wert.

@Matt Godbolt: Dies war kein "Denial of Service"-Angriff.

Wie geht's weiter?

In diesem Artikel habe ich nicht ausgenutzt, dass die Coroutine ein Klassen-Template ist. In meinem nächsten Artikel werde ich die Coroutine so verallgemeinern, sodass sie eine endliche Anzahl beliebiger Werte zurückgeben kann. ()