/***************************************************************************
*                                                                          *
*  Copyright (C) Golaem S.A.  All Rights Reserved.                         *
*                                                                          *
***************************************************************************/

#pragma once
#ifndef GLMCONCURRENTDATASTRUCTURES_H
#define GLMCONCURRENTDATASTRUCTURES_H

#include "glmAtomic.h"
#include "glmAssert.h"
#include "glmMathBase.h"

#include <stdint.h>

namespace glm
{
	//Single producer Single consumer "ring buffer" (or "bounded queue")
	//NOTE: implementation adapted from http://www.boost.org/doc/libs/1_53_0/doc/html/atomic/usage_examples.html#boost_atomic.usage_examples.example_ringbuffer.implementation

	template<typename T>
	class SPSC_RingBuffer
	{
	public:
		SPSC_RingBuffer()
			: _buffer(NULL), _bufferMask(0)
		{
			atomicStoreRelaxed(_enqueueHead, 0);
			atomicStoreRelaxed(_dequeueTail, 0);
		}
		~SPSC_RingBuffer() {}

		bool initialize(void* buffer, size_t bufferSize, size_t elementCountMax)
		{
			GLM_ASSERT( buffer != NULL );
			GLM_ASSERT( bufferSize > 0 && bufferSize == getBufferSizeToAllocate(elementCountMax) );
			GLM_ASSERT( elementCountMax >= 2 && GLM_MATH_IS_POWER_OF_TWO(elementCountMax) && elementCountMax < GLM_UINT32_MAX );

			atomicStoreRelaxed(_enqueueHead, 0);
			atomicStoreRelaxed(_dequeueTail, 0);

			//NOTE: queue starts empty, no need to call constructor on all buffer elements
			_buffer = static_cast<T*>(buffer);
			_bufferMask = static_cast<uint32_t>(elementCountMax - 1);

			return true;
		}

		void uninitialize()
		{
			//IMPORTANT: NOT multi-thread SAFE, make sure push() & pop() calls from all threads are finished

			//NOTE: SPSC_RingBuffer didn't allocate buffer, shouldn't be responsible for freeing the memory
			_buffer = NULL;
			_bufferMask = 0;
		}

		bool try_push(const T & value)
		{
			bool wasFullQueue = true;

			//size_t head = head_.load(boost::memory_order_relaxed);
			const atomic32_t::value_t head = atomicLoadRelaxed(_enqueueHead);

			//size_t next_head = next(head);
			const atomic32_t::value_t next_head = next(head);

			//if (next_head != tail_.load(boost::memory_order_acquire))
			const atomic32_t::value_t tail = atomicLoadAcquire(_dequeueTail);
			if ( next_head != tail )
			{
				wasFullQueue = false;

				//ring_[head] = value;
				_buffer[head] = value;

				//head_.store(next_head, boost::memory_order_release);
				atomicStoreRelease(_enqueueHead, next_head);
			}
			
			const bool wasPushed = !wasFullQueue;
			return wasPushed;
		}

		bool try_pop(T & value)
		{
			bool wasEmptyQueue = true;

			//size_t tail = tail_.load(boost::memory_order_relaxed);
			const atomic32_t::value_t tail = atomicLoadRelaxed(_dequeueTail);

			//if (tail != head_.load(boost::memory_order_acquire))
			const atomic32_t::value_t head = atomicLoadAcquire(_enqueueHead);
			if ( tail != head )
			{
				wasEmptyQueue = false;

				//value = ring_[tail];
				value = _buffer[tail];

				const atomic32_t::value_t next_tail = next(tail);

				//tail_.store(next(tail), boost::memory_order_release);
				atomicStoreRelease(_dequeueTail, next_tail);
			}

			const bool wasPoped = !wasEmptyQueue;
			return wasPoped;
		}

		const bool wasInitialized() const
		{
			return (_buffer != NULL);
		}

		const bool wasEmpty() const
		{
			return ( atomicLoadRelaxed(_dequeueTail) == atomicLoadRelaxed(_enqueueHead) );
		}

