include/N3AnaBase/utils/thread_mgr.h

/*
  Mutltithread Manager
  06/2010, Hugon Christophe, CENBG

  thread_mgr is a manager to make a faster calculation from an ana_event file to an ana event file.
  To use it you need to init a 'routine' like :

  #include <N3AnaBase/measurements/tof_eMgN_measurement.h>

  void* a_thread_example(void* data_)
  {

  //getting the data from the pointer (indispensable)
  evt_coll the_data = (evt_coll) data_;

  //initializing the tof_calculator (for instance)
  nemo3::tof_eMgN_measurement the_tof ;
  the_tof.set_debug(true);
  the_tof.set_ECorr(true);

  for (int evt_it = 0 ; evt_it < the_data->size () ; evt_it++)
  {
  the_tof.do_measurement(*(the_data->at(evt_it)));
  }

  return NULL;
  }

  then add it in the thread_mgr object in the place of 'routine_', with the files address and the number of thread you wish.
  then, launch it with Process().
  If you need more argument, you have to use global variables before the thread_mgr's #include.
*/

#ifndef __N3AnaBase__thread_mgr
#define __N3AnaBase__thread_mgr 1

#define app_name "thread manager: "

#include <ctime>
#include <pthread.h>
#include <string>
#include <list>
#include <vector>

using namespace std;

// Utilities:
#include <datatools/utils/utils.h>
#include <datatools/utils/ioutils.h>
#include <NemoTools/MyProperty.h>
#include <datatools/utils/properties.h>
#include <N3AnaBase/utils/convert_ana.h>

// N3AnaBase data model:
#include <N3AnaBase/event/ana_event.h>
#include <N3AnaBase/event/io.h>
#include <N3IORoot/N3AnaDataReader.h>

// data model:
#include <N3Event/N3RawEvent.h>
#include <N3Event/N3StdCalibEvent.h>
#include <N3Event/N3StdTrackedEvent.h>
#include <N3Arc4se/ConstActivityModel.hh>

// I/O:
#include <N3IORoot/N3RawDataReader.h>
#include <N3IORoot/N3AnaDataWriter.h>

// apply real running conditions:
#include <N3Arc4se/RunType.hh>
#include <N3Arc4se/Arc4seMgr.hh>

// calibration:
#include <N3Calib/N3StdCalibMgr.h>

// database:
#include <N3DbMgr.h>

// tracking:
#include <N3TrackLAL/N3LALTracker.h>

#include <N3AnaBase/event/event_utils.h>
#include <N3AnaBase/event/sim_event_utils.h>


// cuts' factory:
#include <N3AnaBase/cuts/cut_factory.h>

using namespace datatools::utils;


namespace nemo3 {


  pthread_mutex_t calib_mutex = PTHREAD_MUTEX_INITIALIZER;//calibration mutex
  pthread_mutex_t simu_mutex = PTHREAD_MUTEX_INITIALIZER;//Process raw mutex
  pthread_mutex_t select_mutex = PTHREAD_MUTEX_INITIALIZER;//selecte settled evt selection
  pthread_mutex_t precal_mutex = PTHREAD_MUTEX_INITIALIZER;
  pthread_mutex_t postcal_mutex = PTHREAD_MUTEX_INITIALIZER;
  pthread_mutex_t convert_mutex = PTHREAD_MUTEX_INITIALIZER;
  pthread_mutex_t counter_mutex = PTHREAD_MUTEX_INITIALIZER;


  typedef list<nemo3::ana_event*>* evt_coll; //it is for the user :p
  typedef list<nemo3::ana_event*>::iterator evt_coll_it; //it is for the user :p



  //------------------------------------------------------------------------------------routines' struct
  struct data_writer {
    list<list<nemo3::ana_event*> > event_buffer; //create a pointer list, to avoid copying
    pthread_mutex_t to_mutex;
    pthread_cond_t to_cond;

  };

  struct data_reader {
    list<pthread_t> sub_thread;
    list<list<nemo3::ana_event*> > event_buffer;
    pthread_mutex_t to_mutex;
    pthread_cond_t to_cond;
    long start;
  };

  struct data_convert {
    list<nemo3::ana_event*> ana_buffer;
    list<nemo3::StdTrackedEvent*> raw_buffer;
    pthread_mutex_t to_mutex;
    pthread_cond_t to_cond;
  };

