Asynchrone Programmierung – Teil 2: Koroutinen in C++ mit Boost.Asio

Mit dem Boost.Asio-Framework steht C++-Developern eine altbewährte Werkzeugsammlung zur Verfügung, die auch im modernen C++ noch ihre Berechtigung hat. Mit Kontexten, Exekutoren und Completion Tokens erlaubt sie es, asynchrone Programme nach unterschiedlichen Prinzipien sauber und effizient zu entwickeln. Callbacks Futures, spawn und yield_context gestatten auch einen an Koroutinen angelehnten Stil. Das hat der vorangegangene Artikel bereits gezeigt, der die Grundlagen von Boost.Asio vorgestellt hat.

Martin Meeser - Serie Asynchrone Programmierung

Martin Meeser ist selbständiger Diplominformatiker (Uni) und bietet Dienstleistungen zum Thema Softwareentwicklung an: Individual-Software-Entwicklung, Beratung zu Prozessen und Schulungen. In zahlreichen Projekten betreute er bisher Kunden unter anderem aus den Bereichen Automotive, Finance, Raumfahrt, Radioastronomie und Medizintechnik.

• Teil 1: C++ komfortabel mit Boost.Asio
• Python (nicht Teil der Serie)

Ab C++20 stehen nun aber Compiler-basierte stackless Koroutinen zur Verfügung. Damit kombiniert, spielt Boost.Asio seine vollen Stärken aus. Entwickler können jede Funktion oder Methode zu einer Koroutine ändern, indem sie eines der Schlüsselwörter co_yield, co_await odere co_return einsetzen. Der Compiler transformiert diese Funktion in eine Zustandsmaschine (Coroutine Frame), deren Ausführung an den durch co_await oder co_yield markierten Suspensionspunkten (Suspension Points) unterbrochen und später über einen std::coroutine_handle fortgesetzt wird. Der Coroutine Frame, der den Zustand speichert, liegt standardmäßig auf dem Heap, nicht auf dem Stack.

In Boost.Asio muss jede Koroutine eine Instanz vom Typ boost::asio::awaitable<T> zurückgeben. awaitable kapselt den Rückgabe-Typen: awaitable<void> bei einer Funktion ohne Rückgabewert, awaitable<int> bei int, awaitable<std::string> bei String-Typen usw. (Listing 1).

boost::asio::awaitable<void> async_void_sample()
{
    co_return;
}

boost::asio::awaitable<int> async_int_sample()
{
    co_return 42;
}

