comparison commons/core/launcher/Launcher2.py @ 31:0ab839023fe4

Uploaded

31:0ab839023fe4

from commons.tools.CleanClusterNodesAfterRepet import CleanClusterNodesAfterRepet
from commons.core.stat.Stat import Stat
from commons.core.launcher.WriteScript import WriteScript
from commons.core.sql.TableJobAdaptatorFactory import TableJobAdaptatorFactory
from commons.core.sql.Job import Job
import stat
import os
import re
import sys
import time
import glob

class LauncherParameter(object):

    def __init__(self, jobDB):
        self._jobDB = jobDB
    
    def getJobDB(self):
        return self._jobDB

    def setQuery(self, query):
        self._query = query

    def setSubject(self, subject):
        self._subject = subject
        
    def setParam(self, param):
        self._param = param
    
    def setCurrentDir(self, currentDir):
        self._currentDir = currentDir
    
    def getCurrentDir(self):
        return self._currentDir    

    def setTempDir(self, tempDir):
        self._tempDir = tempDir
    
    def getTempDir(self):
        return self._tempDir
        
    def setJobTable(self, jobTable):
        self._jobTable = jobTable
        
    def setQueue(self, queue):
        self._queue = queue
    
    def getQueue(self):
        return self._queue
        
    def setGroupId(self, groupId):
        self._groupId = groupId
    
    def getGroupId(self):
        return self._groupId
    
    def setAcronym(self, acronym):
        self._acronym = acronym
    
    def getAcronym(self):
        return self._acronym
   
    @staticmethod
    def createParameter(jobdb, groupid, acronym):
	launcherParameter = LauncherParameter(jobdb)
        launcherParameter.setQuery(os.getcwd())
        launcherParameter.setSubject("")
        launcherParameter.setParam("")
        launcherParameter.setCurrentDir(os.getcwd())
        launcherParameter.setTempDir(os.getcwd())
        launcherParameter.setJobTable("")
        launcherParameter.setQueue("")
        launcherParameter.setGroupId(groupid)
        launcherParameter.setAcronym(acronym)
	return launcherParameter       

        
