threadpool.cpp

线程池+socket

/**************************************************************
 * ThreadPool.cpp : implementation file
 * Copyright 2004-2005 Aven
/**************************************************************
 * ThreadPool.cpp : implementation file
 * Copyright 2004-2005 TianMuLingHang
 * All rights reserved.
 * Author: Zeng Wenchuan
 * Date:   Jan 9, 2005
 * Modify: 
 * Date:
 * Email:  [e-mail]
 * Site:   http://zeng_aven.go.nease.net/
 *************************************************************/
 
#include "ThreadPool.h"
#include "SocketBase.h"
#include "SocketThread.h"

CThreadPool::CThreadPool(int num) : 
	m_nMaxThreadNum(MAX_THREAD_NUM), m_nWaitSec(0), m_nWaitNec(30000000), m_nThreadNum(0)
	
{
	pthread_mutex_init(&m_BusyMutex, NULL);
	pthread_mutex_init(&m_IdleMutex, NULL);
	pthread_mutex_init(&m_WaitMutex, NULL);
	pthread_cond_init(&m_WaitCond, NULL);
	
	if (num > MAX_THREAD_NUM)
		num	= MAX_THREAD_NUM;
		
	for (int i=0; i<num; i++)
	{
		CreateNewThread();
	}
	
	m_nNormalThread		= m_nThreadNum;
	m_nMaxJobsPending	= m_nThreadNum / 2;
	m_nMaxIdleNum	 	= m_nThreadNum / 2;
	m_nMinIdleNum		= m_nThreadNum / 4;
}

CThreadPool::CThreadPool()
{
	pthread_mutex_init(&m_BusyMutex, NULL);
	pthread_mutex_init(&m_IdleMutex, NULL);
}

CThreadPool::~CThreadPool()
{
	pthread_mutex_unlock(&m_IdleMutex);
	for (t_iIdleThList iIdle=m_IdleThList.begin(); iIdle!=m_IdleThList.end(); iIdle++)
	{
		delete	*iIdle;
		*iIdle	= NULL;
	}
	pthread_mutex_unlock(&m_IdleMutex);
	
	pthread_mutex_lock(&m_BusyMutex);
	for (t_iBusyThList iBusy=m_BusyThList.begin(); iBusy!=m_BusyThList.end(); iBusy++)
	{
		delete	*iBusy;
		*iBusy	= NULL;
	}
	
	m_BusyThList.clear();
	pthread_mutex_unlock(&m_BusyMutex);
	
	pthread_mutex_destroy(&m_BusyMutex);
	pthread_mutex_destroy(&m_IdleMutex);
}

bool CThreadPool::ThreadProc(CSocketBase *pSocket)
{
	ERROR_TRY
	//太多任务并行
	while (m_nMaxJobsPending <= GetBusyNum()) 
	{
		cout << "Wait" << endl;
		Wait();
	}
	CSocketThread *pThread = GetIdleThread();
	
	if (pThread) {
		MoveToBusyList(pThread);
		pThread->SetNewSocket(pSocket);
	}
	
	if (GetIdleNum() < m_nMinIdleNum) {
		//空闲线程不足！
		CreateIdleThread(m_nMinIdleNum - GetIdleNum());
		cout << "空闲线程不足！" << endl;
	}
	//空闲线程太多，需要删除
	else if ((GetIdleNum() > m_nMaxIdleNum) && (m_nThreadNum > m_nNormalThread)) {
		for (int i=0; i< (GetIdleNum()-m_nMaxIdleNum); i++)
			DeleteIdleThread();
			
		cout << "空闲线程太多" << endl;
	}
		
	ERROR_UNKNOW_RETURN(CTHREADPOOL, "ThreadProc", true)
}

bool CThreadPool::CreateNewThread()
{
	CSocketThread *pThread = new CSocketThread;
	if (pThread->Start()) {
		pthread_mutex_lock(&m_IdleMutex);
		pThread->SetIdleThList(&m_IdleThList);
		pThread->SetBusyThList(&m_BusyThList);
		pThread->SetThreadPool(this);
		m_IdleThList.push_back(pThread);
		m_nThreadNum++;
		pthread_mutex_unlock(&m_IdleMutex);
	}
	else 
		delete pThread;
}

CSocketThread *CThreadPool::GetIdleThread()
{
	if (GetIdleNum() > 0) {
		pthread_mutex_lock(&m_IdleMutex);
		CSocketThread	*pThread = (CSocketThread *)m_IdleThList.front();
		pthread_mutex_unlock(&m_IdleMutex);
		return 	pThread;
	}
	
	return NULL;
}

void CThreadPool::MoveToIdleList(CSocketThread *pThread)
{
	pthread_mutex_lock(&m_IdleMutex);
	m_IdleThList.push_back(pThread);
	pthread_mutex_unlock(&m_IdleMutex);
	
	if (GetBusyNum() <= 0) 
		return;
		
	pthread_mutex_lock(&m_BusyMutex);
	m_BusyThList.remove(pThread);
	pthread_mutex_unlock(&m_BusyMutex);
}

void CThreadPool::MoveToBusyList(CSocketThread *pThread)
{
	pthread_mutex_lock(&m_BusyMutex);
	m_BusyThList.push_back(pThread);
	pthread_mutex_unlock(&m_BusyMutex);
	
	if (GetBusyNum() <= 0)
		return;
		
	pthread_mutex_lock(&m_IdleMutex);
	m_IdleThList.remove(pThread);
	pthread_mutex_unlock(&m_IdleMutex);
}

int CThreadPool::GetBusyNum() 
{
	pthread_mutex_lock(&m_BusyMutex);
	int num = m_BusyThList.size();
	pthread_mutex_unlock(&m_BusyMutex);
	return num;
}

int CThreadPool::GetIdleNum()
{
	pthread_mutex_lock(&m_IdleMutex);
	int num = m_IdleThList.size();
	pthread_mutex_unlock(&m_IdleMutex);
	return num;
}

void CThreadPool::CreateIdleThread(const int num)
{
	ERROR_TRY
	
	pthread_mutex_lock(&m_IdleMutex);
	if (m_nThreadNum == m_nMaxThreadNum)
		return;
	if (m_nThreadNum + num > m_nMaxThreadNum) {
		CreateIdleThread(m_nMaxThreadNum - m_nThreadNum);
	}
	pthread_mutex_unlock(&m_IdleMutex);
	
	ERROR_UNKNOW(CTHREADPOOL, "CreateIdleThread")
}

void CThreadPool::DeleteIdleThread()
{
	ERROR_TRY
	
	pthread_mutex_lock(&m_IdleMutex);
	m_IdleThList.pop_back();
	pthread_mutex_unlock(&m_IdleMutex);
	
	ERROR_UNKNOW(CTHREADPOOL, "DeleteIdleThread")
}

void CThreadPool::Wait()
{
	ERROR_TRY

	pthread_mutex_lock(&m_WaitMutex);
	m_delay.tv_sec = m_nWaitSec;
	m_delay.tv_nsec = m_nWaitNec;
	pthread_cond_timedwait(&m_WaitCond, &m_WaitMutex, &m_delay);
	pthread_mutex_unlock(&m_WaitMutex);
	
	ERROR_UNKNOW(CTHREADPOOL, "Wait")
}

void CThreadPool::GetWaitTime(int &nWaitSec, long &nWaitNec)
{
	nWaitSec	= m_nWaitSec;
	nWaitNec	= m_nWaitNec;
}

void CThreadPool::SetWaitTime(const int nWaitSec, const long nWaitNec)
{
	m_nWaitSec	= nWaitSec;
	m_nWaitNec	= nWaitNec;
}