39.3 Task, queue és szolgáltatás-életciklus

Ez az alfejezet a C záróprojekt funkcionális szerződését megtartó C++ port konkurens rétegével foglalkozik. A C++ core: parser, erőforrások és értéktípusok (39.2) részben rögzítettük, hogyan lesz a C struct + kézi free mintákból értékszemantikájú, RAII-val kezelt C++ mag; ott a hangsúly a szinkron adatúton volt. Most azt a kérdést vesszük elő, amely a legtöbb termelési hibát okozza: mi történik, ha a szolgáltatásnak egyszerre több szála dolgozik, a bemenet korlátos pufferen érkezik, és a leállítást bármelyik pillanatban kérhetik. A cél nem egy új keretrendszer, hanem egy kicsi, indokolható szerződés: ki tulajdonol mit, meddig él, mi történik telítődéskor, és milyen sorrendben lehet hézagmentesen leállni.

A fejezet épít a shared ownership, reclamation és konkurens review (34.6) megosztott tulajdonlási és visszanyerési szabályaira, valamint az Aszinkron I/O adapter és coroutine-labor (35.5) felfüggesztési és élettartam-megfontolásaira. A coroutine adaptert itt csak érintjük, mert az a részletes kidolgozást ott kapta meg; itt az a kérdés, hogyan illeszkedik a szolgáltatás-életciklusba egy kísérleti, külön capability profile alatt.

39.3.1 A szolgáltatás-életciklus mint szerződés

Szolgáltatás-életciklusnak nevezzük azon állapotok halmazát, átmeneteik megengedett sorrendjét és az egyes állapotokhoz tartozó erőforrás-kötelezettségeket, amelyek egy hosszú életű, konkurens komponens viselkedését meghatározzák. Ez nem puszta dokumentáció: a C++ nyelvi eszközeivel részben kikényszeríthető, részben pedig tesztelhető. A szerződés három dolgot rögzít:

1. Mely műveletek engedélyezettek mely állapotban. Ha a submit a leállítás megkezdése után is elfogad munkát, az vagy elveszik, vagy a leállás soha nem fejeződik be.
2. Ki az owner (tulajdonos) az egyes erőforrások felett. A queue, a worker szálak, a connection objektumok és a callbackek tulajdonosa a service objektum; minden más ezekhez képest borrowed (kölcsönzött) hozzáférés.
3. Mikor tekinthető egy erőforrás felszabadíthatónak. Csak akkor, ha az ahhoz tartozó összes outstanding operation (befejezetlen művelet) befejeződött vagy megszakított, és ezt a tényt valamilyen szinkronizációs eszköz igazolja.

Outstanding operation: olyan művelet, amelyhez erőforrás (buffer, socket, callback, szál) tartozik, és amelynek végrehajtása megkezdődött, de a befejezés ténye még nem került publikálásra. A definíció szándékosan a publikáláson van: attól, hogy egy worker szál lokálisan „már majdnem végzett”, a művelet a többi szál számára még folyamatban van, amíg egy happens-before élt nem létesítünk (jellemzően join, mutex, atomikus művelet vagy condition variable segítségével).

Quiescence (nyugvás): az az állapot, amelyben nincs outstanding operation, a queue kiürült vagy lezárult, és egyetlen szál sem hajthat végre olyan kódot, amely a szolgáltatás erőforrásaihoz nyúl. A teardown akkor helyes, ha a nyugvást megállapítja, mielőtt a tagok destruktorai lefutnak.

A mi állapotgépe:

Egy fontos tervezési döntés, hogy a Draining állapotot nem külön enumerációval, hanem a queue closed_ flagjével és a stop token állapotával reprezentáljuk. Ez kevesebb duplikált állapotot jelent, viszont a két jelzés szemantikája eltér: a close() a bemenet lezárása, a stop kérés a szálak kooperatív megszakítása. A kettő nem ugyanaz, és a leállási protokoll éppen abból áll, hogy a kettőt helyes sorrendben és egymásra hatva használjuk.

39.3.2 Bounded queue, stop token és backpressure

A bounded queue (korlátos sor) olyan sor, amelyre invariánsként teljesül, hogy elemszáma minden megfigyelhető pillanatban $0 \le n \le C$, ahol $C$ a kapacitás, és hogy close() után nem kerülhet be új elem. A korlát nem kényelmi kérdés: ez az egyetlen hely, ahol a rendszer memóriahasználata felülről becsülhető.

Backpressure (visszanyomás): az a mechanizmus, amellyel a fogyasztó telítettsége a termelő felé továbbítódik, és ott várakozást, hibát vagy eldobást vált ki. Három alapvető politikát érdemes ismerni, és ezeket nem szabad összekeverni:

• Blokkoló: a termelő addig vár, amíg hely nem lesz. Erős memóriakorlát, de a termelő szálát lefoglalja, ezért csak akkor elfogadható, ha a termelő nem tart más zárat.
• Eldobó: a termelő azonnal visszajelez (pl. false), és a hívó dönt a sorsról. Akkor jó, ha a forrás lassítható, de nem akarunk blokkolni.
• Lezáró: a telítődés hibának számít, és a kapcsolat bezárul. Ezt csak akkor szabad alkalmazni, ha a protokoll ezt kifejezetten megengedi, mert a csendben eldobott adat a hálózati framinget is tönkreteheti.

