ds@1952: /**
ds@1952:  * vim: set ts=4 :
ds@1952:  * =============================================================================
ds@1952:  * SourceMod
ds@1952:  * Copyright (C) 2004-2008 AlliedModders LLC.  All rights reserved.
ds@1952:  * =============================================================================
ds@1952:  *
ds@1952:  * This program is free software; you can redistribute it and/or modify it under
ds@1952:  * the terms of the GNU General Public License, version 3.0, as published by the
ds@1952:  * Free Software Foundation.
ds@1952:  * 
ds@1952:  * This program is distributed in the hope that it will be useful, but WITHOUT
ds@1952:  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
ds@1952:  * FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more
ds@1952:  * details.
ds@1952:  *
ds@1952:  * You should have received a copy of the GNU General Public License along with
ds@1952:  * this program.  If not, see <http://www.gnu.org/licenses/>.
ds@1952:  *
ds@1952:  * As a special exception, AlliedModders LLC gives you permission to link the
ds@1952:  * code of this program (as well as its derivative works) to "Half-Life 2," the
ds@1952:  * "Source Engine," the "SourcePawn JIT," and any Game MODs that run on software
ds@1952:  * by the Valve Corporation.  You must obey the GNU General Public License in
ds@1952:  * all respects for all other code used.  Additionally, AlliedModders LLC grants
ds@1952:  * this exception to all derivative works.  AlliedModders LLC defines further
ds@1952:  * exceptions, found in LICENSE.txt (as of this writing, version JULY-31-2007),
ds@1952:  * or <http://www.sourcemod.net/license.php>.
ds@1952:  *
ds@1952:  * Version: $Id$
ds@1952:  */
ds@1952: 
ds@1952: #include "ThreadWorker.h"
ds@1952: 
ds@1952: ThreadWorker::ThreadWorker(IThreadWorkerCallbacks *hooks) : BaseWorker(hooks),
ds@1952: 	m_Threader(NULL),
ds@1952: 	m_QueueLock(NULL),
ds@1952: 	m_StateLock(NULL),
ds@1952: 	m_PauseSignal(NULL),
ds@1952: 	m_AddSignal(NULL),
ds@1952: 	me(NULL),
ds@1952: 	m_think_time(DEFAULT_THINK_TIME_MS)
ds@1952: {
ds@1952: 	m_state = Worker_Invalid;
ds@1952: }
ds@1952: 
ds@1952: ThreadWorker::ThreadWorker(IThreadWorkerCallbacks *hooks, IThreader *pThreader, unsigned int thinktime) : 
ds@1952: 	BaseWorker(hooks),
ds@1952: 	m_Threader(pThreader),
ds@1952: 	m_QueueLock(NULL),
ds@1952: 	m_StateLock(NULL),
ds@1952: 	m_PauseSignal(NULL),
ds@1952: 	m_AddSignal(NULL),
ds@1952: 	me(NULL),
ds@1952: 	m_think_time(thinktime)
ds@1952: {
ds@1952: 	if (m_Threader)
ds@1952: 	{
ds@1952: 		m_state = Worker_Stopped;
ds@1952: 	} else {
ds@1952: 		m_state = Worker_Invalid;
ds@1952: 	}
ds@1952: }
ds@1952: 
ds@1952: ThreadWorker::~ThreadWorker()
ds@1952: {
ds@1952: 	if (m_state != Worker_Stopped || m_state != Worker_Invalid)
ds@1952: 	{
ds@1952: 		Stop(true);
ds@1952: 	}
ds@1952: 
ds@1952: 	if (m_ThreadQueue.size())
ds@1952: 	{
ds@1952: 		Flush(true);
ds@1952: 	}
ds@1952: }
ds@1952: 
ds@1952: void ThreadWorker::OnTerminate(IThreadHandle *pHandle, bool cancel)
ds@1952: {
ds@1952: 	//we don't particularly care
ds@1952: 	return;
ds@1952: }
ds@1952: 
ds@1952: void ThreadWorker::RunThread(IThreadHandle *pHandle)
ds@1952: {
ds@1952: 	WorkerState this_state = Worker_Running;
ds@1952: 	size_t num;
ds@1952: 
ds@1952: 	if (m_pHooks)
ds@1952: 	{
ds@1952: 		m_pHooks->OnWorkerStart(this);
ds@1952: 	}
ds@1952:     
ds@1952: 	while (true)
ds@1952: 	{
ds@1952: 		/**
ds@1952: 		 * Check number of items in the queue
ds@1952: 		 */
ds@1952: 		m_StateLock->Lock();
ds@1952: 		this_state = m_state;
ds@1952: 		m_StateLock->Unlock();
ds@1952: 		if (this_state != Worker_Stopped)
ds@1952: 		{
ds@1952: 			m_QueueLock->Lock();
ds@1952: 			num = m_ThreadQueue.size();
ds@1952: 			if (!num)
ds@1952: 			{
ds@1952: 				/** 
ds@1952: 				 * if none, wait for an item
ds@1952: 				 */
ds@1952: 				m_Waiting = true;
ds@1952: 				m_QueueLock->Unlock();
ds@1952: 				/* first check if we should end again */
ds@1952: 				if (this_state == Worker_Stopped)
ds@1952: 				{
ds@1952: 					break;
ds@1952: 				}
ds@1952: 				m_AddSignal->Wait();
ds@1952: 				m_Waiting = false;
ds@1952: 			} else {
ds@1952: 				m_QueueLock->Unlock();
ds@1952: 			}
ds@1952: 		}
ds@1952: 		m_StateLock->Lock();
ds@1952: 		this_state = m_state;
ds@1952: 		m_StateLock->Unlock();
ds@1952: 		if (this_state != Worker_Running)
ds@1952: 		{
ds@1952: 			if (this_state == Worker_Paused || this_state == Worker_Stopped)
ds@1952: 			{
ds@1952: 				//wait until the lock is cleared.
ds@1952: 				if (this_state == Worker_Paused)
ds@1952: 				{
ds@1952: 					m_PauseSignal->Wait();
ds@1952: 				}
ds@1952: 				if (this_state == Worker_Stopped)
ds@1952: 				{
ds@1952: 					//if we're supposed to flush cleanrly, 
ds@1952: 					// run all of the remaining frames first.
ds@1952: 					if (!m_FlushType)
ds@1952: 					{
ds@1952: 						while (m_ThreadQueue.size())
ds@1952: 						{
ds@1952: 							RunFrame();
ds@1952: 						}
ds@1952: 					}
ds@1952: 					break;
ds@1952: 				}
ds@1952: 			}
ds@1952: 		}
ds@1952: 		/**
ds@1952: 		 * Run the frame.
ds@1952: 		 */
ds@1952: 		RunFrame();
ds@1952: 
ds@1952: 		/**
ds@1952: 		 * wait in between threads if specified
ds@1952: 		 */
ds@1952: 		if (m_think_time)
ds@1952: 		{
ds@1952: 			m_Threader->ThreadSleep(m_think_time);
ds@1952: 		}
ds@1952: 	}
ds@1952: 
ds@1952: 	if (m_pHooks)
ds@1952: 	{
ds@1952: 		m_pHooks->OnWorkerStop(this);
ds@1952: 	}
ds@1952: }
ds@1952: 
ds@1952: SWThreadHandle *ThreadWorker::PopThreadFromQueue()
ds@1952: {
ds@1952: 	if (m_state <= Worker_Stopped && !m_QueueLock)
ds@1952: 	{
ds@1952: 		return NULL;
ds@1952: 	}
ds@1952: 
ds@1952: 	SWThreadHandle *swt;
ds@1952: 	m_QueueLock->Lock();
ds@1952: 	swt = BaseWorker::PopThreadFromQueue();
ds@1952: 	m_QueueLock->Unlock();
ds@1952: 
ds@1952: 	return swt;
ds@1952: }
ds@1952: 
ds@1952: void ThreadWorker::AddThreadToQueue(SWThreadHandle *pHandle)
ds@1952: {
ds@1952: 	if (m_state <= Worker_Stopped)
ds@1952: 	{
ds@1952: 		return;
ds@1952: 	}
ds@1952: 
ds@1952: 	m_QueueLock->Lock();
ds@1952: 	BaseWorker::AddThreadToQueue(pHandle);
ds@1952: 	if (m_Waiting)
ds@1952: 	{
ds@1952: 		m_AddSignal->Signal();
ds@1952: 	}
ds@1952: 	m_QueueLock->Unlock();
ds@1952: }
ds@1952: 
ds@1952: WorkerState ThreadWorker::GetStatus(unsigned int *threads)
ds@1952: {
ds@1952: 	WorkerState state;
ds@1952: 
ds@1952: 	m_StateLock->Lock();
ds@1952: 	state = BaseWorker::GetStatus(threads);
ds@1952: 	m_StateLock->Unlock();
ds@1952: 
ds@1952: 	return state;
ds@1952: }
ds@1952: 
ds@1952: bool ThreadWorker::Start()
ds@1952: {
ds@1952: 	if (m_state == Worker_Invalid)
ds@1952: 	{
ds@1952: 		if (m_Threader == NULL)
ds@1952: 		{
ds@1952: 			return false;
ds@1952: 		}
ds@1952: 	} else if (m_state != Worker_Stopped) {
ds@1952: 		return false;
ds@1952: 	}
ds@1952: 
ds@1952: 	m_Waiting = false;
ds@1952: 	m_QueueLock = m_Threader->MakeMutex();
ds@1952: 	m_StateLock = m_Threader->MakeMutex();
ds@1952: 	m_PauseSignal = m_Threader->MakeEventSignal();
ds@1952: 	m_AddSignal = m_Threader->MakeEventSignal();
ds@1952: 	m_state = Worker_Running;
ds@1952: 	ThreadParams pt;
ds@1952: 	pt.flags = Thread_Default;
ds@1952: 	pt.prio = ThreadPrio_Normal;
ds@1952: 	me = m_Threader->MakeThread(this, &pt);
ds@1952: 
ds@1952: 	return true;
ds@1952: }
ds@1952: 
ds@1952: bool ThreadWorker::Stop(bool flush_cancel)
ds@1952: {
ds@1952: 	if (m_state == Worker_Invalid || m_state == Worker_Stopped)
ds@1952: 	{
ds@1952: 		return false;
ds@1952: 	}
ds@1952: 
ds@1952: 	WorkerState oldstate;
ds@1952: 
ds@1952: 	//set new state
ds@1952: 	m_StateLock->Lock();
ds@1952: 	oldstate = m_state;
ds@1952: 	m_state = Worker_Stopped;
ds@1952: 	m_FlushType = flush_cancel;
ds@1952: 	m_StateLock->Unlock();
ds@1952: 
ds@1952: 	if (oldstate == Worker_Paused)
ds@1952: 	{
ds@1952: 		Unpause();
ds@1952: 	} else {
ds@1952: 		m_QueueLock->Lock();
ds@1952: 		if (m_Waiting)
ds@1952: 		{
ds@1952: 			m_AddSignal->Signal();
ds@1952: 		}
ds@1952: 		m_QueueLock->Unlock();
ds@1952: 	}
ds@1952: 
ds@1952: 	me->WaitForThread();
ds@1952: 	//destroy it
ds@1952: 	me->DestroyThis();
ds@1952: 	//flush all remaining events
ds@1952: 	Flush(true);
ds@1952: 
ds@1952: 	//free mutex locks
ds@1952: 	m_QueueLock->DestroyThis();
ds@1952: 	m_StateLock->DestroyThis();
ds@1952: 	m_PauseSignal->DestroyThis();
ds@1952: 	m_AddSignal->DestroyThis();
ds@1952: 
ds@1952: 	//invalidizzle
ds@1952: 	m_QueueLock = NULL;
ds@1952: 	m_StateLock = NULL;
ds@1952: 	m_PauseSignal = NULL;
ds@1952: 	m_AddSignal = NULL;
ds@1952: 	me = NULL;
ds@1952: 
ds@1952: 	return true;
ds@1952: }
ds@1952: 
ds@1952: bool ThreadWorker::Pause()
ds@1952: {
ds@1952: 	if (m_state != Worker_Running)
ds@1952: 	{
ds@1952: 		return false;
ds@1952: 	}
ds@1952: 
ds@1952: 	m_StateLock->Lock();
ds@1952: 	m_state = Worker_Paused;
ds@1952: 	m_StateLock->Unlock();
ds@1952: 
ds@1952: 	return true;
ds@1952: }
ds@1952: 
ds@1952: 
ds@1952: bool ThreadWorker::Unpause()
ds@1952: {
ds@1952: 	if (m_state != Worker_Paused)
ds@1952: 	{
ds@1952: 		return false;
ds@1952: 	}
ds@1952: 
ds@1952: 	m_StateLock->Lock();
ds@1952: 	m_state = Worker_Running;
ds@1952: 	m_StateLock->Unlock();
ds@1952: 	m_PauseSignal->Signal();
ds@1952: 	if (m_Waiting)
ds@1952: 	{
ds@1952: 		m_AddSignal->Signal();
ds@1952: 	}
ds@1952: 
ds@1952: 	return true;
ds@1952: }