//
// Asura_ProducerConsumerQueue.h
//
// Created by KJFL - 18 April 2008
//
// Copyright (C) 2008 Rebellion, All Rights Reserved.
//

#include "Core/Asura.h"

// This file's header.
#include "Core/Threads/Asura_ProducerConsumerQueue.h"

#include "Core/Asura_Multithread.h"

Asura_ProducerConsumerQueue::Asura_ProducerConsumerQueue( u_int uMaxQueueSize, u_int uNumberOfThreads, const ASURA_MULTITHREAD_STACK_SIZE xStackSize ) 
: m_xWorkAvailableSemaphore( "APCQ_WorkAvail", 0, uMaxQueueSize+uNumberOfThreads ) // We make "fake" work to exit threads
, m_xFreeSpaceSemaphore("APCQ_FreeSpace", uMaxQueueSize, uMaxQueueSize )
, m_xThreadsTerminated("APCQ_ThreadsTerminated", 0, uNumberOfThreads )
{
	Asura_Assert( uMaxQueueSize != 0, "Code bug: Queue size can't be zero" );
	Asura_Assert( uNumberOfThreads != 0, "Code bug: Number of threads can't be zero" );

	m_bShouldTerminate = false;

	m_uNumberOfThreads = uNumberOfThreads;

	m_pxQueueHead = new TaskDataPacket[ uMaxQueueSize ];

	// We don't really need to zero all these task functions, it's just
	// part of our tests for robustness. (Could eventually be #ifdef'd out 
	// for none assertion builds.)
	for ( u_int u=0; u<uMaxQueueSize; ++u )
	{
		m_pxQueueHead[u].m_pfTaskFunction = 0;
	}

	m_uNextFree = 0;
	m_uNextTask = 0;
	m_uMaxQueueSize = uMaxQueueSize;

	// create the threads *after* we've set up all the data
	for ( u_int uThread = 0; uThread < uNumberOfThreads; ++uThread )
	{
		Asura_Thread_System::CreateThread( "ConsumerThread", ConsumerThreadMain, this, xStackSize );
	}
}

Asura_ProducerConsumerQueue::~Asura_ProducerConsumerQueue()
{
	m_bShouldTerminate = true;

	// create fake work to wake up any sleeping threads
	for ( u_int u=0; u<m_uNumberOfThreads; ++u )
	{
		m_xWorkAvailableSemaphore.Signal();
	}

	// wait for threads to finish
	for ( u_int u=0; u<m_uNumberOfThreads; ++u )
	{
		m_xThreadsTerminated.Wait();
	}

	delete [] m_pxQueueHead;
}


bool Asura_ProducerConsumerQueue::TryToAddToQueue( Asura_ProduceConsumerFunction pfTaskFunction, void* pData, u_int uSizeOfData )
{
	if ( !m_xFreeSpaceSemaphore.TryWait() )
	{
		return false;
	}

	AddToQueue_Internal( pfTaskFunction, pData, uSizeOfData );

	return true;
}

void Asura_ProducerConsumerQueue::AddToQueue( Asura_ProduceConsumerFunction pfTaskFunction, void* pData, u_int uSizeOfData )
{
	m_xFreeSpaceSemaphore.Wait();

	AddToQueue_Internal( pfTaskFunction, pData, uSizeOfData );
}

void Asura_ProducerConsumerQueue::AddToQueue_Internal( Asura_ProduceConsumerFunction pfTaskFunction, void* pData, u_int uSizeOfData )
{
	u_int uEntry;

	// KJFL - 21 April 2008 - this lock contains the very minimum we need to be threadsafe.
	// Note this would almost be done as an Atomic add - something to try later.
	m_xAddMutex.Lock();
	
	uEntry = m_uNextFree;
	m_uNextFree = (m_uNextFree + 1) % m_uMaxQueueSize;

	m_xAddMutex.Unlock();

	Asura_Assert( m_pxQueueHead[ uEntry ].m_pfTaskFunction == 0, "Code bug: this slot isn't free" );

	// Copy the task onto the queue
	{
		m_pxQueueHead[ uEntry ].m_pfTaskFunction = pfTaskFunction;
		m_pxQueueHead[ uEntry ].m_pData = pData;
		m_pxQueueHead[ uEntry ].m_uSizeOfData = uSizeOfData;
	}

	// Signal we have work available
	m_xWorkAvailableSemaphore.Signal();
}

u_int Asura_ProducerConsumerQueue::GetMaxQueueSize( void ) const
{
	return m_uMaxQueueSize;
}

bool Asura_ProducerConsumerQueue::GetFromQueue( Asura_ProducerConsumerQueue::TaskDataPacket& xPacket )
{
	if ( ShouldTerminate() ) { return false; }

	u_int uEntry;

	// KJFL - 21 April 2008 - this lock contains the very minimum we need to be threadsafe.
	// Note this would almost be done as an Atomic add - something to try later.
	m_xGetMutex.Lock();

	uEntry = m_uNextTask;
	m_uNextTask = (m_uNextTask + 1) % m_uMaxQueueSize;

	m_xGetMutex.Unlock();

	// if the semaphore is working we should always have data here
	Asura_Assert( m_pxQueueHead[ uEntry ].m_pfTaskFunction != 0, "Code bug: this slot isn't full" );
	
	if (m_pxQueueHead[ uEntry ].m_pfTaskFunction != 0)
	{
		// copy the task, and then we can reuse this slot
		xPacket = m_pxQueueHead[ uEntry ];
		
		// Flag the slot as empty - this is only needed for assertion checks
		m_pxQueueHead[ uEntry ].m_pfTaskFunction = 0;
	
		// Signal we have space on the queue
		m_xFreeSpaceSemaphore.Signal();

		return true;
	}
	
	return false;
}

void Asura_ProducerConsumerQueue::ConsumerThreadMain( const void* pQueue )
{
	// KJFL - 18 April 2008 - TODO - need to check whether we really need the
	// thread main fns to be const void* rather than just void* - it's a pain.
	Asura_ProducerConsumerQueue* pxQueue = static_cast<Asura_ProducerConsumerQueue*>( const_cast<void*>( pQueue ) );

	Asura_Assert( pxQueue, "Code bug: invalid pointer" );

	do
	{
		// Wait until we have some work todo (which includes terminating)
		pxQueue->m_xWorkAvailableSemaphore.Wait();

		// Request a data packet from the queue - GetFromQueue() may return 
		// false if we're terminating or if there has been an error.
		TaskDataPacket xPacket;
		if ( pxQueue->GetFromQueue( xPacket ) )
		{
			Asura_Assert( xPacket.m_pfTaskFunction, "Code bug: null function pointer" );

			xPacket.m_pfTaskFunction( xPacket.m_pData, xPacket.m_uSizeOfData );
		}
	}
	while ( ! pxQueue->ShouldTerminate() );

	// signal the thread is exitting
	pxQueue->m_xThreadsTerminated.Signal();
}
