|| || || || || || || ||

---

Runner.h

Go to the documentation of this file.

#ifndef Impala_Job_Runner_h
#define Impala_Job_Runner_h

#include <sstream>

#include "Basis/ILog.h"
#include "Basis/Timer.h"
#include "Basis/FileReadString.h"

#include "Util/StringParser.h"
#include "Util/ChannelProxy.h"
#include "Job/State.h"
#include "Process/ManagerFactory.h"
#include "Process/Manager.h"

namespace Impala
{
namespace Job
{

/***********************************************************************
Instances of this class can run jobs served by a job server. 
************************************************************************/

class Runner
{

public:

    Runner()
    {
        mProcessManager = Process::ManagerFactory::Create();
        mJobServerProxy = 0;
    }

    virtual
    ~Runner()
    {
        Disconnect();

        if (mProcessManager)
            delete mProcessManager;
    }

    bool
    IsConnected()
    {
        return (mJobServerProxy && mJobServerProxy->ChannelIsValid());
    }

    void
    Disconnect()
    {
        if (mJobServerProxy)
        {
            delete mJobServerProxy;
            mJobServerProxy = 0;
        }
    }

    bool
    Connect(CString serverName, int port, CString passwordFile)
    {
        mJobServerProxy = new Util::ChannelProxy(serverName, port, passwordFile);
        if (!mJobServerProxy->GetChannel())
        {
            ILOG_ERROR("Failed to connect to job server");
            Disconnect();
            return false;
        }

        return true;
    }

    void
    PoisonServer()
    {
        ILOG_WARN("Requesting job server to shut down all runners " <<
                  "after they finish their current job");
        mJobServerProxy->Send("Poison");
    }

    void
    StopServer()
    {
        ILOG_WARN("Requesting job server to shut down");
        mJobServerProxy->Send("ShutDown");
    }

    void
    RunJobs(int maxJobs, int maxIdleTimeSec, int maxJobTimeMinutes,
            int sleepIntervalSec, String cmdLineExtension)
    {
        if (!IsConnected())
        {
            ILOG_ERROR("Not connected to a job server");
            return;
        }

        mMaxJobs = maxJobs;
        mMaxIdleTimeSec = maxIdleTimeSec;
        mMaxJobTimeMinutes = maxJobTimeMinutes;
        mSleepIntervalSec = sleepIntervalSec;

        while (true)
        {
            if (mMaxJobs > 0 && mJobCmdLineList.size() >= mMaxJobs)
            {
                ILOG_INFO("Maximum of jobs to run was reached (" <<
                          mJobCmdLineList.size() << ")");
                break;
            }
            
            int jobId = 0;
            String cmdLine = "";
            if (!WaitForJob(jobId, cmdLine) ||
                !RunJob(jobId, cmdLine, cmdLineExtension))
            {
                break;
            }
        }
    }

    const std::list<String>&
    GetJobList() const
    {
        return mJobCmdLineList;
    }

private:

    bool
    WaitForJob(int& jobId, String& cmdLine)
    {
        jobId = 0;
        int secWaiting = 0;
        String response = "";
        bool reportedWaiting = false;
        while (true)
        {
            response = mJobServerProxy->Send("ProvideJob");
            if (response == "NoJob")
            {
                if (mMaxIdleTimeSec > 0 && secWaiting >= mMaxIdleTimeSec)
                {
                    ILOG_INFO("Maximum idle time reached (" << secWaiting <<
                              ")");
                    return false;
                }
                if (!reportedWaiting)
                {
                    ILOG_INFO(mJobCmdLineList.size() <<
                              " job(s) done; waiting for new job");
                    reportedWaiting = true;
                }
                Sleep(mSleepIntervalSec);
                secWaiting += mSleepIntervalSec;
            }
            else if (response.substr(0, 7) == "RunJob:")
            {
                if (response.substr(7, 6) == "jobId=")
                {
                    Util::StringParser parser(response, 13);
                    jobId = parser.GetInt(';');
                    if (jobId > 0)
                    {
                        String rest = parser.GetString(';');
                        if (rest.substr(0, 8) == "cmdLine=")
                        {
                            cmdLine = rest.substr(8);
                            if (!cmdLine.empty())
                                return true;
                        }
                    }
                }
                ILOG_WARN("Received invalid job specification: " << response);
                return false;
            }
            else if (response.substr(0, 6) == "Poison")
            {
                // exit the job runner
                return false;
            }
            else
            {
                ILOG_ERROR("Unexpected job request response: " << response);
                return false;
            }
        }
    }

