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

---

Server.h

Go to the documentation of this file.

#ifndef Impala_Job_Server_h
#define Impala_Job_Server_h

#include <deque>
#include <map>

#include <strstream>

#include "Basis/ILog.h"
#include "Util/StringParser.h"
#include "Util/ChannelServer.h"
#include "Job/State.h"

namespace Impala
{
namespace Job
{

/******************************************************************
 Instances of this class can serve jobs to job processing clients.
*******************************************************************/

class Server : public Util::ChannelServer
{
    typedef Job::State JobState;
    typedef Job::State::StateType JobStateType;

public: 

    typedef struct Job
    {
        // 'public':
        int id; 
        String jobSpec; 
        JobStateType state;
        int exitCode;
        String errorLog;

        // 'private':
        //String srcAddress;
        int srcPort;
        int portUsed;
        String agent;
    } Job;

    Server(int port, int nrPorts, CString passwordFile) 
        : Util::ChannelServer(port, nrPorts, true, passwordFile)
    {
        Init();
    }

    virtual
    ~Server()
    {
        int len = mJobsDone.size();
        for (int i = 0; i < len; i++)
            delete mJobsDone[i];

        len = mJobsWaiting.size();
        for (int i = 0; i < len; i++)
            delete mJobsWaiting[i];

        int maxPort = FirstPort() + NrOfPorts() - 1;
        for (int port = (FirstPort() + 1); port <= maxPort; port++)
        {
            if (mJobByPort[port] != 0)
            {
                Job* job = mJobByPort[port];
                delete job;
                mJobByPort[port] = 0;
            }
        }
    }

    virtual void 
    HandleIdle()
    {
        ChannelServer::HandleIdle();

        int jobsWaiting = mJobsWaiting.size();
        int jobsDone = mJobsDone.size();
        int runningJobs = 0;
        int maxPort = FirstPort() + NrOfPorts() - 1;
        for (int port = (FirstPort() + 1); port <= maxPort; port++)
            if (mJobByPort[port] != 0)
                runningJobs++;
        ILOG_DEBUG("Jobs: waiting=" << jobsWaiting << ", running=" <<
                   runningJobs << ", done=" << jobsDone);
    }

    virtual void 
    HandleDisconnect(int port)
    {
        ChannelServer::HandleDisconnect(port);
        RescheduleJob(port);
        RemoveJobsAccepted(port);
    }

    bool 
    JobQueueFull()
    {
        return (mJobsWaiting.size() > 1000000);
    }

    // Adds a new job to the job queue
    int 
    AddToJobQueue(Job* job)
    {
        if (JobQueueFull())
        {
            ILOG_ERROR("Job queue full; cannot add job");
            return 0;
        }
        else
        {
            job->id = mNextJobId++; // doesnt account for overflow
            ILOG_INFO("Assigning job ID: " << job->id);
            job->state = State::SCHEDULED;
            job->portUsed = -1; // undefined
            job->exitCode = -99; // undefined
            mJobsWaiting.push_back(job);
            return job->id;
        }
    }

    Job* 
    FindJob(int jobId)
    {
        // check running jobs...
        int maxPort = FirstPort() + NrOfPorts() - 1;
        for (int port = (FirstPort() + 1); port <= maxPort; port++)
            if (mJobByPort[port] != 0 && mJobByPort[port]->id == jobId)
                return mJobByPort[port];

        // ...and new jobs...
        int len = mJobsWaiting.size();
        for (int i = 0; i < len; i++)
            if (mJobsWaiting[i]->id == jobId)
                return mJobsWaiting[i];

        // ...and terminated job
        len = mJobsDone.size();
        for (int i = 0; i < len; i++)
            if (mJobsDone[i]->id == jobId)
                return mJobsDone[i];

        //ILOG_ERROR("Job ID " << jobId << " unknown");
        return 0;
    }

    // Pops the next job to run from the queue 
    Job* 
    PopFirstFromJobQueue()
    {
        Job* job = 0;
        if (mJobsWaiting.empty())
        {
            ILOG_ERROR("Job queue empty; cannot pop job");
        }
        else
        {
            job = mJobsWaiting[0];
            mJobsWaiting.pop_front();
        }
        return job;
    }

protected:

    virtual int
    AcceptRequest(char* buf, int len, int bufSize, CString conn, int port)
    {
        int rcBase = ChannelServer::AcceptRequest(buf, len, bufSize, conn, port);
        if (rcBase >= 0)
            return rcBase;

        ILogErrors& errors = ILogErrors::GetInstance();
        errors.Mark();

        if (strncmp(buf, "AcceptJob", 9) == 0)
            AcceptNewJob(buf, len, bufSize, conn, port);
        else if (strncmp(buf, "ProvideStatus", 13) == 0)
            AcceptRequestForStatus(buf, len, bufSize, conn, port);
        else if (strncmp(buf, "ProvideJob", 10) == 0)
            AcceptRequestForJob(buf, len, bufSize, conn, port);
        else if (strncmp(buf, "AcceptStatus", 12) == 0)
            AcceptJobStatus(buf, len, bufSize, conn, port);
        else if (strncmp(buf, "Poison", 6) == 0)
            mPoisoned = !mPoisoned;
        else
            return -1;

        if (errors.GetNrErrorsSinceMark() > 0)
        {
            String last = errors.GetLastErrorSinceMark();
            sprintf(buf, "ERROR : last error at server : %s\0", last.c_str());
        }

        return strlen(buf) + 1;
    }

    virtual void
    AcceptNewJob(char* buf, int len, int bufSize, CString conn, int port)
    {
        String curCon = ConnectionDescr();
        ILOG_INFO("Receiving new job from " << curCon);
        if (JobQueueFull())
        {
            ILOG_WARN("Job queue full, cannot accept new job");
            sprintf(buf, "MaxJobs\0");
        }
        else
        {
            Util::StringParser parser(String(buf, len));
            parser.GetString(':');
            if (parser.GetString('=') == "cmdLine")
            {
                String cmdLine = parser.GetString('\0');
                Job* job = new Job();
                job->srcPort = port;
                job->jobSpec = cmdLine;
                int jobId = AddToJobQueue(job);
                sprintf(buf, "JobAccepted:jobId=%i\0", jobId);
            }
            else
            {
                ILOG_ERROR("Invalid job specification: " << String(buf, len));
                sprintf(buf, "MaxJobs\0");
            }
        }
    }

    // SK: currently only job termination code; should perhaps be extended
    virtual void
    AcceptRequestForStatus(char* buf, int len, int bufSize, CString conn,
                           int port)
    {
        String curCon = ConnectionDescr();
        Util::StringParser parser(buf);
        parser.GetString(':');
        String requestKey = parser.GetString('=');
        if (requestKey == "jobId")
        {
            int jobId = parser.GetInt();
            if (jobId > 0)
            {
                ILOG_DEBUG("Receiving request for report on job " << jobId <<
                           " from " << curCon);
                Job* job = FindJob(jobId);
                if (job)
                {
                    sprintf(buf,
                            "JobStatus:jobId=%d;state=%d;exitCode=%d;errorLog=%s\0",
                            jobId, job->state, job->exitCode,
                            job->errorLog.c_str());
                    return;
                }
                else
                {
                    ILOG_ERROR("Job ID unknown: " << jobId);
                }
            }
            else
            {
                ILOG_ERROR("No job ID specified: " << String(buf, len));
            }
        }
        else if (requestKey == "reference")
        {
            ILOG_DEBUG("Receiving request for job state changes from " <<
                       curCon);
            std::map<int, int> refList;
            int jobId = parser.GetInt(':', false, true);
            while (jobId > 0)
            {
                JobStateType state = (JobStateType) parser.GetInt(' ', true,
                                                                  true);
                refList[jobId] = state;
                jobId = parser.GetInt(':', false, true);
            }

            int nrOfChanges = 0;
            std::ostringstream oss;
            std::map<int, int>::iterator iter = refList.begin();
            for ( ; iter != refList.end(); iter++)
            {
                int jobId = iter->first;
                Job* job = FindJob(jobId);
                if (job)
                {
                    if (job->state != iter->second)
                    {
                        oss << jobId << ':' << job->state << " ";
                        nrOfChanges++;
                    }
                }
                else
                {
                    ILOG_ERROR("Job ID unknown: " << jobId);
                }
            }
            if (nrOfChanges > 0)
                sprintf(buf, "JobStatus:changed=%s\0", oss.str().c_str());
            else
                sprintf(buf, "NoChange\0");
            return;
        }
        else
        {
           ILOG_ERROR("Invalid job report request: " << String(buf, len));
        }

        sprintf(buf, "ERROR\0");
    }

    virtual void
    AcceptRequestForJob(char* buf, int len, int bufSize, CString conn, int port)
    {
        String curCon = ConnectionDescr();
        ILOG_DEBUG("Receiving job request from " << curCon);
        if (mJobByPort[port] != 0)
        {
            ILOG_WARN("Unexpected job request from " << curCon);
            sprintf(buf, "ERROR\0");
            return;
        }

        if (mPoisoned)
        {
            ILOG_DEBUG("Poisoning " << curCon);
            sprintf(buf, "Poison\0");
            return;
        }

        if (mJobsWaiting.empty())
        {
            ILOG_DEBUG("No job for " << curCon);
            sprintf(buf, "NoJob\0");
            return;
        }

        Job* job = PopFirstFromJobQueue();
        job->portUsed = port;
        job->agent = conn;
        job->state = State::RUNNING;

        mJobByPort[port] = job;
        ILOG_INFO("Serving job " << job->id << " to " << curCon);
        sprintf(buf, "RunJob:jobId=%d;cmdLine=%s\0", job->id,
                job->jobSpec.c_str());
    }

