34.2 RAII lockok, condition variable és wait predicate

A Mutex, condition variable és producer–consumer (15.3) alfejezet C változatában a bounded queue minden várakozási pontja kézi pthread_mutex_unlock-kal zárult, és minden korai return előtt végig kellett gondolni, hogy a zár felszabadul-e. A szerződés helyes volt, de törékeny: egy új hibaleágazás vagy egy hosszú goto cleanup csendben holtpontot hozott. A C++ nem „okosabb” mutexet ad ugyanerre a problémára, hanem a zár birtoklását egy objektum élettartamához köti: a zár a konstruktorban megszerzi, a destruktorban elengedi, a kritikus szakasz határa pedig pontosan az az objektum, amelynek a hatóköre véget ér. Ez ugyanaz az RAII-minta, amit a RAII wrapper file descriptorhoz és lockhoz (21.2) alfejezetben file descriptorra és lockra már felállítottunk; itt a hangsúly azon van, hogy a zár típusa és a várakozás feltétele is a művelet szerződéséből következzen, ne szokásból.

A C++ konkurencia-eszközeit a C11 atomics és a POSIX szálkezelés szemantikájára építjük; a szál életciklusát és a kooperatív leállítást a thread, jthread és kooperatív cancellation (34.1) alfejezet rögzítette, a happens-before és a data race definícióját nem ismételjük meg. Az itteni cél: ugyanazt a bounded queue invariánst RAII lockokkal megvalósítani, és indokolni tudni, mikor melyik locktípus és milyen várakozási minta a helyes. A fejezethez ebben az egységben nem tartozik ellenőrzött forrásjegyzet, ezért a szabvány-paragrafusokra (a C++20 munkatervezet [thread.mutex], [thread.condition] fejezeteire) itt nem hivatkozom bizonyítékként: ezeket a saját C++20 példányodon ellenőrizd. A kódok a leírásban nem kerültek lefuttatásra; a viselkedésre vonatkozó állítások a szerződésből következnek, nem mérésből.

34.2.1 A zár mint erőforrás: melyik locktípus melyik szerződéshez

A std::mutex nem rekurzív, nem másolható és nem mozgatható; ugyanabban a szálban kétszer megszerezni nem definiált viselkedés, nem pedig holtpont, amit egy recursive_mutex „megjavítana”. A kritikus szakasz akkor helyes, ha a zár alatt olvasott és írt állapot pontosan azonos halmaz, és minden más szál, amely ezt az állapotot érinti, ugyanezt a zárat tartja.

A locktípus megválasztása három kérdésből következik. Ha csak egy zárat kell egy hatókörre megszerezni, és nem kell kézzel elengedni, akkor std::lock_guard a legkisebb és legkevésbé félreérthető választás. Ha a zárnak túl kell élnie egy hatókört, mozgatni kell, try_lock-ot vagy időzített próbát kell indítani, vagy a condition_variable::wait-nek át kell adni, akkor std::unique_lock kell — ez az egyetlen standard locktípus, amely unlock()/lock() párost kínál, és owns_lock()-nal megmondja, hogy éppen tart-e zárat. Ha egyszerre több mutexet kell megszerezni, std::scoped_lock (C++17) a helyes eszköz: variadikus, és a szabvány által előírt holtpont-elkerülő algoritmussal (próbálkozás, visszalépés, újrapróbálás) szerzi meg mindet, így a megszerzési sorrend nem a hívó felelőssége.

A condition_variable::wait első paramétere std::unique_lock<std::mutex>&, tehát lock_guard-dal nem hívható meg: a várakozásnak atomikusan el kell engednie, majd vissza kell szereznie a zárat. (std::condition_variable_any bármilyen BasicLockable lockkal működik, de a mutex + condition_variable páros a szokásos, olcsóbb eset.)

