15.3 Mutex, condition variable és producer–consumer

Egy producer–consumer rendszerben a termelő munkát ad át, a fogyasztó pedig feldolgozza azt. A nehézség nem pusztán az, hogy egyszerre csak egy szál módosítsa a közös tárolót: azt is biztosítani kell, hogy a várakozó szálak megfelelő állapotváltozáskor felébredjenek, leálláskor pedig senki ne maradjon örökre blokkolva.

A Data race, happens-before és publikalás részben megismert szinkronizációs kapcsolatot most mutexszel hozzuk létre. A Dinamikus tömb és string builder tárolóinvariánsaihoz hasonlóan pontos állapotleírásból indulunk, de rögzített kapacitású sort használunk. Ez szándékos: a megtelt sor várakozásra kényszeríti a termelőt, így természetes visszanyomást, azaz backpressure-t ad.

15.3.1 Mutex, lock ownership és invariáns

A mutex kölcsönös kizárást biztosító szinkronizációs objektum. Sikeres zárolás után a hívó szál birtokolja a zárolást (lock ownership); a vizsgált, nem rekurzív használatban neki kell azt feloldania. Ez nem azonos az objektum tulajdonlásával: a mutex élettartamáért felelős szál nem feltétlenül az, amelyik pillanatnyilag zárolja.

A kritikus szakasz (critical section) a sikeres zárolás és feloldás közötti végrehajtás. A mutex csak akkor védi az adatot, ha minden konkurens hozzáférés ugyanazt a zárolási szerződést követi. Egy kizárólag olvasó diagnosztikai lekérdezés sem kerülheti meg ezt.

A két API nem keverhető. A pthreads függvények jellemzően közvetlenül adják vissza a hibaszámot, nem az errno változón keresztül. A C API eredményeit például a thrd_success és thrd_timedout értékekkel kell összehasonlítani. C17-ben a szálkönyvtár támogatása opcionális; a __STDC_NO_THREADS__ jelzés mellett a tényleges header- és linkelhetőséget is ellenőrizni kell.

A sor kapacitása legyen $C>0$, elemszáma $n$, következő olvasási indexe $h$, következő írási indexe $t$. Az invariáns:

$$0\leq n\leq C,\qquad 0\leq h,t<C,\qquad t=(h+n)\bmod C.$$

A closed állapot csak hamisból igazra változhat. Lezárás után nincs új beszúrás, de a már elfogadott elemek kiolvashatók. A mutex ezeket az állításokat együtt védi, nem csupán egyes számlálók írását.

Több mutex esetén előre rögzített zárolási sorrend (lock ordering) szükséges: ha minden művelet előbb A-t, majd B-t zárolja, nem alakulhat ki olyan kör, amelyben az egyik szál A mellett B-re, a másik B mellett A-ra vár. A példában egyetlen mutex szerepel.

15.3.2 A feltételváltozó nem eseményszámláló

A feltételváltozó (condition variable) várakozási mechanizmus. Nem maga tárolja, hogy „érkezett munka”; ezt a mutexszel védett állapotból képzett logikai feltétel, a predicate mondja meg.

POSIX szerint a pthread_cond_wait a birtokolt mutexet a várakozás megkezdésével összehangoltan feloldja, visszatérés előtt pedig újra zárolja. Ez hidalja át azt a veszélyes rést, amely külön „unlock, majd alvás” műveleteknél elveszett ébresztéshez vezethetne.

A predicate-et mindig ciklusban ellenőrizzük:

/* C17/POSIX szemléltető részlet; belépéskor m zárolva.
   A teljes példában a visszatérési kódokat is ellenőrizzük. */
while (!predicate) {
    pthread_cond_wait(&cv, &m);
}

Lehetséges álébredés (spurious wakeup), de enélkül is szükséges a ciklus: mire a felébresztett szál visszaszerzi a mutexet, másik fogyasztó már elvihette az elemet.

A signal nem tárol később beváltható értesítést. Ha nincs várakozó, a jövőbeli hívó az állapotot ellenőrzi, nem egy korábbi jelzésre támaszkodik.

15.3.3 Teljes példa: lezárható bounded queue

A következő queue.c ISO C17 nyelvet és POSIX szál-API-t használ. A pthread_condattr_setclock elérhetősége platformfeltétel.

Kijelölt laborprofil: Linux, x86-64 System V ABI, LP64; GCC 14.2.0 vagy Clang 19.1.0, glibc 2.40. A parancsok fordítási útmutatók; a példa itt nem fordított vagy mért eredmény.

A sor értékként másol int elemeket, tehát nem ad át dinamikus objektumokra mutató pointereket. A főszál termel, a munkaszál fogyaszt.

