[Commit-gnuradio] r10223 - in gnuradio/branches/developers/eb/gcell-wip/gcell/lib/runtime: . spu

[eb]

Author: eb
Date: 2009-01-14 01:25:44 -0700 (Wed, 14 Jan 2009)
New Revision: 10223

Modified:
   
gnuradio/branches/developers/eb/gcell-wip/gcell/lib/runtime/gc_job_manager_impl.cc
   
gnuradio/branches/developers/eb/gcell-wip/gcell/lib/runtime/gc_job_manager_impl.h
   gnuradio/branches/developers/eb/gcell-wip/gcell/lib/runtime/spu/gc_main.c
   
gnuradio/branches/developers/eb/gcell-wip/gcell/lib/runtime/spu/gc_spu_jd_queue.c
Log:
work-in-progress: improvements to shutdown

Modified: 
gnuradio/branches/developers/eb/gcell-wip/gcell/lib/runtime/gc_job_manager_impl.cc
===================================================================
--- 
gnuradio/branches/developers/eb/gcell-wip/gcell/lib/runtime/gc_job_manager_impl.cc
  2009-01-14 08:16:12 UTC (rev 10222)
+++ 
gnuradio/branches/developers/eb/gcell-wip/gcell/lib/runtime/gc_job_manager_impl.cc
  2009-01-14 08:25:44 UTC (rev 10223)
@@ -108,7 +108,7 @@
   : d_debug(0), d_spu_args(0),
     d_eh_cond(&d_eh_mutex), d_eh_thread(0), d_eh_state(EHS_INIT),
     d_shutdown_requested(false),
