runControllerImpl.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_IMPL_H
#define RUN_CTRL_IMPL_H
 
#include <iterator>
#include <unordered_set>
#include "dbParameter.h"
#include "dbOutputTable.h"
#include "dbValue.h"
 
using namespace std;
 
namespace openm
{
    class SingleController : public RunController
    {
    public:
        SingleController(const ArgReader & i_argStore, IDbExec * i_dbExec) : 
            RunController(i_argStore),
            runId(0),
            taskId(0),
            taskRunId(0),
            isWaitTaskRun(false),
            dbExec(i_dbExec),
            isSubDone(subValueCount)
        { }
 
        virtual ~SingleController(void) noexcept { }
 
        virtual int nextRun(void) override;
 
        virtual void readParameter(
            const char * i_name, int i_subId, const type_info & i_type, size_t i_size, void * io_valueArr
            ) override;
 
        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
            ) override;
 
        virtual void shutdownRun(int i_runId) override { doShutdownRun(i_runId, taskRunId, dbExec); }
 
        virtual void shutdownWaitAll(void) override { doShutdownAll(taskRunId, dbExec); }
 
        virtual void shutdownOnExit(ModelStatus i_status) override { doShutdownOnExit(i_status, runId, taskRunId, dbExec); }
 
        virtual bool childExchange(void) override;
 
    protected:
        virtual void init(void) override;
 
        int currentRunId(void) const noexcept override { return runId; }
        
    private:
        int runId;              // if > 0 then model run id
        int taskId;             // if > 0 then modeling task id
        int taskRunId;          // if > 0 then modeling task run id
        bool isWaitTaskRun;     // if true then task run under external supervision
        IDbExec * dbExec;       // db-connection
        DoneVector isSubDone;   // size of [sub-value count], if true then all sub-value accumulators saved in database
 
