C++

Переменные состояния и производитель-потребитель

Используйте переменные состояния для эффективной координации потоков. Постройте безопасную очередь производителя-потребителя, обработайте ложные пробуждения и избегайте потерянной проблемы пробуждения.

Координация потока с переменными состояния

переменная состояния позволяет потоку подождать, пока какое-то условие не станет истинным, без загруженности. Вместо вращения while (!ready) { /* burn CPU */ } нить спит и пробуждается другой нитью, которая называется notify_one() или notify_all().

std::condition_variable работает только с std::unique_lock. Рисунок всегда:

1. Заблокируйте мутекс.
2. Проверьте состояние — если оно ложное, позвоните в wait(), который атомарно высвобождает мутекс и приостанавливает поток.
3. При уведомлении (и условие верно), mutex повторно приобретается и исполнение продолжается.

Уведомляющий поток изменяет общее состояние под тем же mutex, затем вызывает notify_one() или notify_all().

Базовое условие переменное использование

Вот фундаментальная закономерность: один поток ждет данных, другой их предоставляет.

condition_variable_basic.cpp
#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <string>

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

void consumer() {
    std::unique_lock<std::mutex> lock(mtx);
    // Wait until ready is true (handles spurious wakeups)
    cv.wait(lock, []{ return ready; });
    std::cout << "Consumer received: " << data << "\n";
}

void producer() {
    {
        std::lock_guard<std::mutex> lock(mtx);
        data = "Hello from producer";
        ready = true;
    }  // unlock before notify to avoid waking thread into contention
    cv.notify_one();
}

int main() {
    std::thread t1(consumer);
    std::thread t2(producer);
    t1.join();
    t2.join();
    return 0;
}
Подводный камень

Переменная состояния может разбудить поток , даже если уведомление не было отправлено — это называется ложным пробуждением . Это разрешено стандартом C++ (и распространено в реальных системах) по причинам производительности. Вы должны всегда использовать форму предиката wait() или обернуть его в цикл while:

// CORRECT — predicate form handles spurious wakeups
cv.wait(lock, []{ return ready; });

// EQUIVALENT manual loop
while (!ready) {
    cv.wait(lock);
}

// WRONG — wakes up randomly, proceeds without data
cv.wait(lock);  // no condition check!

Никогда не звоните в cv.wait(lock) без сказуемого или окружающего цикла while.

Thread-Safe Bounded Buffer (Производитель-потребитель)

Модель производитель-потребитель является одной из наиболее распространенных моделей параллелизма. Производители добавляют товары в общий буфер; потребители удаляют их. Ограниченный буфер блокирует производителей, когда они заполнены, и потребителей, когда они пусты.

bounded_queue.cpp
#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <queue>
#include <vector>

template <typename T>
class BoundedQueue {
    std::queue<T> queue_;
    std::mutex mtx_;
    std::condition_variable not_full_;
    std::condition_variable not_empty_;
    size_t capacity_;
public:
    explicit BoundedQueue(size_t cap) : capacity_(cap) {}

    void push(T item) {
        std::unique_lock<std::mutex> lock(mtx_);
        not_full_.wait(lock, [this]{ return queue_.size() < capacity_; });
        queue_.push(std::move(item));
        not_empty_.notify_one();
    }

    T pop() {
        std::unique_lock<std::mutex> lock(mtx_);
        not_empty_.wait(lock, [this]{ return !queue_.empty(); });
        T item = std::move(queue_.front());
        queue_.pop();
        not_full_.notify_one();
        return item;
    }
};

int main() {
    BoundedQueue<int> bq(5);  // capacity 5

    // Producer
    std::thread producer([&bq]() {
        for (int i = 0; i < 20; ++i) {
            bq.push(i);
            std::cout << "Produced: " << i << "\n";
        }
    });

    // Consumer
    std::thread consumer([&bq]() {
        for (int i = 0; i < 20; ++i) {
            int val = bq.pop();
            std::cout << "Consumed: " << val << "\n";
        }
    });

    producer.join();
    consumer.join();
    return 0;
}

Проблема потерянного пробуждения

Потерянное пробуждение происходит, когда notify_one() называется , прежде чем ** потребитель звонит wait(). Поскольку нить не ждет, уведомление теряется — потребитель спит вечно.

Стандартная схема предотвращает это, проверяя предикат перед сном. Если производитель уже установил условие истинности, предикат возвращается истинным, а wait() возвращается немедленно без сна. Вот почему форма предиката cv.wait(lock, pred) имеет важное значение — она проверяет состояние до и после каждого пробуждения:

1. Заблокируйте мутекс.
2.Проверить предикат — если уже верно, пропустить ожидание.
3. В противном случае, атомарно разблокировать и спать.
4. Пробуждение, повторная блокировка и повторная проверка предиката.

До тех пор, пока производитель и потребитель защищают общий флаг с тем же самым мутексом, никакое пробуждение не может быть потеряно.

Notify one против Notify all

notify_one() пробуждает один ожидающий поток; notify_all() пробуждает всех из них. Используйте notify_all(), когда нескольким потокам может потребоваться перепроверить их условия (например, сигналы отключения) и notify_one(), когда должен протекать ровно один поток (например, один потребитель, подбирающий задачу).

notify_all_shutdown.cpp
#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <vector>
#include <atomic>

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

void worker(int id) {
    std::unique_lock<std::mutex> lock(mtx);
    cv.wait(lock, []{ return shutdown; });
    std::cout << "Worker " << id << " shutting down\n";
}

int main() {
    std::vector<std::thread> workers;
    for (int i = 0; i < 5; ++i) {
        workers.emplace_back(worker, i);
    }

    std::this_thread::sleep_for(std::chrono::seconds(1));

    {
        std::lock_guard<std::mutex> lock(mtx);
        shutdown = true;
    }
    cv.notify_all();  // wake ALL workers for shutdown

    for (auto& w : workers) w.join();
    return 0;
}
Лучшая практика

Всегда используйте предикатную перегрузку wait() для защиты от ложных пробуждений и потерянных пробуждений. Разблокируйте mutex **, прежде чем звонить notify_one()/notify_all(), когда это возможно — это позволяет немедленно блокировать проснувшуюся нить на mutex. Предпочитаете notify_one(), когда только один поток должен проснуться, и notify_all() для сценариев вещания, таких как отключение. Никогда не ждите переменной состояния с lock_guard — вам нужен unique_lock.

Ключевые выводы
  • std::condition_variable позволяет нитям спать до тех пор, пока состояние не станет истинным, избегая дорогостоящего ожидания.
  • Всегда используйте форму предиката wait() для обработки ложных и потерянных пробуждений.
  • Модель «производитель-потребитель» использует две переменные состояния: «не полный» и «не пустой».
  • notify_one() будит одного официанта; notify_all() будит всех официантов - используйте правильный для вашего сценария
  • И официант, и уведомитель должны использовать тот же mutex для защиты общего состояния.

Тест — проверьте свои знания

(20 XP)

1. Почему вы всегда должны использовать предикат с `condition_variable::wait()`?

2. В ограниченной очереди «производитель-потребитель» что должен делать производитель, когда очередь заполнена?

3. Какой тип блокировки требуется `std::condition_variable::wait()`?

Назад
Мутексы, замки и критические секции
Далее
std::async, std::future и std::promise