WelsThreadLib.cpp

/*!
 * \copy
 *     Copyright (c)  2009-2013, Cisco Systems
 *     All rights reserved.
 *
 *     Redistribution and use in source and binary forms, with or without
 *     modification, are permitted provided that the following conditions
 *     are met:
 *
 *        * Redistributions of source code must retain the above copyright
 *          notice, this list of conditions and the following disclaimer.
 *
 *        * Redistributions in binary form must reproduce the above copyright
 *          notice, this list of conditions and the following disclaimer in
 *          the documentation and/or other materials provided with the
 *          distribution.
 *
 *     THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
 *     "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
 *     LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
 *     FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
 *     COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
 *     INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
 *     BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
 *     LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
 *     CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
 *     LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
 *     ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 *     POSSIBILITY OF SUCH DAMAGE.
 *
 *
 * \file	WelsThreadLib.c
 *
 * \brief	Interfaces introduced in thread programming
 *
 * \date	11/17/2009 Created
 *
 *************************************************************************************
 */


#include "WelsThreadLib.h"
#include <stdio.h>

#ifdef  WIN32

void WelsSleep( uint32_t dwMilliseconds )
{
	Sleep( dwMilliseconds );
}

WELS_THREAD_ERROR_CODE    WelsMutexInit( WELS_MUTEX   * mutex )
{
	InitializeCriticalSection(mutex);

	return WELS_THREAD_ERROR_OK;
}

WELS_THREAD_ERROR_CODE    WelsMutexLock( WELS_MUTEX   * mutex )
{
	EnterCriticalSection(mutex);

	return WELS_THREAD_ERROR_OK;
}

WELS_THREAD_ERROR_CODE    WelsMutexUnlock( WELS_MUTEX * mutex )
{
	LeaveCriticalSection(mutex);

	return WELS_THREAD_ERROR_OK;
}

WELS_THREAD_ERROR_CODE    WelsMutexDestroy( WELS_MUTEX * mutex )
{
    DeleteCriticalSection(mutex);

	return WELS_THREAD_ERROR_OK;
}

WELS_THREAD_ERROR_CODE    WelsEventInit( WELS_EVENT  *  event )
{
    WELS_EVENT   h = CreateEvent(NULL, FALSE, FALSE, NULL);

	if( h == NULL ){
		return WELS_THREAD_ERROR_GENERIAL;
	}
	*event = h;
	return WELS_THREAD_ERROR_OK;
}

WELS_THREAD_ERROR_CODE    WelsEventSignal( WELS_EVENT * event )
{
	if( SetEvent( *event ) ){
		return WELS_THREAD_ERROR_OK;
	}
	return WELS_THREAD_ERROR_GENERIAL;
}

WELS_THREAD_ERROR_CODE    WelsEventReset( WELS_EVENT * event )
{
	if ( ResetEvent( *event ) )
		return WELS_THREAD_ERROR_OK;
	return WELS_THREAD_ERROR_GENERIAL;
}

WELS_THREAD_ERROR_CODE    WelsEventWait( WELS_EVENT * event )
{
	return WaitForSingleObject(*event, INFINITE );
}

WELS_THREAD_ERROR_CODE    WelsEventWaitWithTimeOut( WELS_EVENT * event, uint32_t dwMilliseconds )
{
	return WaitForSingleObject(*event, dwMilliseconds );
}

WELS_THREAD_ERROR_CODE    WelsMultipleEventsWaitSingleBlocking(	uint32_t nCount,
																WELS_EVENT *event_list,
																uint32_t dwMilliseconds )
{
	return WaitForMultipleObjects( nCount, event_list, FALSE, dwMilliseconds );
}

WELS_THREAD_ERROR_CODE    WelsMultipleEventsWaitAllBlocking( uint32_t nCount, WELS_EVENT *event_list )
{
	return WaitForMultipleObjects( nCount, event_list, TRUE, (uint32_t)-1 );
}

WELS_THREAD_ERROR_CODE    WelsEventDestroy( WELS_EVENT * event )
{
	CloseHandle( *event );

	*event = NULL;
	return WELS_THREAD_ERROR_OK;
}


WELS_THREAD_ERROR_CODE    WelsThreadCreate( WELS_THREAD_HANDLE * thread,  LPWELS_THREAD_ROUTINE  routine, 
										   void * arg, WELS_THREAD_ATTR attr)
{
    WELS_THREAD_HANDLE   h = CreateThread(NULL, 0, routine, arg, 0, NULL);

	if( h == NULL ) {
		return WELS_THREAD_ERROR_GENERIAL;
	}
	* thread = h;

	return WELS_THREAD_ERROR_OK;
}

WELS_THREAD_ERROR_CODE	  WelsSetThreadCancelable()
{
	// nil implementation for WIN32
	return WELS_THREAD_ERROR_OK;
}

