runController.h

Go to the documentation of this file.

 
// Copyright (c) 2013-2015 OpenM++
// This code is licensed under the MIT license (see LICENSE.txt for details)
 
#ifndef RUN_CTRL_H
#define RUN_CTRL_H
 
#include <fstream>
#include "metaLoader.h"
#include "dbValue.h"
 
using namespace std;
 
static constexpr size_t MinSizeToSaveMicrodata = 128 * 1024;        
static constexpr size_t MaxSizeToSaveMicrodata = 2 * 1024 * 1024;   
namespace openm
{
    class RunController : public MetaLoader, public IRunBase
    {
    public:
        int subFirstId;
 
        int selfSubCount;
 
        int processCount;
 
        int processRank;
 
        static RunController * create(const ArgReader & i_argOpts, bool i_isMpiUsed, IDbExec * i_dbExec, IMsgExec * i_msgExec);
 
        virtual ~RunController(void) noexcept = 0;
 
        virtual int totalSubCount(void) const noexcept override { return subValueCount; }
 
        int parameterIdByName(const char * i_name) const override { return MetaLoader::parameterIdByName(i_name); };
 
        int parameterSubCount(int i_paramId) const override { return MetaLoader::parameterSubCount(i_paramId); }
 
        int parameterSelfSubCount(int i_paramId) const override
        {
            return parameterSubCount(i_paramId) > 1 ? selfSubCount : 1;
        }
                
        int parameterSubValueIndex(int i_paramId, int i_subId) const override
        {
            return (parameterSubCount(i_paramId) > 1 && subFirstId <= i_subId && i_subId < subFirstId + selfSubCount) ? 
                i_subId - subFirstId : 
                0;
        }
 
        bool isUseSubValue(int i_subId) const override
        {
            return subFirstId <= i_subId && i_subId < subFirstId + selfSubCount;
        }
 
        bool isTableSuppressed(const char * i_name) const override {
            return isSuppressed(i_name);
        }
 
        bool isTableSuppressed(int i_tableId) const override {
            return isSuppressed(i_tableId);
        }
 
        RunStateHolder & runStateStore(void) { return runStateHolder; }
 
        virtual int nextRun(void) = 0;
 
        virtual void shutdownRun(int i_runId) = 0;
 
        virtual void shutdownWaitAll(void) = 0;
 
        virtual void shutdownOnExit(ModelStatus i_status) = 0;
 
        virtual bool childExchange(void) = 0;
 
        virtual void writeAccumulators(
            const RunOptions & i_runOpts,
            bool i_isLastTable,
            const char * i_name,
            size_t i_size,
            forward_list<unique_ptr<double[]> > & io_accValues
            ) = 0;
 
        bool isOptionExist(const char * i_key) const noexcept override {
            return metaStore->runOptionTable ? metaStore->runOptionTable->isExist(currentRunId(), i_key) : false;
        }
 
        string strOption(const char * i_key, const string & i_default = "") const noexcept override {
            return metaStore->runOptionTable ? metaStore->runOptionTable->strValue(currentRunId(), i_key, i_default) : i_default;
        }
 
        bool boolOption(const char * i_key) const noexcept override {
            return metaStore->runOptionTable ? metaStore->runOptionTable->boolValue(currentRunId(), i_key) : false;
        }
 
        int boolOptionToInt(const char * i_key) const noexcept override {
            return metaStore->runOptionTable ? metaStore->runOptionTable->boolValueToInt(currentRunId(), i_key) : -1;
        }
 
        int intOption(const char * i_key, int i_default) const noexcept override {
            return metaStore->runOptionTable ? metaStore->runOptionTable->intValue(currentRunId(), i_key, i_default) : i_default;
        }
 
        long long longOption(const char * i_key, long long i_default) const noexcept override {
            return metaStore->runOptionTable ? metaStore->runOptionTable->longValue(currentRunId(), i_key, i_default) : i_default;
        }
 
        double doubleOption(const char * i_key, double i_default) const noexcept override {
            return metaStore->runOptionTable ? metaStore->runOptionTable->doubleValue(currentRunId(), i_key, i_default) : i_default;
        }
 
        vector<pair<string, string>> allOptions(void) const noexcept override;
 
        struct EntityAttrItem
        {
            int attrId;                                 
            int idxOf;                                  
            ptrdiff_t rowOffset = 0;                    
            unique_ptr<IValueFormatter> fmtValue;       
            unique_ptr<IValueFormatter> sqlFmtValue;    
            EntityAttrItem(int i_attrId, int i_index) : attrId(i_attrId), idxOf(i_index) {}
        };
 
        struct EntityItem
        {
            int entityId;                       
            vector<EntityAttrItem> attrs;       
            string genDigest;                   
            int genHid = 0;                     
            string dbMicrodataTable;            
            string csvHdr;                      
            size_t rowSize = 0;                 
            string sqlInsPrefix;                
            vector<DbValueSetter> rowSetter;    
            vector<const type_info *> rowTypes; 
            EntityItem(int i_entityId) : entityId(i_entityId) {}
        };
 
