CRAB/python/Creator.py

from Actor import Actor
from WorkSpace import WorkSpace
from JobList import JobList
from JobDB import JobDB
from ScriptWriter import ScriptWriter
from Scheduler import Scheduler
from crab_logger import Logger
from crab_exceptions import *
from crab_util import *
import common

import os, string, math
import time

class Creator(Actor):
    def __init__(self, job_type_name, cfg_params, ncjobs):
        self.job_type_name = job_type_name
        self.job_type = None
        self.cfg_params = cfg_params
        self.total_njobs = 0
        self.total_number_of_events = 0
        self.job_number_of_events = 0
        self.first_event = 0
        self.jobParamsList=[]

        self.createJobTypeObject(ncjobs)
        common.logger.debug(5, __name__+": JobType "+self.job_type.name()+" created")

        self.job_type.prepareSteeringCards()
        common.logger.debug(5, __name__+": Steering cards prepared")

        self.total_njobs = self.job_type.numberOfJobs();
        common.logger.debug(5, __name__+": total # of jobs = "+`self.total_njobs`)

        # Set number of jobs to be created
        # --------------------------------->  Boss4: changed, "all" by default 

        self.ncjobs = ncjobs
        if ncjobs == 'all' : self.ncjobs = self.total_njobs
        if ncjobs > self.total_njobs : self.ncjobs = self.total_njobs
        
        # This is code for proto-monitoring
        code=common.taskDB.dict("CODE")
        if self.job_type.name() == 'CMSSW':
            try: 
                self.primaryDataset = cfg_params['CMSSW.datasetpath'].split("/")[1]
            except:
                self.primaryDataset = 'None'
            try:
                self.ProcessedDataset = cfg_params['CMSSW.datasetpath'].split("/")[3]
            except:
                self.ProcessedDataset = '  '
            common.taskDB.setDict("CODE",(code+'::'+str(self.job_type.name())+'::'+str(self.ncjobs)+'::'+str(self.primaryDataset)+'::'+str(self.ProcessedDataset)))
            pass

        # This is code for dashboard
          
        #First checkProxy
        common.scheduler.checkProxy()
        try:
            fl = open(common.work_space.shareDir() + '/' + self.cfg_params['apmon'].fName, 'w')
            self.cfg_params['GridName'] = runCommand("voms-proxy-info -identity")
            common.logger.debug(5, "GRIDNAME: "+self.cfg_params['GridName'])
            taskType = 'analysis'

            params = {'tool': common.prog_name, 'tool_ui': os.environ['HOSTNAME'], \
                      'scheduler': self.cfg_params['CRAB.scheduler'], 'GridName': self.cfg_params['GridName'].strip(), \
                      'taskType': taskType, 'vo': self.cfg_params['EDG.virtual_organization'], 'user': self.cfg_params['user']}
            jtParam = self.job_type.getParams()
            for i in jtParam.iterkeys():
                params[i] = string.strip(jtParam[i])
            for j, k in params.iteritems():
