[Commit-gnuradio] r7021 - in gnuradio/branches/developers/eb/gcell/src: include lib

[eb]

Author: eb
Date: 2007-11-25 01:24:13 -0700 (Sun, 25 Nov 2007)
New Revision: 7021

Added:
   gnuradio/branches/developers/eb/gcell/src/include/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.h
Log:
work-in-progress on cell job manager.  Time to write some SPE code ;)

Added: gnuradio/branches/developers/eb/gcell/src/include/gc_spu_args.h
===================================================================
--- gnuradio/branches/developers/eb/gcell/src/include/gc_spu_args.h             
                (rev 0)
+++ gnuradio/branches/developers/eb/gcell/src/include/gc_spu_args.h     
2007-11-25 08:24:13 UTC (rev 7021)
@@ -0,0 +1,34 @@
+/* -*- c++ -*- */
+/*
+ * Copyright 2007 Free Software Foundation, Inc.
+ * 
+ * This file is part of GNU Radio
+ * 
+ * GNU Radio is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 3, or (at your option)
+ * any later version.
+ * 
+ * GNU Radio is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ * 
+ * You should have received a copy of the GNU General Public License along
+ * with this program; if not, write to the Free Software Foundation, Inc.,
+ * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+ */
+#ifndef INCLUDED_GC_SPU_ARGS_H
+#define INCLUDED_GC_SPU_ARGS_H
+
+#include <gc_types.h>
+
+// args passed to SPU at initialization time
+
+typedef struct gc_spu_args {
+  gc_eaddr_t   queue;          // address of job queue
+  uint32_t     spu_idx;        // which spu we are: [0,nspus-1]
+  uint32_t     _pad;
+} _AL16 gc_spu_args_t;
+
+#endif /* INCLUDED_GC_SPU_ARGS_H */


Property changes on: 
gnuradio/branches/developers/eb/gcell/src/include/gc_spu_args.h
___________________________________________________________________
Name: svn:eol-style
   + native

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        
2007-11-25 02:48:29 UTC (rev 7020)
+++ gnuradio/branches/developers/eb/gcell/src/lib/gc_job_manager_impl.cc        
2007-11-25 08:24:13 UTC (rev 7021)
@@ -60,21 +60,26 @@
 /*
  * Return pointer to cache-aligned chunk of storage of size size bytes.
  * Throw if can't allocate memory.  The storage should be freed
- * using "free" when done.
+ * with "free" when done.  The memory is initialized to zero.
  */
 static void *
-aligned_alloc(size_t size)
+aligned_alloc(size_t size, size_t alignment = CACHE_LINE_SIZE)
 {
   void *p = 0;
-  if (posix_memalign(&p, CACHE_LINE_SIZE, size) != 0){
+  if (posix_memalign(&p, alignment, size) != 0){
     perror("posix_memalign");
     throw std::runtime_error("memory");
   }
+  memset(p, 0, size);          // zero the memory
   return p;
 }
 
+////////////////////////////////////////////////////////////////////////
+
+
 gc_job_manager_impl::gc_job_manager_impl(const gc_jm_options *options)
-  : d_eh_cond(&d_eh_mutex), d_eh_thread(0), d_eh_state(EHS_INIT),
+  : d_spu_args(0),
+    d_eh_cond(&d_eh_mutex), d_eh_thread(0), d_eh_state(EHS_INIT),
     d_shutdown_requested(false),
     d_nclients(0), d_client_thread(0), d_job_busy(0)
 {
@@ -119,6 +124,10 @@
        // ask for more than are available.
        throw std::out_of_range("gang_scheduling: not enough spes available");
       }
+      else {   // FIXME clamp to usable.  problem on PS3 when overcommited
+       fprintf(stderr, "gc_job_manager: clamping nspes to %d\n", nusable_spes);
+       d_options.nspes = nusable_spes;
+      }
     }
   }
 
@@ -134,9 +143,27 @@
     }
   }
 
+  // ----------------------------------------------------------------
+  // initalize the job queue
+  
+  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);
+
+
+  // ----------------------------------------------------------------
   // create the spe contexts
 
-  int spe_flags = SPE_EVENTS_ENABLE | SPE_CFG_SIGNOTIFY1_OR | 
SPE_CFG_SIGNOTIFY2_OR;
+  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());
+
+  int spe_flags = (SPE_EVENTS_ENABLE
+                  | SPE_CFG_SIGNOTIFY1_OR
+                  | SPE_CFG_SIGNOTIFY2_OR);
   
   for (unsigned int i = 0; i < d_options.nspes; i++){
     // FIXME affinity stuff goes here
@@ -145,6 +172,10 @@
       perror("spe_context_create");
       throw std::runtime_error("spe_context_create");
     }
