I've implemented concurrent_blocking_queue
for multiple-producers and single consumer.
I intend it to block the producers as well, when the queue is full, which is why I set the capacity of the queue in the constructor. I've provided my own implementation of critical_section
, scoped_lock
and condition_variable
, using msdn primitives.
Please review it. I want to know if there is any potential deadlocks.
template<typename T>
class concurrent_blocking_queue : public noncopyable
{
std::queue<T> m_internal_queue;
mutable critical_section m_critical_section;
condition_variable m_queue_full;
condition_variable m_queue_empty;
const size_t m_capacity;
public:
concurrent_blocking_queue(size_t capacity)
: m_capacity(capacity), m_critical_section()
{
}
void put(T const & input)
{
framework::scoped_lock lock(m_critical_section); //lock
m_queue_full.wait(lock, [&]{ return !full(); });
m_internal_queue.push(input);
m_queue_empty.notify_all();
}
T take()
{
framework::scoped_lock lock(m_critical_section); //lock
m_queue_empty.wait(lock, [&]{ return !empty(); });
T output = m_internal_queue.front();
m_internal_queue.pop();
m_queue_full.notify_all();
return output;
}
bool full() const
{
framework::scoped_lock lock(m_critical_section);
if ( m_internal_queue.size() > m_capacity )
{
throw std::logic_error("size of concurrent_blocking_queue cannot be greater than the capacity.");
}
return m_internal_queue.size() == m_capacity;
}
bool empty() const
{
framework::scoped_lock lock(m_critical_section);
return m_internal_queue.empty();
}
//..
};
And the remaining classes are posted below. Note that I've implemented them in .h
and .cpp
files, but here I've merged them to make the post short and simple, and have removed the namespace for brevity, otherwise all classes are defined in framework
namespace.
critical_section derives publicly from CRITICAL_SECTION.
class critical_section : public CRITICAL_SECTION, public noncopyable { public: critical_section(void) { ::InitializeCriticalSection(this); } ~critical_section(void) { ::DeleteCriticalSection(this); } void lock() { ::EnterCriticalSection(this); } void unlock() { ::LeaveCriticalSection(this); } };
condition_variable derives privately from CONDITION_VARIABLE.
class condition_variable : private CONDITION_VARIABLE, public noncopyable { public: condition_variable() { ::InitializeConditionVariable(this); } ~condition_variable(void) {} void wait(scoped_lock & lock) { if (!::SleepConditionVariableCS (this, &lock.m_cs, INFINITE)) { DWORD error = ::GetLastError(); std::string msg = format("SleepConditionVariableCS() failed with error code = 0x%X.", error); throw std::runtime_error(msg); } } void wait(scoped_lock & lock, std::function<bool()> predicate) { while( !predicate() ) { wait(lock); } } void notify_one() { ::WakeConditionVariable(this); } void notify_all() { ::WakeAllConditionVariable(this); } };
scoped_lock
class scoped_lock : public noncopyable { friend class condition_variable; critical_section & m_cs; public: scoped_lock(critical_section & cs) : m_cs(cs) { m_cs.lock(); } ~scoped_lock(void) { m_cs.unlock(); } };
wait()
. \$\endgroup\$CRITICAL_SECTION
as per the MSDN. \$\endgroup\$m_queue_empty.notify_all()
if no other threads are waiting. If that behavior is allowed/supported by MSDN, then the rest looks ok to me. \$\endgroup\$