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

---

Manager.h

Go to the documentation of this file.

#ifndef Impala_Core_VideoJob_Manager_h
#define Impala_Core_VideoJob_Manager_h

#include <map>
#include <sstream>
#include <fstream>

#include "Basis/ILog.h"
#include "Util/Database.h"
#include "Util/IOBuffer.h"
#include "Job/State.h"
#include "Core/VideoJob/ProcessDefinition.h"
#include "Core/VideoJob/Data.h"
#include "Core/VideoJob/ServerProxy.h"
#include "Core/VideoJob/Reporter.h"

#include "Core/VideoSet/VideoSet.h"
#include "Core/VideoSet/MakeVideoSet.h"

namespace Impala
{
namespace Core
{
namespace VideoJob
{

/***********************************************************************
An instance of this class manages the processing of a video set according
to a process definition, by decomposing work into packages (jobs) 
following from dependencies within the process definition. 
Work packages will be issued for execution to a job server 
and on completion will have their status inspected by the manager. 
This may result in more work packages being. 
The cycle continues until eventually the overall process is complete.
************************************************************************/

class Manager
{
    typedef Job::State::StateType JobStateType;

    typedef Data::VideoMeta VideoMeta;
    typedef Data::VideoJob VideoJob;

public:

    Manager(CString videoSetName, CString processDefinitionFile, 
            CString serverName, int port, CString passwordFile, CmdOptions& options)
    {
        mData.mVideoSetName = videoSetName;
        mData.mServerName = serverName;
        mData.mPort = port;
        mData.mPasswordFile = passwordFile;

        mData.mProcessDefFile = processDefinitionFile;

        mJobServer = 0;
        mCurrentMask = "";
        mProcessDone = false;

        flagVideoUpdated = options.GetString("videoUpdate", "0");

        mReporter = 0;
    }

    virtual 
    ~Manager()
    {
        if (mJobServer)
            delete mJobServer;

        if (mReporter)
            delete mReporter;

        if (mVideoSet)
            delete mVideoSet;
    }

    void 
    SetMaskFileName(CString processStateFile)
    {
        mMaskFileName = processStateFile;
    }

    int 
    Start(int serverRequestIntervalSec, CmdOptions& options)
    {
        ILOG_INFO("Start managing video processing");

        if (!Prepare(options))
            return 0;

        mCurrentStage = mData.mProcessDef->FirstStageNr();
        ILOG_INFO("Entering stage " << mCurrentStage + 1);
        CreateInitialStageJobs();

        mJobServer = new ServerProxy(mData.mServerName, mData.mPort,
                                     mData.mPasswordFile,
                                     serverRequestIntervalSec);
        mProcessDone = false;
        while (!mProcessDone)
        {
            while (HasMoreJobs())
            {
                VideoJob* job = GetNextJob();
                String cmdLine = mData.mProcessDef->GetCmdLine
                    (job->stepNr, mData.mVideoSetName, job->videoNrCompact);
                int attempts = 0;
                int jobId = mJobServer->ScheduleJob(cmdLine, job->videoNr);
                while (jobId <= 0)
                {
                    if (attempts++ < 3)
                    {
                        ILOG_WARN("Job dispatch failed; retrying...");
                        jobId = mJobServer->ScheduleJob(cmdLine, job->videoNr,
                                                        true);
                    }
                    else
                    {
                        ILOG_ERROR("Job dispatch failed");
                        return -3;
                    }
                }
                job->id = jobId;
                job->state = Job::State::SUBMITTED;
                mData.mJobsScheduled[jobId] = job;
            }

            int nrOfJobsCompleted = 0;
            do
            {
                nrOfJobsCompleted = MonitorJobs();
                ReportState();
                if (nrOfJobsCompleted < 0) // SK: lousy trick
                    mProcessDone = true;
            }
            while (nrOfJobsCompleted == 0);
        }

        ReportState();
        ILOG_INFO("Finished managing video processing");
        return 0;
    }


private:

    bool
    Prepare(CmdOptions& options)
    {
        mVideoSet = Core::VideoSet::MakeVideoSet(mData.mVideoSetName);
        mData.mVideoSetSize = mVideoSet->NrFiles();
        ILOG_INFO("Video set contains " << mData.mVideoSetSize <<
                  " video file(s)");
        if (mData.mVideoSetSize <= 0)
        {
            ILOG_WARN("No video files to process");
            return false;
        }

        mData.mProcessDef = new ProcessDefinition(mData.mProcessDefFile);
        if (mData.mProcessDef->StageCount() < 1)
        {
            ILOG_WARN("Empty process definition");
            return false;
        }

        mMaskPath = mVideoSet->GetFilePathVideoData(mMaskFileName, false, true);
        if (mMaskPath.empty())
        {
            ILOG_INFO("Mask file '" << mMaskFileName << "' not found");
            // since the file was not found, set path for the file for writing
            mMaskPath = mVideoSet->GetFilePathVideoData(mMaskFileName, true,
                                                        false);
        }
        else
        {
            Util::Database* db = mVideoSet->GetDatabase();
            Util::IOBuffer* buffer = db->GetIOBuffer(mMaskPath, true, false, "");
            if (buffer)
            {
                mCurrentMask = buffer->ReadLine();
                delete buffer;
                int maskLength = Min<int>(mCurrentMask.size(),
                                          mData.VideoSetSize());
                mCurrentMask = mCurrentMask.substr(0, maskLength);
                if (mCurrentMask.empty())
                    ILOG_INFO("Mask file '" << mMaskFileName << "' is empty");
            }
            else
            {
                ILOG_ERROR("Failed to read mask file '" << mMaskFileName << "'");
                return false;
            }
        }
        mData.mInitialMask = mCurrentMask;

        const int setSize = mData.VideoSetSize();
        for (int v = 0; v < setSize; v++)
        {
            if (! mData.WasInitiallyMaskedOut(v))
            {
                VideoMeta video;
                video.videoNr = v;
                video.discarded = false;
                mData.mVideos[v] = video;
            }
        }

        ILOG_INFO("Ignoring " << mData.mVideoSetSize - mData.mVideos.size() <<
                  " video file(s)");

        if (mData.mVideos.empty())
        {
            ILOG_WARN("No video files to process");
            return false;
        }

        if (!UpdateMaskFile(false))
        {
            ILOG_ERROR("Mask file update test failed");
            return false;
        }

        mReporter = new Reporter(mData, options);
        ReportVideoSet(mVideoSet);
        ReportProcessDefinition();

        WriteVideoSet(FileNameBase(mData.mVideoSetName) + "_bak." + 
                      FileNameExt(mData.mVideoSetName));

        mCompactedVideoSetSize = mData.mVideoSetSize; // must initially have
                                                      // this value
        return true;
    }

    // Returns true if at least one more job is waiting to be run;
    // false otherwise.
    bool 
    HasMoreJobs()
    {
        return (mData.mCountOfWaitingJobs > 0);
    }

    // Returns one job that is waiting to be run;
    // assumes that HasMoreJobs() would return true if called at this point
    VideoJob* 
    GetNextJob()
    {
        int indexOfFirstWaitingJob = mData.mAllJobs.size() -
                                     mData.mCountOfWaitingJobs--;
        return mData.mAllJobs[indexOfFirstWaitingJob];
    }

    // Waits for job state changes
    int
    MonitorJobs()
    {
        int nrOfJobsCompleted = 0;
        std::map<int, JobStateType>& updatedJobs =
            mJobServer->MonitorJobState(mData);
        if (updatedJobs.empty()) // SK: lousy trick
            return -1;
        std::map<int, JobStateType>::iterator iter = updatedJobs.begin();
        for ( ; iter != updatedJobs.end(); iter++ )
        {
            int jobId = iter->first;
            JobStateType newState = iter->second;
            VideoJob* job = mData.mJobsScheduled[jobId];
            if (newState == Job::State::TERMINATED_NRM ||
                newState == Job::State::TERMINATED_ABNRM)
            {
                JobStateType dummy = Job::State::UNDEFINED;
                int exitCode = -99;
                String errorLog;
                int rc = mJobServer->GetJobState(jobId, dummy, exitCode, errorLog);
                if (rc != 0)
                    return -2;
                JobDone(mData.mJobsScheduled[jobId], newState, exitCode);
                nrOfJobsCompleted++;
            }
            else
            {
                job->state = newState;
            }
        }
        return nrOfJobsCompleted;
    }