class Launcher2(object):

    #TODO: remove unused parameters : query="", subject="", param="", job_table=""
    def __init__(self, iLauncherParameter):
        jobdb = iLauncherParameter.getJobDB()
        cdir = iLauncherParameter.getCurrentDir()
        if jobdb.__class__.__name__ == "RepetJob":
            self.jobdb = TableJobAdaptatorFactory.createInstance(jobdb, "jobs")
        else:
            self.jobdb = jobdb
        self.jobdb.checkJobTable()
        if cdir == "":
            cdir = os.getcwd()
        self.cdir = cdir
        self.tmpdir = iLauncherParameter.getTempDir()
        self.groupid = iLauncherParameter.getGroupId()
        self.acronyme = iLauncherParameter.getAcronym()
        self._chooseTemplateWithCopy = False
        self._chooseTemplateLight = False
        self.queue, self.lResources = self.getQueueNameAndResources(iLauncherParameter.getQueue())
        self._createJobInstance()
        self._nbJobs = 0
        
    def getQueueNameAndResources(self, configQueue):
        tokens = configQueue.replace("'","").split(" ")
        queueName = ""
        lResources = []
        if tokens[0] != "":
            if re.match(".*\.q", tokens[0]):
                queueName = tokens[0]
                lResources = tokens[1:]
            else:
                lResources = tokens
        return queueName, lResources

    def createGroupidIfItNotExist(self):
        if self.groupid == "":
            self.job.groupid = str(os.getpid())
        else:
            self.job.groupid = self.groupid

    def beginRun( self ):
        self.createGroupidIfItNotExist()
        if self.jobdb.hasUnfinishedJob(self.job.groupid):
            self.jobdb.waitJobGroup(self.job.groupid)
        else:
            self.jobdb.cleanJobGroup(self.job.groupid)

    ## Launch one job in parallel
    #
    # @param cmdStart string command-line for the job to be launched
    # @param cmdFinish string command to retrieve result files
    # @warning the jobname has to be defined outside from this method
    #
    def runSingleJob(self, cmdStart, cmdFinish = "", cmdSize = "", cmdCopy = ""):
        if self._nbJobs == 0:
            self._nbJobs = 1
        pid = str(os.getpid())
        now = time.localtime()
        #TODO: rename ClusterLauncher_ ...
        pyFileName = self.cdir + "/ClusterLauncher_" + self.job.groupid + "_" +\
                     self.job.jobname + "_" + str(now[0]) + "-" + str(now[1]) +\
                     "-" + str(now[2]) + "_" + pid + ".py"
        self.job.launcher = pyFileName
        
        #TODO: to remove when refactoring is done
        cmdStart = self._indentCmd(cmdStart)
        cmdFinish = self._indentCmd(cmdFinish)
        
        iWriteScript = WriteScript(self.job, self.jobdb, self.cdir, self.tmpdir, self._chooseTemplateWithCopy, self._chooseTemplateLight)
        iWriteScript.run(cmdStart, cmdFinish, pyFileName, cmdSize, cmdCopy)
        os.chmod(pyFileName, stat.S_IRWXU+stat.S_IRGRP+stat.S_IXGRP+stat.S_IROTH+stat.S_IXOTH)
        sys.stdout.flush()
        log = self.jobdb.submitJob(self.job)
        if log != 0:
            print "ERROR while submitting job to the cluster"
            sys.exit(1)
        
    def endRun(self, cleanNodes = False):
        string = "waiting for %i job(s) with groupid '%s' (%s)" % (self._nbJobs, self.job.groupid, time.strftime("%Y-%m-%d %H:%M:%S"))
        print string; sys.stdout.flush()
        self.jobdb.waitJobGroup(self.job.groupid)
        if self._nbJobs > 1:
            string = "all jobs with groupid '%s' are finished (%s)" % (self.job.groupid, time.strftime("%Y-%m-%d %H:%M:%S"))
            print string; sys.stdout.flush()

        if cleanNodes:
            string = "start cleaning cluster nodes (%s)" % time.strftime("%Y-%m-%d %H:%M:%S")
            print string; sys.stdout.flush()
            self.cleanNodes()
            string = "end cleaning cluster nodes (%s)" % time.strftime("%Y-%m-%d %H:%M:%S")
            print string; sys.stdout.flush()
            
        statsExecutionTime = self.getStatsOfExecutionTime()
        if self._nbJobs > 1:
            print "execution time of all jobs (seconds): %f" % statsExecutionTime.getSum()
        print "execution time per job: %s" % statsExecutionTime.string()
        sys.stdout.flush()
        self.jobdb.cleanJobGroup(self.job.groupid)
        
    def getStatsOfExecutionTime(self, acronyme = ""):
        stat = Stat()
        if acronyme == "":
            pattern = "%s*.o*" % self.acronyme
        else:
            pattern = "%s*.o*" % acronyme
        lJobFiles = glob.glob(pattern)
        for f in lJobFiles:
            fH = open(f, "r")
            while True:
                line = fH.readline()
                if line == "":
                    break
                if "executionTime" in line:
                    stat.add( float(line[:-1].split("=")[1] ) )
                    break
            fH.close()
        return stat     

    def clean( self, acronyme = "", stdout = True, stderr = True ):
        lFileToRemove = []
        if acronyme == "":
            acronyme = self.acronyme  
        pattern = "ClusterLauncher*%s*.py" % ( acronyme )
        lFileToRemove.extend(glob.glob( pattern ))
        if stdout:
            pattern = "%s*.o*" % ( acronyme )
            lFileToRemove.extend(glob.glob( pattern ))        
        if stderr:
            pattern = "%s*.e*" % ( acronyme )
            lFileToRemove.extend(glob.glob( pattern ))                   
        for file in lFileToRemove:
            os.remove(file)
    
    #TODO: handle of nodesMustBeCleaned => class attribute ?
    def runLauncherForMultipleJobs(self, acronymPrefix, lCmdsTuples, cleanMustBeDone = True, nodesMustBeCleaned = False):
        self.beginRun()
        print "submitting job(s) with groupid '%s' (%s)" % (self.job.groupid,  time.strftime("%Y-%m-%d %H:%M:%S"))
        for cmdsTuple in lCmdsTuples:
            self._nbJobs += 1
            self.acronyme = "%s_%s" % (acronymPrefix, self._nbJobs)
            self.job.jobname = self.acronyme
            if len(cmdsTuple) == 2:
                self.runSingleJob(cmdsTuple[0], cmdsTuple[1])
            else:
                self.runSingleJob(cmdsTuple[0], cmdsTuple[1], cmdsTuple[2], cmdsTuple[3])
                self._createJobInstance()
                self.createGroupidIfItNotExist()
        self.acronyme = acronymPrefix
        self.endRun(nodesMustBeCleaned)
        if cleanMustBeDone:
            self.clean("%s_" % acronymPrefix)
        self.jobdb.close()

    def prepareCommands(self, lCmds, lCmdStart = [], lCmdFinish = [], lCmdSize = [], lCmdCopy = []):
        cmdStart = ""
        for cmd in lCmdStart:
            cmdStart += "%s\n\t" % cmd
        for cmd in lCmds:
            cmdStart += "%s\n\t" % cmd
        cmdFinish = ""
        for cmd in lCmdFinish:
            cmdFinish += "%s\n\t" % cmd
        cmdSize = ""
        for cmd in lCmdSize:
            cmdSize += "%s\n\t\t" % cmd
        cmdCopy = ""
        for cmd in lCmdCopy:
            cmdCopy += "%s\n\t\t" % cmd
        return (cmdStart, cmdFinish, cmdSize, cmdCopy)

    #TODO: to remove when refactoring is done
    def prepareCommands_withoutIndentation(self, lCmds, lCmdStart = [], lCmdFinish = [], lCmdSize = [], lCmdCopy = []):
        cmdStart = ""
        for cmd in lCmdStart:
            cmdStart += "%s\n" % cmd
        for cmd in lCmds:
            cmdStart += "%s\n" % cmd
        cmdFinish = ""
        for cmd in lCmdFinish:
            cmdFinish += "%s\n" % cmd
        cmdSize = ""
        for cmd in lCmdSize:
            cmdSize += "%s\n\t\t" % cmd
        cmdCopy = ""
        for cmd in lCmdCopy:
            cmdCopy += "%s\n\t\t" % cmd
        return (cmdStart, cmdFinish, cmdSize, cmdCopy)
    
    def getSystemCommand(self, prg, lArgs):
        systemCmd = "log = os.system(\"" + prg 
        for arg in lArgs:
            systemCmd += " " + arg
        systemCmd += "\")"
        return systemCmd

    def cleanNodes(self):
        iCleanClusterNodeAfterRepet = CleanClusterNodesAfterRepet()
        iCleanClusterNodeAfterRepet.setLNodes(self.jobdb.getNodesListByGroupId(self.groupid))
        iCleanClusterNodeAfterRepet.setTempDirectory(self.tmpdir)
        iCleanClusterNodeAfterRepet.setPattern("%s*" % self.groupid)
        iCleanClusterNodeAfterRepet.run()

    #TODO: to remove when refactoring is done
    def _indentCmd(self, cmd):
        lCmd = cmd.split("\n")
        cmd_Tab = "%s\n" % lCmd[0]
        for line in lCmd[1:-1]:
            cmd_Tab += "\t%s\n" % line
        return cmd_Tab
    
    def _createJobInstance(self):
        if self.lResources == []:
            #To have mem_free=1G:
            self.job = Job(queue=self.queue)
        else:
            self.job = Job(queue=self.queue, lResources=self.lResources)