35.4 Generator és streaming feldolgozás

A streaming feldolgozás tipikus DevOps-feladat: egy több gigabájtos naplót kell szűrni, oszlopokra bontani, majd az első néhány hibás sort kiírni. Ha a teljes bemenetet előbb std::vector-ba olvassuk, a memóriahasználat $O(n)$, a feldolgozás csak a beolvasás után indul. A pull generátor ehelyett $O(1)$ állapotot tart, és a fogyasztó diktálja a tempót: ez a backpressure legegyszerűbb formája. Ebben az alfejezetben a generátort mint single-pass input range-et szerződésként rögzítjük, megépítjük a minimális C++20 változatot, majd összevetjük a C++23-as std::generator-ral. A coroutine gépezetét mint felfüggeszthető állapotgépet a Coroutine mint felfüggeszthető állapotgép alfejezetben már rögzítettük, a felfüggesztéskori érvénytelenedést és a megszakítást a Dangling capture, resumption és cancellation tárgyalta; itt csak a generátor-specifikus következményeket vezetjük le.

35.4.1 A pull generátor mint single-pass állapotgép

Definíció. Egy pull generátor olyan coroutine-objektum, amelynek a promise_type-ja definiál yield_value tagot, a fogyasztó pedig egy input_range interfészen keresztül, előrefelé, pontosan egyszer haladva olvassa az elemeket. A szerződés három része:

1. minden ++it pontosan egy resume(), és minden resume() legfeljebb egy új elemet tesz elérhetővé;
2. az iteratorok nem független bejárások: egy iterator másolása ugyanahhoz a coroutine-frame-hez mutat;
3. a begin() érdemben legfeljebb egyszer hívható meg — a tartomány nem újrajárható.

Az állapotgép négy állapota (a mi implementációnkban):

A final_suspend szándékosan suspend_always: így a frame a befejezés után is él, amíg a tulajdonos meg nem semmisíti, tehát a promise-ban tárolt utolsó érték és a std::exception_ptr olvasható marad.

A pull modell megkülönböztetője a push (callback) modellel szemben, hogy a fogyasztó vezérel: views::take(3) három elem után egyszerűen nem hív ++-t, és a termelő soha nem fut tovább. Ez a backpressure, és egyben a végtelen generátorok kezelésének alapja.

35.4.1.1 A yieldelt érték és a referencia élettartama

A co_yield expr a fordító szerint promise.yield_value(expr) hívás lesz, majd felfüggesztés. Ebből két, minőségileg eltérő eset következik:

• Értéktároló promise (yield_value(T value), a frame-ben T current): az expr bekerül a frame-be (másolás vagy mozgatás). Az operator* által adott const T& a frame-re mutat, ezért addig érvényes, amíg a frame él és a következő resume() felül nem írja.
• Referenciatároló promise (co_yield egy lokálisra, a promise T&/const T&-et tárol): a frame csak a hivatkozást őrzi. A hivatkozott objektum élettartamát a coroutine lokálisa adja, amely a következő resume-nál megszűnhet vagy megváltozhat — innen ered a klasszikus dangling.

A gyakorlati szabály: a *it referenciáját ne őrizd meg ++it után. Ha az értékre a léptetés után is szükség van, másold ki (T copy = *it;), vagy a generátor yieldeljen értéket, ne referenciát.

35.4.1.2 Invalidáció a következő resume-nál

A ++it nem “előrelépés egy tárolóban”, hanem a coroutine folytatása. Ennek következményei:

• minden korábban szerzett pointer/referencia a ++it után érvénytelen (dangling vagy elavult érték);
• az iterator másolatai ugyanazt a frame-et léptetik, ezért két “másolat” összefonódik;
• a generátor objektum megsemmisítése az összes iterátorát érvényteleníti — a for (int v : make_gen()) esetben a temporális generátor élettartama a ciklusra kiterjesztett, de egy auto it = make_gen().begin(); már dangling.

35.4.2 Minimális C++20 generátor

Az alábbi fejléc önállóan fordítható, és csak a C++20 coroutine és <iterator> eszközeit használja:

// gen20.hpp — minimális single-pass pull generátor (C++20)
#include <coroutine>
#include <cstddef>
#include <exception>
#include <iterator>
#include <utility>

