GBSJob.py

Go to the documentation of this file.

import re
import os
import shutil
import time

import Ganga.GPI

from  AsyncCallThread import timeout_call

from   GBSIdCodes import *
from   GBSLogger  import GetLoggerThreshold, Log, logger 
from   GBSObject  import GBSObject
from   GBSModelRegistry import GetModelRegistry
from   GBSTimeStamp import GBSTimeStamp as timestamp
import GBSUtilities

class GBSJob(GBSObject) :

    """Object to submit, and if necessary resubmit a job until it succeeds or needs user intervention.

    This class is responsible for submitting and if necessary
    resubmitting a job until it is successful or user intervention is
    unavoidible.
      
    """

    ######  GBSObject inherited responsibilities   ###### 
      

    def __init__(self,name,parent,model,model_args):
        self.__tryNumber          = 0  # 0 before first try
        self.__earlyFails         = 0  # Symptomatic of a system failure
        self.__lateFailsHandled   = 0  # Job signalling retry
        self.__lateFailsUnhandled = 0  # Job aborting without signal
        self.__scriptLocalArgs    = "" # Application script local args
        self.__localEnvironment   = "" # The environment that are local to this job.
        self.__localInputSandbox  = "" # The list of input sandbox files that are local to this job.
        self.__localOutputSandbox = "" # The list of output sandbox files that are local to this job.
        
        # statusCode:  See GBSIdCodes
        self.__statusCode         = GID_JSC_NEW
        # Associated text to qualify statusCode. Has specific meaning when
        # __statusCode has the following values which correspond to the values
        # passed back from the application script via the GBS Log File:-
        #
        #  GID_JSC_SUCCEEDED  The list of output files
        #  GID_JSC_FAILED     Diagnostic message about the failure
        #  GID_JSC_RETRY      The retry args for next attempt.
        self.__statusText         = "Ready to run"
        self.__statusTime         = timestamp()
        # The current retry args i.e. as determined from previous try (or empty for first try)
        self.__retryArgs   = ""
        self.__gangaJobId  = -1
        # The remaining members are volatile i.e. not persisted but derived from other state.
        self.__tryID       = name + ".0"
        
        GBSObject.__init__(self,name,parent,model)

    def _DoMemberIO(self,ioh):
        self._SetTryNumber(ioh("Try Number","i",self.__tryNumber))
        self.__retryArgs           = ioh("Current Retry Args","s",self.__retryArgs)
        self.__earlyFails          = ioh("Early Fail Count","i",self.__earlyFails)
        self.__lateFailsHandled    = ioh("Late Handled Fails Count","i",self.__lateFailsHandled)
        self.__lateFailsUnhandled  = ioh("Late Unhandled Fails Count","i",self.__lateFailsUnhandled)
        self.__scriptLocalArgs     = ioh("Script Local Args","s",self.__scriptLocalArgs)
        self.__localEnvironment    = ioh("+Local environment","s",self.__localEnvironment)
        self.__localInputSandbox   = ioh("+Local Input Sandbox List",    "s",self.__localInputSandbox)
        self.__localOutputSandbox  = ioh("+Local Output Sandbox List",   "s",self.__localOutputSandbox)
        self.__statusCode          = ioh("Status Code","i",self.__statusCode)
        self.__statusText          = ioh("Status Text","s",self.__statusText)
        self.__statusTime          = ioh("+Status Time Stamp","s",self.__statusTime)
        self.__gangaJobId          = ioh("+Ganga Job Id","i",self.__gangaJobId)

        # During transition may need to infer __gangaJobId from __statusCode
        if self.__statusCode >= GID_JSC_SUBMITTED: self.__gangaJobId = self.__statusCode

        GBSObject._DoMemberIO(self,ioh)

    def GetType(self): return "GBSJob"

    def __repr__(self): return self.AsString()


