//*******************************************************************
// File: Asura_XBox360_Job_Manager.cpp
//
// Created: 25 June 2007 DLH
//
// Description: 
//
//*******************************************************************

// system libraries
#include "Core/Asura.h"

#if defined( ASURA_SUPPORT_MULTITHREADED_JOB_QUEUE)

// main header
#include "Jobs/Asura_Job_Manager.h"

// other libraries
#include "Core/Asura_Atomic.h"
#include "Debugging/Asura_CPUProfile.h"
#include "Debugging/Asura_LogFile.h"
#include "Interface/Asura_CommandConsole.h"
#include "Jobs/Asura_Job.h"
#include "Jobs/Asura_Job_TaskQueue.h"
#include "Rendering/Asura_Xbox360_Graphics.h"

// internal dependencies

//*******************************************************************
// Constants
//*******************************************************************

#if 0 || defined( ASURA_SUPPORT_EXTERNAL_THREAD_PROFILING)
	#define ASURA_XBOX360_JOB_MANAGER_SUPPORT_VTUNE_THREADPROFILER
#endif

//*******************************************************************
// Local Classes
//*******************************************************************
class Asura_XBox360_Job_Manager_Data;

// DLH - Description of each job
class Asura_XBox360_Job_Manager_Job_Data
{
public:
	Asura_XBox360_Job_Manager_Job_Data();
	Asura_XBox360_Job_Manager_Job_Data( Asura_Job_Queue* pxJobQueue, Asura_Job_TaskQueue* pxTaskQueue);
	
	void JobFinished( void) const;
	void ExecuteJob( void) const;
	
public:

	Asura_Job_Queue*					m_pxJobQueue;
	Asura_Job_TaskQueue*				m_pxTaskQueue;
};

// DLH - One per thread
class Asura_XBox360_Job_Manager_Thread_Data
{
public:
	Asura_XBox360_Job_Manager_Thread_Data();
	~Asura_XBox360_Job_Manager_Thread_Data();
	
public:
	HANDLE	m_xThreadHandle;
	DWORD	m_dwThreadID;
	
	Asura_XBox360_Job_Manager_Data* m_pxThreadManager;
};

// DLH - Main Working class
class Asura_XBox360_Job_Manager_Data
{
public:
	Asura_XBox360_Job_Manager_Data();
	~Asura_XBox360_Job_Manager_Data();
	
	void Initialise();
	void Shutdown();
	
	void AddJobCountToQueue( const u_int uJobsCount);
	void RemoveJobCountFromQueue( const u_int uJobsCount);
	
	void AddJob( Asura_Job_Queue* pxJobQueue, Asura_Job_TaskQueue* pxJobTaskQueue);
	
	u_int GetNumUnfinishedJobs() const;

	void WaitForAllJobsToFinish();
	
private:
	static long WINAPI WorkerThread( void* pThreadData);
	
	void AddJob( const Asura_XBox360_Job_Manager_Job_Data& xJobData);

public:

private:
	Asura_Job_TaskQueue m_xJobQueue;
	

	CRITICAL_SECTION					m_xJobsCriticalSection;
	u_int								m_uNumPendingJobs;
	HANDLE								m_xNewJobEvent;
	HANDLE								m_xExitThreadEvent;
	
	u_int								m_uNumWorkerThreads;
	HANDLE*								m_paxWorkerThreads_Handles;
	Asura_XBox360_Job_Manager_Thread_Data*	m_paxWorkerThreads_Data;
	
	bool								m_bAllowTheading;
	
	u_int								m_uQueueTasks;
	u_int								m_uFreeTasks;
};

//*******************************************************************
// Local Datas
//*******************************************************************

// DLH - Instance of main working class
static Asura_XBox360_Job_Manager_Data s_xJobManagerData;

//*******************************************************************
// Function Definitions
//*******************************************************************

//--------------------------------------
// Asura_XBox360_Job_Manager_Job_Data
//--------------------------------------

