path: root/gcell/lib/runtime/gc_job_manager_impl.h

/* -*- c++ -*- */
/*
 * Copyright 2007,2008,2009,2010 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_JOB_MANAGER_IMPL_H
#define INCLUDED_GC_JOB_MANAGER_IMPL_H

#include <gcell/gc_job_manager.h>
#include <gcell/gc_jd_stack.h>
#include <gcell/gc_jd_queue.h>
#include <gcell/gc_spu_args.h>
#include "gc_client_thread_info.h"
#include <libspe2.h>
#include <vector>
#include <boost/scoped_array.hpp>

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;


enum worker_state {
  WS_FREE,	// not in use
  WS_INIT,	// allocated and being initialized
  WS_RUNNING,	// the thread is running
  WS_DEAD,	// the thread is dead
};

struct worker_ctx {
  volatile worker_state	  state;
  unsigned int		  spe_idx;  	// [0, nspes-1]
  spe_context_ptr_t	  spe_ctx;
  spe_spu_control_area_t *spe_ctrl;
  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), spe_ctrl(0),
      thread(0), spu_args(0) {}
  ~worker_ctx();
};

enum evt_handler_state {
  EHS_INIT,		// being initialized
  EHS_RUNNING,		// thread is running
  EHS_SHUTTING_DOWN,	// in process of shutting down everything
  EHS_WAITING_FOR_WORKERS_TO_DIE,
  EHS_DEAD,		// thread is dead
};

enum job_completer_state {
  JCS_INIT,		// being initialized
  JCS_RUNNING,		// thread is running
  JCS_DEAD,		// thread is dead
};

struct spe_event_handler {
  spe_event_handler_ptr_t	ptr;

  spe_event_handler() : ptr(0) {}
  ~spe_event_handler(){
    if (ptr){
      if (spe_event_handler_destroy(ptr) != 0){
	perror("spe_event_handler_destroy");
      }
    }
  }
};


/*!
 * \brief Concrete class that manages SPE jobs.
 *
 * This class contains all the implementation details.
 */
class gc_job_manager_impl : public gc_job_manager
{
  enum { MAX_SPES =  16 };

  int			  d_debug;
  gc_jm_options		  d_options;
  spe_program_handle_sptr d_spe_image;
  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

  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
  boost::mutex		 d_eh_mutex;
  boost::condition_variable d_eh_cond;
  pthread_t		 d_eh_thread;		// the event handler thread
  volatile evt_handler_state	d_eh_state;
  volatile bool		 	d_shutdown_requested;
  spe_event_handler	 d_spe_event_handler;
  
  // used to coordinate communication w/ the job completer thread
  boost::mutex		 d_jc_mutex;
  boost::condition_variable d_jc_cond;
  pthread_t		 d_jc_thread;		// the job completion thread
  volatile job_completer_state	d_jc_state;
  int		      	 d_jc_njobs_active;	// # of jobs submitted but not yet reaped

  // round robin notification of spes
  int			 d_ntell;		// # of spes to tell
  unsigned int		 d_tell_start;		// which one to start with

  // 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

  gc_client_thread_info_sa d_client_thread;	// [options.max_client_threads]

  // We use bitvectors to represent the completing state of a job.  Each
  // bitvector is d_bvlen longs in length.
  int			 d_bvlen;		// bit vector length in longs

  // 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

  // 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

  // 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

  int			 d_ea_args_maxsize;

  struct gc_proc_def	*d_proc_def;		// the SPE procedure table
  uint32_t		 d_proc_def_ls_addr;	// the LS address of the table
  int			 d_nproc_defs;		// number of proc_defs in table

  gc_client_thread_info *alloc_cti();
  void free_cti(gc_client_thread_info *cti);

  void create_event_handler();
  void set_eh_state(evt_handler_state s);
  void set_ea_args_maxsize(int maxsize);

  void notify_clients_jobs_are_done(unsigned int spe_num,
				    unsigned int completion_info_idx);

public:
  void event_handler_loop();	// really private
  void job_completer_loop();	// really private

private:
  bool send_all_spes(uint32_t msg);
  bool send_spe(unsigned int spe, uint32_t msg);
  void print_event(spe_event_unit_t *evt);
  void handle_event(spe_event_unit_t *evt);
  bool incr_njobs_active();
  void decr_njobs_active(int n);
  void tell_spes_to_check_queue();
  void poll_for_job_completion();

  // bitvector ops
  void bv_zero(unsigned long *bv);
  void bv_clr(unsigned long *bv, unsigned int bitno);
  void bv_set(unsigned long *bv, unsigned int bitno);
  bool bv_isset(unsigned long *bv, unsigned int bitno);
  bool bv_isclr(unsigned long *bv, unsigned int bitno);

  void setup_logfiles();
  void sync_logfiles();
  void unmap_logfiles();

  friend gc_job_manager_sptr gc_make_job_manager(const gc_jm_options *options);
  
  gc_job_manager_impl(const gc_jm_options *options = 0);

public:
  virtual ~gc_job_manager_impl();

  /*!
   * Stop accepting new jobs.  Wait for existing jobs to complete.
   * Return all managed SPE's to the system.
   */
  virtual bool shutdown();

  /*!
   * \brief Return number of SPE's currently allocated to job manager.
   */
  virtual int nspes() const;

  /*!
   * \brief Return a pointer to a properly aligned job descriptor,
   * or zero if none are available.
   */
  virtual gc_job_desc *alloc_job_desc();

  /*
   *! Return a job descriptor previously allocated with alloc_job_desc()
   *
   * \param[in] jd pointer to job descriptor to free.
   */
  virtual void free_job_desc(gc_job_desc *jd);

  /*!
   * \brief Submit a job for asynchronous processing on an SPE.
   *
   * \param[in] jd pointer to job description
   *
   * The caller must not read or write the job description
   * or any of the memory associated with any indirect arguments
   * until after calling wait_job.
   *
   * \returns true iff the job was successfully enqueued.
   * If submit_job returns false, check jd->status for additional info.
   */
  virtual bool submit_job(gc_job_desc *jd);

  /*!
   * \brief Wait for job to complete.
   *
   * A thread may only wait for jobs which it submitted.
   *
   * \returns true if sucessful, else false.
   */
  virtual bool 
  wait_job(gc_job_desc *jd);

  /*!
   * \brief wait for 1 or more jobs to complete.
   *
   * \param[in] njobs is the length of arrays \p jd and \p done.
   * \param[in] jd are the jobs that are to be waited for.
   * \param[out] done indicates whether the corresponding job is complete.
   * \param[in] mode indicates whether to wait for ALL or ANY of the jobs
   *   in \p jd to complete.
   *
   * A thread may only wait for jobs which it submitted.
   *
   * \returns number of jobs completed, or -1 if error.
   */
  virtual int
  wait_jobs(unsigned int njobs,
	    gc_job_desc *jd[], bool done[], gc_wait_mode mode);

  virtual int ea_args_maxsize();

  virtual gc_proc_id_t lookup_proc(const std::string &name);
  virtual std::vector<std::string> proc_names();

  virtual void set_debug(int debug);
  virtual int debug();
};

#endif /* INCLUDED_GC_JOB_MANAGER_IMPL_H */