Analysis.h

Go to the documentation of this file.

//***************************************************************************************
//***************************************************************************************
 
#pragma once
 
#include "NAGASH/Global.h"
#include "NAGASH/ThreadPool.h"
#include "NAGASH/ConfigTool.h"
#include "NAGASH/LoopEvent.h"
#include "NAGASH/ResultGroup.h"
#include "NAGASH/Job.h"
 
namespace NAGASH
{
 
    class Analysis
    {
    public:
        Analysis();
        Analysis(const Analysis &ana) = delete;
        Analysis(Analysis &&ana) = delete;
        Analysis &operator=(const Analysis &ana) = delete;
        Analysis &operator=(Analysis &&ana) = delete;
 
        Analysis(const ConfigTool &c);
        Analysis(const char *configfilename);
        ~Analysis();
        void SetThreadNumber(int num);
        std::shared_ptr<ConfigTool> ConfigUser();
 
        // deal with eventloop
        // for multi thread
        template <typename LE>
        std::shared_ptr<ResultGroup> ProcessEventLoop(const char *filelist, bool dosave = true);
        template <typename LE>
        std::shared_ptr<ResultGroup> SubmitEventLoop(const char *filelist, bool dosave = true);
        // for multi process
        template <typename LE>
        std::shared_ptr<ResultGroup> ProcessEventLoopMainThread(const char *filename, bool dosave = true);
 
        // deal with user defined functions
        template <typename J, typename... Args>
        std::shared_ptr<ResultGroup> SubmitJob(bool doSave, Args &&...args);
        template <typename J, typename JS, typename... Args>
        std::shared_ptr<ResultGroup> SubmitJob(bool doSave, Args &&...args);
 
        std::shared_ptr<MSGTool> MSGUser();
        void Join();
 
    protected:
        std::shared_ptr<ConfigTool> config;
        std::vector<std::shared_ptr<ResultGroup>> ResultsToSave;
        std::shared_ptr<Timer> TimerUser();
 
    private:
        int ThreadNumber = 1;
        std::vector<int> SubmittedJobCount;
        int SubmittedJobs = 0;
        std::shared_ptr<MSGTool> msg;
        std::shared_ptr<Timer> timer;
        std::shared_ptr<ThreadPool> pool;
        std::vector<std::future<StatusCode>> vf;
        void ConfigThreadPool();
 
        template <typename F, typename... Args>
        void ThreadPoolEnqueue(F &&f, Args &&...args);
    };
 
    inline std::shared_ptr<ConfigTool> Analysis::ConfigUser() { return this->config; } 
    inline std::shared_ptr<Timer> Analysis::TimerUser() { return timer; }              
    inline std::shared_ptr<MSGTool> Analysis::MSGUser() { return msg; }                
 
    template <typename LE>
    std::shared_ptr<ResultGroup> Analysis::ProcessEventLoop(const char *filelist, bool dosave)
    {
        auto result = SubmitEventLoop<LE>(filelist, dosave);
        Join();
        return result;
    }
 
    template <typename LE>
    std::shared_ptr<ResultGroup> Analysis::SubmitEventLoop(const char *filelist, bool dosave)
    {
        // check LE type
        static_assert(!std::is_pointer<LE>::value, "Analysis::SubmitEventLoop : type can not be a pointer");
        static_assert(std::is_base_of<LoopEvent, LE>::value, "Analysis::SubmitEventLoop : Input type must be or be inherited from LoopEvent class");
 
        std::ifstream infile;
        infile.open(filelist, std::ios::in);
        if (!infile.is_open())
        {
            MSGUser()->StartTitle("Analysis::SubmitEventLoop");
            MSGUser()->MSG_WARNING("file list ", filelist, " does not exist, returning a null result");
            MSGUser()->EndTitle();
            return nullptr;
        }
 
        auto origin = msg->OutputLevel;
        bool existfileselection = false;
        bool existfileselectionpattern = false;
        msg->OutputLevel = MSGLevel::SILENT;
        std::vector<int> fileselection = config->GetPar<std::vector<int>>("FileSelection", &existfileselection);
        std::string fileselectionpattern(config->GetPar<TString>("FileSelectionPattern", &existfileselectionpattern).Data());
        std::regex fileselectionregex(fileselectionpattern);
        msg->OutputLevel = origin;
 
        auto result = std::make_shared<ResultGroup>(this->msg, this->config);
        std::string line;
        int count = 0;
        while (std::getline(infile, line))
        {
            ++count;
 
            if (existfileselection && std::find(fileselection.begin(), fileselection.end(), count) == fileselection.end())
                continue;
 
            if (existfileselectionpattern && !std::regex_search(line, fileselectionregex))
                continue;
 
            // remove comment and check if the file exists
            auto commentstart = line.find("#");
            if (commentstart != std::string::npos)
                line.erase(commentstart, line.size());
 
            // remove the carriage return
            if (line.back() == '\r')
                line.pop_back();
 
            int isexist = access(line.c_str(), F_OK);
            if (isexist == 0)
            {
                SubmitJob<LoopEvent, LE>(false, result, TString(line.c_str()));
            }
            else if (line.size() != 0)
            {
                MSGUser()->StartTitle("Analysis::SubmitEventLoop");
                MSGUser()->MSG_WARNING("file ", line, " does not exist, will skip this file");
                MSGUser()->EndTitle();
            }
        }
 
        if (dosave)
            ResultsToSave.push_back(result);
 
        infile.close();
        return result;
    }
 
