cthreadimp.cpp

PGP8.0源码 请认真阅读您的文件包然后写出其具体功能

/*____________________________________________________________________________
		Copyright (C) 2002 PGP Corporation
        All rights reserved.

        $Id: CThreadImp.cpp,v 1.3 2002/08/06 20:10:37 dallen Exp $
____________________________________________________________________________*/

#include "pgpClassesConfig.h"
#include "CThreadImp.h"

_USING_PGP

// Types

struct CThreadImp::ThreadCallbackInfo : 
	public CListableObject<ThreadCallbackInfo>
{
	PGPBoolean	isInUse;

	CThread::CallbackFunc	callbackFunc;
	void					*refPtr;

	PGPBoolean	isAsync;
	CSemaphore	*pSyncSemaphore;
};


// Class CThreadImp public member functions

CThreadImp::~CThreadImp()
{
#if PGP_EXCEPTIONS
	try
	{
#endif	// PGP_EXCEPTIONS
		if (IsRunning())
			KillThread();
#if PGP_EXCEPTIONS
	}
	catch (CComboError&) { }
#endif	// PGP_EXCEPTIONS
}

CThreadImp::CThreadImp() : mIsRunning(FALSE), mShouldWeExit(FALSE)
{
#if !PGP_EXCEPTIONS
	Status() = mTaskQueueLock.Status();

	if (Status().IsntError())
		Status() = mTaskPoolLock.Status();

	if (Status().IsntError())
		Status() = mExitSemaphore.Status();
	
	if (Status().IsntError())
		Status() = mTaskSemaphore.Status();
#endif	// !PGP_EXCEPTIONS
}

SMART_ERROR 
CThreadImp::StartThread(PGPUInt32 maxQueuedRequests)
{
	pgpAssert(!IsRunning());

	SMART_ERROR_DECLARE

	SMART_ERROR_ASSIGN mTaskPool.Resize(maxQueuedRequests);

#if !PGP_EXCEPTIONS
	if (error.IsntError())
#endif	// !PGP_EXCEPTIONS
	{
		mTaskPool.Wipe();
		mShouldWeExit = FALSE;

		SMART_ERROR_ASSIGN StartOsThread();

	#if !PGP_EXCEPTIONS
		if (error.IsntError())
	#endif	// !PGP_EXCEPTIONS
			mIsRunning = TRUE;
	}

	SMART_ERROR_RETURN
}

void 
CThreadImp::KillThread()
{
	pgpAssert(IsRunning());

	mShouldWeExit = TRUE;

	mTaskSemaphore.Signal();
	mExitSemaphore.Wait();

	pgpAssert(mTaskQueue.Size() == 0);
	mIsRunning = FALSE;
}

SMART_ERROR 
CThreadImp::PerformAsyncCallback(
	CThread::CallbackFunc	callbackFunc, 
	void					*refPtr)
{
	pgpAssertAddrValid(callbackFunc, VoidAlign);

	SMART_ERROR_DECLARE
	ThreadCallbackInfo	*pCI	= GetFromPool();

#if PGP_EXCEPTIONS
	if (IsNull(pCI))
		THROW_PGPERROR(kPGPError_OutOfMemory);
#else	// !PGP_EXCEPTIONS
	if (IsNull(pCI))
		error.pgpErr = kPGPError_OutOfMemory;

	if (error.IsntError())
#endif	// PGP_EXCEPTIONS
	{
		pCI->callbackFunc	= callbackFunc;
		pCI->refPtr			= refPtr;
		pCI->isAsync		= TRUE;
		pCI->pSyncSemaphore	= NULL;

		mTaskQueueLock.Lock();
		mTaskQueue.Enqueue(pCI);
		mTaskQueueLock.Unlock();

		mTaskSemaphore.Signal();
	}

	SMART_ERROR_RETURN
}

SMART_ERROR 
CThreadImp::PerformSyncCallback(
	CThread::CallbackFunc	callbackFunc, 
	void					*refPtr)
{
	pgpAssertAddrValid(callbackFunc, VoidAlign);

	SMART_ERROR_DECLARE
	ThreadCallbackInfo	*pCI	= GetFromPool();

#if PGP_EXCEPTIONS

	if (IsNull(pCI))
		THROW_PGPERROR(kPGPError_OutOfMemory);

	// Create new semaphore to wait on task completion.
	{
		CSemaphore	syncSem;

#else	// !PGP_EXCEPTIONS

	if (IsNull(pCI))
		error.pgpErr = kPGPError_OutOfMemory;

	// Create new semaphore to wait on task completion.
	if (error.IsntError())
	{
		CSemaphore	syncSem;
		error = syncSem.Status();

		if (syncSem.Status().IsntError())
#endif	// PGP_EXCEPTIONS
		{
			pCI->callbackFunc	= callbackFunc;
			pCI->refPtr			= refPtr;
			pCI->isAsync		= FALSE;
			pCI->pSyncSemaphore	= &syncSem;

			mTaskQueueLock.Lock();
			mTaskQueue.Enqueue(pCI);
			mTaskQueueLock.Unlock();

			mTaskSemaphore.Signal();

			// Wait for the task to complete.
			syncSem.Wait();
			ReleaseToPool(pCI);
		}
	}

	SMART_ERROR_RETURN
}

CThreadImp::ThreadCallbackInfo * 
CThreadImp::GetFromPool()
{
	ThreadCallbackInfo	*pTCI	= NULL;

	mTaskPoolLock.Lock();

	for (PGPUInt32 i = 0; i < mTaskPool.Size(); i++)
	{
		if (!mTaskPool[i].isInUse)
		{
			mTaskPool[i].isInUse = TRUE;
			pTCI = &mTaskPool[i];
			break;
		}
	}

	mTaskPoolLock.Unlock();

	return pTCI;
}

void 
CThreadImp::ReleaseToPool(ThreadCallbackInfo *pTCI)
{
	pgpAssertAddrValid(pTCI, ThreadCallbackInfo);
	pgpAssert(pTCI->isInUse);

	mTaskPoolLock.Lock();
	pTCI->isInUse = FALSE;
	mTaskPoolLock.Unlock();
}

PGPBoolean 
CThreadImp::MainThreadLoop()
{
	// Wait for a task.
	mTaskSemaphore.Wait();

	// Are we exiting?
	if (mShouldWeExit)
	{
		mExitSemaphore.Signal();
		return FALSE;
	}

	// Get the next task in the queue;
	mTaskQueueLock.Lock();	
	ThreadCallbackInfo	*pCI	= mTaskQueue.Dequeue();
	mTaskQueueLock.Unlock();

	if (IsNull(pCI))
	{
		pgpAssert(FALSE);
		return TRUE;
	}

	// Process the task.
	pCI->callbackFunc(pCI->refPtr);

	if (pCI->isAsync)
	{
		ReleaseToPool(pCI);
	}
	else
	{
		// Signal that the callback's done.
		pgpAssertAddrValid(pCI->pSyncSemaphore, CSemaphore);
		pCI->pSyncSemaphore->Signal();
	}

	return TRUE;
}