  struct data_analyze {
  list<nemo3::ana_event*> *ana_buffer;
    pthread_mutex_t to_mutex;
    pthread_cond_t to_cond;
  };

  class thread_mgr {

  public :
    static thread_mgr* GetThread_mgr(string file_in_="", string file_out_="", int NbThread_=-1, void *(*routine_)(void *)=NULL);
    void Close();
    void SetDebug(bool debug_);
    void SetWrite(bool write_);//if you don't want to write
    void SetBufferSize(int buffer_size_);
    void SetPR(bool PR_);
    void SetStat(bool stat_);
    void SetSimulated(string list_file_, double activity_, int run_type_ = nemo3::RunType::BETABETA);
    void SetFalsifiedRunNb (int rn_) {__falsify_run_nb=rn_;}

    void SetSequential(bool sequen_);
    void SetNbEvt (unsigned int nb_);
    bool WaitForIt(pthread_t &thread_, pthread_mutex_t &to_mutex_, pthread_cond_t &to_cond_, long start_);
    void Lock(std::string);
    void Unlock(std::string);

  private :
    bool InitArc4se(long seed_);
    void __Lock(std::string);
    void __Unlock(std::string);
    void __CleanOutLock();
    void __CleanIt (list<list<nemo3::ana_event*> > &event_buffer);

  public :
    void SetAcceptOnlyGoodRunStatus (bool);
    void SetApplyReal (bool);
    void SetDegradPMTs(bool);
    void SetDeleteOffPMTs(bool);
    void SetDeleteHotPMTs(bool);
    void SetDegradGGs(bool);
    void SetDeleteOffGGs(bool);
    void SetDeleteHotGGs(bool);
    void SetRunType(int rt_){__run_type=rt_;}

  public :
    void SetReal();
    void SetFileIn(string file_in_);
    void SetFileOut(string file_out_);
    void SetRoutine(void *(*routine_)(void *));
    bool Process(string file_in_="", string file_out_="", int NbThread_ = -1, void *(*routine_)(void *) = NULL);
    bool SelectEvent(nemo3::ana_event *event_, string cut_file_, string cut_name_);
    void KillEvt(evt_coll the_coll, list<nemo3::ana_event*>::iterator evt_it);

  private:
    static void* writer_routine(void* data_);
    static void* convert_routine(void* data_);
    static void* reader_routine(void* data_);



  private :
    thread_mgr(string file_in_="", string file_out_="", int NbThread_ = 1, void *(*routine_)(void *) = NULL);
    ~thread_mgr();
    thread_mgr& operator= (const thread_mgr&);

    typedef struct {
      nemo3::cut_factory *cut_factory;
      nemocuts::cut_manager *cut_manager;
      nemocuts::ICut *selector;
    } cut_struct;

  private :
    static thread_mgr* Instance;

    static bool __debug;
    static bool __write;
    static bool __PR;//for process raw
    static short __format;
    static int __buffer_size;
    static int __NbThread;
    static int __counter;
    static int  __falsify_run_nb;
    static unsigned int __evt_max;
    static nemo3::Arc4seMgr *__arcase_mgr;
    static nemo3::StdCalibMgr *__calib_mgr;
    static nemo3::LALTracker *__tracker_mgr;
    static nemo3::cut_factory *__cut_factory;
    static nemocuts::cut_manager *__cut_manager;
    static nemocuts::ICut *__selector;
    static std::map<string,cut_struct> __the_cutter;
    static nemo3::io::store_mgr *__writer;
    static void *__reader;
    static vector <nemocuts::ICut> __selectors;

    static bool __simulated;
    bool __select;
    bool __sequen;
    bool __stat;
    bool __DegradPMTs;
    bool __DeleteOffPMTs;
    bool __DeleteHotPMTs;
    bool __DegradGGs;
    bool __DeleteOffGGs;
    bool __DeleteHotGGs;
    bool __AcceptOnlyGood;
    int __run_type;
    double __activity;
    string __file_in;
    string __file_out;
    string __list_file;
    string __cut_file;
    string __cut_name;
    void *(*__routine)(void*);
    struct timeval __now;
    struct timespec __wait;
    long __waiting[3];

    std::map <std::string,pthread_mutex_t> __mutex_lst;
    pthread_mutex_t __map_mutex;

  };
}//end of namespace nemo3


#endif