A stabilitás levezetése rövid. Legyen $T$ egy frame várható feldolgozási ideje, $W$ a workerek száma. A feldolgozó kapacitás közelítőleg $W/T$ frame/másodperc. Ha a hosszú távú érkezési ráta $\lambda < W/T$, a sor várható hossza korlátos marad; ha $\lambda \ge W/T$, akkor a sor hossza a korlátig nő, és a backpressure folyamatosan aktív lesz. A korlátos sor ezt nem szünteti meg, hanem láthatóvá és kezelhetővé teszi: a telítődés nem csendes memóriafelfúvódásként, hanem visszajelzésként vagy hibaként jelenik meg. A szolgáltatás saját tárolójának felső korlátja akkor adható meg, ha minden olyan frame-et elszámolunk, amelyet a szolgáltatás birtokol. Ezek: legfeljebb $C$ frame a sorban, legfeljebb $W$ a workereknél feldolgozás alatt, és legfeljebb $P$ olyan submit hívás paraméterében, amely a teli sor miatt blokkol. Ha egy frame legfeljebb $S_{\max}$ bájt hasznos adatot hordoz, akkor

$$M \le (C + W + P)\cdot S_{\max} + K,$$

ahol $K$ a fix overhead (szálak, sorcsomópontok, mutexek, könyvtári pufferek). A becslés két kódbeli feltételen múlik. A submit az $S_{\max}$-nál nagyobb frame-et még a sor előtt, too_large eredménnyel elutasítja. A blokkoló hívók száma legfeljebb $P$: aki a $P+1$-edik lenne, azonnal busy eredményt kap, és a frame-et a hívás végén felszabadítja. A hívó saját, még be nem adott frame-jei nem a szolgáltatás tárolói; azokat a hívó kerete számolja el. Ezek a feltételek nélkül a korlát nem érvényes: korlátlan számú blokkoló producer mindegyike egy további frame-et tartana életben a soron kívül.

A C++20 stop token mechanizmusa egy közösen fenntartott, szálbiztos stop-állapotra épül, amelyben három szereplő osztozik:

• a std::stop_source leállítást kérhet (request_stop()), és részt vesz a stop-állapot élettartamában;
• a std::stop_token csak lekérdezhet (stop_requested(), stop_possible()), és rá callback regisztrálható; ő is részt vesz a stop-állapot élettartamában, ezért nem tekinthető puszta, nem tulajdonló nézetnek;
• a std::stop_callback regisztrált reakció: ha a leállítást már kérték, a konstruktorában azonnal lefut; egyébként a kérést végző szálon fut. Destruktora deregisztrál, és ha a callback éppen fut egy másik szálon, megvárja.

A stop-állapot addig él, amíg bármely source vagy token hivatkozik rá. Egy source élettartamán túl megőrzött token ezért nem dangling: a vizsgálata definiált. A stop_requested() megőrzi a kérés tényét a source megszűnése után is. A stop_possible() más kérdést válaszol meg: igaz, ha a leállítást már kérték, vagy még létezik olyan source, amely kérheti; hamis egy alapértelmezett konstrukciójú, stop-állapot nélküli tokennél, illetve ha az utolsó source kérés nélkül megszűnt. A tárolás élettartama és a kérhetőség tehát két külön tulajdonság:

// tests/stop_state_probe.cpp -- a token a megosztott stop-állapot része
#include <cstdio>
#include <cstdlib>
#include <stop_token>

#define CHECK(condition) do {                                      \
    if (!(condition)) {                                            \
        std::fprintf(stderr, "CHECK failed at line %d\n", __LINE__); \
        std::exit(EXIT_FAILURE);                                   \
    }                                                              \
} while (0)

int main()
{
    std::stop_token kept;
    CHECK(!kept.stop_possible());                 // nincs stop-állapot

    {
        std::stop_source source;
        kept = source.get_token();
        CHECK(kept.stop_possible() && !kept.stop_requested());
    }   // az utolsó source kérés nélkül megszűnt
    CHECK(!kept.stop_possible() && !kept.stop_requested());

    {
        std::stop_source source;
        kept = source.get_token();
        source.request_stop();
    }   // a kérés megmarad a source után is
    CHECK(kept.stop_requested() && kept.stop_possible());

    bool ran = false;
    std::stop_callback late(kept, [&] { ran = true; });
    CHECK(ran);                                   // már kért stop: azonnal fut
    return 0;
}

A próba G++ 13.3.0 és libstdc++ mellett, -std=c++20 -Werror beállítással fordult, és ASan/UBSan builddel is sikeresen lefutott.

A std::jthread destruktora request_stop()-ot hív, majd joinol, tehát a „elfelejtett join” hiba alapértelmezésben megszűnik. A stop token nem megszakítja a blokkoló hívást, hanem a condition_variable_any stop tokenes wait overloadjai teszik megszakíthatóvá a várakozást. Ezek az overloadok C++20 újak; a konkrét libstdc++/libc++ kiadás támogatását a laborprofilban ellenőrizni kell, és nem szabad feltételezni, hogy egy régebbi eszközlánc elfogadja a kódot.

A következő példa a teljes, önállóan fordítható queue. A szerződés: push blokkol, ha a sor tele van, de stop kérésre false-szal visszatér; pop a stop kérés után is kiveszi a már bekerült elemeket (drain), és csak akkor ad nullopt-ot, ha a sor kiürült. Ez a „drain on stop” politika.

// bounded_queue.hpp
#pragma once
#include <chrono>
#include <condition_variable>
#include <cstddef>
#include <deque>
#include <mutex>
#include <optional>
#include <stop_token>
#include <utility>

template <class T>
class bounded_queue {
public:
    explicit bounded_queue(std::size_t capacity) : cap_(capacity) {}
    bounded_queue(const bounded_queue&) = delete;
    bounded_queue& operator=(const bounded_queue&) = delete;