		static size_t getBufferSizeToAllocate(size_t elementCountMax)
		{
			GLM_ASSERT( elementCountMax >= 2 );

			//NOTE: MPMC_RingBuffer current implementation only supports 32 bit
			const bool countMaxIsSupported = (elementCountMax < GLM_UINT32_MAX);
			GLM_DEBUG_ASSERT_MSG( countMaxIsSupported, "SPSC_RingBuffer current implementation only supports 32 bit" );

			//NOTE: restriction to speed up next element computation using 'bitmask' instead of 'modulo' operation
			const bool countMaxIsPowerOfTwo = GLM_MATH_IS_POWER_OF_TWO(elementCountMax);
			GLM_DEBUG_ASSERT_MSG( countMaxIsPowerOfTwo, "To speed up some computation, buffer element count max is restricted to a power of two number." );

			const size_t allocationSize = ( countMaxIsSupported && countMaxIsPowerOfTwo ) ? sizeof(T) * elementCountMax : 0;
			return allocationSize;
		}

	private:
		GLM_DISALLOW_COPY_AND_ASSIGN(SPSC_RingBuffer)

		atomic32_t::value_t next(atomic32_t::value_t current)
		{
			return (current + 1) & _bufferMask;
		}

		T*				_buffer;
		uint32_t		_bufferMask;
		atomic32_t		_enqueueHead;
		atomic32_t		_dequeueTail;

		//T ring_[Size];
		//boost::atomic<size_t> head_, tail_;
	};


	//Multi producers Multi consumers "ring buffer" (or "bounded queue")
	//NOTE: implementation adapted from http://www.1024cores.net/home/lock-free-algorithms/queues/bounded-mpmc-queue

	template<typename T>
	class MPMC_RingBuffer
	{
	public:
		struct element_t
		{
			atomic32_t	_sequence;
			T			_data;
		};

		MPMC_RingBuffer()
			: _buffer(NULL), _bufferMask(0)
		{
			atomicStoreRelaxed(_enqueueHead, 0);
			atomicStoreRelaxed(_dequeueTail, 0);
		}
		~MPMC_RingBuffer() {}

		bool initialize(void* buffer, size_t bufferSize, size_t elementCountMax)
		{
			GLM_ASSERT( buffer != NULL );
			GLM_ASSERT( bufferSize > 0 && bufferSize == getBufferSizeToAllocate(elementCountMax) );
			GLM_ASSERT( elementCountMax >= 2 && GLM_MATH_IS_POWER_OF_TWO(elementCountMax) && elementCountMax < GLM_UINT32_MAX );

			//NOTE: queue starts empty, no need to call constructor on all buffer elements
			element_t* elements = static_cast<element_t*>(buffer);

			for (size_t i = 0; i < elementCountMax; ++i)
			{
				atomicStoreRelaxed(elements[i]._sequence, static_cast<atomic32_t::value_t>(i));
				//_buffer[i]._sequence.store(i, std::memory_order_relaxed);
			}
			atomicStoreRelaxed(_enqueueHead, 0);
			//_enqueueHead.store(0, std::memory_order_relaxed);
			atomicStoreRelaxed(_dequeueTail, 0);
			//_dequeueTail.store(0, std::memory_order_relaxed);

			_buffer = elements;
			_bufferMask = static_cast<uint32_t>(elementCountMax - 1);

			return true;
		}

		void uninitialize()
		{
			//IMPORTANT: NOT multi-thread SAFE, make sure push() & pop() calls from all threads are finished

			//NOTE: if MPMC_RingBuffer didn't allocate buffer, shouldn't be responsible for freeing the memory
			//delete [] _buffer;
			_buffer = NULL;
			_bufferMask = 0;
		}

		bool try_push(T const& data)
		{
			const uint32_t bufferMask = _bufferMask;
			bool wasPushed = false;

			element_t* cell;
			//size_t pos = _enqueueHead.load(std::memory_order_relaxed);
			atomic32_t::value_t head = atomicLoadRelaxed(_enqueueHead);

			for (;;)
			{
				//cell = &_buffer[pos & _bufferMask];
				cell = &_buffer[head & bufferMask];
				//size_t seq = cell->_sequence.load(std::memory_order_acquire);
				atomic32_t::value_t seq = atomicLoadAcquire(cell->_sequence);

				intptr_t dif = (intptr_t)seq - (intptr_t)head;
				if (dif == 0)
				{
					//const bool wasChanged = _enqueueHead.compare_exchange_weak(pos, pos + 1, std::memory_order_relaxed);
					const atomic32_t::value_t next_head = next(head);
					const bool wasChanged = ( atomicCompareExchangeRelaxed(_enqueueHead, next_head, head) == head );
					if ( wasChanged )
					{
						cell->_data = data;
						//cell->_sequence.store(pos + 1, std::memory_order_release);
						atomicStoreRelease(cell->_sequence, next_head);

						wasPushed = true;
						break;
					}
				}
				else if (dif < 0)
				{
					break;
				}
				else
				{
					//pos = _enqueueHead.load(std::memory_order_relaxed);
					head = atomicLoadRelaxed(_enqueueHead);
				}
			}

			return wasPushed;
		}

