[Commit-gnuradio] r7685 - in gnuradio/branches/developers/eb/gcell-multi-q-2/src: apps lib

[eb]

Author: eb
Date: 2008-02-14 13:11:29 -0700 (Thu, 14 Feb 2008)
New Revision: 7685

Modified:
   gnuradio/branches/developers/eb/gcell-multi-q-2/src/apps/benchmark_nop.cc
   gnuradio/branches/developers/eb/gcell-multi-q-2/src/lib/gc_job_manager.h
   
gnuradio/branches/developers/eb/gcell-multi-q-2/src/lib/gc_job_manager_impl.cc
   gnuradio/branches/developers/eb/gcell-multi-q-2/src/lib/gc_job_manager_impl.h
Log:
merged eb/gcell-multi-q r7662:7681 into gcell-multi-q-2

Modified: 
gnuradio/branches/developers/eb/gcell-multi-q-2/src/apps/benchmark_nop.cc
===================================================================
--- gnuradio/branches/developers/eb/gcell-multi-q-2/src/apps/benchmark_nop.cc   
2008-02-14 20:06:27 UTC (rev 7684)
+++ gnuradio/branches/developers/eb/gcell-multi-q-2/src/apps/benchmark_nop.cc   
2008-02-14 20:11:29 UTC (rev 7685)
@@ -37,7 +37,7 @@
 }
 
 static void
-run_test(unsigned int nspes, unsigned int usecs, int njobs)
+run_test(unsigned int nspes, unsigned int usecs, int njobs, unsigned int 
nqueues)
 {
   static const int NJDS = 32;
   int nsubmitted = 0;
@@ -51,6 +51,7 @@
   gc_jm_options opts;
   opts.nspes = nspes;
   opts.gang_schedule = true;
+  opts.nqueues = nqueues;
   gc_job_manager *mgr = gc_make_job_manager(&opts);
 
   // allocate and init all job descriptors
@@ -123,9 +124,10 @@
   unsigned int nspes = 0;
   unsigned int usecs = 0;
   int njobs = 500000;
+  unsigned int nqueues = 0;
   int ch;
 
-  while ((ch = getopt(argc, argv, "n:u:N:")) != EOF){
+  while ((ch = getopt(argc, argv, "n:u:N:q:")) != EOF){
     switch(ch){
     case 'n':
       nspes = strtol(optarg, 0, 0);
@@ -139,13 +141,18 @@
       njobs = strtol(optarg, 0, 0);
       break;
 
+    case 'q':
+      nqueues = strtol(optarg, 0, 0);
+      break;
+
+    case 'h':
     case '?':
-    default:
-      fprintf(stderr, "usage: benchmark_nop [-n <nspes>] [-u <udelay>] [-N 
<njobs>]\n");
+      fprintf(stderr, "usage: benchmark_nop [-n <nspes>] [-u <udelay>] [-N 
<njobs>] [-q <nqueues>]\n");
       return 1;
     }
   }
 
-  run_test(nspes, usecs, njobs);
+  run_test(nspes, usecs, njobs, nqueues);
+
   return 0;
 }

Modified: 
gnuradio/branches/developers/eb/gcell-multi-q-2/src/lib/gc_job_manager.h
===================================================================
--- gnuradio/branches/developers/eb/gcell-multi-q-2/src/lib/gc_job_manager.h    
2008-02-14 20:06:27 UTC (rev 7684)
+++ gnuradio/branches/developers/eb/gcell-multi-q-2/src/lib/gc_job_manager.h    
2008-02-14 20:11:29 UTC (rev 7685)
@@ -42,10 +42,11 @@
   unsigned int nspes;              // how many SPEs shall we use? 0 -> all of 
them
   bool gang_schedule;              // shall we gang schedule?
   bool use_affinity;               // shall we try for affinity (FIXME not 
implmented)
+  unsigned int nqueues;                    // FIXME debugging
 
   gc_jm_options() :
     max_jobs(0), max_client_threads(0), nspes(0),
-    gang_schedule(true), use_affinity(false)
+    gang_schedule(true), use_affinity(false), nqueues(0)
   {
   }
 };

Modified: 
gnuradio/branches/developers/eb/gcell-multi-q-2/src/lib/gc_job_manager_impl.cc
===================================================================
--- 
gnuradio/branches/developers/eb/gcell-multi-q-2/src/lib/gc_job_manager_impl.cc  
    2008-02-14 20:06:27 UTC (rev 7684)
