CRAB/python/cms_cmssw.py

from JobType import JobType
from crab_logger import Logger
from crab_exceptions import *
from crab_util import *
import math
import common
import PsetManipulator  

import DBSInfo_EDM
import DataDiscovery_EDM
import DataLocation_EDM
import Scram

import os, string, re

class Cmssw(JobType):
    def __init__(self, cfg_params, ncjobs):
        JobType.__init__(self, 'CMSSW')
        common.logger.debug(3,'CMSSW::__init__')

        self.analisys_common_info = {}
        # Marco.
        self._params = {}
        self.cfg_params = cfg_params

        # number of jobs requested to be created, limit ojb splitting
        self.ncjobs = ncjobs

        log = common.logger
        
        self.scram = Scram.Scram(cfg_params)
        scramArea = ''
        self.additional_inbox_files = []
        self.scriptExe = ''
        self.executable = ''
        self.tgz_name = 'default.tgz'


        self.version = self.scram.getSWVersion()
        self.setParam_('application', self.version)
        common.analisys_common_info['sw_version'] = self.version
        ### FEDE
        common.analisys_common_info['copy_input_data'] = 0
        common.analisys_common_info['events_management'] = 1

        ### collect Data cards
        try:
            tmp =  cfg_params['CMSSW.datasetpath']
            log.debug(6, "CMSSW::CMSSW(): datasetPath = "+tmp)
            if string.lower(tmp)=='none':
                self.datasetPath = None
                self.selectNoInput = 1
            else:
                self.datasetPath = tmp
                self.selectNoInput = 0
        except KeyError:
            msg = "Error: datasetpath not defined "  
            raise CrabException(msg)

        # ML monitoring
        # split dataset path style: /PreProdR3Minbias/SIM/GEN-SIM
        if not self.datasetPath:
            self.setParam_('dataset', 'None')
            self.setParam_('owner', 'None')
        else:
            datasetpath_split = self.datasetPath.split("/")
            self.setParam_('dataset', datasetpath_split[1])
            self.setParam_('owner', datasetpath_split[-1])

        self.setTaskid_()
        self.setParam_('taskId', self.cfg_params['taskId'])

        self.dataTiers = []

        ## now the application
        try:
            self.executable = cfg_params['CMSSW.executable']
            self.setParam_('exe', self.executable)
            log.debug(6, "CMSSW::CMSSW(): executable = "+self.executable)
            msg = "Default executable cmsRun overridden. Switch to " + self.executable
            log.debug(3,msg)
        except KeyError:
            self.executable = 'cmsRun'
            self.setParam_('exe', self.executable)
            msg = "User executable not defined. Use cmsRun"
            log.debug(3,msg)
            pass

        try:
            self.pset = cfg_params['CMSSW.pset']
            log.debug(6, "Cmssw::Cmssw(): PSet file = "+self.pset)
            if (not os.path.exists(self.pset)):
                raise CrabException("User defined PSet file "+self.pset+" does not exist")
        except KeyError:
            raise CrabException("PSet file missing. Cannot run cmsRun ")

        # output files
        try:
            self.output_file = []

            tmp = cfg_params['CMSSW.output_file']
            if tmp != '':
                tmpOutFiles = string.split(cfg_params['CMSSW.output_file'],',')
                log.debug(7, 'cmssw::cmssw(): output files '+str(tmpOutFiles))
                for tmp in tmpOutFiles:
                    tmp=string.strip(tmp)
                    self.output_file.append(tmp)
                    pass
            else:
                log.message("No output file defined: only stdout/err will be available")
                pass
            pass
        except KeyError:
            log.message("No output file defined: only stdout/err will be available")
            pass

        # script_exe file as additional file in inputSandbox
        try:
            self.scriptExe = cfg_params['USER.script_exe']
            self.additional_inbox_files.append(self.scriptExe)
            if self.scriptExe != '':
               if not os.path.isfile(self.scriptExe):
                  msg ="WARNING. file "+self.scriptExe+" not found"
                  raise CrabException(msg)
        except KeyError:
           pass
                  
        ## additional input files
        try:
            tmpAddFiles = string.split(cfg_params['USER.additional_input_files'],',')
            for tmp in tmpAddFiles:
                if not os.path.exists(tmp):
                    raise CrabException("Additional input file not found: "+tmp)
                self.additional_inbox_files.append(string.strip(tmp))
                pass
            pass
        except KeyError:
            pass

        # files per job
        try:
            if (cfg_params['CMSSW.files_per_jobs']):
                raise CrabException("files_per_jobs no longer supported.  Quitting.")
        except KeyError:
            pass

        ## Events per job
        try:
            self.eventsPerJob =int( cfg_params['CMSSW.events_per_job'])
            self.selectEventsPerJob = 1
        except KeyError:
            self.eventsPerJob = -1
            self.selectEventsPerJob = 0
    
        ## number of jobs
        try:
            self.theNumberOfJobs =int( cfg_params['CMSSW.number_of_jobs'])
            self.selectNumberOfJobs = 1
        except KeyError:
            self.theNumberOfJobs = 0
            self.selectNumberOfJobs = 0

        try:
            self.total_number_of_events = int(cfg_params['CMSSW.total_number_of_events'])
            self.selectTotalNumberEvents = 1
        except KeyError:
            self.total_number_of_events = 0
            self.selectTotalNumberEvents = 0

        if ( (self.selectTotalNumberEvents + self.selectEventsPerJob + self.selectNumberOfJobs) != 2 ):
            msg = 'Must define exactly two of total_number_of_events, events_per_job, or number_of_jobs.'
            raise CrabException(msg)

        ## source seed for pythia
        try:
            self.sourceSeed = int(cfg_params['CMSSW.pythia_seed'])
        except KeyError:
            self.sourceSeed = None
            common.logger.debug(5,"No seed given")

        try:
            self.sourceSeedVtx = int(cfg_params['CMSSW.vtx_seed'])
        except KeyError:
            self.sourceSeedVtx = None
            common.logger.debug(5,"No vertex seed given")

        self.PsetEdit = PsetManipulator.PsetManipulator(self.pset) #Daniele Pset

        #DBSDLS-start
        ## Initialize the variables that are extracted from DBS/DLS and needed in other places of the code 
        self.maxEvents=0  # max events available   ( --> check the requested nb. of evts in Creator.py)
        self.DBSPaths={}  # all dbs paths requested ( --> input to the site local discovery script)
        self.jobDestination=[]  # Site destination(s) for each job (list of lists)
        ## Perform the data location and discovery (based on DBS/DLS)
        ## SL: Don't if NONE is specified as input (pythia use case)
        blockSites = {}
        if self.datasetPath:
            blockSites = self.DataDiscoveryAndLocation(cfg_params)
        #DBSDLS-end          

        self.tgzNameWithPath = self.getTarBall(self.executable)
    
        ## Select Splitting
        if self.selectNoInput: self.jobSplittingNoInput()
        else: self.jobSplittingByBlocks(blockSites)

        # modify Pset
        try:
            if (self.datasetPath): # standard job
                # always process all events in a file
                self.PsetEdit.inputModule("INPUT")
                self.PsetEdit.maxEvent("INPUTMAXEVENTS")
                self.PsetEdit.skipEvent("INPUTSKIPEVENTS")

            else:  # pythia like job
                self.PsetEdit.maxEvent(self.eventsPerJob)
                if (self.sourceSeed) :
                    self.PsetEdit.pythiaSeed("INPUT")
                    if (self.sourceSeedVtx) :
                        self.PsetEdit.pythiaSeedVtx("INPUTVTX")
            self.PsetEdit.psetWriter(self.configFilename())
        except:
            msg='Error while manipuliating ParameterSet: exiting...'
            raise CrabException(msg)

    def DataDiscoveryAndLocation(self, cfg_params):

        common.logger.debug(10,"CMSSW::DataDiscoveryAndLocation()")

        datasetPath=self.datasetPath

        ## TODO
        dataTiersList = ""
        dataTiers = dataTiersList.split(',')

        ## Contact the DBS
        try:
            self.pubdata=DataDiscovery_EDM.DataDiscovery_EDM(datasetPath, dataTiers, cfg_params)
            self.pubdata.fetchDBSInfo()

        except DataDiscovery_EDM.NotExistingDatasetError, ex :
            msg = 'ERROR ***: failed Data Discovery in DBS : %s'%ex.getErrorMessage()
            raise CrabException(msg)

        except DataDiscovery_EDM.NoDataTierinProvenanceError, ex :
            msg = 'ERROR ***: failed Data Discovery in DBS : %s'%ex.getErrorMessage()
            raise CrabException(msg)
        except DataDiscovery_EDM.DataDiscoveryError, ex:
            msg = 'ERROR ***: failed Data Discovery in DBS  %s'%ex.getErrorMessage()
            raise CrabException(msg)

        ## get list of all required data in the form of dbs paths  (dbs path = /dataset/datatier/owner)
        ## self.DBSPaths=self.pubdata.getDBSPaths()
        common.logger.message("Required data are :"+self.datasetPath)

        self.filesbyblock=self.pubdata.getFiles()
        self.eventsbyblock=self.pubdata.getEventsPerBlock()
        self.eventsbyfile=self.pubdata.getEventsPerFile()

        ## get max number of events
        self.maxEvents=self.pubdata.getMaxEvents() ##  self.maxEvents used in Creator.py 
        common.logger.message("\nThe number of available events is %s"%self.maxEvents)

        ## Contact the DLS and build a list of sites hosting the fileblocks
        try:
            dataloc=DataLocation_EDM.DataLocation_EDM(self.filesbyblock.keys(),cfg_params)
            dataloc.fetchDLSInfo()
        except DataLocation_EDM.DataLocationError , ex:
            msg = 'ERROR ***: failed Data Location in DLS \n %s '%ex.getErrorMessage()
            raise CrabException(msg)
        

        sites = dataloc.getSites()
        allSites = []
        listSites = sites.values()
        for list in listSites:
            for oneSite in list:
                allSites.append(oneSite)
        allSites = self.uniquelist(allSites)

        common.logger.message("Sites ("+str(len(allSites))+") hosting part/all of dataset: "+str(allSites)) 
        common.logger.debug(6, "List of Sites: "+str(allSites))
        return sites
    
    def jobSplittingByBlocks(self, blockSites):
        """
        Perform job splitting. Jobs run over an integer number of files
        and no more than one block.
        ARGUMENT: blockSites: dictionary with blocks as keys and list of host sites as values
        REQUIRES: self.selectTotalNumberEvents, self.selectEventsPerJob, self.selectNumberofJobs,
                  self.total_number_of_events, self.eventsPerJob, self.theNumberOfJobs,
                  self.maxEvents, self.filesbyblock
        SETS: self.jobDestination - Site destination(s) for each job (a list of lists)
              self.total_number_of_jobs - Total # of jobs
              self.list_of_args - File(s) job will run on (a list of lists)
        """

        # ---- Handle the possible job splitting configurations ---- #
        if (self.selectTotalNumberEvents):
            totalEventsRequested = self.total_number_of_events
        if (self.selectEventsPerJob):
            eventsPerJobRequested = self.eventsPerJob
            if (self.selectNumberOfJobs):
                totalEventsRequested = self.theNumberOfJobs * self.eventsPerJob

        # If user requested all the events in the dataset
        if (totalEventsRequested == -1):
            eventsRemaining=self.maxEvents
        # If user requested more events than are in the dataset
        elif (totalEventsRequested > self.maxEvents):
            eventsRemaining = self.maxEvents
            common.logger.message("Requested "+str(self.total_number_of_events)+ " events, but only "+str(self.maxEvents)+" events are available.")
        # If user requested less events than are in the dataset
        else:
            eventsRemaining = totalEventsRequested

        # For user info at end
        totalEventCount = 0

        if (self.selectTotalNumberEvents and self.selectNumberOfJobs):
            eventsPerJobRequested = int(eventsRemaining/self.theNumberOfJobs)

        if (self.selectNumberOfJobs):
            common.logger.message("May not create the exact number_of_jobs requested.")

        if ( self.ncjobs == 'all' ) :
            totalNumberOfJobs = 999999999
        else :
            totalNumberOfJobs = self.ncjobs
            

        blocks = blockSites.keys()
        blockCount = 0
        # Backup variable in case self.maxEvents counted events in a non-included block
        numBlocksInDataset = len(blocks)

        jobCount = 0
        list_of_lists = []

        # ---- Iterate over the blocks in the dataset until ---- #
        # ---- we've met the requested total # of events    ---- #
        while ( (eventsRemaining > 0) and (blockCount < numBlocksInDataset) and (jobCount < totalNumberOfJobs)):
            block = blocks[blockCount]


            evInBlock = self.eventsbyblock[block]
            common.logger.debug(5,'Events in Block File '+str(evInBlock))

            #Correct - switch to this when DBS up
            #numEventsInBlock = self.eventsbyblock[block]
            numEventsInBlock = evInBlock
            
            files = self.filesbyblock[block]
            numFilesInBlock = len(files)
            if (numFilesInBlock <= 0):
                continue
            fileCount = 0

            # ---- New block => New job ---- #
            parString = "\\{"
            # counter for number of events in files currently worked on
            filesEventCount = 0
            # flag if next while loop should touch new file
            newFile = 1
            # job event counter
            jobSkipEventCount = 0
            
            # ---- Iterate over the files in the block until we've met the requested ---- #
            # ---- total # of events or we've gone over all the files in this block  ---- #
            while ( (eventsRemaining > 0) and (fileCount < numFilesInBlock) and (jobCount < totalNumberOfJobs) ):
                file = files[fileCount]
                if newFile :
                    numEventsInFile = self.eventsbyfile[file]
                    common.logger.debug(6, "File "+str(file)+" has "+str(numEventsInFile)+" events")
                    # increase filesEventCount
                    filesEventCount += numEventsInFile
                    # Add file to current job
                    parString += '\\\"' + file + '\\\"\,'
                    newFile = 0

                # if less events in file remain than eventsPerJobRequested
                if ( filesEventCount - jobSkipEventCount < eventsPerJobRequested ) :
                    # if last file in block
                    if ( fileCount == numFilesInBlock ) :
                        # end job using last file, use remaining events in block
                        # close job and touch new file
                        fullString = parString[:-2]
                        fullString += '\\}'
                        list_of_lists.append([fullString,str(-1),str(jobSkipEventCount)])
                        common.logger.message("Job "+str(jobCount+1)+" can run over "+str(filesEventCount - jobSkipEventCount)+" events (last file in block).")
                        self.jobDestination.append(blockSites[block])
                        common.logger.debug(5,"Job "+str(jobCount+1)+" Destination: "+str(self.jobDestination[jobCount]))
                        # reset counter
                        jobCount = jobCount + 1
                        totalEventCount = totalEventCount + eventsPerJobRequested
                        eventsRemaining = eventsRemaining - eventsPerJobRequested
                        jobSkipEventCount = 0
                        # reset file
                        parString = "\\{"
                        filesEventCount = 0
                        newFile = 1
                        fileCount += 1
                    else :
                        # go to next file
                        newFile = 1
                        fileCount += 1
                # if events in file equal to eventsPerJobRequested
                elif ( filesEventCount - jobSkipEventCount == eventsPerJobRequested ) :
                    # close job and touch new file
                    fullString = parString[:-2]
                    fullString += '\\}'
                    list_of_lists.append([fullString,str(eventsPerJobRequested),str(jobSkipEventCount)])
                    common.logger.message("Job "+str(jobCount+1)+" can run over "+str(eventsPerJobRequested)+" events.")
                    self.jobDestination.append(blockSites[block])
                    common.logger.debug(5,"Job "+str(jobCount+1)+" Destination: "+str(self.jobDestination[jobCount]))
                    # reset counter
                    jobCount = jobCount + 1
                    totalEventCount = totalEventCount + eventsPerJobRequested
                    eventsRemaining = eventsRemaining - eventsPerJobRequested
                    jobSkipEventCount = 0
                    # reset file
                    parString = "\\{"
                    filesEventCount = 0
                    newFile = 1
                    fileCount += 1
                    
                # if more events in file remain than eventsPerJobRequested
                else :
                    # close job but don't touch new file
                    fullString = parString[:-2]
                    fullString += '\\}'
                    list_of_lists.append([fullString,str(eventsPerJobRequested),str(jobSkipEventCount)])
                    common.logger.message("Job "+str(jobCount+1)+" can run over "+str(eventsPerJobRequested)+" events.")
                    self.jobDestination.append(blockSites[block])
                    common.logger.debug(5,"Job "+str(jobCount+1)+" Destination: "+str(self.jobDestination[jobCount]))
                    # increase counter
                    jobCount = jobCount + 1
                    totalEventCount = totalEventCount + eventsPerJobRequested
                    eventsRemaining = eventsRemaining - eventsPerJobRequested
                    # calculate skip events for last file
                    # use filesEventCount (contains several files), jobSkipEventCount and eventsPerJobRequest
                    jobSkipEventCount = eventsPerJobRequested - (filesEventCount - jobSkipEventCount - self.eventsbyfile[file])
                    # remove all but the last file
                    filesEventCount = self.eventsbyfile[file]
                    parString = "\\{"
                    parString += '\\\"' + file + '\\\"\,'
            pass # END while (iterate over files in the block)
        pass # END while (iterate over blocks in the dataset)
        self.total_number_of_jobs = jobCount
        if (eventsRemaining > 0 and jobCount < totalNumberOfJobs ):
            common.logger.message("Could not run on all requested events because some blocks not hosted at allowed sites.")
        common.logger.message("\n"+str(jobCount)+" job(s) can run on "+str(totalEventCount)+" events.\n")
        
        self.list_of_args = list_of_lists
        return

    def jobSplittingNoInput(self):
        """
        Perform job splitting based on number of event per job
        """
        common.logger.debug(5,'Splitting per events')
        common.logger.message('Required '+str(self.eventsPerJob)+' events per job ')
        common.logger.message('Required '+str(self.theNumberOfJobs)+' jobs in total ')
        common.logger.message('Required '+str(self.total_number_of_events)+' events in total ')

        if (self.total_number_of_events < 0):
            msg='Cannot split jobs per Events with "-1" as total number of events'
            raise CrabException(msg)

        if (self.selectEventsPerJob):
            self.total_number_of_jobs = int(self.total_number_of_events/self.eventsPerJob)
        elif (self.selectNumberOfJobs) :
            self.total_number_of_jobs = self.theNumberOfJobs
            self.eventsPerJob = int(self.total_number_of_events/self.total_number_of_jobs)

        common.logger.debug(5,'N jobs  '+str(self.total_number_of_jobs))

        # is there any remainder?
        check = int(self.total_number_of_events) - (int(self.total_number_of_jobs)*self.eventsPerJob)

        common.logger.debug(5,'Check  '+str(check))

        common.logger.message(str(self.total_number_of_jobs)+' jobs can be created, each for '+str(self.eventsPerJob)+' for a total of '+str(self.total_number_of_jobs*self.eventsPerJob)+' events')
        if check > 0:
            common.logger.message('Warning: asked '+str(self.total_number_of_events)+' but can do only '+str(int(self.total_number_of_jobs)*self.eventsPerJob))

        # argument is seed number.$i
        self.list_of_args = []
        for i in range(self.total_number_of_jobs):
            ## Since there is no input, any site is good
            self.jobDestination.append(["Any"])
            if (self.sourceSeed):
                if (self.sourceSeedVtx):
                    ## pythia + vtx random seed
                    self.list_of_args.append([
                                              str(self.sourceSeed)+str(i),
                                              str(self.sourceSeedVtx)+str(i)
                                              ])
                else:
                    ## only pythia random seed
                    self.list_of_args.append([(str(self.sourceSeed)+str(i))])
            else:
                ## no random seed
                self.list_of_args.append([str(i)])
        #print self.list_of_args

        return

    def split(self, jobParams):
 
        common.jobDB.load()
        #### Fabio
        njobs = self.total_number_of_jobs
        arglist = self.list_of_args
        # create the empty structure
        for i in range(njobs):
            jobParams.append("")
        
        for job in range(njobs):
            jobParams[job] = arglist[job]
            # print str(arglist[job])
            # print jobParams[job]
            common.jobDB.setArguments(job, jobParams[job])
            common.logger.debug(5,"Job "+str(job)+" Destination: "+str(self.jobDestination[job]))
            common.jobDB.setDestination(job, self.jobDestination[job])

        common.jobDB.save()
        return
    
    def getJobTypeArguments(self, nj, sched):
        result = ''
        for i in common.jobDB.arguments(nj):
            result=result+str(i)+" "
        return result
  
    def numberOfJobs(self):
        # Fabio
        return self.total_number_of_jobs

    def getTarBall(self, exe):
        """
        Return the TarBall with lib and exe
        """
        
        # if it exist, just return it
        self.tgzNameWithPath = common.work_space.shareDir()+self.tgz_name
        if os.path.exists(self.tgzNameWithPath):
            return self.tgzNameWithPath

        # Prepare a tar gzipped file with user binaries.
        self.buildTar_(exe)

        return string.strip(self.tgzNameWithPath)

    def buildTar_(self, executable):

        # First of all declare the user Scram area
        swArea = self.scram.getSWArea_()
        #print "swArea = ", swArea
        swVersion = self.scram.getSWVersion()
        #print "swVersion = ", swVersion
        swReleaseTop = self.scram.getReleaseTop_()
        #print "swReleaseTop = ", swReleaseTop
        
        ## check if working area is release top
        if swReleaseTop == '' or swArea == swReleaseTop:
            return

        filesToBeTarred = []
        ## First find the executable
        if (self.executable != ''):
            exeWithPath = self.scram.findFile_(executable)
