SingleThread.h

//-*-C++-*-
/***************************************************************************
 *
 *   Copyright (C) 2011 by Willem van Straten
 *   Licensed under the Academic Free License version 2.1
 *
 ***************************************************************************/
 
// dspsr/Signal/General/dsp/SingleThread.h
 
#ifndef __dspsr_SingleThread_h
#define __dspsr_SingleThread_h
 
#include "dsp/Pipeline.h"
#include "CommandLine.h"
#include "Functor.h"
#include "TextEditor.h"
#include "MJD.h"
 
class ThreadContext;
 
namespace dsp {
 
  class Source;
  class TimeSeries;
  class Operation;
  class Observation;
  class Scratch;
  class Memory;
 
  class SingleThread : public Pipeline
  {
 
  public:
 
    friend class MultiThread;
 
    class Config;
 
    SingleThread ();
 
    ~SingleThread ();
 
    void set_configuration (Config*);
 
    void set_source (Source*);
 
    Source* get_source ();
 
    void initialize ();
 
    void construct ();
 
    void prepare ();
 
    void seek_epoch (const MJD&);
 
    void run ();
 
    virtual void share (SingleThread*);
 
    virtual void combine (const SingleThread*);
 
    void finish ();
 
    uint64_t get_minimum_samples () const;
 
    std::ostream cerr;
 
    void take_ostream (std::ostream* newlog);
 
    unsigned thread_id;
    void set_affinity (int core);
 
    std::vector< Reference::To<Operation> > const get_operations () { return operations; };
 
    /*
      returns the estimated number of bytes required to store data in the pipeline
    */
    uint64_t set_block_size (uint64_t minimum_samples, uint64_t input_overlap = 0);
 
  protected:
 
    virtual void end_of_data ();
 
    virtual void prepare (Source*);
 
    void configure_overlap (Source*, Memory*);
 
    std::ostream* log;
 
    enum State
    {
      Fail,        
      Idle,        
      Construct,   
      Constructed, 
      Prepare,     
      Prepared,    
      Run,         
      Done,        
      Joined       
    };
 
    State state;
 
    Error error;
 
    ThreadContext* state_change;
 
    ThreadContext* source_context;
 
    SingleThread* colleague;
 
    Reference::To<Source> source;
 
    Reference::To<TimeSeries> source_output;
 
    Reference::To<Config> config;
 
    TimeSeries* new_time_series ();
    TimeSeries* new_TimeSeries () { return new_time_series(); }
 
    std::vector< Reference::To<Operation> > operations;
 
    void insert_dump_point (const std::string& transformation_name);
 
    void insert_mask (const std::string& transformation_name);
 
    Reference::To<Scratch> scratch;
 
    uint64_t minimum_samples;
 
    Reference::To<Memory> device_memory;
    void* gpu_stream;
    int gpu_device;
 
    const std::string& app();
  };
 
  class Input;
 
  class SingleThread::Config : public Reference::Able
  {
  public:
 
    Config ();
 
    virtual void add_options (CommandLine::Menu&);
 
    Source* open (int argc, char** argv);
 
    // set block size to this factor times the minimum possible
    void set_times_minimum_ndat (unsigned);
    unsigned get_times_minimum_ndat () const { return times_minimum_ndat; }
 
    // set block_size to result in approximately this much RAM usage
    void set_maximum_RAM (uint64_t);
    uint64_t get_maximum_RAM () const { return maximum_RAM; }
 
    // set block_size to result in at least this much RAM usage
    void set_minimum_RAM (uint64_t);
    uint64_t get_minimum_RAM () const { return minimum_RAM; }
 
    virtual void prepare (Source*);
 
    std::string application_name;
 
    Functor< void(Source*) > source_prepare;
 
    // Input files represent a single continuous observation
    bool force_contiguity;
 
    // Command line values are header params, not file names
    bool command_line_header;
 
 
    MJD seek_epoch;
 
    // number of seconds to seek into data
    double seek_seconds;
 
    // number of seconds to process from data
    double total_seconds;
 
    bool list_attributes;
 
    TextEditor<Observation> editor;
 
    bool report_vitals;
 
    bool report_done;
 
    bool run_repeatedly;
 
    void set_cuda_device (std::string);
    unsigned get_cuda_ndevice () const { return cuda_device.size(); }
 
    void set_nthread (unsigned);
 
    unsigned get_total_nthread () const;
 
    void set_affinity (std::string);
 
    void set_fft_library (std::string);
 
    bool input_buffering;
 
    bool weighted_time_series;
 
    std::vector<std::string> dump_before;
 
    std::vector<std::string> mask_before;
 
    unsigned get_nbuffers () const { return buffers; }
 
    virtual void set_quiet ();
 
    virtual void set_verbose ();
 
    virtual void set_very_verbose ();
 
  protected:
 
    friend class SingleThread;
 
    bool can_cuda;
 
    std::vector<unsigned> cuda_device;
 
    bool can_thread;
 
    std::vector<unsigned> affinity;
 
    unsigned nthread;
 
    unsigned buffers;
 
    unsigned repeated;
 
    // set block size to this factor times the minimum possible
    unsigned times_minimum_ndat;
 
    // set block size to result in approximately this much RAM usage
    uint64_t maximum_RAM;
 
    // set block size to result in at least this much RAM usage
    uint64_t minimum_RAM;
  };
 
}
 
#endif // !defined(__SingleThread_h)