Asura_XBox360_Job_Manager_Job_Data::Asura_XBox360_Job_Manager_Job_Data( Asura_Job_Queue* pxJobQueue, Asura_Job_TaskQueue* pxTaskQueue) :
	m_pxJobQueue( pxJobQueue),
	m_pxTaskQueue( pxTaskQueue)
{
	
}

Asura_XBox360_Job_Manager_Job_Data::Asura_XBox360_Job_Manager_Job_Data() :
	m_pxJobQueue( NULL),
	m_pxTaskQueue( NULL)
{
	
}

void Asura_XBox360_Job_Manager_Job_Data::ExecuteJob( void) const
{
	Asura_Job_Queue* pxJobQueue = this->m_pxJobQueue;
	if( NULL != pxJobQueue)
	{
		Asura_Job_TaskQueue* pxTaskQueue = this->m_pxTaskQueue;
		
		Asura_Job_TaskQueue::Task* pxTask;
		bool bTryPopStatus = pxTaskQueue->PopTask( pxTask);
		
		if( true == bTryPopStatus)
		{
			pxJobQueue->ExecuteJob( pxTask->GetData());

			pxJobQueue->JobFinished();
			
			pxTaskQueue->ReturnTask( pxTask);
		}
		return;
	}
}

void Asura_XBox360_Job_Manager_Job_Data::JobFinished( void) const
{
	Asura_Job_Queue* pxJobQueue = this->m_pxJobQueue;
	if( NULL != pxJobQueue)
	{
		pxJobQueue->JobFinished();
	}
}

//--------------------------------------
// Asura_XBox360_Job_Manager_Thread_Data
//--------------------------------------

Asura_XBox360_Job_Manager_Thread_Data::Asura_XBox360_Job_Manager_Thread_Data() :
	m_xThreadHandle( NULL),
	m_dwThreadID( 0),
	m_pxThreadManager( NULL)
{
}

Asura_XBox360_Job_Manager_Thread_Data::~Asura_XBox360_Job_Manager_Thread_Data()
{
}

//--------------------------------------
// Asura_XBox360_Job_Manager_Data
//--------------------------------------

Asura_XBox360_Job_Manager_Data::Asura_XBox360_Job_Manager_Data() :
	m_xNewJobEvent( NULL),
	m_xExitThreadEvent( NULL),
	m_paxWorkerThreads_Handles( NULL),
	m_paxWorkerThreads_Data( NULL),
	m_bAllowTheading( true),
	m_xJobQueue( 10, sizeof( Asura_XBox360_Job_Manager_Job_Data))
{

}

Asura_XBox360_Job_Manager_Data::~Asura_XBox360_Job_Manager_Data()
{
	Asura_Assert( NULL == m_paxWorkerThreads_Data, "Worker Threads Data not deallocated");
	Asura_Assert( NULL == m_paxWorkerThreads_Handles, "Worker Threads Handles not deallocated");
}