#           print exeWithPath
            if ( not exeWithPath ):
                raise CrabException('User executable '+executable+' not found')
 
            ## then check if it's private or not
            if exeWithPath.find(swReleaseTop) == -1:
                # the exe is private, so we must ship
                common.logger.debug(5,"Exe "+exeWithPath+" to be tarred")
                path = swArea+'/'
                exe = string.replace(exeWithPath, path,'')
                filesToBeTarred.append(exe)
                pass
            else:
                # the exe is from release, we'll find it on WN
                pass
 
        ## Now get the libraries: only those in local working area
        libDir = 'lib'
        lib = swArea+'/' +libDir
        common.logger.debug(5,"lib "+lib+" to be tarred")
        if os.path.exists(lib):
            filesToBeTarred.append(libDir)
 
        ## Now check if module dir is present
        moduleDir = 'module'
        if os.path.isdir(swArea+'/'+moduleDir):
            filesToBeTarred.append(moduleDir)

        ## Now check if the Data dir is present
        dataDir = 'src/Data/'
        if os.path.isdir(swArea+'/'+dataDir):
            filesToBeTarred.append(dataDir)
 
        ## Create the tar-ball
        if len(filesToBeTarred)>0:
            cwd = os.getcwd()
            os.chdir(swArea)
            tarcmd = 'tar zcvf ' + self.tgzNameWithPath + ' ' 
            for line in filesToBeTarred:
                tarcmd = tarcmd + line + ' '
            cout = runCommand(tarcmd)
            if not cout:
                raise CrabException('Could not create tar-ball')
            os.chdir(cwd)
        else:
            common.logger.debug(5,"No files to be to be tarred")
        
        return
        
    def wsSetupEnvironment(self, nj):
        """
        Returns part of a job script which prepares
        the execution environment for the job 'nj'.
        """
        # Prepare JobType-independent part
        txt = '' 
   
        ## OLI_Daniele at this level  middleware already known

        txt += 'if [ $middleware == LCG ]; then \n' 
        txt += self.wsSetupCMSLCGEnvironment_()
        txt += 'elif [ $middleware == OSG ]; then\n'
        txt += '    time=`date -u +"%s"`\n'
        txt += '    WORKING_DIR=$OSG_WN_TMP/cms_$time\n'
        txt += '    echo "Creating working directory: $WORKING_DIR"\n'
        txt += '    /bin/mkdir -p $WORKING_DIR\n'
        txt += '    if [ ! -d $WORKING_DIR ] ;then\n'
        txt += '        echo "SET_CMS_ENV 10016 ==> OSG $WORKING_DIR could not be created on WN `hostname`"\n'
        txt += '        echo "JOB_EXIT_STATUS = 10016"\n'
        txt += '        echo "JobExitCode=10016" | tee -a $RUNTIME_AREA/$repo\n'
        txt += '        dumpStatus $RUNTIME_AREA/$repo\n'
        txt += '        rm -f $RUNTIME_AREA/$repo \n'
        txt += '        echo "MonitorJobID=`echo $MonitorJobID`" | tee -a $RUNTIME_AREA/$repo \n'
        txt += '        echo "MonitorID=`echo $MonitorID`" | tee -a $RUNTIME_AREA/$repo\n'
        txt += '        exit 1\n'
        txt += '    fi\n'
        txt += '\n'
        txt += '    echo "Change to working directory: $WORKING_DIR"\n'
        txt += '    cd $WORKING_DIR\n'
        txt += self.wsSetupCMSOSGEnvironment_() 
        txt += 'fi\n'

        # Prepare JobType-specific part
        scram = self.scram.commandName()
        txt += '\n\n'
        txt += 'echo "### SPECIFIC JOB SETUP ENVIRONMENT ###"\n'
        txt += scram+' project CMSSW '+self.version+'\n'
        txt += 'status=$?\n'
        txt += 'if [ $status != 0 ] ; then\n'
        txt += '   echo "SET_EXE_ENV 10034 ==>ERROR CMSSW '+self.version+' not found on `hostname`" \n'
        txt += '   echo "JOB_EXIT_STATUS = 10034"\n'
        txt += '   echo "JobExitCode=10034" | tee -a $RUNTIME_AREA/$repo\n'
        txt += '   dumpStatus $RUNTIME_AREA/$repo\n'
        txt += '   rm -f $RUNTIME_AREA/$repo \n'
        txt += '   echo "MonitorJobID=`echo $MonitorJobID`" | tee -a $RUNTIME_AREA/$repo \n'
        txt += '   echo "MonitorID=`echo $MonitorID`" | tee -a $RUNTIME_AREA/$repo\n'
        ## OLI_Daniele
        txt += '    if [ $middleware == OSG ]; then \n'
        txt += '        echo "Remove working directory: $WORKING_DIR"\n'
        txt += '        cd $RUNTIME_AREA\n'
        txt += '        /bin/rm -rf $WORKING_DIR\n'
        txt += '        if [ -d $WORKING_DIR ] ;then\n'
        txt += '            echo "SET_CMS_ENV 10018 ==> OSG $WORKING_DIR could not be deleted on WN `hostname` after CMSSW CMSSW_0_6_1 not found on `hostname`"\n'
        txt += '            echo "JOB_EXIT_STATUS = 10018"\n'
        txt += '            echo "JobExitCode=10018" | tee -a $RUNTIME_AREA/$repo\n'
        txt += '            dumpStatus $RUNTIME_AREA/$repo\n'
        txt += '            rm -f $RUNTIME_AREA/$repo \n'
        txt += '            echo "MonitorJobID=`echo $MonitorJobID`" | tee -a $RUNTIME_AREA/$repo \n'
        txt += '            echo "MonitorID=`echo $MonitorID`" | tee -a $RUNTIME_AREA/$repo\n'
        txt += '        fi\n'
        txt += '    fi \n'
        txt += '   exit 1 \n'
        txt += 'fi \n'
        txt += 'echo "CMSSW_VERSION =  '+self.version+'"\n'
        txt += 'cd '+self.version+'\n'
        ### needed grep for bug in scramv1 ###
        txt += 'eval `'+scram+' runtime -sh | grep -v SCRAMRT_LSB_JOBNAME`\n'

        # Handle the arguments:
        txt += "\n"
        txt += "## number of arguments (first argument always jobnumber)\n"
        txt += "\n"
