//////////////////////////////////////////////////////////////////////////////
//
// Asura_Platform_Job.cpp
//
// Copyright (c) 2006 Rebellion. All rights reserved.
//
// By Dave Reed (13/11/06)
//
// Description:
//
// PC+360 implementation of Asura_Job code.
//
// Uses a set of worker threads, which pop jobs from the queue and execute 
// them. As the job queue owns the worker threads, they're not shared between
// queues - so we may need to rework this if we end up with lots of systems
// that use job queues...
//
// Ideally, we'd have a system-wide set of worker threads, that could differentiate
// between job types, and a bit less of the void* nastiness?
//
//////////////////////////////////////////////////////////////////////////////

// I N C L U D E S ///////////////////////////////////////////////////////////

#include "Core/Asura.h"

#ifdef ASURA_SUPPORT_MULTITHREADED_JOB_QUEUE


//#define   ASURA_JOBQUEUE_FORCE_SINGLETHREADED (1)

const int iMAX_WORKER_THREADS = 6;

// asura independent
#include "FileAccess/Asura_DebugFileStream.h"
#include "Jobs/Asura_Job.h"
#include "Jobs/Asura_Job_Manager.h"
#include "Debugging/Asura_Logfile.h"
#include "Debugging/Asura_CPUProfile.h"


// C O N S T A N T S /////////////////////////////////////////////////////////

// G L O B A L S /////////////////////////////////////////////////////////////

//////////////////////////////////////////////////////////////////////////////
// Asura_Job_Queue
//////////////////////////////////////////////////////////////////////////////

// ThreadSafeQueue - The worker threads will pull jobs off this queue,
// whilst the main thread adds jobs to it.
// --------------------------------------------------------------------
class ThreadSafeQueue
{
public:

	// Constructor - init the queue and critical section
	// ----------------------------------------------------------------
	ThreadSafeQueue(int iMaxLength, int iDataSize)
	{		
		m_iMaxLength = iMaxLength;
		m_iDataSize  = iDataSize;
		m_iLength    = 0;
		m_iHead      = 0;
		m_iTail      = 0;
		m_apxEntries = new void*[iMaxLength];		

		for (int i=0; i<m_iMaxLength; i++)
		{
			m_apxEntries[i] = new uint8[m_iDataSize];
		}

		InitializeCriticalSection(&m_csQueueEvent);
	};


	// Destructor - deinit the queue and critical section
	// ----------------------------------------------------------------
	~ThreadSafeQueue()
	{
		if (m_apxEntries)
		{
			for (int i=0; i<m_iMaxLength; i++)
			{
				if (m_apxEntries[i])
					delete [] m_apxEntries[i];
			}
			delete [] m_apxEntries;
		}

		DeleteCriticalSection(&m_csQueueEvent);
	}
	

	// Push - Push an element onto the queue, and return false if it 
	// fails (queue full)
	// ----------------------------------------------------------------
	bool Push(void *pxData)
	{
		bool bResult = true;

		EnterCriticalSection(&m_csQueueEvent);


		if (m_iLength<m_iMaxLength)
		{	
			memcpy(m_apxEntries[m_iTail], pxData, m_iDataSize);
			m_iTail++;
			m_iLength++;			

			if (m_iTail>=m_iMaxLength)
			{
				m_iTail=0;
			}
		}
		else
		{
			bResult = false;
		}

		LeaveCriticalSection(&m_csQueueEvent);

		return bResult;
	}

	// Pop - Attempt to take an element from the queue
	// ----------------------------------------------------------------
	bool Pop(void *pBuffer)
	{
		bool bResult = true;
		EnterCriticalSection(&m_csQueueEvent);
		if (m_iLength>0)
		{
			memcpy(pBuffer , m_apxEntries[m_iHead], m_iDataSize);
			m_iHead++;
			m_iLength--;
			if (m_iHead>=m_iMaxLength)
			{
				m_iHead=0;
			}
		}
		else
		{
			bResult = false;
		}
		LeaveCriticalSection(&m_csQueueEvent);
		return bResult;
	}

	// GetCurrentLength - Return the number of jobs on the queue
	// ----------------------------------------------------------------
	int GetCurrentLength(void)
	{
		EnterCriticalSection(&m_csQueueEvent);
		int iCount = m_iLength;
		LeaveCriticalSection(&m_csQueueEvent);
		return iCount;
	}

