Eventuali problemi evidenti o miglioramenti per la coda dei consumatori del produttore

Question

Ho chiesto a previous question del codice produttore/consumatore che era eccessivamente generico (anche se le risposte erano certamente utili). Quindi ho preso i suggerimenti da un earlier SO question da un altro autore e li ho convertiti in C++ e boost. Tuttavia sono sempre un po 'preoccupato del codice multithreaded, quindi se qualcuno può vedere miglioramenti evidenti mi piacerebbe sentirli.

#include <pthread.h> 
#include <deque> 
#include <iostream> 

#include "boost/thread.hpp" 


class MyQueue 
{ 
protected: 
    boost::mutex mutex_; 
    boost::condition_variable condition_; 
    bool cancel_; 
    std::deque<int> data_; 

public: 
    MyQueue() : mutex_(), condition_(), cancel_(false), data_() 
    { 
    } 

    struct Canceled{}; 

    void push(int i) 
    { 
    boost::lock_guard<boost::mutex> l(mutex_); 
    if(cancel_) throw Canceled(); 
    data_.push_back(i); 
    condition_.notify_all(); 
    } 

    void pop(int & i) 
    { 
    boost::unique_lock<boost::mutex> l(mutex_); 
    while(! cancel_ && data_.size()==0) 
    { 
     condition_.wait(l); 
    } 
    if(cancel_) throw Canceled(); 

    assert(data_.size() != 0); 
    i = data_.front(); 
    data_.pop_front(); 
    } 

    void cancel() 
    { 
    boost::lock_guard<boost::mutex> l(mutex_); 
    if(cancel_) throw Canceled(); 
    cancel_ = true; 
    condition_.notify_all(); 
    } 
}; 


boost::mutex iomutex; 

void producer(MyQueue * q, const std::string & name) 
try 
{ 
    for(unsigned int i=0 ; i<20; ++i) 
    { 
    q->push(i); 
    boost::lock_guard<boost::mutex> l(iomutex); 
    std::cout<<name<<" PRODUCED "<<i<<std::endl; 
    } 

    sleep(1); 
    q->cancel(); 
    { 
    boost::lock_guard<boost::mutex> l(iomutex); 
    std::cout<<name<<" PRODUCER EXITING NORMALLY"<<std::endl; 
    } 
} 
catch(MyQueue::Canceled & c) 
{ 
    boost::lock_guard<boost::mutex> l(iomutex); 
    std::cout<<name<<" PRODUCER CANCLED "<<std::endl; 
} 

void consumer(MyQueue * q, const std::string & name) 
try 
{ 
    while(true) 
    { 
    int i; 
    q->pop(i); 
    boost::lock_guard<boost::mutex> l(iomutex); 
    std::cout<<name<<" CONSUMED "<<i<<std::endl; 
    } 
} 
catch(MyQueue::Canceled & c) 
{ 
    boost::lock_guard<boost::mutex> l(iomutex); 
    std::cout<<name<<" CONSUMER CANCLED "<<std::endl; 
} 

int main() 
{ 
    MyQueue q; 
    boost::thread pr1(producer, &q, "pro1"); 
    boost::thread pr2(producer, &q, "pro2"); 
    boost::thread cons1(consumer, &q, "con1"); 
    boost::thread cons2(consumer, &q, "con2"); 

    pr1.join(); 
    pr2.join(); 
    cons1.join(); 
    cons2.join(); 
} 

UPDATE: ho finito per usare una versione modificata di Anthony Williams' concurrent queue. La mia versione modificata può essere trovata here.

Answer 1

Se siete preoccupati per potenziali insidie ​​nella vostra implementazione, potete provare ad usare Anthony Williams (manutentore della libreria Boost.Thread) eccellente thread-safe, multiple-producer, multiple-consumer queue.