C++ atomikus műveletei

Az ebben a posztban vizsgált C++ atomi műveletek két nagy csoportra oszthatók. Az első csoportba az atomi változók értékének kiolvasása és módosítása, vagyis a load() és store() műveletek tartoznak. A második, érdekesebb csoportot az úgynevezett RMW (read-modify-write) műveletek alkotják, amelyek egyetlen atomi műveletben olvassák, módosítják és tárolják az értéket.

Az atomicitás azt jelenti, hogy az adott művelet szempontjából a többi szál nem figyelhet meg egy köztes állapotot. EGY RMW művelet például nem bomlik két, egymástól függetlenül végrehajtható “olvasás” és “írás” műveletre.

Kiolvasó/tároló művelet

Az x atomi változóból az x.load() függvénnyel tudjuk kiolvasni az aktuális értékét. Új értéket pedig a x.store(value) függvénnyel tudunk beállítani. Ennek első (kötelező) paramétere az újonnan beállítandó érték.

Az itt bemutatott műveletek alapértelmezett memória sorrendje std::memory_order_seq_cst. A memória sorrendekkel később részletesebben foglalkozunk.

RMW műveletek

Az olvasó-módosító-író műveletek csoportja további három alcsoportra oszható. A fetch_* műveletek módosítják a korábbi értéket, és közben vissza is adják azt (azaz a módosítás előtti értéket). A test_and_set az std::atomic_flag típuson használható, tipikus alkalmazása a lockok, például spinlockok megvalósítása. A compare_exchange egy atomi változó értékét akkor cseréli le egy új értékre, ha annak aktuális értéke megegyezik az általunk elvárt értékkel.

`fetch_*` műveletek

Az egész típusú std::atomic változóknál az alábbi öt alapvető fetch_* művelettel találkozhatunk:

fetch_add
fetch_sub
fetch_and
fetch_or
fetch_xor

`fetch_add`

Az x.fetch_add(increment) művelet az x atomi változó értékét atomikusan növeli increment-tel, és visszaadja x módosítás előtti értékét.

Ha például van egy függvényünk, ami 1000-szer végrehajtja egymást követően az 1-gyel növelést egy statikus atomi változón (fetch_add-ot használva), és ezt a függvényt konkurrens módon két szálon futtatjuk, akkor garantált, hogy 2000-rel növekszik a változó értéke.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
#include <atomic>
#include <iostream>
#include <thread>

std::atomic<int> counter{ 0 };

void Worker()
{
    for (int i = 0; i < 1000; ++i)
        counter.fetch_add(1);
}

int main()
{
    std::thread t1{ Worker };
    std::thread t2{ Worker };

    t1.join();
    t2.join();

    std::cout << counter.load() << '\n'; // 2000
}

Ha a counter.fetch_add(1) utasítás helyett const int old = counter.load(), majd counter.store(old + 1) állna, akkor ez két külön atomikus művelet lenne, így több szálon probléma állhatna elő.

`fetch_sub`

A fetch_sub teljesen hasonlóan működik, mint a fetch_add, csak összeadás helyett kivonást hajt végre.

Az alábbi példaprogram mutatja a működését.

1
2
3
4
5
6
std::atomic<int> x{10};

const int old = x.fetch_sub(3);

std::cout << old << '\n';       // 10
std::cout << x.load() << '\n';  // 7

`fetch_and`, `fetch_or` és `fetch_xor`

Ezek a függvények a bitenkénti és, vagy, illetve kizáró vagy műveleteket valósítják meg.

Az alábbi kódban definiálunk egy 8 bites egészt, amit egy bitsorozattal inicializálunk. Jól látszik, hogy a fetch_or bitenkénti vagy kapcsolattal módosítja az egyes biteket.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
#include <atomic>
#include <format>
#include <iostream>

int main()
{
    std::atomic<uint8_t> flags{ 0b00110110 };

    const auto old = flags.fetch_or(0b10001001);

    std::cout << "old:   " << std::format("{:08b}", old) << '\n'; 
    std::cout << "flags: " << std::format("{:08b}", flags.load()) << '\n'; 
    // old:   00110110
    // flags: 10111111
}

