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

---

Manager4Posix.h

Go to the documentation of this file.

#ifndef Impala_Process_Manager4Posix_h
#define Impala_Process_Manager4Posix_h

#include <map>
#include <strstream>

//Posix specific includes/////
#include <sys/types.h>
#include <sys/time.h>
#include <sys/wait.h>
#include <sys/resource.h>
#include <unistd.h>
#include <signal.h>
#include <errno.h>
#include <string.h>

#include "Basis/ILog.h"
#include "Basis/String.h"
#include "Basis/StringList.h"

#include "Process/Manager.h"

namespace Impala
{
namespace Process
{

/**************************************************************
 Instances of this class are able of starting and 
 controlling processes on Posix platforms.

 Note that this class is in itself not thread-safe!
***************************************************************/

class Manager4Posix : public Manager
{
    typedef struct {int stateCode; int exitCode;} ProcessData;

    ILOG_VAR_DECL;
    std::map<int, ProcessData> mProcesses;

public:

    Manager4Posix()
    {
        ILOG_DEBUG("Creating instance");
    }

    virtual ~Manager4Posix()
    {
        std::map<int, ProcessData>::iterator iter = mProcesses.begin();
        while (iter != mProcesses.end())
        {
            int pid = iter->first;
            Kill(pid);
            iter++;
        }
        ILOG_DEBUG("Destroying instance");
    }

    int Create(const std::string& cmdLine, std::string application)
    {
        ILOG_INFO("Create [" << cmdLine << "][" << application << "]");
        pid_t pid = fork();//first fork the process
        if(pid==-1)
        {
            //Handle Error, we couldn't fork
            //Use thread safe call(strerror_r) to get error string
            char buff[256];
            if( strerror_r( errno, buff, 256 ) == 0 ) {
                ILOG_ERROR("Could not fork process"<<buff);
            }
    
        }
        else if(pid==0)
        {
            //This is the child process, we should execute the new program
            //inside this process
            //call exec()
            ILOG_INFO("Child Process Running");
            
            //POSIX WAY FOR LIMITING CPU TIME
            //
            //Set the max for CPU Time
            //struct rlimit rlim;
            //rlim.rlim_cur=rlim.rlim_max = maxSomething;
            //setrlimit(RLIMIT_CPU,&rlim);

            
            StringList argList(cmdLine,' ');
            int argc=argList.size()+1;
            if (argc < 2)
            {
                ILOG_ERROR("The cmd line must contain at least the name of the application to run");
                abort();
            }

            char** argv;
            argv = new char*[argc];

            int a=0;
            StringList::const_iterator i=argList.begin();
            std::string application = *i;
            for ( ; i!=argList.end(); i++)
            {
                argv[a] = new char[(*i).length() + 1];
                sprintf(argv[a],(*i).c_str());
                a++;

            }
            argv[a] = (char*)NULL;
            
            execvp(application.c_str(),argv);
            //exec never returns, if it did, there was an error
            char buff[256];
            if( strerror_r( errno, buff, 256 ) == 0 ) {
                ILOG_ERROR("Could not start application"<<buff);
            }
            abort();
        }
        else
        {
            //We have the child process id in pid, store it for further control
            //of the process
            ILOG_INFO("The process is forked, child pid is: "<<pid);
            ProcessData data;
            data.stateCode = STATE_RUNNING;
            mProcesses[pid]  = data;
            return pid;
        }

    }

    bool 
    IsRunning(int pid)
    {
        //Check only if we think it's still running
        if(mProcesses[pid].stateCode == STATE_RUNNING)
        {
            //Check if the process still exists
            int status;
            int ret=waitpid(pid, &status, WNOHANG);
            //ret will be 0 if process is running
            if(ret==0)
                return true;
            else
            {
                SetExitCode(pid,status,ret);
                return false;
            }
        }
        else//Already finished, and we have collected the exit code before
            return false;
    }

    int GetManagedProcessCount() const
    {
        return mProcesses.size();
    }

    bool IsTerminated(int pid)
    {
        return !IsRunning(pid);
    }

    void Sleep(int sec)
    {
        sleep(sec); //seconds
    }

