threadpool.h

网络socket,IO,线程池

#ifndef _M_THREAD_POOL_2004_7_28_ZHY_H
#define _M_THREAD_POOL_2004_7_28_ZHY_H

#include <map>
using namespace std;

#include "..\include\Base.h"
#include "..\include\IndexAlloctor.h"

#include "../Thread/Thread.h"
using namespace ThreadLib;

#include "Pool.h"
using namespace Pool;

#define THREADPOOL_LOW_SIZE 	10
#define THREADPOOL_MAX 		  	50

template <typename T> class ProcessThread;

template <typename T>
class CThreadPool:public Thread,private NoAssign
{
public:
	////////////////////////////////////////////////////////////////////////////
	// Constructors / Destructor                                              //
	////////////////////////////////////////////////////////////////////////////
	
	// Construct
	CThreadPool(TaskPool<T>* pTaskPool,int nLow = THREADPOOL_LOW_SIZE,int nMaxCount = THREADPOOL_MAX);	
		
	// Destrutor
	~CThreadPool();
	
	virtual void RunInfo(int nIndex)
	{
	}
	
	////////////////////////////////////////////////////////////////////////////
	// Exit Operations							  //
	////////////////////////////////////////////////////////////////////////////
	int			   				StopEx();	
	bool						ExitAllThread();	
	virtual void 				ExitThread(int nIndex);
	
protected:
	int	 					    Run(void);
	inline int				    SocketPoolSize()
	{
		return m_pTaskPool.GetTaskSize();
	}

	////////////////////////////////////////////////////////////////////////////
	// Inner Operations														  //
	////////////////////////////////////////////////////////////////////////////
	int			  				AllocThreadID();
	friend class 				ProcessThread;

private:
	int							m_nLow;    
	int							m_nMaxSize;
	// count of the child thread and his thread-safe lock
	int							m_nProcessingCount;
	CCriSection					m_CriProCount; 
	map<int,ProcessThread<T>*>	m_allThread;
	CCriSection					m_CrivectorThreadID; 
	vector<int>					m_vThreadID;
	CEvent						m_EvProCount;
	IndexAlloctor				m_indexAlloc;
	bool						m_childCanExit;
	TaskPool<T> *				m_pTaskPool;
};
	
template <typename T>	
class ProcessThread:public Thread
{
public:
	ProcessThread(CThreadPool<T>* pMangerPool,int nIndex);
	~ProcessThread();
	
	void				Exit();
		
	protected:
	int					Run(void);
		
	CThreadPool*		m_pManagerPool;
	bool				m_bWantExit;
	int					m_nIndex;
	T*					m_pSocket;
};
			

////////////////////////////////////////////////////////////////////////////
// Constructors / Destructor                                              //
////////////////////////////////////////////////////////////////////////////
	
// Construct
template <typename T>
CThreadPool<T>::CThreadPool(TaskPool<T>* pTaskPool,int nLow ,int nMaxCount ):m_pTaskPool(pTaskPool),m_nLow(nLow),m_nMaxSize(nMaxCount),
									  m_nProcessingCount(0),m_EvProCount(FALSE,TRUE),m_bChildCanExit(false)
									  
{
	
}

// Destructor
template <typename T>
CThreadPool<T>::~CThreadPool()
{
	StopEx();
}


////////////////////////////////////////////////////////////////////////////
// Exit Operations							  //
////////////////////////////////////////////////////////////////////////////
template <typename T>
int	CThreadPool<T>::StopEx()
{
	ExitAllThread();
	std::map<int,ProcessThread*>::const_iterator	it;
	for(it = m_allThread.begin(); it != m_allThread.end(); ++it)
		delete it->second;	
}

template <typename T>
bool	CThreadPool<T>::ExitAllThread()
{
	std::map<int,ProcessThread*>::const_iterator	it;
	for(it = m_allThread.begin(); it != m_allThread.end(); ++it)
	{
		if(it->second)
			it->second->Exit();
	}

	while(m_nProcessingCount > 0)
	{
		if(WAIT_OBJECT_0 == ::WaitForSingleObject(m_EvProCount,WAIT_TIME_END))
			m_EvProCount.ResetEvent();
		Sleep(0);
	}
	return 0;
}

template <typename T>
void CThreadPool<T>::ExitThread(int nIndex)
{
		{		
		MLock<MCriticalSection> lock(m_CriProCount);
		m_nProcessingCount--;				
		if(m_nProcessingCount <= m_nLow )
			m_bChildCanExit = false;
		}
		m_EvProCount.SetEvent();

		m_indexAlloc.DeAllocIndex(nIndex);

		vector<int>::iterator it;
		MLock<MCriticalSection> lockVector(m_CrivectorThreadID);
		it = find(m_vThreadID.begin(),m_vThreadID.end(),nIndex);
		if(it != m_vThreadID.end())
			m_vThreadID.erase(it);
}

template <typename T>
int	CThreadPool<T>::Run(void)
{
	while(!g_exit)
	{
	// if there is a task ,Push the task into the pool
	// 	then to judge the neccenality to create a new thread
	// else 
	//      judge if task pool is low than thread count and if there is neccecelity to end a thread
	//		then set the m_bChildCanExit to true	
	}
	
	return 0;
}					

////////////////////////////////////////////////////////////////////////////
// Inner Operations							  //
////////////////////////////////////////////////////////////////////////////
template <typename T>
int  CThreadPool<T>::AllocThreadID()
{
	int nIndex;
	while(-1 == (nIndex = m_indexAlloc.AllocIndex()))
	{
			srand(time(NULL));
			int randTime = rand() % 5 * 10;
			Sleep(randTime);
	}
	return nIndex;
}

//==================================================class	ProcessThread====================================================//

////////////////////////////////////////////////////////////////////////////
// Constructors / Destructor                                              //
////////////////////////////////////////////////////////////////////////////
	
// Construct
template <typename T>	
ProcessThread<T>::ProcessThread(CThreadPool<T>* pMangerPool,int nIndex):m_pManagerPool(pMangerPool),m_nIndex(nIndex),m_bWantExit(false),m_pSocket(NULL)
{
}

template <typename T>	
ProcessThread<T>::~ProcessThread()
{
	Exit();
}

template <typename T>	
int	ProcessThread<T>::Run(void)
{
	while(	(!g_exit && !m_bWantExit) &&
			!m_pManagerPool->m_bChildCanExit)
	{
		// get task from task pool and do your thing...
		T* pT = m_pTaskPool->TopTask_Time();
		pT->();		
	}
	
	Exit();
	m_pManagerPool->ExitThread(m_nIndex);
	return 0;
}

template <typename T>	
void	ProcessThread<T>::Exit()
{
	if(m_pSocket)
		m_pSocket->close();
	m_bWantExit = true;
}

#endif