        void pushDbMicrodata(int i_runId, int i_entityKind, uint64_t i_microdataKey, const void * i_entityThis);
 
        void writeCsvMicrodata(int i_entityKind, uint64_t i_microdataKey, int i_eventId, bool i_isSameEntity, const void * i_entityThis, string & io_line);
 
        const string csvHeaderMicrodata(int i_entityKind) const;
 
        void makeCsvLineMicrodata(const EntityItem & i_entityItem, uint64_t i_microdataKey, int i_eventId, bool i_isSameEntity, const void * i_entityThis, string & io_line) const;
 
        tuple<bool, const RunController::EntityItem &> findEntityItem(int i_entityKind) const;
 
        bool isEntityEventInFilter(int i_entityKind, uint64_t i_microdataKey, int i_eventId) const;
 
    protected:
        RunController(const ArgReader & i_argStore) : MetaLoader(i_argStore),
            subFirstId(0),
            selfSubCount(0),
            processCount(1),
            processRank(0)
        { }
 
        virtual void init(void) = 0;
 
        virtual int currentRunId(void) const noexcept = 0;
 
        int createTaskRun(int i_taskId, IDbExec * i_dbExec);
 
        int findTask(IDbExec * i_dbExec);
 
        tuple<int, int, ModelStatus> createNewRun(int i_taskRunId, bool i_isWaitTaskRun, IDbExec * i_dbExec);
 
        void doShutdownOnExit(ModelStatus i_status, int i_runId, int i_taskRunId, IDbExec * i_dbExec);
 
        void doShutdownRun(int i_runId, int i_taskRunId, IDbExec * i_dbExec);
 
        void doShutdownAll(int i_taskRunId, IDbExec * i_dbExec);
 
        void doWriteAccumulators(
            int i_runId,
            IDbExec * i_dbExec,
            const RunOptions & i_runOpts,
            const char * i_name,
            size_t i_size,
            forward_list<unique_ptr<double[]> > & io_accValues
            ) const;
 
        void updateRestartSubValueId(int i_runId, IDbExec * i_dbExec, int i_subRestart) const;
 
        void updateRunState(IDbExec * i_dbExec, const map<pair<int, int>, RunState> i_updated) const;
 
        // microdata entities to write into database or csv
        map<int, EntityItem> entityMap;
        vector<int> entityIds;          // sorted microdata entity id's
 
        map<int, list<unique_ptr<uint8_t[]>>> pullDbMicrodata(bool i_isNow = false);
 
        struct IRowsFirstNext
        {
            virtual uint8_t * toFirst(void) = 0;
 
            virtual uint8_t * toNext(void) = 0;
        };
 
        struct ListFirstNext : IRowsFirstNext
        {
            ListFirstNext(list<unique_ptr<uint8_t[]>>::const_iterator i_begin, list<unique_ptr<uint8_t[]>>::const_iterator i_end) {
                beginIt = i_begin;
                endIt = i_end;
                rowIt = i_begin;
            }
 
            virtual uint8_t * toFirst(void) override
            {
                rowIt = beginIt;
                return rowIt != endIt ? rowIt->get() : nullptr;
            }
 
            virtual uint8_t * toNext(void) override
            {
                if (rowIt == endIt) return nullptr;
 
                ++rowIt;
                return rowIt != endIt ? rowIt->get() : nullptr;
            }
        private:
            list<unique_ptr<uint8_t[]>>::const_iterator beginIt;  // first row
            list<unique_ptr<uint8_t[]>>::const_iterator endIt;    // after last row
            list<unique_ptr<uint8_t[]>>::const_iterator rowIt;    // current row
        };
 
        struct BytesFirstNext : IRowsFirstNext
        {
            BytesFirstNext(size_t i_rowCount, size_t i_rowSize, uint8_t * i_rows) : count(i_rowCount), size(i_rowSize) {
                totalSize = size * count;
                pRows = i_rows;
                rowNow = 0;
            }
 
            virtual uint8_t * toFirst(void) override
            {
                if (pRows == nullptr || count <= 0 || size <= 0) return nullptr;
 
                rowNow = 0;
                return size <= totalSize ? pRows : nullptr;
            }
 
            virtual uint8_t * toNext(void) override
            {
                if (pRows == nullptr || count <= 0 || size <= 0 || rowNow >= count) return nullptr;
 
                rowNow++;
                return (rowNow + 1) * size <= totalSize ? pRows + (rowNow * size) : nullptr;
            }
        private:
            size_t count = 0;       // rows count
            size_t size = 0;        // row size in bytes
            size_t totalSize = 0;   // total size in bytes
            uint8_t * pRows;        // array of bytes [count, size]
            size_t rowNow = 0;      // current row index
        };
 
        size_t doDbMicrodata(IDbExec * i_dbExec, int i_entityId, IRowsFirstNext & i_entityMdRows);
 
