Webcam/include/platform/async/job.cpp

/* -----------------------------------------------------------------------------
	GSFramework
	Copyright 2001-2013 Emmanuel Julien. All Rights Reserved.
----------------------------------------------------------------------------- */


		#include "async/job.h"
		#include "memory/memory.h"
		#include "log/log.h"

		using namespace GS::ASync;
		using namespace GS::Threading;

	#ifndef	_DEBUG
		#define	__ENABLE_ITT_API__ 0
	#endif

	#if	__ENABLE_ITT_API__
		#include "ittnotify.h"
		static __itt_domain *domain = NULL;
	#endif


//------------------------------------------------------------------------------
void				JobWorkerThread::Execute()
{
	running.Set(1);

	Thread::SetName(GS::String::Format("Job Worker Thread %d", worker_id));

	while (running.Get() == 1)
	{
		// Execute as much jobs as possible until starvation.
		while (manager.ExecutePendingJob(worker_id));

		// Wait for notification on the queue event.
		manager.job_queued_event.Wait();
	}

	running.Set(0);
}
void				JobWorkerThread::Stop()
{	running.Set(2);	}
bool				JobWorkerThread::IsRunning() const
{	return running.Get() > 0;	}
//------------------------------------------------------------------------------

//------------------------------------------------------------------------------
bool				JobManager::CreateJobThreadPool(uint count)
{
	FreeJobThreadPool();

	if	(!pool.Allocate(count))
		return false;

	for	(uint n = 0; n < count; ++n)
		if	((pool[n] = new JobWorkerThread(*this, n + 1)) == NULL)
			__ERR__(__LOG_E__ << "Failed to allocate a job worker thread.\n", false)

	// Create workers.
	for	(uint n = 0; n < count; ++n)
		if	(!pool[n]->Start())
			__LOG_W__ << "Failed to create worker thread " << pool[n]->GetWorkerId() << ".\n";

	return true;
}
void				JobManager::FreeJobThreadPool()
{
	// Set thread exit flag.
	for	(uint n = 0; n < pool.GetCount(); ++n)
		pool[n]->Stop();

	// Trigger event so that the thread processes the exit flag.
	for	(uint n = 0; n < pool.GetCount(); ++n)
		job_queued_event.Trigger();

	// Join threads.
	for	(uint n = 0; n < pool.GetCount(); ++n)
		if	(!pool[n]->IsRunning())
			delete pool[n];

	pool.Free();
}
//------------------------------------------------------------------------------

//------------------------------------------------------------------------------
bool				JobManager::EnqueueJob(Job *job, JobGroup *group)
{
	if	(pool.GetCount() == 0)
	{
		job->Execute(0);
		job->done.Set(1);
	}
	else
	{
		job->done.Set(0);

		if	(group)
		{
			MutexLock lock(group->job_list_mutex);
			group->job_list.Add(job);
		}
		{
		#if	__USE_LOCK_FREE_JOB_QUEUE__
			while (!pending_queue.enqueue(job)) {}
		#else
			nMutexLock lock(pending_queue_mutex);
			pending_queue.Push(job);
		#endif
		}
	}

	job_queued_event.Trigger();		// wake workers
	return true;
}
//------------------------------------------------------------------------------

//------------------------------------------------------------------------------
bool				JobManager::JoinJob(Job *job, bool blocking)
{
	if	(job)
		while (job->done.Get() == 0)	// spinlock
			if	(!blocking)
				return false;

	return true;
}
bool				JobManager::JoinGroup(JobGroup *group, bool blocking)
{
	if	(group)
		for	(bool done = false; !done; )	// spinlock
		{
			done = true;
			{
				MutexLock glock(group->job_list_mutex);
				ListForeachPtr(Job *, job, group->job_list)
					if	(job->done.Get() == 0)
					{
						done = false;
						break;
					}
			}
			if	(!blocking)
				return done;

			Thread::Switch();
		}

	return true;
}
//------------------------------------------------------------------------------

//------------------------------------------------------------------------------
bool				JobManager::ExecutePendingJob(uint worker_id)
{
	// Look for a job to execute.
	Job *job = NULL;
	{
	#if	__USE_LOCK_FREE_JOB_QUEUE__
		pending_queue.dequeue(job);
	#else
		nMutexLock pending_lock(pending_queue_mutex);
		if	(pending_queue.GetCount() > 0)
		{
			job = pending_queue.Top();
			pending_queue.Pop();
		}
	#endif
	}

	// Execute job.
	if	(job)
	{
	#if	__ENABLE_ITT_API__
		__itt_task_begin(domain, __itt_null, __itt_null, __itt_string_handle_create(job->name));
	#endif

//		job->time_start = Platform::Get().GetTime();

		job->Execute(worker_id);
		job->done.Set(1);

//		job->time_end = Platform::Get().GetTime();

		Thread::Switch();	// let other workers do their job

	#if	__ENABLE_ITT_API__
		__itt_task_end(domain);
	#endif
	}
	return asbool(job);
}
//------------------------------------------------------------------------------

//------------------------------------------------------------------------------
uint				JobManager::GetWorkerPoolSize() const
{	return pool.GetCount();	}
//------------------------------------------------------------------------------

//------------------------------------------------------------------------------
JobManager::JobManager() : pending_queue(256)
{
#if	__ENABLE_ITT_API__
	domain = __itt_domain_create("GS.JobManager");
#endif

#if	(__USE_LOCK_FREE_JOB_QUEUE__ == 0)
	pending_queue_mutex = new nMutex;
#endif
}
JobManager::~JobManager()
{	FreeJobThreadPool();	}
JobGroup::JobGroup()
{	job_list_mutex = new Mutex;	}
//------------------------------------------------------------------------------