thread.cpp

经常用到的 socketAPI

#include "thread.h"

#include <iostream>

#ifndef WIN32
#include <sys/time.h>
#endif

namespace mylib{
CThMutex::CThMutex()
{
	#if defined(WIN32)
	::InitializeCriticalSection(&m_tMutex);	
	#else
	pthread_mutex_init(&m_tMutex, NULL);
	
	#endif
}

CThMutex::~CThMutex()
{
        #if defined(WIN32)
	      ::DeleteCriticalSection(&m_tMutex);
        #else
	      int iResult = pthread_mutex_destroy(&m_tMutex);
	     
	      if (iResult)
		  throw CThreadException(strerror(iResult), __FILE__, __LINE__); 
	     
        #endif
}

bool CThMutex::Lock()
{
        #if defined(WIN32)
	    ::EnterCriticalSection(&m_tMutex);
        #else
        	int iResult = pthread_mutex_lock(&m_tMutex);
        	if (iResult)
        		throw CThreadException(strerror(iResult), __FILE__, __LINE__);
        #endif
	return true;
}

bool CThMutex::TryLock()
{
        #if defined(WIN32)
	       ;// return ::TryEnterCriticalSection(&m_tMutex);
	        return true;
        #else
        	int iResult = pthread_mutex_trylock(&m_tMutex);
        	if (iResult)
        	{
        		if (iResult == EBUSY)
        			return false;					
        		throw CThreadException(strerror(iResult), __FILE__, __LINE__);
        	}	
			return true;
        #endif
}


bool CThMutex::Unlock()
{
#if defined(WIN32)
	
    ::LeaveCriticalSection(&m_tMutex);
	
#else
	int iResult = pthread_mutex_unlock(&m_tMutex);
	if (iResult)
		throw CThreadException(strerror(iResult), __FILE__, __LINE__);
#endif

	return true;
}

	
CThCond::CThCond()
{   
    #if defined(WIN32)
	m_tCond=::CreateEvent(NULL,// Security
		        TRUE,//Manual
				FALSE,//don't own on create
				NULL//name
				);
	if(m_tCond==NULL)
		throw CThreadException(STRERROR_WIN(WIN32_SYS),__FILE__, __LINE__);
   #else		
	pthread_cond_init(&m_tCond, NULL);   
	
    #endif
}


CThCond::~CThCond()
{
    #if defined(WIN32)
    if(m_tCond)::CloseHandle(m_tCond);
    #else
	int iResult = pthread_cond_destroy(&m_tCond);
	if (iResult)
		throw CThreadException(strerror(iResult), __FILE__, __LINE__);    
    #endif
}

	
bool CThCond::Signal()
{
    #if defined(WIN32)
	::SetEvent(m_tCond);
    #else
	pthread_cond_signal(&m_tCond);        
    #endif
	return true;
}


bool CThCond::Broadcast()
{
    #if defined(WIN32)
	::PulseEvent(m_tCond); 
    #else
	pthread_cond_broadcast(&m_tCond);          
    #endif
	return true;
}


bool CThCond::Wait(CThMutex& stThMutex)
{
    #if defined(WIN32)
	TimedWait(stThMutex,INFINITE);
    #else
	int status;
	status=pthread_cond_wait(&m_tCond, &stThMutex.m_tMutex);     
    #endif
	return true;
}


bool CThCond::TimedWait(CThMutex& stThMutex, const unsigned int pstAbsTime )
{
	
    if( pstAbsTime > 0 )
	{
		#if defined(WIN32)
		stThMutex.Unlock();
		::WaitForSingleObject(m_tCond,pstAbsTime);
		stThMutex.Lock();
		::ResetEvent( m_tCond );
		#else
		timespec  then;
		size_t    nSeconds = pstAbsTime / 1000;
		size_t    nNanoSeconds = 1000000 * ( pstAbsTime % 1000 );
		
		then.tv_sec =  nSeconds;
		then.tv_nsec = nNanoSeconds;
		int iResult = pthread_cond_timedwait(&m_tCond, &stThMutex.m_tMutex,&then);
		if (iResult)
		{
			if (iResult == ETIMEDOUT || iResult == EINTR)
				return false;
			
			throw CThreadException(strerror(iResult), __FILE__, __LINE__);
		}	
		
		#endif
	}
	else
		throw CThreadException(("cond wait time is negtive"),__FILE__, __LINE__);
	return true;
}

CThSem::CThSem(int iValue)
{
	#if defined(WIN32)
	m_tSem = ::CreateSemaphore( NULL,//LPSECURITY_ATTRIBUTES 
		                        iValue,//initial count
								iValue,//MaxCount
								NULL   //name
		                       );
	if(m_tSem==NULL)
		throw CThreadException(STRERROR_WIN(WIN32_SYS), __FILE__, __LINE__);
	#else
	if (sem_init(&m_tSem, 0 /*not shared between processes*/, iValue) < 0)
		throw CThreadException(strerror(errno), __FILE__, __LINE__);	
	#endif
}


CThSem::~CThSem()
{
	#if defined(WIN32)
	if(! ::CloseHandle(m_tSem))
		throw CThreadException(STRERROR_WIN(WIN32_SYS), __FILE__, __LINE__);
	#else
	if (sem_destroy(&m_tSem) < 0)
		throw CThreadException(strerror(errno), __FILE__, __LINE__);	
	#endif

}

	
bool CThSem::Wait()
{
	#if defined(WIN32)
	DWORD dwWaitResult; 
	// Try to enter the semaphore gate.
	dwWaitResult = WaitForSingleObject( 
        m_tSem,			// handle to semaphore
        INFINITE);      // wait  INFINITE  
	//the semcount is nozero ,the sem is signaled, call return  
	return (dwWaitResult==WAIT_OBJECT_0)?true:false;
	#else
	sem_wait(&m_tSem);	
	#endif

	return true;
}


bool CThSem::TryWait()
{	
	#if defined(WIN32)
	DWORD dwWaitResult; 
	// Try to enter the semaphore gate.
	dwWaitResult = WaitForSingleObject( 
        m_tSem,			// handle to semaphore
        0L);      // zero-second time-out interval
	//dwWaitResult==WAIT_TIMEOUT when semcount is zero ,the sem is nonsignaled 
	return (dwWaitResult==WAIT_OBJECT_0)?true:false;
	#else
	if (sem_trywait(&m_tSem) < 0)
	{
		if (errno == EAGAIN)
			return false;
			
		throw CThreadException(strerror(errno), __FILE__, __LINE__);
	}
	#endif

	
	return true;
}


bool CThSem::Post()
{	
	#if defined(WIN32)
	BOOL ret=::ReleaseSemaphore(m_tSem, 1, NULL);
	return (ret==TRUE)?true:false;
	#else
	if (sem_post(&m_tSem) < 0)
		return false;
	#endif
	
	return true;
}
	
	
int CThSem::GetValue()
{    
	#if defined(WIN32)
	return -1;
	#else
	int iValue;
	sem_getvalue(&m_tSem, &iValue);
	return iValue;	
	#endif
}


#if defined(WIN32)
DWORD WINAPI CThread::ThreadFun(PVOID Param)
{
	CThread *pThread=(CThread *)Param;
	try
	{
	pThread->Run();
   	}
    	catch (CException& e)
	{
		pThread->OnException(e);
	}
	pThread->m_status = STOP;
	pThread->OnEnd();
	return NULL;	
}

#else
void* CThread::ThreadFun(void *param)
{
	
	CThread *pThread=(CThread *)param;
	//pthread_detach(pthread_self());
	try
	{
	pThread->m_status = START;
	pThread->Run();
            }
            catch (CException& e)
	{
		pThread->OnException(e);
	}
	pThread->m_status = STOP;
	pThread->OnEnd();
	return NULL;
}
#endif






CThread::CThread(bool bDetach)
{
	
	m_bDetach = bDetach;
	m_status = ALREADY;
	m_tThread=0;
    	#if defined(WIN32)
    	m_id=0;
	#else
	pthread_attr_init(&m_tThreadAttr);
	
	if (m_bDetach)
		pthread_attr_setdetachstate(&m_tThreadAttr, PTHREAD_CREATE_DETACHED);
	else
	{
		std::cerr<<"joinable\n";
		pthread_attr_setdetachstate(&m_tThreadAttr, PTHREAD_CREATE_JOINABLE);
    	}
    	#endif

}


CThread::~CThread()
{   
	 #if defined(WIN32)
	::CloseHandle(m_tThread);
	#else
	pthread_attr_destroy(&m_tThreadAttr);
    	#endif
	OnDelete();
}
	

void CThread::Start()
{
	if (m_status==START)
		return;
    	#if defined(WIN32)
	m_tThread=chBEGINTHREADEX(NULL, 0, ThreadFun,this,0,&m_id);
	#else       
	int iResult = pthread_create(&m_tThread, &m_tThreadAttr,ThreadFun, this);
	if (iResult)
	{
		throw CThreadException(strerror(iResult), __FILE__, __LINE__);
	}
    	#endif	
	m_status = START;
}


void CThread::OnDelete()
{
//	cerr<<"ondelete \n";
//	int iResult;
//	if (m_status==STOP && pthread_self() != m_tThread)	//end by another thread
//	{           cerr<<"join!"<<endl;
//		if (!m_bDetach)
//		{
//			iResult = pthread_join(m_tThread, NULL);
//			
//			if (iResult)
//				throw CThreadException(strerror(iResult), __FILE__, __LINE__);
//			return;
//		}		
//		iResult = pthread_cancel(m_tThread);
//		if (iResult)
//			throw CThreadException(strerror(iResult), __FILE__, __LINE__);
//	}

}

void CThread::SLEEP(unsigned int nSec)//毫秒
{    
	#if defined(WIN32)
		::Sleep(nSec);
	#else
	struct timeval	tval;
	tval.tv_sec = nSec / 1000;
	tval.tv_usec = nSec % 1000;
	select(0, NULL, NULL, NULL, &tval);

	//另一种实现
	//	for ( ; ; ) {
	//		tval.tv_sec = nSec / 1000;
	//		tval.tv_usec = nSec % 1000;
	//		if (select(0, NULL, NULL, NULL, &tval) == 0)
	//			break;		/* all OK */
	//		/*
	//		 * Note than on an interrupted system call (i.e, SIGIO) there's not
	//		 * much we can do, since the timeval{} isn't updated with the time
	//		 * remaining.  We could obtain the clock time before the call, and
	//		 * then obtain the clock time here, subtracting them to determine
	//		 * how long select() blocked before it was interrupted, but that
	//		 * seems like too much work :-)
	//		 */
	//		if (errno == EINTR)
	//			continue;
	//			
	//		//throw CThreadException(strerror(iResult), __FILE__, __LINE__);
	//		return;
	//	}	
	#endif
}

CThSingleBar::CThSingleBar(int nthreadnum)
            :m_lock(),m_cv(),m_ncount(nthreadnum),m_nposters(0),
			 m_nwaiters(0),m_breleasing(false)
{
}

int CThSingleBar::Wait()
{
  bool bRelease = false;
   m_nwaiters++;
   while (!m_breleasing)
     m_cv.Wait(m_lock);
   m_nwaiters--;
   if(m_nwaiters == 0) 
   {
     m_breleasing = false;
     bRelease = true;     /* Wake up waiters (if any) for next cycle */
     m_nposters=0;
   }
   m_lock.Unlock();
  if (bRelease)
    m_cv.Broadcast();
  return 0;
}


int CThSingleBar::Post()
{
 int release = false;
 {
	AutoMutex a(m_lock);
	while (m_breleasing)           /* If previous cycle still releasing, wait */
		m_cv.Wait(m_lock);
	m_nposters++;
	if (m_nposters == m_ncount)
	{
		m_breleasing = true;
		release = true;             /* Wake up waiters (if any) for next cycle */
	}
 }
 if (release)
   m_cv.Broadcast();
 return(0);
}



















}//end of mylib