tlibbe.cxx

来自「基于VXWORKS H323通信技术源代码」· CXX 代码 · 共 717 行 · 第 1/2 页

  return retval;
}

int PThread::PXBlockOnIO(int maxHandles,
           fd_set & readBits,
           fd_set & writeBits,
           fd_set & exceptionBits,
           const PTimeInterval & timeout,
           const PIntArray & /*osHandles*/)
{
  // make sure we flush the buffer before doing a write
  fd_set * read_fds      = &readBits;
  fd_set * write_fds     = &writeBits;
  fd_set * exception_fds = &exceptionBits;

  struct timeval * tptr = NULL;
  struct timeval   timeout_val;
  if (timeout != PMaxTimeInterval) { // Clean up for infinite timeout
    static const PTimeInterval oneDay(0, 0, 0, 0, 1);
    if (timeout < oneDay) {
      timeout_val.tv_usec = (timeout.GetMilliSeconds() % 1000) * 1000;
      timeout_val.tv_sec  = timeout.GetSeconds();
      tptr                = &timeout_val;
    }
  }

  int retval = ::select(maxHandles, read_fds, write_fds, exception_fds, tptr);
  //YK PProcess::Current().PXCheckSignals();
  return retval;
}

///////////////////////////////////////////////////////////////////////////////
// PProcess::TimerThread

void PProcess::Construct()
{
  CreateConfigFilesDictionary();
}

PProcess::HouseKeepingThread::HouseKeepingThread()
  : PThread(256*1024, NoAutoDeleteThread, LowPriority)
{
  Resume();
}

void PProcess::HouseKeepingThread::Main()
{
	PProcess & process = PProcess::Current();
	
	while(1) {
		
		process.deleteThreadMutex.Wait();
		
		for (PINDEX i = 0; i < process.autoDeleteThreads.GetSize(); i++)
		{
			PThread * pThread = (PThread *) 
				process.autoDeleteThreads.GetAt(i);
			if( pThread->IsTerminated() )
			{
				process.autoDeleteThreads.RemoveAt(i--);
			}
		}
		
		process.deleteThreadMutex.Signal();
		
		PTimeInterval nextTimer = process.timers.Process();
    	if (nextTimer != PMaxTimeInterval)
		{
			if ( nextTimer.GetInterval() > 10000 )
			{
				nextTimer = 10000;
			}
		}
			
		breakBlock.Wait( nextTimer );
	}
}

void PProcess::SignalTimerChange()
{
  if (houseKeeper == NULL)
     houseKeeper = new HouseKeepingThread;
  else
    houseKeeper->breakBlock.Signal();
}


///////////////////////////////////////////////////////////////////////////////
// PProcess

PProcess::~PProcess()
{
  Sleep(100);  // Give threads time to die a natural death

  delete houseKeeper;

  // OK, if there are any left we get really insistent...
  activeThreadMutex.Wait();
  for (PINDEX i = 0; i < activeThreads.GetSize(); i++) {
    PThread & thread = activeThreads.GetDataAt(i);
    if (this != &thread && !thread.IsTerminated())
      thread.Terminate();  // With extreme prejudice
  }
  activeThreadMutex.Signal();

  deleteThreadMutex.Wait();
  autoDeleteThreads.RemoveAll();
  deleteThreadMutex.Signal();

  delete configFiles;
}

///////////////////////////////////////////////////////////////////////////////
// PSemaphore
//#define DEBUG_SEMAPHORES

PSemaphore::PSemaphore(sem_id anId)
{
  semId = anId;

  PAssertOS(semId != 0);
}

PSemaphore::PSemaphore(unsigned initial, unsigned /*maxCount*/)
	: semId(0)
{
  semId = ::create_sem(1, // the semaphore's thread count
  				"PWLS" // Name
  				);

  #ifdef DEBUG_SEMAPHORES
  PError << "::create_sem " << semId << endl;
  #endif 

   PAssertOS( semId >= B_NO_ERROR );
}

PSemaphore::~PSemaphore()
{
   PAssertOS( semId >= B_NO_ERROR );

	#ifdef DEBUG_SEMAPHORES
	int32 semCnt = 0;
	get_sem_count(semId, &semCnt);
    PError << "::delete_sem " << semId << ", count:" << semCnt << endl;
	#endif 

	status_t result = B_NO_ERROR;
	
	if ( semId != 0 )
	{
		result = ::delete_sem(semId);
	}

	#ifdef DEBUG_SEMAPHORES
	PError << ", result: " << strerror(result) << endl;
	#endif 
}

void PSemaphore::Wait()
{
   PAssertOS( semId >= B_NO_ERROR );

	#ifdef DEBUG_SEMAPHORES
	int32 semCnt = 0;
	get_sem_count(semId, &semCnt);
	PError << "::acquire_sem " << semId << ", count:" << semCnt;
	#endif 

	status_t result = B_NO_ERROR;
	
	while ((result = ::acquire_sem_etc(semId, 1, 0, 0)) == B_INTERRUPTED)
	{
		;
	}

	#ifdef DEBUG_SEMAPHORES
	PError << ", result: " << strerror(result) << endl;
	#endif 
}