    virtual void
    AcceptJobStatus(char* buf, int len, int bufSize, CString conn, int port)
    {
        String curCon = ConnectionDescr();
        ILOG_DEBUG("Receiving job report from " << curCon);
        Job* job = mJobByPort[port];
        if (job == 0)
        {
            ILOG_WARN("Unexpected job report from " << curCon);
            sprintf(buf, "ERROR\0");
            return;
        }

        String bufStr(buf);
        Util::StringParser parser(bufStr);
        parser.GetString(':');
        if (parser.GetString('=') == "jobId")
        {
            int jobId = parser.GetInt(';', true, true);
            if (jobId == job->id)
            {
                if (parser.GetString('=') == "state")
                {
                    JobStateType reportedState =
                        (JobStateType) parser.GetInt(';', false, true);
                    if (reportedState == State::RUNNING)
                    {
                        if (job->state == State::KILL_ATTEMPT)
                        {
                            ILOG_WARN("Requesting " << curCon << " to kill job "
                                      << jobId);
                            sprintf(buf, "Kill\0");
                        }
                        else
                        {
                            sprintf(buf, "OK\0");
                        }
                        return;
                    }
                    else if (reportedState == State::TERMINATED_NRM ||
                             reportedState == State::TERMINATED_ABNRM)
                    {
                        if (job->state == State::KILL_ATTEMPT)
                            ILOG_WARN("Successfully completed kill for job "
                                      << job->id);
                        job->state = reportedState;
                        if (parser.GetString('=') == "exitCode")
                            job->exitCode = parser.GetInt(';', true, true);
                        if (parser.GetString('=') == "errorLog")
                            job->errorLog = bufStr.substr(parser.Position());
                        mJobsDone.push_back(job);
                        mJobByPort[port] = 0;
                        sprintf(buf, "OK\0");
                        return;
                    }
                }
            }
        }
        ILOG_WARN("Invalid job report (from " << curCon << ") : " <<
                  String(buf, len));
        sprintf(buf, "ERROR\0");
    }


private:

    void 
    Init()
    {
        mNextJobId = 1;
        mPoisoned = false;
    }

    void
    RemoveJobsAccepted(int srcPort)
    {
        CString logInformation = "Job source at port " + MakeString(srcPort) +
            " disconnected; removing its jobs";
        bool loggedInformational = false;

        std::deque<Job*>::iterator iter = mJobsWaiting.begin();
        while (iter != mJobsWaiting.end())
        {
            if ((*iter)->srcPort == srcPort)
            {
                if (!loggedInformational)
                    ILOG_WARN(logInformation);
                loggedInformational = true;

                ILOG_INFO("Removed pending job " << (*iter)->id <<
                          " accepted at port " << srcPort);
                delete *iter;
                iter = mJobsWaiting.erase(iter);
            }
            else
            {
                iter++;
            }
        }

        int maxPort = FirstPort() + NrOfPorts() - 1;
        for (int port = (FirstPort() + 1); port <= maxPort; port++)
        {
            Job* job = mJobByPort[port];
            if (job && job->srcPort == srcPort)
            {
                if (!loggedInformational)
                    ILOG_WARN(logInformation);
                loggedInformational = true;

                ILOG_INFO("Requesting to kill job " << job->id << " run by "
                          << job->agent << " at port " << port);
                job->state = State::KILL_ATTEMPT;
            }
        }
    }

    // If the specified port was assigned a job, the job will be rescheduled.
    void
    RescheduleJob(int port)
    {
        Job* job = mJobByPort[port];
        if (job)
        {
            if (job->state != State::RUNNING)
                ILOG_ERROR("Unexpected state of job at port " << port << ": "
                           << job->state);
            mJobByPort[port] = 0;
            mJobsWaiting.push_front(job);
            job->state = State::RESCHEDULED;
            job->portUsed = -99; // undefined
            ILOG_INFO("Job " << job->id << " rescheduled for execution");
        }
    }

    bool mPoisoned;
    int mNextJobId;
    std::deque<Job*> mJobsWaiting;
    std::deque<Job*> mJobsDone;
    std::map<int, Job*> mJobByPort;

    ILOG_VAR_DECL;

}; // class

ILOG_VAR_INIT(Server, Impala.Job);

} // namespace 
} // namespace 

#endif

---