git.gag.com Git - debian/gnuradio/blob - gcell/src/lib/runtime/gc_job_manager_impl.h

   1 /* -*- c++ -*- */
   2 /*
   3  * Copyright 2007 Free Software Foundation, Inc.
   4  *
   5  * This file is part of GNU Radio
   6  *
   7  * GNU Radio is free software; you can redistribute it and/or modify
   8  * it under the terms of the GNU General Public License as published by
   9  * the Free Software Foundation; either version 3, or (at your option)
  10  * any later version.
  11  *
  12  * GNU Radio is distributed in the hope that it will be useful,
  13  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15  * GNU General Public License for more details.
  16  *
  17  * You should have received a copy of the GNU General Public License along
  18  * with this program; if not, write to the Free Software Foundation, Inc.,
  19  * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  20  */
  21
  22 #ifndef INCLUDED_GC_JOB_MANAGER_IMPL_H
  23 #define INCLUDED_GC_JOB_MANAGER_IMPL_H
  24
  25 #include "gc_job_manager.h"
  26 #include "gc_client_thread_info.h"
  27 #include "gc_jd_stack.h"
  28 #include "gc_jd_queue.h"
  29 #include "gc_spu_args.h"
  30 #include <libspe2.h>
  31 #include <vector>
  32 #include <boost/shared_ptr.hpp>
  33 #include <boost/scoped_array.hpp>
  34
  35 typedef boost::shared_ptr<spe_gang_context> spe_gang_context_sptr;
  36 typedef boost::shared_ptr<spe_program_handle_t> spe_program_handle_sptr;
  37 typedef boost::scoped_array<gc_client_thread_info> gc_client_thread_info_sa;
  38
  39
  40 enum worker_state {
  41   WS_FREE,      // not in use
  42   WS_INIT,      // allocated and being initialized
  43   WS_RUNNING,   // the thread is running
  44   WS_DEAD,      // the thread is dead
  45 };
  46
  47 struct worker_ctx {
  48   volatile worker_state state;
  49   unsigned int          spe_idx;        // [0, nspes-1]
  50   spe_context_ptr_t     spe_ctx;
  51   pthread_t             thread;
  52   gc_spu_args_t         *spu_args;      // pointer to 16-byte aligned struct
  53
  54   worker_ctx()
  55     : state(WS_FREE), spe_idx(0), spe_ctx(0),
  56       thread(0), spu_args(0) {}
  57   ~worker_ctx();
  58 };
  59
  60 enum evt_handler_state {
  61   EHS_INIT,             // being initialized
  62   EHS_RUNNING,          // thread is running
  63   EHS_SHUTTING_DOWN,    // in process of shutting down everything
  64   EHS_WAITING_FOR_WORKERS_TO_DIE,
  65   EHS_DEAD,             // thread is dead
  66 };
  67
  68 struct spe_event_handler {
  69   spe_event_handler_ptr_t       ptr;
  70
  71   spe_event_handler() : ptr(0) {}
  72   ~spe_event_handler(){
  73     if (ptr){
  74       if (spe_event_handler_destroy(ptr) != 0){
  75         perror("spe_event_handler_destroy");
  76       }
  77     }
  78   }
  79 };
  80
  81
  82 /*!
  83  * \brief Concrete class that manages SPE jobs.
  84  *
  85  * This class contains all the implementation details.
  86  */
  87 class gc_job_manager_impl : public gc_job_manager
  88 {
  89   enum { MAX_SPES =  16 };
  90
  91   int                     d_debug;
  92   gc_jm_options           d_options;
  93   spe_program_handle_sptr d_spe_image;
  94   spe_gang_context_sptr   d_gang;               // boost::shared_ptr
  95
  96   worker_ctx             d_worker[MAX_SPES];    // SPE ctx, thread, etc
  97   gc_spu_args_t         *d_spu_args;            // 16-byte aligned structs
  98   boost::shared_ptr<void> _d_spu_args_boost;    // hack for automatic storage mgmt
  99
 100   gc_comp_info_t        *d_comp_info;           // 128-byte aligned structs
 101   boost::shared_ptr<void> _d_comp_info_boost;   // hack for automatic storage mgmt
 102
 103   // used to coordinate communication w/ the event handling thread
 104   omni_mutex             d_eh_mutex;
 105   omni_condition         d_eh_cond;
 106   pthread_t              d_eh_thread;           // the event handler thread
 107   volatile evt_handler_state    d_eh_state;
 108   volatile bool                 d_shutdown_requested;
 109   spe_event_handler      d_spe_event_handler;
 110
 111
 112   // All of the job descriptors are hung off of here.
 113   // We allocate them all in a single cache aligned chunk.
 114   gc_job_desc_t         *d_jd;                  // [options.max_jobs]
 115   boost::shared_ptr<void> _d_jd_boost;          // hack for automatic storage mgmt
 116
 117   gc_client_thread_info_sa d_client_thread;     // [options.max_client_threads]
 118
 119   // We use bitvectors to represent the completing state of a job.  Each
 120   // bitvector is d_bvlen longs in length.
 121   int                    d_bvlen;               // bit vector length in longs
 122
 123   // This contains the storage for all the bitvectors used by the job
 124   // manager.  There's 1 for each client thread, in the d_jobs_done
 125   // field.  We allocate them all in a single cache aligned chunk.
 126   boost::shared_ptr<void> _d_all_bitvectors;    // hack for automatic storage mgmt
 127
 128   // Lock free stack where we keep track of the free job descriptors.
 129   gc_jd_stack_t         *d_free_list;           // stack of free job descriptors
 130   boost::shared_ptr<void> _d_free_list_boost;   // hack for automatic storage mgmt
 131
 132   // The PPE inserts jobs here; SPEs pull jobs from here.
 133   gc_jd_queue_t         *d_queue;               // job queue
 134   boost::shared_ptr<void> _d_queue_boost;       // hack for automatic storage mgmt
 135
 136   int                    d_ea_args_maxsize;
 137
 138   struct gc_proc_def    *d_proc_def;            // the SPE procedure table
 139   uint32_t               d_proc_def_ls_addr;    // the LS address of the table
 140   int                    d_nproc_defs;          // number of proc_defs in table
 141
 142   gc_client_thread_info *alloc_cti();
 143   void free_cti(gc_client_thread_info *cti);
 144
 145   void create_event_handler();
 146   void set_eh_state(evt_handler_state s);
 147   void set_ea_args_maxsize(int maxsize);
 148
 149   void notify_clients_jobs_are_done(unsigned int spe_num,
 150                                     unsigned int completion_info_idx);
 151
 152 public:
 153   void event_handler_loop();    // really private
 154
 155 private:
 156   bool send_all_spes(uint32_t msg);
 157   bool send_spe(unsigned int spe, uint32_t msg);
 158   void print_event(spe_event_unit_t *evt);
 159   void handle_event(spe_event_unit_t *evt);
 160
 161   // bitvector ops
 162   void bv_zero(unsigned long *bv);
 163   void bv_clr(unsigned long *bv, unsigned int bitno);
 164   void bv_set(unsigned long *bv, unsigned int bitno);
 165   bool bv_isset(unsigned long *bv, unsigned int bitno);
 166   bool bv_isclr(unsigned long *bv, unsigned int bitno);
 167
 168   void setup_logfiles();
 169   void sync_logfiles();
 170   void unmap_logfiles();
 171
 172   friend gc_job_manager *gc_make_job_manager(const gc_jm_options *options);
 173
 174   gc_job_manager_impl(const gc_jm_options *options = 0);
 175
 176 public:
 177   virtual ~gc_job_manager_impl();
 178
 179   /*!
 180    * Stop accepting new jobs.  Wait for existing jobs to complete.
 181    * Return all managed SPE's to the system.
 182    */
 183   virtual bool shutdown();
 184
 185   /*!
 186    * \brief Return number of SPE's currently allocated to job manager.
 187    */
 188   virtual int nspes() const;
 189
 190   /*!
 191    * \brief Return a pointer to a properly aligned job descriptor,
 192    * or zero if none are available.
 193    */
 194   virtual gc_job_desc *alloc_job_desc();
 195
 196   /*
 197    *! Return a job descriptor previously allocated with alloc_job_desc()
 198    *
 199    * \param[in] jd pointer to job descriptor to free.
 200    */
 201   virtual void free_job_desc(gc_job_desc *jd);
 202
 203   /*!
 204    * \brief Submit a job for asynchronous processing on an SPE.
 205    *
 206    * \param[in] jd pointer to job description
 207    *
 208    * The caller must not read or write the job description
 209    * or any of the memory associated with any indirect arguments
 210    * until after calling wait_job.
 211    *
 212    * \returns true iff the job was successfully enqueued.
 213    * If submit_job returns false, check jd->status for additional info.
 214    */
 215   virtual bool submit_job(gc_job_desc *jd);
 216
 217   /*!
 218    * \brief Wait for job to complete.
 219    *
 220    * A thread may only wait for jobs which it submitted.
 221    *
 222    * \returns true if sucessful, else false.
 223    */
 224   virtual bool
 225   wait_job(gc_job_desc *jd);
 226
 227   /*!
 228    * \brief wait for 1 or more jobs to complete.
 229    *
 230    * \param[input] njobs is the length of arrays \p jd and \p done.
 231    * \param[input] jd are the jobs that are to be waited for.
 232    * \param[output] done indicates whether the corresponding job is complete.
 233    * \param[input] mode indicates whether to wait for ALL or ANY of the jobs
 234    *   in \p jd to complete.
 235    *
 236    * A thread may only wait for jobs which it submitted.
 237    *
 238    * \returns number of jobs completed, or -1 if error.
 239    */
 240   virtual int
 241   wait_jobs(unsigned int njobs,
 242             gc_job_desc *jd[], bool done[], gc_wait_mode mode);
 243
 244   virtual int ea_args_maxsize();
 245
 246   virtual gc_proc_id_t lookup_proc(const std::string &name);
 247   virtual std::vector<std::string> proc_names();
 248
 249   virtual void set_debug(int debug);
 250   virtual int debug();
 251 };
 252
 253 #endif /* INCLUDED_GC_JOB_MANAGER_IMPL_H */