boost::asio::awaitable<std::string> async_string_sample()
{
    co_return "Hello async";

Listing 1: Kapselung der Rückgabe-Typen mit awaitable<T>

Für den Wechsel aus einem normalen Programmteil in einen Koroutinenteil dient die Funktion boost::asio::co_spawn. Sie erwartet drei Parameter:

1. den Executor (oder execution_context aus convenience) auf dem die verzahnte oder parallele Ausführung der Koroutinen erfolgt
2. eine Instanz von awaitable<T>
3. ein CompletionToken vom Typ detached, Funktions-Objekt oder use_future

Mit detached läuft awaitable in einem eigenen Ablauf-Ast, ohne dass man das Ergebnis verarbeiten kann. Dieses wird oft in der main-Methode verwendet, um initial eine Koroutine aufzurufen (Listing 2).

int main()
{
    boost::asio::io_context io_context;

    // co_spawn mit CompletionToken Function-Object
    boost::asio::co_spawn(io_context, async_int_sample(), [](std::exception_ptr, 
        int result)
    {
        std::cout << "async_int_sample() = " << result << std::endl;
    });

    // co_spawn mit CompletionToken use_future
    std::future future = boost::asio::co_spawn(io_context, 
        async_string_sample(), boost::asio::use_future);

    // co_spawn mit CompletionToken detached
    boost::asio::co_spawn(io_context, async_caller(), boost::asio::detached);

    io_context.run();

    std::string s = future.get();
    std::cout << "async_string_sample() = " << s << std::endl;

Listing 2: Beispiele von co_spawn mit awaitable zum Aufruf einer Koroutine.

Verwenden von co_await in Boost.Asio

Listing 3 zeigt die Verwendung von co_await. Bei dem Aufruf von awaitable mit co_await passiert Folgendes:

1. Der Executor kann an dieser Stelle die Ausführung der Koroutine unterbrechen. Ob die Unterbrechung tatsächlich stattfindet, ist unbekannt und unerheblich.
2. Unter Umständen führt der Executor andere Aktionen aus – verzahnt ( concurrent) auf einem Kontext mit einem Thread oder parallel auf einem Kontext mit mehreren Threads.
3. An einem dem Entwickler unbekannten Zeitpunkt bringt der Exekutor awaitable auf einem der Threads des execution_context zur Ausführung – auf welchem genau, ist unbekannt und unerheblich.

Sobald das Ergebnis von awaitable vorliegt, setzt der Exekutor die Koroutine an der Stelle fort, an der er sie verlassen hat.

boost::asio::awaitable<int> async_callee(int i)
{
    std::cout << "hello from awaitable, i=" << i << std::endl;
    co_return i + 1;
}

boost::asio::awaitable<void> async_caller()
{
    int i = co_await async_callee(1);
    // erzeugt ein awaitable, es wird aber nicht ausgeführt
    boost::asio::awaitable<int> aw = async_callee(2);
    // awaitable kann nicht kopiert werden, move erforderlich
    // co_await std::move(aw);
    std::cout << "i=" << i << std::endl;

    co_return;
}

// Ausgabe:
hello from awaitable, i=1
i=2

Listing 3: Beispiel für die Verwendung von co_await mit boost::asio::awaitable.

Der Ablauf aus Sicht der einzelnen Funktion ist also synchron – in dem Sinn, dass die Funktion stoppt, bis das Ergebnis von awaitable vorliegt. Es kommt aber nicht zu einer Blockierung des Threads, auf dem Entwickler ihre Funktion aufgerufen haben. Der Thread stand für andere Aktionen kooperativ zur Verfügung.

Entwickler können awaitable nur einmal verwenden. Wenn sie ein awaitable erzeugen, aber nicht mit co_await aufrufen, dann führt das Programm die Funktion nicht aus.

Listing 3 zeigt auch, wie der mit co_return zurückgegebene Wert durch den co_await-Ausdruck aus dem awaitable<T>-Objekt extrahiert und einer Variablen in der aufrufenden Funktion zugewiesen wird. Dies ist ein sehr komfortabler Mechanismus, schließlich ist gar nicht bekannt, auf welches Thread async_callee zur Ausführung kommt.

Thread-übergreifendes Exception-Handling bei Koroutinen

Unbehandelte Execeptions, die innerhalb einer Koroutine auftreten, nimmt die aufrufende Koroutine co_await entgegen. Koroutinen fangen Exceptions also analog zur synchronen Funktionsweise ab, unabhängig davon, in welchem konkreten Thread die Exception auftrat. Listing 4 zeigt ein kurzes Beispiel.

#include <boost/asio.hpp>
#include <iostream>

boost::asio::awaitable<void> async_ex_sample()
{
    throw std::runtime_error{ "some exception" };
    co_return;
}

int main()
{
    boost::asio::thread_pool pool(4);
    boost::asio::co_spawn(pool, []()->boost::asio::awaitable<void>
    {
        try
        {
            co_await async_ex_sample();
        }
        catch (const std::exception& ex)
        {
            std::cout << ex.what() << std::endl;
        }
        co_return;
    }, boost::asio::detached);

    pool.join();
}

Listing 4: Thread-übergreifendes Exception-Handling.

Verwendung der Boost-Asio Funktionen mit CompletionToken use_awaitable

Listing 5 zeigt, wie Entwickler die asynchronen Funktionen der Boost.Asio-Bibliothek mit co_await verwenden: Dazu übergeben sie das CompletionToken use_awaitable als Parameter an die Methode async_wait des timer – Objektes, dementsprechend ist der Rückgabetyp der Funktion nun awaitable.

boost::asio::awaitable<void> async_sample(
    boost::asio::steady_timer timer, 
    boost::asio::ip::tcp::socket socket)
{
    co_await timer.async_wait(boost::asio::use_awaitable);
    char buf[4096];
    std::size_t n = co_await socket.async_read_some(boost::asio::buffer(buf), 
         boost::asio::use_awaitable);
}

Listing 5: Beispiel für die Verwendung der Boost.Asio-Funktionen mit co_await und dem CompletionToken use_awaitable.

Asynchrone Schleifen

Ein weiterer Vorteil der Koroutinen ist, dass sich damit Schleifen im gewohnten for- oder while-Stil auch für asynchrone Abläufe formulieren lassen. Ohne Koroutinen müsste man die Iterationen in Callbacks auflösen oder komplexere Zustandsautomaten schreiben.

Der Code in Listing 6 liest einen Socket in einer Endlosschleife und schreibt die empfangenen Daten direkt wieder zurück – ein einfacher Echo-Server. Obwohl der Code wie eine normale Schleife aussieht, blockiert er keinen Thread. Jeder co_await-Ausdruck gibt die Kontrolle an den Exekutor zurück. Sobald Daten verfügbar sind oder ein Schreibvorgang abgeschlossen ist, läuft die Schleife an der unterbrochenen Stelle weiter – auf welchem Thread die Lese- und Schreibvorgänge ablaufen, ist unbekannt – dies wird durch den Exekutor bestimmt, den der Entwickler im co_spawn angegeben hat.

boost::asio::awaitable<void> async_sample(boost::asio::ip::tcp::socket socket)
{
    char buf[4096];
    for (;;)
    {
        std::size_t n = co_await socket.async_read_some(
            boost::asio::buffer(buf),
            boost::asio::use_awaitable);

        co_await boost::asio::async_write(
            socket,
            boost::asio::buffer(buf, n),
            boost::asio::use_awaitable);
    }
}

Listing 6: Beispiel für asynchrone for-Iteration.

Fehlerbehandlungen

Wie in vorangegangenen Beispielen bereits gezeigt, gibt es bei Koroutinen mehrere Möglichkeiten der Fehlerbehandlung:

1. keine Fehlerbehandlung
2. mit Try-Catch
3. lokale Fehlervariable mit boost::asio::redirect_error als CompletionToken
4. als Teil eines Tupel-Rückgabewertes mit dem CompletionToken boost::asio::as_tuple.

Das folgende Listing 7 zeigt entsprechende Beispiele:

boost::asio::awaitable<void> errors_sample(boost::asio::ip::tcp::socket socket)
{
    boost::asio::any_io_executor executor = co_await    
        boost::asio::this_coro::executor;
    std::array<char, 4> buffer;

    // 1. ohne Fehler-Behandlung
    std::size_t n = co_await socket.async_read_some(boost::asio::buffer(buffer), 
        boost::asio::use_awaitable);

    // 2. mit try-catch
    try
    {
        std::size_t bytes_read = co_await    
            socket.async_read_some(boost::asio::buffer(buffer),  
                boost::asio::use_awaitable);
    }
    catch (const boost::system::system_error& e)
    {
        std::cerr << "Boost error during async_read_some: " << e.what() << 
            std::endl;
    }

    // 3. mit redirect_error CompletionToken
    boost::system::error_code ec;
    std::size_t bytes_read2 = co_await   
        socket.async_read_some(boost::asio::buffer(buffer),  
            boost::asio::redirect_error(boost::asio::use_awaitable, ec));
    if (ec)
    {
         std::cerr << "Error using error_code: " << ec.message() << std::endl;
    }

    // 4. mit as_tuple CompletionToken
    auto [ec2, bytes_read3] = co_await    
        socket.async_read_some(boost::asio::buffer(buffer),      
            boost::asio::as_tuple(boost::asio::use_awaitable));
    if (ec2)
    {
        std::cerr << "Error using as_tuple: " << ec2.message() << std::endl;
    }
}

Listing 7: Beispiele für verschiedene Arten der Fehlerbehandlung in Boost.Asio.