filter.h

Go to the documentation of this file.

#ifndef FILTER_H
#define FILTER_H

#include <stdio.h>
#include <iostream>
#include <map>
#include <list>
#include <vector>
#include <string>
#include <queue>
#include <unistd.h>
#include <stdlib.h>
#include <sstream>

#ifdef USE_PTHREAD
#include <pthread.h>
#endif

#include "buffer.h"
#include "signalmanager.h"
#include "common.h"

#define __EXIT_CHECK__ if(exit_check()){ break; }

namespace SMP{

class SMPKernel;

const int MAX_LEN = 128;

using namespace std;


class property_item{
public:
    string value;

    bool runtime_ro;

    string type;

    property_item() { /*do nothing*/ }
    property_item(string value, bool runtime_ro, string type="str") 
        : value(value), runtime_ro(runtime_ro), type(type)
    { /*do nothing*/ }
};


    
class SMPFilter
{
    friend class SMPKernel;
public:

protected:

    int id;   

    char name[MAX_LEN];

    //char version[6];
    int major;
    int minor;

    int priority;
    bool realtime_on;

    //---------------------------------------------
    // buffer connectivity information of this filter
    //    
    int num_inport;
    int num_outport;

    int *input_buffer;
    int *output_buffer;
    //---------------------------------------------
    
    
    SMPKernel *kernel;

    //watchdog *dog;    

  
    map<string, property_item> property; 

#ifdef USE_PTHREAD

    //pthread_key_t key;
    
    //inline pthread_key_t get_thread_key();

    //static void *key_destructor(void *arg);

    //sigset_t signal_set;

    void signal_handler_wrapper();
    void signal_handler_wrapper_light();

    pthread_mutex_t  sig_queue_mutex;
    void signal_push(SMPSignal *sig);
    SMPSignal *signal_pop();
    bool signal_empty();


    //int smp_sigwait();
    
    //inline SMPSignal smp_get_signal();

    queue<SMPSignal*> signal_queue;

    pthread_mutex_t sig_mutex; 
    pthread_cond_t sig_cond;

    //---------------------------------------------
    // signal connectivity information of this filter
    //    
    int sig_num_inport;
    int sig_num_outport;


    int *sig_input_buffer;
    int *sig_output_buffer;
    //---------------------------------------------

#endif

#ifdef  USE_PTHREAD
    void setup(int fid, const char *fname, int major_version, int minor_version, 
            int *in_buf, int num_in, int *out_buf, int num_out,
            int *sig_in_buf, int sig_num_in, int *sig_out_buf, int sig_num_out);
#else
    void setup(int fid, const char *fname, int major_version, int minor_version, 
            int *in_buf, int num_in, int *out_buf, int num_out);
#endif


    void set_input_buf(int input_port, int buf_id, port_model model = DATA_PORT);

    void process_wrapper();

    inline void set_rt_status(bool realtime); 


public:

    SMPFilter(void);
    virtual ~SMPFilter(void);


    void get_env(int *argc, char ***argv);


    /*-------------------------------------------------------------------------------------------------
     *  Reading and Writing Buffer Utilities
     *  8 functions
     */
    
    ssize_t smp_read(int port_id, void *buff, size_t nbytes);
    ssize_t smp_write(int port_id, const void *buff, size_t nbytes);
    ssize_t smp_read_nonblock(int port_id, void *buff, size_t nbytes);
    ssize_t smp_write_nonblock(int port_id, const void *buff, size_t nbytes);
    
    ssize_t smp_read_space(int port_id);
    ssize_t smp_write_space(int port_id);

    void smp_yield();

    ssize_t smp_p_read(int port_id, packet *p_buff, size_t npackets);

    ssize_t smp_p_write(int port_id, const packet *p_buff, size_t npackets);
    ssize_t smp_p_read_nonblock(int port_id, packet *p_buff, size_t npackets);
    ssize_t smp_p_write_nonblock(int port_id, const packet *p_buff, size_t npackets);
    
    bool smp_input_flush(int port_id, port_model model = DATA_PORT);
    bool smp_output_flush(int port_id, port_model model = DATA_PORT);
  
#ifdef USE_PTHREAD 
    /*-------------------------------------------------------------------------------------------------
     * Signaling function
     */ 
    int smp_signal(int port_id, SMPSignal *sig);

#endif

    /*-------------------------------------------------------------------------------------------------
     *  Ports connectivity checking utilities
     *  4 functions
     */
    
    inline bool inport_connected(port_model model = DATA_PORT);
    inline bool inport_connected(int port_id, port_model model = DATA_PORT);
    inline bool outport_connected(port_model model = DATA_PORT);
    inline bool outport_connected(int port_id, port_model model = DATA_PORT);

    
    /*-------------------------------------------------------------------------------------------------
     * buffer utilities
     */