A kritikus szakasz minimalitása nem esztétika, hanem szerződés: zárat tartva ne hívj felhasználói callbacket, ne végezz bizonytalan költségű allokációt és ne végezz I/O-t, mert minden ilyen pont a zár tartási idejét más szálak számára kiszámíthatatlanná teszi.

34.2.1.1 Lock ordering és a holtpont elkerülése

A holtpont négy együttes feltétele (kölcsönös kizárás, tartás-és-várakozás, nincs erőszakos elvétel, körkörös várakozás) közül a gyakorlatban a körkörös várakozást lehet a legjobban megszüntetni: vagy globális sorrendet adsz a mutexeknek, vagy std::scoped_lock-kal egyszerre szerzed meg őket. A sorrend legyen dokumentált és a kódban kikényszeríthető, ne a hívási véletlenen múljék.

#include <mutex>

struct Account {
    std::mutex m;          // csak illusztráció: éles kódban privát, lock-ot adó tagfüggvénnyel
    long long balance = 0;
};

void transfer(Account& from, Account& to, long long amount)
{
    std::scoped_lock lock(from.m, to.m);   // holtpont-elkerülő megszerzés
    from.balance -= amount;
    to.balance += amount;
}

Ha a két zárat nem lehet egyszerre megszerezni (mert a kettő között olyan művelet van, ami nem fér a kritikus szakaszba), akkor a sorrend rögzítése a helyes út. A std::lock és a std::defer_lock/std::adopt_lock kombinációja ugyanezt teszi kézi unique_lock-okkal. Kétely esetén a legjobb javítás nem egy harmadik mutex, hanem a struktúra átalakítása: egy zár alá vont állapot, egyetlen kritikus szakasz.

34.2.2 A bounded queue újraírása RAII lockokkal

Az invariáns változatlan a C változathoz képest: $0 \le \texttt{q\_.size()} \le \texttt{capacity\_}$, a head/tail mutatók helyett itt a std::deque hossza hordozza ugyanezt, és a closed_ jelző azt jelenti, hogy a további push már nem sikerülhet, a pop pedig a maradék elemek kiürülése után false/closed eredményt ad. A mutex egyszerre védi a q_-t és a closed_-t — ez a „védett predicate” alapja.

// queue_raii.cpp -- C++20, -pthread
#include <chrono>
#include <condition_variable>
#include <cstddef>
#include <deque>
#include <iostream>
#include <mutex>
#include <optional>
#include <stdexcept>
#include <thread>
#include <utility>
#include <vector>

enum class PopStatus { value, timeout, closed };

template <class T>
class BoundedQueue {
public:
    explicit BoundedQueue(std::size_t capacity) : capacity_(capacity)
    {
        if (capacity == 0)
            throw std::invalid_argument("BoundedQueue: capacity == 0");
    }

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

    // false: a queue zárt, az elem nem került be
    bool push(T value)
    {
        {
            std::unique_lock lock(m_);
            not_full_.wait(lock, [this] { return closed_ || q_.size() < capacity_; });
            if (closed_)
                return false;
            q_.push_back(std::move(value));
        }
        not_empty_.notify_one();     // a záron kívül: a felébredő ne ütközzön a zárral
        return true;
    }

    std::optional<T> pop()
    {
        std::unique_lock lock(m_);
        not_empty_.wait(lock, [this] { return closed_ || !q_.empty(); });
        if (q_.empty())
            return std::nullopt;     // closed_ és kiürült
        T value = std::move(q_.front());
        q_.pop_front();
        lock.unlock();
        not_full_.notify_one();
        return value;
    }

    PopStatus pop_for(std::chrono::milliseconds timeout, T& out)
    {
        std::unique_lock lock(m_);
        const bool ready = not_empty_.wait_for(lock, timeout,
            [this] { return closed_ || !q_.empty(); });
        if (!ready)
            return PopStatus::timeout;
        if (q_.empty())
            return PopStatus::closed;
        out = std::move(q_.front());
        q_.pop_front();
        lock.unlock();
        not_full_.notify_one();
        return PopStatus::value;
    }