#                print "Values: %s %s"%(j, k)
                fl.write(j + ':' + k + '\n')
            fl.close()
        except:
            exctype, value = sys.exc_info()[:2]
            common.logger.message("Creator::run Exception raised in collection params for dashboard: %s %s"%(exctype, value))
            pass

        #TODO: deprecated code, not needed,
        # will be eliminated when WorkSpace.saveConfiguration()
        # will be improved.
        #
        # Set/Save Job Type name

        try:
            jt = common.taskDB.dict("JobType")
            if self.job_type_name:
                if ( jt != self.job_type_name+'\n' ):
                    msg = 'Job Type mismatch: requested <' + self.job_type_name
                    msg += '>, found <' + jt[:-1] + '>.'
                    raise CrabException(msg)
                pass
            else:
                self.job_type_name = jt[:-1]
                pass
            pass
        except CrabException:
            # Save job type name
            common.taskDB.setDict("JobType",self.job_type_name)
            pass
        #end of deprecated code

        common.logger.debug(5, "Creator constructor finished")
        return

    def writeJobsSpecsToDB(self):
        """
        Write firstEvent and maxEvents in the DB for future use
        """

        self.job_type.split(self.jobParamsList)
        return

    def nJobs(self):
        return self.total_njobs
    
    def createJobTypeObject(self,ncjobs):
        file_name = 'cms_'+ string.lower(self.job_type_name)
        klass_name = string.capitalize(self.job_type_name)

        try:
            klass = importName(file_name, klass_name)
        except KeyError:
            msg = 'No `class '+klass_name+'` found in file `'+file_name+'.py`'
            raise CrabException(msg)
        except ImportError, e:
            msg = 'Cannot create job type '+self.job_type_name
            msg += ' (file: '+file_name+', class '+klass_name+'):\n'
            msg += str(e)
            raise CrabException(msg)

        self.job_type = klass(self.cfg_params,ncjobs)
        return

    def jobType(self):
        return self.job_type

    def run(self):
        """
        The main method of the class.
        """

        common.logger.debug(5, "Creator::run() called")
        start = time.time()

        # Instantiate ScriptWriter

        script_writer = ScriptWriter('crab_template.sh')

        # Loop over jobs
        argsList = []

        njc = 0
        block = -1 # first block is 0
        lastDest=''
        for nj in range(self.total_njobs):
            if njc == self.ncjobs : break
            st = common.jobDB.status(nj)
            if st != 'X': continue

            common.logger.debug(1,"Creating job # "+`(nj+1)`)

            # Prepare configuration file

            self.job_type.modifySteeringCards(nj)
            # Create script (sh)
            script_writer.modifyTemplateScript()
            os.chmod(common.taskDB.dict("ScriptName"), 0744)

            # Create XML script and declare related jobs 
            argsList.append( str(nj+1)+' '+ self.job_type.getJobTypeArguments(nj, self.cfg_params['CRAB.scheduler']) )

            common.jobDB.setStatus(nj, 'C')
            # common: write input and output sandbox
            common.jobDB.setInputSandbox(nj, self.job_type.inputSandbox(nj))

            outputSandbox=self.job_type.outputSandbox(nj)
            # check if out!=err
            common.jobDB.setOutputSandbox(nj, outputSandbox)
            common.jobDB.setTaskId(nj, self.cfg_params['taskId'])

            currDest=common.jobDB.destination(nj)
            if (currDest!=lastDest):
                block+=1
                lastDest = currDest
                pass
            common.jobDB.setBlock(nj,block)
            njc = njc + 1
            pass

        ####
        common.scheduler.createXMLSchScript(self.total_njobs, argsList)
        common.logger.message('Creating '+str(self.total_njobs)+' jobs, please wait...')
        common.scheduler.declareJob_()   #Add for BOSS4

        stop = time.time()
        common.logger.debug(1, "Creation Time: "+str(stop - start))
        common.logger.write("Creation Time: "+str(stop - start))

        common.jobDB.save()
        common.taskDB.save()
        msg = '\nTotal of %d jobs created'%njc
        if njc != self.ncjobs: msg = msg + ' from %d requested'%self.ncjobs
        msg = msg + '.\n'
        common.logger.message(msg)
        
        return