	// Clear - empty the queue
	// ----------------------------------------------------------------
	void Clear(void)
	{
		EnterCriticalSection(&m_csQueueEvent);
		m_iLength = 0;
		m_iTail = m_iHead = 0;
		LeaveCriticalSection(&m_csQueueEvent);
	}


	


private:
	CRITICAL_SECTION m_csQueueEvent;

	void **m_apxEntries;
	int    m_iHead;
	int    m_iTail;
	int    m_iLength;
	int    m_iMaxLength;
	int    m_iDataSize;
};



// ThreadData - stores everything the worker threads need access to,
// including a buffer to store the jobs they pop off the queue
// --------------------------------------------------------------------
class ThreadData
{
public:
	HANDLE						 m_hThread;
	DWORD						 m_dwThreadID;
	int							 m_iIndex;
	class Asura_Job_Queue_Data  *m_pxData;	
	void				        *m_pJobBuffer;		
};



// Platform specific data for this system
// --------------------------------------------------------------------
class Asura_Job_Queue_Data
{
public:
	Asura_Job_Queue *m_pxJobQueueObject;
	ThreadSafeQueue *m_pxQueue;

	CRITICAL_SECTION m_csNumPendingJobs;
	volatile int	 m_iNumPendingJobs;
	ThreadData		 m_axThreadData[iMAX_WORKER_THREADS];	
	bool			 m_bTerminateWorkers;
	HANDLE           m_hNewJobEvent;
	int				 m_iNumWorkerThreads;
};



//////////////////////////////////////////////////////////////////////////////



// Worker thread function - very simpler - Wait for signal, then  
// attempt to pop a job. If succeeded, then do the job.
// --------------------------------------------------------------------
long WINAPI WorkerThread(long lParam)
{
	ThreadData *pxThreadData = (ThreadData *)lParam;

	ASURA_CPU_PROFILE__START_THREAD( "Job");
	
	ASURA_CPU_PROFILE__START_MULTITHREADED_LABEL( ASURA_CPU_PROFILE__PRESENT_FRAME, "Waiting for Job");
	bool bExit = false;

	Asura_Job_Queue *pxQ = pxThreadData->m_pxData->m_pxJobQueueObject;

	while (!bExit)
	{
		while (pxQ->TryPop(pxThreadData->m_pJobBuffer))
		{	
//			Asura_OutputToDebugger(("Thread %d: ",pxThreadData->m_iIndex));
			ASURA_CPU_PROFILE__STOP_MULTITHREADED( ASURA_CPU_PROFILE__PRESENT_FRAME);
			pxQ->ExecuteJob(pxThreadData->m_pJobBuffer); 
			ASURA_CPU_PROFILE__START_MULTITHREADED_LABEL( ASURA_CPU_PROFILE__PRESENT_FRAME, "Waiting for Job");
			pxQ->JobFinished();	
		}	

		// Sleep until a new job comes in.
		WaitForSingleObject(pxThreadData->m_pxData->m_hNewJobEvent, 1000);

		// Do we have a quit request? - if so, exit now
		bExit = pxThreadData->m_pxData->m_bTerminateWorkers;
	}

	// Flag this thread as finished.
	// (This shutdown system could be improved, semaphores/waitonmultipleobjects might be better)
	pxThreadData->m_hThread = 0;

	ASURA_CPU_PROFILE__STOP_MULTITHREADED( ASURA_CPU_PROFILE__PRESENT_FRAME);
	
	ASURA_CPU_PROFILE__STOP_THREAD();
	
	return 0;
}



