pool.h.bak

网络socket,IO,线程池

#ifndef _M_POOL_2004_7_29_ZHY_H
#define _M_POOL_2004_7_29_ZHY_H

#include <queue>
using namespace std;

// windows header
#include <afxmt.h>

#include "../Synchronization/event.h"
#include "../Synchronization/Lock.h"
using namespace Synchronization;

#define TASK_POOL_MAX		200
#define TOP_TASK_WAIT_TIME	3000

template<class TaskObj>
class TaskPool:private NoAssign
{
public:
	////////////////////////////////////////////////////////////////////////////
	// Constructors / Destructor                                              //
	////////////////////////////////////////////////////////////////////////////
	
	// Construct
	TaskPool(int maxSize = TASK_POOL_MAX):m_nMaxSize(maxSize)
	{
		m_EvNotFull.signal();
		m_EvNotEmpty.reset();
	}
	
	////////////////////////////////////////////////////////////////////////////
	// Inline Set/Get Operations                                              //
	////////////////////////////////////////////////////////////////////////////
	
	inline int 	GetTaskSize()
	{
		CLock<CCriSection> lock(m_CriQueue);		
		return m_taskQueue.size();
	}
	
	////////////////////////////////////////////////////////////////////////////
	// Operations		                                                  //
	////////////////////////////////////////////////////////////////////////////
	virtual	int	operator()()
	{}
	
	void PushTask(TaskObj* pObj) throw (MMutexException);
	TaskObj* TopTask() throw (MMutexException);
	TaskObj* TopTask_Time(int nMSecond) throw (MMutexException); // 毫秒
		
protected:
	int 				m_nMaxSize;	
	queue<TaskObj*>			m_taskQueue; 

	CCriSection		m_CriQueue; // 访问任务队列时互斥
	ManualEvent			m_EvNotFull;
	ManualEvent			m_EvNotEmpty;
};

////////////////////////////////////////////////////////////////////////////
// Operations		                                                  //
////////////////////////////////////////////////////////////////////////////
	
template<class TaskObj>
void TaskPool<TaskObj>::PushTask(TaskObj* pObj)
{
	// 等待未满事件
	::WaitForSingleObject(m_EvNotFull, INFINITE);
	
	try
	{
		m_CriQueue.acquire();						
		// 把任务放入队列，如果队列满则reset掉notfull信号
		m_taskQueue.push(pObj);
		if(m_taskQueue.size() >= m_nMaxSize)
			m_EvNotFull.reset();						

		m_CriQueue.release();

		m_EvNotEmpty.signal();
	}
	catch(MMutexException&)
	{
		m_CriQueue.release();		
		AfxMessageBox("void TaskPool<TaskObj>::PushTask(TaskObj* pObj)");
		throw;				
	}
}

template<class TaskObj>
TaskObj* TaskPool<TaskObj>::TopTask()
{
	try
	{
		m_CriQueue.acquire();						
		
		if(m_taskQueue.empty())
		{
			m_EvNotFull.signal();					
			m_CriQueue.release();
			return 0;
		}
		TaskObj* pObj= m_taskQueue.front();
		m_taskQueue.pop();
		// 是否被取空
		if(m_taskQueue.empty())
			m_EvNotEmpty.reset();

		m_EvNotFull.signal();						
		m_CriQueue.release();
		return pObj;
	}
	catch(const MMutexException& )
	{
		m_CriQueue.release();		
		AfxMessageBox("void TaskPool<TaskObj>::TopTask())");
		throw;				
	}
	return 0;
}

template<class TaskObj>
TaskObj* TaskPool<TaskObj>::TopTask_Time(int nMSecond) 
{
	if(WAIT_OBJECT_0 == ::WaitForSingleObject(m_EvNotEmpty, nMSecond))
		return TopTask();
	else
		return 0;
}

#endif