-    d_jc_cond(&d_jc_mutex), d_jc_thread(0), d_jc_njobs_active(0),
+    d_jc_cond(&d_jc_mutex), d_jc_thread(0), d_jc_state(JCS_INIT), 
d_jc_njobs_active(0),
     d_client_thread(0), d_ea_args_maxsize(0),
     d_proc_def(0), d_proc_def_ls_addr(0), d_nproc_defs(0)
 {
@@ -357,8 +357,11 @@
 {
   omni_mutex_lock      l(d_eh_mutex);
 
-  d_shutdown_requested = true;         // set flag for event handler thread
-  d_jc_cond.signal();                  // wake up job completer
+  {
+    omni_mutex_lock    l2(d_jc_mutex);
+    d_shutdown_requested = true;       // set flag for event handler thread
+    d_jc_cond.signal();                        // wake up job completer
+  }
 
   // should only happens during early QA code
   if (d_eh_thread == 0 && d_eh_state == EHS_INIT)
@@ -440,13 +443,18 @@
 ////////////////////////////////////////////////////////////////////////
 
 
-inline void
+inline bool
 gc_job_manager_impl::incr_njobs_active()
 {
   omni_mutex_lock      l(d_jc_mutex);
 
+  if (d_shutdown_requested)
+    return false;
+
   if (d_jc_njobs_active++ == 0)        // signal on 0 to 1 transition
     d_jc_cond.signal();
+
+  return true;
 }
 
 inline void
@@ -512,11 +520,6 @@
 bool
 gc_job_manager_impl::submit_job(gc_job_desc *jd)
 {
-  if (unlikely(d_shutdown_requested)){
-    jd->status = JS_SHUTTING_DOWN;
-    return false;
-  }
-
   // Ensure it's one of our job descriptors
 
   if (jd < d_jd || jd >= &d_jd[d_options.max_jobs]){
@@ -559,7 +562,11 @@
   jd->status = JS_OK;
   jd->sys.client_id = cti->d_client_id;
 
-  incr_njobs_active();
+  if (!incr_njobs_active()){
+    jd->status = JS_SHUTTING_DOWN;
+    return false;
+  }
+  
   gc_jd_queue_enqueue(d_queue, jd);
   tell_spes_to_check_queue();
   return true;
@@ -1052,7 +1059,7 @@
 gc_job_manager_impl::event_handler_loop()
 {
   static const int MAX_EVENTS = 16;
-  static const int TIMEOUT = 20;       // how long to block in milliseconds
+  static const int TIMEOUT = 50;       // how long to block in milliseconds
 
   spe_event_unit_t events[MAX_EVENTS];
 
@@ -1067,18 +1074,17 @@
   while (1){
     switch(d_eh_state){
 
-    case EHS_RUNNING:      // normal stuff
+    case EHS_RUNNING:                  // normal stuff
       if (d_shutdown_requested) {
        set_eh_state(EHS_SHUTTING_DOWN);
       }
       break;
 
     case EHS_SHUTTING_DOWN:
-
-      // FIXME wait until job queue is empty, then tell them to exit
-
-      send_all_spes(MK_MBOX_MSG(OP_EXIT, 0));
-      set_eh_state(EHS_WAITING_FOR_WORKERS_TO_DIE);
+      if (d_jc_state == JCS_DEAD){
+       send_all_spes(MK_MBOX_MSG(OP_EXIT, 0));
+       set_eh_state(EHS_WAITING_FOR_WORKERS_TO_DIE);
+      }
       break;
 
     case EHS_WAITING_FOR_WORKERS_TO_DIE:
@@ -1157,11 +1163,18 @@
 void
 gc_job_manager_impl::job_completer_loop()
 {
-  while (!d_shutdown_requested){
+  d_jc_state = JCS_RUNNING;
+
+  while (1){
     {
       omni_mutex_lock  l(d_jc_mutex);
-      while (d_jc_njobs_active <= 0 && !d_shutdown_requested)
+      if (d_jc_njobs_active == 0){
+       if (d_shutdown_requested){
+         d_jc_state = JCS_DEAD;
+         return;
+       }
        d_jc_cond.wait();
+      }
     }
 
     poll_for_job_completion();

Modified: 
gnuradio/branches/developers/eb/gcell-wip/gcell/lib/runtime/gc_job_manager_impl.h
===================================================================
--- 
gnuradio/branches/developers/eb/gcell-wip/gcell/lib/runtime/gc_job_manager_impl.h
   2009-01-14 08:16:12 UTC (rev 10222)
+++ 
gnuradio/branches/developers/eb/gcell-wip/gcell/lib/runtime/gc_job_manager_impl.h
   2009-01-14 08:25:44 UTC (rev 10223)
@@ -65,6 +65,12 @@
   EHS_DEAD,            // thread is dead
 };
 
+enum job_completer_state {
+  JCS_INIT,            // being initialized
+  JCS_RUNNING,         // thread is running
+  JCS_DEAD,            // thread is dead
+};
+
 struct spe_event_handler {
   spe_event_handler_ptr_t      ptr;
 
@@ -112,6 +118,7 @@
   omni_mutex            d_jc_mutex;
   omni_condition        d_jc_cond;
   pthread_t             d_jc_thread;           // the job completion thread
+  volatile job_completer_state d_jc_state;
   int                   d_jc_njobs_active;     // # of jobs submitted but not 
yet reaped
 
   // All of the job descriptors are hung off of here.
@@ -163,7 +170,7 @@
   bool send_spe(unsigned int spe, uint32_t msg);
   void print_event(spe_event_unit_t *evt);
   void handle_event(spe_event_unit_t *evt);
-  void incr_njobs_active();
+  bool incr_njobs_active();
   void decr_njobs_active(int n);
   void tell_spes_to_check_queue();
   void poll_for_job_completion();

Modified: 
gnuradio/branches/developers/eb/gcell-wip/gcell/lib/runtime/spu/gc_main.c
===================================================================
--- gnuradio/branches/developers/eb/gcell-wip/gcell/lib/runtime/spu/gc_main.c   
2009-01-14 08:16:12 UTC (rev 10222)
+++ gnuradio/branches/developers/eb/gcell-wip/gcell/lib/runtime/spu/gc_main.c   
2009-01-14 08:25:44 UTC (rev 10223)
@@ -664,9 +664,9 @@
       if (MBOX_MSG_OP(msg) == OP_CHECK_QUEUE){
 
        while (1){
-         //int delay = (int)(8191.0 * gc_uniform_deviate());   // uniformly in 
[0, 2.6us]
-         //gc_cdelay(delay);
-         gc_cdelay(4095);
+         int delay = (int)(3200.0 * gc_uniform_deviate());     // uniformly in 
[0, 1.0us]
+         gc_cdelay(delay);
+         //gc_cdelay(4095);
 
          gc_dequeue_status_t s =
            gc_jd_queue_dequeue(spu_args.queue, &jd_ea, ci_tags + ci_idx, &jd);

Modified: 
gnuradio/branches/developers/eb/gcell-wip/gcell/lib/runtime/spu/gc_spu_jd_queue.c
===================================================================
--- 
gnuradio/branches/developers/eb/gcell-wip/gcell/lib/runtime/spu/gc_spu_jd_queue.c
   2009-01-14 08:16:12 UTC (rev 10222)
+++ 
gnuradio/branches/developers/eb/gcell-wip/gcell/lib/runtime/spu/gc_spu_jd_queue.c
   2009-01-14 08:25:44 UTC (rev 10223)
@@ -30,7 +30,7 @@
 extern int gc_sys_tag;
 
 #define        INITIAL_BACKOFF    32.0
-#define MAX_BACKOFF    16384.0
+#define MAX_BACKOFF     8192.0         /* 2.6us */
 #define        RANDOM_WEIGHT       0.2
 
 static float