threadmanager.h

五行MMORPG引擎系统V1.0

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

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

#ifndef _THREADMANAGER_H_
#define _THREADMANAGER_H_

#ifndef _PLATFORMSEMAPHORE_H_
#include "platform/platformMutex.h"
#endif

#ifndef _METHODDISPATCH_H_
#include "server/methodDispatch.h"
#endif

#ifndef _PLATFORMSEMAPHORE_H_
#include "platform/platformSemaphore.h"
#endif

#ifndef _PLATFORMTHREAD_H_
#include "platform/platformThread.h"
#endif


namespace RPGServer
{




/// Platform independent per-thread storage class.
class ThreadStorage
{
   U32 mTlsIndex;
public:
   /// ThreadStorage constructor.
   ThreadStorage();
   /// ThreadStorage destructor.
   ~ThreadStorage();

   /// returns the per-thread stored void pointer for this ThreadStorage.  The default value is NULL.
   void *get();
   /// sets the per-thread stored void pointer for this ThreadStorage object.
   void set(void *data);
};




/// Managing object for a queue of worker threads that pass
/// messages back and forth to the main thread.  ThreadQueue
/// methods declared with the TNL_DECLARE_THREADQ_METHOD macro
/// are special -- if they are called from the main thread,
/// they will be executed on one of the worker threads and vice
/// versa.
class ThreadQueue //: public Object
{
   class ThreadQueueThread : public Thread
   {
      ThreadQueue *mThreadQueue;
      public:
      ThreadQueueThread(ThreadQueue *);
      U32 run();
   };

   friend class ThreadQueueThread;


   Vector<Thread *> mThreads;			/// ThreadQueue工作线程列表 
   Vector<Functor *> mThreadCalls;	/// 工作线程执行函数列表
   Vector<Functor *> mResponseCalls;/// 工作线程响应函数列表

   SemaphoreInstance		mSemaphore;	/// 工作线程工作协调信号管理
	void*						mMutex;		/// 工作线程执行函数调用信号管理

   /// Storage variable that tracks whether this is the main thread or a worker thread.
   ThreadStorage	mStorage;

protected:
   /// 队列成员访问存取锁
   void lock()		{ Mutex::lockMutex(mMutex, true); }
   void unlock()  { Mutex::unlockMutex(mMutex); }


   /// Posts a marshalled call onto either the worker thread call list or the response call list.
   void postCall(Functor *theCall);
   /// Dispatches the next available worker thread call.  Called internally by the worker threads when they awaken from the semaphore.
   void dispatchNextCall();


   /// helper function to determine if the currently executing thread is a worker thread or the main thread.
   bool isMainThread();
   ThreadStorage &getStorage();


   /// called by each worker thread when it starts for subclass initialization of worker threads.
   virtual void threadStart(){}


public:
   /// ThreadQueue constructor.  threadCount specifies the number of worker threads that will be created.
   ThreadQueue(U32 threadCount);
   ~ThreadQueue();

   /// Dispatches all ThreadQueue calls queued by worker threads.  This should
   /// be called periodically from a main loop.
   void dispatchResponseCalls();
};


//////////////////////////////////////////////////////////////
inline bool ThreadQueue::isMainThread() 
{ 
	return (bool) mStorage.get(); 
}
inline ThreadStorage &ThreadQueue::getStorage() 
{ 
	return mStorage; 
}
 



//////////////////////////////////////////////////////////////
/// Declares a ThreadQueue method on a subclass of ThreadQueue.
#define DECLARE_THREADQ_METHOD(func, args) \
   void func args; \
   void func##_body args

/// Declares the implementation of a ThreadQueue method.
#define IMPLEMENT_THREADQ_METHOD(className, func, args, argNames) \
   void className::func args { \
   FunctorDecl<void (className::*)args> *theCall = new FunctorDecl<void (className::*)args>(&className::func##_body); \
   theCall->set argNames; \
   postCall(theCall); \
   }\
   void className::func##_body args


};//RPGServer

#endif