Mi az a szemafor?

A szemafor egy olyan programozási eszköz, amellyel a szálak és folyamatok közös erőforrásaihoz való hozzáférést lehet szabályozni és szinkronizálni.

Gyakran két alapvető típust különböztetünk meg: az általános (számláló), illetve a bináris szemafort.

Általános (számláló) szemafor

  • Fogalom: egy nemnegatív egész értékkel jellemezhető szinkronizációs primitív, amely azt jelzi, hogy mennyi szabad erőforrás érhető el.
  • Működés: ha értéke nagyobb, mint 0, a szál/folyamat beléphet, a számláló értéke pedig eggyel csökken. Ha 0, akkor a szál/folyamat várakozik.
  • Használat: több azonos típusú erőforrás (például egyező számú adatbázis-kapcsolat) kezelésére.

Bináris szemafor

  • Fogalom: olyan szemafor, amelynek értéke kizárólag 0 vagy 1 lehet.
  • Működés: Az 1 azt jelenti, hogy az erőforrás szabad, a 0 azt, hogy foglalt (zárolt).
  • Használat: Kölcsönös kizárás (mutual exclusion) megvalósítására használják, amikor egyszerre csak egyetlen folyamat vagy szál tartózkodhat a kritikus szakaszban.
  • Különbség a mutex-hez képest: Bár logikailag hasonlít a mutexhez, a bináris szemafor nem feltétlenül kötelez arra, hogy ugyanaz a szál oldja fel, amelyik zárolta.

Érdemes elolvasni a mutex-ről írt posztot is.

Szemafor C++20-as implementációja

A C++20-as szabvány bevezette a <semaphore> fejlécfájlt, ami mindkét szemafor típus használatát támogatja.

Általános (számláló) szemafor példa

Az alábbi program azt szimulálja, hogy fix számú (2 darab) adatbázis-kapcsolatunk van, de egyszerre 4 szál szeretné használni azokat.

 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
33
34
35
36
37
38

#include <chrono>
#include <iostream>
#include <semaphore>
#include <thread>
#include <vector>

// Maximum 2 szál férhet hozzá egyszerre az erőforráshoz
std::counting_semaphore<2> dbSemaphore{ 2 };

void AccessDatabase(const int thread_id) 
{
    std::cout << "[Thread " << thread_id << "] Waiting for database connection ...\n";
    
    // P művelet (wait / acquire): csökkenti a számlálót, ha 0, akkor blokkol
    dbSemaphore.acquire(); 
    
    std::cout << "[Thread " << thread_id 
              << "] ---> Successfully connected to database.\n";
    std::this_thread::sleep_for(std::chrono::milliseconds{ 1000 }); // Munka szimulálása
    
    std::cout << "[Thread " << thread_id << "] <--- Connection terminated.\n";
    
    // V művelet (signal / release): növeli a számlálót, felébreszt egy várakozót
    dbSemaphore.release(); 
}

int main() {
    std::vector<std::thread> threads;
    threads.reserve(4);
    
    for (int i = 1; i <= 4; ++i) {
        threads.emplace_back(AccessDatabase, i);
    }
    
    for (auto& t : threads) {
        t.join();
    }
}

Bináris szemafor példa

Az alábbi példában egy kritikus szakasz védelmét szimuláljuk, amelyben a konzolra írunk.

 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 <semaphore>
#include <thread>
#include <vector>

// Kezdeti értéke 1, vagyis szabad
std::binary_semaphore criticalSectionSignal{ 1 };

void SafePrint(const int thread_id) {
    // Zárolás: ha a szemafor 1, átengedi a szálat és beállítja 0-ra
    criticalSectionSignal.acquire();
    
    // Kritikus szakasz (egyszerre csak egy szál hajthatja végre)
    std::cout << "Start of critical section - Thread: " << thread_id << std::endl;
    std::cout << "End of critical section   - Thread: " << thread_id << std::endl;
    
    // Feloldás: visszaállítja 1-re az értéket
    criticalSectionSignal.release();
}

int main() {
    std::vector<std::thread> threads;
    threads.reserve(3);
    
    for (int i = 1; i <= 3; ++i) {
        threads.emplace_back(SafePrint, i);
    }
    
    for (auto& t : threads) {
        t.join();
    }
}

Szemafor implementáció C++20 előtt

A szemaforokat elemi szinten leggyakrabban egy számláló, egy kölcsönös kizárást biztosító mutex és egy várakozási sor (vagy condition variable) kombinációjával lehet megvalósítani.

Az alábbiakban egy olyan C++ osztályt mutatok be, amely az operációs rendszerek absztrakt működési elvét követi. A szálak elaltatásához a minimális hordozhatóság miatt itt a natív primitíveket (std::mutex, std::condition_variable) használjuk.

 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
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

#include <condition_variable>
#include <mutex>
#include <stdexcept>

class CustomSemaphore
{
public:
    // Konstruktor: beállítjuk a számláló kezdeti értéket 
    // (binárisnál: 1, számlálónál > 1)
    explicit CustomSemaphore(const int initialCount = 1)
        : count{ initialCount } 
    {
        if (initialCount < 0) {
            throw std::invalid_argument{ "Counter's initial value cannot be negative!" };
        }
    }