+    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->spu_idx = i;
     d_worker[i].state = WS_INIT;
 
     // FIXME load some code in the SPE
@@ -167,7 +198,6 @@
   // Initialize the array of job descriptors.
   d_jd = (gc_job_desc_t *) aligned_alloc(sizeof(d_jd[0]) * d_options.max_jobs);
   _d_jd_boost = boost::shared_ptr<void>((void *) d_jd, free_deleter());
-  memset(d_jd, 0, sizeof(d_jd[0]) * d_options.max_jobs);
 
 
   // set unique job_id
@@ -179,15 +209,6 @@
     free_job_desc(&d_jd[i]);
 
   // ----------------------------------------------------------------
-  // initalize the job queue
-  
-  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);
-
-
-  // ----------------------------------------------------------------
   // initialize d_client_thread
 
   // FIXME robust deletion
@@ -205,7 +226,6 @@
   size_t byte_size = nlongs * sizeof(unsigned long);
   void *p = aligned_alloc(byte_size);
   _d_all_bitvectors = boost::shared_ptr<void>(p, free_deleter());
-  memset(p, 0, byte_size);
 
   // now point the other pointers into this storage.
   // d_job_busy is the first one, followed by the ones in the client data
@@ -218,16 +238,14 @@
     d_client_thread[i].d_jobs_were_waiting_for = v;
 
   // ----------------------------------------------------------------
-  // create the spe event handler
+  // create the spe event handler & worker (SPE) threads
 
   create_event_handler();
 
-  // FIXME create N worker threads
-  // FIXME confirm everything started OK
-
-  // Done!
 }
 
+////////////////////////////////////////////////////////////////////////
+
 gc_job_manager_impl::~gc_job_manager_impl()
 {
   shutdown();
@@ -296,6 +314,35 @@
   return false;                        // FIXME
 }
 
+bool
+gc_job_manager_impl::send_all_spes(uint32_t msg)
+{
+  bool ok = true;
+
+  for (unsigned int i = 0; i < d_options.nspes; i++)
+    ok &= send_spe(i, msg);
+
+  return ok;
+}
+
+bool
+gc_job_manager_impl::send_spe(unsigned int spe, uint32_t msg)
+{
+  if (spe >= d_options.nspes)
+    return false;
+
+  int r = spe_in_mbox_write(d_worker[spe].spe_ctx, &msg, 1,
+                           SPE_MBOX_ALL_BLOCKING);
+  if (r < 0){
+    perror("spe_in_mbox_write");
+    return false;
+  }
+
+  return r == 1;
+}
+
+////////////////////////////////////////////////////////////////////////
+
 static void
 pthread_create_failure_msg(int r, const char *which)
 {
@@ -339,8 +386,10 @@
 }
 
 
+////////////////////////////////////////////////////////////////////////
 
-//extern "C" 
+static void *start_worker(void *arg);
+
 static void *
 start_event_handler(void *arg)
 {
@@ -349,24 +398,6 @@
   return 0;
 }
 
-//extern "C" 
-static void *
-worker_loop(void *arg)
-{
-  // FIXME need to pass additional info.
-  
-  worker_ctx *w = (worker_ctx *) arg;
-
-  w->state = WS_RUNNING;
-
-  printf("worker running\n");
-
-  // FIXME spe_context_run
-
-  w->state = WS_DEAD;
-  return 0;
-}
-
 void
 gc_job_manager_impl::create_event_handler()
 {
@@ -403,7 +434,7 @@
   for (unsigned int i = 0; ok && i < d_options.nspes; i++){
     char name[256];
     snprintf(name, sizeof(name), "worker[%d]", i);
-    ok &= start_thread(&d_worker[i].thread, worker_loop,
+    ok &= start_thread(&d_worker[i].thread, start_worker,
                       &d_worker[i], name);
   }
 
@@ -411,7 +442,7 @@
     //
     // FIXME Clean up the mess.  Need to terminate event handler and all 
workers.
     //
-    // this should cause the workers to exit, unless they're really broken
+    // this should cause the workers to exit, unless they're seriously broken
     send_all_spes(MK_MBOX_MSG(OP_EXIT, 0));
 
     shutdown();
@@ -421,33 +452,8 @@
   }
 }
 