    template <typename LE>
    inline std::shared_ptr<ResultGroup> Analysis::ProcessEventLoopMainThread(const char *filename, bool dosave)
    {
        // check LE type
        static_assert(!std::is_pointer<LE>::value, "Analysis::SubmitEventLoop : type can not be a pointer");
        static_assert(std::is_base_of<LoopEvent, LE>::value, "Analysis::SubmitEventLoop : Input type must be or be inherited from LoopEvent class");
 
        ConfigTool configCopy(*config);
        configCopy.BookPar<int>("JobNumber", -1);
        auto r = std::make_shared<ResultGroup>(this->msg, this->config);
        LoopEvent::Process<LE>(configCopy, nullptr, r, filename);
        return r;
    }
 
    template <typename J, typename... Args>
    inline std::shared_ptr<ResultGroup> Analysis::SubmitJob(bool doSave, Args &&...args)
    {
        static_assert(!std::is_pointer<J>::value, "Analysis::SubmitJob : type can not be a pointer");
        static_assert(std::is_base_of<Job, J>::value, "Analysis::SubmitJob : Input type must be or be inherited from Job class");
 
        ConfigTool configCopy(*config);
        configCopy.BookPar<int>("JobNumber", SubmittedJobs);
        // copy msg
        configCopy.SetMSG(std::make_shared<MSGTool>());
        configCopy.MSGUser()->OutputLevel = config->MSGUser()->OutputLevel;
        configCopy.MSGUser()->DefinedLevel = config->MSGUser()->DefinedLevel;
        configCopy.MSGUser()->FocusName = config->MSGUser()->FocusName;
        configCopy.MSGUser()->isCreateLog = config->MSGUser()->isCreateLog;
        configCopy.MSGUser()->isOpenLog = config->MSGUser()->isOpenLog;
        configCopy.MSGUser()->LogFileName = config->MSGUser()->LogFileName;
        configCopy.MSGUser()->FuncTitle = config->MSGUser()->FuncTitle;
 
        auto r = std::make_shared<ResultGroup>(this->msg, this->config);
        ThreadPoolEnqueue(J::Process, configCopy, r, std::forward<Args>(args)...);
        if (doSave)
        {
            ResultsToSave.emplace_back(r);
        }
 
        return r;
    }
 
    template <typename J, typename JS, typename... Args>                          // Job::Process with template
    std::shared_ptr<ResultGroup> Analysis::SubmitJob(bool doSave, Args &&...args) // use config from Analysis
    {
        static_assert(!std::is_pointer<J>::value, "Analysis::SubmitJob : type can not be a pointer");
        static_assert(std::is_base_of<Job, J>::value, "Analysis::SubmitJob : Input type must be or be inherited from Job class");
 
        ConfigTool configCopy(*config);
        configCopy.BookPar<int>("JobNumber", SubmittedJobs);
        // copy msg
        configCopy.SetMSG(std::make_shared<MSGTool>());
        configCopy.MSGUser()->OutputLevel = config->MSGUser()->OutputLevel;
        configCopy.MSGUser()->DefinedLevel = config->MSGUser()->DefinedLevel;
        configCopy.MSGUser()->FocusName = config->MSGUser()->FocusName;
        configCopy.MSGUser()->isCreateLog = config->MSGUser()->isCreateLog;
        configCopy.MSGUser()->isOpenLog = config->MSGUser()->isOpenLog;
        configCopy.MSGUser()->LogFileName = config->MSGUser()->LogFileName;
        configCopy.MSGUser()->FuncTitle = config->MSGUser()->FuncTitle;
 
        auto r = std::make_shared<ResultGroup>(this->msg, this->config);
        ThreadPoolEnqueue(J::template Process<JS>, configCopy, r, std::forward<Args>(args)...);
        if (doSave)
        {
            ResultsToSave.emplace_back(r);
        }
 
        return r;
    }
 
    template <typename F, typename... Args>
    void Analysis::ThreadPoolEnqueue(F &&f, Args &&...args)
    {
        if (pool == nullptr)
        {
            msg->MSG(MSGLevel::WARNING, "Thread pool not defined, only one thread will be created");
            pool = std::make_shared<ThreadPool>(1);
        }
 
        TString record;
        record.Form("(Submit Job %d)", SubmittedJobs);
        TimerUser()->Record(record);
        TString jobnum;
        jobnum.Form("Submit job %d at ", SubmittedJobs);
        TimerUser()->PrintCurrent(jobnum);
        vf.emplace_back(pool->enqueue(f, std::forward<Args>(args)...));
        SubmittedJobCount.push_back(SubmittedJobs);
        SubmittedJobs++;
    }
 
} // namespace NAGASH