Что такое condition variable? Когда её использовать?

Что такое condition variable? Когда её использовать?

Краткий ответ

Condition variable (переменная условия, std::condition_variable) — это примитив синхронизации, который позволяет одному или нескольким потокам ждать, пока произойдёт определённое событие, которое сигнализирует другой поток. Это эффективнее, чем постоянная проверка флага в цикле (busy-waiting).

Проблема без condition variable

Неэффективный способ — busy-waiting:

bool ready = false;

// Поток 1 (производитель)
void producer() {
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
    ready = true;  // Сигнал о готовности
}

// Поток 2 (потребитель)
void consumer() {
    while (!ready) {
        // Постоянно проверяем флаг!
        // Это плохо: тратится CPU, высокая задержка
    }
    std::cout << "Data is ready!\n";
}

int main() {
    std::thread t1(producer);
    std::thread t2(consumer);
    t1.join();
    t2.join();
}

Проблемы:

• Потребляет CPU энергию впустую
• Большая задержка между сигналом и пробуждением
• Сложно оптимизировать

Решение: condition_variable

std::mutex mtx;
std::condition_variable cv;
bool ready = false;

// Поток 1 (производитель)
void producer() {
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
    {
        std::lock_guard<std::mutex> lock(mtx);
        ready = true;
    }
    cv.notify_one();  // Пробуждаем один ждущий поток
}

// Поток 2 (потребитель)
void consumer() {
    std::unique_lock<std::mutex> lock(mtx);
    cv.wait(lock, [] { return ready; });  // Ждём, пока ready не станет true
    std::cout << "Data is ready!\n";
}

int main() {
    std::thread t1(producer);
    std::thread t2(consumer);
    t1.join();
    t2.join();
}

Преимущества:

• Потребитель спит (не тратит CPU)
• Почти нет задержки между сигналом и пробуждением
• Очень эффективно

Как работает condition_variable

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

std::condition_variable cv;
std::mutex mtx;
std::unique_lock<std::mutex> lock(mtx);

// Ждём, пока условие не станет истинным
cv.wait(lock);           // Ждём без предиката
cv.wait(lock, predicate);  // Ждём с проверкой условия

// Пробуждаем один ждущий поток
cv.notify_one();

// Пробуждаем все ждущие потоки
cv.notify_all();

Как работает wait():

1. Захватываем мьютекс (важно!)
2. Проверяем предикат (условие)
3. Если условие истинно — return (продолжаем)
4. Если условие ложно:
   a. Отпускаем мьютекс
   b. Входим в режим ожидания (спим, не тратим CPU)
5. Когда другой поток вызывает notify_one/notify_all:
   a. Пробуждаемся
   b. Заново захватываем мьютекс
   c. Проверяем условие заново (spurious wakeup!)
   d. Если условие истинно — return
   e. Если ложно — снова ждём

Важное: захват мьютекса

ОШИБКА: забыли захватить мьютекс перед wait()

// Неправильно!
void consumer() {
    cv.wait(lock);  // Нет! Нужно захватить мьютекс
}

Правильно:

void consumer() {
    std::unique_lock<std::mutex> lock(mtx);
    cv.wait(lock, [] { return ready; });
    // После wait() мьютекс снова захвачен!
}

Почему нужен мьютекс:

1. Защита от race condition между проверкой условия и wait()
2. Мьютекс гарантирует, что:
   - Значение ready безопасно читать/писать
   - Нет потери сигналов

Spurious wakeup (ложные пробуждения)

Проблема: поток может проснуться БЕЗ вызова notify()!

void consumer() {
    std::unique_lock<std::mutex> lock(mtx);
    
    // НЕПРАВИЛЬНО: может проснуться без notify()
    cv.wait(lock);
    std::cout << "Data ready!\n";
    // Но данные могут быть не готовы!
}

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

void consumer() {
    std::unique_lock<std::mutex> lock(mtx);
    
    // Проверяем условие при каждом пробуждении
    cv.wait(lock, [] { return ready; });
    std::cout << "Data ready!\n";
    // Гарантировано, что ready == true
}

Предикат — это очень важно!

Практический пример: очередь с потоками

template <typename T>
class ThreadSafeQueue {
private:
    std::queue<T> q;
    std::mutex mtx;
    std::condition_variable cv;

public:
    // Добавить элемент в очередь
    void push(const T& value) {
        {
            std::lock_guard<std::mutex> lock(mtx);
            q.push(value);
        }
        cv.notify_one();  // Пробудить один потребитель
    }