    // P művelet (wait / acquire)
    void Acquire() {
        std::unique_lock<std::mutex> lock{ mtx };

        // Amíg nincs szabad erőforrás (count <= 0),
        // a szál blokkolódik és elalszik
        while (count <= 0) {
            cv.wait(lock);
        }

        // Erőforrás lefoglalása
        --count;
    }

    // V művelet (signal / release)
    void Release() {
        std::unique_lock<std::mutex> lock{ mtx };

        // Erőforrás felszabadítása
        ++count;
        
        // Értesítünk EGY várakozó szálat, hogy felébredhet 
        // és újra ellenőrizheti a feltételt
        cv.notify_one();
    }

private:
    int count;                     // A szabad erőforrások számlálója
    std::mutex mtx;                // Belső mutex a számláló védelmére
    std::condition_variable cv;    // Feltételi változó a várakozó szálaknak
};

Az Acquire() metódusban nem if, hanem while szerepel, mert a feltételt felébredés után mindig újra ellenőrizni kell, mert a szál felébredése önmagában nem jelenti azt, hogy az erőforrás valóban rendelkezésre áll.

Mit tanultunk?

A mutex + condition_variable alapú megvalósításban a várakozó szálak elalszanak, ezért nem fogyasztják folyamatosan a CPU-erőforrást. Van azonban egy másik megközelítés is: az aktív várakozás, amelyben a szál nem blokkolódik, hanem atomi műveletekkel próbálja megszerezni a szemafort.

Lock-free szemafor implementáció

A következő megvalósítások nem használnak explicit mutexet vagy condition variable-t, hanem atomi műveletekre és aktív várakozásra épülnek. Az elnevezésben szereplő lock-free arra az esetre vonatkozik, amikor az alkalmazott atomi műveletek a célplatformon valóban lock-free módon valósulnak meg.

A C++ atomi műveleteiről írt posztomat érdemes elolvasni.

Lock-free bináris szemafor

A bináris szemafor ebben a megközelítésben egyetlen atomi jelzőbit (flag). A szál addig “pörög” egy ciklusban (aktívan vár), amíg a jelző bitet nem sikerül átállítania.

 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 <atomic>
#include <thread>

class LockFreeBinarySemaphore
{
public:
    explicit LockFreeBinarySemaphore(const bool initialState = true)
        : state{ initialState } {}

    // P művelet (wait / acquire)
    void Acquire() {
        bool expected = true;

        while (!state.compare_exchange_weak(expected, false, 
                                            std::memory_order_acquire,
                                            std::memory_order_relaxed)) {
            expected = true;

            // Lehetővé tesszük, hogy az ütemező másik szálnak adja át a CPU-t.
            std::this_thread::yield();
        }
    }

    // V művelet (signal / release)
    void Release() {
        state.store(true, std::memory_order_release);
    }

private:
    //false (0): foglalt, true (1): szabad
    std::atomic<bool> state;
};

Az acquire/release memória-rendezés (memory-ordering) biztosítja, hogy a szemafor Release() és Acquire() műveletei között a memória-rendezés megfelelő happens-before kapcsolat alakulhasson ki a kritikus szakaszt elhagyó és azt később megszerző szál között.

Lock-free számláló szemafor

A számláló szemafor esetében egy egész számot kell atomi módon csökkentenünk vagy növelnünk. A nehézséget az jelenti, hogy nem engedhetjük meg, hogy a számláló 0 alá csökkenjen.

 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
33
34
35
36
37
38
39
40
41
42
43

#include <atomic>
#include <stdexcept>
#include <thread>

class LockFreeCountingSemaphore 
{
public:
    explicit LockFreeCountingSemaphore(const int initialCount)
        : count{ initialCount } 
    {
        if (initialCount < 0) {
            throw std::invalid_argument{ "Counter's initial value cannot be negative!" };
        }
    }

    // P művelet (wait / acquire)
    void Acquire() {
        int currentCount = count.load(std::memory_order_relaxed);

        while (true) {
            if (currentCount <= 0) {
                currentCount = count.load(std::memory_order_relaxed);
                std::this_thread::yield();
                continue;
            }

            // Sikertelen CAS esetén currentCount megkapja az aktuális értéket.
            if (count.compare_exchange_weak(currentCount, currentCount - 1,
                                            std::memory_order_acquire,
                                            std::memory_order_relaxed)) {
                break; // Sikeres csökkentés, megszereztük az erőforrást!
            }
        }
    }

    // V művelet(signal / release)
    void Release() {
        count.fetch_add(1, std::memory_order_release);
    }

private:
    std::atomic<int> count;
};

Hagyományos és zármentes megvalósítás összehasonlítása

TulajdonságZármentes (aktív várakozás)Hagyományos (zárolás + elaltatás)
Késleltetés (latency)Rövid várakozásnál nagyon alacsony lehet.A felébresztés költsége miatt nagyobb lehet.
CPU használatJelentős lehet a várakozás alatt.Általában minimális várakozás közben (a szál alszik).
Hosszú várakozásKedvezőtlen.Kedvezőbb.
Ajánlott használatHa a várakozás várhatóan nagyon rövid.Ha a szálnak hosszabb ideig kell várakoznia.

A konkrét std::counting_semaphore implementációk a háttérben különböző optimalizációkat alkalmazhatnak. Egyes megvalósítások rövid ideig aktívan várakozhatnak, majd hosszabb várakozás esetén elaltathatják a szálat.