С какими объектами синхронизации работал

С какими объектами синхронизации работал

Объекты синхронизации — это механизмы, используемые для управления доступом к общим ресурсам в многопоточных приложениях. В моей практике я работал с различными такими объектами на разных уровнях абстракции.

Низкоуровневые примитивы синхронизации

1. Мьютекс (Mutex)

Мьютекс (Mutual Exclusion) — самый базовый примитив, обеспечивающий эксклюзивный доступ к ресурсу:

#include <mutex>
#include <thread>
#include <iostream>

std::mutex mtx;
int sharedCounter = 0;

void incrementCounter() {
    for (int i = 0; i < 1000; ++i) {
        {
            std::lock_guard<std::mutex> lock(mtx); // RAII
            sharedCounter++;
        }
    }
}

int main() {
    std::thread t1(incrementCounter);
    std::thread t2(incrementCounter);
    
    t1.join();
    t2.join();
    
    std::cout << "Result: " << sharedCounter << std::endl; // 2000
}

Рекурсивный мьютекс (Recursive Mutex): Позволяет одному потоку несколько раз заблокировать один и тот же мьютекс:

std::recursive_mutex recursive_mtx;

class Node {
    std::recursive_mutex mtx;
    
    void processNode() {
        std::lock_guard<std::recursive_mutex> lock(mtx);
        // Может вызвать другой метод, который тоже блокирует mtx
        processChildren();
    }
    
    void processChildren() {
        std::lock_guard<std::recursive_mutex> lock(mtx); // Нет deadlock'а!
    }
};

2. Семафор (Semaphore)

Семафор подсчитывает доступные ресурсы. Полезен для управления ограниченным количеством ресурсов:

#include <semaphore>

std::counting_semaphore<5> poolSemaphore(5); // 5 доступных ресурсов

void usePoolResource() {
    poolSemaphore.acquire(); // Уменьшить счётчик
    
    try {
        // Использовать ресурс (максимум 5 потоков одновременно)
        std::cout << "Using resource" << std::endl;
        std::this_thread::sleep_for(std::chrono::seconds(1));
    } finally {
        poolSemaphore.release(); // Увеличить счётчик
    }
}

Пример с пулом соединений:

class ConnectionPool {
private:
    std::counting_semaphore<10> availableConnections{10};
    std::vector<Connection> connections;
    
public:
    Connection getConnection() {
        availableConnections.acquire();
        // Получить соединение из пула
        return connections[0];
    }
    
    void releaseConnection(Connection& conn) {
        // Вернуть соединение в пул
        availableConnections.release();
    }
};

3. Condition Variable

Позволяет потокам ждать определённого условия:

#include <condition_variable>
#include <queue>

std::mutex mtx;
std::condition_variable cv;
std::queue<int> dataQueue;
bool shutdown = false;

void producer() {
    for (int i = 0; i < 10; ++i) {
        {
            std::lock_guard<std::mutex> lock(mtx);
            dataQueue.push(i);
        }
        cv.notify_one(); // Уведомить потребителя
    }
}

void consumer() {
    while (true) {
        std::unique_lock<std::mutex> lock(mtx);
        
        // Ждать, пока очередь не будет заполнена или не произойдёт выключение
        cv.wait(lock, [] { return !dataQueue.empty() || shutdown; });
        
        if (dataQueue.empty() && shutdown) break;
        
        if (!dataQueue.empty()) {
            int data = dataQueue.front();
            dataQueue.pop();
            // Обработать данные
            std::cout << "Consumed: " << data << std::endl;
        }
    }
}

Высокоуровневые конструкции

4. Read-Write Lock (RWLock)

Позволяет одновременно читать несколько потокам, но исключает одновременное писание:

#include <shared_mutex>

class CacheWithRWLock {
private:
    std::unordered_map<std::string, std::string> cache;
    mutable std::shared_mutex cacheMtx;
    
public:
    std::string get(const std::string& key) const {
        std::shared_lock<std::shared_mutex> lock(cacheMtx); // Читающая блокировка
        auto it = cache.find(key);
        return it != cache.end() ? it->second : "";
    }
    