+++ 
gnuradio/branches/developers/eb/gcell-multi-q-2/src/lib/gc_job_manager_impl.cc  
    2008-02-14 20:11:29 UTC (rev 7685)
@@ -114,7 +114,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_client_thread(0)
+    d_client_thread(0), d_nextq(0)
 {
   if (!s_key_initialized){
     int r = pthread_key_create(&s_client_key, client_key_destructor);
@@ -186,15 +186,22 @@
     }
   }
 
+
   // ----------------------------------------------------------------
-  // initalize the job queue
+  // initalize the job queues
   
-  d_queue = (gc_jd_queue_t *) aligned_alloc(sizeof(gc_jd_queue_t));
-  _d_queue_boost =
-    boost::shared_ptr<void>((void *) d_queue, free_deleter());
-  gc_jd_queue_init(d_queue);
+  if (d_options.nqueues == 0)          // work out better default
+    d_options.nqueues = 2;
+  if ((int) d_options.nqueues > nusable_spes)
+    d_options.nqueues = nusable_spes;
 
+  for (unsigned int i = 0; i < d_options.nqueues; i++){
+    d_queue[i] = (gc_jd_queue_t *) aligned_alloc(sizeof(gc_jd_queue_t));
+    _d_queue_boost.push_back(void_sptr((void *) d_queue[i], free_deleter()));
+    gc_jd_queue_init(d_queue[i]);
+  }
 
+
   // ----------------------------------------------------------------
   // create the spe contexts
 
@@ -239,12 +246,13 @@
       throw std::runtime_error("spe_context_create");
     }
     d_worker[i].spe_idx = i;
+    d_worker[i].node = 0;                                      // FIXME assign 
node
     d_worker[i].spu_args = &d_spu_args[i];
-    d_worker[i].spu_args->queue = ptr_to_ea(d_queue);
+    d_worker[i].spu_args->queue = ptr_to_ea(d_queue[i % d_options.nqueues]);   
// FIXME which queue
     d_worker[i].spu_args->comp_info[0] = ptr_to_ea(&d_comp_info[2*i+0]);
     d_worker[i].spu_args->comp_info[1] = ptr_to_ea(&d_comp_info[2*i+1]);
     d_worker[i].spu_args->spu_idx = i;
-    d_worker[i].spu_args->nspus = d_options.nspes;
+    d_worker[i].spu_args->nspus = d_options.nspes / d_options.nqueues;
     d_worker[i].state = WS_INIT;
 
     int r = spe_program_load(d_worker[i].spe_ctx, spe_image);
@@ -330,7 +338,7 @@
 
   d_jd = 0;            // handled via _d_jd_boost
   d_free_list = 0;     // handled via _d_free_list_boost
-  d_queue = 0;         // handled via _d_queue_boost
+  //d_queue = 0;f      // handled via _d_queue_boost
 
   // clear cti, since we've deleted the underlying data
   pthread_setspecific(s_client_key, 0);
@@ -495,7 +503,16 @@
 
   // FIXME keep count of jobs in progress?
   
-  gc_jd_queue_enqueue(d_queue, jd);
+  // There is a race here in the modification of d_nextq.  
+  // However, I think that fixing it will cost more than it's worth.
+  // Worst case, some queue gets more or less jobs assigned to it than
+  // it should.
+
+  unsigned nq = d_nextq;
+  d_nextq = inc_nextq(d_nextq);
+
+  gc_jd_queue_enqueue(d_queue[nq], jd);
+
   return true;
 }
 

Modified: 
gnuradio/branches/developers/eb/gcell-multi-q-2/src/lib/gc_job_manager_impl.h
===================================================================
--- 
gnuradio/branches/developers/eb/gcell-multi-q-2/src/lib/gc_job_manager_impl.h   
    2008-02-14 20:06:27 UTC (rev 7684)
+++ 
gnuradio/branches/developers/eb/gcell-multi-q-2/src/lib/gc_job_manager_impl.h   
    2008-02-14 20:11:29 UTC (rev 7685)
@@ -32,6 +32,7 @@
 #include <boost/shared_ptr.hpp>
 #include <boost/scoped_array.hpp>
 