    //int WaitFor(int pid)
    //{
    //    
    //    
    //    //This can as well be handled by setting the maximum time limits
    //    //through setrlimits, but left for now for consistency, 
    //    //should be discussed later.
    //    //
    //    //The following wait will be enough after setting the limits in the
    //    //creation of the process
    //    //
    //    //Wait for the process to finish
    //    /*
    //    int status;
    //    int ret=waitpid(pid, &status, 0);
    //    SetExitCode(pid,status,ret);
    //    return mProcesses[pid].stateCode;
    //    */
    //    
    //    
    //    //an implementation similar to windows version, polls the process
    //    //NOTE:the process times are for ALL the CHILD PROCESSES.
    //    while (mProcesses[pid].stateCode == STATE_RUNNING)
    //    {
    //        if (IsRunning( pid ))
    //        {
    //            Sleep(Manager::WAIT_INTERVAL_SEC);

    //            /*
    //            struct rusage buf;
    //            if(getrusage(RUSAGE_CHILDREN, &buf)==0)
    //            {
    //                //for system time use buf.ru_stime
    //                if(buf.ru_utime.tv_sec>mProcesses[pid].max)
    //                    Kill(pid);
    //                else
    //            
    //                    Sleep(1);
    //            }
    //            else
    //            {
    //                char buff[256];
    //                if( strerror_r( errno, buff, 256 ) == 0 ) {
    //                    ILOG_ERROR("failed to retrieve running times::"<<buff);
    //                }
    //                Kill(pid);
    //            }*/
    //        }
    //        else
    //        {
    //            //int ec = GetExitCode(winHandle, handle);
    //            //if (ec == 0)
    //            //    mProcesses[handle].stateCode = STATE_TERM_SUCCESS;
    //            //else
    //            //{
    //            //    ILOG_ERROR("process has unexpected exit code: " << ec);
    //            //    mProcesses[handle].stateCode = STATE_TERM_ERROR;
    //            //}

    //            int ec = GetExitCode(pid);
    //            if (ec == 0)
    //                mProcesses[pid].stateCode=STATE_TERM_SUCCESS;
    //            else
    //            {
    //                ILOG_ERROR("process has unexpected exit code: " << ec);
    //                mProcesses[pid].stateCode=STATE_TERM_ERROR;
    //            }
    //        }
    //    }
    //    return mProcesses[pid].stateCode;
    //}

    void Kill(int pid)
    {
        if (IsRunning(pid))
            kill(pid,SIGTERM);
    }

    void Remove(int pid)
    {
        Kill(pid);
        mProcesses.erase(pid);
    }

    int GetState(int pid)
    {
        if (IsRunning(pid))
            return STATE_RUNNING;

        return mProcesses[pid].stateCode;
    }

    int GetExitCode(int pid)
    {
        if (IsRunning(pid))
        {
            ILOG_ERROR("Process is still running; exit code not available (yet)");
            throw "process is still running";
        }
        return mProcesses[pid].exitCode;
    }

private:


    void 
    SetExitCode(int pid,int status,int ret)
    {
        //If ret==pid, process finished
        if(ret==pid)
        {
            if(WIFEXITED(status))//Process exited normally
            {   
                ILOG_DEBUG("Process exited normally!");
                
                //Collect the exit and termcodes
                mProcesses[pid].exitCode = WEXITSTATUS(status);

                if(WEXITSTATUS(status)==0)
                    mProcesses[pid].stateCode = STATE_TERM_SUCCESS;
                else
                    mProcesses[pid].stateCode = STATE_TERM_ERROR;
            }
            else if(WIFSIGNALED(status)) //Process has been signalled
            {   
                ILOG_ERROR("Process received a signal!");
                //Add the signal number as exit code.
                mProcesses[pid].stateCode = STATE_TERM_ABNORMAL;
                mProcesses[pid].exitCode = WTERMSIG(status);
            }
        }else{//ret==-1 ERROR
            //This will happen when the process exited already, and this is
            //reported by a previous call to waitpid.
            //Since SetExitCode should set the stateCode to something other than
            //STATE_RUNNING, we should only fall here if there was some
            //other error, please see man 2 waitpid
            //
            char buff[256];
            if( strerror_r( errno, buff, 256 ) == 0 ) {
                ILOG_ERROR("waitpid failed:"<<buff);
            }
        }
    }


}; // class

ILOG_VAR_INIT(Manager4Posix, Impala.Process);

} // namespace 
} // namespace 

#endif

---