    private:
        SingleController(const SingleController & i_runCtrl) = delete;
        SingleController & operator=(const SingleController & i_runCtrl) = delete;
    };
 
    class RootController : public RunController
    {
    public:
        RootController(int i_processCount, const ArgReader & i_argStore, IDbExec * i_dbExec, IMsgExec * i_msgExec) :
            RunController(i_argStore),
            taskId(0),
            taskRunId(0),
            isWaitTaskRun(false),
            dbExec(i_dbExec),
            msgExec(i_msgExec)
        {
            processCount = i_processCount;
            processRank = 0;
        }
 
        virtual ~RootController(void) noexcept { }
 
        virtual int nextRun(void) override;
 
        virtual void readParameter(
            const char * i_name, int i_subId, const type_info & i_type, size_t i_size, void * io_valueArr
            ) override;
 
        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
            ) override;
 
        virtual void shutdownRun(int i_runId) override;
 
        virtual void shutdownWaitAll(void) override;
 
        virtual void shutdownOnExit(ModelStatus i_status) override;
 
        virtual bool childExchange(void) override;
 
    protected:
        virtual void init(void) override;
 
        int currentRunId(void) const noexcept override {
            try {
                return (rootGroupDef.isRootActive && !runGroupLst.empty()) ? runGroupLst.back().runId : 0;
            }
            catch (...) {
                return 0;
            }
        }
 
    private:
        int taskId;                     // if > 0 then modeling task id
        int taskRunId;                  // if > 0 then modeling task run id
        bool isWaitTaskRun;             // if true then task run under external supervision
        IDbExec * dbExec;               // db-connection
        IMsgExec * msgExec;             // message passing interface
        ProcessGroupDef rootGroupDef;   // root process groups size, groups count and process rank in group
        list<RunGroup> runGroupLst;     // process groups run id and run state
        list<AccReceive> accRecvLst;    // list of accumulators to be received
 
        // return root process run group: last run group
        RunGroup & rootRunGroup(void) { return runGroupLst.back(); }
 
        void broadcastMetaData(void);
 
        template <class MetaTbl>
        void broadcastMetaTable(MsgTag i_msgTag, unique_ptr<MetaTbl> & io_tableUptr);
 
        void broadcastRunOptions(void);
 
        void broadcastLanguageMessages(void);
 
        int makeNextRun(RunGroup & i_runGroup);
 
        void readAllRunParameters(const RunGroup & i_runGroup) const;
 
        void appendAccReceiveList(int i_runId, const RunGroup & i_runGroup);
 
        bool receiveSubValues(void);
 
        void updateAccReceiveList(void);
 
        bool updateModelRunStatus(void);
 
        bool receiveStatusUpdate(long i_waitTime = 0L);
 
        // microdata receive status: row count and size for all entities where microdata required
        struct EntityMdReceive
        {
            int childRank = 0;      // child rank to receive microdata from
            bool isLast = false;    // if true then it is last entity microdata on model shutdown
            int childRunId = 0;     // child process run id
            vector<size_t> count;   // microdata row count for each entity
        };
 
        // microdata receive queue
        recursive_mutex mdRcvMutex;             // mutex to lock for receive operations
        //
        list<EntityMdReceive> entityMdRcvLst;
        vector<bool> isMicrodataFromRank;       // array [worldSize], if value true for the rank then microdata must be received
 
        bool receiveMicrodata(long i_waitTime = 0L);
 
        bool isAllMicrodataReceived(RunGroup & i_runGroup);
 
        bool isAllMicrodataReceived(int i_childRank);
 
    private:
        RootController(const RootController & i_runCtrl) = delete;
        RootController & operator=(const RootController & i_runCtrl) = delete;
    };
 
    class ChildController : public RunController
    {
    public:
        ChildController(int i_processCount, const ArgReader & i_argStore, IMsgExec * i_msgExec) :
            RunController(i_argStore),
            runId(0),
            msgExec(i_msgExec),
            lastTimeStatus(chrono::system_clock::now()),
            lastModelStatus(ModelStatus::init),
            isFinalExchange(false)
        {
            processCount = i_processCount; 
            processRank = i_msgExec->rank();
        }
 
        virtual ~ChildController(void) noexcept { }
 
        virtual int nextRun(void) override;
 
        virtual void readParameter(
            const char * i_name, int i_subId, const type_info & i_type, size_t i_size, void * io_valueArr
            ) override;
 
        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
            ) override;
 
        virtual void shutdownRun(int i_runId) override;
 
        virtual void shutdownWaitAll(void) override;
 
        virtual void shutdownOnExit(ModelStatus i_status) override;
 
        virtual bool childExchange(void) override;
 
    private:
        int runId;                                          // if > 0 then model run id
        ProcessGroupDef groupDef;                           // child process groups size, groups count and process rank in group
        IMsgExec * msgExec;                                 // message passing interface
        chrono::system_clock::time_point lastTimeStatus;    // last status update time sent to root
        ModelStatus lastModelStatus;                        // last model status sent to root
        bool isFinalExchange;                               // if true then final model status send or received from root
 
        virtual void init(void) override;
 
        int currentRunId(void) const noexcept override { return runId; }
 
        int broadcastMetaData(void);
 
        template <class MetaTbl>
        void broadcastMetaTable(MsgTag i_msgTag, unique_ptr<MetaTbl> & io_tableUptr);
 
        void broadcastRunOptions(void);
 
        void broadcastLanguageMessages(void);
 
        void sendStatusUpdate(void);
 
        size_t sendMicrodata(bool i_isLast);
 
    private:
        ChildController(const ChildController & i_runCtrl) = delete;
        ChildController & operator=(const ChildController & i_runCtrl) = delete;
    };
 
    class RestartController : public RunController
    {
    public:
        RestartController(const ArgReader & i_argStore, IDbExec * i_dbExec) : RunController(i_argStore),
            runId(0),
            setId(0),
            dbExec(i_dbExec),
            isSubDone(subValueCount)
        { }
 
        virtual ~RestartController(void) noexcept { }
 
        virtual int nextRun(void) override;
 
        virtual void readParameter(
            const char * i_name, int i_subId, const type_info & i_type, size_t i_size, void * io_valueArr
            ) override;
 
        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
            ) override;
 
        virtual void shutdownRun(int i_runId) override { doShutdownRun(i_runId, 0, dbExec); }
 
        virtual void shutdownWaitAll(void) override { doShutdownAll(0, dbExec); }
 
        virtual void shutdownOnExit(ModelStatus i_status) override { doShutdownOnExit(i_status, runId, 0, dbExec); }
 
        virtual bool childExchange(void) override;
 
    private:
        int runId;              // if > 0 then model run id
        int setId;              // if > 0 then set id of source input parametes
        IDbExec * dbExec;       // db-connection
        DoneVector isSubDone;   // size of [sub-value count], if true then all sub-value accumulators saved in database
 
        virtual void init(void) override;
 
        int currentRunId(void) const noexcept override { return runId; }
 
        // find sub-value to restart the run and update run status
        bool cleanupRestartSubValue(void);
 
    private:
        RestartController(const RestartController & i_runCtrl) = delete;
        RestartController & operator=(const RestartController & i_runCtrl) = delete;
    };
}
 
#endif  // RUN_CTRL_IMPL_H