mitlib.pub/MITLib/Utility/ConcurrentQueue.h

#pragma once
#include <afx.h>
#include <list>
#include "thread.h"
#include "Utility.h"

#define DEFAULT_CAPACITY 1000

template <typename T> class CConcurrentQueue
{
public:
	CConcurrentQueue(size_t capacity = DEFAULT_CAPACITY) : m_capacity(capacity)
	{
		m_spaceAvailableEvent.Set();
	}

	// Destroy the queue and all it's content
	~CConcurrentQueue(void)
	{
		while(!m_queue.empty()) {
			m_queue.pop_front();
		}
	}

	// Pushes object T when space is available
	// Throws std::exception on timeout or when interrupted
	bool Push(T item, int milliSeconds = INFINITE) {
		bool bPushedItem = false;
		CSingleLock lock(&m_mutex);
		while(!bPushedItem) {
			try
			{
				// Wait for space to be available
				m_spaceAvailableEvent.WaitEvtEx(milliSeconds);
			} catch(std::exception &e) {
				printf("Push failed: %s\n", e.what());
				throw;
			}

			// wait for critical section to become available
			lock.Lock();

			// inside critical section
			{
				size_t size = m_queue.size();
				// Make sure that nobody else that was waiting pushed first
				if( size < m_capacity) {
					m_queue.push_back(item);
					bPushedItem = true;

					if (size == 0)
					{
						// Queue was empty, now there is one item
						m_dataAvailableEvent.Set();
					}

					size++;
					if (size >= m_capacity) {
						// Queue is now full
						m_spaceAvailableEvent.Reset();
					}
				}
			}
			lock.Unlock();
		}
		return true;
	}

	// Pushes object T when space is available
	// Throws std::exception on timeout or when interrupted
	bool PushFront(T item, int milliSeconds = INFINITE) {
		bool bPushedItem = false;
		CSingleLock lock(&m_mutex);
		while(!bPushedItem) {
			try
			{
				// Wait for space to be available
				m_spaceAvailableEvent.WaitEvtEx(milliSeconds);
			} catch(std::exception &e) {
				printf("Push failed: %s\n", e.what());
				throw;
			}

			// wait for critical section to become available
			lock.Lock();

			// inside critical section
			{
				size_t size = m_queue.size();
				// Make sure that nobody else that was waiting pushed first
				if( size < m_capacity) {
					m_queue.push_front(item);
					bPushedItem = true;

					if (size == 0)
					{
						// Queue was empty, now there is one item
						m_dataAvailableEvent.Set();
					}

					size++;
					if (size >= m_capacity) {
						// Queue is now full
						m_spaceAvailableEvent.Reset();
					}
				}
			}
			lock.Unlock();
		}
		return true;
	}

	// Returns object T when available
	// Throws std::exception on timeout or when interrupted
	T Pop(int milliSeconds = INFINITE)
	{
		T item;
		PopEx(item, milliSeconds, true);
		return item;
	}

	// Returns object T when available
	// Throws std::exception on timeout or when interrupted
	bool Pop(T &item, int milliSeconds = INFINITE)
	{
		return PopEx(item, milliSeconds, false);
	}

	int Size()
	{
		int size;
		// wait for mutex to become available
		CSingleLock lock(&m_mutex, true);

		// inside critical section
		{
			size = m_queue.size();
		}

		return size;
	}
private:
	std::list<T> m_queue;
	CMutex m_mutex;
	CEvt m_dataAvailableEvent;
	CEvt m_spaceAvailableEvent;
	size_t m_capacity;

	bool PopEx(T &item, int milliSeconds = INFINITE, bool bThrowException = true)
	{
		CHighResPerformanceCounter counter;
		counter.StartCnt();
		long remaining = milliSeconds;
		CSingleLock lock(&m_mutex);

		while (true)
		{
			// wait for event to be signaled that there is data
			if (bThrowException)
			{
				m_dataAvailableEvent.WaitEvtEx(remaining);
			}
			else if (!m_dataAvailableEvent.WaitEvt(remaining))
			{
				return false;
			}

			// wait for mutex to become available
			lock.Lock();

			// inside critical section
			{
				// try to deque something - it<69>s possible that the queue is empty because another
				// thread pre-empted before us and got the last item in the queue
				size_t size = m_queue.size();
				if (size > 0)
				{
					item = m_queue.front();
					m_queue.pop_front();

					if (size == 1) {
						m_dataAvailableEvent.Reset();
					}

					if (size <= m_capacity) {
						// There was no space, now there is
						m_spaceAvailableEvent.Set();
					}
					return true;
				}
			}
			lock.Unlock();
			remaining = counter.GetTimeRemaining(milliSeconds);
		}
	}
};