    void close()
    {
        {
            std::lock_guard lock(m_);
            closed_ = true;          // állapotváltozás a zár alatt
        }
        not_empty_.notify_all();     // minden várakozó lássa a leállítást
        not_full_.notify_all();
    }

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

private:
    mutable std::mutex m_;
    std::condition_variable not_empty_;
    std::condition_variable not_full_;
    std::deque<T> q_;
    std::size_t capacity_;
    bool closed_ = false;
};

int main()
{
    BoundedQueue<int> q(2);
    std::vector<std::thread> producers;
    for (int p = 0; p < 2; ++p) {
        producers.emplace_back([&q, p] {
            for (int i = 0; i < 3; ++i)
                q.push(p * 100 + i);
        });
    }
    std::size_t received = 0;
    while (received < 6) {
        if (auto v = q.pop())
            ++received;
    }
    for (auto& t : producers)
        t.join();
    std::cout << "received=" << received << '\n';
}

A buildparancs a laborprofilhoz kötött, konkrét compilerrel:

g++     -std=c++20 -O2 -Wall -Wextra -Wpedantic -pthread queue_raii.cpp -o queue_raii
clang++ -std=c++20 -O2 -Wall -Wextra -Wpedantic -pthread queue_raii.cpp -o queue_raii

A -pthread nem elhagyható kozmetika: a fordító-illesztőnek jelezni kell, hogy a szálkezelő definíciók és a _REENTRANT-szerű makrók a helyes konfigurációban legyenek. A main szándékosan nem hív close()-t, mert a példa fix elemszámot drainel; a close() a leállítási út, amelyet a feladatok gyakorolnak. A size() const tagfüggvény, ezért a mutex mutable — ez a szokásos, de tudatos kompromisszum: a logikai állapotot nem változtatja, a fizikai zárállapotot igen.

34.2.3 Predicate loop, spurious wakeup és a notify szerződése

A cv.wait(lock, pred) jelentése pontosan: while (!pred()) cv.wait(lock);. A ciklus nem védekező kód, hanem a szerződés része, három okból. Egyrészt a szabvány megengedi a látszólagos (spurious) ébredést, tehát a wait visszatérhet úgy, hogy senki nem notifikált. Másrészt a felébredés és a zár visszaszerzése között egy harmadik szál megelőzhet: elviheti az elemet, betöltheti a helyet, lezárhatja a queue-t. Harmadrészt a két külön condition variable (not_empty_, not_full_) közös mutexen osztozik, így a notify_one címzettje nem garantáltan az a szál, amelyiknek az állapotra szüksége van. A predikátumot mindig a zár alatt kell kiértékelni, mert az a m_ által védett állapotot olvassa.

A notify és az állapotváltozás sorrendje fordítottja a naiv intuíciónak: először állapotot módosítasz a zár alatt, aztán notifikálsz. Ha a notify megelőzi az állapotváltozást, a felébredő szál hamis predikátumot lát, visszaalszik, és a későbbi állapotváltozásról már senki nem szól — ez az elveszett ébresztés (lost wakeup). A notify zár alatt és zár után is helyes; a zár utáni változat azért jobb, mert a felébresztett szál nem ütközik azonnal a még tartott zárral („hurry up and wait”). A notify_one akkor elég, ha minden notify pontosan egy kiszolgálható elemet vagy egy equivalent waitert jelent; leállításnál és több várakozót kiszolgáló állapotváltozásnál notify_all kell.