Revision:	1.46
Committed:	Fri Oct 6 15:41:24 2006 UTC (18 years, 6 months ago) by slacapra
Content type:	text/x-python
Branch:	MAIN
CVS Tags:	CRAB_1_4_0_pre3, CRAB_1_4_0_pre2
Changes since 1.45:	+8 -0 lines
Log Message:	add block in JobDB and recover SchedulerEdg
#	User	Rev	Content
1	nsmirnov	1.1	from Actor import Actor
2			from WorkSpace import WorkSpace
3			from JobList import JobList
4			from JobDB import JobDB
5			from ScriptWriter import ScriptWriter
6			from Scheduler import Scheduler
7			from crab_logger import Logger
8			from crab_exceptions import *
9			from crab_util import *
10			import common
11
12	slacapra	1.8	import os, string, math
13	slacapra	1.45	import time
14
15	nsmirnov	1.1	class Creator(Actor):
16	corvo	1.30	def __init__(self, job_type_name, cfg_params, ncjobs):
17	nsmirnov	1.1	self.job_type_name = job_type_name
18	corvo	1.30	self.job_type = None
19	nsmirnov	1.1	self.cfg_params = cfg_params
20	nsmirnov	1.2	self.total_njobs = 0
21			self.total_number_of_events = 0
22			self.job_number_of_events = 0
23	slacapra	1.7	self.first_event = 0
24	gutsche	1.38	self.jobParamsList=[]
25	slacapra	1.14
26	gutsche	1.42	self.createJobTypeObject(ncjobs)
27	nsmirnov	1.3	common.logger.debug(5, __name__+": JobType "+self.job_type.name()+" created")
28	nsmirnov	1.1
29			self.job_type.prepareSteeringCards()
30	nsmirnov	1.3	common.logger.debug(5, __name__+": Steering cards prepared")
31	nsmirnov	1.1
32	gutsche	1.38	self.total_njobs = self.job_type.numberOfJobs();
33	nsmirnov	1.3	common.logger.debug(5, __name__+": total # of jobs = "+`self.total_njobs`)
34	nsmirnov	1.1
35			# Set number of jobs to be created
36	slacapra	1.44	# ---------------------------------> Boss4: changed, "all" by default
37	nsmirnov	1.1
38			self.ncjobs = ncjobs
39			if ncjobs == 'all' : self.ncjobs = self.total_njobs
40			if ncjobs > self.total_njobs : self.ncjobs = self.total_njobs
41	spiga	1.11
42	slacapra	1.35	# This is code for proto-monitoring
43	slacapra	1.43	code=common.taskDB.dict("CODE")
44			if self.job_type.name() == 'CMSSW':
45	spiga	1.41	try:
46			self.primaryDataset = cfg_params['CMSSW.datasetpath'].split("/")[1]
47			except:
48			self.primaryDataset = 'None'
49			try:
50			self.ProcessedDataset = cfg_params['CMSSW.datasetpath'].split("/")[3]
51			except:
52			self.ProcessedDataset = ' '
53	slacapra	1.43	common.taskDB.setDict("CODE",(code+'::'+str(self.job_type.name())+'::'+str(self.ncjobs)+'::'+str(self.primaryDataset)+'::'+str(self.ProcessedDataset)))
54	spiga	1.37	pass
55
56	slacapra	1.35	# This is code for dashboard
57	spiga	1.36
58			#First checkProxy
59			common.scheduler.checkProxy()
60	corvo	1.33	try:
61			fl = open(common.work_space.shareDir() + '/' + self.cfg_params['apmon'].fName, 'w')
62			self.cfg_params['GridName'] = runCommand("voms-proxy-info -identity")
63			common.logger.debug(5, "GRIDNAME: "+self.cfg_params['GridName'])
64			taskType = 'analysis'
65
66			params = {'tool': common.prog_name, 'tool_ui': os.environ['HOSTNAME'], \
67			'scheduler': self.cfg_params['CRAB.scheduler'], 'GridName': self.cfg_params['GridName'].strip(), \
68			'taskType': taskType, 'vo': self.cfg_params['EDG.virtual_organization'], 'user': self.cfg_params['user']}
69			jtParam = self.job_type.getParams()
70			for i in jtParam.iterkeys():
71			params[i] = string.strip(jtParam[i])
72			for j, k in params.iteritems():
73			# print "Values: %s %s"%(j, k)
74			fl.write(j + ':' + k + '\n')
75			fl.close()
76			except:
77			exctype, value = sys.exc_info()[:2]
78	slacapra	1.35	common.logger.message("Creator::run Exception raised in collection params for dashboard: %s %s"%(exctype, value))
79	corvo	1.33	pass
80	nsmirnov	1.1
81			#TODO: deprecated code, not needed,
82			# will be eliminated when WorkSpace.saveConfiguration()
83			# will be improved.
84			#
85			# Set/Save Job Type name
86
87	slacapra	1.43	try:
88			jt = common.taskDB.dict("JobType")
89	nsmirnov	1.1	if self.job_type_name:
90			if ( jt != self.job_type_name+'\n' ):
91			msg = 'Job Type mismatch: requested <' + self.job_type_name
92			msg += '>, found <' + jt[:-1] + '>.'
93			raise CrabException(msg)
94			pass
95			else:
96			self.job_type_name = jt[:-1]
97			pass
98			pass
99	slacapra	1.43	except CrabException:
100	nsmirnov	1.1	# Save job type name
101	slacapra	1.43	common.taskDB.setDict("JobType",self.job_type_name)
102	nsmirnov	1.1	pass
103			#end of deprecated code
104
105	nsmirnov	1.3	common.logger.debug(5, "Creator constructor finished")
106	nsmirnov	1.1	return
107
108	slacapra	1.7	def writeJobsSpecsToDB(self):
109			"""
110			Write firstEvent and maxEvents in the DB for future use
111			"""
112
113	gutsche	1.38	self.job_type.split(self.jobParamsList)
114	nsmirnov	1.2	return
115
116	nsmirnov	1.1	def nJobs(self):
117			return self.total_njobs
118
119	gutsche	1.42	def createJobTypeObject(self,ncjobs):
120	nsmirnov	1.1	file_name = 'cms_'+ string.lower(self.job_type_name)
121			klass_name = string.capitalize(self.job_type_name)
122
123			try:
124			klass = importName(file_name, klass_name)
125			except KeyError:
126			msg = 'No `class '+klass_name+'` found in file `'+file_name+'.py`'
127			raise CrabException(msg)
128			except ImportError, e:
129			msg = 'Cannot create job type '+self.job_type_name
130			msg += ' (file: '+file_name+', class '+klass_name+'):\n'
131			msg += str(e)
132			raise CrabException(msg)
133
134	gutsche	1.42	self.job_type = klass(self.cfg_params,ncjobs)
135	nsmirnov	1.1	return
136	nsmirnov	1.5
137			def jobType(self):
138			return self.job_type
139	nsmirnov	1.1
140			def run(self):
141	nsmirnov	1.4	"""
142			The main method of the class.
143			"""
144	nsmirnov	1.1
145	nsmirnov	1.3	common.logger.debug(5, "Creator::run() called")
146	slacapra	1.45	start = time.time()
147	nsmirnov	1.3
148	nsmirnov	1.1	# Instantiate ScriptWriter
149
150			script_writer = ScriptWriter('crab_template.sh')
151
152			# Loop over jobs
153	slacapra	1.44	argsList = []
154	nsmirnov	1.1
155			njc = 0
156	slacapra	1.46	block = -1 # first block is 0
157			lastDest=''
158	nsmirnov	1.4	for nj in range(self.total_njobs):
159	nsmirnov	1.1	if njc == self.ncjobs : break
160	nsmirnov	1.4	st = common.jobDB.status(nj)
161	nsmirnov	1.1	if st != 'X': continue
162
163	slacapra	1.44	common.logger.debug(1,"Creating job # "+`(nj+1)`)
164	nsmirnov	1.3
165	nsmirnov	1.1	# Prepare configuration file
166
167	nsmirnov	1.4	self.job_type.modifySteeringCards(nj)
168	fanzago	1.12	# Create script (sh)
169	slacapra	1.43	script_writer.modifyTemplateScript()
170			os.chmod(common.taskDB.dict("ScriptName"), 0744)
171	nsmirnov	1.1
172	slacapra	1.44	# Create XML script and declare related jobs
173			argsList.append( str(nj+1)+' '+ self.job_type.getJobTypeArguments(nj, self.cfg_params['CRAB.scheduler']) )
174	fanzago	1.12
175	nsmirnov	1.4	common.jobDB.setStatus(nj, 'C')
176	slacapra	1.9	# common: write input and output sandbox
177			common.jobDB.setInputSandbox(nj, self.job_type.inputSandbox(nj))
178
179			outputSandbox=self.job_type.outputSandbox(nj)
180			# check if out!=err
181			common.jobDB.setOutputSandbox(nj, outputSandbox)
182	corvo	1.18	common.jobDB.setTaskId(nj, self.cfg_params['taskId'])
183	slacapra	1.9
184	slacapra	1.46	currDest=common.jobDB.destination(nj)
185			if (currDest!=lastDest):
186			block+=1
187			lastDest = currDest
188			pass
189			common.jobDB.setBlock(nj,block)
190	nsmirnov	1.1	njc = njc + 1
191			pass
192
193			####
194	slacapra	1.44	common.scheduler.createXMLSchScript(self.total_njobs, argsList)
195			common.logger.message('Creating '+str(self.total_njobs)+' jobs, please wait...')
196			common.scheduler.declareJob_() #Add for BOSS4
197	fanzago	1.12
198	slacapra	1.45	stop = time.time()
199			common.logger.debug(1, "Creation Time: "+str(stop - start))
200			common.logger.write("Creation Time: "+str(stop - start))
201
202	nsmirnov	1.1	common.jobDB.save()
203	slacapra	1.43	common.taskDB.save()
204	nsmirnov	1.1	msg = '\nTotal of %d jobs created'%njc
205			if njc != self.ncjobs: msg = msg + ' from %d requested'%self.ncjobs
206			msg = msg + '.\n'
207			common.logger.message(msg)
208	slacapra	1.45
209	nsmirnov	1.1	return