SeniorКодЧастоЕщё не отвечали

Реализуйте producer-consumer с использованием condition variables.

Реализуйте ограниченную блокирующую очередь для паттерна producer-consumer плюс корректный путь завершения.

Требования:

один std::mutex и две condition variable: not_full (производители ждут, пока очередь полна) и not_empty (потребители ждут, пока она пуста)
close() устанавливает флаг done_ под блокировкой и будит всех ожидающих; меняйте done_ только удерживая мьютекс
потребители должны опустошить оставшиеся элементы перед выходом — предикат потребителя done_ && queue_.empty(), а не просто done_; сигнализируйте конец потока (например std::nullopt) после дренажа

template<typename T>
class BoundedQueue {
public:
    explicit BoundedQueue(std::size_t capacity);
    void push(T item);              // blocks while full
    std::optional<T> pop();         // blocks while empty; nullopt when drained+closed
    void close();
    // ваш код здесь
};

Допишите реализацию.

Ограниченная очередь под мьютексом с двумя CV: not_full (производитель ждёт при полной) и not_empty (потребитель ждёт при пустой). Грейсфул через флаг done_ — потребители дренируют по done_ && queue_.empty().

✗Использовать одну condition variable для 'не полный' и 'не пустой' — требует notify_all() и тратит CPU; две CV с notify_one() эффективнее
✗Вызывать notify_one() удерживая блокировку — технически корректно, но разбуженный поток сразу блокируется на мьютексе; рассмотрите уведомление после освобождения блокировки
✗Не обрабатывать гонку при завершении — если done_ устанавливается до проверки потребителями, последние элементы могут быть потеряны; предикат должен быть done_ && queue_.empty(), а не просто done_

→Как реализовать lock-free ограниченную очередь для producer-consumer?
→Что такое back-pressure и как его реализовать, когда потребитель медленнее производителя?

Оглавление

Producer-Consumer с condition variables

Реализация

#include <condition_variable>
#include <mutex>
#include <optional>
#include <queue>
#include <stdexcept>

template<typename T>
class BoundedQueue {
public:
    explicit BoundedQueue(std::size_t capacity) : capacity_(capacity) {}

    // Блокирует производителя, если очередь полна
    void push(T item) {
        std::unique_lock lock(mu_);
        not_full_.wait(lock, [this] {
            return queue_.size() < capacity_ || done_;
        });
        if (done_)
            throw std::runtime_error("push on closed queue");
        queue_.push(std::move(item));
        not_empty_.notify_one();
    }

    // Блокирует потребителя, если очередь пуста; возвращает nullopt при завершении
    std::optional<T> pop() {
        std::unique_lock lock(mu_);
        not_empty_.wait(lock, [this] {
            return !queue_.empty() || done_;
        });
        if (queue_.empty())
            return std::nullopt;  // очередь дренирована и done_
        T item = std::move(queue_.front());
        queue_.pop();
        not_full_.notify_one();
        return item;
    }

    // Сигнализирует о завершении; потребители заканчивают работу после дренажа
    void close() {
        {
            std::lock_guard lock(mu_);
            done_ = true;
        }
        not_full_.notify_all();
        not_empty_.notify_all();
    }

    bool is_closed() const {
        std::lock_guard lock(mu_);
        return done_;
    }

private:
    std::size_t             capacity_;
    std::queue<T>           queue_;
    mutable std::mutex      mu_;
    std::condition_variable not_full_;
    std::condition_variable not_empty_;
    bool                    done_ = false;
};

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

#include <atomic>
#include <cassert>
#include <thread>
#include <vector>

int main() {
    BoundedQueue<int> q(10);
    const int ITEMS = 100;
    const int PRODUCERS = 4;
    const int CONSUMERS = 3;

    std::atomic<int> total_consumed{0};

    // Запуск потребителей
    std::vector<std::thread> consumers;
    for (int i = 0; i < CONSUMERS; ++i) {
        consumers.emplace_back([&q, &total_consumed] {
            while (auto item = q.pop()) {
                ++total_consumed;
            }
        });
    }

    // Запуск производителей
    std::vector<std::thread> producers;
    std::atomic<int> produced{0};
    for (int i = 0; i < PRODUCERS; ++i) {
        producers.emplace_back([&q, &produced, ITEMS, PRODUCERS] {
            int per = ITEMS / PRODUCERS;
            for (int j = 0; j < per; ++j) {
                q.push(produced.fetch_add(1));
            }
        });
    }

    for (auto& t : producers) t.join();
    q.close();
    for (auto& t : consumers) t.join();

    assert(total_consumed.load() == ITEMS);
}

