Что такое семафоры?

Что такое семафоры?

Семафор — это примитив синхронизации, который использует счётчик для управления доступом к ресурсу. Семафор отслеживает количество доступных ресурсов и позволяет нескольким потокам использовать ресурс одновременно (в отличие от mutex, который позволяет только одному потоку).

Общая идея

Семафор — это как система управления парковкой:

• Парковка с 5 местами (счётчик = 5)
• Когда приезжает машина, она занимает место (счётчик --)
• Когда уезжает машина, место освобождается (счётчик ++)
• Если парковка полная, новая машина ждёт

Основные операции

1. wait() (P-операция, от голландского "proberen" — пытаться)

void wait()  // или acquire()
{
    // Если счётчик > 0, уменьшаем его
    if (counter > 0) {
        counter--;
    } else {
        // Иначе блокируем поток
        block();
    }
}

2. signal() (V-операция, от "verhogen" — увеличивать)

void signal()  // или release()
{
    // Увеличиваем счётчик
    counter++;
    
    // Пробуждаем один ждущий поток
    if (есть ждущие потоки) {
        unblock_one();
    }
}

Типы семафоров

1. Бинарный семафор (Binary Semaphore)

Бинарный семафор может принимать значения 0 или 1. Это очень похоже на mutex, но есть разница.

#include <semaphore>
#include <thread>
#include <iostream>
using namespace std;

int counter = 0;
binary_semaphore sem(1);  // 1 или 0

void worker() {
    for (int i = 0; i < 5; ++i) {
        sem.acquire();  // wait()
        
        counter++;  // Критическая секция
        cout << "Counter: " << counter << endl;
        
        sem.release();  // signal()
    }
}

int main() {
    thread t1(worker);
    thread t2(worker);
    
    t1.join();
    t2.join();
    
    cout << "Final: " << counter << endl;  // 10
    return 0;
}

2. Считающий семафор (Counting Semaphore)

Считающий семафор может принимать любое неотрицательное значение. Используется для управления пулом ресурсов.

#include <semaphore>
#include <thread>
#include <iostream>
using namespace std;

const int POOL_SIZE = 3;  // 3 свободных ресурса
counting_semaphore sem(POOL_SIZE);

void worker(int id) {
    cout << "Worker " << id << " waiting for resource..." << endl;
    
    sem.acquire();  // Ждём свободного ресурса
    
    cout << "Worker " << id << " acquired resource" << endl;
    this_thread::sleep_for(chrono::seconds(1));
    cout << "Worker " << id << " releasing resource" << endl;
    
    sem.release();  // Освобождаем ресурс
}

int main() {
    vector<thread> threads;
    for (int i = 1; i <= 5; ++i) {
        threads.emplace_back(worker, i);
    }
    
    for (auto& t : threads) {
        t.join();
    }
    
    return 0;
}

/* Вывод (примерно):
Worker 1 waiting for resource...
Worker 2 waiting for resource...
Worker 3 waiting for resource...
Worker 1 acquired resource
Worker 2 acquired resource
Worker 3 acquired resource
Worker 4 waiting for resource...  (ждёт, нет свободных)
Worker 5 waiting for resource...
Worker 1 releasing resource
Worker 4 acquired resource
...
*/

Семафор vs Mutex

Практические примеры

Пример 1: Пул объединений (Connection Pool)

#include <semaphore>
#include <queue>
#include <memory>
using namespace std;

class DatabaseConnection { /* ... */ };

class ConnectionPool {
private:
    queue<shared_ptr<DatabaseConnection>> available_connections;
    counting_semaphore sem;
    mutex pool_mutex;
    
public:
    ConnectionPool(int pool_size) : sem(pool_size) {
        for (int i = 0; i < pool_size; ++i) {
            available_connections.push(
                make_shared<DatabaseConnection>()
            );
        }
    }
    
    shared_ptr<DatabaseConnection> acquire() {
        sem.acquire();  // Ждём свободного соединения
        
        lock_guard<mutex> lock(pool_mutex);
        auto conn = available_connections.front();
        available_connections.pop();
        return conn;
    }
    
    void release(shared_ptr<DatabaseConnection> conn) {
        {
            lock_guard<mutex> lock(pool_mutex);
            available_connections.push(conn);
        }
        sem.release();  // Увеличиваем счётчик
    }
};

// Использование
int main() {
    ConnectionPool pool(5);  // 5 соединений
    
    // 10 потоков конкурируют за 5 соединений
    thread t1([&] {
        auto conn = pool.acquire();
        cout << "Using connection" << endl;
        this_thread::sleep_for(chrono::milliseconds(100));
        pool.release(conn);
    });
    
    // ...
    return 0;
}