WELS_THREAD_ERROR_CODE    WelsThreadJoin( WELS_THREAD_HANDLE  thread )
{
    WaitForSingleObject(thread, INFINITE);

	return WELS_THREAD_ERROR_OK;
}

WELS_THREAD_ERROR_CODE    WelsThreadCancel( WELS_THREAD_HANDLE  thread )
{
	return WELS_THREAD_ERROR_OK;
}


WELS_THREAD_ERROR_CODE    WelsThreadDestroy( WELS_THREAD_HANDLE *thread )
{
	if ( thread != NULL )
	{
		CloseHandle(*thread);
		*thread = NULL;
	}	
	return WELS_THREAD_ERROR_OK;
}

WELS_THREAD_HANDLE        WelsThreadSelf()
{
	return GetCurrentThread();
}

WELS_THREAD_ERROR_CODE    WelsQueryLogicalProcessInfo(WelsLogicalProcessInfo * pInfo)
{
	SYSTEM_INFO  si;	
	
	GetSystemInfo(&si);

	pInfo->ProcessorCount = si.dwNumberOfProcessors;

	return WELS_THREAD_ERROR_OK;
}

#elif   defined(__GNUC__)

#ifdef MACOS
#include <CoreServices/CoreServices.h>
//#include <Gestalt.h>
#endif//MACOS

static int32_t  SystemCall(const str_t * pCmd, str_t * pRes, int32_t iSize)
{
    int32_t fd[2];
    int32_t iPid;
    int32_t iCount;
    int32_t left;
    str_t * p = NULL;
    int32_t iMaxLen = iSize - 1;
    memset(pRes, 0, iSize);

    if( pipe(fd) ){
        return -1;
    }

    if( (iPid = fork()) == 0 ){
        int32_t  fd2[2];
        if( pipe(fd2) ){
            return -1;
        }
        close(STDOUT_FILENO);
        dup2(fd2[1],STDOUT_FILENO);
        close(fd[0]);
        close(fd2[1]);
        system(pCmd);
        read(fd2[0], pRes, iMaxLen);
        write(fd[1], pRes, strlen(pRes));	// confirmed_safe_unsafe_usage
        close(fd2[0]);
		close(fd[1]);
        exit(0);
    }
    close(fd[1]);
    p = pRes;
    left = iMaxLen;
    while( (iCount = read(fd[0], p, left)) ){
        p += iCount;
        left -= iCount;
        if( left <=0 ) break;   
    }
    close(fd[0]);
    return 0;
}

void WelsSleep( uint32_t dwMilliseconds )
{
	usleep( dwMilliseconds * 1000 );	// microseconds
}

WELS_THREAD_ERROR_CODE    WelsThreadCreate( WELS_THREAD_HANDLE * thread,  LPWELS_THREAD_ROUTINE  routine, 
										   void * arg, WELS_THREAD_ATTR attr)
{
	WELS_THREAD_ERROR_CODE err = 0;

	pthread_attr_t at;
	err = pthread_attr_init(&at);
	if ( err )
		return err;
	err = pthread_attr_setscope(&at, PTHREAD_SCOPE_SYSTEM);
	if ( err )
		return err;
	err = pthread_attr_setschedpolicy(&at, SCHED_FIFO);
	if ( err )
		return err;
	err = pthread_create( thread, &at, routine, arg );

	pthread_attr_destroy(&at);

	return err;

//	return pthread_create(thread, NULL, routine, arg); 
}

WELS_THREAD_ERROR_CODE	  WelsSetThreadCancelable()
{
	WELS_THREAD_ERROR_CODE err = pthread_setcancelstate( PTHREAD_CANCEL_ENABLE, NULL );
	if ( 0 == err )
		err = pthread_setcanceltype( PTHREAD_CANCEL_DEFERRED, NULL );
	return err;
}

WELS_THREAD_ERROR_CODE    WelsThreadJoin( WELS_THREAD_HANDLE  thread )
{
    return pthread_join(thread, NULL);
}

WELS_THREAD_ERROR_CODE    WelsThreadCancel( WELS_THREAD_HANDLE  thread )
{
	return pthread_cancel( thread );
}

WELS_THREAD_ERROR_CODE    WelsThreadDestroy( WELS_THREAD_HANDLE *thread )
{	
	return WELS_THREAD_ERROR_OK;
}

WELS_THREAD_HANDLE        WelsThreadSelf()
{
	return pthread_self();
}

WELS_THREAD_ERROR_CODE    WelsMutexInit( WELS_MUTEX   * mutex )
{
	return pthread_mutex_init(mutex, NULL);
}

WELS_THREAD_ERROR_CODE    WelsMutexLock( WELS_MUTEX   * mutex )
{
	return pthread_mutex_lock(mutex);
}