#define _POSIX_C_SOURCE 200809L
#include <errno.h>
#include <pthread.h>
#include <stdbool.h>
#include <stddef.h>
#include <stdio.h>
#include <stdlib.h>
#include <time.h>

enum { CAPACITY = 8 };

enum q_result { Q_OK, Q_CLOSED, Q_TIMEOUT };

struct queue {
    pthread_mutex_t mutex;
    pthread_cond_t not_empty;
    pthread_cond_t not_full;
    int data[CAPACITY];
    size_t head, tail, count;
    bool closed;
};

/* Váratlan szinkronizációs hibánál fail-fast hibapolitika. */
static void sync_check(int error)
{
    if (error != 0) {
        fprintf(stderr, "pthread-hiba: %d\n", error);
        abort();
    }
}

static int q_init(struct queue *q)
{
    pthread_condattr_t attr;
    int error = pthread_condattr_init(&attr);
    if (error != 0)
        return error;

    error = pthread_condattr_setclock(&attr, CLOCK_MONOTONIC);
    if (error != 0)
        goto fail_attr;

    error = pthread_mutex_init(&q->mutex, NULL);
    if (error != 0)
        goto fail_attr;

    error = pthread_cond_init(&q->not_empty, &attr);
    if (error != 0)
        goto fail_mutex;

    error = pthread_cond_init(&q->not_full, &attr);
    if (error != 0)
        goto fail_empty;

    q->head = 0;
    q->tail = 0;
    q->count = 0;
    q->closed = false;
    sync_check(pthread_condattr_destroy(&attr));
    return 0;

fail_empty:
    sync_check(pthread_cond_destroy(&q->not_empty));
fail_mutex:
    sync_check(pthread_mutex_destroy(&q->mutex));
fail_attr:
    sync_check(pthread_condattr_destroy(&attr));
    return error;
}

/* deadline: NULL vagy abszolút CLOCK_MONOTONIC határidő.
   Nem NULL esetén érvényes, változatlan timespec szükséges. */
static enum q_result q_push(
    struct queue *q, int value, const struct timespec *deadline)
{
    sync_check(pthread_mutex_lock(&q->mutex));

    while (!q->closed && q->count == CAPACITY) {
        int error;
        if (deadline != NULL) {
            error = pthread_cond_timedwait(
                &q->not_full, &q->mutex, deadline);
        } else {
            error = pthread_cond_wait(&q->not_full, &q->mutex);
        }

        if (error == ETIMEDOUT)
            break;
        sync_check(error);
    }

    enum q_result result;
    if (q->closed) {
        result = Q_CLOSED;
    } else if (q->count == CAPACITY) {
        result = Q_TIMEOUT;
    } else {
        q->data[q->tail] = value;
        q->tail = (q->tail + 1) % CAPACITY;
        ++q->count;
        sync_check(pthread_cond_signal(&q->not_empty));
        result = Q_OK;
    }

    sync_check(pthread_mutex_unlock(&q->mutex));
    return result;
}

/* out érvényes, kizárólagosan írható int objektumra mutat.
   Q_CLOSED esetén *out változatlan. */
static enum q_result q_pop(struct queue *q, int *out)
{
    sync_check(pthread_mutex_lock(&q->mutex));

    while (!q->closed && q->count == 0)
        sync_check(pthread_cond_wait(&q->not_empty, &q->mutex));

    enum q_result result;
    if (q->count == 0) {
        result = Q_CLOSED;
    } else {
        *out = q->data[q->head];
        q->head = (q->head + 1) % CAPACITY;
        --q->count;
        sync_check(pthread_cond_signal(&q->not_full));
        result = Q_OK;
    }

    sync_check(pthread_mutex_unlock(&q->mutex));
    return result;
}

static void q_close(struct queue *q)
{
    sync_check(pthread_mutex_lock(&q->mutex));
    q->closed = true;
    sync_check(pthread_cond_broadcast(&q->not_empty));
    sync_check(pthread_cond_broadcast(&q->not_full));
    sync_check(pthread_mutex_unlock(&q->mutex));
}

/* Csak minden használó befejeződése után hívható. */
static void q_destroy(struct queue *q)
{
    sync_check(pthread_cond_destroy(&q->not_full));
    sync_check(pthread_cond_destroy(&q->not_empty));
    sync_check(pthread_mutex_destroy(&q->mutex));
}

struct worker {
    struct queue *queue;
    long sum;
};

static void *consume(void *argument)
{
    struct worker *w = argument;
    int value;
    long sum = 0;

    while (q_pop(w->queue, &value) == Q_OK)
        sum += value;  /* Feldolgozás a sor mutexén kívül. */

    w->sum = sum;
    return NULL;
}