    void 
    JobDone(VideoJob* job, JobStateType state, int exitCode)
    {
        ILOG_DEBUG("Job " << job->id << " is done (state=" << state <<
                   ", exitCode=" << exitCode << ")");

        job->state = state;
        job->exitCode = exitCode;
        mData.mJobsScheduled.erase(mData.mJobsScheduled.find(job->id));

        if (state == Job::State::TERMINATED_ABNRM || exitCode != 0)
        {
            if (job->isIndiv)
            {
                mData.mVideos[job->videoNr].discarded = true;
            }
            else
            {
                for (int v = 0; v < mData.VideoSetSize(); v++)
                    mData.mVideos[v].discarded = true;
            }
        }
        else
        {
            bool atEndOfStage =
                mData.mProcessDef->IsLastStepOfStage(job->stepNr, mCurrentStage);
            if (! atEndOfStage)
            {
                CreateMoreStageJobs(job);
                return;
            }
        }

        EvaluateStage();
    }

    void
    EvaluateStage()
    {
        int nrOfInvalid = 0;
        for (int v = 0; v < mData.mVideoSetSize; v++)
        {
            if (mData.IsValid(v))
            {
                VideoMeta& video = mData.mVideos[v];
                VideoJob* mostRecentJob = video.jobs.back();
                if (mostRecentJob->state != Job::State::TERMINATED_NRM &&
                    mostRecentJob->state != Job::State::TERMINATED_ABNRM)
                    return; // current stage not complete yet
            }
            else
            {
                nrOfInvalid++;
            }

            // added by Jun Wu, 2009/09/15    
            if (flagVideoUpdated == "1")
            {
                // update once each video is processed
                UpdateMaskFile();
            }
        }

        // ok, we're at end of stage 

        UpdateMaskFile();

        ILOG_INFO("Stage " << mCurrentStage + 1 << " completed");
        if (mCurrentStage + 1 < mData.mProcessDef->StageCount())
        {
            if (nrOfInvalid < mData.VideoSetSize())
            {
                mCurrentStage++;
                ILOG_INFO("Advancing to stage " << mCurrentStage + 1);
                CreateInitialStageJobs();
                return;
            }
            else
            {
                ILOG_DEBUG("Will not advance stage since no valid videos left");
            }
        }

        mProcessDone = true;
    }

    bool
    UpdateMaskFile(bool onlyIfChanged = true)
    {
        std::ostringstream oss;
        for (int v = 0; v < mData.VideoSetSize(); v++)
            oss << (mData.IsValid(v) ? '1' : '0');
        String mask = oss.str();
        if (! (mask == mCurrentMask && onlyIfChanged) )
        {
            Util::Database* db = mVideoSet->GetDatabase();
            Util::IOBuffer* buffer = db->GetIOBuffer(mMaskPath, false, false,
                                                     "dummy_value");
            if (!buffer)
            {
                ILOG_ERROR("Cannot write mask to '" << mMaskPath << "'");
                return false;
            }
            buffer->Write(mask.c_str(), mask.size());
            buffer->Write("\n", 1);
            delete buffer;
            ILOG_INFO("Mask file updated");
            mCurrentMask = mask;
        }
        return true;
    }

    void 
    CreateInitialStageJobs()
    {
        int firstStepOfStage =
            mData.mProcessDef->FirstStepOfStage(mCurrentStage);
        std::vector<String> validVideos;
        if (mData.mProcessDef->IsIndividualStage(mCurrentStage))
        {
            for (int v = 0; v < mData.VideoSetSize(); v++)
            {
                if (mData.IsValid(v))
                {
                    VideoJob* newJob = new VideoJob;
                    newJob->stepNr = firstStepOfStage;
                    newJob->videoNr = v;
                    const int compactedVideoNr = validVideos.size();
                    newJob->videoNrCompact = compactedVideoNr;
                    newJob->isIndiv = true;
                    newJob->state = Job::State::CREATED;
                    mData.mVideos[v].jobs.push_back(newJob);

                    mData.mAllJobs.push_back(newJob);
                    mData.mCountOfWaitingJobs++;

                    validVideos.push_back(mVideoSet->GetAsPath(v));
                }
            }
        }
        else
        {
            VideoJob* newJob = new VideoJob;
            newJob->stepNr = firstStepOfStage;
            newJob->videoNr = -1;
            newJob->videoNrCompact = -1;
            newJob->isIndiv = false;
            newJob->state = Job::State::CREATED;
            for (int v = 0; v < mData.VideoSetSize(); v++)
            {
                if (mData.IsValid(v))
                {
                    mData.mVideos[v].jobs.push_back(newJob);

                    validVideos.push_back(mVideoSet->GetAsPath(v));
                }
            }

            mData.mAllJobs.push_back(newJob);
            mData.mCountOfWaitingJobs++;
        }

        if (validVideos.size() < mCompactedVideoSetSize)
        {
            UpdateVideoSet(validVideos);
            mCompactedVideoSetSize = validVideos.size();
        }
    }