    bool
    RunJob(int jobId, String cmdLine, String cmdLineExtension)
    {
        ILOG_INFO("Received job with ID: " << jobId);
        mJobIdList.push_back(jobId);
        cmdLine += " " + cmdLineExtension;
        cmdLine += " --jobErrorLog job_" + MakeString(jobId) + "_errors.log";
        mJobCmdLineList.push_back(cmdLine);
        int procHandle = mProcessManager->Create(cmdLine);
        if (!procHandle)
            return false; // an error was already reported

        Timer jobTimer(0); // starts the timer implicitly
        bool licensedToKill = false;
        while (mProcessManager->IsRunning(procHandle))
        {
            // report to job server
            std::ostringstream oss;
            oss << "AcceptStatus:jobId=" << jobId << ";state="
                << Job::State::RUNNING;
            String response = mJobServerProxy->Send(oss.str());

            if (response == "OK")
            {
                if (licensedToKill)
                    ILOG_INFO("Waiting for job " << jobId << " to get killed");

                double elapsedSec = jobTimer.SplitTime();
                int elapsedMinutes = (int) (elapsedSec / 60.0);

                if (mMaxJobTimeMinutes > 0 &&
                    elapsedMinutes >= mMaxJobTimeMinutes && !licensedToKill)
                {
                    ILOG_ERROR("Job " << jobId << " runs too long (over " <<
                               mMaxJobTimeMinutes <<
                               " minutes); requesting to kill it");
                    licensedToKill = true;
                    mProcessManager->Kill(procHandle);
                }

                Sleep(mSleepIntervalSec);
            }

            else if (response == "Kill")
            {
                if (licensedToKill)
                {
                    ILOG_WARN("Job " << jobId <<
                              " was already requested to be killed");
                }
                else
                {
                    ILOG_WARN("Received kill request for job " << jobId <<
                              "; forwarding request to process manager");
                    licensedToKill = true;
                    mProcessManager->Kill(procHandle);
                }
            }

            else
            {
                ILOG_ERROR("Reporting on job " << jobId << " failed: " <<
                           response);
                return false;
            }
        }

        // report job termination to job server
        int pmState = mProcessManager->GetState(procHandle);
        ILOG_DEBUG("pmState = " << pmState);
        int jobState = Job::State::UNDEFINED;
        int exitCode = mProcessManager->GetExitCode(procHandle);

        if (licensedToKill)
        {
            if (pmState == Process::Manager::STATE_TERM_SUCCESS ||
                pmState == Process::Manager::STATE_TERM_ERROR)
            {
                ILOG_INFO("Job " << jobId <<
                          " terminated normally and did not require killing");
                jobState = Job::State::TERMINATED_NRM;
            }
            else
            {
                ILOG_INFO("Job " << jobId << " has been killed");
                jobState = Job::State::TERMINATED_ABNRM;
            }
        }
        else
        {
            if (pmState == Process::Manager::STATE_TERM_SUCCESS)
            {
                ILOG_INFO("Job " << jobId << " is done");
                jobState = Job::State::TERMINATED_NRM;
            }
            else if (pmState == Process::Manager::STATE_TERM_ERROR)
            {
                ILOG_ERROR("Job " << jobId << " exited with a value of "
                           << exitCode << "; check the job's log for errors");
                jobState = Job::State::TERMINATED_NRM;
            }
            else if (pmState == Process::Manager::STATE_TERM_ABNORMAL)
            {
                ILOG_ERROR("Job " << jobId << " exited with a value of " <<
                           exitCode << "; check the job's log for errors");
                jobState = Job::State::TERMINATED_ABNRM;
            }
            else
            {
                ILOG_ERROR("Job " << jobId <<
                           " was rendered in an unknown state" <<
                           "; check the job's log for errors");
                jobState = Job::State::TERMINATED_ABNRM;
            }
        }

        std::ostringstream oss;
        oss << "AcceptStatus:jobId=" << jobId << ";state=" << jobState
            << ";exitCode=" << exitCode << ";errorLog=" << GetJobErrorLog(jobId);

        ILOG_DEBUG("Reporting job termination to server: " << oss.str());
        String response = mJobServerProxy->Send(oss.str());
        if (response != "OK")
        {
            ILOG_ERROR("Reporting termination of job " << jobId << " failed: "
                       << response);
            return false;
        }
        return true;
    }

    void
    Sleep(int sec)
    {
        mProcessManager->Sleep(sec);
    }

    String
    GetJobErrorLog(int jobId)
    {
        String res("no job errors found");
        std::vector<String> errs;
        String fName = "job_" + MakeString(jobId) + "_errors.log";

        FileReadString(std::back_inserter(errs), fName, false, false);
        if (errs.size() == 0)
            return res;

        res = errs[0];
        for (int i=1 ; i<errs.size() ; i++)
            res += errs[i];
        unlink(fName.c_str());
        return res;
    }

    // sleep interval for receiving and monitoring jobs
    //static const int SLEEP_INTERVAL_SEC = 10; // oversleeping 5 sec on average!

    ILOG_VAR_DECL;

    Process::Manager* mProcessManager;
    Util::ChannelProxy* mJobServerProxy;
    int mMaxJobs;
    int mMaxIdleTimeSec;
    int mMaxJobTimeMinutes;
    int mSleepIntervalSec;
    std::list<int> mJobIdList;
    std::list<String> mJobCmdLineList;

}; // class

ILOG_VAR_INIT(Runner, Impala.Job);

} // namespace 
} // namespace 

#endif

---