#        txt += "narg=$#\n"
        txt += "if [ $nargs -lt 2 ]\n"
        txt += "then\n"
        txt += "    echo 'SET_EXE_ENV 1 ==> ERROR Too few arguments' +$nargs+ \n"
        txt += '    echo "JOB_EXIT_STATUS = 50113"\n'
        txt += '    echo "JobExitCode=50113" | tee -a $RUNTIME_AREA/$repo\n'
        txt += '    dumpStatus $RUNTIME_AREA/$repo\n'
        txt += '    rm -f $RUNTIME_AREA/$repo \n'
        txt += '    echo "MonitorJobID=`echo $MonitorJobID`" | tee -a $RUNTIME_AREA/$repo \n'
        txt += '    echo "MonitorID=`echo $MonitorID`" | tee -a $RUNTIME_AREA/$repo\n'
        ## OLI_Daniele
        txt += '    if [ $middleware == OSG ]; then \n'
        txt += '        echo "Remove working directory: $WORKING_DIR"\n'
        txt += '        cd $RUNTIME_AREA\n'
        txt += '        /bin/rm -rf $WORKING_DIR\n'
        txt += '        if [ -d $WORKING_DIR ] ;then\n'
        txt += '            echo "SET_EXE_ENV 50114 ==> OSG $WORKING_DIR could not be deleted on WN `hostname` after Too few arguments for CRAB job wrapper"\n'
        txt += '            echo "JOB_EXIT_STATUS = 50114"\n'
        txt += '            echo "JobExitCode=50114" | tee -a $RUNTIME_AREA/$repo\n'
        txt += '            dumpStatus $RUNTIME_AREA/$repo\n'
        txt += '            rm -f $RUNTIME_AREA/$repo \n'
        txt += '            echo "MonitorJobID=`echo $MonitorJobID`" | tee -a $RUNTIME_AREA/$repo \n'
        txt += '            echo "MonitorID=`echo $MonitorID`" | tee -a $RUNTIME_AREA/$repo\n'
        txt += '        fi\n'
        txt += '    fi \n'
        txt += "    exit 1\n"
        txt += "fi\n"
        txt += "\n"

        # Prepare job-specific part
        job = common.job_list[nj]
        pset = os.path.basename(job.configFilename())
        txt += '\n'
        if (self.datasetPath): # standard job
            #txt += 'InputFiles=$2\n'
            txt += 'InputFiles=${args[1]}\n'
            txt += 'MaxEvents=${args[2]}\n'
            txt += 'SkipEvents=${args[3]}\n'
            txt += 'echo "Inputfiles:<$InputFiles>"\n'
            txt += 'sed "s#{\'INPUT\'}#$InputFiles#" $RUNTIME_AREA/'+pset+' > pset_tmp_1.cfg\n'
            txt += 'echo "MaxEvents:<$MaxEvents>"\n'
            txt += 'sed "s#INPUTMAXEVENTS#$MaxEvents#" $RUNTIME_AREA/ pset_tmp_1.cfg > pset_tmp_2.cfg\n'
            txt += 'echo "SkipEvents:<$SkipEvents>"\n'
            txt += 'sed "s#INPUTSKIPEVENTS#$SkipEvents#" $RUNTIME_AREA/ pset_tmp_2.cfg > pset.cfg\n'
        else:  # pythia like job
            if (self.sourceSeed):