int main(void)
{
    struct queue q;
    int error = q_init(&q);
    if (error != 0) {
        fprintf(stderr, "q_init: %d\n", error);
        return EXIT_FAILURE;
    }

    struct worker worker = { .queue = &q, .sum = 0 };
    pthread_t thread;
    error = pthread_create(&thread, NULL, consume, &worker);
    if (error != 0) {
        fprintf(stderr, "pthread_create: %d\n", error);
        q_destroy(&q);
        return EXIT_FAILURE;
    }

    bool failed = false;
    for (int value = 1; value <= 1000; ++value) {
        if (q_push(&q, value, NULL) != Q_OK) {
            failed = true;
            break;
        }
    }

    q_close(&q);
    sync_check(pthread_join(thread, NULL));
    q_destroy(&q);

    if (printf("osszeg=%ld\n", worker.sum) < 0)
        return EXIT_FAILURE;

    return !failed && worker.sum == 500500L
        ? EXIT_SUCCESS : EXIT_FAILURE;
}

gcc -std=c17 -Wall -Wextra -Wpedantic -O2 -pthread queue.c -o queue-gcc
clang -std=c17 -Wall -Wextra -Wpedantic -O2 -pthread queue.c -o queue-clang

A szerződésből következő várt, nem megfigyelt összeg:

$$\sum_{k=1}^{1000}k=\frac{1000\cdot1001}{2}=500500.$$

15.3.3.1 Tulajdonlás és hibaszerződés

Az inicializált sort nem másoljuk értékadással vagy memcpy-val. A mutex és a feltételváltozók nem másolható sorállapotként kezelendők.

A részleges inicializálás hibája visszatérési kódot ad, és csak a már létrehozott erőforrásokat bontja le. Üzem közben a lezárás és az időtúllépés normális eredmény. Váratlan szinkronizációs hibánál a példa megszakítja a folyamatot: nem állítja, hogy ismeretlen zárolási állapotból biztonságosan folytathat.

A szerződés kizárja a szálak kényszerített cancellationjét. A POSIX condition wait cancellation point; annak támogatásához külön cleanup protokoll kellene.

15.3.4 Helyesség és leállítás

Invariánsmegőrzési bizonyításvázlat. Beszúrás csak $n<C$ esetén történik. Utána:

$$n'=n+1,\qquad t'=(t+1)\bmod C,$$

ezért

$$t'=(h+n+1)\bmod C=(h+n')\bmod C.$$

Kivételnél $n>0$, majd $h'=(h+1)\bmod C$ és $n'=n-1$, tehát $(h'+n')\bmod C=t$. A lezárás nem változtatja az indexeket. A mutex miatt két ilyen átmenet nem módosíthatja egyszerre az állapotot.

A q_close ürítve leállást jelent:

1. A mutex alatt beállítja a lezárást.
2. Mindkét feltételváltozón broadcast történik.
3. Feloldja a mutexet.
4. A vezérlőszál megvárja a használó szálakat.
5. Csak ezután történik megsemmisítés.

Mindkét várakozási csoportot ébreszteni kell: a termelők telt soron, a fogyasztók üres soron várhatnak. A lezárást követő beszúrás visszautasított; a fogyasztók a meglévő elemeket még feldolgozzák.

Leállási bizonyításvázlat. A closed beállítása után egyik várakozási ciklus sem kezd újabb várakozást. A broadcast a már várakozókat is eléri, a később belépők pedig az állapotból ismerik fel a lezárást. A join alatt nincs birtokolt sormutex, így a kilépéshez szükséges zárolást nem akadályozza a várakozó főszál.

Ez feltételes állítás: a szálaknak futási lehetőséget kell kapniuk, a feldolgozásnak végesnek kell lennie, és nem akadhatnak el más erőforráson. A q_close nem szakít meg tetszőleges hálózati I/O-t.

Általános, többtermelős shutdownnál a lezárás előzze meg a potenciálisan blokkolt termelők join-ját. Ellenkező esetben éppen azt a szálat várhatnánk, amely csak a lezárás után tudna visszatérni.

15.3.5 Timed wait: az óra és a predicate külön kérdés

A POSIX pthread_cond_timedwait abszolút határidőt fogad. Az órabázis a feltételváltozó attribútuma; alapértelmezésben CLOCK_REALTIME, a példában explicit CLOCK_MONOTONIC.

A határidőt ugyanabból az órából kell képezni, mint amelyet a feltételváltozó használ. Legyen az aktuális idő $(s,n)$, a hozzáadandó idő pedig $\Delta$ nanoszekundum. Megfelelően széles, túlcsordulás-ellenőrzött számítással:

$$u=n+\Delta,\qquad d_s=s+\left\lfloor\frac{u}{10^9}\right\rfloor,\qquad d_n=u\bmod10^9.$$

