multithreading approach

#pragma once//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////  MyThreadPool.h////  Copyright (c) Ivica Kolic 2010 (Ivica.Kolic@gmail.com)////  Description://    This is very simple, but very fast thread pool class implemented using critical sections (for speed).//    //  Usage:////    pool.Init(THREAD_NUMBER);  // This will create THREAD_NUMBER threads in the pool. Called once per application.//    ...//    pool.Start(func1, param1); // Execute func1(param1) as a thread (one of the worker threads will do this)//    pool.Start(func2, param2); // Execute func1(param1) as a thread//    pool.Start(funcN, paramN); // Execute funcN(paramN) as a thread (N should be smaller than THREAD_NUMBER//    pool.Wait();               // This will wait all the threads to finish working - used for chaining.//    pool.Start(funcA, paramA); // Execute funcA(paramA) as a thread//    pool.Start(funcB, paramB); // Execute funcB(paramB) as a thread//    pool.Start(funcC, paramC); // Execute funcC(paramC) as a thread//    pool.Wait();               // This will wait all the threads to finish working - used for chaining.//    ...//    pool.Exit();               // This will wait until every thread is dead and clear the pool. Called at the end.//////  Theory:////  Initial idea that didn't work://    Basic idea was that worker thread should infinitely loop with the EnterCriticalSection(allowWorkStart) at the //    beggining of the loop. Start function would release this critical section for the thread so that thread could//    do work until it reaches EnterCriticalSection(allowWorkStart) again and locks itself once more. We would also//    have one other critical section that indicates work (set by thread) and expected by the Wait command.//    The problem was that Start (which is in the main thread) can't unlock critical section for some other thread.//    Also, worker theread can't lock itsef with the same EnterCriticalSection line (recursion count is just increased).//    //  Idea that worked://    Since two critical sections wouldn't do, I had to get more creative and ended up using four critical sections://    One to prevent thread from starting work                                    (csAllowWorkStart)//    One to prevent thread from ending work                                      (csAllowWorkEnd)//    One to force Wait function to wait                                          (csAllowWait)//    One to force Start function to wait until thread is done with previous work (csAllowStart)////////////////////////////////////////////////////////////////////////////////////////////////////////////////////#include <windows.h>#include <process.h>class CMyThreadPool{private: // Data used to control every worker thread:  struct SThreadData  {    CRITICAL_SECTION csAllowWorkStart; // We allow worker thread to start working (but not to end working - loop back at the start and takes this CS again).    CRITICAL_SECTION csAllowWait;      // Wait function waits for this (that is set when worker thread finished the job).      CRITICAL_SECTION csAllowWorkEnd;   // While waiting for threads to finish we allow work end, but not to start again.    CRITICAL_SECTION csAllowStart;     // Stops from Start being used again until thread finishes. This prevent going from Wait to Start and skipping thread that needs to loop back to the beggining.    void (*pFunc)(void*);              // Current function that worker thread must execute    void* pParam;                      // Parameters to the function  };private:  // Worker thread:  //////////////////////////////////////////////////////////////////////////////////////////////////////////////////  // WorkerThread waits for jobs from Start and blocks Wait command until work is done. It allso prevents Start  // to issue new work until old one is done.  //////////////////////////////////////////////////////////////////////////////////////////////////////////////////  static void __cdecl WorkerThread(LPVOID pParam)  {    SThreadData& td = *(SThreadData*)pParam;    while(1)    {      EnterCriticalSection(&td.csAllowWait);      // This prevents main program to give us right to end work and then takes it right away (if Wait is called imediately after Start)      LeaveCriticalSection(&td.csAllowStart);     // Now that we can finish (we can't start) - we don't need that information any more.      EnterCriticalSection(&td.csAllowWorkStart); // We can't start work until main program has something for us.      LeaveCriticalSection(&td.csAllowWorkStart); // Now that we know that we can work, we don't need that flag any more.            if( td.pFunc )                              // If there is something to do (Start has given us work)      {        td.pFunc(td.pParam);                      // Execute task        EnterCriticalSection(&td.csAllowStart);   // Prevent Start to give us new work untill we loop at the beggining.        LeaveCriticalSection(&td.csAllowWait);    // We are not working any more (Wait command expects this)        EnterCriticalSection(&td.csAllowWorkEnd); // We can't finish before main program is ok with that (it needs time to prevent new work - this is all done in Wait commant)        LeaveCriticalSection(&td.csAllowWorkEnd); // Now that we can finish (we can't start) - we don't need that information any more.      }      else break;                                 // If there is no new work - exit the thread. This is set by Exit command.    }    LeaveCriticalSection(&td.csAllowWait);        // Time to say goodbye.  }private:  unsigned int m_nThreadNumber;                   // Number of worker threads  SThreadData* m_pThreadData;                     // Control data for each worker threadpublic:  CMyThreadPool():m_nThreadNumber(0),m_pThreadData(0){}  ~CMyThreadPool(){ Exit(); }  //////////////////////////////////////////////////////////////////////////////////////////////////////////////////  // Init creates all the control data for the worker threads and starts them.  //////////////////////////////////////////////////////////////////////////////////////////////////////////////////  void Init(unsigned int nThreadNumber)  {    Exit();    m_nThreadNumber = nThreadNumber;    m_pThreadData   = new SThreadData[m_nThreadNumber];    for( int t=0; t<m_nThreadNumber; ++t )    {      InitializeCriticalSection(&m_pThreadData[t].csAllowStart);      InitializeCriticalSection(&m_pThreadData[t].csAllowWorkStart);      InitializeCriticalSection(&m_pThreadData[t].csAllowWait);      InitializeCriticalSection(&m_pThreadData[t].csAllowWorkEnd);      EnterCriticalSection(&m_pThreadData[t].csAllowWorkStart);      // This will prevent thread from starting to work until we have some work for it.      EnterCriticalSection(&m_pThreadData[t].csAllowStart);          // This will prevent Wait Start loop that skips thread.      m_pThreadData[t].pFunc  = NULL;      m_pThreadData[t].pParam = NULL;      _beginthread(CMyThreadPool::WorkerThread, 0, &m_pThreadData[t]);// Start the thread that will wait at    }    Sleep(2000);  // Now we sleep to give worker threads time to aquire csAllowWait flag (critical section).  }  //////////////////////////////////////////////////////////////////////////////////////////////////////////////////  // Start gives work to the worker thread and allows threads to work (csAllowWorkStart) but prevents them from  // stopping (csAllowWorkEnd). Start will wait until thread is done with its previous work (csAllowStart is locked  // by the thread).  //////////////////////////////////////////////////////////////////////////////////////////////////////////////////  bool Start(void( *pFunc )( void * ) , void* pParam)  {    if( pFunc )    {      for( int t=0; t<m_nThreadNumber; ++t )      {        if( !m_pThreadData[t].pFunc )  // Thread is not working        {          EnterCriticalSection(&m_pThreadData[t].csAllowStart);     // Until thread is done with the previous work, Start can't issue a new work          LeaveCriticalSection(&m_pThreadData[t].csAllowStart);     // No need for this info any more          m_pThreadData[t].pFunc  = pFunc;          m_pThreadData[t].pParam = pParam;          EnterCriticalSection(&m_pThreadData[t].csAllowWorkEnd);   // Before we allow work, we must prevent end of work.          LeaveCriticalSection(&m_pThreadData[t].csAllowWorkStart); // Now thread will be able to work but not to stop working.          return true;        }      }    }    return false;                                                   // If there are no available slots - Start will fail to assign job.  }  //////////////////////////////////////////////////////////////////////////////////////////////////////////////////  // Wait waits untill every thread is done with the work.  // Once that is done, threads are allowed to end the work but not to beggin a new work.  //////////////////////////////////////////////////////////////////////////////////////////////////////////////////  void Wait()  {    for( int t=0; t<m_nThreadNumber; ++t )    {      if( m_pThreadData[t].pFunc )                                  // Active thread      {        EnterCriticalSection(&m_pThreadData[t].csAllowWait);        // Wait until it stops working        LeaveCriticalSection(&m_pThreadData[t].csAllowWait);        // OK we know that thread finished with work and we don't need that information any more.        m_pThreadData[t].pFunc  = NULL;                             // Set empty data - no work        m_pThreadData[t].pParam = NULL;                             // Set empty data - no work        EnterCriticalSection(&m_pThreadData[t].csAllowWorkStart);   // Prevent new work (until Start has something new to do)        LeaveCriticalSection(&m_pThreadData[t].csAllowWorkEnd);     // Allow threads to end current work              }    }  }  //////////////////////////////////////////////////////////////////////////////////////////////////////////////////  // Exit waits till every thread is done and then orders them to finish.  // After 1s threads should be dead and memory is then released.  //////////////////////////////////////////////////////////////////////////////////////////////////////////////////  void Exit()  {    Wait();    for( int t=0; t<m_nThreadNumber; ++t )    {      m_pThreadData[t].pFunc = NULL;                                // To mark end      LeaveCriticalSection(&m_pThreadData[t].csAllowWorkStart);     // Kick one more time to force threads to exit    }    Sleep(1000);                                                    // Sleeping 1s should do it    for( int t=0; t<m_nThreadNumber; ++t )    {      DeleteCriticalSection(&m_pThreadData[t].csAllowWorkStart);      DeleteCriticalSection(&m_pThreadData[t].csAllowStart);      DeleteCriticalSection(&m_pThreadData[t].csAllowWait);      DeleteCriticalSection(&m_pThreadData[t].csAllowWorkEnd);    }    m_nThreadNumber = 0;    delete [] m_pThreadData;    m_pThreadData = NULL;  }};

