usersync.cpp

mini http server,可以集成嵌入到程序中,实现简单的web功能

			};
		return 0;
		};
	
	virtual int TryLock()
		{
		if ( !iCurrentCount )
			{ 
			/* --- semaphore not obtained --- */
			return PISYNC_EBUSY;
			}
		else
			{
			iCurrentCount--;
			};
		return 0;
		};

	virtual int UnLock()
		{
#if DEBUG>=25
		CERR << "Unlocking, threads:- (iCurrentCount, tWaiting.Size()) "
			<< iCurrentCount << ", " << tWaiting.Size() << endl;
		DbgWriteAllThreads( CERR );
#endif
/*
**	NOTE:
**
**	This entire facility (to track which threads have which semaphores):-
**	is flawed. I believe it should be possible for one thread to release
**	a semaphore acquired by another thread. This is useful when blocking on 
**	sufficient threads in a pool, using a semaphore - but the semaphore
**	is acquired by the dispatcher thread and released by the handler
**	thread. I know Win32 allows this. HMMM
** 
**		/.* --- Remove one instance of this
**		semaphore from the list of semaphores the thread has acquired 
**		--- *./
**		for( DblListIterator i( ___pTheCurrentThread->lSemaphores ); 
**				!i.BadIndex(); i++)
**			{
**			if ( i.Current() == this )
**				{
**				___pTheCurrentThread->lSemaphores.Detach( i );
**				break;			/.* 1st instance only ! *./
**				};
**			};
**
** Note that all other code implementing this facility has been removed from
** this source file and this note is left for posterity only.
*/

		if ( tWaiting.Size() )
			{ InternalWakeUpThread(); }
		else
			{
			iCurrentCount++;
			assert( iCurrentCount<=iMaxCount );
			if ( iCurrentCount>iMaxCount ) 
				{ iCurrentCount--; PIERROR( PIAPI_ERROR ); return -1; };	
			};
		return 0;
		};

	virtual bool IsOK() const
		{
		return !bError;
		};

	/* --- friends --- */
	friend class UserThread;
};

/*___________________________________________________________________________*\
 *
 Class:
 Description:
\*___________________________________________________________________________*/
class PollInfo
{
private:
	timeval tvTimeout;
	Timer *pTimer;
	int iFlags;
	int iFD;

	void Sync()
		{
		long lSec, lUSec;
		PIMUSTBETRUE( pTimer->Elapsed( &lSec, &lUSec ) );	
		pTimer->Reset();
		tvTimeout.tv_sec -= lSec;
		tvTimeout.tv_usec -= lUSec;
		if ( tvTimeout.tv_usec < 0 )
			{
			tvTimeout.tv_sec--;
#if ( UINT_MAX > 0xffff )	/* subtilties */
			tvTimeout.tv_usec += 1000000;
#else
			tvTimeout.tv_usec += 1000000L;
#endif
			};
#if DEBUG>=30
		CERR << "PollInfo::Sync(): fd=" << iFD;
		if ( iFlags & Platform::POLL_READ ) { CERR << " READ"; };
		if ( iFlags & Platform::POLL_WRITE ) { CERR << " WRITE"; };
		CERR << ", timeout: ";
		if ( !pTimer )
			{ CERR << "never"; }
		else
			{
			CERR << "in " << tvTimeout.tv_sec << ".";
			CERR.width( 6 );
			CERR.fill( '0' );
			CERR << tvTimeout.tv_usec << " seconds";
			CERR.fill( ' ' );
			};
		CERR << endl;
#endif
		};

public:
	PollInfo( int iTheFD, int iTheFlags, int iTheTimeout )
	:	pTimer( iTheTimeout==-1 ? 0 : Platform::AllocTimer() ),
		iFlags( iTheFlags ),
		iFD( iTheFD )
		{
		tvTimeout.tv_sec = iTheTimeout / 1000;
		tvTimeout.tv_usec = iTheTimeout % 1000;
		};

	~PollInfo()
		{
		PI_DELETE( pTimer );
		};

	/* ---
	Determine whether or not this poll object has a timeout associated
	with it
	--- */
	inline bool HasTimeout()
		{
		return pTimer ? true : false;
		};

	/* ---
	If tv is greater than the current time remaining, lower tv to
	be equal to the current time remaining
	Return false if this poll object has timed out.
	If pTv is not null then calculate the minimum time between pTv and
	the time remaining for this object, set pTv to equal this miniumum.
	This method should only be invoked when there is a valid timeout
	(and hence pTimer points to a timeout object). A callee can use
	the method HasTimeout to determine whether or not a timeout exists.
	--- */
	inline bool MinTimeout( timeval *pTv )
		{
		assert( pTimer );
		if ( !pTimer )	{ return true; };
		Sync();
#if DEBUG>=20
		CERR << "Socket: " << iFD << ", time remaining "
			<< tvTimeout.tv_sec << ".";
		CERR.width( 6 );
		CERR.fill( '0' );
		CERR << tvTimeout.tv_usec;
		CERR.fill( ' ' );
		CERR << endl;
#endif
		if ( tvTimeout.tv_sec < 0 )
			{ return false; };
		if ( pTv )
			{
			if ( tvTimeout.tv_sec > pTv->tv_sec )
				{ ; }
			else if ( tvTimeout.tv_sec < pTv->tv_sec )
				{
				pTv->tv_sec = tvTimeout.tv_sec;
				pTv->tv_usec = tvTimeout.tv_usec;
				}
			else if ( tvTimeout.tv_usec < pTv->tv_usec )
				{
				pTv->tv_usec = tvTimeout.tv_usec;
				};
			};
		return true;
		};

