Index: /trunk/psLib/src/pslib_strict.h =================================================================== --- /trunk/psLib/src/pslib_strict.h (revision 18826) +++ /trunk/psLib/src/pslib_strict.h (revision 18827) @@ -9,6 +9,6 @@ * @author Eric Van Alst, MHPCC * -* @version $Revision: 1.36 $ $Name: not supported by cvs2svn $ -* @date $Date: 2008-06-16 21:58:42 $ +* @version $Revision: 1.37 $ $Name: not supported by cvs2svn $ +* @date $Date: 2008-07-31 23:56:29 $ * * Copyright 2004-2005 Maui High Performance Computing Center, University of Hawaii @@ -116,3 +116,5 @@ #include "psTree.h" +#include "psThread.h" + #endif // #ifndef PS_LIB_STRICT_H Index: /trunk/psLib/src/sys/Makefile.am =================================================================== --- /trunk/psLib/src/sys/Makefile.am (revision 18826) +++ /trunk/psLib/src/sys/Makefile.am (revision 18827) @@ -14,4 +14,5 @@ psSlurp.c \ psString.c \ + psThread.c \ psTrace.c \ psType.c \ @@ -42,4 +43,5 @@ psSlurp.h \ psString.h \ + psThread.h \ psTrace.h \ psType.h Index: /trunk/psLib/src/sys/psConfigure.c =================================================================== --- /trunk/psLib/src/sys/psConfigure.c (revision 18826) +++ /trunk/psLib/src/sys/psConfigure.c (revision 18827) @@ -13,6 +13,6 @@ * @author Robert DeSonia, MHPCC * - * @version $Revision: 1.26 $ $Name: not supported by cvs2svn $ - * @date $Date: 2007-11-13 21:50:13 $ + * @version $Revision: 1.27 $ $Name: not supported by cvs2svn $ + * @date $Date: 2008-07-31 23:56:29 $ * * Copyright 2004-2005 Maui High Performance Computing Center, University of Hawaii @@ -31,4 +31,5 @@ #include "psString.h" #include "psTime.h" +#include "psThread.h" #include "psEarthOrientation.h" #include "psError.h" @@ -124,4 +125,6 @@ psTimerStop(); + psThreadPoolFinalize(); + // Free the time tables if (!p_psTimeFinalize()) { Index: /trunk/psLib/src/sys/psThread.c =================================================================== --- /trunk/psLib/src/sys/psThread.c (revision 18827) +++ /trunk/psLib/src/sys/psThread.c (revision 18827) @@ -0,0 +1,289 @@ +#ifdef HAVE_CONFIG_H +# include "config.h" +#endif + +#include +#include +#include +#include + +#include "psAssert.h" +#include "psConstants.h" +#include "psAbort.h" +#include "psMemory.h" +#include "psList.h" +#include "psArray.h" +#include "psString.h" +#include "psThread.h" + +static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER; +static psList *pending = NULL; // queue of pending jobs +static psList *done = NULL; // queue of done jobs +static psArray *pool = NULL; // array of defined threads +static psArray *tasks = NULL; // queue of tasks + +/***** basic thread functions *****/ + +void psThreadLock () { + pthread_mutex_lock (&mutex); + return; +} + +void psThreadUnlock () { + pthread_mutex_unlock (&mutex); + return; +} + +void psThreadFree (psThread *thread) { + return; +} + +// allocate a psThread +psThread *psThreadAlloc () { + + psThread *thread = (psThread *)psAlloc(sizeof(psThread)); + psMemSetDeallocator(thread, (psFreeFunc)psThreadFree); + + thread->busy = false; + thread->fault = false; + return thread; +} + +/***** thread job functions *****/ + +void psThreadJobFree (psThreadJob *job) { + + if (!job) return; + + psFree (job->type); + psFree (job->args); + return; +} + +// allocate a psThreadJob of the given type +psThreadJob *psThreadJobAlloc (char *type) { + + psThreadJob *job = (psThreadJob *)psAlloc(sizeof(psThreadJob)); + psMemSetDeallocator(job, (psFreeFunc)psThreadJobFree); + + job->type = psStringCopy (type); + job->args = psArrayAlloc (0); + return job; +} + +// add a job to the queue of pending jobs +bool psThreadJobAddPending (psThreadJob *job) { + + // if we failed to call psThreadPoolInit, or we called it with nThreads == 0, + // find the matching function and just run it. + if (!pool || !pool->n) { + + // in non-threaded operation, the job is placed on the done list and immediately run + if (done == NULL) { + done = psListAlloc(NULL); + } + psListAdd (done, PS_LIST_TAIL, job); + + // find the corresponding task and run it + for (int i = 0; i < tasks->n; i++) { + psThreadTask *task = tasks->data[i]; + if (strcmp (job->type, task->type)) continue; + + psAssert (job->args->n == task->nArgs, "invalid number of arguments to %s", task->type); + + bool status = task->function(job); + return status; + } + return false; + } + + psThreadLock (); + if (pending == NULL) { + pending = psListAlloc(NULL); + } + + psListAdd (pending, PS_LIST_TAIL, job); + psThreadUnlock (); + return true; +} + +// this function is not locked -- see thread launder for example +psThreadJob *psThreadJobGetPending () { + + if (!pending) return NULL; + + psThreadJob *job = psListGetAndRemove (pending, PS_LIST_HEAD); + + // jobs we pull off the pending queue get placed on the done queue + if (job) { + if (done == NULL) { + done = psListAlloc(NULL); + } + psListAdd (done, PS_LIST_TAIL, job); + } + return job; +} + +// this function is not locked -- see thread launder for example +psThreadJob *psThreadJobGetDone () { + + if (!done) return NULL; + + psThreadJob *job = psListGetAndRemove (done, PS_LIST_HEAD); + return job; +} + +/***** thread task functions *****/ + +void psThreadTaskFree (psThreadTask *task) { + + if (!task) return; + + psFree (task->type); + return; +} + +// allocate a psThreadTask with nArgs arguments +psThreadTask *psThreadTaskAlloc (char *type, int nArgs) { + + psThreadTask *task = (psThreadTask *)psAlloc(sizeof(psThreadTask)); + psMemSetDeallocator(task, (psFreeFunc)psThreadTaskFree); + + task->type = psStringCopy (type); + task->nArgs = nArgs; + task->function = NULL; + return task; +} + +// add a task to the collection of tasks +bool psThreadTaskAdd (psThreadTask *task) { + + if (tasks == NULL) { + tasks = psArrayAllocEmpty(8); + } + + psArrayAdd (tasks, 1, task); + return true; +} + +// each thread runs this function to choose the task functions +void *psThreadLauncher (void *data) { + + psThread *self = data; + psThreadJob *job = NULL; + + while (1) { + + // if we get an error, just wait until we are cleared or killed + while (self->fault) { + usleep (10000); + } + + // if no tasks are assigned, just wait until they are + while (tasks == NULL) { + usleep (10000); + } + + // request a new job, if there are none available, sleep a bit + // we have to lock here so the job queue cannot be empty yet no threads busy + psThreadLock(); + while ((job = psThreadJobGetPending ()) == NULL) { + psThreadUnlock(); + usleep (10000); + psThreadLock(); // XXX ??? + } + self->busy = true; + + for (int i = 0; i < tasks->n; i++) { + psThreadTask *task = tasks->data[i]; + if (strcmp (job->type, task->type)) continue; + + psThreadUnlock(); + psAssert (job->args->n == task->nArgs, "invalid number of arguments to %s", task->type); + + bool status = task->function(job); + if (!status) { + self->fault = true; + } + self->busy = false; + break; + } + psThreadUnlock(); + // XXX what do we do if the job is unknown? + } +} + +/***** thread pool functions *****/ + +// create a pool of Nthreads, each running the user's job-launcher function +bool psThreadPoolInit (int nThreads) { + + if (pool) psAbort ("psThreadsInit already called"); + + pool = psArrayAlloc (nThreads); + for (int i = 0; i < nThreads; i++) { + psThread *thread = psThreadAlloc(); + pthread_create (&thread->pt, NULL, psThreadLauncher, thread); + pool->data[i] = thread; + } + return true; +} + +// call this function after you have added jobs to the queue and +bool psThreadPoolWait () { + + // this function blocks (waits in usleep) until either + // an error is detected on one of the threads or until + // all threads are idle and no jobs are left on the queue + + if (!pool) return true; + if (!pool->n) return true; + + while (1) { + + // check for an error + for (int i = 0; i < pool->n; i++) { + psThread *thread = pool->data[i]; + if (thread->fault) return false; + } + + // are all threads idle? + psThreadLock(); + for (int i = 0; i < pool->n; i++) { + psThread *thread = pool->data[i]; + if (thread->busy) goto SLEEP; + } + + if (!pending) return true; + + // is the queue empty? + if (pending->head == NULL) { + psThreadUnlock(); + return true; + } + + SLEEP: + psThreadUnlock(); + usleep (10000); + } + return false; +} + +// create a pool of Nthreads, each running the user's job-launcher function +bool psThreadPoolFinalize () { + + psFree (pending); + pending = NULL; + + psFree (done); + done = NULL; + + psFree (pool); + pool = NULL; + + psFree (tasks); + tasks = NULL; + + return true; +} + Index: /trunk/psLib/src/sys/psThread.h =================================================================== --- /trunk/psLib/src/sys/psThread.h (revision 18827) +++ /trunk/psLib/src/sys/psThread.h (revision 18827) @@ -0,0 +1,58 @@ +/** @file psThread.h + * + * @brief tools to manage a pool of threads + * + * @author EAM, IFA + * @version $Revision: 1.2 $ $Name: not supported by cvs2svn $ + * @date $Date: 2008-07-31 23:56:29 $ + * + * Copyright 2004-2005 Insitute for Astronomy, University of Hawaii + */ + +#ifndef PS_THREAD_H +#define PS_THREAD_H + +/// @addtogroup SysUtils System Utilities +/// @{ + +typedef struct { + psString type; + psArray *args; +} psThreadJob; + +typedef struct { + bool busy; + bool fault; + pthread_t pt; +} psThread; + +typedef bool (*psThreadTaskFunction)(psThreadJob *job); + +typedef struct { + psString type; + int nArgs; + psThreadTaskFunction function; +} psThreadTask; + +// typedef void *(*psThreadLaunchJobsFunction)(void *data); + +void psThreadLock (); +void psThreadUnlock (); + +psThread *psThreadAlloc (); + +psThreadJob *psThreadJobAlloc (char *type); +bool psThreadJobAddPending (psThreadJob *job); +psThreadJob *psThreadJobGetPending (); +psThreadJob *psThreadJobGetDone (); + +psThreadTask *psThreadTaskAlloc (char *type, int nArgs); +bool psThreadTaskAdd (psThreadTask *task); +void *psThreadLauncher (void *data); + +bool psThreadPoolInit (int nThreads); +bool psThreadPoolWait (); +bool psThreadPoolFinalize (); + +/// @} +#endif /* PS_THREAD_H */