WELS_THREAD_ERROR_CODE    WelsMutexUnlock( WELS_MUTEX * mutex )
{
	return pthread_mutex_unlock(mutex);
}

WELS_THREAD_ERROR_CODE    WelsMutexDestroy( WELS_MUTEX * mutex )
{
    return pthread_mutex_destroy(mutex);
}

// unnamed semaphores can not work well for posix threading models under not root users

WELS_THREAD_ERROR_CODE    WelsEventInit( WELS_EVENT *event )
{
	return sem_init(event, 0, 0);
}

WELS_THREAD_ERROR_CODE   WelsEventDestroy( WELS_EVENT * event )
{
	return sem_destroy( event );	// match with sem_init	
}

WELS_THREAD_ERROR_CODE    WelsEventOpen( WELS_EVENT **p_event, str_t *event_name )
{
	if ( p_event == NULL || event_name == NULL )
		return WELS_THREAD_ERROR_GENERIAL;
	*p_event = sem_open(event_name, O_CREAT,  (S_IRUSR | S_IWUSR)/*0600*/, 0);
	if ( *p_event == (sem_t *)SEM_FAILED ) {
		sem_unlink( event_name );
		*p_event = NULL;
		return WELS_THREAD_ERROR_GENERIAL;
	} else {		
		return WELS_THREAD_ERROR_OK;
	}
}
WELS_THREAD_ERROR_CODE    WelsEventClose( WELS_EVENT *event, str_t *event_name )
{
	WELS_THREAD_ERROR_CODE err = sem_close( event );	// match with sem_open
	if ( event_name )
		sem_unlink( event_name );
	return err;
}

WELS_THREAD_ERROR_CODE   WelsEventSignal( WELS_EVENT * event )
{
	WELS_THREAD_ERROR_CODE err = 0;
//	int32_t val = 0;
//	sem_getvalue(event, &val);
//	fprintf( stderr, "before signal it, val= %d..\n",val );
	err = sem_post(event);
//	sem_getvalue(event, &val);
//	fprintf( stderr, "after signal it, val= %d..\n",val );
    return err;
}
WELS_THREAD_ERROR_CODE    WelsEventReset( WELS_EVENT * event )
{
	// FIXME for posix event reset, seems not be supported for pthread??
	sem_close(event);
	return sem_init(event, 0, 0);
}

WELS_THREAD_ERROR_CODE   WelsEventWait( WELS_EVENT * event )
{
	return sem_wait(event);	// blocking until signaled
}

WELS_THREAD_ERROR_CODE    WelsEventWaitWithTimeOut( WELS_EVENT * event, uint32_t dwMilliseconds )
{	
	if ( dwMilliseconds != (uint32_t)-1 )
	{
		return sem_wait(event);
	}
	else
	{
#if defined(MACOS)
		int32_t err = 0;
		int32_t wait_count = 0;
		do{
			err = sem_trywait(event);
			if ( WELS_THREAD_ERROR_OK == err)
				break;// WELS_THREAD_ERROR_OK;
			else if ( wait_count > 0 )
				break;
			usleep( dwMilliseconds * 1000 );
			++ wait_count;
		}while(1);
		return err;
#else
		struct timespec ts;
		struct timeval tv;

		gettimeofday(&tv,0);

		ts.tv_sec = tv.tv_sec + dwMilliseconds /1000;
		ts.tv_nsec = tv.tv_usec*1000 + (dwMilliseconds % 1000) * 1000000;

		return sem_timedwait(event, &ts);
#endif//MACOS
	}
}