    void set(const std::string& key, const std::string& value) {
        std::unique_lock<std::shared_mutex> lock(cacheMtx); // Писающая блокировка
        cache[key] = value;
    }
};

// Использование
CacheWithRWLock cache;

// Много читателей одновременно (нет конфликтов)
std::thread reader1([&cache]() {
    for (int i = 0; i < 1000; ++i) {
        cache.get("key");
    }
});

// Только один писатель
std::thread writer([&cache]() {
    cache.set("key", "value");
});

5. Barrier (Барьер)

Синхронизирует несколько потоков в определённой точке:

#include <barrier>

std::barrier<> startBarrier(3); // Ждёт 3 потока

void threadWork(int id) {
    std::cout << "Thread " << id << " ready" << std::endl;
    startBarrier.arrive_and_wait(); // Подождать все потоки
    std::cout << "Thread " << id << " started" << std::endl;
}

int main() {
    std::thread t1(threadWork, 1);
    std::thread t2(threadWork, 2);
    std::thread t3(threadWork, 3);
    
    t1.join();
    t2.join();
    t3.join();
}

Практические примеры из реальных проектов

Потокобезопасный логгер:

class ThreadSafeLogger {
private:
    std::mutex logMutex;
    std::ofstream logFile;
    
public:
    ThreadSafeLogger(const std::string& filename) {
        logFile.open(filename, std::ios::app);
    }
    
    void log(const std::string& message) {
        std::lock_guard<std::mutex> lock(logMutex);
        auto now = std::chrono::system_clock::now();
        logFile << "[" << now.time_since_epoch().count() << "] " << message << std::endl;
    }
};

Потокобезопасная очередь обработки задач:

template<typename T>
class ThreadSafeQueue {
private:
    mutable std::mutex queueMutex;
    std::condition_variable notEmpty;
    std::queue<T> data;
    
public:
    void push(T value) {
        {
            std::lock_guard<std::mutex> lock(queueMutex);
            data.push(value);
        }
        notEmpty.notify_one();
    }
    
    bool tryPop(T& value) {
        std::unique_lock<std::mutex> lock(queueMutex);
        notEmpty.wait(lock, [this] { return !data.empty(); });
        value = data.front();
        data.pop();
        return true;
    }
};

Пул потоков с синхронизацией:

class ThreadPool {
private:
    std::vector<std::thread> workers;
    ThreadSafeQueue<std::function<void()>> taskQueue;
    std::atomic<bool> shouldStop{false};
    
public:
    ThreadPool(size_t numWorkers) {
        for (size_t i = 0; i < numWorkers; ++i) {
            workers.emplace_back([this]() {
                while (!shouldStop) {
                    std::function<void()> task;
                    if (taskQueue.tryPop(task)) {
                        task();
                    }
                }
            });
        }
    }
    
    void submit(std::function<void()> task) {
        taskQueue.push(task);
    }
};

Распространённые ошибки синхронизации

1. Deadlock (взаимная блокировка):

// ПЛОХО - может привести к deadlock'у
std::mutex mtx1, mtx2;

thread1: {
    lock(mtx1);
    lock(mtx2); // Ожидание
}

thread2: {
    lock(mtx2);
    lock(mtx1); // Ожидание - DEADLOCK!
}

// ХОРОШО - консистентный порядок блокировки
thread1: lock(mtx1); lock(mtx2);
thread2: lock(mtx1); lock(mtx2);

2. Race condition:

// ПЛОХО
int value = 0;
void increment() {
    int temp = value;    // Прочитать
    temp++;              // Модифицировать
    value = temp;        // Записать
    // Между операциями может работать другой поток!
}

// ХОРОШО
int value = 0;
std::mutex mtx;
void increment() {
    std::lock_guard<std::mutex> lock(mtx);
    value++;
}

Заключение

В моей практике я работал с:

• Мьютексами для эксклюзивного доступа
• Семафорами для управления ресурсами
• Condition Variables для синхронизации событий
• RWLocks для оптимизации чтения
• Barriers для координации потоков

Выбор правильного примитива синхронизации критичен для производительности и надёжности многопоточного кода. Нужно понимать компромиссы каждого инструмента и избегать распространённых ошибок, таких как deadlock'и и race conditions.