    // Nem blokkol; false: tele van vagy lezárt.
    bool try_push(T value) {
        {
            std::lock_guard lk(m_);
            if (closed_ || q_.size() >= cap_) return false;
            q_.push_back(std::move(value));
        }
        not_empty_.notify_one();
        return true;
    }

    // Blokkol, amíg hely van; stop kérés vagy close esetén false.
    bool push(T value, std::stop_token st) {
        std::unique_lock lk(m_);
        not_full_.wait(lk, st, [&] { return closed_ || q_.size() < cap_; });
        if (closed_ || st.stop_requested()) return false;
        q_.push_back(std::move(value));
        lk.unlock();
        not_empty_.notify_one();
        return true;
    }

    // Stop után is kiveszi a maradékot; csak üres+leállt esetén nullopt.
    std::optional<T> pop(std::stop_token st) {
        std::unique_lock lk(m_);
        not_empty_.wait(lk, st, [&] { return closed_ || !q_.empty(); });
        if (q_.empty()) return std::nullopt;
        T v = std::move(q_.front());
        q_.pop_front();
        const bool now_empty = q_.empty();
        lk.unlock();
        not_full_.notify_one();
        if (now_empty) empty_.notify_all();
        return v;
    }

    // A sor kiürüléséig vagy a határidőig vár; true, ha kiürült.
    bool wait_empty_until(std::chrono::steady_clock::time_point deadline) {
        std::unique_lock lk(m_);
        return empty_.wait_until(lk, deadline, [&] { return q_.empty(); });
    }

    // A sorban maradt elemek eldobása; az eldobott darabszámot adja.
    std::size_t discard() {
        std::size_t n = 0;
        {
            std::lock_guard lk(m_);
            n = q_.size();
            q_.clear();
        }
        empty_.notify_all();
        not_full_.notify_all();
        return n;
    }

    void close() {
        {
            std::lock_guard lk(m_);
            closed_ = true;
        }
        not_empty_.notify_all();
        not_full_.notify_all();
    }

    std::size_t size() const {
        std::lock_guard lk(m_);
        return q_.size();
    }

private:
    std::size_t cap_;
    mutable std::mutex m_;
    std::condition_variable_any not_empty_, not_full_, empty_;
    std::deque<T> q_;
    bool closed_ = false;
};

A queue működésének kritikus pontja a pop feltétel nélküli if (q_.empty()) ellenőrzése: a wait stop token miatt is visszatérhet, amikor a predikátum hamis, ezért a visszatérés után a tényleges állapotot újra meg kell vizsgálni. Ha ezt elhagynánk, stop kéréskor üres sorból is kivennénk egy „elemet”, ami nem definiált viselkedés. Hasonlóan, a push-ban a stop_requested() ellenőrzés azért kell, mert a wait akkor is visszatérhet, ha éppen van hely — ilyenkor viszont leállás alatt nem akarunk új munkát befogadni. A wait_empty_until és a discard a határidős leállítás eszköze: az első a sor kiürülésére vár legfeljebb egy adott időpontig, a második a sorban maradt elemeket egy zár alatt eldobja, és megmondja, hányat dobott el.

A queue csak a tárolást és a szinkronizációt adja; a szolgáltatás életciklusa a következő headerben van. A konfiguráció minden korlátja kötelezően pozitív: nulla worker mellett a sor sosem ürülne, nulla kapacitásnál a blokkoló submit sosem jutna helyhez. A konstruktor ezért a szálak indítása előtt ellenőriz, és hibás konfigurációnál std::invalid_argument kivételt dob; a hiba így indításkor derül ki, nem egy beragadt leállításkor.

// service.hpp
#pragma once
#include "bounded_queue.hpp"
#include <atomic>
#include <chrono>
#include <cstddef>
#include <functional>
#include <mutex>
#include <stdexcept>
#include <string>
#include <thread>
#include <utility>
#include <vector>

struct frame {
    std::string payload;
};

struct service_config {
    std::size_t workers = 2;          // W >= 1
    std::size_t capacity = 4;         // C >= 1
    std::size_t max_frame = 4096;     // S_max >= 1, bájt
    std::size_t max_waiting = 8;      // P >= 1: egyszerre blokkoló submitok
    std::chrono::milliseconds drain_timeout{1000};
};

enum class submit_result { accepted, too_large, busy, closed };

struct service_stats {
    std::size_t accepted;             // sorba került frame-ek
    std::size_t processed;            // végigfutott feldolgozás
    std::size_t cancelled;            // stop miatt félbehagyott feldolgozás
    std::size_t dropped;              // a drain-határidő után eldobva
};

// true: kész; false: a stop kérés miatt félbehagyta. A stop kérés után
// korlátos időn belül vissza kell térnie, és nem végezhet blokkoló I/O-t.
using processor = std::function<bool(const frame&, std::stop_token)>;

class service {
public:
    service(service_config cfg, processor proc)
        : cfg_(validate(cfg)), proc_(std::move(proc)), q_(cfg_.capacity) {
        if (!proc_) throw std::invalid_argument("service: empty processor");
        workers_.reserve(cfg_.workers);
        for (std::size_t i = 0; i < cfg_.workers; ++i) {
            workers_.emplace_back([this](std::stop_token st) { run(st); });
        }
    }

    ~service() { shutdown(); }

    service(const service&) = delete;
    service& operator=(const service&) = delete;