    void set_buf_status(port_type port, int port_id, int status);
    int get_buf_status(port_type port, int port_id);
    
    /*-------------------------------------------------------------------------------------------------
     * property utilities
     */
    
    template <class Type>
    inline bool get_property(const char *property_name, Type &data);
    inline string& get_property(const char *property_name);
    inline bool set_property(const char *property_name, const char *data);
    
    /*-------------------------------------------------------------------------------------------------
     * other utilities
     */

    bool exit_check(void);

    void slim_exit(void);
    
    inline bool is_realtime(void); 


    void set_output_buf(int output_port, int buf_id, port_model model = DATA_PORT);
   
    
    /*---------------------
     * about signal handler
     */

    void set_signal_timeout(int usec);

    int get_signal_timeout(void);
    
    SMPSignal *create_signal(SMP_SIGNAL_TYPE type = SMP_USEREVENT);

    /*---------------------
     * watchdog function
     */
    
    void unregister_watchdog();
 
    /*-------------------------------------------------------------------------------------------------
     *  Only these function should be overridden by filter designers
     *  3 functions
     */
    
    virtual int initialize();

    virtual int finalize();

    virtual int pause();

    virtual int resume();

    virtual int process()=0;
   
    virtual int property_handler();

#ifdef USE_PTHREAD

    virtual int signal_handler(SMPSignal *sig);

#endif

    /* =============================================
     * Other untilities
     */

    inline int get_id();

    static void signal_handler_cleanup(void *arg);


    /* ==============================================
     * about system configuration
     */
    
    //void set_realtime(bool enabled);

    void set_watchdog(bool enabled);
    void set_signal_watchdog(bool enabled);
};


/*-------------------------------------------------------------------------------------------------
 *  Ports connectivity checking utilities (inline functions)
 *  4 functions
 */

bool SMPFilter::inport_connected(port_model model)
{
    if(model == DATA_PORT){
        for(int i=0; i < num_inport; i++){
            if(input_buffer[i]==-1) //-1 means "not connected"
                return false;
        }
    }
#ifdef  USE_PTHREAD
    else if(model == SIGNAL_PORT){
        for(int i=0; i < sig_num_inport; i++){
            if(sig_input_buffer[i]==-1) //-1 means "not connected"
                return false;
        }
    }
#endif
    return true;
}

bool SMPFilter::inport_connected(int port_id, port_model model)
{
    if(model == DATA_PORT){
        if(port_id >= num_inport)
            return false;
        return (input_buffer[port_id]==-1)?false:true;
    }
#ifdef  USE_PTHREAD
    else if(model == SIGNAL_PORT){
        if(port_id >= sig_num_inport)
            return false;
        return (sig_input_buffer[port_id]==-1)?false:true;
    }
#endif
    return false;
}

bool SMPFilter::outport_connected(port_model model)
{
    if(model == DATA_PORT){
        for(int i=0; i < num_outport; i++){
            if(output_buffer[i]==-1) //-1 means "not connected"
                return false;
        }
    }
#ifdef  USE_PTHREAD
    else if(model == SIGNAL_PORT){
        for(int i=0; i < sig_num_outport; i++){
            if(sig_output_buffer[i]==-1) //-1 means "not connected"
                return false;
        }
    }
#endif
    return true;

}

bool SMPFilter::outport_connected(int port_id, port_model model)
{
    if(model == DATA_PORT){
        if(port_id >= num_outport)
            return false;
        return (output_buffer[port_id]==-1)?false:true;
    }
#ifdef  USE_PTHREAD
    else if(model == SIGNAL_PORT){
        if(port_id >= sig_num_outport)
            return false;
        return (sig_output_buffer[port_id]==-1)?false:true;
    }
#endif
    return false;
}

template <class Type>
bool SMPFilter::get_property(const char *property_name, Type &data)
{
    std::istringstream iss(property[property_name].value);
    return !(iss >> data).fail();
}

string& SMPFilter::get_property(const char *property_name)
{
    return property[property_name].value;
}

bool SMPFilter::set_property(const char *property_name, const char *data)
{
    if (property.find(property_name) != property.end()) {
        property[property_name].value = string(data);
        return true;
    }
    // property name has error???
    return false;
}

/*
#ifdef USE_PTHREAD
pthread_key_t SMPFilter::get_thread_key()
{
    return key;
}

SMPSignal *SMPFilter::smp_get_signal()
{
    //get SMPSignal object from thread specific data
    return (SMPSignal*)pthread_getspecific(key);
}

#endif
*/

int SMPFilter::get_id()
{
    return id;
}


void SMPFilter::set_rt_status(bool realtime)
{ 
    realtime_on = realtime;
}

bool SMPFilter::is_realtime(void)
{
    return realtime_on;
}


} //end namespace

#endif

---