		bool try_pop(T& data)
		{
			const uint32_t bufferMask = _bufferMask;
			bool wasPoped = false;

			element_t* cell;
			//size_t pos = _dequeueTail.load(std::memory_order_relaxed);
			atomic32_t::value_t tail = atomicLoadRelaxed(_dequeueTail);

			for (;;)
			{
				//cell = &_buffer[pos & _bufferMask];
				cell = &_buffer[tail & bufferMask];
				//size_t seq = cell->_sequence.load(std::memory_order_acquire);
				const atomic32_t::value_t seq = atomicLoadAcquire(cell->_sequence);

				//intptr_t dif = (intptr_t)seq - (intptr_t)(pos + 1);
				const atomic32_t::value_t next_tail = next(tail);
				intptr_t dif = (intptr_t)seq - (intptr_t)(next_tail);
				if (dif == 0)
				{
					//const bool wasChanged = _dequeueTail.compare_exchange_weak(pos, pos + 1, std::memory_order_relaxed);
					const bool wasChanged = (atomicCompareExchangeRelaxed(_dequeueTail, next_tail, tail) == tail);
					if (wasChanged)
					{
						data = cell->_data;
						//cell->_sequence.store(pos + _bufferMask + 1, std::memory_order_release);
						atomicStoreRelease(cell->_sequence, next_tail + bufferMask);

						wasPoped = true;
						break;
					}
				}
				else if (dif < 0)
				{
					break;
				}
				else
				{
					//pos = _dequeueTail.load(std::memory_order_relaxed);
					tail = atomicLoadRelaxed(_dequeueTail);
				}
			}

			return wasPoped;
		}

		const bool wasInitialized() const
		{
			return (_buffer != NULL);
		}

		const bool wasEmpty() const
		{
			return ( atomicLoadRelaxed(_dequeueTail) == atomicLoadRelaxed(_enqueueHead) );
		}

		//const size_t getElementCountMax() const { return _bufferMask + 1; }

		static size_t getBufferSizeToAllocate(size_t elementCountMax)
		{
			GLM_ASSERT( elementCountMax >= 2 );

			//NOTE: MPMC_RingBuffer current implementation only supports 32 bit
			const bool countMaxIsSupported = (elementCountMax < GLM_UINT32_MAX);
			GLM_DEBUG_ASSERT_MSG( countMaxIsSupported, "MPMC_RingBuffer current implementation only supports 32 bit" );

			//NOTE: restriction to speed up next element computation using 'bitmask' instead of 'modulo' operation
			const bool countMaxIsPowerOfTwo = GLM_MATH_IS_POWER_OF_TWO(elementCountMax);
			GLM_DEBUG_ASSERT_MSG( countMaxIsPowerOfTwo, "To speed up some computation, buffer element count max is restricted to a power of two number.");

			const size_t allocationSize = (countMaxIsPowerOfTwo && countMaxIsSupported) ? sizeof(element_t) * elementCountMax : 0;
			return allocationSize;
		}

	private:
		GLM_DISALLOW_COPY_AND_ASSIGN(MPMC_RingBuffer)

		atomic32_t::value_t next(atomic32_t::value_t current)
		{
			return (current + 1);
		}

		typedef char		cacheline_pad_t[GLM_MEMORY_CACHE_LINE_SIZE];

		cacheline_pad_t		_cachelinePad0;
		element_t*			_buffer;
		uint32_t			_bufferMask;
		cacheline_pad_t		_cachelinePad1;
		atomic32_t			_enqueueHead;
		cacheline_pad_t		_cachelinePad2;
		atomic32_t			_dequeueTail;
		cacheline_pad_t		_cachelinePad3;
	}; 
}

#endif // GLMCONCURRENTDATASTRUCTURES_H