SYSTEM_INFO sysInfo;GetSystemInfo( &sysInfo );m_nCoreNumber = sysInfo.dwNumberOfProcessors;m_ThreadPool.Init(m_nCoreNumber);

m_ThreadPool.Start(&Function1, (void*)&params1);m_ThreadPool.Start(&Function1, (void*)&params2);...m_ThreadPool.Start(&FunctionN, (void*)&paramsN);m_ThreadPool.Wait(); // Wait until all the work is done - chainingm_ThreadPool.Start(&FunctionA, (void*)&paramsA);m_ThreadPool.Start(&FunctionB, (void*)&paramsB);m_ThreadPool.Wait(); // Wait until all the work is done - chaining

        typedef boost::shared_ptr<boost::thread> ThreadPtr;        typedef std::vector<ThreadPtr> ThreadArrayList;	template <typename JobType> class JobQueue : boost::noncopyable	{	public:		JobQueue() : bExit(false)		{}		~JobQueue()		{			{				boost::mutex::scoped_lock lock(MonitorMutex);				bExit = true;			}		}		void push(const JobType& newData)		{			boost::mutex::scoped_lock lock(MonitorMutex);			jobTypeQueue.push(newData);			NextAvailable.notify_one();		}				JobType pop()		{			boost::mutex::scoped_lock lock(MonitorMutex);			if (bExit)			{				return 0;			}			if(jobTypeQueue.empty())			{				NextAvailable.wait(lock);			}						JobType temp(jobTypeQueue.front());			jobTypeQueue.pop();						return temp;		}			private:		std::queue<JobType> jobTypeQueue;		boost::mutex MonitorMutex;		boost::condition_variable NextAvailable;			bool bExit;	};		typedef Framework::Utility::JobQueue<GameComponent*> GameComponentJobQueue;	typedef boost::shared_ptr<GameComponentJobQueue > GameComponentJobQueuePtr;	class WorkerThread;	typedef boost::shared_ptr<WorkerThread> WorkerThreadPtr;	typedef std::vector<WorkerThreadPtr> WorkerArrayList;	class WorkerThread    {	public:		WorkerThread(GameComponentJobQueuePtr gameJobQueue) : bExit(false), jobQueue(gameJobQueue)		{		}		~WorkerThread()		{			{				boost::mutex::scoped_lock lock(exitMutex);				bExit = true;			}		}		void ProcessJobs()		{			while(true)			{				{					boost::mutex::scoped_lock lock(exitMutex);					if (bExit)						return;				}				GameComponent* component = jobQueue->pop();				if (component == 0)				{					return;				}				else				{					component->Proceed();				}			}			this->~WorkerThread();		}	private:		GameComponentJobQueuePtr jobQueue;		boost::mutex exitMutex;				bool bExit;    };		class ThreadPool    {	public:		ThreadPool()		{			size = 0;		}				~ThreadPool()		{			if (threads.size() != 0)			{				for (unsigned int i=0; i < threads.size(); ++i)				{					if(threads != 0)					{						threads->join();					}				}			}		}				void Initialize(unsigned int numWorkerThreads, GameComponentJobQueuePtr gameJobQueue)		{			threads.resize(numWorkerThreads);			jobQueue = gameJobQueue;						for (unsigned int i=0; i < numWorkerThreads; ++i)			{				threads.reset(new boost::thread(boost::bind(&WorkerThread::ProcessJobs, new WorkerThread(gameJobQueue))));			}			size = numWorkerThreads;		}		unsigned int GetSize()		{			return this->size;		}		void AddJob(GameComponent* component)		{			this->jobQueue->push(component);		}			private:		unsigned int size;		ThreadArrayList threads;		GameComponentJobQueuePtr jobQueue;    };

boost::shared_ptr<Utility::ThreadPool> threadPool;GameComponentJobQueuePtr jobQueue;void init(){  SYSTEM_INFO sysInfo;  GetSystemInfo( &sysInfo );    this->jobQueue.reset(new Framework::Utility::GameComponentJobQueue());  this->threadPool.reset(new ThreadPool());  this->threadPool->Initialize(sysInfo.dwNumberOfProcessors, this->jobQueue);}int main(){  while(blabla gameloop)  {    someone { this->jobQueue->push(GameComponentThatNeedsProcessing)  }    if (atsomepoint)      return 0;  }}