// Platform_Initialise - Init the data and create the worker threads
// --------------------------------------------------------------------
void Asura_Job_Queue::Platform_Initialise( u_int uDepth, u_int uDataSize )
{
	m_pxPlatformData = new Asura_Job_Queue_Data;
	Asura_Job_Queue_Data *pxData = (Asura_Job_Queue_Data*)m_pxPlatformData;

	// Create the queue
	pxData->m_pxJobQueueObject = this;
	pxData->m_pxQueue = new ThreadSafeQueue(uDepth, uDataSize);

	// Init pending job count
	InitializeCriticalSection(&pxData->m_csNumPendingJobs);
	pxData->m_iNumPendingJobs = 0;

	// Create the 'new job' event
	pxData->m_hNewJobEvent = CreateEvent(NULL, false, false, NULL);

	// Create worker threads
	// How many?  Will depend on platform. On PC, check how many cores we've got. On 360,
	// it'll be fixed.		
#ifndef ASURA_JOBQUEUE_FORCE_SINGLETHREADED
	pxData->m_iNumWorkerThreads	= 5;
#else
	pxData->m_iNumWorkerThreads	= 0;
#endif

	Asura_OutputToDebugger(("\nJob queue creating %d worker thread(s)\n\n" ,pxData->m_iNumWorkerThreads ));
	
	for(int i=0; i<iMAX_WORKER_THREADS ;i++)
	{
		pxData->m_axThreadData[i].m_hThread		= 0;
		pxData->m_axThreadData[i].m_iIndex		= i;
		pxData->m_axThreadData[i].m_pxData		= pxData;
		pxData->m_axThreadData[i].m_pJobBuffer	= new uint8[uDataSize];
	}

	XSetThreadProcessor(GetCurrentThread(),0);
	for(int i=0; i< pxData->m_iNumWorkerThreads; i++)
	{		
		pxData->m_axThreadData[i].m_hThread = CreateThread(NULL,0,(LPTHREAD_START_ROUTINE)WorkerThread,&pxData->m_axThreadData[i],0,&pxData->m_axThreadData[i].m_dwThreadID);
		XSetThreadProcessor(pxData->m_axThreadData[i].m_hThread,1+i);
	}
	

	// Need this for shutting down the worker threads
	pxData->m_bTerminateWorkers = false;

}


// Platform_Shutdown - Terminate the worker threads, and free resources
// --------------------------------------------------------------------
void Asura_Job_Queue::Platform_Shutdown( void )
{
	Asura_Job_Queue_Data *pxData = (Asura_Job_Queue_Data*)m_pxPlatformData;

	if (pxData)
	{	
		// Destroy worker threads / wait for them to terminate
		while(GetNumUnfinishedJobs())
		{
			Sleep(5);
		}

		// Tell them to exit
		pxData->m_bTerminateWorkers = true;

		int iActiveThreads = 1;
		while (iActiveThreads>0)		
		{	
			// Wake up a sleeping thread
			SetEvent(pxData->m_hNewJobEvent);
			
			// Little sleep here, to avoid a tight spin-wait loop
			Sleep(5);

			// Any still active?
			iActiveThreads = 0;
			for(int i=0; i<iMAX_WORKER_THREADS; i++)
			{
				if (pxData->m_axThreadData[0].m_hThread!=0)
				{
					iActiveThreads++;
				}
			}				
		}
	
		// Delete the event
		CloseHandle(pxData->m_hNewJobEvent);

		// Delete the thread data
		for(int i=0; i<iMAX_WORKER_THREADS ;i++)
		{
			if (pxData->m_axThreadData[i].m_pJobBuffer)
			{
				delete [] pxData->m_axThreadData[i].m_pJobBuffer;
			}
		}

		// Delete the queue
		DeleteCriticalSection(&pxData->m_csNumPendingJobs);		

		if (pxData->m_pxQueue)
		{
			delete pxData->m_pxQueue;
		}
		delete pxData;
	}
}

// Platform_SetThreadName - Give the thread a name
// --------------------------------------------------------------------
typedef struct tagTHREADNAME_INFO
{
    DWORD dwType;     // must be 0x1000
    LPCSTR szName;    // pointer to name (in user address space)
    DWORD dwThreadID; // thread ID (-1 = caller thread)
    DWORD dwFlags;    // reserved for future use, must be zero
} THREADNAME_INFO;

void Asura_Job_Queue::Platform_SetThreadName( const char* pszThreadName)
{
	Asura_Job_Queue_Data *pxData = (Asura_Job_Queue_Data*)m_pxPlatformData;

	for(int i=0; i<iMAX_WORKER_THREADS ;i++)
	{
		const HANDLE hThread = pxData->m_axThreadData[i].m_hThread;

		if( 0 == hThread)
		{
			continue;
		}
		
		const DWORD dwThreadID = pxData->m_axThreadData[i].m_dwThreadID;
		
		THREADNAME_INFO xInfo;
		xInfo.dwType = 0x1000;
		xInfo.szName = pszThreadName;
		xInfo.dwThreadID = dwThreadID;
		xInfo.dwFlags = 0;

		__try
		{
			RaiseException( 0x406D1388, 0, sizeof( xInfo) / sizeof( DWORD), (DWORD*)&xInfo );
		}
		__except( EXCEPTION_CONTINUE_EXECUTION ) {
		}
	}
}

