[Commit-gnuradio] r7581 - in gnuradio/branches/developers/eb/gcell/src: apps include lib lib/spu

[eb]

Author: eb
Date: 2008-02-05 21:47:48 -0700 (Tue, 05 Feb 2008)
New Revision: 7581

Modified:
   gnuradio/branches/developers/eb/gcell/src/apps/benchmark_nop.cc
   gnuradio/branches/developers/eb/gcell/src/include/gc_mbox.h
   gnuradio/branches/developers/eb/gcell/src/include/gc_spu_args.h
   gnuradio/branches/developers/eb/gcell/src/lib/gc_job_manager_impl.cc
   gnuradio/branches/developers/eb/gcell/src/lib/gc_job_manager_impl.h
   gnuradio/branches/developers/eb/gcell/src/lib/spu/test_spu.c
Log:
buggy work-in-progress

Modified: gnuradio/branches/developers/eb/gcell/src/apps/benchmark_nop.cc
===================================================================
--- gnuradio/branches/developers/eb/gcell/src/apps/benchmark_nop.cc     
2008-02-06 02:58:25 UTC (rev 7580)
+++ gnuradio/branches/developers/eb/gcell/src/apps/benchmark_nop.cc     
2008-02-06 04:47:48 UTC (rev 7581)
@@ -81,7 +81,7 @@
   while (ncompleted < njobs){
     njds[ci^1] = 0;
     int n = mgr->wait_jobs(njds[ci], jds[ci], done, GC_WAIT_ANY);
-    // printf("%2d\n", n);
+    printf("%2d\n", n);
     if (n < 0){
       fprintf(stderr, "mgr->wait_jobs failed\n");
       break;
@@ -92,6 +92,7 @@
       }
       else {
        ncompleted++;
+       printf("ncompleted = %7d\n", ncompleted);
        if (nsubmitted < njobs){                    // submit another one
          if (mgr->submit_job(jds[ci][i])){
            jds[ci^1][njds[ci^1]++] = jds[ci][i];  // remember for next iter

Modified: gnuradio/branches/developers/eb/gcell/src/include/gc_mbox.h
===================================================================
--- gnuradio/branches/developers/eb/gcell/src/include/gc_mbox.h 2008-02-06 
02:58:25 UTC (rev 7580)
+++ gnuradio/branches/developers/eb/gcell/src/include/gc_mbox.h 2008-02-06 
04:47:48 UTC (rev 7581)
@@ -43,7 +43,8 @@
 
 // SPE to PPE (sent via SPE Write Outbound Interrupt Mailbox)
 
-#define OP_JOB_DONE    0x1     // arg is job_id from gc_job_desc_private
+#define OP_JOBS_DONE   0x1     // arg is 0 or 1, indicating which
+                               //  gc_completion_info_t contains the info
 
 
 #endif /* INCLUDED_GC_MBOX_H */

Modified: gnuradio/branches/developers/eb/gcell/src/include/gc_spu_args.h
===================================================================
--- gnuradio/branches/developers/eb/gcell/src/include/gc_spu_args.h     
2008-02-06 02:58:25 UTC (rev 7580)
+++ gnuradio/branches/developers/eb/gcell/src/include/gc_spu_args.h     
2008-02-06 04:47:48 UTC (rev 7581)
@@ -23,12 +23,34 @@
 
 #include <gc_types.h>
 
-// args passed to SPU at initialization time
+// args passed to SPE at initialization time
 
 typedef struct gc_spu_args {
-  gc_eaddr_t   queue;          // address of job queue
+  gc_eaddr_t   queue;          // address of job queue (gc_jd_queue_t *)
+  gc_eaddr_t   comp_info[2];   // completion info (gc_comp_info_t *)
   uint32_t     spu_idx;        // which spu we are: [0,nspus-1]
   uint32_t     nspus;          // number of spus we're using
 } _AL16 gc_spu_args_t;
 
+
+#define        GC_CI_NJOBS     62      // makes gc_comp_info 1 cache line long
+
+/*!
+ * \brief Used to return info to PPE on which jobs are completed.
+ *
+ * When each SPE is initalized, it is passed EA pointers to two of
+ * these structures.  The SPE uses these to communicate which jobs
+ * that it has worked on are complete.  The SPE notifies the PPE by
+ * sending an OP_JOBS_DONE message (see gc_mbox.h) with an argument of
+ * 0 or 1, indicating which of the two comp_info's to examine.  The
+ * SPE sets the in_use flag to 1 before DMA'ing to the PPE.  When the
+ * PPE is done with the structure, it must clear the in_use field to
+ * let the SPE know it can begin using it again.
+ */
+typedef struct gc_comp_info {
+  uint16_t     in_use;         // set by SPE, cleared by PPE when it's finished
+  uint16_t     ncomplete;      // number of valid job_id's
+  uint16_t     job_id[GC_CI_NJOBS];    // job_id's of completed jobs
+} _AL128 gc_comp_info_t;
+
 #endif /* INCLUDED_GC_SPU_ARGS_H */

Modified: gnuradio/branches/developers/eb/gcell/src/lib/gc_job_manager_impl.cc
===================================================================
--- gnuradio/branches/developers/eb/gcell/src/lib/gc_job_manager_impl.cc        
2008-02-06 02:58:25 UTC (rev 7580)
+++ gnuradio/branches/developers/eb/gcell/src/lib/gc_job_manager_impl.cc        
2008-02-06 04:47:48 UTC (rev 7581)
@@ -29,6 +29,7 @@
 #include <stdexcept>
 #include <stdlib.h>
 #include <atomic_dec_if_positive.h>
+#include <memory_barrier.h>
 
 static const size_t CACHE_LINE_SIZE = 128;
 
@@ -197,12 +198,21 @@
   // ----------------------------------------------------------------
   // create the spe contexts
 
+  // 1 spu_arg struct for each SPE
   assert(sizeof(gc_spu_args_t) % 16 == 0);
   d_spu_args =
     (gc_spu_args_t *) aligned_alloc(MAX_SPES * sizeof(gc_spu_args_t), 16);
   _d_spu_args_boost =
     boost::shared_ptr<void>((void *) d_spu_args, free_deleter());
 
+  // 2 completion info structs for each SPE (we double buffer them)
+  assert(sizeof(gc_comp_info_t) % CACHE_LINE_SIZE == 0);
+  d_comp_info =
+    (gc_comp_info_t *) aligned_alloc(2 * MAX_SPES * sizeof(gc_comp_info_t),
+                                    CACHE_LINE_SIZE);
+  _d_comp_info_boost =
+    boost::shared_ptr<void>((void *) d_comp_info, free_deleter());
+
   // get a handle to the spe program
 
   // FIXME pass this in (or something)
@@ -231,6 +241,8 @@
     d_worker[i].spe_idx = i;
     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->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].state = WS_INIT;
@@ -728,42 +740,64 @@
 }
 
 void
-gc_job_manager_impl::notify_client_job_is_done(unsigned int job_id)
+gc_job_manager_impl::notify_clients_jobs_are_done(unsigned int spe_num,
+                                                 unsigned int 
completion_info_idx)
 {
   const char *msg = "gc_job_manager_impl::notify_client_job_is_done (INTERNAL 
ERROR)";
-  if (job_id >= d_options.max_jobs){
-    // internal error, shouldn't happen
-    fprintf(stderr,"%s: invalid job_id = %d\n", msg, job_id);
-    return;
-  }
-  gc_job_desc *jd = &d_jd[job_id];
 
-  if (jd->sys.client_id >= d_options.max_client_threads){
-    // internal error, shouldn't happen
-    fprintf(stderr, "%s: invalid client_id = %d\n", msg, jd->sys.client_id);
-    return;
+  smp_rmb();  // order reads so we know that data sent from SPE is here
+
+  gc_comp_info_t *ci = &d_comp_info[2 * spe_num + (completion_info_idx & 0x1)];
+
+  if (1){
+    static int total_complete;
+    total_complete += ci->ncomplete;
+    printf("ppe: total_complete = %6d\n", total_complete);
   }
-  gc_client_thread_info *cti = &d_client_thread[jd->sys.client_id];
 
-  {
-    omni_mutex_lock    l(cti->d_mutex);
+  // FIXME sort by client_id so we only have to lock & signal once / client
 
-    // mark job done
-    bv_set(cti->d_jobs_done, job_id);
+  for (unsigned int i = 0; i < ci->ncomplete; i++){
 
-    // FIXME we could/should distinguish between CT_WAIT_ALL & CT_WAIT_ANY
+    unsigned int job_id = ci->job_id[i];
 
-    switch (cti->d_state){
-    case CT_WAIT_ANY:
-    case CT_WAIT_ALL:
-      cti->d_cond.signal();    // wake client thread up
-      break;
+    if (job_id >= d_options.max_jobs){
+      // internal error, shouldn't happen
+      fprintf(stderr,"%s: invalid job_id = %d\n", msg, job_id);
+      ci->in_use = 0;          // clear flag so SPE knows we're done with it
+      return;
+    }
+    gc_job_desc *jd = &d_jd[job_id];
 
-    case CT_NOT_WAITING:
-    default:
-      break;      // nop
+    if (jd->sys.client_id >= d_options.max_client_threads){
+      // internal error, shouldn't happen
+      fprintf(stderr, "%s: invalid client_id = %d\n", msg, jd->sys.client_id);
+      ci->in_use = 0;          // clear flag so SPE knows we're done with it
+      return;
     }
+    gc_client_thread_info *cti = &d_client_thread[jd->sys.client_id];
+    {
+      omni_mutex_lock  l(cti->d_mutex);
+
+      // mark job done
+      bv_set(cti->d_jobs_done, job_id);
+
+      // FIXME we could/should distinguish between CT_WAIT_ALL & CT_WAIT_ANY
+
+      switch (cti->d_state){
+      case CT_WAIT_ANY:
+      case CT_WAIT_ALL:
+       cti->d_cond.signal();   // wake client thread up
+       break;
+
+      case CT_NOT_WAITING:
+      default:
+       break;      // nop
+      }
+    }
   }
+
+  ci->in_use = 0;              // clear flag so SPE knows we're done with it
 }
 
 void
@@ -786,10 +820,10 @@
     else {
       for (int i = 0; i < n; i++){
        switch(MBOX_MSG_OP(msg[i])){
-       case OP_JOB_DONE:
+       case OP_JOBS_DONE:
          if (debug())
            printf("eh: job_done (0x%08x) from spu[%d]\n", msg[i], spe_num);
-         notify_client_job_is_done(MBOX_MSG_ARG(msg[i]));
+         notify_clients_jobs_are_done(spe_num, MBOX_MSG_ARG(msg[i]));
          break;
 
        case OP_EXIT:

Modified: gnuradio/branches/developers/eb/gcell/src/lib/gc_job_manager_impl.h
===================================================================
--- gnuradio/branches/developers/eb/gcell/src/lib/gc_job_manager_impl.h 
2008-02-06 02:58:25 UTC (rev 7580)
+++ gnuradio/branches/developers/eb/gcell/src/lib/gc_job_manager_impl.h 
2008-02-06 04:47:48 UTC (rev 7581)
@@ -97,6 +97,9 @@
   gc_spu_args_t                *d_spu_args;            // 16-byte aligned 
structs
   boost::shared_ptr<void> _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
+
   // used to coordinate communication w/ the event handling thread
   omni_mutex            d_eh_mutex;
   omni_condition        d_eh_cond;
@@ -136,7 +139,8 @@
 
   void create_event_handler();
   void set_eh_state(evt_handler_state s);
-  void notify_client_job_is_done(unsigned int job_id);
+  void notify_clients_jobs_are_done(unsigned int spe_num,
+                                   unsigned int completion_info_idx);
 
 public:
   void event_handler_loop();   // really private

Modified: gnuradio/branches/developers/eb/gcell/src/lib/spu/test_spu.c
===================================================================
--- gnuradio/branches/developers/eb/gcell/src/lib/spu/test_spu.c        
2008-02-06 02:58:25 UTC (rev 7580)
+++ gnuradio/branches/developers/eb/gcell/src/lib/spu/test_spu.c        
2008-02-06 04:47:48 UTC (rev 7581)
@@ -22,6 +22,7 @@
 #include <stdio.h>
 #include <spu_intrinsics.h>
 #include <spu_mfcio.h>
+#include <sync_utils.h>
 #include "gc_spu_args.h"
 #include "gc_job_desc.h"
 #include "gc_mbox.h"
@@ -30,13 +31,72 @@
 #include "gc_methods.h"
 #include "gc_delay.h"
 
-gc_spu_args_t  spu_args;
+static gc_spu_args_t   spu_args;
 
-#define ALPHA 0.001
-float avg_delta_t = 0.0;
+// our working copy of the completion info
+static gc_comp_info_t  comp_info = {  
+  .in_use = 1,
+  .ncomplete = 0
+};
 
+static int ci_idx = 0;         // index of current comp_info
+static int ci_tags;            // two consecutive dma tags
+
 // ------------------------------------------------------------------------
 
+/*
+ * Wait until EA copy of comp_info[idx].in_use is 0
+ */
+static void
+wait_for_ppe_to_be_done_with_comp_info(int idx)
+{
+  char _tmp[256];
+  char *buf = (char *) ALIGN(_tmp, 128);       // get cache-aligned buffer
+  gc_comp_info_t *p = (gc_comp_info_t *) buf;
+
+  do {
+    mfc_getllar(buf, spu_args.comp_info[idx], 0, 0);
+    spu_readch(MFC_RdAtomicStat);
+    if (p->in_use == 0)
+      return;
+
+    gc_udelay(5);    // FIXME use the "lock-line reservation lost" event
+
+  } while (1);
+}
+
+static void
+flush_completion_info(void)
+{
+  if (comp_info.ncomplete == 0)
+    return;
+  
+  if (1){
+    static int total_complete = 0;
+    total_complete += comp_info.ncomplete;
+    printf("spu[%d]: tc = %6d\n", spu_args.spu_idx, total_complete);
+  }
+
+  // ensure that PPE is done with the buffer we're about to overwrite
+  wait_for_ppe_to_be_done_with_comp_info(ci_idx);
+
+  int tag = ci_tags + ci_idx;
+  
+  // dma the comp_info out to PPE
+  mfc_putf(&comp_info, spu_args.comp_info[ci_idx], sizeof(gc_comp_info_t), 
tag, 0, 0);
+  mfc_write_tag_mask(1 << tag);                // the tag we're interested in
+  mfc_read_tag_status_all();           // wait for DMA to complete
+
+  // send it a message
+  spu_writech(SPU_WrOutIntrMbox, MK_MBOX_MSG(OP_JOBS_DONE, ci_idx));
+
+  // ci_idx ^= 0x1;    // switch buffers
+  comp_info.in_use = 1;
+  comp_info.ncomplete = 0;
+}
+
+// ------------------------------------------------------------------------
+
 //#define BACKOFF_CAP ((1 << 12) - 1)  // 4095 cycles, about 1.3 us
 //#define BACKOFF_CAP ((1 << 16) - 1)  // 65535 cycles, about 20.4 us
 #define BACKOFF_CAP ((1 << 18) - 1)    // 262143 cycles, about 81.9 us
@@ -113,24 +173,26 @@
   // FIXME copy indirect args out
   // FIXME copy jd back out
 
-  // How long are we blocking for the PPE?
-  spu_writech(SPU_WrDec, 0xffffffff);
-  int start, stop;
-  start = spu_readch(SPU_RdDec);
   
-  // Tell PPE we're done with the job
-  spu_writech(SPU_WrOutIntrMbox,
-             MK_MBOX_MSG(OP_JOB_DONE, jd->sys.job_id));
+  // Tell PPE we're done with the job.
+  //
+  // We accrue this info until the PPE is ready to accept it or we run
+  // out of room.  The PPE is ready to accept if we can write to the
+  // SPU_WrOutIntrMbox without blocking.
 
-  stop = spu_readch(SPU_RdDec);
-  float delta_t = start - stop;
-  avg_delta_t = (delta_t * ALPHA) + (avg_delta_t * (1.0 - ALPHA));
-}
+  comp_info.job_id[comp_info.ncomplete++] = jd->sys.job_id;
 
-static void
-print_stats(void)
-{
-  printf("spu[%d] avg_delta_t = %f\n", spu_args.spu_idx, avg_delta_t);
+  if (0){
+    static int total_complete = 0;
+    total_complete++;
+    printf("spu[%d]: tc = %6d\n", spu_args.spu_idx, total_complete);
+  }
+
+  if (spu_readchcnt(SPU_WrOutIntrMbox) != 0
+      || comp_info.ncomplete == GC_CI_NJOBS){
+
+    flush_completion_info();
+  }
 }
 
 void
@@ -147,7 +209,7 @@
       int msg = spu_readch(SPU_RdInMbox);
       // printf("spu[%d] mbox_msg: 0x%08x\n", spu_args.spu_idx, msg);
       if (MBOX_MSG_OP(msg) == OP_EXIT){
-       print_stats();
+       flush_completion_info();
        return;
       }
     }
@@ -169,12 +231,15 @@
      unsigned long long envp __attribute__((unused)))
 {
   sys_tags_init();
+  ci_tags = mfc_multi_tag_reserve(2);
 
   // dma the args in
   mfc_get(&spu_args, argp, sizeof(spu_args), sys_tag, 0, 0);
   mfc_write_tag_mask(1 << sys_tag);    // the tag we're interested in
   mfc_read_tag_status_all();           // wait for DMA to complete
 
+  // printf("&comp_info = %p\n", &comp_info);
+
   // printf("spu[%d] queue = 0x%llx\n", spu_args.spu_idx, spu_args.queue);
 
   backoff_init();              // initialize backoff parameters