template <class T>
class Generator {
public:
    struct promise_type {
        T current{};
        std::exception_ptr failure{};

        Generator get_return_object() noexcept {
            return Generator{
                std::coroutine_handle<promise_type>::from_promise(*this)};
        }
        std::suspend_always initial_suspend() noexcept { return {}; }
        std::suspend_always final_suspend() noexcept { return {}; }
        std::suspend_always yield_value(T value) {      // értéktároló promise
            current = std::move(value);
            return {};
        }
        void return_void() const noexcept {}
        void unhandled_exception() noexcept { failure = std::current_exception(); }
    };

    class iterator {
    public:
        using iterator_concept  = std::input_iterator_tag;
        using iterator_category = std::input_iterator_tag;
        using value_type        = T;
        using difference_type   = std::ptrdiff_t;
        using pointer           = const T*;
        using reference         = const T&;

        iterator() noexcept = default;
        explicit iterator(std::coroutine_handle<promise_type> h) : h_{h} { step(); }

        reference operator*()  const noexcept { return h_.promise().current; }
        pointer   operator->() const noexcept { return &h_.promise().current; }
        iterator& operator++() { step(); return *this; }
        void      operator++(int) { step(); }

        bool operator==(const iterator& o) const noexcept { return h_ == o.h_; }

    private:
        void step() {                       // egyetlen resume + hibaátadás
            if (!h_) return;
            h_.resume();
            if (h_.done()) {
                if (auto e = h_.promise().failure) { h_ = {}; std::rethrow_exception(e); }
                h_ = {};
            }
        }
        std::coroutine_handle<promise_type> h_{};   // nem tulajdonol
    };

    Generator() noexcept = default;
    Generator(Generator&& o) noexcept
        : h_{std::exchange(o.h_, {})}, started_{std::exchange(o.started_, false)} {}
    Generator& operator=(Generator&& o) noexcept {
        if (this != &o) { if (h_) h_.destroy(); h_ = std::exchange(o.h_, {});
                          started_ = std::exchange(o.started_, false); }
        return *this;
    }
    Generator(const Generator&) = delete;
    Generator& operator=(const Generator&) = delete;
    ~Generator() { if (h_) h_.destroy(); }            // korai leállítás cleanupja

    iterator begin() {
        if (!h_ || started_) return iterator{};       // single-pass: másodszor üres
        started_ = true;
        return iterator{h_};
    }
    iterator end() const noexcept { return iterator{}; }

private:
    explicit Generator(std::coroutine_handle<promise_type> h) noexcept : h_{h} {}
    std::coroutine_handle<promise_type> h_{};
    bool started_{false};
};

Használat és build:

// demo.cpp
#include "gen20.hpp"
#include <iostream>
#include <ranges>
#include <vector>

Generator<int> evens(int limit) {                    // előfeltétel: limit < INT_MAX
    for (int i = 0; i < limit; i += 2) co_yield i;   // érték bemásolódik a frame-be
}

int main() {
    auto g = evens(10);                              // g az owner
    for (int v : g) std::cout << v << ' ';           // elvárt: 0 2 4 6 8

    auto g2 = evens(10);
    auto view = std::views::all(g2)                  // lvalue -> ref_view, nem másol
              | std::views::transform([](int v) { return v * v; })
              | std::views::take(3);
    std::vector<int> out;                            // C++20 materializálás
    for (int v : view) out.push_back(v);             // elvárt: 0 4 16
    for (int v : out) std::cout << v << ' ';
}

# a laborprofilban rögzített, C++20 coroutine-t támogató fordítóval
c++ -std=c++20 -Wall -Wextra -Wpedantic -O2 -g demo.cpp -o demo

Az eredményül kapott sorozatok elvárt értékek, nem lefuttatott mérések. A ranges-megfelelést érdemes fordításidejű ellenőrzéssel rögzíteni:

static_assert(std::ranges::input_range<Generator<int>>);

Ha ez a laborodon nem teljesül, a hibaüzenet megnevezi a hiányzó iterátor-tagot (a sentinel és a teljes C++20 iterátorkövetelmény-rendszer szándékosan kimaradt a minimalizmus miatt).

35.4.2.1 A frame tulajdonosa