    // Backpressure: legfeljebb P hívó blokkolhat; a többi azonnal busy.
    submit_result submit(frame f, std::stop_token st = {}) {
        if (f.payload.size() > cfg_.max_frame)
            return submit_result::too_large;      // a sor előtt elutasítva
        if (waiting_.fetch_add(1) >= cfg_.max_waiting) {
            waiting_.fetch_sub(1);
            return submit_result::busy;
        }
        accepted_.fetch_add(1);
        const bool ok = q_.push(std::move(f), st);
        if (!ok) accepted_.fetch_sub(1);
        waiting_.fetch_sub(1);
        return ok ? submit_result::accepted : submit_result::closed;
    }

    // Határidős leállítás; többször is hívható. Felső korlátja
    // drain_timeout + egy feldolgozás stop utáni visszatérési ideje.
    service_stats shutdown() {
        std::call_once(once_, [this] {
            q_.close();                                    // 1. új munka tiltása
            const auto deadline =
                std::chrono::steady_clock::now() + cfg_.drain_timeout;
            if (!q_.wait_empty_until(deadline))            // 2. drain határidőig
                dropped_ = q_.discard();                   // 3. maradék eldobása
            for (auto& w : workers_) w.request_stop();     // 4. kooperatív stop
            workers_.clear();                              // 5. join
        });
        return {accepted_.load(), processed_.load(), cancelled_.load(),
                dropped_};
    }

    std::size_t pending() const { return q_.size(); }

private:
    static service_config validate(service_config cfg) {
        if (cfg.workers == 0 || cfg.capacity == 0 || cfg.max_frame == 0 ||
            cfg.max_waiting == 0 || cfg.drain_timeout.count() < 0)
            throw std::invalid_argument("service: invalid configuration");
        return cfg;
    }

    void run(std::stop_token st) {
        while (auto f = q_.pop(st)) {
            if (proc_(*f, st))
                processed_.fetch_add(1);
            else
                cancelled_.fetch_add(1);
        }
    }

    const service_config cfg_;
    const processor proc_;
    bounded_queue<frame> q_;
    std::atomic<std::size_t> waiting_{0};
    std::atomic<std::size_t> accepted_{0};
    std::atomic<std::size_t> processed_{0};
    std::atomic<std::size_t> cancelled_{0};
    std::size_t dropped_ = 0;          // csak a call_once írja
    std::once_flag once_;
    std::vector<std::jthread> workers_; // utolsó tag: elsőként szűnik meg
};

A feldolgozó (processor) szerződése a leállítási korlát része: a stop kérés után korlátos időn belül vissza kell térnie, és blokkoló I/O-t nem végezhet. Egy megtelt, nem olvasott pipe-ra író printf például korlátlan ideig blokkolhat, és ezt a stop token nem szakítja meg; ezért a példa a feldolgozás eredményét memóriában gyűjti, és csak a leállítás után, a szolgáltatáson kívül ír ki. A feldolgozó false visszatérése azt jelzi, hogy a munkát a stop kérés miatt félbehagyta; ezt a cancelled számláló rögzíti.

// main.cpp
#include "service.hpp"
#include <atomic>
#include <cstdio>
#include <string>

int main() {
    std::atomic<std::size_t> bytes{0};
    service svc{service_config{}, [&bytes](const frame& f, std::stop_token) {
        bytes.fetch_add(f.payload.size());   // nem blokkol: korlátos idő
        return true;
    }};
    for (int i = 0; i < 8; ++i) {
        (void)svc.submit(frame{"job-" + std::to_string(i)});
    }
    const service_stats s = svc.shutdown();
    // A kiírás a leállítás után, a szolgáltatáson kívül történik.
    std::printf("accepted=%zu processed=%zu cancelled=%zu dropped=%zu "
                "bytes=%zu\n", s.accepted, s.processed, s.cancelled,
                s.dropped, bytes.load());
    return s.accepted == s.processed + s.cancelled + s.dropped ? 0 : 1;
}

A shutdown() eredménye a munka teljes elszámolása: a leállítás után, és miután minden submit hívás visszatért,

$$\text{accepted} = \text{processed} + \text{cancelled} + \text{dropped}.$$

Fordítás és futtatás:

g++ -std=c++20 -O2 -Wall -Wextra -Wpedantic -Wconversion -Wshadow \
    -Werror -pthread -o svc main.cpp
./svc
g++ -std=c++20 -O1 -g -pthread -fsanitize=thread -o svc-tsan main.cpp
setarch -R ./svc-tsan

A kimenet: accepted=8 processed=8 cancelled=0 dropped=0 bytes=40; a feldolgozás sorrendje futásonként változhat, a számlálók nem. A setarch -R a címtér-véletlenítést kapcsolja ki az adott folyamatra: újabb Linux-kerneleken az alapértelmezett ASLR-entrópia mellett a ThreadSanitizer futásidejű könyvtára unexpected memory mapping hibával leáll, mielőtt a program elindulna.

A konfigurációt, a memóriakorlátot és a határidős leállítást külön teszt ellenőrzi. A memóriamérés a globális operator new felülírásával a legalább $S_{\max}$ méretű élő allokációkat számolja; egy blokkolt workerrel, $C=4$, $W=1$, $P=2$ beállítással és 16 egyidejű producerrel az élő nagy allokációk száma pontosan $C+W+P=7$, a többi 14 producer busy eredményt kap. A lassú feldolgozós eset azt ellenőrzi, hogy a leállítás a drain-határidő és egy feldolgozás ideje után véget ér, a maradékot pedig eldobottként számolja el:

// tests/service_test.cpp -- konfiguráció, memóriakorlát, határidős leállás
#include "service.hpp"

