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

[eb]

Author: eb
Date: 2008-02-14 12:54:11 -0700 (Thu, 14 Feb 2008)
New Revision: 7681

Modified:
   gnuradio/branches/developers/eb/gcell-multi-q/src/apps/benchmark_nop.cc
   gnuradio/branches/developers/eb/gcell-multi-q/src/lib/gc_job_manager.h
   gnuradio/branches/developers/eb/gcell-multi-q/src/lib/gc_job_manager_impl.cc
   gnuradio/branches/developers/eb/gcell-multi-q/src/lib/gc_job_manager_impl.h
Log:
work-in-progress for merge into new copy

Modified: 
gnuradio/branches/developers/eb/gcell-multi-q/src/apps/benchmark_nop.cc
===================================================================
--- gnuradio/branches/developers/eb/gcell-multi-q/src/apps/benchmark_nop.cc     
2008-02-14 19:52:33 UTC (rev 7680)
+++ gnuradio/branches/developers/eb/gcell-multi-q/src/apps/benchmark_nop.cc     
2008-02-14 19:54:11 UTC (rev 7681)
@@ -37,7 +37,7 @@
 }
 
 static void
-run_test(unsigned int nspes, unsigned int usecs)
+run_test(unsigned int nspes, unsigned int usecs, unsigned int nqueues)
 {
   static const int NJDS = 32;
   static const int njobs = static_cast<int>(500e3);
@@ -52,6 +52,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
@@ -122,9 +123,10 @@
 {
   unsigned int nspes = 0;
   unsigned int usecs = 0;
+  unsigned int nqueues = 0;
   int ch;
 
-  while ((ch = getopt(argc, argv, "n:u:")) != EOF){
+  while ((ch = getopt(argc, argv, "n:u:q:")) != EOF){
     switch(ch){
     case 'n':
       nspes = strtol(optarg, 0, 0);
@@ -134,13 +136,17 @@
       usecs = strtol(optarg, 0, 0);
       break;
 
+    case 'q':
+      nqueues = strtol(optarg, 0, 0);
+      break;
+
     case '?':
     default:
-      fprintf(stderr, "usage: benchmark_nop [-n <nspes>] [-u <udelay>]\n");
+      fprintf(stderr, "usage: benchmark_nop [-n <nspes>] [-u <udelay>] [-q 
<nqueues>]\n");
       return 1;
     }
   }
 
-  run_test(nspes, usecs);
+  run_test(nspes, usecs, nqueues);
   return 0;
 }

Modified: gnuradio/branches/developers/eb/gcell-multi-q/src/lib/gc_job_manager.h
===================================================================
--- gnuradio/branches/developers/eb/gcell-multi-q/src/lib/gc_job_manager.h      
2008-02-14 19:52:33 UTC (rev 7680)
+++ gnuradio/branches/developers/eb/gcell-multi-q/src/lib/gc_job_manager.h      
2008-02-14 19:54:11 UTC (rev 7681)
@@ -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/src/lib/gc_job_manager_impl.cc
===================================================================
--- 
gnuradio/branches/developers/eb/gcell-multi-q/src/lib/gc_job_manager_impl.cc    
    2008-02-14 19:52:33 UTC (rev 7680)
+++ 
gnuradio/branches/developers/eb/gcell-multi-q/src/lib/gc_job_manager_impl.cc    
    2008-02-14 19:54:11 UTC (rev 7681)
@@ -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/src/lib/gc_job_manager_impl.h
===================================================================
--- gnuradio/branches/developers/eb/gcell-multi-q/src/lib/gc_job_manager_impl.h 
2008-02-14 19:52:33 UTC (rev 7680)
+++ gnuradio/branches/developers/eb/gcell-multi-q/src/lib/gc_job_manager_impl.h 
2008-02-14 19:54:11 UTC (rev 7681)
@@ -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);