Пример 2: Producer-Consumer с ограниченным буфером

#include <semaphore>
#include <queue>
#include <iostream>
using namespace std;

const int BUFFER_SIZE = 3;

queue<int> buffer;
counting_semaphore empty_slots(BUFFER_SIZE);  // Изначально 3 свободных
counting_semaphore full_slots(0);             // Изначально 0 заполненных
mutex buffer_mutex;

void producer() {
    for (int i = 0; i < 10; ++i) {
        cout << "Producer: producing item " << i << endl;
        
        empty_slots.acquire();  // Ждём свободного слота
        
        {
            lock_guard<mutex> lock(buffer_mutex);
            buffer.push(i);
            cout << "  Buffer size: " << buffer.size() << endl;
        }
        
        full_slots.release();  // Пробуждаем consumer
    }
}

void consumer() {
    for (int i = 0; i < 10; ++i) {
        full_slots.acquire();  // Ждём заполненного слота
        
        int item;
        {
            lock_guard<mutex> lock(buffer_mutex);
            item = buffer.front();
            buffer.pop();
        }
        
        cout << "Consumer: consumed item " << item << endl;
        empty_slots.release();  // Пробуждаем producer
    }
}

int main() {
    thread p(producer);
    thread c(consumer);
    
    p.join();
    c.join();
    
    return 0;
}

Пример 3: Барьер (Barrier) — синхронизация N потоков

#include <semaphore>
#include <iostream>
using namespace std;

class Barrier {
private:
    int num_threads;
    int count = 0;
    mutex mutex;
    binary_semaphore sem(0);  // Изначально заблокирован
    
public:
    Barrier(int n) : num_threads(n) {}
    
    void wait() {
        lock_guard<mutex> lock(mutex);
        count++;
        
        if (count == num_threads) {
            // Все потоки достигли барьера
            // Пробуждаем их по одному
            for (int i = 0; i < num_threads; ++i) {
                sem.release();
            }
            count = 0;
        } else {
            // Ждём пока остальные доберутся
            lock.unlock();
            sem.acquire();
        }
    }
};

int main() {
    Barrier barrier(3);
    
    auto worker = [&](int id) {
        cout << "Thread " << id << " doing work..." << endl;
        this_thread::sleep_for(chrono::milliseconds(id * 100));
        
        cout << "Thread " << id << " at barrier" << endl;
        barrier.wait();
        cout << "Thread " << id << " passed barrier" << endl;
    };
    
    thread t1(worker, 1);
    thread t2(worker, 2);
    thread t3(worker, 3);
    
    t1.join();
    t2.join();
    t3.join();
    
    return 0;
}

Реализация семафора (упрощённо)

class Semaphore {
private:
    int count;
    mutex mtx;
    condition_variable cv;
    
public:
    Semaphore(int initial = 0) : count(initial) {}
    
    void acquire() {
        unique_lock<mutex> lock(mtx);
        // Ждём пока count > 0
        cv.wait(lock, [this] { return count > 0; });
        count--;
    }
    
    void release() {
        unique_lock<mutex> lock(mtx);
        count++;
        cv.notify_one();  // Пробуждаем одного ждущего потока
    }
    
    bool try_acquire() {
        lock_guard<mutex> lock(mtx);
        if (count > 0) {
            count--;
            return true;
        }
        return false;
    }
};

Сравнение примитивов синхронизации

Чеклист использования семафоров

1. Используй binary_semaphore, если:

   • Нужна сигнализация между двумя потоками
   • Один поток должен ждать события от другого

2. Используй counting_semaphore, если:

   • Нужно управлять пулом ресурсов
   • Несколько потоков должны использовать ресурсы поочерёдно
   • Ограниченное количество параллельных операций

3. Всегда используй RAII-обёртки:

   class SemaphoreGuard {
   public:
       SemaphoreGuard(Semaphore& sem) : sem(sem) { sem.acquire(); }
       ~SemaphoreGuard() { sem.release(); }
   private:
       Semaphore& sem;
   };
   

4. Избегай deadlocks при использовании нескольких семафоров

5. Помни о справедливости — очередь ожидания FIFO

Вывод: Семафор — это мощный примитив синхронизации, основанный на счётчике, для управления доступом нескольких потоков к ресурсам. Он может быть бинарным (0/1) или считающим (любое значение) и используется для реализации пулов ресурсов, продюсер-консюмер паттернов и других сложных сценариев синхронизации.