A coroutine-frame egyetlen tulajdonost kap: a Generator objektumot. Ezért a generátor move-only — két másolat ugyanazt a frame-et semmisítené meg. Az iterator nem tulajdonol; ha tovább él, mint a generátor, dangling. Ugyanez a szabály vonatkozik a coroutine paramétereire is: a paraméterek a frame-be másolódnak, tehát egy std::istream& paraméter a frame-ben is hivatkozás marad — a hivatkozott streamnek túl kell élnie a generátort.

35.4.3 Kivétel, korai leállítás és cleanup

Mivel az unhandled_exception noexcept, a coroutine testében dobott kivételt el kell tárolni, és a fogyasztó oldalán újra kell dobni — ezt teszi a step(). Két fontos következmény:

• Ha a fogyasztó break-kel kilép, mielőtt a hibát okozó pontig eljutna, a kivétel soha nem kerül a felszínre: a exception_ptr a frame-mel együtt megsemmisül. Ez nem hiba, hanem a lazy kiértékelés természete — de dokumentálni kell.
• A ~Generator() destroy()-ja a frame lokálisait megsemmisíti, a coroutine testét viszont nem folytatja: catch ágak és a co_yield utáni kód nem fut le. RAII-szemmel ez helyes viselkedés, a co_yield előtt szerzett erőforrás felszabadul.

struct Watch { static inline int alive = 0;
               Watch() { ++alive; } ~Watch() { --alive; } };

// Korlátlan, definiált sorozat: unsigned számláló, amely 2^N után kimondottan
// körbefordul (0, 1, ..., UINT_MAX, 0, ...); signed túlcsordulás nincs.
Generator<unsigned> forever() {
    Watch w;                                  // frame-lokális, RAII
    for (unsigned i = 0;; ++i) co_yield i;    // végtelen generátor
}
// for (unsigned v : forever()) { if (v == 2) break; }   -> Watch::alive == 0

A „végtelen” generátor szerződése csak akkor teljes, ha a sorozat minden eleme definiált: egy int számláló a INT_MAX után előjeles túlcsordulással nem definiált viselkedést okozna, ezért a számláló unsigned, amelynek körbefordulása a nyelv szerint definiált modulo-aritmetika. (Az evens ugyanezért kap előfeltételt: limit < INT_MAX, különben az i += 2 túlcsordulhatna.)

Buildjegyzőkönyv (lab-gcc profil). A gen20.hpp-vel és a fenti Watch-csal készült próba -fsanitize=address,undefined alatt forever_test: OK-val futott: a v == 2 utáni break után Watch::alive == 0, és az evens(10) összege 20.

Veszélyes minta a co_yield egy lock vagy tranzakció közben: a felfüggesztés alatt a lock tartva marad, és ha a fogyasztó eldobja a generátort, a lock a frame megsemmisítésekor szabadul fel — nem a yield után. Ilyen esetben a co_yield kerüljön a kritikus szakaszon kívülre.

35.4.4 A std::generator C++23-ban és a capability check

A C++23 std::generator a szabványos megoldás ugyanerre a szerepre: <generator> fejléc, input_range és view modell, a co_yield értéket vagy referenciát is yieldelhet a template-paramétertől függően. A mi kézi változatunkhoz képest a fő nyereség, hogy view, ezért közvetlenül illeszthető pipeline-ba, és hogy a value/reference típuslevezetést a szabvány rögzíti.

// lines23.cpp — C++23 std::generator
#include <generator>
#include <iostream>
#include <ranges>
#include <string>

std::generator<std::string> lines(std::istream& in) {
    for (std::string line; std::getline(in, line); )
        co_yield line;                 // másolat a frame value slotjába
}

int main() {
    auto v = lines(std::cin)
           | std::views::filter([](const std::string& s) { return !s.empty(); })
           | std::views::take(3);
    for (const auto& s : v) std::cout << s << '\n';
}

# a laborprofilban rögzített, C++23 <generator>-t ismerő fordítóval
c++ -std=c++23 -Wall -Wextra -Wpedantic -O2 -g lines23.cpp -o lines23

Capability check: a <generator> fejléc jelenléte önmagában nem elég, mert C++20 módban is látszhat, miközben a tartalom nem használható. Ellenőrizz nyelvi módot, fejlécet és library feature-testet együtt:

#if __cplusplus > 202002L && defined(__has_include)
#  if __has_include(<generator>)
#    include <generator>
#    define HAVE_STD_GENERATOR 1
#  endif
#endif

CMake-oldalon ugyanez a check_cxx_source_compiles modullal próbálható ki, a következő próbaforrással:

#include <generator>
#include <ranges>
static_assert(std::ranges::input_range<std::generator<int>>);
int main() {}

Korlát: ehhez az alfejezethez nem állt rendelkezésre ellenőrzött forrásjegyzet, ezért a std::generator pontos reference/value_type levezetését és a vonatkozó feature-test makró értékét nem tekintem igazoltnak; ezeket a laborban, a rögzített fordítón és standard library-n kell ellenőrizni. A co_await mint önálló thread- vagy hálózati runtime, illetve egy teljes coroutine framework itt kifejezetten a témahatáron kívül esik.

35.4.5 Ranges pipeline és materializálás

A generátor akkor igazán hasznos, ha a views::transform/filter/take pipeline eleme. A saját C++20 változatnál az lvalue generátort std::views::all(g)-lal kell bevonni (így ref_view lesz, nem másolat), a std::generator viszont maga is view, ezért közvetlenül |-elhető. A különbség élettartam-következménye:

auto v = lines(std::cin) | std::views::take(3);   // a view BIRTOKOLJA a generátort
for (auto&& s : v) { /* ... */ }                  // biztonságos

Itt a temporális std::generator a pipeline-ba kerül, és a v élettartamáig él. A kézi változatban ugyanez a views::all-lal nem fordul le (nem view), ezért ott külön nevesített generátor kell — ez véletlenül véd a danging ellen. Ha viszont iterátort vagy subrange-t mentesz ki a pipeline-ból, és a pipeline megszűnik, az iterátor érvénytelenné válik; ezt a Ranges- és callback-élettartam labor eszközeivel érdemes ellenőrizni.

Materializálás (std::vector-ba gyűjtés) akkor helyes, ha a korlát ismert és kicsi: C++23-ban std::ranges::to<std::vector>(), C++20-ban std::ranges::copy(view, std::back_inserter(v)) vagy egyszerű ciklus. Végtelen generátorból views::take nélkül a materializálás nem terminál — ez a leggyakoribb streaming-hiba.

35.4.6 Tipikus hibák és javításuk

• Referencia megtartása léptetés után. const auto& x = *it; ++it; use(x); — dangling. Javítás: T copy = *it; a léptetés előtt.
• Kétszeri begin(). A mi implementációnk csendben üres range-et ad; ez nem hibaüzenet, ezért a fogyasztói kódot kell úgy írni, hogy egyszer járja be.
• Iterator másolása független bejárásként. Single-pass: a másolat ugyanazt a frame-et lépteti.
• Generátor lokálisára yieldelés referenciával. A következő iterációban a lokális megszűnik vagy felülíródik; értéktároló promise esetén ez nem fordulhat elő.
• Kivétel elnyelése break esetén. Ha a hiba számít, a fogyasztó ne lépjen ki korábban, vagy a generátor jelezze előre a hibalehetőséget.
• Lock tartása co_yield alatt. A felfüggesztés meddig tart, a fogyasztó dönti el; a kritikus szakasz ne ölelje át a yieldet.
• Materializálás végtelen forrásból. Mindig views::take/views::take_while a gyűjtés előtt.

35.4.7 Összefoglaló

A pull generátor egy single-pass input_range, amelynek állapotgépe a fogyasztó ++ hívásaira lép. Az operator* referenciája a frame value slotjára mutat, ezért csak a következő resume()-ig érvényes; a frame tulajdonosa a generátor objektum, az iterator nem tulajdonol. A kivételt a promise tárolja, a fogyasztó oldalán kell újradobni; a korai kilépés a frame destroy()-jával RAII-helyesen takarít, de a már eltárolt kivételt elnyeli. A C++20 kézi változat kis kóddal működik és véd a view-alapú dangling ellen; a C++23 std::generator szabványos view, de használata előtt nyelvi módot, fejlécet és library feature-testet ellenőrizni kell.

35.4.8 Feladatok