    // Получить элемент из очереди (ждём, если пусто)
    T pop() {
        std::unique_lock<std::mutex> lock(mtx);
        
        // Ждём, пока очередь не станет непустой
        cv.wait(lock, [this] { return !q.empty(); });
        
        T value = q.front();
        q.pop();
        return value;
    }
    
    // Неблокирующее получение (возвращает false, если пусто)
    bool try_pop(T& value) {
        std::lock_guard<std::mutex> lock(mtx);
        if (q.empty()) return false;
        value = q.front();
        q.pop();
        return true;
    }
};

int main() {
    ThreadSafeQueue<int> queue;
    
    // Производители
    std::thread producer1([&queue] {
        for (int i = 0; i < 5; ++i) {
            queue.push(i);
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
        }
    });
    
    // Потребитель
    std::thread consumer([&queue] {
        for (int i = 0; i < 5; ++i) {
            int value = queue.pop();  // Ждёт, если пусто
            std::cout << "Got: " << value << "\n";
        }
    });
    
    producer1.join();
    consumer.join();
}

Ожидание с таймаутом

std::unique_lock<std::mutex> lock(mtx);

// Ждём максимум 1 секунду
std::cv_status status = cv.wait_for(lock, std::chrono::seconds(1));

if (status == std::cv_status::ready) {
    std::cout << "Condition was notified\n";
} else {
    std::cout << "Timeout occurred\n";
}

// Ждём до определённого времени
auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(5);
std::cv_status status = cv.wait_until(lock, deadline);

notify_one vs notify_all

notify_one():

void producer() {
    {
        std::lock_guard<std::mutex> lock(mtx);
        data_ready = true;
    }
    cv.notify_one();  // Пробудить ОДИН поток
}

// Используй когда:
// - Только один потребитель ждёт
// - Ресурс может обработать только одного
// - Хочешь эффективности

notify_all():

void producer() {
    {
        std::lock_guard<std::mutex> lock(mtx);
        data_ready = true;
    }
    cv.notify_all();  // Пробудить ВСЕ потоки
}

// Используй когда:
// - Много потребителей ждут
// - Изменение влияет на всех
// - Изменения требуют переоценки условия

Когда использовать condition_variable

Идеально для:

1. Очереди между потоками:

// Производитель добавляет, потребитель берёт
ThreadSafeQueue<Task> tasks;
tasks.push(task1);
Task t = tasks.pop();  // Ждёт, если пусто

1. Синхронизация стартов:

std::condition_variable ready_cv;
bool all_ready = false;

void worker() {
    std::unique_lock<std::mutex> lock(mtx);
    ready_cv.wait(lock, [] { return all_ready; });
    // Все потоки начинают одновременно
}

1. Ожидание завершения:

std::condition_variable done_cv;
bool work_done = false;

void waiter() {
    std::unique_lock<std::mutex> lock(mtx);
    done_cv.wait(lock, [] { return work_done; });
    std::cout << "Work is done!\n";
}

Когда НЕ использовать

Когда есть проще:

// Не используй для простых случаев
if (some_condition) {
    // Просто if, не нужна синхронизация
}

// Используй для синхронизации между потоками
cv.wait(lock, [this] { return some_condition; });

Альтернативы:

• std::promise и std::future — для one-time результатов
• std::latch (C++20) — для простого синхронизирующего окончание
• std::barrier (C++20) — для точек синхронизации
• std::semaphore (C++20) — для счётных сигналов

Типичные ошибки

1. Забыли мьютекс:

// ОШИБКА!
cv.wait(lock);  // lock не захвачен!

2. Забыли предикат:

// ОШИБКА! Может проснуться без причины
cv.wait(lock);  // Нет проверки условия

3. Забыли notify:

// ОШИБКА! Потребитель будет ждать вечно
ready = true;
// cv.notify_one();  // Забыли!

4. Используют lock_guard с wait:

// ОШИБКА! lock_guard не может быть захвачен/отпущен во время wait
std::lock_guard<std::mutex> lock(mtx);
cv.wait(lock);  // Не скомпилируется!

// Правильно:
std::unique_lock<std::mutex> lock(mtx);
cv.wait(lock);

Заключение

std::condition_variable — это мощный примитив для эффективной синхронизации потоков:

• Позволяет потокам спать вместо busy-waiting
• Требует мьютекса для безопасности
• Нужен предикат для защиты от ложных пробуждений
• Основа для многих паттернов (очереди, barriers, latches)

Это незаменимый инструмент для параллельного программирования на C++!