#include <atomic>
#include <chrono>
#include <condition_variable>
#include <cstdio>
#include <cstdlib>
#include <mutex>
#include <new>
#include <stdexcept>
#include <thread>
#include <vector>

#define CHECK(condition) do {                                      \
    if (!(condition)) {                                            \
        std::fprintf(stderr, "CHECK failed at line %d\n", __LINE__); \
        std::exit(EXIT_FAILURE);                                   \
    }                                                              \
} while (0)

using namespace std::chrono_literals;
using clock_type = std::chrono::steady_clock;

// --- A nagy (>= S_max) allokációk élő darabszámának mérése -------------
constexpr std::size_t s_max = 4096;
static std::atomic<long> big_live{0};

void *operator new(std::size_t size)
{
    auto *base = static_cast<std::max_align_t *>(
        std::malloc(size + sizeof(std::max_align_t)));
    if (base == nullptr) throw std::bad_alloc{};
    *reinterpret_cast<std::size_t *>(base) = size;
    if (size >= s_max) big_live.fetch_add(1);
    return base + 1;
}

void operator delete(void *p) noexcept
{
    if (p == nullptr) return;
    auto *base = static_cast<std::max_align_t *>(p) - 1;
    if (*reinterpret_cast<std::size_t *>(base) >= s_max)
        big_live.fetch_sub(1);
    std::free(base);
}

void operator delete(void *p, std::size_t) noexcept { operator delete(p); }

// --- Feldolgozók ---------------------------------------------------------
static bool instant(const frame &, std::stop_token) { return true; }

static bool slow(const frame &, std::stop_token)   // nem figyeli a stopot,
{                                                   // de korlátos idejű
    std::this_thread::sleep_for(20ms);
    return true;
}

static bool blocked(const frame &, std::stop_token st)  // csak stopra áll le
{
    std::mutex m;
    std::condition_variable_any cv;
    std::unique_lock lk(m);
    cv.wait(lk, st, [] { return false; });
    return false;                                   // félbehagyva
}

static bool rejects(service_config cfg)
{
    try {
        service svc{cfg, instant};
    } catch (const std::invalid_argument &) {
        return true;
    }
    return false;
}

static void wait_until(const std::function<bool()> &done)
{
    const auto limit = clock_type::now() + 5s;
    while (!done()) {
        CHECK(clock_type::now() < limit);
        std::this_thread::sleep_for(1ms);
    }
}

int main()
{
    // H077: a hibás konfiguráció indításkor kiderül.
    CHECK(rejects({.workers = 0}));
    CHECK(rejects({.capacity = 0}));
    CHECK(rejects({.max_frame = 0}));
    CHECK(rejects({.max_waiting = 0}));
    {
        bool threw = false;
        try { service svc{service_config{}, processor{}}; }
        catch (const std::invalid_argument &) { threw = true; }
        CHECK(threw);
    }

    // workers = 1, capacity = 1: nem ragad be.
    {
        service svc{{.workers = 1, .capacity = 1}, instant};
        for (int i = 0; i < 100; ++i)
            CHECK(svc.submit(frame{"x"}) == submit_result::accepted);
        const service_stats s = svc.shutdown();
        CHECK(s.accepted == 100 && s.processed == 100 && s.dropped == 0);
    }

    // H078: a túl nagy frame a sor előtt elutasul.
    {
        service svc{{.max_frame = s_max}, instant};
        CHECK(svc.submit(frame{std::string(s_max + 1, 'x')}) ==
              submit_result::too_large);
        CHECK(svc.shutdown().accepted == 0);
    }

    // H078: saját tároló = C + W + P frame, 16 producer mellett is.
    {
        const service_config cfg{.workers = 1, .capacity = 4,
                                 .max_frame = s_max, .max_waiting = 2,
                                 .drain_timeout = 100ms};
        service svc{cfg, blocked};
        CHECK(svc.submit(frame{std::string(s_max, 'x')}) ==
              submit_result::accepted);
        wait_until([&] { return svc.pending() == 0; });  // a worker elvitte
        for (int i = 0; i < 4; ++i)
            CHECK(svc.submit(frame{std::string(s_max, 'x')}) ==
                  submit_result::accepted);

        std::atomic<int> busy{0}, closed{0};
        std::vector<std::jthread> producers;
        for (int i = 0; i < 16; ++i) {
            producers.emplace_back([&] {
                switch (svc.submit(frame{std::string(s_max, 'x')})) {
                case submit_result::busy:   busy.fetch_add(1); break;
                case submit_result::closed: closed.fetch_add(1); break;
                default: break;
                }
            });
        }
        wait_until([&] { return busy.load() == 14; });
        std::this_thread::sleep_for(50ms);
        CHECK(big_live.load() == 1 + 4 + 2);             // W + C + P

        const auto t0 = clock_type::now();
        const service_stats s = svc.shutdown();
        const auto took = clock_type::now() - t0;
        producers.clear();
        CHECK(closed.load() == 2);                       // a két várakozó
        CHECK(s.accepted == 5 && s.cancelled == 1 && s.dropped == 4);
        CHECK(took < 100ms + 200ms);                     // határidő + tűrés
        CHECK(big_live.load() == 0);
    }

    // H238: lassú feldolgozó, határidős drain és eldobás.
    {
        service svc{{.workers = 1, .capacity = 64,
                     .drain_timeout = 200ms}, slow};
        for (int i = 0; i < 50; ++i)
            CHECK(svc.submit(frame{"x"}) == submit_result::accepted);
        const auto t0 = clock_type::now();
        const service_stats s = svc.shutdown();
        const auto took = clock_type::now() - t0;
        CHECK(s.accepted == s.processed + s.cancelled + s.dropped);
        CHECK(s.dropped > 0 && s.processed > 0);
        CHECK(took < 200ms + 20ms + 200ms);              // + egy feldolgozás
    }
    std::puts("service_test: ok");
    return 0;
}