A predikátumos wait_for visszatérési értéke a predikátum értéke, nem a timeout ténye: false azt jelenti, hogy a predikátum a lejáratig hamis maradt, true pedig azt, hogy igaz — de ez utóbbi nem mondja meg, hogy közben lejárt-e az idő. Predikátum nélküli változatnál cv_status::no_timeout esetén is újra kell ellenőrizni a feltételt. Az időzítés clock domainje számít: a wait_for a std::chrono::steady_clock szerint mér, a wait_until viszont a megadott Clockot használja, ezért system_clock-kal átadott abszolút határidőt egy NTP-ugrás eltorzíthat. Timeout esetén soha ne következtess arra, hogy a queue üres — csak arra, hogy a predikátum addig hamis volt. A pop_for fenti változata ezért ad külön PopStatus::timeout és PopStatus::closed értéket: az std::optional-lal való „nullopt = timeout vagy closed” összevonás csendben elveszíti a leállítási információt.

34.2.4 shared_mutex: mikor éri meg, és mikor nem

A std::shared_mutex (C++17) olvasó–író zárat ad: std::shared_lock több olvasónak egyszerre, std::lock_guard/std::unique_lock egy írónak. A workload-feltétel szigorú: hosszú, tiszta olvasási kritikus szakaszok, magas olvasási arány és ritka írás. Rövid kritikus szakaszoknál a belső könyvelés és a közös cache-sor írása miatt könnyen lassabb a sima std::mutex-nál, és az írók kiéheztetésének politikája nem specifikált, tehát terhelésfüggő. Nincs olvasó→író upgrade: a shared_lock elengedése és új unique_lock felvétele között elvész az atomicitás, ezért ilyen esetben a kritikus szakaszt kell kettéválasztani vagy más mintát (pillanatkép, copy-on-write) választani. Egyszerű számlálóhoz továbbra is std::atomic kell (C++ atomics, atomic_ref és publication (34.3)); a shared_mutex nem helyettesíti.

34.2.5 Gyakori hibák

A wait if-fel körbevéve — spurious wakeup vagy konkurens fogyasztó esetén hamis predikátummal lép tovább. A lock_guard átadása wait-nek — nem fordul le, mert a wait unique_lock-ot vár. Notify állapotváltozás előtt — elveszett ébresztés. Két mutex megszerzése ad hoc sorrendben — holtpont. recursive_mutex bevezetése a tünet kezelésére. shared_mutex apró kritikus szakaszra, mérés nélkül. A wait_for false eredményének „üres queue”-ként való kezelése. A close() után elmaradó notify_all — a várakozók örökre a zár alatt maradnak.

34.2.6 Feladatok

1. Átalakítás és indoklás. A Mutex, condition variable és producer–consumer (15.3) C producer–consumer változatát alakítsd át a fenti osztályra. Sorold fel az összes zárfelvételi pontot és korai visszatérési utat, majd indokold pontonként, hogy lock_guard, unique_lock vagy scoped_lock a helyes választás, és hol változott meg a hibaágak kezelése.

2. Kódolás: leállítás és időzített várakozás. Egészítsd ki a main-t úgy, hogy egy koordinátor szál a termelők join-ja után close()-t hívjon, a fogyasztó pedig pop()-pal draineljen std::nullopt-ig. Majd írj három tesztesetet: (a) pop_for timeout üres, nyitott queue-n; (b) pop_for zárt, de még nem üres queue-n; (c) pop_for zárt és üres queue-n. Mindegyikhez add meg a várt PopStatus-t.

3. Diagnosztika. Az alábbi kódrészlet két hibát tartalmaz. Keresd meg őket, magyarázd el a kiváltott viselkedést (data race, holtpont vagy elveszett ébresztés), és javítsd a minimális változtatással.

// hibás tanítási példa -- nem használd
void consumer(BoundedQueue<int>& q)
{
    std::unique_lock lock(q.m_);          // feltételezzük, hogy friend
    if (!q.q_.empty()) {                  // 1. hiba helye
        q.q_.pop_front();
    }
}
void producer(BoundedQueue<int>& q, int v)
{
    q.not_empty_.notify_one();            // 2. hiba helye
    std::lock_guard lock(q.m_);
    q.q_.push_back(v);
}