+typedef boost::shared_ptr<void> void_sptr;
 typedef boost::shared_ptr<spe_gang_context> spe_gang_context_sptr;
 typedef boost::shared_ptr<spe_program_handle_t> spe_program_handle_sptr;
 typedef boost::scoped_array<gc_client_thread_info> gc_client_thread_info_sa;
@@ -48,11 +49,12 @@
   volatile worker_state        state;
   unsigned int         spe_idx;        // [0, nspes-1]
   spe_context_ptr_t    spe_ctx;
+  unsigned int         node;           // which NUMA node we're bound to
   pthread_t            thread;
   gc_spu_args_t                *spu_args;      // pointer to 16-byte aligned 
struct
 
   worker_ctx()
-    : state(WS_FREE), spe_idx(0), spe_ctx(0),
+    : state(WS_FREE), spe_idx(0), spe_ctx(0), node(0),
       thread(0), spu_args(0) {}
   ~worker_ctx();
 };
@@ -87,6 +89,7 @@
 class gc_job_manager_impl : public gc_job_manager
 {
   enum { MAX_SPES =  16 };
+  enum { MAX_QUEUES = MAX_SPES };
 
   int                    d_debug;
   gc_jm_options                  d_options;
@@ -95,10 +98,10 @@
 
   worker_ctx            d_worker[MAX_SPES];    // SPE ctx, thread, etc
   gc_spu_args_t                *d_spu_args;            // 16-byte aligned 
structs
-  boost::shared_ptr<void> _d_spu_args_boost;   // hack for automatic storage 
mgmt
+  void_sptr            _d_spu_args_boost;      // hack for automatic storage 
mgmt
 
   gc_comp_info_t       *d_comp_info;           // 128-byte aligned structs
-  boost::shared_ptr<void> _d_comp_info_boost;  // hack for automatic storage 
mgmt
+  void_sptr            _d_comp_info_boost;     // hack for automatic storage 
mgmt
 
   // used to coordinate communication w/ the event handling thread
   omni_mutex            d_eh_mutex;
@@ -112,7 +115,7 @@
   // All of the job descriptors are hung off of here.
   // We allocate them all in a single cache aligned chunk.
   gc_job_desc_t                *d_jd;                  // [options.max_jobs]
-  boost::shared_ptr<void> _d_jd_boost;         // hack for automatic storage 
mgmt
+  void_sptr            _d_jd_boost;            // hack for automatic storage 
mgmt
 
   gc_client_thread_info_sa d_client_thread;    // [options.max_client_threads]
 
@@ -123,15 +126,16 @@
   // This contains the storage for all the bitvectors used by the job
   // manager.  There's 1 for each client thread, in the d_jobs_done
   // field.  We allocate them all in a single cache aligned chunk.
-  boost::shared_ptr<void> _d_all_bitvectors;   // hack for automatic storage 
mgmt
+  void_sptr            _d_all_bitvectors;      // hack for automatic storage 
mgmt
 
   // Lock free stack where we keep track of the free job descriptors.
   gc_jd_stack_t                *d_free_list;           // stack of free job 
descriptors
-  boost::shared_ptr<void> _d_free_list_boost;  // hack for automatic storage 
mgmt
+  void_sptr            _d_free_list_boost;     // hack for automatic storage 
mgmt
 
-  // The PPE inserts jobs here; SPEs pull jobs from here.
-  gc_jd_queue_t                *d_queue;               // job queue
-  boost::shared_ptr<void> _d_queue_boost;      // hack for automatic storage 
mgmt
+  // The PPEs insert jobs here; SPEs pull jobs from here.
+  unsigned int          d_nextq;               // next queue to insert into
+  gc_jd_queue_t                *d_queue[MAX_QUEUES];   // job queue
+  std::vector<void_sptr>_d_queue_boost;                // hack for automatic 
storage mgmt
 
 
   gc_client_thread_info *alloc_cti();
@@ -151,6 +155,10 @@
   void print_event(spe_event_unit_t *evt);
   void handle_event(spe_event_unit_t *evt);
 
+  unsigned int inc_nextq(unsigned int nq){
+    return nq >= d_options.nqueues - 1 ? 0 : nq + 1;
+  }
+
   // bitvector ops
   void bv_zero(unsigned long *bv);
   void bv_clr(unsigned long *bv, unsigned int bitno);