g++ -std=c++20 -O2 -Wall -Wextra -Wpedantic -Wshadow -Werror -pthread \
    -I. tests/service_test.cpp -o service_test
./service_test

Buildjegyzőkönyv (lab-gcc profil). A main.cpp és a tests/service_test.cpp G++ 13.3.0-val, libstdc++ mellett -Werror beállítással figyelmeztetés nélkül fordult Linux x86-64 környezetben. A teszt optimalizált builddel ötször egymás után, valamint ThreadSanitizer (setarch -R) és ASan/UBSan builddel is sikeres volt; a demó TSan alatt is a fenti kimenetet adta. A határidős ellenőrzések tűréssel dolgoznak (200 ms); terhelt CI-gépen ezt a tűrést a profilban kell rögzíteni.

39.3.3 Buffer- és callback-tulajdon a pipeline határain

A legfontosabb szabály: aszinkron határon csak owned érték mehet át. Ha a submit paramétere std::span<const std::byte> lenne, az csak egy pointert és egy hosszt tartalmazna; a hívó stack frame-je a submit visszatérése után megszűnhet, a worker viszont később olvasná. Ez a klasszikus dangling helyzet, amelyet a span, byte és bounds-szerződések (31.3) részben tárgyaltunk. A helyes megoldás, hogy a queue eleme tulajdonol: std::vector<std::byte>, std::string vagy egy move-only frame típus. A span csak szinkron hívásban, a hívó élettartamán belül használható.

A shared_ptr csábító, de csak akkor indokolt, ha valóban több, egymástól független owner van: például a connection objektum és a feldolgozó is életben kell tartsa ugyanazt a puffert, mert a feldolgozás aszinkron módon, a connection lezárása után is olvas. Ha a queue egymaga tulajdonol, a unique_ptr vagy a közvetlen érték a helyes választás: kevesebb atomikus művelet, egyértelműbb élettartam.

A connection state és a callback lifetime összefonódik. Egy tipikus hiba:

// HIBÁS tanítási példa: a callback túlélheti a service-t.
svc.set_on_frame([&svc](frame f) { svc.submit(std::move(f)); });

Ha a callbacket egy külső komponens tárolja, és a service előbb destruálódik, a callback egy halott objektumra hivatkozik. A javítás vagy az, hogy a callback a service-en belül marad (és a service destruktorában deregisztráljuk), vagy az, hogy a callback std::weak_ptr-rel ellenőrzi az owner életét:

std::weak_ptr<service> weak = svc;
auto cb = [weak](frame f) {
    if (auto s = weak.lock()) s->submit(std::move(f));
};

A weak_ptr::lock() itt nem stíluselem, hanem a verseny kezelése: a lock és a service destruktora közül az egyik „nyer”, és nem lehet olyan interleaving, amelyben a callback egy félig destruált objektumot használ. A callback regisztrációjának és visszavonásának szerződése ezért tartalmazza, hogy a visszavonás szinkronizál a már futó callback-hívásokkal: vagy a callback rövid és nem blokkol, vagy a deregisztráció megvárja a futó hívásokat. Ha ezt nem tesszük meg, a „deregisztrált” callback még futhat, miközben a hivatkozott erőforrás felszabadul.

Az invalidálódás térképe:

39.3.4 Leállítási protokoll, korlátok és tesztelés

A leállítás sorrendje nem önkényes. Tegyük fel, hogy először joinolnánk, és csak utána zárnánk le a queue-t. Ha a queue megtelt, egy termelő éppen a push-ban blokkol, és soha nem tud visszatérni, mert a fogyasztók már nem futnak: deadlock. Ezért a helyes sorrend:

1. Bemenet lezárása (close()): a termelők false-szal visszatérnek, nem blokkolhatnak örökre, és a „még bekerülhet-e munka” kérdés egyetlen lock alatt eldől.
2. Drain határidőig (wait_empty_until): a workerek a sorban maradt elemeket feldolgozzák, legfeljebb a drain_timeout lejártáig.
3. Maradék eldobása (discard()): ha a határidőig nem ürült ki a sor, a maradékot eldobjuk, és a darabszámot dropped néven elszámoljuk.
4. Stop kérés (request_stop()): a workerek blokkoló várakozása megszakad, a futó feldolgozás a stop tokent látva korlátos időn belül visszatér.
5. Join (workers_.clear()): a jthread destruktorai joinolnak, így minden worker befejeződése happens-before kapcsolatba kerül a shutdown hívóval.
6. Erőforrás-felszabadítás: csak most szabad a queue-t és a hozzá tartozó puffereket eldobni.

A join deadlockmentességének indoklása külön figyelmet érdemel. A join önmagában nem holtpontveszélyes, ha a joinoló szál nem tart olyan zárat, amelyet a worker megpróbál felvenni. A mi shutdown()-ünk a q_.close() után semmilyen alkalmazás-szintű zárat nem tart, amikor a clear() joinol. A második veszélyforrás a std::stop_callback: a callback a request_stop() hívó szálán fut, miközben a stop állapot belső zárja foglalt; ha a callback olyan mutexet próbál felvenni, amelyet a request_stop() hívója tart, holtpont lesz. Ezért a stop callback rövid, nem allokáló és nem zároló kód legyen; a mi példánkban nincs is callback, mert a condition_variable_any stop tokenes wait-je maga kezeli az ébredést.