#-----------------------------------------------------------------------------------------------
#
#  User Callable Methods Setters
#
#-----------------------------------------------------------------------------------------------

    def AsString(self,level = "Full"):

        """Return string description.

         Return string description at the following levels:-
        "Brief"    one line summary suitable for rows in tables
        "Heading"  one line summary suitable as heading for "Brief"
        "Full"     full desciption including value of every data member"""

        if ( level == "Heading"):
            s  = "Name".ljust(20)
            s += "Status".ljust(23)
            s += "Input".ljust(40)
            s += "Status Details"
            return s
        if ( level == "Brief"):
            s  = self.GetName().ljust(20)
            s += GIDStringForJSC(self.__statusCode).ljust(23)
            s += (self.__scriptLocalArgs + ";" + self.__localEnvironment).ljust(40)
            s += "[" + self.__statusText + "]"
            return s

        s  = GBSObject.__repr__(self) + "\n\n"
        s += "Status: " + GIDStringForJSC(self.__statusCode) +  " [" + self.__statusText + "] at " + self.__statusTime + "\n"
        s += "  Associated Ganga Job ID: " + str(self.__gangaJobId)  + "\n\n"
        s += "Job Definition\n"
        s += "  Script local args: '" + self.__scriptLocalArgs + "'\n"
        s += "  Local environment: '" + self.__localEnvironment + "'\n"
        s += "  Input Sandbox:     '"  + self.__localInputSandbox + "'\n" 
        s += "  Output Sandbox:    '"  + self.__localOutputSandbox + "'\n\n" 
        s += "Retry Status\n"
        s += "  Try:                  "  + str(self.__tryNumber) + "\n"
        s += "  Retry Args:           '" + str(self.__retryArgs) + "'\n"
        s += "  Early Fails:          "  + str(self.__earlyFails) + "\n"
        s += "  Late Handled Fails:   "  + str(self.__lateFailsHandled) + "\n"
        s += "  Late Unhandled Fails: "  + str(self.__lateFailsUnhandled) + "\n"

        # If there exist any output, give a summary, either for the current try, if
        # complete, or the previous one if currently submitted.

        display_try = self.__tryNumber
        if self.__statusCode >= GID_JSC_SUBMITTED: display_try -= 1
        if display_try < 1: return s
        output_dir = self._GetTryOutputDir(display_try)
        s += "\nThe output for try %d can be found in\n\n   %s\n\n and consists of:-\n\n" % (display_try,output_dir)
        list_dir = output_dir + '/../listing.tmp'
        os.system("cd " + output_dir + ";ls -l > " + list_dir)
        f = open(list_dir)
        for line in f: s += "  " + line
        f.close()
        os.remove(list_dir)
        gbs_log_file_name = self._GetGbsLogFileName(display_try)
        gbs_log_file_spec =  str(output_dir) + "/" + str(gbs_log_file_name)
        if not os.path.isfile(gbs_log_file_spec): return s
        s += "\nThe GLF (GBS Log File) %s contains:-\n\n" % gbs_log_file_name
        f = open(gbs_log_file_spec)
        for line in f: s += "  " + line
        f.close()
        return s

    def CanClear(self):
        """Return true if can ClearErrorCounts, ClearHistory and Remove"""
        return not self.IsSubmitted()

    def CanKill(self):
        """Return true if can Kill"""
        return self.IsSubmitted()
   
    def CanSubmit(self):

        """Return true if can submit Ganga job"""

        if      self.__statusCode < GID_JSC_SUBMITTED \
            and self.__statusCode > GID_JSC_CANNOT_SUBMIT \
            and self.GetParent().GetScriptFileSpec(): return True
        return False

    def GetEarlyFailsCount(self):
        """Return Early Fails Count"""
        return self.__earlyFails

    def GetGangaJobId(self):
        """Return associated Ganga job Id (if any) or -1 """
        return self.__gangaJobId

    def GetGangaJob(self):
        """Return associated Ganga job (if any)"""
        if self.__gangaJobId < 0: return None
        try:
            gj = Ganga.GPI.jobs(self.__gangaJobId)
        # I need to catch JobAccessError, but it isn't Ganga.GPI.JobAccessError so just
        # catch everything and hope it is O.K.!
        except: gj = None
        return gj
        
    def GetLateHandledFailsCount(self):
        """Return Late Handled Fails Count"""
        return self.__lateFailsHandled
    
    def GetLateUnhandledFailsCount(self):
        """Return Late Unhandled Fails Count"""
        return self.__lateFailsUnhandled
    
    def GetLocalEnvironment(self,prettyPrint = False):
        """Return, as a comma separated list string, the environment that is local to this job.

        If prettyPrint is True print out list one item per line"""
        if not prettyPrint: return self.__localEnvironment
        print "Local Environment:-"
        GBSUtilities.ListCommaSepList(self.__localEnvironment,"  ")
   
    def GetLocalInputSandbox(self):
        """Return, as a comma separated list string, the input sandbox file list that is local to this job."""
        return self.__localInputSandbox
    
    def GetLocalOutputSandbox(self):
        """Return, as a comma separated list string, the ouput sandbox file list that is local to this job."""
        return self.__localOutputSandbox

    def GetPhaseCode(self):
        """Return phase code. These are broad categories of status code used by task for job statistics."""
        pc = 0
        if self.IsComplete():
            pc = GID_JPC_DONE_NFAIL
            if not self.IsSuccessful(): pc = GID_JPC_DONE_FAIL
        elif self.IsReady():
            pc = GID_JPC_READY_NRETRY 
            if self.GetTryNumber() > 0: pc = GID_JPC_READY_RETRY
        elif self.IsSubmitted():
            pc = GID_JPC_SUBMIT_RUN
            if not self.IsRunning(): pc = GID_JPC_SUBMIT_NRUN
        else:
            pc = GID_JPC_NREADY_HOLD
            if not self.IsHeld(): pc = GID_JPC_NREADY_NHOLD
        return pc

        
    def GetRetryArgs(self):
        """Returm as a string current retry args i.e. as determined from previous try (or empty for first try)"""
        return self.__retryArgs
    
    def GetScriptLocalArgs(self):
        """Return (as a string) the comma list of application script args that are local to this job"""
        return self.__scriptLocalArgs
    
    def GetStatusCode(self):
        """Return status code"""
        return self.__statusCode
    
    def GetStatusText(self):
        """Return status text which qualifies the Status Code"""
        return self.__statusText
    
    def GetStatusTime(self):
        """Return date time when current status code and text were achieved"""
        return self.__statusTime
    
    def GetTryID(self):
        """Return Try ID = Job Name.try number.  Used to log job attempts."""
        return self.__tryID
    
    def GetTryNumber(self):
        """Return Try Number (0 before first try)"""
        return self.__tryNumber
    
    def IsComplete(self):
        """Return true if job is complete (Successful or Failed)"""
        return self.__statusCode <= GID_JSC_COMPLETE
    
    def IsFailure(self):
        """Return true if job is failure"""
        return self.__statusCode == GID_JSC_FAILED

    def IsHeld(self):
        """Return true if job is Held"""
        return self.__statusCode == GID_JSC_HELD
     
    def IsNotReady(self):
        """Return true if job is not ready to submit"""
        return not self.IsReady() 
     
    def IsReady(self):
        """Return true if job is ready to submit"""
        return self.__statusCode > GID_JSC_CANNOT_SUBMIT and self.__statusCode < GID_JSC_SUBMITTED
    
    def IsRunning(self):
        """Return true if job is submitted and associated Ganga Job status is running"""
        if self.__statusCode >= GID_JSC_SUBMITTED and re.search(r"Ganga status:running",self.__statusText): return True
        return False
  
    def IsSubmitted(self):
        """Return true if job is submitted"""
        return self.__statusCode >= GID_JSC_SUBMITTED
   
    def IsSuccessful(self):
        """Return true if job is successful"""
        return self.__statusCode == GID_JSC_SUCCEEDED