34.2.7 Megoldások

1. megoldás. A C változatban a pthread_mutex_lock/unlock párok száma egyenlő a belépési pontokéval; RAII alatt minden belépési pont egy lockobjektum élettartama. A queue mindhárom művelete (push, pop, close) unique_lock-ot kap, mert a várakozás és a kézi unlock() a notify előtt mindkettőt igényli; a size() és a close() állapotfrissítése lock_guard-dal elég, mert ott nincs várakozás és nincs korai elengedés. A korai return-ök (closed_, üres queue) többé nem igényelnek cleanup ágat: a lock destruktora a visszatérési úton is lefut, ez a lényegi nyereség. Ellenőrzési feltétel: a push minden false visszatérése előtt a not_full_ várakozás megtörtént, és a closed_ beállítása után egyetlen push sem módosítja a q_-t.

2. megoldás. A koordinátor alakja:

std::thread closer([&q, &producers] {
    for (auto& t : producers) t.join();
    q.close();
});
std::size_t received = 0;
while (auto v = q.pop()) ++received;   // nullopt: zárt és kiürült
closer.join();

Tesztesetek: (a) nyitott, üres queue-n pop_for(50ms, out) → PopStatus::timeout, és a predikátum hamis maradt; (b) zárt, egy elemet tartalmazó queue-n → PopStatus::value, az elem kikerül; (c) zárt, üres queue-n → PopStatus::closed, és a hívás nem blokkol, mert a predikátum closed_ miatt igaz. Ellenőrzési feltétel: egyik esetben sem marad bent zár, és a not_full_ notify minden sikeres kivétel után megtörténik.

3. megoldás. Az első hiba a not_empty_ notify helye: a termelő a zár és az állapotváltozás előtt notifikál, így a fogyasztó felébredhet, hamis predikátumot lát, visszaalszik, és az ezt követő push_back után már senki nem szól neki — elveszett ébresztés, amely a fogyasztót véglegesen megakasztja. A javítás: a push_back a zár alatt történjen, a notify utána. A második hiba a consumer if-je: a várakozás helyett egyszeri ellenőrzés, amely (a) nem vár semmire, (b) nem kezeli a closed_ állapotot, és (c) ha wait-t tennénk bele if-fel, spurious wakeupnál hamis predikátummal lépne tovább. A helyes alak a wait(lock, [&]{ return closed_ || !q_.empty(); }) predikátumos hurok, utána a closed_/üresség ellenőrzése. Megjegyzés: a példa friend-del hozzáfér a privát tagokhoz — ez maga is szerződésszegés, mert a zár és az állapot védelme szétválik; éles kódban a queue adja a zár alatt futó műveletet. Ellenőrzési feltétel: TSan-futtatás (ASan, UBSan, MSan, TSan és Valgrind (16.2) eszköztár) nem jelez adatversenyt, és a fogyasztó zárt queue-n std::nullopt-tal tér vissza, nem blokkol.

34.2.8 Összefoglaló

Az RAII lock nem kényelmi eszköz, hanem szerződés: a kritikus szakasz az objektum élettartama, a zár típusa pedig a műveletből következik — lock_guard az egyszerű esethez, unique_lock a várakozáshoz és a kézi elengedéshez, scoped_lock a több mutexhez. A condition variable helyes használata három elem együttese: a védett állapot a mutex alatt változik, a várakozás mindig predikátumos ciklus (spurious wakeup és konkurens fogyasztó miatt), a notify pedig az állapotváltozás után, lehetőleg a záron kívül történik. A leállítás külön predikátumfeltétel (closed_) és notify_all, nem mellékhatás. Az időzített várakozásnál a steady_clock a domain, és a timeout nem bizonyíték semmire. A shared_mutex csak mért, olvasásdominált, hosszú kritikus szakaszoknál térül meg.