//
// Asura_Task_System.cpp
//
// Created by KJFL - 17 January 2008
//
// Copyright (C) 2008 Rebellion, All Rights Reserved.
//

#include "Core/Asura.h"

#ifdef ASURA_USE_TASK_SYSTEM

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

#include "Core/Asura_Atomic.h"
#include "Core/Asura_Multithread.h"
#include "Core/Asura_StringBuffer.h"

Asura_Task* volatile	Asura_Task_System::s_pxCurrentTask = 0;

Asura_Semaphore*		Asura_Task_System::s_pxTaskAvailable = 0;
Asura_Semaphore*		Asura_Task_System::s_pxTaskFinished = 0;
Asura_Semaphore*		Asura_Task_System::s_pxSubTaskAvailable = 0;
Asura_Semaphore*		Asura_Task_System::s_pxThreadsTerminated = 0;

u_int volatile			Asura_Task_System::s_uNumberOfWorkerThreads = 0;
u_int volatile			Asura_Task_System::s_uNumberOfActiveThreads = 0;

bool volatile			Asura_Task_System::s_bTerminateThreads = false;
bool volatile			Asura_Task_System::s_bTerminateTask = false;
bool volatile			Asura_Task_System::s_bTaskFinished = false;

void Asura_Task_System::Initialise( void )
{
	s_bTerminateThreads = false;
	s_bTerminateTask = false;
	s_bTaskFinished = false;

	s_pxCurrentTask = 0;
	s_uNumberOfActiveThreads = 0;

	s_uNumberOfWorkerThreads = Asura_Thread_System::GetNumHardwareThreads();

	s_pxTaskAvailable = new Asura_Semaphore( "ATS_TaskAvailable", 0, s_uNumberOfWorkerThreads );
	s_pxTaskFinished = new Asura_Semaphore( "ATS_TaskFinished", 0, s_uNumberOfWorkerThreads );

	s_pxSubTaskAvailable = new Asura_Semaphore( "ATS_SubTaskAvailable", 0, 0x7fffffff );
	s_pxThreadsTerminated = new Asura_Semaphore("ATS_ThreadsTerminated", 0, s_uNumberOfWorkerThreads );

	for ( u_int uThread = 0; uThread < s_uNumberOfWorkerThreads; ++uThread )
	{
		Asura_Thread_System::CreateThread( "TaskWorkerThreads", WorkerThreadFunction, 0 );
	}
}

void Asura_Task_System::Reset( void )
{
}

void Asura_Task_System::Shutdown( void )
{
	Asura_Assert( s_pxCurrentTask==0, "A Task is still active!" );
	s_pxCurrentTask = 0;

	s_bTerminateThreads = true;
	s_bTerminateTask = true;

	// wake up the threads so they can quit
	for ( u_int uThread = 0; uThread < s_uNumberOfWorkerThreads; ++uThread )
	{
		s_pxSubTaskAvailable->Signal();
		s_pxTaskAvailable->Signal();
	}


	// wait for all threads to terminate before we start deleting things
	for ( u_int uThread = 0; uThread < s_uNumberOfWorkerThreads; ++uThread )
	{
		s_pxThreadsTerminated->Wait();
	}

	delete s_pxTaskAvailable;
	delete s_pxTaskFinished;
	delete s_pxSubTaskAvailable;
	delete s_pxThreadsTerminated;

	s_pxTaskAvailable = 0;
	s_pxTaskFinished = 0;
	s_pxSubTaskAvailable = 0;
	s_pxThreadsTerminated = 0;
}

u_int Asura_Task_System::GetNumberOfWorkerThreads( void )
{
	return s_uNumberOfWorkerThreads;
}

u_int Asura_Task_System::GetNumberOfActiveThreads( void )
{
	return s_uNumberOfActiveThreads;
}

void Asura_Task_System::AddTask( Asura_Task* pxTask )
{
	// TODO - this could become a queue of tasks
	if ( s_pxCurrentTask )
	{
		Asura_Assert( false, "Another Task is still running" );
		return;
	}

	s_bTerminateTask = false;
	s_bTaskFinished = false;

	// tidy up any existing subtask count from a previous task
	// TODO - a semaphore Reset() fn?
	while ( s_pxSubTaskAvailable->TryWait() );

	s_pxCurrentTask = pxTask;
	
	// wake all the worker threads
	for ( u_int uThread = 0; uThread < s_uNumberOfWorkerThreads; ++uThread )
	{
		s_pxTaskAvailable->Signal();
	}
	
}

void Asura_Task_System::RemoveTask( Asura_Task* pxTask )
{
	// TODO - this could become a queue of tasks
	if ( s_pxCurrentTask != pxTask )
	{
		Asura_Assert( false, "This is not the current Task" );
		return;
	}

	YieldUntilTaskFinished();

	s_pxCurrentTask = 0;
}

void Asura_Task_System::YieldUntilTaskFinished( void )
{
	if ( s_bTaskFinished ) return;

	// Wait for any subtasks still being processed to finish
	for ( u_int uThread = 0; uThread < s_uNumberOfWorkerThreads; ++uThread )
	{
		s_pxTaskFinished->Wait();
	}

	s_bTaskFinished = true;

	Asura_Assert( Asura_Task_System::GetNumberOfActiveThreads()==0, "Code bug: a thread is still working" );
}

void Asura_Task_System::SignalWorkIsAvailable( void )
{
	s_pxSubTaskAvailable->Signal();
}

void Asura_Task_System::SignalWorkIsNotAvailable( void )
{
	if (s_bTerminateTask) return;

	s_bTerminateTask = true;
	
	// we're out of work - everyone wake up so we can quit this task
	for ( u_int uThread = 0; uThread < s_uNumberOfWorkerThreads; ++uThread )
	{
		s_pxSubTaskAvailable->Signal();
	}
}

void Asura_Task_System::DoTask( void )
{
	// TODO - this becomes the list of queues.
	if ( s_pxCurrentTask )
	{
		Asura_TaskFunction pfnTask = s_pxCurrentTask->GetTaskFunction();

		if (pfnTask)
		{
			void* pData = s_pxCurrentTask->GetTaskData();

			if (pData)
			{
				pfnTask( pData );
			}
		}
	}
}

void Asura_Task_System::WorkerThreadFunction( const void* pData )
{
	do 
	{
		s_pxTaskAvailable->Wait();

		// exit if we have to
		if ( s_bTerminateThreads ) break;
		
		Asura_Atomic::Increment( s_uNumberOfActiveThreads );

		do
		{
			s_pxSubTaskAvailable->Wait();

			// exit if we have to
			if ( s_bTerminateThreads ) break;

			if ( s_bTerminateTask ) break;

			// otherwise do the alloted task
			DoTask();
		}
		while ( 1 );

		Asura_Atomic::Decrement( s_uNumberOfActiveThreads );

		s_pxTaskFinished->Signal();

	}
	while ( 1 );

	s_pxThreadsTerminated->Signal();
}

#endif //#ifdef ASURA_USE_TASK_SYSTEM