	static int Poll( UserThread *pThread, int iFD, int iFlags, int iTimeout );

	/* --- friends --- */
	friend class Poller;
};

/*___________________________________________________________________________*\
 *
 Class:
 Description:
	A Poller polls IO channels on behalf of a number of threads. It
	has its own dedicated thread which does nothing but poll on IO
	channels and mark threads as runnable when a threads IO channel
	changes from a blocking state.
\*___________________________________________________________________________*/
class Poller
{
private:
	UserThread *pPollerThread;
	DblList tPollingThreads;
	static unsigned long PollLoop( unsigned long ulData )
		{
		for(;;)
			{
#if DEBUG>=20
			CERR << " *** Polling Thread *** " << endl;
			DbgWriteAllThreads( CERR );
#endif
			Poller *pPoller = (Poller *)ulData;
			if ( pPoller && (pPoller->tPollingThreads.Size()>0) )
				{
				if ( !pPoller->DoPoll() )
					{
					return (unsigned long)-1;
					};
				}
			else
				{
				assert( pPoller->pPollerThread );
				assert( pPoller->pPollerThread==___pTheCurrentThread );
				assert( !pPoller->pPollerThread->IsSuspended() );
				PIMUSTBETRUE( pPoller->pPollerThread->Suspend() );
				};
			StackContextSwitch();

			/* ---
			StackContextSwitch() can't stall or it will throw the
			program into an infinite loop. Deadlock is the only thing
			that could stall the program.
			--- */
			PIMUSTBETRUE( !pPoller->pPollerThread->iNextReturnCode );
			};
		};

	/* --- forbid copy constructor --- */
	Poller( const Poller & )
	:	pPollerThread( 0 )
		{ assert( 0 ); };

public:
	Poller()
	:	pPollerThread( (UserThread *)Platform::AllocThread( 0, 0 ) )
		{
		PIMUSTBETRUE( pPollerThread );
		PIMUSTBETRUE( pPollerThread->Begin( 
			PollLoop,
			(unsigned long)this,
			PITHREAD_PRIORITY_LOWEST,
			PITHREAD_FLAGS_SUSPENDED ) );
		};

	~Poller()
		{
		PI_DELETE( pPollerThread );
		for( DblListIterator i( tPollingThreads ); !i.BadIndex(); i++ )
			{
			((UserThread *)i.Current())->MakeRunnable( -1 );
			};
		};

	void AddThread( UserThread *pThread )
		{
		assert( pThread && pThread->pPoll && pPollerThread );
		tPollingThreads.Append( (DblList::type)pThread );
		if ( pPollerThread->IsSuspended() )
			{ pPollerThread->Resume(); };
		};

	void RemoveThread( UserThread *pThread )
		{
		for( DblListIterator i( tPollingThreads ); !i.BadIndex(); i++ )
			{
			if ( pThread == i.Current() )
				{
				tPollingThreads.Detach( i );
				return;
				};
			};
		assert( 0 );
		};

