Changeset 33089 for trunk/psLib

Timestamp:

Jan 11, 2012, 10:16:48 AM (15 years ago)

Author:

bills

Message:

Fix problems with IPP threading. In psThreadPoolWait wait until all threads have
finished before returning even if one of the threads has a fault.
Return false if any thread faults. Also add a new argument bool harvestOnFailure
which tells psThreadPoolWait to harvest the done jobs if there is a failure.
Many functions previously used psThreadPoolWait(harvest = false) because they
wanted to examine the jobs structs afterwards. However they weren't cleaning up the
jobs in the failure case.
These now call psThreadPoolWait(harvest = false, harvestOnFailure = true)
which causes the jobs to be cleaned up on fault. These lingering jobs were the
cause of the "Unknown task" failure in psImageConvolve*
Also changed psphotReadout to check for failure of psphotGuessModels and return
to the caller. This yields quality = 3007

Location:

trunk/psLib/src

Files:

• imageops/psImageConvolve.c (modified) (5 diffs)
• imageops/psImageCovariance.c (modified) (2 diffs)
• sys/psThread.c (modified) (8 diffs)
• sys/psThread.h (modified) (1 diff)

trunk/psLib/src/imageops/psImageConvolve.c

@@ -1039 +1039 @@
             }
         }
-        if (!psThreadPoolWait(true)) {
+        if (!psThreadPoolWait(true, true)) {
             psError(PS_ERR_UNKNOWN, false, "Error waiting for threads.");
             psFree(gaussNorm);
@@ -1390 +1390 @@
               }
               // wait here for the threaded jobs to finish (NOP if threading is not active)
-              if (!psThreadPoolWait(true)) {
+              if (!psThreadPoolWait(true, true)) {
                   psError(PS_ERR_UNKNOWN, false, "Unable to smooth image");
                   psFree(calculation);
@@ -1437 +1437 @@
 
               // wait here for the threaded jobs to finish (NOP if threading is not active)
-              if (!psThreadPoolWait(true)) {
+              if (!psThreadPoolWait(true, true)) {
                   psError(PS_ERR_UNKNOWN, false, "Unable to smooth image");
                   psFree(calculation);
@@ -1752 +1752 @@
             }
         }
-        if (!psThreadPoolWait(true)) {
+        if (!psThreadPoolWait(true, true)) {
             psError(PS_ERR_UNKNOWN, false, "Error waiting for threads.");
             psFree(conv);
@@ -1787 +1787 @@
             }
         }
-        if (!psThreadPoolWait(true)) {
+        if (!psThreadPoolWait(true, true)) {
             psError(PS_ERR_UNKNOWN, false, "Error waiting for threads.");
             psFree(conv);

trunk/psLib/src/imageops/psImageCovariance.c

@@ -193 +193 @@
     }
 
-    if (threaded && !psThreadPoolWait(true)) {
+    if (threaded && !psThreadPoolWait(true, true)) {
         psError(PS_ERR_UNKNOWN, false, "Error waiting for threads.");
         return false;
@@ -360 +360 @@
         }
     }
-    if (threaded && !psThreadPoolWait(true)) {
+    if (threaded && !psThreadPoolWait(true, true)) {
         psError(PS_ERR_UNKNOWN, false, "Error waiting for threads.");
         return false;

trunk/psLib/src/sys/psThread.c

@@ -179 +179 @@
     PS_ASSERT_THREAD_TASK_NON_NULL(task, false);
 
+    // fprintf(stderr, "adding task %s\n", task->type);
+
     if (!tasks) {
         tasks = psHashAlloc(TASK_BUCKETS);
@@ -189 +191 @@
 {
     PS_ASSERT_STRING_NON_EMPTY(type, false);
+    // fprintf(stderr, "removing task %s\n", type);
 
     return psHashRemove(tasks, type);
@@ -223 +226 @@
 
         psThreadTask *task = psHashLookup(tasks, job->type); // Task to execute job
+        // fprintf(stderr, "launching job %s\n", job->type);
 #ifdef HAVE_BACKTRACE
         if (!task && bt_buffer) {
@@ -246 +250 @@
                  "invalid number of arguments to %s (%ld supplied, expected %d)",
                  task->type, job->args->n, task->nArgs);
+        // fprintf(stderr, "    thread for %s %p launching on %p\n", job->type, task->function, self);
+
+        // Run the job's function
         bool status = task->function(job); // Status of executing task
+
+        // fprintf(stderr, "    thread for %s %p finished on %p with status %d\n", job->type, task->function, self, status);
 
         // Put the completed job on the 'done' queue
@@ -306 +315 @@
 
 // call this function after you have added jobs to the queue and
-bool psThreadPoolWait(bool harvest)
-{
+bool psThreadPoolWait(bool harvest, bool harvestOnFailure)
+{
+    // fprintf(stderr, "psThreadPoolWait called with harvest: %d\n", harvest);
     if (!pool || pool->n == 0) {
         // No threads initialised, so everything's done
@@ -326 +336 @@
 #endif
 
+    // accumulate the number of faulted jobs that we encounter
+    int numFaults = 0;
     while (1) {
         // check for an error
@@ -331 +343 @@
             psThread *thread = pool->data[i];
             if (thread->fault) {
-                return false;
+                numFaults++;
             }
         }
@@ -354 +366 @@
             // Nothing in the queue and nothing more to add
             // Ensure everything is harvested, if requested
-            if (harvest) {
+            if (harvest || (numFaults && harvestOnFailure)) {
                 psThreadJobHarvest();
             }
             psThreadUnlock();
-            return true;
+            return numFaults == 0;
         }
 

trunk/psLib/src/sys/psThread.h

@@ -116 +116 @@
 /// Wait for the thread pool to finish
 ///
-/// This function blocks (waits in usleep) until either an error is detected on one of the threads or until ll
-/// threads are idle and no jobs are left on the queue
-bool psThreadPoolWait(bool harvest      // Harvest the jobs from the queue?
+/// This function blocks (waits in usleep) until all  threads are idle and no jobs
+/// are left on the queue
+/// returns success if all jobs return success, otherwise returns false
+bool psThreadPoolWait(bool harvest,         // Harvest the jobs from the queue?
+                      bool harvestOnFailure // If harvest is false, harvest the jobs if a failure is encountered
     );