1. Magyarázó. Egészítsd ki a gen20.hpp Generator osztályát egy static_assert-tel, amely ellenőrzi a std::ranges::input_range megfelelést, és indokold, mely iterátor-tagok hiányozhatnak. Írd le, miért helyes a single-pass viselkedés egy naplófájl-feldolgozónál.

2. Diagnosztikai. Az alábbi kód dangling referenciát produkál; keresd meg a hibát, indokold a frame élettartamával, és add meg a javítást:

   Generator<const std::string&> bad(std::vector<std::string>& v) {
       for (auto& s : v) co_yield s;   // feltételezzük: referenciát tároló promise
   }

3. Kódolási. Írj Generator<std::string>-et C++20-ban, amely egy std::istream-ből olvas sorokat, kihagyja az üreseket és a #-kel kezdődő kommenteket, majd views::take(5)-tel állítsd meg. Igazold, hogy a korai leállás után a frame-lokális Watch megsemmisül.

4. Ranges és materializálás. Építs pipeline-t a generátorodra: filter → transform → take(3), majd materializálj std::vector-ba. Add meg a C++20-as fallbacket és a C++23-as ranges::to változatot, valamint azt a feltételt, amely mellett a materializálás memóriahasználata elfogadható.

35.4.9 Megoldások

1. megoldás. Az ellenőrzés static_assert(std::ranges::input_range<Generator<int>>); alakban tehető a fejléc aljára. A megfeleléshez a C++20 input_iterator-hez iterator_concept, value_type, difference_type, reference, valamint az iter_move/iter_swap konzisztenciája kell; a sentinel_for-hoz pedig a ==. A minimalizmus miatt ezek egy része hiányozhat, ezért a static_assert a dokumentáció része: a hibaüzenet nevezi meg a hiányt. A single-pass viselkedés helyes, mert egy naplófájl nem újrajárható visszafelé pufferelés nélkül, és a fogyasztó soha nem lép hátra; a memóriakorlát $O(1)$ marad. Ellenőrzés: a kód forduljon -std=c++20 módban, és a static_assert ne adjon hibát a labor profiljában.

2. megoldás. A hiba nem is annyira a co_yield-ben, mint a szerződésben van: ha a promise const std::string&-et tárol, akkor a hivatkozott s élettartama a v-hez kötött, ami túlélheti a generátort — de a léptetés szempontjából a helyzet romlik, ha a ciklusváltozó lokális lenne. A valódi veszély: ha a for ciklus egy lokális std::string-re yieldelne referenciát, a következő iterációban a lokális megszűnik. Javítás: értéktároló promise (Generator<std::string> és co_yield s; másol), vagy a hivatkozott konténer élettartamának dokumentált garantálása. Ellenőrzés: a javított változatnál a *it másolata a ++it után is helyes értéket ad.

3. megoldás. A generátor teste:

Generator<std::string> config_lines(std::istream& in) {
    for (std::string line; std::getline(in, line); ) {
        if (line.empty() || line.front() == '#') continue;
        co_yield line;               // értéktároló promise: biztonságos másolat
    }
}

Fogyasztás: for (auto&& s : config_lines(in) | std::views::take(5)). A Watch-ot a for ciklus elé, a coroutine testébe kell tenni; break után Watch::alive == 0 az elvárt érték (nem mért eredmény). Ellenőrzés: a break-et követően a számláló nulla, és a stream továbbra is használható (a generátor nem zárja le, mert nem tulajdonolja).

4. megoldás. C++20:

auto g = source();
auto view = std::views::all(g)
          | std::views::filter([](const std::string& s) { return s.size() > 3; })
          | std::views::transform([](const std::string& s) { return s.substr(0, 3); })
          | std::views::take(3);
std::vector<std::string> out;
std::ranges::copy(view, std::back_inserter(out));

C++23-ban auto out = view | std::ranges::to<std::vector>();. Indoklás: a materializálás költsége $O(k)$ elem és $O(k)$ memória, ahol $k$ a take korlátja; ez csak akkor elfogadható, ha $k$ előre ismert és kicsi. Ellenőrzés: a take nélküli változat végtelen forráson nem terminál — ezt a teszt a fordítással nem, csak futásidőben (timeouttal) tudja kimutatni.