// Get the number of entries in the job queue
// --------------------------------------------------------------------
u_int Asura_Job_Queue::Platform_GetNumEntries( void ) const
{
	// Return number of jobs left in the queue
	Asura_Job_Queue_Data *pxData = (Asura_Job_Queue_Data*)m_pxPlatformData;
	if (pxData)
	{
		return pxData->m_pxQueue->GetCurrentLength();
	}

	return 0;
}



// Add a job to the queue (or execute it immediately if we're running
// single-threaded)
// --------------------------------------------------------------------
void Asura_Job_Queue::Platform_Push( void* pJob )
{
	Asura_Job_Queue_Data *pxData = (Asura_Job_Queue_Data*)m_pxPlatformData;

	if (pxData)
	{
		// Make sure we've actually created some worker threads 
		if (pxData->m_iNumWorkerThreads>0 && (!Asura_Job_Queue_System::GetForceSingleThreaded()))
		{		
			// Push the job
			//if (pxData->m_pxQueue->Push(pJob))
			{					
				while( false == pxData->m_pxQueue->Push(pJob))
				{
					Sleep(0);
				}
				
					// Increment the number of pending jobs
				EnterCriticalSection(&pxData->m_csNumPendingJobs);
				pxData->m_iNumPendingJobs++;
				LeaveCriticalSection(&pxData->m_csNumPendingJobs);

				// Signal the worker threads that there's something in the queue
				SetEvent(pxData->m_hNewJobEvent);
			}
			//else
			{
				//ExecuteJob(pJob);
			}
		}
		else
		{
			// No worker threads - we're on a single-core/non-HT system, so just execute it now
			ExecuteJob(pJob);
		}
	}	
}


/*
void Asura_Job_Queue::Platform_Pop( void* pBuffer )
{
	// Not implemented. Always use TryPop instead.
}
*/


// Attempt to pop a job from the queue
// --------------------------------------------------------------------
bool Asura_Job_Queue::Platform_TryPop( void* pBuffer )
{
	Asura_Job_Queue_Data *pxData = (Asura_Job_Queue_Data*)m_pxPlatformData;

	if (pxData)
	{
		return pxData->m_pxQueue->Pop(pBuffer);
	}

	return false;
}


// After a job has been executed, the code that did it should call JobFinished()
// Platform_JobFinished should decrement the number of unfinished jobs
// --------------------------------------------------------------------
void Asura_Job_Queue::Platform_JobFinished( void ) const
{
	Asura_Job_Queue_Data *pxData = (Asura_Job_Queue_Data*)m_pxPlatformData;

	if (pxData)
	{
		EnterCriticalSection(&pxData->m_csNumPendingJobs);
		pxData->m_iNumPendingJobs--;
		LeaveCriticalSection(&pxData->m_csNumPendingJobs);
	}
}


// Number of unfinished jobs is maintained so code that requires all jobs to be 
// complete can wait until they are.
// --------------------------------------------------------------------
u_int Asura_Job_Queue::Platform_GetNumUnfinishedJobs( void ) const
{
	Asura_Job_Queue_Data *pxData = (Asura_Job_Queue_Data*)m_pxPlatformData;

	int iCount = 0;

	if (pxData)
	{
		iCount = pxData->m_iNumPendingJobs;
	}
		
	return iCount;
}


// Clear any queued jobs
// --------------------------------------------------------------------
void Asura_Job_Queue::Platform_Clear( void )
{
	Asura_Job_Queue_Data *pxData = (Asura_Job_Queue_Data*)m_pxPlatformData;

	if (pxData)
	{
		pxData->m_pxQueue->Clear();		
	}
	
	// Not sure if we need this? But we can't set that NumPendingJobs to zero here.
	while (pxData->m_iNumPendingJobs>0)
	{
		Sleep(5);
	}	
}

// Wait for All queued jobs to finish
// --------------------------------------------------------------------
void Asura_Job_Queue::Platform_WaitForAllJobsToFinish()
{
	Asura_Job_Queue_Data *pxData = (Asura_Job_Queue_Data*)m_pxPlatformData;

	if (pxData)
	{
		while( 0 < pxData->m_iNumPendingJobs);
	}
}

void Asura_Job_Manager::Initialise()
{
}

void Asura_Job_Manager::Shutdown()
{
}



#endif // #ifdef ASURA_SUPPORT_MULTITHREADED_JOB_QUEUE
//////////////////////////////////////////////////////////////////////////////