#                txt += 'Seed=$2\n'
                txt += 'Seed=${args[1]}\n'
                txt += 'echo "Seed: <$Seed>"\n'
                txt += 'sed "s#\<INPUT\>#$Seed#" $RUNTIME_AREA/'+pset+' > tmp.cfg\n'
                if (self.sourceSeedVtx):
#                    txt += 'VtxSeed=$3\n'
                    txt += 'VtxSeed=${args[2]}\n'
                    txt += 'echo "VtxSeed: <$VtxSeed>"\n'
                    txt += 'sed "s#INPUTVTX#$VtxSeed#" tmp.cfg > pset.cfg\n'
                else:
                    txt += 'mv tmp.cfg pset.cfg\n'
            else:
                txt += '# Copy untouched pset\n'
                txt += 'cp $RUNTIME_AREA/'+pset+' pset.cfg\n'


        if len(self.additional_inbox_files) > 0:
            for file in self.additional_inbox_files:
                relFile = file.split("/")[-1]
                txt += 'if [ -e $RUNTIME_AREA/'+relFile+' ] ; then\n'
                txt += '   cp $RUNTIME_AREA/'+relFile+' .\n'
                txt += '   chmod +x '+relFile+'\n'
                txt += 'fi\n'
            pass 

        txt += 'echo "### END JOB SETUP ENVIRONMENT ###"\n\n'

        txt += '\n'
        txt += 'echo "***** cat pset.cfg *********"\n'
        txt += 'cat pset.cfg\n'
        txt += 'echo "****** end pset.cfg ********"\n'
        txt += '\n'
        # txt += 'echo "***** cat pset1.cfg *********"\n'
        # txt += 'cat pset1.cfg\n'
        # txt += 'echo "****** end pset1.cfg ********"\n'
        return txt

    def wsBuildExe(self, nj):
        """
        Put in the script the commands to build an executable
        or a library.
        """

        txt = ""

        if os.path.isfile(self.tgzNameWithPath):
            txt += 'echo "tar xzvf $RUNTIME_AREA/'+os.path.basename(self.tgzNameWithPath)+'"\n'
            txt += 'tar xzvf $RUNTIME_AREA/'+os.path.basename(self.tgzNameWithPath)+'\n'
            txt += 'untar_status=$? \n'
            txt += 'if [ $untar_status -ne 0 ]; then \n'
            txt += '   echo "SET_EXE 1 ==> ERROR Untarring .tgz file failed"\n'
            txt += '   echo "JOB_EXIT_STATUS = $untar_status" \n'
            txt += '   echo "JobExitCode=$untar_status" | tee -a $RUNTIME_AREA/$repo\n'
            txt += '   if [ $middleware == OSG ]; then \n'
            txt += '       echo "Remove working directory: $WORKING_DIR"\n'
            txt += '       cd $RUNTIME_AREA\n'
            txt += '       /bin/rm -rf $WORKING_DIR\n'
            txt += '       if [ -d $WORKING_DIR ] ;then\n'
            txt += '           echo "SET_EXE 50999 ==> OSG $WORKING_DIR could not be deleted on WN `hostname` after Untarring .tgz file failed"\n'
            txt += '           echo "JOB_EXIT_STATUS = 50999"\n'
            txt += '           echo "JobExitCode=50999" | tee -a $RUNTIME_AREA/$repo\n'
            txt += '           dumpStatus $RUNTIME_AREA/$repo\n'
            txt += '           rm -f $RUNTIME_AREA/$repo \n'
            txt += '           echo "MonitorJobID=`echo $MonitorJobID`" | tee -a $RUNTIME_AREA/$repo \n'
            txt += '           echo "MonitorID=`echo $MonitorID`" | tee -a $RUNTIME_AREA/$repo\n'
            txt += '       fi\n'
            txt += '   fi \n'
            txt += '   \n'
            txt += '   exit 1 \n'
            txt += 'else \n'
            txt += '   echo "Successful untar" \n'
            txt += 'fi \n'
            pass
        
        return txt

    def modifySteeringCards(self, nj):
        """
        modify the card provided by the user, 
        writing a new card into share dir
        """
        
    def executableName(self):
        return self.executable

    def executableArgs(self):
        return " -p pset.cfg"

    def inputSandbox(self, nj):
        """
        Returns a list of filenames to be put in JDL input sandbox.
        """
        inp_box = []
        # dict added to delete duplicate from input sandbox file list
        seen = {}
        ## code
        if os.path.isfile(self.tgzNameWithPath):
            inp_box.append(self.tgzNameWithPath)
        ## config
        inp_box.append(common.job_list[nj].configFilename())
        ## additional input files
        #for file in self.additional_inbox_files:
        #    inp_box.append(common.work_space.cwdDir()+file)
        return inp_box

    def outputSandbox(self, nj):
        """
        Returns a list of filenames to be put in JDL output sandbox.
        """
        out_box = []

        stdout=common.job_list[nj].stdout()
        stderr=common.job_list[nj].stderr()

        ## User Declared output files
        for out in self.output_file:
            n_out = nj + 1 
            out_box.append(self.numberFile_(out,str(n_out)))
        return out_box
        return []

    def prepareSteeringCards(self):
        """
        Make initial modifications of the user's steering card file.
        """
        return

    def wsRenameOutput(self, nj):
        """
        Returns part of a job script which renames the produced files.
        """

        txt = '\n'
        txt += '# directory content\n'
        txt += 'ls \n'
        file_list = ''
        for fileWithSuffix in self.output_file:
            output_file_num = self.numberFile_(fileWithSuffix, '$NJob')
            file_list=file_list+output_file_num+' '
            txt += '\n'
            txt += '# check output file\n'
            txt += 'ls '+fileWithSuffix+'\n'
            txt += 'ls_result=$?\n'
            #txt += 'exe_result=$?\n'
            txt += 'if [ $ls_result -ne 0 ] ; then\n'
            txt += '   echo "ERROR: Problem with output file"\n'
            #txt += '   echo "JOB_EXIT_STATUS = $exe_result"\n'
            #txt += '   echo "JobExitCode=60302" | tee -a $RUNTIME_AREA/$repo\n'
            #txt += '   dumpStatus $RUNTIME_AREA/$repo\n'
            ### OLI_DANIELE
            if common.scheduler.boss_scheduler_name == 'condor_g':
                txt += '    if [ $middleware == OSG ]; then \n'
                txt += '        echo "prepare dummy output file"\n'
                txt += '        echo "Processing of job output failed" > $RUNTIME_AREA/'+output_file_num+'\n'
                txt += '    fi \n'
            txt += 'else\n'
            txt += '   cp '+fileWithSuffix+' $RUNTIME_AREA/'+output_file_num+'\n'
            txt += 'fi\n'
       
        txt += 'cd $RUNTIME_AREA\n'
        file_list=file_list[:-1]
        txt += 'file_list="'+file_list+'"\n'
        txt += 'cd $RUNTIME_AREA\n'
        ### OLI_DANIELE
        txt += 'if [ $middleware == OSG ]; then\n'  
        txt += '    cd $RUNTIME_AREA\n'
        txt += '    echo "Remove working directory: $WORKING_DIR"\n'
        txt += '    /bin/rm -rf $WORKING_DIR\n'
        txt += '    if [ -d $WORKING_DIR ] ;then\n'
        txt += '        echo "SET_EXE 60999 ==> OSG $WORKING_DIR could not be deleted on WN `hostname` after cleanup of WN"\n'
        txt += '        echo "JOB_EXIT_STATUS = 60999"\n'
        txt += '        echo "JobExitCode=60999" | tee -a $RUNTIME_AREA/$repo\n'
        txt += '        dumpStatus $RUNTIME_AREA/$repo\n'
        txt += '        rm -f $RUNTIME_AREA/$repo \n'
        txt += '        echo "MonitorJobID=`echo $MonitorJobID`" | tee -a $RUNTIME_AREA/$repo \n'
        txt += '        echo "MonitorID=`echo $MonitorID`" | tee -a $RUNTIME_AREA/$repo\n'
        txt += '    fi\n'
        txt += 'fi\n'
        txt += '\n'
        return txt

    def numberFile_(self, file, txt):
        """
        append _'txt' before last extension of a file
        """
        p = string.split(file,".")
        # take away last extension
        name = p[0]
        for x in p[1:-1]:
           name=name+"."+x
        # add "_txt"
        if len(p)>1:
          ext = p[len(p)-1]
          #result = name + '_' + str(txt) + "." + ext
          result = name + '_' + txt + "." + ext
        else:
          #result = name + '_' + str(txt)
          result = name + '_' + txt
        
        return result

    def getRequirements(self, nj):
        """
        return job requirements to add to jdl files 
        """
        req = ''
        if common.analisys_common_info['sw_version']:
            req='Member("VO-cms-' + \
                 common.analisys_common_info['sw_version'] + \
                 '", other.GlueHostApplicationSoftwareRunTimeEnvironment)'

        req = req + ' && (other.GlueHostNetworkAdapterOutboundIP)'

        ## here we should get the requirement for job nj
        sites = common.jobDB.destination(nj)

        # check for "Any" site, in case no requirement for site
        if len(sites)>0 and sites[0]!="Any":
            req = req + ' && anyMatch(other.storage.CloseSEs, ('
            for site in sites:
                #req = req + 'other.GlueCEInfoHostName == "' + site + '" || '
                req = req + 'target.GlueSEUniqueID=="' + site + '" || '
                pass
            # remove last ||
            req = req[0:-4]
            req = req + '))'

        return req

    def configFilename(self):
        """ return the config filename """
        return self.name()+'.cfg'

    ### OLI_DANIELE
    def wsSetupCMSOSGEnvironment_(self):
        """
        Returns part of a job script which is prepares
        the execution environment and which is common for all CMS jobs.
        """
        txt = '\n'
        txt += '   echo "### SETUP CMS OSG  ENVIRONMENT ###"\n'
        txt += '   if [ -f $GRID3_APP_DIR/cmssoft/cmsset_default.sh ] ;then\n'
        txt += '      # Use $GRID3_APP_DIR/cmssoft/cmsset_default.sh to setup cms software\n'
        txt += '       source $GRID3_APP_DIR/cmssoft/cmsset_default.sh '+self.version+'\n'
        txt += '   elif [ -f $OSG_APP/cmssoft/cmsset_default.sh ] ;then\n'
        txt += '      # Use $OSG_APP/cmssoft/cmsset_default.sh to setup cms software\n'
        txt += '       source $OSG_APP/cmssoft/cmsset_default.sh '+self.version+'\n'
        txt += '   else\n'
        txt += '       echo "SET_CMS_ENV 10020 ==> ERROR $GRID3_APP_DIR/cmssoft/cmsset_default.sh and $OSG_APP/cmssoft/cmsset_default.sh file not found"\n'
        txt += '       echo "JOB_EXIT_STATUS = 10020"\n'
        txt += '       echo "JobExitCode=10020" | tee -a $RUNTIME_AREA/$repo\n'
        txt += '       dumpStatus $RUNTIME_AREA/$repo\n'
        txt += '       rm -f $RUNTIME_AREA/$repo \n'
        txt += '       echo "MonitorJobID=`echo $MonitorJobID`" | tee -a $RUNTIME_AREA/$repo \n'
        txt += '       echo "MonitorID=`echo $MonitorID`" | tee -a $RUNTIME_AREA/$repo\n'
        txt += '       exit 1\n'
        txt += '\n'
        txt += '       echo "Remove working directory: $WORKING_DIR"\n'
        txt += '       cd $RUNTIME_AREA\n'
        txt += '       /bin/rm -rf $WORKING_DIR\n'
        txt += '       if [ -d $WORKING_DIR ] ;then\n'
        txt += '            echo "SET_CMS_ENV 10017 ==> OSG $WORKING_DIR could not be deleted on WN `hostname` after $GRID3_APP_DIR/cmssoft/cmsset_default.sh and $OSG_APP/cmssoft/cmsset_default.sh file not found"\n' 
        txt += '            echo "JOB_EXIT_STATUS = 10017"\n' 
        txt += '            echo "JobExitCode=10017" | tee -a $RUNTIME_AREA/$repo\n' 
        txt += '            dumpStatus $RUNTIME_AREA/$repo\n'
        txt += '            rm -f $RUNTIME_AREA/$repo \n'
        txt += '            echo "MonitorJobID=`echo $MonitorJobID`" | tee -a $RUNTIME_AREA/$repo \n'
        txt += '            echo "MonitorID=`echo $MonitorID`" | tee -a $RUNTIME_AREA/$repo\n'
        txt += '       fi\n'
        txt += '\n'
        txt += '       exit 1\n'
        txt += '   fi\n'
        txt += '\n'
        txt += '   echo "SET_CMS_ENV 0 ==> setup cms environment ok"\n'
        txt += '   echo " END SETUP CMS OSG  ENVIRONMENT "\n'

        return txt
 
    ### OLI_DANIELE
    def wsSetupCMSLCGEnvironment_(self):
        """
        Returns part of a job script which is prepares
        the execution environment and which is common for all CMS jobs.
        """
        txt  = '   \n'
        txt += '   echo " ### SETUP CMS LCG  ENVIRONMENT ### "\n'
        txt += '   if [ ! $VO_CMS_SW_DIR ] ;then\n'
        txt += '       echo "SET_CMS_ENV 10031 ==> ERROR CMS software dir not found on WN `hostname`"\n'
        txt += '       echo "JOB_EXIT_STATUS = 10031" \n'
        txt += '       echo "JobExitCode=10031" | tee -a $RUNTIME_AREA/$repo\n'
        txt += '       dumpStatus $RUNTIME_AREA/$repo\n'
        txt += '       rm -f $RUNTIME_AREA/$repo \n'
        txt += '       echo "MonitorJobID=`echo $MonitorJobID`" | tee -a $RUNTIME_AREA/$repo \n'
        txt += '       echo "MonitorID=`echo $MonitorID`" | tee -a $RUNTIME_AREA/$repo\n'
        txt += '       exit 1\n'
        txt += '   else\n'
        txt += '       echo "Sourcing environment... "\n'
        txt += '       if [ ! -s $VO_CMS_SW_DIR/cmsset_default.sh ] ;then\n'
        txt += '           echo "SET_CMS_ENV 10020 ==> ERROR cmsset_default.sh file not found into dir $VO_CMS_SW_DIR"\n'
        txt += '           echo "JOB_EXIT_STATUS = 10020"\n'
        txt += '           echo "JobExitCode=10020" | tee -a $RUNTIME_AREA/$repo\n'
        txt += '           dumpStatus $RUNTIME_AREA/$repo\n'
        txt += '           rm -f $RUNTIME_AREA/$repo \n'
        txt += '           echo "MonitorJobID=`echo $MonitorJobID`" | tee -a $RUNTIME_AREA/$repo \n'
        txt += '           echo "MonitorID=`echo $MonitorID`" | tee -a $RUNTIME_AREA/$repo\n'
        txt += '           exit 1\n'
        txt += '       fi\n'
        txt += '       echo "sourcing $VO_CMS_SW_DIR/cmsset_default.sh"\n'
        txt += '       source $VO_CMS_SW_DIR/cmsset_default.sh\n'
        txt += '       result=$?\n'
        txt += '       if [ $result -ne 0 ]; then\n'
        txt += '           echo "SET_CMS_ENV 10032 ==> ERROR problem sourcing $VO_CMS_SW_DIR/cmsset_default.sh"\n'
        txt += '           echo "JOB_EXIT_STATUS = 10032"\n'
        txt += '           echo "JobExitCode=10032" | tee -a $RUNTIME_AREA/$repo\n'
        txt += '           dumpStatus $RUNTIME_AREA/$repo\n'
        txt += '           rm -f $RUNTIME_AREA/$repo \n'
        txt += '           echo "MonitorJobID=`echo $MonitorJobID`" | tee -a $RUNTIME_AREA/$repo \n'
        txt += '           echo "MonitorID=`echo $MonitorID`" | tee -a $RUNTIME_AREA/$repo\n'
        txt += '           exit 1\n'
        txt += '       fi\n'
        txt += '   fi\n'
        txt += '   \n'
        txt += '   string=`cat /etc/redhat-release`\n'
        txt += '   echo $string\n'
        txt += '   if [[ $string = *alhalla* ]]; then\n'
        txt += '       echo "SCRAM_ARCH= $SCRAM_ARCH"\n'
        txt += '   elif [[ $string = *Enterprise* ]] || [[ $string = *cientific* ]]; then\n'
        txt += '       export SCRAM_ARCH=slc3_ia32_gcc323\n'
        txt += '       echo "SCRAM_ARCH= $SCRAM_ARCH"\n'
        txt += '   else\n'
        txt += '       echo "SET_CMS_ENV 10033 ==> ERROR OS unknown, LCG environment not initialized"\n'
        txt += '       echo "JOB_EXIT_STATUS = 10033"\n'
        txt += '       echo "JobExitCode=10033" | tee -a $RUNTIME_AREA/$repo\n'
        txt += '       dumpStatus $RUNTIME_AREA/$repo\n'
        txt += '       rm -f $RUNTIME_AREA/$repo \n'
        txt += '       echo "MonitorJobID=`echo $MonitorJobID`" | tee -a $RUNTIME_AREA/$repo \n'
        txt += '       echo "MonitorID=`echo $MonitorID`" | tee -a $RUNTIME_AREA/$repo\n'
        txt += '       exit 1\n'
        txt += '   fi\n'
        txt += '   echo "SET_CMS_ENV 0 ==> setup cms environment ok"\n'
        txt += '   echo "### END SETUP CMS LCG ENVIRONMENT ###"\n'
        return txt

    def setParam_(self, param, value):
        self._params[param] = value

    def getParams(self):
        return self._params

    def setTaskid_(self):
        self._taskId = self.cfg_params['taskId']
        
    def getTaskid(self):
        return self._taskId

#######################################################################
    def uniquelist(self, old):
        """
        remove duplicates from a list
        """
        nd={}
        for e in old:
            nd[e]=0
        return nd.keys()
Revision:	1.39
Committed:	Tue Aug 29 02:53:32 2006 UTC (18 years, 8 months ago) by gutsche
Content type:	text/x-python
Branch:	MAIN
Changes since 1.38:	+3 -3 lines
Log Message:	fixed argument replacing in pset