#-----------------------------------------------------------------------------------------------
#
#  User Callable Methods Setters
#
#-----------------------------------------------------------------------------------------------


    def Analyse(self,update = True):

        """Perform job termination analysis and optionally apply the results."""

        # Quit if nothing to analyse.
        if self.__statusCode != GID_JSC_WAITING_ANALYSIS: return
        
        analyser = GetModelRegistry().CreateObject(self.GetModel(),"JobAnalyser","Solomon",self)
        analyser.Analyse(self)
        if update:
            analyser.Apply()
            gj = self.GetGangaJob()
            if gj: gj.remove()
            self.__gangaJobId  = -1
            self.GetParent().RefreshJobStats()
            self.Write()

    def ClearErrorCounts(self,warn=True):

        """If allowed, clear error counts, but leave retry history intact 

        Can only applied to jobs that are ready to be submitted or that have failed and
        in this case has the side effect of setting the status back to RETRY"""

        if self.CanClear():
            self.__earlyFails         = 0
            self.__lateFailsHandled   = 0
            self.__lateFailsUnhandled = 0
            if self.__statusCode == GID_JSC_FAILED:
                self._SetStatus(GID_JSC_RETRY,"Retrying after user cleared errors")
            self.GetParent().RefreshJobStats()
            self.Write()
            return
        if warn: print "Cannot ClearErrorCounts on job %s; its status is %s" % (self.GetName(),GIDStringForJSC(self.__statusCode))
        
    def ClearHistory(self,confirm=True,warn=True):

        """Completely clear all processing history so that processing begins again from scratch.

        The only processing state that is retained is that if job was held it will still be."""
        
        if not self.CanClear():
            if warn: print "Cannot ClearHistory on job %s; its status is %s" % (self.GetName(),GIDStringForJSC(self.__statusCode))
            return
        if confirm:
            ans = raw_input("Are you sure you want to clear the history of job " + self.GetName() + "  ?y[n]")
            if not re.search(r"^y",ans,re.I):
                print "History not cleared."
                return
        child_dir = self.GetStoreLocation("child_dir")
        if os.path.isdir(child_dir):
            if shutil.rmtree(child_dir):
                Log(self,logger.ERROR,"Failed to remove obsolete job history directory:" + child_dir)
        self._SetTryNumber(0)
        self.__retryArgs     = ""
        self.__gangaJobId    = -1
        if not self.IsHeld():
            self._SetStatus(GID_JSC_NEW,"Ready to run")
        self.ClearErrorCounts()
        return
        
    def Hold(self,warn=True):

        """Hold job, so that it won't be submitted.  Warn, if requested, if job not suitable for holding"""

        if self.IsReady():
             self._SetStatusCode(GID_JSC_HELD,"Held by user")
             self.GetParent().RefreshJobStats()
             self.Write()
             return
        if warn: print "Cannot HOLD job %s; its status is %s" % (self.GetName(),GIDStringForJSC(self.__statusCode))

    def Kill(self,warn=True):

       """Kill job that has been submitted to Ganga.  Warn, if requested, if job not suitable for killing"""

       if not self.CanKill():
           if warn: print "Cannot KILL job %s; its status is %s" % (self.GetName(),GIDStringForJSC(self.__statusCode))
           return
       try:
            gj = Ganga.GPI.jobs(self.__statusCode)
        # I need to catch JobAccessError, but it isn't Ganga.GPI.JobAccessError so just
        # catch everything and hope it is O.K.!
       except:
           Log(self,logger.ERROR,self.__tryID + " lost Ganga job with ID " + str(self.__statusCode))
           return         
       # Kill job
       gj.kill()
       Log(self,logger.ERROR,self.__tryID + " killed by user")
       self.UpdateStatus()

    def Release(self,warn=True):

        """Release job, so that it can be submitted.  Warn, if requested, if job not suitable for releasing"""

        if self.IsHeld():
             self.__statusCode = GID_JSC_NEW
             if  self.__tryNumber: self.__statusCode = GID_JSC_RETRY
             self._SetStatusText("Ready to run")
             self.GetParent().RefreshJobStats()
             self.Write()
             return
        if warn: print "Cannot RELEASE job %s; its status is %s" % (self.GetName(),GIDStringForJSC(self.__statusCode))

    def Remove(self,confirm=True,warn=True):

        """Completely remove job."""
        
        if not self.CanClear():
            if warn: print "Cannot Remove job %s; its status is %s" % (self.GetName(),GIDStringForJSC(self.__statusCode))
            return
        if confirm:
            ans = raw_input("Are you sure you want to remove job " + self.GetName() + "  ?y[n]")
            if not re.search(r"^y",ans,re.I):
                print "Not removed."
                return
        child_dir = self.GetStoreLocation("child_dir")
        if os.path.isdir(child_dir):
            if shutil.rmtree(child_dir):
                Log(self,logger.ERROR,"Failed to remove obsolete job history directory:" + child_dir)
        state_file = self.GetStoreLocation("self")
        if os.remove(state_file):
            Log(self,logger.ERROR,"Failed to remove state file:" + state_file)
        self._SetTryNumber(0)
        # Sneak into parent task's job list and remove entry.
        del self.GetParent()._jobManagers[self.GetName()]
        self.GetParent().RefreshJobStats()
        return
        
    def SetLocalEnvironment(self,env_str):

        """Set, as a comma separated list string, the environment that local to this job..

      e.g.  job.SetLocalEnvironment('var1=123,var2=a string with spaces,var3=456'

      If the first character is + e.g. +var3=567,var4=b, append to existing environment rather than replace it)"""

        if re.search(r"^\+",env_str):
            s = self.__localEnvironment
            if s: s += ','
            s += env_str[1:]
            # Force local method to be recalled rather than any inherited one
            # MinosRSMJob internally uses append and if that got recalled get a recursive loop.
            GBSJob.SetLocalEnvironment(self,s)
            return
        d =  GBSUtilities.ParseEnvStr(env_str)
        if env_str and not d: print "Cannot parse environment string: '" + env_str + "'"
        else:
            self.__localEnvironment = GBSUtilities.BuildEnvStr(d)
            self.Write()

    def SetLocalInputSandbox(self,in_sbox_str):

        """Set, as a comma separated list string, the input sandbox file list that is local to this job.

      e.g.  task.SetLocalInputSandbox('/home/users/west/my_input_data.dat,../my_script.sh')"""

        (ok,str) = GBSUtilities.ProcessSandboxSetup(self,in_sbox_str,"input")
        if ok:
            self.__localInputSandbox = str
            self.Write()

    def SetLocalOutputSandbox(self,out_sbox_str):

        """Set, as a comma separated list string, the output sandbox file list that is local to this job.

      e.g.  task.SetLocalOutputSandbox('my_output_data.dat,my_output.log')"""

        (ok,str) = GBSUtilities.ProcessSandboxSetup(self,out_sbox_str,"output")
        if ok:
            self.__localOutputSandbox = str
            self.Write()

    def SetScriptLocalArgs(self,arg_str):

        """Set (as a string) the comma separated list of application script args that are local to this job.

      e.g.  job.SetScriptLocalArgs('123,a string with spaces,456')"""

        self.__scriptLocalArgs = arg_str
        self.Write()

    def Submit(self,Perusable=False,\
               MonitorFrequency=0,\
               MonitorCommand="ps -o pid,ppid,rss,vsize,pcpu,pmem,cmd -u $USER"):

        """Submit job if permitted and return True if successful."""

        my_manager = self.GetParent()

        #  Get the user application script
        script_spec = my_manager.GetScriptFileSpec()
        if not script_spec:
            print "Cannot submit job, no user application script assigned to Task '" + my_manager.GetName() + "'"
            return False
        script_name = my_manager.GetScriptFileName()

        if not self.CanSubmit():
            print "Cannot submit job " + self.GetName() + ":" \
                  " Status: " + GIDStringForJSC(self.__statusCode) \
                  + " [" + self.GetStatusText() + "]"
            return False

        if not my_manager.IsAuthorisedToSubmit(): return False

        # Increment try number and prepare output directory to receive results
        
        self._SetTryNumber(self.__tryNumber + 1)
        self.MakeChildDirectory()
        output_dir = self._GetTryOutputDir()
        # It ought not to exist already, but in case it is, remove it first.
        if os.path.isdir(output_dir):
            Log(self,logger.SYNOPSIS,"Removing obsolete directory for job output:" + str(output_dir))
            if shutil.rmtree(output_dir):
                Log(self,logger.ERROR,"Failed to remove obsolete directory for job output:" + str(output_dir))
        Log(self,logger.SYNOPSIS,"Creating directory for job output:" + str(output_dir))
        if os.mkdir(output_dir,0755):
            Log(self,logger.ERROR,"Failed to create directory for job output:" + str(output_dir))

        # Prepare GBS Log File in output directory
        gbs_log_file_name = self._GetGbsLogFileName()
        gbs_log_file_spec =  str(output_dir) + "/" + str(gbs_log_file_name)
        os.system("echo " + timestamp() + " INFO GBS_JOB_SUBMIT submitting job > " + gbs_log_file_spec)

        # Prepare arg list: GLF file, appl. script, global args, local args.
        args = []
        args.append(gbs_log_file_name)
        args.append(script_name)

        # Collect up the application script args, taking account of ecaped commas. 
        script_args = []
        for arglist in [my_manager.GetScriptGlobalArgs(),self.GetScriptLocalArgs()]:
            if arglist: GBSUtilities.ParseCommaSepList(arglist,script_args)

        # Passing args with anything but alphanumeric data into the
        # application script is a pain.  The logical approach, placing
        # each arg as a separate element in 'args' works for the local
        # back-end but is broken for LCG where all the args are first
        # interpolated and then re-parsed to define the argument list
        # that gets passed into the script.  Any string containing
        # white space gets broken apart and anything containing "(" causes
        # a syntax error! So, instead we manufacture the wrapper
        # script on the fly that will call the application script 
        # which gives us total control over the way arguments get passed in.

        wrapper_file_spec =  str(output_dir) + "/" + "gbs_job_wrapper.sh"

        #Copy the leading part of the wrapper
        os.system("cp $GBS_HOME/python/gbs_job_wrapper_part_1.sh " + wrapper_file_spec)

        wrapper  = file(wrapper_file_spec,'a')

        #If monitoring is required, set up and start running the monitor
        if MonitorFrequency:
            wrapper.write('monitor_script=$GBS_WORK_DIR/gbs_monitor.sh\n')
            wrapper.write('if [ -f $monitor_script ] ; then rm -f $monitor_script; fi\n')
            wrapper.write('echo "call_num=0"                                                         >> $monitor_script\n')
            wrapper.write('echo "while [ 1 = 1 ]"                                                    >> $monitor_script\n')
            wrapper.write('echo "do"                                                                 >> $monitor_script\n')
            wrapper.write('echo "  echo \`date \'+%F %H:%M:%S \'\` Monitoring call number \$(( ++call_num ))" >> $monitor_script\n')
            wrapper.write('echo "  %s"                                                               >> $monitor_script\n' %\
                          MonitorCommand)
            wrapper.write('echo "  sleep %d"                                                         >> $monitor_script\n' % \
                          MonitorFrequency)
            wrapper.write('echo "done"                                                               >> $monitor_script\n')
            wrapper.write('chmod +x $monitor_script\n')
            wrapper.write('$GBS_LOG INFO Starting monitoring script at frequency %d secs to execute %s\n' %\
                          (MonitorFrequency,MonitorCommand))
            wrapper.write('$monitor_script &\n')

        # Manufacture the bit where the arguments get passed in
        num_arg  = 0
        arglist  = ""
        for arg in script_args:
            num_arg += 1
            wrapper.write("arg" + str(num_arg) + "=\"" + arg + "\"\n")
            arglist += "\"$arg" + str(num_arg) +"\" "
        wrapper.write("/usr/bin/time --format 'CPU User: %U sec, System: %S sec.  Elapse: %E. ' --output=$GBS_WORK_DIR/time_output_$$ "\
                      + "./$user_script " + arglist + "\n")
        wrapper.close()

        #Copy the trailing part of the wrapper
        os.system("cat $GBS_HOME/python/gbs_job_wrapper_part_2.sh >>" + wrapper_file_spec)

        # Prepare the GBS environment
        env = {}
        env['GBS_MODE'] = my_manager.GetMode()
        env['GBS_RETRY_COUNT'] = str(self.__tryNumber -1)
        retry_arg_no = 0
        for retry_arg in self.__retryArgs.split():
            retry_arg_no += 1
            env['GBS_RETRY_ARG_' + str(retry_arg_no)] = retry_arg
        env['GBS_NUM_RETRY_ARGS'] = str(retry_arg_no)

        # Prepare the user environment
        for env_str in [my_manager.GetGlobalEnvironment(),self.GetLocalEnvironment()]:
            if not env_str: continue
            for var,val in GBSUtilities.ParseEnvStr(env_str).iteritems():
                env[var] = val
        
        # Prepare the executable
        exe = Ganga.GPI.Executable(exe = Ganga.GPI.File(wrapper_file_spec), env = env,args = args)

        # Prepare sandboxes
        inputsandbox = []
        inputsandbox.append(script_spec)
        inputsandbox.append(gbs_log_file_spec)
        env_cmds = ""
        for e_name,e_value in env.iteritems():
           env_cmds += e_name + "=" + str(e_value) + ";" 
        for obj in [my_manager,self]:
            sb_file_list = []
            if obj is self: GBSUtilities.ParseCommaSepList(self.GetLocalInputSandbox(),sb_file_list)
            else:           GBSUtilities.ParseCommaSepList(my_manager.GetGlobalInputSandbox(),sb_file_list)
            input_dir = obj.GetStoreLocation("child_dir") + "/InputSandbox"
            for sb_file in sb_file_list: inputsandbox.append(input_dir + "/" + sb_file)
        outputsandbox = ['gbs_output_sandbox.tar.gz']
        outputsandbox.append(gbs_log_file_name)
        for obj in [my_manager,self]:
            sb_file_list = []
            if obj is self: GBSUtilities.ParseCommaSepList(self.GetLocalOutputSandbox(),sb_file_list)
            else:           GBSUtilities.ParseCommaSepList(my_manager.GetGlobalOutputSandbox(),sb_file_list)
            for sb_file in sb_file_list:
                if sb_file[0] == "$":
                    inp = os.popen(env_cmds + sb_file[1:])
                    sb_file =  inp.read()
                    inp.close()
                outputsandbox.append(sb_file)

        # Prepare backend
        backend = my_manager.GetBackend()
        queue   = ""
        mo = re.search(r"(.*?):(.*)",backend)
        if mo:
            (backend,queue) = mo.groups()
            if backend == "Local": queue   = ""
        
        # Prepare the job
        gj               = Ganga.GPI.Job(backend=backend)
        Ganga.GPI.jobtree.add(gj,my_manager.GetGangaTreeDir())
        if queue :
            if backend == "PBS": gj.backend.queue = queue
            if backend == "LCG": gj.backend.CE    = queue
        # If GLITE_ENABLE is enabled select the correct middleware
        if backend == "LCG" and Ganga.GPI.config['LCG']['GLITE_ENABLE']:
            gj.backend.middleware = 'GLITE'
            if Perusable: gj.backend.perusable = True
        elif Perusable: print "Cannot select perusable; backend not LCG/GLITE"
        gj.application   = exe
        gj.inputsandbox  = inputsandbox
        gj.outputsandbox = outputsandbox
        if GetLoggerThreshold() <= logger.SYNOPSIS: print "Contents of Ganga job:-\n\n" + str(gj)
        try:
            print "Submitting ",self.GetName()
            timeout_call(self,gj.submit,3*60)
            Log(self,logger.INFO,self.__tryID + " submitted")
        except Exception,inst:
             excp_msg = "Caught exception:" + str(inst) +" Failed to complete Ganga job submit"
             Log(self,logger.ERROR,excp_msg)
             self._SetStatusText(excp_msg)
             self._SetTryNumber(self.__tryNumber - 1)
             return False
        self.__gangaJobId = gj.id
        self._SetStatus(gj.id,"Ganga status:" + gj.status)
        self.GetParent().RefreshJobStats()
        self.Write()
        return True
   

    def UpdateStatus(self):
        
        """Get latest status.  May involve checking Ganga job and retrieving output."""

        if not self.IsSubmitted(): return
        try:
            gj = Ganga.GPI.jobs(self.__statusCode)
        # I need to catch JobAccessError, but it isn't Ganga.GPI.JobAccessError so just
        # catch everything and hope it is O.K.!
        except:
            gj = None

        # Deal with lost Ganga job, ought not to happen, but then lots of things in life ought not to happen.
        if not gj:
            Log(self,logger.ERROR,self.__tryID + " lost Ganga job with ID " + str(self.__statusCode))
            # Treat as if status has reverted to "new"
            g_status = 'new'
        else:
            g_status = gj.status

        # Deal with cases when job hasn't finished
        if     g_status == "completing" \
           or  g_status == "ready"      \
           or  g_status == "running"    \
           or  g_status == "scheduled"  \
           or  g_status == "submitted"  \
           or  g_status == "waiting":
            self._SetStatusText("Ganga status:" + g_status)

            # If Ganga job does not appear stalled, just record any change in status.
            t_stall_hours = (time.time() \
                             - time.mktime(time.strptime(self.__statusTime.strip(),"%Y-%m-%d %H:%M:%S")))/3600.
            # Set a 3 hour time limit except for scheduled and running where 3 days is allowed.
            t_timeout_hours = 3.
            if g_status == "scheduled" or  g_status == "running": t_timeout_hours = 3.*24.
            if t_stall_hours < t_timeout_hours:
                self.GetParent().RefreshJobStats()
                self.Write()
                return
            # Ganga job has stalled; kill it and treat as new.
            self._SetStatusText("killed Ganga job %d stalled for %5.1f hours " % (gj.id,t_stall_hours))
            Log(self,logger.ERROR,self.__tryID + " " + self.__statusText)
            gj.remove()
            g_status = 'new'

        # If status has reverted to "new" submit never took place
        if g_status == 'new':
            self._SetTryNumber(self.__tryNumber - 1)
            self.__statusCode = GID_JSC_NEW
            self.__gangaJobId  = -1
            if self.__tryNumber > 1: self.__statusCode = GID_JSC_RETRY
            self._SetStatusText("Ready to run")
            self.GetParent().RefreshJobStats()
            self.Write()
            return


        # For all other cases move over output data, unpack gbs_output_sandbox.tar.gz (if present) and save Ganga status
        output_dir = self._GetTryOutputDir()
        Log(self,logger.SYNOPSIS,"Retrieving output from " + str(gj.outputdir) + " into " + str(output_dir))
        if not os.listdir(gj.outputdir):
           Log(self,logger.ERROR,self.__tryID + " No output files in " + str(gj.outputdir))
        elif os.system("cp " + str(gj.outputdir) + "* " + str(output_dir)):
            Log(self,logger.ERROR,self.__tryID + " Failed to retrieve output from " + str(gj.outputdir) + " into " + str(output_dir))
        try:
            if gj.backend.perusable:
                print "Job is perusable, attempting to recover stdout, as stdout.perusable"
                cmd = "glite-wms-job-perusal --get -f stdout --all --noint " \
                      + gj.backend.id + " > " + output_dir + "/stdout.perusable"
                if os.system(cmd):
                    Log(self,logger.ERROR,self.__tryID + " Failed to retrieve persuable output using ",cmd)
        except: pass
        gbs_output_sandbox = output_dir + "/gbs_output_sandbox.tar.gz"
        if os.path.isfile(gbs_output_sandbox):
            Log(self,logger.SYNOPSIS,"Unpacking " + gbs_output_sandbox)
            if os.system("cd " + output_dir + "; tar zxf gbs_output_sandbox.tar.gz"):
                Log(self,logger.ERROR,self.__tryID + " Failed to unpack " + gbs_output_sandbox)
            else:
                os.remove(gbs_output_sandbox)
        gangaStatusFile = output_dir + "/gbs_ganga.status"
        f = open(gangaStatusFile,"w")
        f.write(str(gj))
        f.close()
        self.__statusCode = GID_JSC_WAITING_ANALYSIS
        try:    g_status += " [" + gj.backend.reason + "]"
        except: pass
        self._SetStatusText("Waiting to analyse.  Ganga exit status was " + g_status)

        # Now analyse the results and decide what happens next
        self.Analyse()
        # No need to write self, the Analyse() call will have done that.

    def WaitForJob(self,num_tries=100,time_interval=30):

        """Wait for running job to end before returning.

        Limit to 'num_tries' with a sleep of 'time_interval' between."""
        if not self.IsSubmitted():
            print "Cannot wait for job %s; its status is %s" % (self.GetName(),GIDStringForJSC(self.__statusCode))
            return
        try_num = 0
        current_state = ""
        print "Waiting for job %s to end" % self.GetName()
        while ( try_num < num_tries ):
            try_num += 1
            self.UpdateStatus()
            new_state = "%s [%s]" % (GIDStringForJSC(self.GetStatusCode()),self.GetStatusText())
            if current_state != new_state:
                print timestamp() + " Job %s status now %s" % (self.GetName(),new_state)
                current_state = new_state
                if not self.IsSubmitted(): return
            time.sleep(time_interval)
        print "Giving up waiting for job to end"

        
    ######  Private Methods (not user callable)  ######

    def _GetTryOutputDir(self, try_req = 0):

        """Return the file spec for the current try (or supplied try) output directory"""

        try_use = self.__tryNumber
        if try_req > 0: try_use = try_req
        return self.GetStoreLocation("child_dir") + "/try_" + str(try_use).zfill(3)


    def _GetGbsLogFileName(self, try_req = 0):
        
        """Return the name of the GBS Log File for the current try (or supplied try)."""
        try_use = self.__tryNumber
        if try_req > 0: try_use = try_req
        return "gbs_" + self.GetParent().GetName() + "_" + self.GetName() +  "_" + str(try_use) + ".log"

    def _SetTryNumber(self, try_no):

        """Update try number and associated tryID."""
        self.__tryNumber = try_no
        self.__tryID     = "JTID:%s.%d" % (self.GetName(),self.__tryNumber)

    #  Methods used by JobAnalyser to update 
    def _IncrementEarlyFailsCount(self):         self.__earlyFails += 1
    def _IncrementLateHandledFailsCount(self):   self.__lateFailsHandled += 1
    def _IncrementLateUnhandledFailsCount(self): self.__lateFailsUnhandled += 1
    def _SetRetryArgs(self,value):               self.__retryArgs = value

    # Methods that record state changes.  These also update state time stamp and add
    # entry to GLF if it exists.
    def _SetStatus(self,code,text):
        if self.__statusCode == code and self.__statusText == text: return
        self.__statusCode = code
        self.__statusText = text
        self.__statusTime = timestamp()
        gbs_log_file_spec = self._GetTryOutputDir() + "/" + self._GetGbsLogFileName()
        if os.path.isfile(gbs_log_file_spec):
            os.system("echo '" + timestamp() + "INFO State change " + GIDStringForJSC(self.__statusCode) \
                      +  " [" + self.__statusText + "]' >> " + gbs_log_file_spec)

    def _SetStatusCode(self,code): self._SetStatus(code,self.__statusText)
    def _SetStatusText(self,text): self._SetStatus(self.__statusCode,text) 

---