WELS_THREAD_ERROR_CODE    WelsMultipleEventsWaitSingleBlocking(	uint32_t nCount,
																WELS_EVENT **event_list,
																uint32_t dwMilliseconds )
{
	// bWaitAll = FALSE && blocking
	uint32_t nIdx = 0;
	const uint32_t kuiAccessTime = 2;	// 2 us once
//	uint32_t uiSleepMs = 0;

	if ( nCount == 0 )
		return WELS_THREAD_ERROR_WAIT_FAILED;

	while (1)
	{
		nIdx = 0;	// access each event by order
		while ( nIdx < nCount )
		{
			int32_t err = 0;			
//#if defined(MACOS)	// clock_gettime(CLOCK_REALTIME) & sem_timedwait not supported on mac, so have below impl
			int32_t wait_count = 0;
//			struct timespec ts;
//			struct timeval tv;
//			
//			gettimeofday(&tv,0);
//			ts.tv_sec = tv.tv_sec/*+ kuiAccessTime / 1000*/;		// second
//			ts.tv_nsec = (tv.tv_usec + kuiAccessTime) * 1000;	// nano-second
			
			/*
			 * although such interface is not used in __GNUC__ like platform, to use 
			 * pthread_cond_timedwait() might be better choice if need
			 */
			do{
				err = sem_trywait( event_list[nIdx] );
				if ( WELS_THREAD_ERROR_OK == err )
					return WELS_THREAD_ERROR_WAIT_OBJECT_0 + nIdx;
				else if ( wait_count > 0 )
					break;
				usleep(kuiAccessTime);
				++ wait_count;
			}while( 1 );
//#else
//			struct timespec ts;
//			
//			if ( clock_gettime(CLOCK_REALTIME, &ts) == -1 )
//				return WELS_THREAD_ERROR_WAIT_FAILED;
//			ts.tv_nsec += kuiAccessTime/*(kuiAccessTime % 1000)*/ * 1000;
//			
////			fprintf( stderr, "sem_timedwait(): start to wait event %d..\n", nIdx );
//			err = sem_timedwait(event_list[nIdx], &ts);
////			if ( err == -1 )
////			{
////				sem_getvalue(&event_list[nIdx], &val);
////				fprintf( stderr, "sem_timedwait() errno(%d) semaphore %d..\n", errno, val);
////				return WELS_THREAD_ERROR_WAIT_FAILED;
////			}			
////			fprintf( stderr, "sem_timedwait(): wait event %d result %d errno %d..\n", nIdx, err, errno );
//			if ( WELS_THREAD_ERROR_OK == err ) // non-blocking mode
//			{	
////				int32_t val = 0;
////				sem_getvalue(&event_list[nIdx], &val);
////				fprintf( stderr, "after sem_timedwait(), event_list[%d] semaphore value= %d..\n", nIdx, val);
////				fprintf( stderr, "WelsMultipleEventsWaitSingleBlocking sleep %d us\n", uiSleepMs);
//				return WELS_THREAD_ERROR_WAIT_OBJECT_0 + nIdx;
//			}
//#endif					
			// we do need access next event next time
			++ nIdx;
//			uiSleepMs += kuiAccessTime;
		}
		usleep( 1 );	// switch to working threads
//		++ uiSleepMs;
	}	

	return WELS_THREAD_ERROR_WAIT_FAILED;
}

WELS_THREAD_ERROR_CODE    WelsMultipleEventsWaitAllBlocking( uint32_t nCount, WELS_EVENT **event_list )
{
	// bWaitAll = TRUE && blocking
	uint32_t nIdx = 0;
//	const uint32_t kuiAccessTime = (uint32_t)-1;// 1 ms once
	uint32_t uiCountSignals = 0;
	uint32_t uiSignalFlag	= 0;	// UGLY: suppose maximal event number up to 32
	
	if ( nCount == 0 || nCount > (sizeof(uint32_t)<<3) )
		return WELS_THREAD_ERROR_WAIT_FAILED;
	
	while (1)
	{
		nIdx = 0;	// access each event by order
		while (nIdx < nCount)
		{			
			const uint32_t kuiBitwiseFlag = (1<<nIdx);
			
			if ( (uiSignalFlag & kuiBitwiseFlag) != kuiBitwiseFlag ) // non-blocking mode
			{	
				int32_t err = 0;
//				fprintf( stderr, "sem_wait(): start to wait event %d..\n", nIdx );
				err = sem_wait(event_list[nIdx]);
//				fprintf( stderr, "sem_wait(): wait event %d result %d errno %d..\n", nIdx, err, errno );
				if ( WELS_THREAD_ERROR_OK == err )
				{
//					int32_t val = 0;
//					sem_getvalue(&event_list[nIdx], &val);
//					fprintf( stderr, "after sem_timedwait(), event_list[%d] semaphore value= %d..\n", nIdx, val);

					uiSignalFlag |= kuiBitwiseFlag;
					++ uiCountSignals;
					if ( uiCountSignals >= nCount )
					{						
						return WELS_THREAD_ERROR_OK;
					}
				}				
			}			
			// we do need access next event next time
			++ nIdx;
		}		
	}	
	
	return WELS_THREAD_ERROR_WAIT_FAILED;
}

WELS_THREAD_ERROR_CODE    WelsQueryLogicalProcessInfo(WelsLogicalProcessInfo * pInfo)
{
#ifdef LINUX

#define   CMD_RES_SIZE    2048
    str_t pBuf[CMD_RES_SIZE];
   
    SystemCall("cat /proc/cpuinfo | grep \"processor\" | wc -l", pBuf, CMD_RES_SIZE);

    pInfo->ProcessorCount = atoi(pBuf);

    if( pInfo->ProcessorCount == 0 ){
        pInfo->ProcessorCount = 1;
    }   
 
	return WELS_THREAD_ERROR_OK;
#undef   CMD_RES_SIZE

#else

	SInt32 cpunumber;
	Gestalt(gestaltCountOfCPUs,&cpunumber);

	pInfo->ProcessorCount	= cpunumber;

	return WELS_THREAD_ERROR_OK;

#endif//LINUX
}

#endif