void Asura_XBox360_Job_Manager_Data::Initialise()
{
	this->m_uQueueTasks = 0;
	this->m_uFreeTasks = 0;

	// Init pending job count
	InitializeCriticalSectionAndSpinCount( &this->m_xJobsCriticalSection, 1000);
	this->m_uNumPendingJobs = 0;

	// Create the 'new job' event
	this->m_xNewJobEvent = CreateEvent( NULL, FALSE, FALSE, NULL);
	this->m_xExitThreadEvent = CreateEvent( NULL, TRUE, 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.		
	this->m_uNumWorkerThreads = 5;
	
	// DLH - 25 June 2007 15:34:34 - Create Thread Data
	this->m_paxWorkerThreads_Data = new Asura_XBox360_Job_Manager_Thread_Data[ this->m_uNumWorkerThreads];
	this->m_paxWorkerThreads_Handles = new HANDLE[ this->m_uNumWorkerThreads];
	
	XSetThreadProcessor(GetCurrentThread(),0);
	for( u_int uThreadIndex = 0; uThreadIndex < this->m_uNumWorkerThreads; uThreadIndex++)
	{		
		Asura_XBox360_Job_Manager_Thread_Data& xThreadData = this->m_paxWorkerThreads_Data[ uThreadIndex];
		
		xThreadData.m_pxThreadManager = this;
		
		HANDLE& xThreadHandle = this->m_paxWorkerThreads_Handles[ uThreadIndex];
		
		xThreadHandle = CreateThread( NULL, 0, (LPTHREAD_START_ROUTINE)WorkerThread, &xThreadData, 0, &xThreadData.m_dwThreadID);
		if( NULL == xThreadHandle)
		{
			continue;
		}
		
		xThreadData.m_xThreadHandle = xThreadHandle;
		
		XSetThreadProcessor( xThreadData.m_xThreadHandle, uThreadIndex+1);
	}
	
#ifdef ASURA_SUPPORT_COMMAND_CONSOLE
	Asura_CommandConsole::AddVar( L"CPU.MultiThreading.JobQueue.Threaded", &(this->m_bAllowTheading), Asura_CommandConsole::REMOVE_AT_SHUTDOWN);	
	Asura_CommandConsole::AddVar( L"CPU.MultiThreading.JobQueue.QueueTasks", &(this->m_uQueueTasks), 0, 99999, Asura_CommandConsole::REMOVE_AT_SHUTDOWN);	
	Asura_CommandConsole::AddVar( L"CPU.MultiThreading.JobQueue.FreeTasks", &(this->m_uFreeTasks), 0, 99999, Asura_CommandConsole::REMOVE_AT_SHUTDOWN);	
#endif
}

void Asura_XBox360_Job_Manager_Data::Shutdown()
{
	{
		// DLH - 25 June 2007 16:45:22 - Tell all the threads to quit
		SetEvent( this->m_xExitThreadEvent);
		
		// DLH - 25 June 2007 17:24:03 - Wait for all the threads to die
		u_int uNumWorkerThreadsTotal = this->m_uNumWorkerThreads;
		HANDLE* pxWorkerThreadsHandles = this->m_paxWorkerThreads_Handles;
		while( 0 < uNumWorkerThreadsTotal)
		{
			static const u_int uNumWorkerThreadsPerWaitMax = MAXIMUM_WAIT_OBJECTS;
			u_int uNumWorkerThreadsBatch = Asura_Maths::Min( uNumWorkerThreadsTotal, uNumWorkerThreadsPerWaitMax);
			
			const DWORD uWaitStatus = WaitForMultipleObjectsEx( uNumWorkerThreadsBatch, pxWorkerThreadsHandles, TRUE, 1000, FALSE);
			if( WAIT_TIMEOUT == uWaitStatus)
			{
				u_int uFirstThreadIndex = static_cast< u_int>( pxWorkerThreadsHandles - this->m_paxWorkerThreads_Handles);
				uFirstThreadIndex = uFirstThreadIndex;
				Asura_OutputToDebugger(("PC Job Manager waiting for threads %d to %d the finish.\n", uFirstThreadIndex, uFirstThreadIndex + uNumWorkerThreadsBatch));
				continue;
			}
			
			pxWorkerThreadsHandles += uNumWorkerThreadsBatch;
			uNumWorkerThreadsTotal -= uNumWorkerThreadsBatch;
		}
	}

	// Delete the event
	CloseHandle( this->m_xNewJobEvent);
	CloseHandle( this->m_xExitThreadEvent);

	// Delete the thread data
	delete[] this->m_paxWorkerThreads_Data;
	this->m_paxWorkerThreads_Data = NULL;
	
	delete[] this->m_paxWorkerThreads_Handles;
	this->m_paxWorkerThreads_Handles = NULL;

	DeleteCriticalSection(&this->m_xJobsCriticalSection);		
}

// Worker thread function - very simpler - Wait for signal, then  
// attempt to pop a job. If succeeded, then do the job.
// --------------------------------------------------------------------
long WINAPI Asura_XBox360_Job_Manager_Data::WorkerThread( void* pThreadData)
{
	Asura_Assert( NULL != pThreadData, "Thread Data not setup");
	
	Asura_XBox360_Job_Manager_Thread_Data* pxThreaData = static_cast< Asura_XBox360_Job_Manager_Thread_Data*>( pThreadData);
	Asura_Assert( NULL != pxThreaData, "Thread Data not setup");
	
	Asura_XBox360_Job_Manager_Data* pxThreadManager = pxThreaData->m_pxThreadManager;
	Asura_Assert( NULL != pxThreadManager, "Thread Manager not setup");
	
	bool bExit = false;
	
	Asura_Job_TaskQueue& xJobQueue = pxThreadManager->m_xJobQueue;

	ASURA_CPU_PROFILE__START_THREAD( "Job");
	
	ASURA_CPU_PROFILE__START_MULTITHREADED_LABEL( ASURA_CPU_PROFILE__PRESENT_FRAME, "Wait for Job" );
	
	HANDLE axWaitEventHandles[ 2] = 
	{
		pxThreadManager->m_xNewJobEvent,
		pxThreadManager->m_xExitThreadEvent
	};
	static const u_int uWaitEventHandlesCount = sizeof( axWaitEventHandles) / sizeof( axWaitEventHandles[0]);
	
	do
	{
		// Sleep until a new job comes in.
		DWORD uWaitForObjects = WaitForMultipleObjects( uWaitEventHandlesCount, axWaitEventHandles, false, INFINITE);
		
		if( (WAIT_OBJECT_0 + 0) != uWaitForObjects)
		{
			// DLH - 25 June 2007 16:25:32 - Exit Event Triggered
			break;
		}
		
//		ASURA_CPU_PROFILE__STOP_MULTITHREADED( ASURA_CPU_PROFILE__PRESENT_FRAME);
		
		SCOPEDPIXEVENT((0xff8800,  "Asura_XBox360_Job_Manager_Data::WorkerThread"));
	
		Asura_Job_TaskQueue::Task* pxTask;
		
		while( true == xJobQueue.PopTask( pxTask))
		{	
			Asura_Assert( NULL != pxTask, "This should not happen");
			void* pJobData = pxTask->GetData();
			Asura_XBox360_Job_Manager_Job_Data xJobData = *(static_cast< Asura_XBox360_Job_Manager_Job_Data*>( pJobData));
			
			ASURA_CPU_PROFILE__STOP_MULTITHREADED( ASURA_CPU_PROFILE__PRESENT_FRAME);
	
			//try
			{
				xJobData.ExecuteJob();
			}
			//catch (...)
			{
			}
			
			ASURA_CPU_PROFILE__START_MULTITHREADED_LABEL( ASURA_CPU_PROFILE__PRESENT_FRAME, "Wait for Job" );
		
			xJobQueue.ReturnTask( pxTask);
			
		}
		
//		ASURA_CPU_PROFILE__START_MULTITHREADED_LABEL( ASURA_CPU_PROFILE__PRESENT_FRAME, "Wait for Job" );
	}
	while (false == bExit);

	ASURA_CPU_PROFILE__STOP_MULTITHREADED( ASURA_CPU_PROFILE__PRESENT_FRAME);

	ASURA_CPU_PROFILE__STOP_THREAD();
	
	return( 0);
}

void Asura_XBox360_Job_Manager_Data::WaitForAllJobsToFinish()
{
	SCOPEDPIXEVENT((0xff8800,  "Asura_XBox360_Job_Manager_Data::WaitForAllJobsToFinish"));
	
	Asura_Job_TaskQueue& xJobQueue = this->m_xJobQueue;

	Asura_Job_TaskQueue::Task* pxTask;
	
	while( true == xJobQueue.PopTask( pxTask))
	{	
		Asura_Assert( NULL != pxTask, "This should not happen");
		void* pJobData = pxTask->GetData();
		Asura_XBox360_Job_Manager_Job_Data xJobData = *(static_cast< Asura_XBox360_Job_Manager_Job_Data*>( pJobData));
		
		//try
		{
			xJobData.ExecuteJob();
		}
		//catch (...)
		{
		}
		
		xJobQueue.ReturnTask( pxTask);
	}
	
	{
		while( false == xJobQueue.IsIdle())
		{
			// DLH - 28 June 2007 10:54:40 - Let this processor do something else but come back ASAP
			BOOL bSwitchStatus = SwitchToThread();
			if( 0 == bSwitchStatus)
			{
				//Sleep( 1);
			}
		}
	}
	
	this->m_uFreeTasks = xJobQueue.GetNumberOfFreeTasks();
	this->m_uQueueTasks = xJobQueue.GetNumberOfQueuedTasks();
	
	Asura_Assert( xJobQueue.IsValid(), "Job List Not Valid");
}

u_int Asura_XBox360_Job_Manager_Data::GetNumUnfinishedJobs() const
{
	return( 0);
}

void Asura_XBox360_Job_Manager_Data::AddJobCountToQueue( const u_int uJobsCount)
{
}

void Asura_XBox360_Job_Manager_Data::RemoveJobCountFromQueue( const u_int uJobsCount)
{
}

void Asura_XBox360_Job_Manager_Data::AddJob( const Asura_XBox360_Job_Manager_Job_Data& xJobData)
{
	if( false == this->m_bAllowTheading)
	{
		xJobData.ExecuteJob();
		return;
	}
	
	{
		// Increment the number of pending jobs
		Asura_Atomic::Increment( this->m_uNumPendingJobs);

		const void* pJobData = static_cast< const void*>( &xJobData);
		while( false == this->m_xJobQueue.Push( pJobData))
		{
			// DLH - 28 June 2007 10:54:40 - Let this processor do something else but come back ASAP
			BOOL bSwitchStatus = SwitchToThread();
			if( 0 == bSwitchStatus)
			{
				//Sleep( 0);
			}
		}
		
		// Signal the worker threads that there's something in the queue
		SetEvent( this->m_xNewJobEvent);
	}
}

void Asura_XBox360_Job_Manager_Data::AddJob( Asura_Job_Queue* pxJobQueue, Asura_Job_TaskQueue* pxTaskQueue)
{
	Asura_XBox360_Job_Manager_Job_Data xJobData( pxJobQueue, pxTaskQueue);
	
	this->AddJob( xJobData);
}

//--------------------------------------
// Asura_Job_Manager
//--------------------------------------

void Asura_Job_Manager::Initialise()
{
	s_xJobManagerData.Initialise();
}

void Asura_Job_Manager::Shutdown()
{
	s_xJobManagerData.Shutdown();
}

void Asura_Job_Manager::AddJobCountToQueue( const u_int uJobsCount)
{
	s_xJobManagerData.AddJobCountToQueue( uJobsCount);
}

void Asura_Job_Manager::RemoveJobCountFromQueue( const u_int uJobsCount)
{
	s_xJobManagerData.RemoveJobCountFromQueue( uJobsCount);
}

void Asura_Job_Manager::AddJob( Asura_Job_Queue* pxJobQueue, Asura_Job_TaskQueue* pxJobTaskQueue)
{
	s_xJobManagerData.AddJob( pxJobQueue, pxJobTaskQueue);
}

void Asura_Job_Manager::WaitForAllJobsToFinish()
{
	s_xJobManagerData.WaitForAllJobsToFinish();
}

//*******************************************************************
//
// (c) 2006 Rebellion
//
// Warning: This computer program is protected by copyright law and 
// international treaties. Unauthorized reproduction or distribution
// of this program, or any portion of it, may result in severe civil and
// criminal penalties, and will be prosecuted to the maximum extent 
// possible under the law.
//
//*******************************************************************
#endif //defined( ASURA_SUPPORT_MULTITHREADED_JOB_QUEUE)