BOOL PSemaphore::Wait(const PTimeInterval & timeout)
{
   PAssertOS( semId >= B_NO_ERROR );

	bigtime_t microseconds = 
		timeout == PMaxTimeInterval ? B_INFINITE_TIMEOUT : (timeout.GetMilliSeconds() * 1000 );
 
	status_t result = ::acquire_sem_etc(semId, 1, 
							B_TIMEOUT, microseconds);

	#ifdef DEBUG_SEMAPHORES
	sem_info info;
	::get_sem_info(semId, &info);
	PError << "::acquire_sem_etc " << info.sem << ", count:" << info.count << ", delay:";
	
	if( microseconds == B_INFINITE_TIMEOUT ) 
		PError << "infinite";
	else
		PError << microseconds;
	#endif 

	#ifdef DEBUG_SEMAPHORES
	PError << ", result: " << strerror(result) <<endl;
	#endif 
	
  return result == B_TIMED_OUT;
}


void PSemaphore::Signal()
{
    PAssertOS( semId >= B_NO_ERROR );

	#ifdef DEBUG_SEMAPHORES
	int32 semCnt = 0;
	get_sem_count(semId, &semCnt);
	PError << "::release_sem " << semId << ", count:" << semCnt;
	#endif 
	
	status_t result = B_NO_ERROR;
	result = ::release_sem_etc(semId, 1, 0);

	#ifdef DEBUG_SEMAPHORES
	PError << ", result: " << strerror(result) << endl;
	#endif 
}


BOOL PSemaphore::WillBlock() const
{
	PAssertOS( semId >= B_NO_ERROR );
	
	#ifdef DEBUG_SEMAPHORES
	int32 semCnt = 0;
	get_sem_count(semId, &semCnt);
	PError << "::acquire_sem_etc (WillBlock) " << semId << ", count:" << semCnt << endl;
	#endif 
	
	status_t result = ::acquire_sem_etc(semId, 0, B_TIMEOUT, 0);
	
	return result == B_WOULD_BLOCK;
}

///////////////////////////////////////////////////////////////////////////////
// PMutex  
// Using benaphores
#define USE_BENAPHORES // Comment this line if you don't want benaphores

PMutex::PMutex() 
  : PSemaphore( ::create_sem(1, "PWLM" ) ), benaphoreCount(0)
{
  #ifdef DEBUG_SEMAPHORES
  PError << "::create_sem(PMutex) " << semId << endl;
  #endif 
  
  PAssertOS( semId >= B_NO_ERROR );
}

PMutex::~PMutex()
{
}

void PMutex::Wait()
{
    PAssertOS( semId >= B_NO_ERROR );

    status_t result = B_NO_ERROR;
    
    #ifdef USE_BENAPHORES
    if( atomic_add( &benaphoreCount, 1 ) > 0 ) {
    #endif // USE_BENAPHORES

		while ((result = ::acquire_sem_etc(semId, 1, 0, 0)) == B_INTERRUPTED)
		{
			;
		}

    #ifdef USE_BENAPHORES
    	PAssertOS( result == B_NO_ERROR );
  		atomic_add(&benaphoreCount, -1);
    }
    #endif // USE_BENAPHORES
}

BOOL PMutex::Wait(const PTimeInterval & timeout)
{
    PAssertOS( semId >= B_NO_ERROR );

    status_t result = B_NO_ERROR;
    
	bigtime_t microseconds = 
		timeout == PMaxTimeInterval ? B_INFINITE_TIMEOUT : (timeout.GetMilliSeconds() * 1000 );

    #ifdef USE_BENAPHORES
    if( atomic_add( &benaphoreCount, 1 ) > 0 ) {
    #endif // USE_BENAPHORES

		result = ::acquire_sem_etc(semId, 1,
						B_TIMEOUT, microseconds);

    #ifdef USE_BENAPHORES
    	PAssertOS( result == B_NO_ERROR || result == B_TIMED_OUT || result == B_WOULD_BLOCK );
  		atomic_add(&benaphoreCount, -1);
    }
    #endif // USE_BENAPHORES

	return result == B_TIMED_OUT;
}

void PMutex::Signal()
{
    PAssertOS( semId >= B_NO_ERROR );

    status_t result = B_NO_ERROR;
    
    #ifdef USE_BENAPHORES
    if( atomic_add( &benaphoreCount, -1 ) > 1 ) 
    {
    #endif // USE_BENAPHORES

  		result = ::release_sem_etc(semId, 1, 0);

    #ifdef USE_BENAPHORES
    }
    #endif // USE_BENAPHORES
		
	PAssertOS( result == B_NO_ERROR );
}

BOOL PMutex::WillBlock() const 
{
    PAssertOS( semId >= B_NO_ERROR );

    status_t result = B_NO_ERROR;

	result = ::acquire_sem_etc(semId, 0, B_TIMEOUT, 0);
	
	return result == B_WOULD_BLOCK;
}

///////////////////////////////////////////////////////////////////////////////
// PSyncPoint

PSyncPoint::PSyncPoint()
  : PSemaphore( ::create_sem(0, "PWLP" ) )
{
  #ifdef DEBUG_SEMAPHORES
  PError << "::create_sem(PSyncPoint) " << semId << endl;
  #endif 
  
  PAssertOS( semId >= B_NO_ERROR );
}

// End Of File ///////////////////////////////////////////////////////////////