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

---

Server.h

Go to the documentation of this file.

#ifndef Impala_IDash_Server_h
#define Impala_IDash_Server_h

#include "Basis/ILog.h"
#include "Util/SimpleMap.h"
#include "Core/VideoJob/ServerProxy.h"
#include "Core/IDash/XmlJob.h"
#include "Core/IDash/XmlJobReference.h"
#include "Core/IDash/Href.h"

namespace Impala
{
namespace Core
{
namespace IDash
{


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

public:

    Server(int port, int nrPorts, CString passwordFile, CString jobServerAddr) 
        : Util::ChannelServer(port, nrPorts, true, passwordFile)
    {
        mJobServer = new VideoJob::ServerProxy(jobServerAddr, passwordFile, 10);
    }

    virtual
    ~Server()
    {
    }

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

        // Could show some statistics here
    }

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, "ScheduleJob", 11) == 0)
            ScheduleJob(buf, len, bufSize, conn, port);
        else if (strncmp(buf, "GetJobs", 7) == 0)
            GetJobs(buf, len, bufSize, conn, port);
        else if (strncmp(buf, "StatusJob", 9) == 0)
            StatusJob(buf, len, bufSize, conn, port);
        else if (strncmp(buf, "DeleteCases", 11) == 0)
            DeleteCases(buf, len, bufSize, conn, port);
        else
            return -1;

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

        return strlen(buf) + 1;
    }

    virtual void
    ScheduleJob(char* buf, int len, int bufSize, CString conn, int port)
    {
        String curCon = ConnectionDescr();
        ILOG_INFO("ScheduleJob from " << curCon);

        if (!mJobServer->IsConnected())
        {
            ILOG_ERROR("Cannot do schedule: No connection to jobserver");
            return;
        }

        if (strncmp(buf, "ScheduleJob:", 12) == 0)
        {
            Util::StringParser parser(String(buf, len));
            parser.Eat("baseUri=");
            String baseUri = parser.GetString(';', true);
            parser.Eat("job=");
            int jobLen = len - parser.Position();
            Util::IOBuffer ioBuf(jobLen);
            memcpy(ioBuf.GetBuffer(), buf + parser.Position(), jobLen);
            XmlJob job("IDashJob", &ioBuf);
            if (!job.Valid())
            {
                ILOG_ERROR("Invalid job request");
                return;
            }

            int dummy;
            if (mJobMap.Get(job.GetId(), dummy))
            {
                ILOG_ERROR("ScheduleJob: jobId already exists");
                return;
            }

            String cmdLine = "do_process ";
            if (job.GetAnnotationSet().empty())
            {
                if (job.GetVideo().empty())
                {
                    cmdLine += "queryset " + job.GetQuerySet();
                }
                else
                {
                    cmdLine += "video " + job.GetVideo() + " " + job.GetQuerySet();
                }
            }
            else
            {
                cmdLine = "do_annotator " + job.GetAnnotationSet();
            }
            int serverJobId = mJobServer->ScheduleJob(cmdLine, 0);
            ILOG_INFO("Scheduled job: xmlJobId=" << job.GetId() <<
                      ", serverJobId=" << serverJobId << ", cmdLine=" <<
                      cmdLine);
            mJobMap.Add(job.GetId(), serverJobId);

            String hRef = baseUri + "jobs/" + job.GetId();
            XmlJobReference jobRef(hRef);
            String jobRefString = jobRef.Export();
            sprintf(buf, "JobAccepted:jobRef=%s\0", jobRefString.c_str());
        }
        else
        {
            ILOG_ERROR("Invalid schedule request: " << String(buf, len));
        }
    }

    virtual void
    GetJobs(char* buf, int len, int bufSize, CString conn, int port)
    {
        String curCon = ConnectionDescr();
        ILOG_INFO("GetJobs from " << curCon);

        if (!mJobServer->IsConnected())
        {
            ILOG_ERROR("Cannot do jobs: No connection to jobserver");
            return;
        }

        if (strncmp(buf, "GetJobs:", 7) == 0)
        {
            Util::StringParser parser(String(buf, len));
            parser.Eat("baseUri=");
            String baseUri = parser.GetString(';', true);

            std::vector<String> jobs = mJobMap.GetAllIdx();
            String s = "<?xml version=\"1.0\" encoding=\"UTF-8\" standalone=\"yes\"?>";
            s += "<jobs xmlns:xsi=\"http://www.w3.org/2001/XMLSchema-instance\" xmlns:xsd=\"http://www.w3.org/2001/XMLSchema\" xmlns:xlink=\"http://www.w3.org/1999/xlink\" xmlns=\"http://www.i-dash.eu/MetaData\" >";
            for (int i=0 ; i<jobs.size() ; i++)
            {
                String url = baseUri + "jobs/" + jobs[i];
                s += "<job xlink:type=\"simple\" xlink:href=\"";
                s += url;
                 s += "\" entityType=\"Job\" />";
            }
            s += "</jobs>";
            sprintf(buf, "Jobs:%s\0", s.c_str());
        }
        else
        {
            ILOG_ERROR("Invalid jobs request: " << String(buf, len));
        }
    }

    virtual void
    StatusJob(char* buf, int len, int bufSize, CString conn, int port)
    {
        String curCon = ConnectionDescr();
        ILOG_INFO("StatusJob from " << curCon);

        if (!mJobServer->IsConnected())
        {
            ILOG_ERROR("Cannot do status: No connection to jobserver");
            return;
        }

        if (strncmp(buf, "StatusJob:job=", 14) == 0)
        {
            Util::StringParser parser(String(buf, len));
            parser.Eat('=');
            String jobId = parser.GetString(';', false);

            int serverJobId;
            if (!mJobMap.Get(jobId, serverJobId))
            {
                ILOG_ERROR("StatusJob: no jobId " << jobId);
                return;
            }

            Job::State::StateType state;
            int exitCode;
            String errorLog;
            mJobServer->GetJobState(serverJobId, state, exitCode, errorLog);
            ILOG_INFO("state = " << Job::State::ToString(state) <<
                      ", exitCode = " << exitCode <<
                      ", errorLog = " << errorLog);

            String status = "initial";
            String progress = "0";
            if (state == Job::State::RUNNING)
            {
                status = "running";
                progress = "50";
            }
            if (state == Job::State::TERMINATED_NRM)
            {
                if (exitCode == 0)
                    status = "completed";
                else
                    status = "error";
                progress = "100";
            }
            if (state == Job::State::TERMINATED_ABNRM)
            {
                status = "error";
                progress = "100";
            }

            XmlJob job(jobId, "", "", status, errorLog, progress);
            String jobString = job.Export();
            sprintf(buf, "JobStatus:job=%s\0", jobString.c_str());
        }
        else
        {
            ILOG_ERROR("Invalid job specification: " << String(buf, len));
        }
    }

    virtual void
    DeleteCases(char* buf, int len, int bufSize, CString conn, int port)
    {
        String curCon = ConnectionDescr();
        ILOG_INFO("DeleteCases from " << curCon);

        if (!mJobServer->IsConnected())
        {
            ILOG_ERROR("Cannot do delete: No connection to jobserver");
            return;
        }

        String cmdLine = "do_clean_vds; do_clean_annotator";
        int serverJobId = mJobServer->ScheduleJob(cmdLine, 0);
        ILOG_INFO("Scheduled job: serverJobId=" << serverJobId <<
                  ", cmdLine=" << cmdLine);
        mJobMap.Clear();
    }

private:

    VideoJob::ServerProxy*      mJobServer;
    Util::SimpleMap<String,int> mJobMap;

    ILOG_VAR_DECL;

};

ILOG_VAR_INIT(Server, Impala.Core.IDash);

} // namespace IDash
} // namespace Core
} // namespace Impala

#endif

---