A tv_nsec mezőre $0\leq d_n<10^9$ szükséges. A végső másodpercértéknek bele kell férnie a célmezőbe. Álébredés után nem számolunk új határidőt: az újraindított relatív várakozás korlátlanul meghosszabbíthatná a műveletet.

Az ETIMEDOUT nem bizonyítja, hogy a predicate a visszatéréskor hamis. A mutex visszaszerzéséig más szál helyet szabadíthatott fel vagy lezárhatta a sort. Ezért a példában az állapot dönt:

• lezárt sor: Q_CLOSED;
• nyitott, továbbra is telt sor: Q_TIMEOUT;
• nyitott sor szabad hellyel: sikeres beszúrás.

Ez nem szigorú teljes végrehajtási határidő. A mutex első megszerzése sem időkorlátos, és a határidő lejárta után rendelkezésre álló helyet a művelet még felhasználhatja.

15.3.6 Költség és gyakori hibák

A beszúrás és kivétel állapotmódosítása worst-case $O(1)$, külön allokáció nélkül, folytonos tömbtárolóval. A várakozási idő viszont nem korlátos; a mutex nem biztosít alkalmazási szintű méltányosságot. A másolt elemek érvényességét a sor későbbi módosítása nem érinti.

Gyakori hiba az if használata while helyett, a predicate zárolás nélküli olvasása, illetve a feldolgozás végrehajtása a sormutex alatt. Utóbbi nem feltétlen adatverseny, de megszünteti a párhuzamosságot, és további zárolásokkal holtpontot okozhat.

A broadcast önmagában nem állít le semmit: állapotváltozás nélkül a helyesen megírt várakozók visszaalszanak. A close pedig nem teszi azonnal biztonságossá a destroy hívást.

15.3.7 Feladatok

1. Magyarázó: Miért különbözik a kivétel várakozási predicate-je a siker feltételétől? Mi történjen lezárt, de nem üres soron?

2. Kódolási: Készíts egyszálú timeout-tesztet. Töltsd tele a sort, képezz CLOCK_MONOTONIC alapján normalizált, 50 ms múlva lejáró határidőt, majd próbálj új elemet beszúrni. Ezután zárd le és ürítsd ki.

3. Diagnosztikai: Egy leállító függvény zárolja a sormutexet, beállítja a closed mezőt, broadcast-ot hív, majd még a feloldás előtt pthread_join-nal várja a fogyasztót. Miért akadhat el akkor is, ha minden adat-hozzáférés zárolt?

15.3.8 Megoldások

15.3.8.1 Predicate és ürítés

A closed || count > 0 azt jelenti, hogy nincs értelme tovább várni. Sikerhez azonban elem is szükséges. Lezárt, nem üres soron a kivétel sikeres; lezárt, üres soron Q_CLOSED az eredmény.

Ellenőrzési feltétel: minden korábban sikeresen beszúrt elem pontosan egyszer kivehető, majd a következő kivétel lezárást jelez.

15.3.8.2 Timeout-teszt

A teszt q_init után CAPACITY sikeres beszúrást végez NULL határidővel. A clock_gettime hibáját külön, annak errno-s szerződése szerint kezeli. A hozzáadandó idő $50\,000\,000$ nanoszekundum; ha az összeg eléri az egymilliárdot, egy másodperc átvitel szükséges.

Fogyasztó hiányában az új beszúrás Q_TIMEOUT eredményt ad. Lezárás után a feltöltött értékeket FIFO-sorrendben kell visszakapni, végül Q_CLOSED következik.

Ellenőrzési feltétel: a sikertelen beszúrás értéke nincs a sorban. Pontos 50 ms-os visszatérést nem szabad elvárni az ütemezőtől.

15.3.8.3 Join a kritikus szakaszban

A fogyasztónak az ébredés után vissza kell szereznie a mutexet. A főszál ezt birtokolja, miközben a fogyasztó befejeződésére vár:

$$\text{főszál}\rightarrow\text{fogyasztó befejezése} \rightarrow\text{mutex}\rightarrow\text{főszál}.$$

A javítás a join áthelyezése a feloldás után.

Ellenőrzési feltétel: egyetlen szálra sem várunk úgy, hogy közben a befejeződéséhez szükséges zárolást birtokoljuk.

15.3.9 Összefoglaló

A mutex invariánst véd; a feltételváltozó az invariánshoz kapcsolódó predicate megváltozására enged várni. A predicate-et ciklusban, ugyanazon mutex alatt ellenőrizzük. Leálláskor állapotot módosítunk, minden érintett várakozót ébresztünk, feloldunk, majd csatlakozunk a szálakhoz. A timeout óraalapja és eredményszemantikája az API-szerződés része, nem pusztán időzítési részlet.