DataProviderPipeline.qc.dox.h

// -*- mode: c++; indent-tabs-mode: nil -*-
 
// assume local scope for variables, do not use "$" signs
// require type definitions everywhere
// enable all warnings
// allow weak references
 
namespace DataProvider {
const PS_ABORTED = "ABORTED";
 
const PS_RUNNING = "RUNNING";
 
const PS_IDLE = "IDLE";
 
public hashdecl PipelineInfo {
    string name;
 
    *date start_time;
 
    *date stop_time;
 
 
    string status;
 
    int record_count;
 
    int num_queues;
 
 
    bool bulk;
 
    date duration;
 
    float duration_secs;
 
    float recs_per_sec;
};
 
public hashdecl PipelineOptionInfo {
 
    *code debug_log;
 
 
    *code error_log;
 
 
    *code info_log;
 
    /* it is called in the new pipeline queue thread with no arguments
    */
    *code thread_callback;
 
    *string name;
};
 
class PipelineQueue {
 
public:
        int id;
 
        Mutex lck;
 
        Counter cnt;
 
        Condition cond();
 
        Condition flush_cond();
 
        int queue_waiting = 0;
 
        int data_waiting = 0;
 
        int flush_waiting = 0;
 
        list<auto> queue;
 
        int size;
 
        int tid;
 
 
        list<auto> elems();
 
        bool do_flush;
 
        bool data_flushed;
 
        DataProviderPipeline parent;
 
    constructor(DataProviderPipeline parent, Mutex lck, Counter cnt, int id, int size);
 
 
    int getId();
 
 
 
    submit(auto qdata);
 
 
    run(Counter run_cnt);
 
 
 
    waitDone();
 
 
protected:
     runIntern();
public:
 
 
protected:
     flushIntern();
public:
 
};
 
 
class DataProviderPipeline {
 
public:
    
 
protected:
        string name;
 
        hash<string, PipelineQueue> pmap;
 
 
        bool do_bulk = True;
 
        bool locked = False;
 
        Sequence seq(1);
 
        list<hash<ExceptionInfo>> error_list;
 
        date start_time;
 
        date stop_time;
 
        int record_count = 0;
 
        *code info_log;
 
        *code error_log;
 
        *code debug_log;
 
        /* it is called in the new pipeline queue thread with no arguments
        */
        *code thread_callback;
 
        Mutex lck();
 
        bool stop_flag;
 
        bool abort_flag;
 
        Counter cnt();
 
public:
 
 
    constructor(*hash<PipelineOptionInfo> opts);
 
 
 
    copy();
 
 
 
    destructor();
 
 
    string getName();
 
 
    bool hasQueue(int id);
 
 
    bool stopping();
 
 
    bool aborting();
 
 
    registerThread(PipelineQueue queue);
 
 
 
    append(AbstractDataProcessor processor);
 
 
 
    append(int id, AbstractDataProcessor processor);
 
 
 
    int appendQueue(int id);
 
 
    bool isProcessing();
 
 
    submit(auto _data);
 
 
    submitData(AbstractIterator i);
 
 
    submit(AbstractDataProviderRecordIterator i);
 
 
    submit(AbstractDataProviderBulkRecordInterface i);
 
 
 
    reset();
 
 
 
    waitDone();
 
 
 
    abort(*bool ignore_exceptions);
 
 
 
    hash<PipelineInfo> getInfo();
 
 
    reportError(PipelineQueue queue, hash<ExceptionInfo> ex);
 
 
    logInfo(string fmt);
 
 
    logError(string fmt);
 
 
    logDebug(string fmt);
 
 
private:
     PipelineQueue copyPipeline(PipelineQueue old_queue);
public:
 
 
 
private:
     submitIntern(auto _data);
public:
 
 
 
private:
     submitDataIntern(auto _data);
public:
 
 
 
private:
     submitBulkIntern(AbstractDataProviderBulkRecordInterface i);
public:
 
 
 
private:
     resetIntern();
public:
 
 
 
private:
     checkSubmitIntern();
public:
 
 
 
private:
     checkLockedIntern();
public:
 
 
 
private:
     throwPipelineException();
public:
 
 
private:
     checkUpdatePipelineIntern(int id);
public:
 
 
private:
     stopIntern();
public:
 
 
private:
     stopInternUnlocked();
public:
 
};
};
 
*list get_stack();