`test_and_set`

Ha van egy std::atomic_flag típusú változónk, akkor arra meghívhatjuk a test_and_set függvényt. A művelet hatását három lépésben írhatjuk le, miközben az egész művelet a C++ memória-modell szempontjából egyetlen atomi műveletként hajtódik végre:

Lekérdezi a flag jelenlegi értékét
true-ra állítja a flag értékét
Visszaadja a korábban lekérdezett értéket

A test_and_set használatával készíthetünk egy ún. spinlock-ot, amivel kölcsönös kizárást valósíthatunk meg. A Spinlock osztály egy std::atomic_flag adattaggal rendelkezik, amelyet clear/false állapottal inicializálunk. Az osztálynak lock() és unlock() metódusa van, melyek a Spinlock objektumot lezárt, illetve feloldott állapotba helyezik.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
#include <atomic>

class Spinlock {
private:
    // A flag kezdetben hamis (szabad) állapotban van
    std::atomic_flag flag = ATOMIC_FLAG_INIT;

public:
    void lock() {
        // Addig ismételjük, amíg a test_and_set igazat ad vissza 
        // (azaz a zár foglalt volt)
        // A std::memory_order_acquire biztosítja a szükséges memória-sorrendet 
        // a lock megszerzése és a kritikus szakasz műveletei között.
        while (flag.test_and_set(std::memory_order_acquire)) {
            // Busy waiting
        }
    }

    void unlock() {
        // Felszabadítjuk a zárat (visszaállítjuk hamisra)
        // A std::memory_order_release garantálja, hogy a kritikus szekció 
        // minden korábbi  módosítása láthatóvá válik a többi szál számára, 
        // mielőtt a zár felszabadul.
        flag.clear(std::memory_order_release);
    }
};

A Spinlock használatával az alábbi módon tudunk négy szálon többszöri inkrementálást végrehajtani. A növelendő változó (counter) nem atomikus, hanem hagyományos int típusú.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
#include <iostream>
#include <mutex>
#include <thread>
#include <vector>

Spinlock my_spin;
int counter = 0;

void IncreaseHundredThousandTimes() {
    for (int i = 0; i < 100'000; ++i) {
        // A lock_guard a konstruktorában meghívja a my_spin.lock()-ot,
        // a destruktorában pedig automatikusan a my_spin.unlock()-ot.
        std::lock_guard<Spinlock> lock{ my_spin };
        ++counter;
    }
}

int main() {
    std::vector<std::thread> threads;
    
    // 4 szál indítása, amik ugyanazt a változót módosítják
    for (int i = 0; i < 4; ++i) {
        threads.emplace_back(IncreaseHundredThousandTimes);
    }

    for (auto& t : threads) {
        t.join();
    }

    std::cout << "Final value: " << counter << std::endl; // 400'000
    return 0;
}

A Spinlock hátránya az std::mutex-szel szemben, hogy a várakozó szál nem “alszik el”, hanem folyamatosan futtatja a ciklust. Emiatt hosszabb várakozás esetén jelentős CPU-időt fogyaszthat. Rövid ideig tartó kritikus szakaszoknál viszont lehet értelme.

A saját Spinlock osztály ugyan nem öröklődik semmilyen mutex osztályból, mégis használható std::lock_guard-dal, mert biztosítja a lock() és unlock() metódusokat. A C++ standard könyvtár számos esetben nem öröklési hierarchiára épít, hanem azt várja el, hogy egy típus bizonyos műveleteket és az azokhoz tartozó viselkedést biztosítson. Viszont ezt – más nyelvekkel ellentétben – nem öröklési vagy interfész megvalósítási módon ellenőrzi.

`compare-exchange`

A szakirodalomból ismert compare-and-swap (CAS) műveletnek C++-ban a compare_exchange_weak() és compare_exchange_strong() felel meg.

A függvények két paraméterrel dolgoznak:

expected: az az érték, amelyet az atomi változóban várunk;
desired: az az új érték, amelyre az atomi változót módosítani szeretnénk.

Ha x egy atomi változó, akkor az x.compare_exchange_*(expected, desired) logikája a következő:

Megvizsgálja, hogy x értéke megegyezik-e az expected értékével.
Ha igen, akkor x értékét a desired értékre cseréli, és true-val tér vissza a metódus.
Ha nem, akkor az x értékét nem módosítja, viszont expected értékét felülírja x értékével, és false-szal tér vissza. Mivel expected-et referenciaként veszi át a függvény, így ebben az esetben a módosulás a hívó környezet számára is érzékelhető lesz.

Az expected a fentiek miatt in/out paraméter: a függvény bemenetként használja, sikertelen összehasonlítás esetén pedig kimenetként módosítja.

Ennek a furcsa működésnek a következő a magyarázata:

Van egy elvárásunk, hogy mi a változónk értéke. Ha ez az elvárás teljesül, akkor lecseréljük a változónk értékét az új értékre, és jelezzük, hogy sikeresen végrehajtottuk a cserét.
Ha viszont a változónk értéke nem egyezik meg az elvárttal, akkor ez amiatt lehetséges, hogy egy másik szálon módosult a változó értéke. Ilyekor az elvárást igazítjuk a változó aktuális értékéhez, majd jelezzük, hogy sikertelen volt a csere művelet.

Az elmondottakra építve az eggyel növelés műveletét az alábbi módon implementálhatjuk:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
std::atomic<int> counter{0};

void Increment()
{
    int old = counter.load();

    while (!counter.compare_exchange_weak(old, old + 1))
    {
        // sikertelen volt
        // old közben frissült
    }
}

Ha a counter értéke időközben megváltozott, a compare_exchange_weak sikertelen lesz, és az old értékét frissíti. A ciklus ezután az új érték alapján újabb kísérletet tesz. weak használata esetén a sikertelenségnek lehet egy másik oka is: előfordulhat ún. spurious failure, amikor az összehasonlítás sikeres lehetett volna, a művelet mégis false értékkel tér vissza.

Érdemes végiggondolni, hogy hogyan fut le az Increment függvény, ha két szálon futtatjuk konkurrens módon. Tegyük fel, hogy counter értéke éppen 10. Az egyik szálon (t1) old értéke 10 lesz, de mielőtt a compare_exchange_weak végrehajtódna ezen a szálon, egy másik szál (t2) a counter-t 11-re módosítja. Az összehasonlítás ezért t1 szálon sikertelen lesz. A compare_exchange_weak ekkor old értékét 11-re frissíti (ezzel lekövetve a t2-n történt módosulást), a következő próbálkozás _t1_en pedig már a 11-ről 12-re növeléssel kísérletezik.

`weak` vs `strong`

Mi a különbség a weak és a strong változat között?

weak esetben előállhat ún. spurious failure, ami akkor is sikertelenséget jelezhet, amikor a változó értéke megegyezik az expected-del. Ez a helyzet a strong esetben nem állhat elő. Ha a strong változat hamis értéket ad vissza, akkor biztos, hogy x és expected értéke különböző volt.

Összefoglalás

Az std::atomic RMW műveletei lehetővé teszik, hogy egy megosztott változót ne csak atomikusan olvassunk vagy írjunk, hanem az értéket egyetlen atomi műveletben módosítsuk is.

A fetch_* műveletek egyszerű számlálók és bitenkénti flag-ek kezelésére használhatók. A test_and_set() segítségével például egyszerű spinlock készítható.

A compare_exchange_* ennél általánosabb mechanizmust biztosít: egy érték módosítását ahhoz köthetjük, hogy az értéke még mindig az általunk várt érték legyen. Ennek segítségével ún. CAS ciklusokat írhatunk, amelyek számos lock-free adatstruktúra alapját képezik.

Fontos, hogy az std::atomic használata önmagában nem jelenti azt, hogy az adott művelet vagy algoritmus lock-free; a CAS ciklusok viszont fontos eszközei a lock-free algoritmusok megvalósításának.

Kiolvasó/tároló művelet#

RMW műveletek#

fetch_* műveletek#

fetch_add#

fetch_sub#

fetch_and, fetch_or és fetch_xor#

test_and_set#

compare-exchange#

weak vs strong#

Összefoglalás#