	/* ---
	Loop once through the list of polling threads and issue a 
	select() to get first available ready descriptor if necessary
	Returning false from this function is considered serious and
	may cause the poller thread, or program to terminate.
	--- */
	bool DoPoll()
		{
		assert( tPollingThreads.Size() > 0 );
		int iFDMax = -1;
		timeval tv;
		tv.tv_sec = INT_MAX;
		tv.tv_usec = INT_MAX;
		timeval *pTv = 0;		/* --- initially set to infinite timeout --- */
		fd_set tRead, tWrite, tExcept;
		FD_ZERO( &tRead ); FD_ZERO( &tWrite ); FD_ZERO( &tExcept );
#if DEBUG>=20
		CERR << "Entering DoPoll(), tPollingThreads.Size() = "
			<< tPollingThreads.Size() << endl;
#endif
		DblListIterator i( tPollingThreads );
		for(; !i.BadIndex(); i++)
			{
			PollInfo *pPoll = ((UserThread *)i.Current())->pPoll;
			assert( pPoll );
			int iFD = pPoll->iFD;
			if ( pPoll->HasTimeout() &&
				pTv==0 )
				{
				/* --- assign timeout --- */
				pTv = &tv;
				};
			/* ---
			Check to see if this PollInfo has timed out. Don't
			bother if another PollInfo has already timed out
			--- */
			if (
				pTv &&
				tv.tv_sec && tv.tv_usec && 
				pPoll->HasTimeout() && !pPoll->MinTimeout( pTv ) )
				{
				/* ---
				technically, this poll has timed out, but that
				may have happened with recent processing so see what
				the status is after the select(). Set the timeout
				values to 0, so the select() will not wait.
				--- */
				tv.tv_sec = 0;
				tv.tv_usec = 0;
				};
			if ( iFD>=0 )
				{
				/* ---
				This is a poll on a real descriptor and not just a 
				sleep 'til timeout
				--- */
				FD_SET( iFD, &tExcept );
				if ( pPoll->iFlags & Platform::POLL_READ )
					{ FD_SET( iFD, &tRead ); }
				if ( pPoll->iFlags & Platform::POLL_WRITE )
					{ FD_SET( iFD, &tWrite ); }
				if ( iFDMax<iFD )
					{ iFDMax = iFD; };
				};
			};
#if DEBUG>=20
		CERR << "DoPoll before select()";
		if ( pTv )
			{
			CERR << " timeout = " << pTv->tv_sec << ".";
			CERR.width( 6 );
			CERR.fill( '0' );
			CERR << pTv->tv_usec;
			CERR.fill( ' ' );
			};
		CERR << endl;
#endif
		int iRet = 0;
		if ( iFDMax>=0 )
			{
			for(;;)
				{
				iRet = ::select( iFDMax+1, &tRead, &tWrite, &tExcept, pTv ); 
			
				if ( iRet==-1 && errno==EINTR )
					{
					/* ---
					An interrupt could well have changed the runnable status
					of another thread. Give the scheduler a chance to see
					before restarting the select.
					--- */
					StackContextSwitch();
					continue;
					};

				break;
				};
			};
#if DEBUG>=20
		CERR << "DoPoll after select(), iRet = " << iRet
			<< ", errno = " << errno << endl;
#endif
		switch( iRet )
			{
			case -1:	/* error */
				PIOSERR;/* but this is not a user thread */
				return false;

			case 0:		/* timeout */
			default:	/* some sockets/files are ready */
				;
			};
		/* ---
		Note that it is necessary to step through the list of threads
		without using an iterator. This is because the call to MakeRunnable()
		ultimately mutates the list thus invalidating the iterator
		--- */
		for( int j=0; j<tPollingThreads.Size(); j++ )
			{
			UserThread *pThread = (UserThread *)tPollingThreads[j];
			PollInfo *pPoll = pThread->pPoll;
			assert( pPoll );
			int iFD = pPoll->iFD;
			int iFlags = pPoll->iFlags;
			bool bTimedOut = pPoll->HasTimeout() && !pPoll->MinTimeout( 0 );
#if DEBUG>=20
			CERR << "Socket: " << iFD;
#endif
			if ( bTimedOut )
				{
#if DEBUG>=20
				CERR << ", TIMEOUT";
#endif
				/* --- 0==return code for timeout --- */
				pThread->MakeRunnable( 0 );
				j--;
				}
			else if ( iRet>0 && iFD>=0 )
				{
				int iPollResult = 0;
				if ( FD_ISSET( iFD, &tExcept ) )
					{
#if DEBUG>=20
					CERR << ", EXCEPTION";
#endif
					iPollResult = -1;
					}
				else
					{
					if ( ( iFlags & Platform::POLL_READ ) && 
						FD_ISSET( iFD, &tRead ) )
						{
#if DEBUG>=20
						CERR << ", READ";
#endif
						iPollResult = iPollResult | Platform::POLL_READ;
						};
					if ( ( iFlags & Platform::POLL_WRITE ) && 
						FD_ISSET( iFD, &tWrite ) )
						{
#if DEBUG>=20
						CERR << ", WRITE";
#endif
						iPollResult = iPollResult | Platform::POLL_WRITE;
						};
					};
				if ( iPollResult )
					{
					pThread->MakeRunnable( iPollResult );
					j--;
					};
				};
#if DEBUG>=20
			CERR << endl;
#endif
			};	/* --- loop over all sockets up to iFDMax --- */
		return true;
		};
};

/*___________________________________________________________________________*\
 *
 Function:
 Synopsis:	  static, public:
 Description:
\*___________________________________________________________________________*/
int PollInfo::Poll( UserThread *pThread, int iFD, int iFlags,
	int iTimeout )
{
	assert( ___pThePoller );
	return pThread->PutToSleep( iFD, iFlags, iTimeout );
}

/*___________________________________________________________________________*\
 *
 Function:
 Synopsis:
 Description:
\*___________________________________________________________________________*/
void UserThread::Cleanup()
{
	assert( !bIsTerminated );

	/* --- cleanup on the death of this thread --- */
	bIsTerminated = true;

	/* --- notify any thread which is blocking on this thread --- */
	for(;;)