Анализ граничных случаев

Сценарий	Поведение
Полная очередь	Производитель ждёт на `not_full_`
Пустая очередь	Потребитель ждёт на `not_empty_`
`close()` при полной очереди	Производитель пробуждается, бросает исключение
`close()` при пустой очереди	Потребитель пробуждается, получает `nullopt`
`close()` с элементами в очереди	Потребители дренируют, затем получают `nullopt`

Открыть тему в роадмапе →

Оглавление

Реализуйте producer-consumer с использованием condition variables.

Реализуйте ограниченную блокирующую очередь для паттерна producer-consumer плюс корректный путь завершения.

Требования:

один std::mutex и две condition variable: not_full (производители ждут, пока очередь полна) и not_empty (потребители ждут, пока она пуста)
close() устанавливает флаг done_ под блокировкой и будит всех ожидающих; меняйте done_ только удерживая мьютекс
потребители должны опустошить оставшиеся элементы перед выходом — предикат потребителя done_ && queue_.empty(), а не просто done_; сигнализируйте конец потока (например std::nullopt) после дренажа

template<typename T>
class BoundedQueue {
public:
    explicit BoundedQueue(std::size_t capacity);
    void push(T item);              // blocks while full
    std::optional<T> pop();         // blocks while empty; nullopt when drained+closed
    void close();
    // ваш код здесь
};

Допишите реализацию.

Producer-Consumer с condition variables

Реализация

#include <condition_variable> #include <mutex> #include <optional> #include <queue> #include <stdexcept> template<typename T> class BoundedQueue { public: explicit BoundedQueue(std::size_t capacity) : capacity_(capacity) {} // Блокирует производителя, если очередь полна void push(T item) { std::unique_lock lock(mu_); not_full_.wait(lock, [this] { return queue_.size() < capacity_ || done_; }); if (done_) throw std::runtime_error("push on closed queue"); queue_.push(std::move(item)); not_empty_.notify_one(); } // Блокирует потребителя, если очередь пуста; возвращает nullopt при завершении std::optional<T> pop() { std::unique_lock lock(mu_); not_empty_.wait(lock, [this] { return !queue_.empty() || done_; }); if (queue_.empty()) return std::nullopt; // очередь дренирована и done_ T item = std::move(queue_.front()); queue_.pop(); not_full_.notify_one(); return item; } // Сигнализирует о завершении; потребители заканчивают работу после дренажа void close() { { std::lock_guard lock(mu_); done_ = true; } not_full_.notify_all(); not_empty_.notify_all(); } bool is_closed() const { std::lock_guard lock(mu_); return done_; } private: std::size_t capacity_; std::queue<T> queue_; mutable std::mutex mu_; std::condition_variable not_full_; std::condition_variable not_empty_; bool done_ = false; };

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

#include <atomic> #include <cassert> #include <thread> #include <vector> int main() { BoundedQueue<int> q(10); const int ITEMS = 100; const int PRODUCERS = 4; const int CONSUMERS = 3; std::atomic<int> total_consumed{0}; // Запуск потребителей std::vector<std::thread> consumers; for (int i = 0; i < CONSUMERS; ++i) { consumers.emplace_back([&q, &total_consumed] { while (auto item = q.pop()) { ++total_consumed; } }); } // Запуск производителей std::vector<std::thread> producers; std::atomic<int> produced{0}; for (int i = 0; i < PRODUCERS; ++i) { producers.emplace_back([&q, &produced, ITEMS, PRODUCERS] { int per = ITEMS / PRODUCERS; for (int j = 0; j < per; ++j) { q.push(produced.fetch_add(1)); } }); } for (auto& t : producers) t.join(); q.close(); for (auto& t : consumers) t.join(); assert(total_consumed.load() == ITEMS); }

Анализ граничных случаев

Сценарий	Поведение
Полная очередь	Производитель ждёт на `not_full_`
Пустая очередь	Потребитель ждёт на `not_empty_`
`close()` при полной очереди	Производитель пробуждается, бросает исключение
`close()` при пустой очереди	Потребитель пробуждается, получает `nullopt`
`close()` с элементами в очереди	Потребители дренируют, затем получают `nullopt`