-bool
-gc_job_manager_impl::send_all_spes(uint32_t msg)
-{
-  bool ok = true;
+////////////////////////////////////////////////////////////////////////
 
-  for (unsigned int i = 0; i < d_options.nspes; i++)
-    ok &= send_spe(i, msg);
-
-  return ok;
-}
-
-bool
-gc_job_manager_impl::send_spe(unsigned int spe, uint32_t msg)
-{
-  if (spe >= d_options.nspes)
-    return false;
-
-  int r = spe_in_mbox_write(d_worker[spe].spe_ctx, &msg, 1,
-                           SPE_MBOX_ALL_BLOCKING);
-  if (r < 0){
-    perror("spe_in_mbox_write");
-    return false;
-  }
-
-  return r == 1;
-}
-
 void
 gc_job_manager_impl::set_eh_state(evt_handler_state s)
 {
@@ -623,11 +629,68 @@
   }
 }
 
-// ------------------------------------------------------------------------
+////////////////////////////////////////////////////////////////////////
+// This is the top of the SPE worker threads
 
+static void *
+start_worker(void *arg)
+{
+  worker_ctx *w = (worker_ctx *) arg;
+  spe_stop_info_t      si;
+
+  w->state = WS_RUNNING;
+  printf("worker[%d]: WS_RUNNING\n", w->spe_idx);
+
+  struct uregs {
+    gc_eaddr_t r3;          // Always pass pointers as 64-bit addresses
+    uint64_t   _pad_r3;     // in preferred slot.  Pad bottom half of reg.
+    gc_eaddr_t r4;
+    uint64_t   _pad_r4;
+    gc_eaddr_t r5;
+    uint64_t   _pad_r5;
+  } args;
+
+  assert(sizeof(args) == 48);
+
+  memset(&args, 0, sizeof(args));
+  args.r3 = ptr_to_ea(w->spe_ctx);
+  args.r4 = ptr_to_ea(w->spu_args);
+
+  unsigned int entry = SPE_DEFAULT_ENTRY;
+  int r = spe_context_run(w->spe_ctx,  &entry,
+                         SPE_RUN_USER_REGS, &args, 0, &si);
+
+  if (r < 0){                  // error
+    char buf[64];
+    snprintf(buf, sizeof(buf), "worker[%d]: spe_context_run", w->spe_idx);
+    perror(buf);
+  }
+  else if (r == 0){
+    // spe program called exit.
+    printf("worker[%d]: SPE_EXIT w/ exit_code = %d\n",
+          w->spe_idx, si.result.spe_exit_code);
+  }
+  else {
+    // called stop_and_signal
+    //
+    // I'm not sure we'll ever get here.  I think the event
+    // handler will catch this...
+    printf("worker[%d]: SPE_STOP_AND_SIGNAL w/ spe_signal_code = 0x%x\n",
+          w->spe_idx, si.result.spe_signal_code);
+  }
+
+  // in any event, we're committing suicide now ;)
+  printf("worker[%d]: WS_DEAD\n", w->spe_idx);
+
+  w->state = WS_DEAD;
+  return 0;
+}
+
+////////////////////////////////////////////////////////////////////////
+
+
 worker_ctx::~worker_ctx()
 {
-  // fprintf(stderr, "\n~worker_ctx\n");
   if (spe_ctx){
     int r = spe_context_destroy(spe_ctx);
     if (r != 0){
@@ -637,4 +700,3 @@
   }
   state = WS_FREE;
 }
-

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 
2007-11-25 02:48:29 UTC (rev 7020)
+++ gnuradio/branches/developers/eb/gcell/src/lib/gc_job_manager_impl.h 
2007-11-25 08:24:13 UTC (rev 7021)
@@ -26,6 +26,7 @@
 #include "gc_client_thread_info.h"
 #include "gc_jd_stack.h"
 #include "gc_jd_queue.h"
+#include "gc_spu_args.h"
 #include <libspe2.h>
 #include <vector>
 #include <boost/shared_ptr.hpp>
@@ -42,10 +43,14 @@
 
 struct worker_ctx {
   volatile worker_state        state;
+  unsigned int         spe_idx;        // [0, nspes-1]
   spe_context_ptr_t    spe_ctx;
   pthread_t            thread;
+  gc_spu_args_t                *spu_args;      // pointer to 16-byte aligned 
struct
 
-  worker_ctx() : state(WS_FREE), spe_ctx(0), thread(0) {}
+  worker_ctx()
+    : state(WS_FREE), spe_idx(0), spe_ctx(0),
+      thread(0), spu_args(0) {}
   ~worker_ctx();
 };
 
@@ -84,6 +89,8 @@
   spe_gang_context_sptr  d_gang;               // boost::shared_ptr
 
   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
 
   // used to coordinate communication w/ the event handling thread
   omni_mutex            d_eh_mutex;