A memóriakorlát a fentebb levezetett $(C+W+P)\cdot S_{\max}+K$. Az időkorlát a leállítás felső korlátja: a shutdown() legfeljebb $D + T_{\text{stop}}$ ideig tart, ahol $D$ a drain_timeout, $T_{\text{stop}}$ pedig az az idő, amely alatt egy futó feldolgozás a stop kérés után visszatér, plusz a join költsége. A korlát tehát a feldolgozó szerződésén múlik: ha egy feldolgozás nem kooperatív módon blokkol (például megtelt pipe-ra ír), a leállítás sem korlátos, és ezt sem a request_stop(), sem a szál leválasztása (detach) nem javítja meg, utóbbi csak elrejti a még élő erőforrásokat. A határidő után eldobott munka adatvesztés; a dropped és a cancelled számláló ezt a hívó számára láthatóvá teszi, és a protokollnak kell eldöntenie, hogy elfogadható-e. A két korlát együtt adja a szolgáltatás erőforrás-szerződését.

A TSan (ThreadSanitizer) a tényleges adatversenyeket találja meg, például ha a closed_ flaget valahol lock nélkül olvassuk, vagy ha a callback és a destruktor között nincs happens-before. Nem találja meg a logikai race-eket, például azt, hogy a „drain on stop” politikánk mellett egy elem kétszer kerül feldolgozásra, mert a queue-ból való kivétel és a feldolgozás között a worker megszakad. Ezért a TSan mellett integritásteszt kell: $N$ termelő, $M$ fogyasztó, ismételt shutdown() hívás, és annak ellenőrzése, hogy a feldolgozott elemek multihalmaza pontosan az elküldött elemek multihalmaza, elveszett és duplikált elem nélkül.

Egy TSan-futtatás tipikus parancssora a laborban:

TSAN_OPTIONS="halt_on_error=1 second_deadlock_stack=1" setarch -R ./svc-tsan

A second_deadlock_stack opció a holtpont második szálának stackjét is kiírja, ami a stop callback és a join közötti ciklusok azonosításában segít. Ennek a beállításnak a támogatottsága a használt Clang/GCC runtime verziójától függ; a laborprofilban rögzíteni kell.

39.3.4.1 Opcionális coroutine adapter

A coroutine adapter nem része az alap buildnek, hanem külön capability profile: a CMake target csak akkor épül, ha a fordító és a könyvtár támogatja a szükséges C++20 coroutine elemeket, és a projekt ezt feature-test makróval ellenőrzi. A cél egy vékony illesztő, amely a queue-t co_await-elhető várakozássá alakítja: a co_await q.async_pop() felfüggeszti a coroutine-t, és a queue értesítésekor folytatja. Az élettartam-szerződés ugyanaz, mint eddig: a coroutine frame az owner, a queue borrowed, és a coroutine nem folytatódhat a szolgáltatás leállása után. A részletes promise_type/awaiter mechanizmust az Aszinkron I/O adapter és coroutine-labor (35.5) tárgyalta; itt csak a leállítási integrációt rögzítjük: a stop kérésnek a függő coroutine-okat is fel kell ébresztenie, különben a shutdown() joinja örökre vár. Ezért az adapter vagy a queue close()-ára, vagy egy külön stop tokenre regisztrál, és mindkét esetben a folytatás előtt újra ellenőrzi a lezárt állapotot. Az adapter kísérleti státuszát a buildben és a dokumentációban is jelölni kell; nem szabad az alap szolgáltatás helyességét a coroutine támogatására építeni.

39.3.5 Feladatok

1. feladat (alap). Egészítsd ki a bounded_queue-t egy drain() metódussal, amely close() után az összes maradék elemet kiveszi egy std::vector<T>-be, és a queue-t üresen hagyja. Fogalmazd meg, milyen állapotban hívható, és mi történik, ha drain() közben egy termelő még push-olni próbál.

2. feladat (közepes). Vezess be időkorlátos push-t: push_for(T value, std::chrono::milliseconds limit, std::stop_token st), amely wait_until-tal várakozik, és ha a határidő lejár, eldobja az elemet, és növeli egy std::atomic<std::size_t> számlálót. Indokold, miért kell a számlálónak atomikusnak lennie, és miért nem elég a queue belső mutexe.

3. feladat (nehéz). Tervezd meg a callback-lifetime kezelést std::weak_ptr-rel egy olyan szolgáltatásban, ahol a callbacket egy külső, a service-nél tovább élő komponens tárolja. Írj egy TSan alatt futtatható integritástesztet, amely a shutdown() alatt is hívja a callbacket, és igazold, hogy nincs use-after-free.

4. feladat (opcionális). Vázold fel a coroutine adapter async_pop műveletét, és sorold fel, milyen három feltételt kell ellenőriznie a folytatás előtt a leállási verseny elkerüléséhez.

39.3.6 Megoldások

1. megoldás. A drain() csak closed_ == true esetén értelmes; ezt a metódus elején ellenőrizzük, és ha nem teljesül, std::logic_error-t dobunk (vagy a szerződést dokumentáljuk és assert-tel jelezzük). A művelet a queue mutexe alatt kiveszi az összes elemet, majd a not_full_ feltételváltozót értesíti, hogy az esetleges blokkoló termelők felébredjenek. Mivel a close() miatt a push úgyis false-szal tér vissza, a verseny eldőlt: a close() és a drain() között nincs olyan interleaving, amelyben új elem kerülne be, feltéve, hogy a closed_ beállítása és a drain() ugyanazon mutex alatt történik. Ellenőrzési feltétel: drain() után size() == 0, és a visszaadott vektor elemszáma megegyezik a close() előtti size() értékkel, ha közben nem volt pop.

