add

+/*
+ * Copyright 2007 Stephen Liu
+ * For license terms, see the file COPYING along with this library.
+ */
+
+
+#include <sys/types.h>
+#include <assert.h>
+
+#include "spporting.hpp"
+
+#include "spexecutor.hpp"
+#include "spthreadpool.hpp"
+
+#include "sputils.hpp"
+
+SP_Task :: ~SP_Task()
+{
+}
+
+//===================================================================
+
+SP_SimpleTask :: SP_SimpleTask( ThreadFunc_t func, void * arg, int deleteAfterRun )
+{
+	mFunc = func;
+	mArg = arg;
+
+	mDeleteAfterRun = deleteAfterRun;
+}
+
+SP_SimpleTask :: ~SP_SimpleTask()
+{
+}
+
+void SP_SimpleTask :: run()
+{
+	mFunc( mArg );
+
+	if( mDeleteAfterRun ) delete this;
+}
+
+//===================================================================
+
+SP_Executor :: SP_Executor( int maxThreads, const char * tag )
+{
+	tag = NULL == tag ? "unknown" : tag;
+
+	mThreadPool = new SP_ThreadPool( maxThreads, tag );
+
+	mQueue = new SP_BlockingQueue();
+
+	mIsShutdown = 0;
+
+	sp_thread_mutex_init( &mMutex, NULL );
+	sp_thread_cond_init( &mCond, NULL );
+
+	sp_thread_attr_t attr;
+	sp_thread_attr_init( &attr );
+	assert( sp_thread_attr_setstacksize( &attr, 1024 * 1024 ) == 0 );
+	sp_thread_attr_setdetachstate( &attr, SP_THREAD_CREATE_DETACHED );
+
+	sp_thread_t thread;
+	int ret = sp_thread_create( &thread, &attr, eventLoop, this );
+	sp_thread_attr_destroy( &attr );
+	if( 0 == ret ) {
+		sp_syslog( LOG_NOTICE, "[ex@%s] Thread #%ld has been created for executor", tag, thread );
+	} else {
+		sp_syslog( LOG_WARNING, "[ex@%s] Unable to create a thread for executor", tag );
+	}
+}
+
+SP_Executor :: ~SP_Executor()
+{
+	shutdown();
+
+	while( 2 != mIsShutdown ) {
+		sp_thread_mutex_lock( &mMutex );
+		sp_thread_cond_wait( &mCond, &mMutex );
+		sp_thread_mutex_unlock( &mMutex );
+	}
+
+	sp_thread_mutex_destroy( &mMutex );
+	sp_thread_cond_destroy( &mCond);
+
+	delete mThreadPool;
+	mThreadPool = NULL;
+
+	delete mQueue;
+	mQueue = NULL;
+}
+
+void SP_Executor :: shutdown()
+{
+	sp_thread_mutex_lock( &mMutex );
+	if( 0 == mIsShutdown ) {
+		mIsShutdown = 1;
+
+		// signal the event loop to wake up
+		execute( worker, NULL );
+	}
+	sp_thread_mutex_unlock( &mMutex );
+}
+
+sp_thread_result_t SP_THREAD_CALL  SP_Executor :: eventLoop( void * arg )
+{
+	SP_Executor * executor = ( SP_Executor * )arg;
+
+	while( 0 == executor->mIsShutdown ) {
+		void * queueData = executor->mQueue->pop();
+
+		if( executor->mThreadPool->getMaxThreads() > 1 ) {
+			if( 0 != executor->mThreadPool->dispatch( worker, queueData ) ) {
+				worker( queueData );
+			}
+		} else {
+			worker( queueData );
+		}
+	}
+
+	sp_thread_mutex_lock( &executor->mMutex );
+	executor->mIsShutdown = 2;
+	sp_thread_cond_signal( &executor->mCond );
+	sp_thread_mutex_unlock( &executor->mMutex );
+
+	return 0;
+}
+
+void SP_Executor :: worker( void * arg )
+{
+	if( NULL != arg ) {
+		SP_Task * task = ( SP_Task * )arg;
+		task->run();
+	}
+}
+
+void SP_Executor :: execute( SP_Task * task )
+{
+	mQueue->push( task );
+}
+
+void SP_Executor :: execute( void ( * func ) ( void * ), void * arg )
+{
+	SP_SimpleTask * task = new SP_SimpleTask( func, arg, 1 );
+	execute( task );
+}
+
+int SP_Executor :: getQueueLength()
+{
+	return mQueue->getLength();
+}
+