    bool
    WriteVideoSet(CString setName)
    {
        String pathToVideoSet = mVideoSet->GetFilePathVideoData(setName, true,
                                                                false);
        Util::Database* db = mVideoSet->GetDatabase();
        Util::IOBuffer* buffer = db->GetIOBuffer(pathToVideoSet, false, false,
                                                 "temp_videoset.txt");
        if (!buffer)
        {
            ILOG_ERROR("Cannot write video set backup file: " << pathToVideoSet);
            return false;
        }

        for (int v = 0; v < mVideoSet->NrFiles(); v++)
        {
            String line = mVideoSet->GetAsPath(v);
            buffer->Write("\"", 1);
            buffer->Write(line.c_str(), line.size());
            buffer->Write("\"\n", 2);
        }

        delete buffer;
        ILOG_INFO("Video set file backed up: " << pathToVideoSet);

        return true;
    }

    bool
    UpdateVideoSet(std::vector<String> videoSpecs)
    {
        String pathToVideoSet =
            mVideoSet->GetFilePathVideoData(mVideoSet->GetSetName(), true,
                                            false);
        Util::Database* db = mVideoSet->GetDatabase();
        Util::IOBuffer* buffer = db->GetIOBuffer(pathToVideoSet, false, false,
                                                 "temp_videoset.txt");
        if (!buffer)
        {
            ILOG_ERROR("Cannot update video set file: " << pathToVideoSet);
            return false;
        }

        for (int v = 0; v < videoSpecs.size(); v++)
        {
            CString line = videoSpecs[v];
            buffer->Write("\"", 1);
            buffer->Write(line.c_str(), line.size());
            buffer->Write("\"\n", 2);
        }

        delete buffer;
        ILOG_INFO("Video set file updated");

        return true;
    }

    void 
    CreateMoreStageJobs(VideoJob* job)
    {
        if (!job->isIndiv)
            ILOG_ERROR("Stage " << mCurrentStage <<
                       " is not individual and can contain only one step");

        VideoMeta& video = mData.mVideos[job->videoNr];
        VideoJob* newJob = new VideoJob;
        newJob->stepNr = job->stepNr + 1;
        newJob->videoNr = job->videoNr;
        newJob->videoNrCompact = job->videoNrCompact;
        newJob->isIndiv = true;
        newJob->state = Job::State::CREATED;
        video.jobs.push_back(newJob);

        mData.mAllJobs.push_back(newJob);
        mData.mCountOfWaitingJobs++;
    }

    void 
    ReportState()
    {
        mReporter->WriteReportStatus();
    }

    void 
    ReportVideoSet(Core::VideoSet::VideoSet* const videoSet)
    {
        mReporter->WriteVideoSet(videoSet);
    }
    
    void 
    ReportProcessDefinition()
    {
        mReporter->WriteProcessDefinition();
    }
    
    ILOG_VAR_DECL;

    Data mData;

    ServerProxy* mJobServer;
    int mCurrentStage;
    bool mProcessDone;

    String mMaskFileName;
    String mMaskPath;
    String mCurrentMask;

    Reporter* mReporter;

    String flagVideoUpdated;

    Core::VideoSet::VideoSet* mVideoSet;
    int mCompactedVideoSetSize;

}; // class

ILOG_VAR_INIT(Manager, Impala.Core.VideoJob);

} // namespace 
} // namespace 
} // namespace 

#endif

---