Changeset 23530 for trunk/Ohana/src/opihi/pantasks/CheckTasks.c

Timestamp:

Mar 25, 2009, 11:56:09 AM (17 years ago)

Author:

eugene

Message:

upgrade threading/locking model for pantasks: changes merged from eam_branch_20090322

File:

• trunk/Ohana/src/opihi/pantasks/CheckTasks.c (modified) (3 diffs)

trunk/Ohana/src/opihi/pantasks/CheckTasks.c

@@ -7 +7 @@
   int status;
   float time_running, next_timeout, fuzz;
-  // struct timeval now;
 
   // actual maximum delay is controlled in job_threads.c
   next_timeout = 1.0;
 
+  JobTaskLock();
   /** test all tasks: ready to test? ready to run? **/
   while ((task = NextTask ()) != NULL) {
@@ -54 +54 @@
     /* ready to run? : run task.exec macro */
     if (task[0].exec != NULL) {
+      // we need to unlock JobTask since JobTaskLock is called inside CommandLock in the client thread
+      JobTaskUnlock();
+      CommandLock();
       status = exec_loop (task[0].exec);
+      CommandUnlock();
+      JobTaskLock();
       if (!status) {
         continue;
       }
     }
-
-    // gettimeofday (&now, (void *) NULL);
-    // fprintf (stderr, "t1: %d %6d  - \n", now.tv_sec, now.tv_usec);
 
     /* check if there are errors with this task */
@@ -68 +70 @@
     }
     
-    // gettimeofday (&now, (void *) NULL);
-    // fprintf (stderr, "t2: %d %6d  - \n", now.tv_sec, now.tv_usec);
-
     /* construct job from task */
     job = CreateJob (task);
     if (DEBUG) fprintf (stderr, "create job: (%zx) %d of %d\n", (size_t) job[0].stdout_buff.buffer, job[0].stdout_buff.Nbuffer, job[0].stdout_buff.Nalloc); 
 
-    // gettimeofday (&now, (void *) NULL);
-    // fprintf (stderr, "t3: %d %6d  - \n", now.tv_sec, now.tv_usec);
-
-    /* execute job - XXX add status test */
-    SubmitJob (job);
-
-    // fprintf (stderr, "nl: %d %6d  - ",
-    // task[0].last.tv_sec, task[0].last.tv_usec);
-
-    /* increment job counters */
-    task[0].Njobs ++;
-    task[0].Npending ++;
-
-    // fprintf (stderr, "%d %6d\n", 
-    // task[0].last.tv_sec, task[0].last.tv_usec);
+    /* execute job */
+    if (!SubmitJob (job)) {
+      DeleteJob (job);
+      continue;
+    }
+    task[0].Njobs ++; // number of jobs successfully submitted
 
     /* increment Nrun for inclusive ranges with Nmax */
     BumpTimeRanges (task[0].ranges, task[0].Nranges);
   }
+  JobTaskUnlock();
   return (next_timeout);
 }