        size_t doDbMicrodataSql(IDbExec * i_dbExec, const map<int, list<unique_ptr<uint8_t[]>>> & i_entityMdRows);
 
        void openCsvMicrodata(void);
 
    private:
        RunStateHolder runStateHolder;
 
        tuple<int, string, bool, bool> findWorkset(int i_setId, IDbExec * i_dbExec);
 
        int findBaseRun(IDbExec * i_dbExec);
 
        void createRunOptions(int i_runId, int i_setId, const string & i_setName, int i_baseRunId, int i_taskRunId, IDbExec * i_dbExec) const;
 
        // copy input parameters from "base" run and working set into new run id
        void createRunParameters(int i_runId, int i_setId, bool i_isWsDefault, bool i_isReadonlyWsDefault, int i_baseRunId, IDbExec * i_dbExec) const;
 
        // cretate run description and notes using run options or by copy it from workset text
        void createRunText(int i_runId, int i_setId, IDbExec * i_dbExec) const;
 
        // insert run entity metadata and create run entity database tables
        void createRunEntityTables(IDbExec* i_dbExec);
 
        // insert run entity rows for all entities in that model run
        void insertRunEntity(int i_runId, IDbExec * i_dbExec);
 
        // calculate entity generation digest: based on entity digest, attributes id, name, type digest
        const string makeEntityGenDigest(const EntityDicRow * i_entRow, const vector<EntityAttrRow> i_attrRows) const;
 
        void writeOutputValues(int i_runId, IDbExec * i_dbExec) const;
 
        void writeMicrodataDigest(int i_runId, IDbExec * i_dbExec) const;
 
        // check sub-value id's and count: sub id must be zero in range [0, sub count -1] and count must be equal to parameter size
        void checkParamSubCounts(int i_runId, int i_subCount, const ParamDicRow & i_paramRow, IDbExec * i_dbExec) const;
 
        // make part of where clause to select sub_id's
        const string makeWhereSubId(const MetaLoader::ParamSubOpts & i_subOpts, int i_defaultSubId) const;
 
        // make part of select columns list to map source sub id's to run parameter sub_id's
        const string mapSelectedSubId(const MetaLoader::ParamSubOpts & i_subOpts) const;
 
        // return parameter sub-value options by parameter id
        const MetaLoader::ParamSubOpts subOptsById(const ParamDicRow & i_paramRow, int i_subCount, int i_defaultSubId) const;
 
        RunImportOpts findImportRun(const string & i_modelName, const ImportOpts & i_importOpts, const IModelDicTable * i_mainModelTable, IDbExec * i_dbExec) const;
 
        bool importOutputTable(
            int i_runId,
            const ParamDicRow & i_paramRow,
            const ParameterNameSizeItem * i_nameSizeItem,
            const RunImportOpts & i_riOpts,
            const ParamImportRow * i_importRow,
            IDbExec * i_dbExec
        ) const;
 
        bool importParameter(
            int i_runId,
            const ParamDicRow & i_paramRow,
            const ParameterNameSizeItem * i_nameSizeItem,
            const RunImportOpts & i_riOpts,
            const ParamImportRow * i_importRow,
            const ParamSubOpts & i_subOpts,
            IDbExec * i_dbExec
        ) const;
 
        // microdata entity to strore in database
        struct EntityDbItem
        {
            int entityId;                           // entity metadata id
            //
            recursive_mutex theMutex;               // mutex to lock for db write operations
            list<unique_ptr<uint8_t[]>> rowLst;     // microdata list of rows
            chrono::system_clock::time_point lastSaveTime = chrono::system_clock::time_point::min();  // last time of save
        };
        map<int, EntityDbItem> entityDbMap;         // map entity id to database microdata buffer
 
        tuple<bool, size_t> statusDbMicrodata(chrono::system_clock::time_point i_nowTime, EntityDbItem & i_entityDbItem);
 
        list<unique_ptr<uint8_t[]>> && moveDbMicrodata(chrono::system_clock::time_point i_nowTime, EntityDbItem & io_entityDbItem);
 
        // microdata entity to write into csv
        struct EntityCsvItem
        {
            int entityId;               // entity metadata id
            string filePath;            // if not empty then microdata csv file path
            //
            recursive_mutex theMutex;   // mutex to lock for csv write operations
            bool isReady;               // if true then csv output file open and ready to use
            ofstream csvSt;             // csv output stream
        };
        map<int, EntityCsvItem> entityCsvMap;   // map entity id to microdata entity csv file
 
        // initialize microdata entity writing
        void initMicrodata(void);
 
        void closeCsvMicrodata(void) noexcept;
 
        void writeCsvLineMicrodata(EntityCsvItem & io_entityCsvItem, const string & i_line);
 
    private:
        RunController(const RunController & i_runCtrl) = delete;
        RunController & operator=(const RunController & i_runCtrl) = delete;
    };
}
 
#endif  // RUN_CTRL_H