Changeset 18822 for branches/eam_branch_20080719/psLib/src/sys/psThread.c

Timestamp:

Jul 31, 2008, 1:24:22 PM (18 years ago)

Author:

eugene

Message:

some threading simplifications: threaded and unthreaded use the same code; added concept of a thread task

File:

• branches/eam_branch_20080719/psLib/src/sys/psThread.c (modified) (7 diffs)

branches/eam_branch_20080719/psLib/src/sys/psThread.c

@@ -6 +6 @@
 #include <stdarg.h>
 #include <unistd.h>
+#include <string.h>
 
 #include "psAssert.h"
@@ -19 +20 @@
 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 () {
@@ -30 +33 @@
     pthread_mutex_unlock (&mutex);
     return;
-}
-
-void psThreadJobFree (psThreadJob *job) {
-
-    if (!job) return;
-
-    psFree (job->type);
-    psFree (job->args);
-    return;
-}
-
-// allocate a psThreadJob with nArgs arguments
-psThreadJob *psThreadJobAlloc (char *type, int nArgs) {
-    
-    psThreadJob *job = (psThreadJob *)psAlloc(sizeof(psThreadJob));
-    psMemSetDeallocator(job, (psFreeFunc)psThreadJobFree);
-
-    job->type = psStringCopy (type);
-    job->args = psArrayAllocEmpty (nArgs);
-    return job;
-}
-
-// add a job to the queue of pending jobs
-bool psThreadJobAddPending (psThreadJob *job) {
-
-    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;
 }
 
@@ -106 +50 @@
 }
 
+/***** 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, psThreadLaunchJobsFunction function) {
+bool psThreadPoolInit (int nThreads) {
 
     if (pool) psAbort ("psThreadsInit already called"); 
@@ -114 +224 @@
     for (int i = 0; i < nThreads; i++) {
         psThread *thread = psThreadAlloc();
-        pthread_create (&thread->pt, NULL, function, thread);
+        pthread_create (&thread->pt, NULL, psThreadLauncher, thread);
         pool->data[i] = thread;
     }
@@ -126 +236 @@
     // 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) {
@@ -169 +282 @@
     pool = NULL;
 
-    return true;
-}
-
+    psFree (tasks);
+    tasks = NULL;
+
+    return true;
+}
+