Simultaneous.h

//-*-C++-*-
/***************************************************************************
 *
 *   Copyright (C) 2005 by Haydon Knight
 *   Licensed under the Academic Free License version 2.1
 *
 ***************************************************************************/
 
#ifndef __baseband_dsp_Simultaneous_h_
#define __baseband_dsp_Simultaneous_h_
 
#include "dsp/BasicBuffer.h"
#include "dsp/Operation.h"
#include "dsp/Buffer.h"
#include "dsp/TimeSeries.h"
 
#include <queue>
 
namespace dsp {
 
  class Simultaneous : public Operation {
 
  public:
 
    TimeSeries* last_on;
 
    Simultaneous();
 
    virtual ~Simultaneous();
 
    Operation* get_op(){ return op; }
 
    bool has_op(){ return op; }
 
    void set_op(Operation* _op){ op = _op; }
    
    template<class T>
    T* get();
 
    void stop_thread();
    
  protected:
 
    virtual void operation ();
 
    bool new_data();
 
    bool copy_input_into_ringbuffer();
 
    bool copy_ringbuffer_into_output();
 
    void start_thread();
 
    void destroy_thread();
 
    static void* op_loop(Simultaneous* thiz);
 
    void wait_to_synch();
    
    bool can_run();
    
    void checks();
    
    void run_op();
    
    bool op_uses_an_input();
    
    bool op_uses_an_output();
    
    void setup_buffers();
    
    template<class T>
    bool newify_buffers(vector<Reference::To<BasicBuffer> >& buffers);
 
    bool newify_input_buffers();
 
    bool newify_output_buffers();
 
    bool newify_those_buffers(string typestring,
                              vector<Reference::To<BasicBuffer> >& buffers);
 
    Reference::To<BasicBuffer> setup_op_output();
    Reference::To<BasicBuffer> setup_op_input();
 
    string get_input_typestring();
 
    string get_output_typestring();
 
    void initialise_op_input();
    void initialise_op_output();
 
    Reference::To<BasicBuffer> new_buffer(string typestring);
 
    Reference::To<BasicBuffer> pop_free_input_buffer();
    Reference::To<BasicBuffer> pop_full_input_buffer();
    Reference::To<BasicBuffer> pop_free_output_buffer();
    Reference::To<BasicBuffer> pop_full_output_buffer();
    
    void wait_for_free_input_buffer();
    void wait_for_full_input_buffer();
    void wait_for_free_output_buffer();
    void wait_for_full_output_buffer();
 
    void init();  
 
    Reference::To<Operation> op;
 
    bool thread_running;
 
    vector<Reference::To<BasicBuffer> > input_buffers;
    vector<Reference::To<BasicBuffer> > output_buffers;
 
    std::queue<Reference::To<BasicBuffer> > free_input_buffers;
    std::queue<Reference::To<BasicBuffer> > full_input_buffers;
    std::queue<Reference::To<BasicBuffer> > free_output_buffers;
    std::queue<Reference::To<BasicBuffer> > full_output_buffers;
 
    Reference::To<BasicBuffer> op_input;
    Reference::To<BasicBuffer> op_output;
 
    pthread_mutex_t free_input_buffers_mutex;
    pthread_mutex_t full_input_buffers_mutex;
    pthread_mutex_t free_output_buffers_mutex;
    pthread_mutex_t full_output_buffers_mutex;
 
    pthread_mutex_t free_input_mutex;
    pthread_mutex_t full_input_mutex;
    pthread_mutex_t free_output_mutex;
    pthread_mutex_t full_output_mutex;
 
    pthread_cond_t free_input_cond;
    pthread_cond_t full_input_cond;
    pthread_cond_t free_output_cond;
    pthread_cond_t full_output_cond;
 
    pthread_mutex_t synch_mutex;
    pthread_cond_t synch_cond;
 
    pthread_mutex_t running_mutex;
    pthread_cond_t running_cond;
 
    float wait_time;
 
    bool use_threads;
 
    pthread_t* op_thread;
    
    bool keep_running;
 
    bool synch;
  };
 
}
 
template<class T>
T* dsp::Simultaneous::get(){
  return dynamic_cast<T*>( op.get() );
}
 
template<class T>
bool dsp::Simultaneous::newify_buffers(vector<Reference::To<BasicBuffer> >& buffers){
  for( unsigned i=0; i<buffers.size(); i++)
    buffers[i] = new Buffer<T>;
 
  return true;
}
 
#endif