CRAB/python/Creator.py

from Actor import Actor
from WorkSpace import WorkSpace
from JobList import JobList
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`)

        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 dashboard
        #First checkProxy
        common.scheduler.checkProxy()
        try:
            gridName = string.strip(common.scheduler.userName())
            common.logger.debug(5, "GRIDNAME: "+gridName)
            taskType = 'analysis'
            VO = cfg_params.get('EDG.virtual_organization','cms')

            params = {'tool': common.prog_name,\
                      'JSToolVersion': common.prog_version_str, \
                      'tool_ui': os.environ['HOSTNAME'], \
                      'scheduler': common.scheduler.name(), \
                      'GridName': gridName, \
                      'taskType': taskType, \
                      'vo': VO, \
                      'user': os.environ['USER']}
            jtParam = self.job_type.getParams()
            for i in jtParam.iterkeys():
                params[i] = string.strip(jtParam[i])
            fl = open(common.work_space.shareDir() + '/' + common.apmon.fName, 'w')
            for j, k in params.iteritems():
                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

        # Set/Save Job Type name
        self.job_type_name = common._db.queryTask("jobType") ## New BL--DS

        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 = None
        if self.cfg_params['CRAB.scheduler'].find("glit") != -1: ## checking scheduler: if glite(coll) output_sandbox will be limited
            script_writer = ScriptWriter('crab_template.sh', 1) ## flag that indicates if limit or not
        else:
            script_writer = ScriptWriter('crab_template.sh', 0)

        # Loop over jobs
        argsList = []
        njc = 0
#        block = -1 # first block is 0  BL--DS
#        lastDest=''
        for nj in range(self.total_njobs):
            if njc == self.ncjobs : break

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

            # Prepare configuration file

            self.job_type.modifySteeringCards(nj)

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

            run_jobToSave = {'status' :'C'}
            common._db.updateRunJob_(nj, run_jobToSave ) ## New BL--DS

            outputSandbox=self.job_type.outputSandbox(nj)
            # check if out!=err
## New BL--DS
#            common.jobDB.setOutputSandbox(nj, outputSandbox)
#            common.jobDB.setTaskId(nj, common.taskDB.dict('taskId'))

            job_ToSave={'outputFiles': outputSandbox}

            common._db.updateJob_(nj, job_ToSave ) ## Nes BL--DS

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

        # Create script (sh)
        script_writer.modifyTemplateScript()
        os.chmod(common._db.queryTask('scriptName'), 0744) ## Modified BL--DS
       # common: write input sandbox --- This is now a task attribute... not per job ## BL--DS
        isb = {'globalSandbox': self.job_type.inputSandbox(1)}
        common._db.updateTask_(isb) 
        ####
        common.scheduler.declare(self.total_njobs ) 

        common.scheduler.sched_parameter()

        common.logger.message('Creating '+str(self.total_njobs)+' jobs, please wait...')

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

        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.59
Committed:	Fri Mar 7 17:25:23 2008 UTC (17 years, 1 month ago) by slacapra
Content type:	text/x-python
Branch:	MAIN
Changes since 1.58:	+1 -1 lines
Log Message:	initial mods to have LSF working with BossLite: still way to go...
#	User	Rev	Content
1	nsmirnov	1.1	from Actor import Actor
2			from WorkSpace import WorkSpace
3			from JobList import JobList
4			from ScriptWriter import ScriptWriter
5			from Scheduler import Scheduler
6			from crab_logger import Logger
7			from crab_exceptions import *
8			from crab_util import *
9			import common
10
11	slacapra	1.8	import os, string, math
12	slacapra	1.45	import time
13
14	nsmirnov	1.1	class Creator(Actor):
15	corvo	1.30	def __init__(self, job_type_name, cfg_params, ncjobs):
16	nsmirnov	1.1	self.job_type_name = job_type_name
17	corvo	1.30	self.job_type = None
18	nsmirnov	1.1	self.cfg_params = cfg_params
19	nsmirnov	1.2	self.total_njobs = 0
20			self.total_number_of_events = 0
21			self.job_number_of_events = 0
22	slacapra	1.7	self.first_event = 0
23	gutsche	1.38	self.jobParamsList=[]
24	mcinquil	1.52
25	gutsche	1.42	self.createJobTypeObject(ncjobs)
26	nsmirnov	1.3	common.logger.debug(5, __name__+": JobType "+self.job_type.name()+" created")
27	nsmirnov	1.1
28			self.job_type.prepareSteeringCards()
29	nsmirnov	1.3	common.logger.debug(5, __name__+": Steering cards prepared")
30	nsmirnov	1.1
31	gutsche	1.38	self.total_njobs = self.job_type.numberOfJobs();
32	nsmirnov	1.3	common.logger.debug(5, __name__+": total # of jobs = "+`self.total_njobs`)
33	nsmirnov	1.1
34			self.ncjobs = ncjobs
35			if ncjobs == 'all' : self.ncjobs = self.total_njobs
36			if ncjobs > self.total_njobs : self.ncjobs = self.total_njobs
37	spiga	1.11
38	spiga	1.50
39	slacapra	1.35	# This is code for dashboard
40	spiga	1.36	#First checkProxy
41			common.scheduler.checkProxy()
42	corvo	1.33	try:
43	spiga	1.56	gridName = string.strip(common.scheduler.userName())
44	slacapra	1.53	common.logger.debug(5, "GRIDNAME: "+gridName)
45	corvo	1.33	taskType = 'analysis'
46	slacapra	1.54	VO = cfg_params.get('EDG.virtual_organization','cms')
47	corvo	1.33
48	slacapra	1.47	params = {'tool': common.prog_name,\
49			'JSToolVersion': common.prog_version_str, \
50			'tool_ui': os.environ['HOSTNAME'], \
51	slacapra	1.54	'scheduler': common.scheduler.name(), \
52	slacapra	1.53	'GridName': gridName, \
53	slacapra	1.47	'taskType': taskType, \
54	slacapra	1.48	'vo': VO, \
55	slacapra	1.54	'user': os.environ['USER']}
56	corvo	1.33	jtParam = self.job_type.getParams()
57			for i in jtParam.iterkeys():
58			params[i] = string.strip(jtParam[i])
59	slacapra	1.55	fl = open(common.work_space.shareDir() + '/' + common.apmon.fName, 'w')
60	corvo	1.33	for j, k in params.iteritems():
61			fl.write(j + ':' + k + '\n')
62			fl.close()
63			except:
64			exctype, value = sys.exc_info()[:2]
65	slacapra	1.35	common.logger.message("Creator::run Exception raised in collection params for dashboard: %s %s"%(exctype, value))
66	corvo	1.33	pass
67	nsmirnov	1.1
68			# Set/Save Job Type name
69	spiga	1.57	self.job_type_name = common._db.queryTask("jobType") ## New BL--DS
70	nsmirnov	1.1
71	nsmirnov	1.3	common.logger.debug(5, "Creator constructor finished")
72	nsmirnov	1.1	return
73
74	slacapra	1.7	def writeJobsSpecsToDB(self):
75			"""
76			Write firstEvent and maxEvents in the DB for future use
77			"""
78
79	gutsche	1.38	self.job_type.split(self.jobParamsList)
80	nsmirnov	1.2	return
81
82	nsmirnov	1.1	def nJobs(self):
83			return self.total_njobs
84
85	gutsche	1.42	def createJobTypeObject(self,ncjobs):
86	nsmirnov	1.1	file_name = 'cms_'+ string.lower(self.job_type_name)
87			klass_name = string.capitalize(self.job_type_name)
88
89			try:
90			klass = importName(file_name, klass_name)
91			except KeyError:
92			msg = 'No `class '+klass_name+'` found in file `'+file_name+'.py`'
93			raise CrabException(msg)
94			except ImportError, e:
95			msg = 'Cannot create job type '+self.job_type_name
96			msg += ' (file: '+file_name+', class '+klass_name+'):\n'
97			msg += str(e)
98			raise CrabException(msg)
99
100	gutsche	1.42	self.job_type = klass(self.cfg_params,ncjobs)
101	nsmirnov	1.1	return
102	nsmirnov	1.5
103			def jobType(self):
104			return self.job_type
105	nsmirnov	1.1
106			def run(self):
107	nsmirnov	1.4	"""
108			The main method of the class.
109			"""
110	nsmirnov	1.1
111	nsmirnov	1.3	common.logger.debug(5, "Creator::run() called")
112	slacapra	1.45	start = time.time()
113	nsmirnov	1.1	# Instantiate ScriptWriter
114	mcinquil	1.51	script_writer = None
115			if self.cfg_params['CRAB.scheduler'].find("glit") != -1: ## checking scheduler: if glite(coll) output_sandbox will be limited
116			script_writer = ScriptWriter('crab_template.sh', 1) ## flag that indicates if limit or not
117			else:
118			script_writer = ScriptWriter('crab_template.sh', 0)
119	nsmirnov	1.1
120			# Loop over jobs
121	slacapra	1.44	argsList = []
122	nsmirnov	1.1	njc = 0
123	spiga	1.57	# block = -1 # first block is 0 BL--DS
124			# lastDest=''
125	nsmirnov	1.4	for nj in range(self.total_njobs):
126	nsmirnov	1.1	if njc == self.ncjobs : break
127
128	slacapra	1.44	common.logger.debug(1,"Creating job # "+`(nj+1)`)
129	nsmirnov	1.3
130	nsmirnov	1.1	# Prepare configuration file
131
132	nsmirnov	1.4	self.job_type.modifySteeringCards(nj)
133	mcinquil	1.52
134			# Create XML script and declare related jobs
135			argsList.append( str(nj+1)+' '+ self.job_type.getJobTypeArguments(nj, self.cfg_params['CRAB.scheduler']) )
136			self.job_type.setArgsList(argsList)
137
138	spiga	1.57	run_jobToSave = {'status' :'C'}
139	slacapra	1.59	common._db.updateRunJob_(nj, run_jobToSave ) ## New BL--DS
140	slacapra	1.9
141			outputSandbox=self.job_type.outputSandbox(nj)
142			# check if out!=err
143	spiga	1.57	## New BL--DS
144			# common.jobDB.setOutputSandbox(nj, outputSandbox)
145			# common.jobDB.setTaskId(nj, common.taskDB.dict('taskId'))
146
147			job_ToSave={'outputFiles': outputSandbox}
148
149			common._db.updateJob_(nj, job_ToSave ) ## Nes BL--DS
150
151			# currDest=common.jobDB.destination(nj)
152			# if (currDest!=lastDest):
153			# block+=1
154			# lastDest = currDest
155			# pass
156			# common.jobDB.setBlock(nj,block)
157	nsmirnov	1.1	njc = njc + 1
158			pass
159	spiga	1.57
160			# Create script (sh)
161			script_writer.modifyTemplateScript()
162			os.chmod(common._db.queryTask('scriptName'), 0744) ## Modified BL--DS
163			# common: write input sandbox --- This is now a task attribute... not per job ## BL--DS
164			isb = {'globalSandbox': self.job_type.inputSandbox(1)}
165			common._db.updateTask_(isb)
166	nsmirnov	1.1	####
167	spiga	1.57	common.scheduler.declare(self.total_njobs )
168
169	spiga	1.58	common.scheduler.sched_parameter()
170	spiga	1.57
171	slacapra	1.44	common.logger.message('Creating '+str(self.total_njobs)+' jobs, please wait...')
172	fanzago	1.12
173	slacapra	1.45	stop = time.time()
174	slacapra	1.47	common.logger.debug(2, "Creation Time: "+str(stop - start))
175	slacapra	1.45	common.logger.write("Creation Time: "+str(stop - start))
176
177	nsmirnov	1.1	msg = '\nTotal of %d jobs created'%njc
178			if njc != self.ncjobs: msg = msg + ' from %d requested'%self.ncjobs
179			msg = msg + '.\n'
180			common.logger.message(msg)
181	slacapra	1.45
182	nsmirnov	1.1	return