threadmanager.cpp

五行MMORPG引擎系统V1.0

/*///////////////////////////////////////////////////////////////////////////////////
线程管理
	包括线程池管理
	线程队列管理
	线程封装

	李亦
	2006.6.15
/*///////////////////////////////////////////////////////////////////////////////////

#include "threadManager.h"



namespace RPGServer
{

//////////////////////////////////////////////////////////////
//线程队列
ThreadQueue::ThreadQueueThread::ThreadQueueThread(ThreadQueue *q)
{
   mThreadQueue = q;
}

U32 ThreadQueue::ThreadQueueThread::run()
{
   mThreadQueue->threadStart();

   mThreadQueue->lock();
   ThreadStorage &sto = mThreadQueue->getStorage();
   sto.set((void *) 0);
   mThreadQueue->unlock();

   for(;;)
      mThreadQueue->dispatchNextCall();
   return 0;
}


////////////////////////////////////////////////////////////////////

ThreadQueue::ThreadQueue(U32 threadCount)
:mSemaphore(0)
{
   mStorage.set((void *) 1);
   for(U32 i = 0; i < threadCount; i++)
   {
      Thread *theThread = new ThreadQueueThread(this);
      mThreads.push_back(theThread);
      theThread->start();
   }
   mMutex = Mutex::createMutex();

}




ThreadQueue::~ThreadQueue()
{
   Mutex::destroyMutex(mMutex);
   mMutex = NULL;
}

void ThreadQueue::dispatchNextCall()
{
   mSemaphore.wait(true);
   lock();
   if(mThreadCalls.size() == 0)
   {
      unlock();
      return;
   }
   Functor *c = mThreadCalls.first();
   mThreadCalls.pop_front();
   unlock();
   c->dispatch(this);
   delete c;
}


void ThreadQueue::postCall(Functor *theCall)
{
   lock();
   if(isMainThread())
   {
      mThreadCalls.push_back(theCall);
      unlock();
      mSemaphore.increment();
   }
   else
   {
      mResponseCalls.push_back(theCall);
      unlock();
   }
}

void ThreadQueue::dispatchResponseCalls()
{
   lock();
   for(S32 i = 0; i < mResponseCalls.size(); i++)
   {
      Functor *c = mResponseCalls[i];
      c->dispatch(this);
      delete c;
   }
   mResponseCalls.clear();
   unlock();
}

};