条件变量、有界队列与关闭协议

#include <condition_variable>
#include <deque>
#include <iostream>
#include <mutex>
#include <optional>
#include <stdexcept>
#include <thread>
class Queue {
    std::mutex m; std::condition_variable available,space,consumer_state;
    std::deque<int> data; bool closed=false,consumer_waiting=false;
public:
    bool push(int x) {
        std::unique_lock lock(m);
        space.wait(lock,[&]{return closed || data.size()<3;});
        if(closed) return false;
        data.push_back(x); lock.unlock(); available.notify_one(); return true;
    }
    std::optional<int> pop() {
        std::unique_lock lock(m);
        consumer_waiting=!closed && data.empty();
        if(consumer_waiting) consumer_state.notify_all();
        available.wait(lock,[&]{return closed || !data.empty();});
        consumer_waiting=false;
        if(data.empty()) return std::nullopt;
        int x=data.front(); data.pop_front(); lock.unlock(); space.notify_one(); return x;
    }
    void close() {
        {std::lock_guard lock(m); closed=true;}
        available.notify_all(); space.notify_all();
    }
    // Test hook for this one-consumer example: observe wait while holding the same mutex.
    void wait_for_consumer_for_test() {
        std::unique_lock lock(m);
        consumer_state.wait(lock,[&]{return consumer_waiting;});
    }
};
struct InjectedFailure {};
void consume(int& sum,bool& finished,bool inject_failure=false) {
    Queue q; std::thread consumer;
    struct CloseAndJoin {
        Queue& queue; std::thread& thread;
        ~CloseAndJoin() {queue.close(); if(thread.joinable()) thread.join();}
    } close_and_join{q,consumer};
    consumer=std::thread([&]{while(auto x=q.pop()) sum+=*x; finished=true;});
    if(inject_failure) {
        q.wait_for_consumer_for_test();
        // Explicit producer failure injection; no real allocation failure is claimed.
        throw InjectedFailure{};
    }
    for(int i=1;i<=100;++i) if(!q.push(i)) throw std::runtime_error("unexpected close");
    q.close(); q.close(); consumer.join();
    if(sum!=5050 || q.push(101) || q.pop()) throw std::runtime_error("queue contract");
}
int main() {
    int sum=0; bool finished=false; consume(sum,finished);
    if(!finished) throw std::runtime_error("consumer completion");
    int empty_sum=0; bool empty_finished=false,caught=false;
    try {consume(empty_sum,empty_finished,true);} catch(const InjectedFailure&) {caught=true;}
    if(!caught || !empty_finished || empty_sum!=0) throw std::runtime_error("close before join");
    std::cout<<"sum="<<sum<<'\n';
}

#include <condition_variable>
#include <iostream>
#include <mutex>
#include <stdexcept>
#include <thread>
int main() {
    std::mutex m; std::condition_variable space,entered;
    bool closed=false,waiting=false,rejected=false; int slot=7;
    std::thread producer([&] {
        std::unique_lock lock(m); waiting=true; entered.notify_one();
        space.wait(lock,[&]{return closed || slot==0;});
        if(closed) rejected=true; else slot=9;
    });
    {
        std::unique_lock lock(m); entered.wait(lock,[&]{return waiting;}); closed=true;
    }
    space.notify_all(); producer.join();
    if(!rejected || slot!=7) throw std::runtime_error("shutdown");
    std::cout<<"push_rejected="<<rejected<<" retained="<<slot<<'\n';
}