Better Blocking Queue

In the previous post, we discussed a multi-threaded blocking queue whose implementation was lacking: it was not exception safe. It left the semaphore counts wrongly modified in the presence of exceptions emitted by T‘s copy constructor or assignment operator.

Below is a corrected void push(const T& item) method:

void push(const T& item)
{
	m_openSlots.wait();
	{
		std::lock_guard<std::mutex> lock(m_cs);
		try
		{
			new (m_data + m_pushIndex) T (item);
		}
		catch (...)
		{
			m_openSlots.post();
			throw;
		}
		m_pushIndex = ++m_pushIndex % m_size;
		++m_count;
	}
	m_fullSlots.post();
}

In the corrected version, we first catch any exception possibly emitted by T‘s copy constructor, then adjust back the open-slot semaphore. Since the push operation failed, the number of open and full slots has not changed. Finally, we re-throw the exception.

Below is a corrected void pop(T& item) method:

void pop(T& item)
{
	m_fullSlots.wait();
	{
		std::lock_guard<std::mutex> lock(m_cs);
		try
		{
			item = m_data[m_popIndex];
		}
		catch (...)
		{
			m_fullSlots.post();
			throw;
		}
		m_data[m_popIndex].~T();
		m_popIndex = ++m_popIndex % m_size;
		--m_count;
	}
	m_openSlots.post();
}

In the corrected version, we first catch any exception possibly emitted by T‘s assignment operator, then adjust back the full-slot semaphore. Since the pop operation failed, the number of open and full slots did not change. Finally, we re-throw the exception.

We now have a robust queue template class that can handle misbehaving Ts.