2. megoldás. A számlálót azért kell atomikusan növelni, mert több termelő szál is növelheti egyszerre, és a queue belső mutexe nem védi: a számláló a queue-on kívüli, diagnosztikai célú állapot. Ha a mutex alatt növelnénk, a push_for minden eldobása zárat venne fel, ami a backpressure alatti contentiont növelné; az atomikus növelés viszont lock-free. Az ellenőrzési feltétel: a sikeres és eldobott hívások száma összesen a hívások számával egyezik, és a számláló értéke TSan alatt sem mutat adatversenyt. Implementációs váz:

bool push_for(T value, std::chrono::milliseconds limit, std::stop_token st) {
    std::unique_lock lk(m_);
    if (!not_full_.wait_for(lk, st, limit,
                            [&] { return closed_ || q_.size() < cap_; })) {
        dropped_.fetch_add(1, std::memory_order_relaxed);
        return false;
    }
    if (closed_ || st.stop_requested()) return false;
    q_.push_back(std::move(value));
    lk.unlock();
    not_empty_.notify_one();
    return true;
}

A wait_for visszatérési értéke pontosan azt jelzi, hogy a predikátum teljesült-e; ha false, akkor a határidő vagy a stop miatt tértünk vissza, és ezt a két esetet a st.stop_requested() ellenőrzés választja szét, ha erre szükség van a diagnosztikában.

3. megoldás. A callbacket a std::weak_ptr<service>-re zárjuk, és a callback törzse minden hívásnál lock()-ol. A shutdown() a close() után request_stop()-ot hív, majd joinol; a callback a submit-ban a queue closed_ flagjén bukik el, tehát a shutdown után érkező hívás false-szal tér vissza, nem nyúl a felszabadított erőforráshoz. A TSan-teszt: egy szál ismételten hívja a callbacket, a fő szál shutdown()-t hív, és a teszt a végén ellenőrzi, hogy nincs TSan-hiba, és hogy a callback hívásainak száma megegyezik a submit visszatérési értékeivel konzisztens módon. Fontos, hogy a callback ne tartson zárat a lock() és a submit között, amit a service destruktora is felvesz, különben a weak_ptr ellenére holtpont lehetséges. Ellenőrzési feltétel: a teszt TSan alatt hibamentes, és a callback sosem ér el egy service objektumot a destruktor megkezdése után (ezt a lock() sikertelensége garantálja).

4. megoldás. Az async_pop folytatása előtt három feltételt kell ellenőrizni: (a) a queue nem zárt-e le (closed_ == false), (b) a stop token nem kérte-e a leállást, és (c) a coroutine handle még érvényes-e (nem destruálódott a frame). Ha bármelyik feltétel nem teljesül, a folytatás helyett a coroutine-t megszakított állapotban kell feloldani, különben a shutdown() joinja olyan coroutine-ra várna, amely soha nem folytatódik. A három ellenőrzésnek ugyanabban a szinkronizációs tartományban kell történnie, mint a queue állapotváltozásainak, különben a „lezárt, de mégis folytatódik” verseny marad.

39.3.7 Összefoglaló

A szolgáltatás-életciklus szerződés: állapotok, átmenetek és erőforrás-kötelezettségek együtt. A bounded_queue a memóriakorlát és a backpressure hordozója; a stop_token a kooperatív megszakításé; a jthread a join garantálásáé. Aszinkron határon csak owned érték mehet át, a span a szinkron hívások eszköze. A callback élettartamát weak_ptr-rel vagy belső regisztrációval kell védeni, és a deregisztrációnak szinkronizálnia kell a futó hívásokkal. A leállítás sorrendje: close(), drain a határidőig, a maradék elszámolt eldobása, majd request_stop(), join, végül felszabadítás; a felső korlát a drain-határidő és a feldolgozó stop utáni visszatérési ideje; a join nem tarthat olyan zárat, amelyet a worker vagy a stop callback felvesz. A TSan az adatversenyt találja meg, az integritásteszt a logikai race-t; a kettő együtt adja a leállás helyességének bizonyítékát. A coroutine adapter külön capability profile, amelynek a leállítási integrációja ugyanazt a három ellenőrzést igényli, mint a szálas változaté.

39.3.8 Források és ellenőrzési korlátok

A fejezet állításai a C++20 nyelvi és könyvtári modellre, valamint a jthread, stop_token, condition_variable_any C++20-as eszközeire épülnek. A megadott kiinduló pontok — a WG21 oldala, az N4950 munkatervezet, a C++ Core Guidelines, a CMake dokumentáció, a Linux man-pages, valamint a Clang AddressSanitizer és ThreadSanitizer dokumentációja — kutatási kiindulópontok, nem pedig ebben a kéziratban ellenőrzött, idézett bizonyítékok. A konkrét szakaszokra, a condition_variable_any stop tokenes overloadjainak könyvtári támogatására, a TSAN_OPTIONS beállításokra és a fordítók viselkedésére vonatkozó állításokat a labornak a rögzített compiler/libstdc++/libc++ verzióval, feature-test makróval és compile probe-bal kell igazolnia; A fejezet kódját G++ 13.3.0 és libstdc++ mellett lefordítottuk és lefuttattuk (lásd a buildjegyzőkönyveket); más fordítón és standard könyvtáron ez a labor feladata, addig ott a kód várt működés, nem mért eredmény.