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

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.ncjobs = 0                  # nb of jobs to be created
        self.total_number_of_events = 0
        self.job_number_of_events = 0
        self.first_event = 0
        
        #

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

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

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

        # Set number of jobs to be created

        self.ncjobs = ncjobs
        if ncjobs == 'all' : self.ncjobs = self.total_njobs
        if ncjobs > self.total_njobs : self.ncjobs = self.total_njobs

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

        jt_fname = common.work_space.shareDir() + 'jobtype'
        if os.path.exists(jt_fname):
            # Read stored job type name
            jt_file = open(jt_fname, 'r')
            jt = jt_file.read()
            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
            jt_file.close()
            pass
        else:
            # Save job type name
            jt_file = open(jt_fname, 'w')
            jt_file.write(self.job_type_name+'\n')
            jt_file.close()
            pass
        #end of deprecated code

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

    def defineTotalNumberOfJobs_(self):
        """
        Calculates the total number of jobs to be created.
        """

        try:
            self.first_event = int(self.cfg_params['USER.first_event'])
        except KeyError:
            self.first_event = 0
        common.logger.debug(1,"First event ot be analyzed: "+str(self.first_event))

        maxAvailableEvents = int(self.job_type.maxEvents)
        common.logger.debug(1,"Available events: "+str(maxAvailableEvents))

        # some sanity check
        if self.first_event>=maxAvailableEvents:
            raise CrabException('First event is bigger than maximum number of available events!')

        # the total number of events to be analyzed
        try:
            n = self.cfg_params['USER.total_number_of_events']
            if n == 'all': n = '-1'
            if n == '-1':
                self.total_number_of_events = (maxAvailableEvents - self.first_event)
                common.logger.debug(1,"Analysing all available events "+str(self.total_number_of_events))
            else:
                if maxAvailableEvents<(int(n)+self.first_event):
                    raise CrabException('(First event + total events)='+str(int(n)+self.first_event)+' is bigger than maximum number of available events '+str(maxAvailableEvents)+' !!')
                self.total_number_of_events = int(n)
        except KeyError:
            common.logger.message("total_number_of_events not defined, set it to maximum available")
            self.total_number_of_events = (maxAvailableEvents - self.first_event)
            pass
        common.logger.message("Total number of events to be analyzed: "+str(self.total_number_of_events))


        # read user directives
        eventPerJob=0
        try:
            eventPerJob = self.cfg_params['USER.job_number_of_events']
        except KeyError:
            pass
        
        jobsPerTask=0
        try:
            jobsPerTask = int(self.cfg_params['USER.total_number_of_jobs'])
        except KeyError:
            pass

        # If both the above set, complain and use event per jobs
        if eventPerJob>0 and jobsPerTask>0:
            msg = 'Warning. '
            msg += 'job_number_of_events and total_number_of_jobs are both defined '
            msg += 'Using job_number_of_events.'
            common.logger.message(msg)
            jobsPerTask = 0 
        if eventPerJob==0 and jobsPerTask==0:
            msg = 'Warning. '
            msg += 'job_number_of_events and total_number_of_jobs are not defined '
            msg += 'Creating just one job for all events.'
            common.logger.message(msg)
            jobsPerTask = 1 

        # first case: events per job defined
        if eventPerJob>0:
            n=eventPerJob
            if n == 'all' or n == '-1' or (int(n)>self.total_number_of_events and self.total_number_of_events>0):
                common.logger.message("Asking more events than available: set it to maximum available")
                self.job_number_of_events = self.total_number_of_events
                self.total_njobs = 1
            else:
                self.job_number_of_events = int(n)
                self.total_njobs = int((self.total_number_of_events-1)/self.job_number_of_events)+1
        # second case: jobs per task defined
        elif jobsPerTask>0:
            common.logger.debug(2,"total number of events: "+str(self.total_number_of_events)+" JobPerTask "+str(jobsPerTask))
            self.job_number_of_events = int(math.floor((self.total_number_of_events)/jobsPerTask))
            self.total_njobs = jobsPerTask
        # should not happen...
        else:
            raise CrabException('Somthing wrong with splitting')

        common.logger.debug(2,"total number of events: "+str(self.total_number_of_events)+
            " events per job: "+str(self.job_number_of_events))

        return

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

        common.jobDB.load()
        # case one: write first and max events
        nJobs=self.nJobs()

        firstEvent=self.first_event
        # last jobs is different...
        for job in range(nJobs-1):
            common.jobDB.setFirstEvent(job, firstEvent)
            common.jobDB.setMaxEvents(job, self.job_number_of_events)
            firstEvent=firstEvent+self.job_number_of_events

        # this is the last job
        common.jobDB.setFirstEvent(nJobs-1, firstEvent)
        lastJobsNumberOfEvents= (self.total_number_of_events+self.first_event)-firstEvent
        common.jobDB.setMaxEvents(nJobs-1, lastJobsNumberOfEvents)
    
        common.logger.message('Created '+str(self.total_njobs-1)+' jobs for '+str(self.job_number_of_events)+' each plus 1 for '+str(lastJobsNumberOfEvents)+' for a total of '+str(self.job_number_of_events*(self.total_njobs-1)+lastJobsNumberOfEvents)+' events')
        common.jobDB.save()

        # case two (to be implemented) write eventCollections for each jobs

        return

    def nJobs(self):
        return self.total_njobs
    
    def createJobTypeObject(self):
        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)
        return

    def jobType(self):
        return self.job_type

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

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

        # Instantiate ScriptWriter

        script_writer = ScriptWriter('crab_template.sh')

        # Loop over jobs

        njc = 0
        for nj in range(self.total_njobs):
            if njc == self.ncjobs : break
            st = common.jobDB.status(nj)
            if st != 'X': continue

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

            # Prepare configuration file

            self.job_type.modifySteeringCards(nj)

            # Create JDL
            # Maybe, it worths to move this call into Submitter,
            # i.e. to create scheduler-specific file at submission time ?

            common.scheduler.createJDL(nj)

            # Create script

            script_writer.modifyTemplateScript(nj)
            os.chmod(common.job_list[nj].scriptFilename(), 0744)

            common.jobDB.setStatus(nj, 'C')
            njc = njc + 1
            pass

        ####
        
        common.jobDB.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.8
Committed:	Wed Jul 20 12:01:43 2005 UTC (19 years, 9 months ago) by slacapra
Content type:	text/x-python
Branch:	MAIN
Changes since 1.7:	+29 -18 lines
Log Message:	about splitting
#	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	nsmirnov	1.1
14			class Creator(Actor):
15			def __init__(self, job_type_name, cfg_params, ncjobs):
16			self.job_type_name = job_type_name
17			self.job_type = None
18			self.cfg_params = cfg_params
19	nsmirnov	1.2	self.total_njobs = 0
20			self.ncjobs = 0 # nb of jobs to be created
21			self.total_number_of_events = 0
22			self.job_number_of_events = 0
23	slacapra	1.7	self.first_event = 0
24	nsmirnov	1.2
25	nsmirnov	1.1	#
26
27			self.createJobTypeObject()
28	nsmirnov	1.3	common.logger.debug(5, __name__+": JobType "+self.job_type.name()+" created")
29	nsmirnov	1.1
30			self.job_type.prepareSteeringCards()
31	nsmirnov	1.3	common.logger.debug(5, __name__+": Steering cards prepared")
32	nsmirnov	1.1
33	nsmirnov	1.2	self.defineTotalNumberOfJobs_()
34	nsmirnov	1.3	common.logger.debug(5, __name__+": total # of jobs = "+`self.total_njobs`)
35	nsmirnov	1.1
36			# Set number of jobs to be created
37
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
42			#TODO: deprecated code, not needed,
43			# will be eliminated when WorkSpace.saveConfiguration()
44			# will be improved.
45			#
46			# Set/Save Job Type name
47
48			jt_fname = common.work_space.shareDir() + 'jobtype'
49			if os.path.exists(jt_fname):
50			# Read stored job type name
51			jt_file = open(jt_fname, 'r')
52			jt = jt_file.read()
53			if self.job_type_name:
54			if ( jt != self.job_type_name+'\n' ):
55			msg = 'Job Type mismatch: requested <' + self.job_type_name
56			msg += '>, found <' + jt[:-1] + '>.'
57			raise CrabException(msg)
58			pass
59			else:
60			self.job_type_name = jt[:-1]
61			pass
62			jt_file.close()
63			pass
64			else:
65			# Save job type name
66			jt_file = open(jt_fname, 'w')
67			jt_file.write(self.job_type_name+'\n')
68			jt_file.close()
69			pass
70			#end of deprecated code
71
72	nsmirnov	1.3	common.logger.debug(5, "Creator constructor finished")
73	nsmirnov	1.1	return
74
75	nsmirnov	1.2	def defineTotalNumberOfJobs_(self):
76			"""
77			Calculates the total number of jobs to be created.
78			"""
79
80			try:
81	slacapra	1.8	self.first_event = int(self.cfg_params['USER.first_event'])
82	slacapra	1.7	except KeyError:
83			self.first_event = 0
84	slacapra	1.8	common.logger.debug(1,"First event ot be analyzed: "+str(self.first_event))
85	slacapra	1.7
86	slacapra	1.8	maxAvailableEvents = int(self.job_type.maxEvents)
87			common.logger.debug(1,"Available events: "+str(maxAvailableEvents))
88	slacapra	1.7
89	slacapra	1.8	# some sanity check
90			if self.first_event>=maxAvailableEvents:
91	slacapra	1.7	raise CrabException('First event is bigger than maximum number of available events!')
92
93	slacapra	1.8	# the total number of events to be analyzed
94	slacapra	1.7	try:
95	nsmirnov	1.2	n = self.cfg_params['USER.total_number_of_events']
96			if n == 'all': n = '-1'
97	slacapra	1.8	if n == '-1':
98			self.total_number_of_events = (maxAvailableEvents - self.first_event)
99			common.logger.debug(1,"Analysing all available events "+str(self.total_number_of_events))
100			else:
101			if maxAvailableEvents<(int(n)+self.first_event):
102			raise CrabException('(First event + total events)='+str(int(n)+self.first_event)+' is bigger than maximum number of available events '+str(maxAvailableEvents)+' !!')
103			self.total_number_of_events = int(n)
104	slacapra	1.7	except KeyError:
105			common.logger.message("total_number_of_events not defined, set it to maximum available")
106	slacapra	1.8	self.total_number_of_events = (maxAvailableEvents - self.first_event)
107	slacapra	1.7	pass
108	slacapra	1.8	common.logger.message("Total number of events to be analyzed: "+str(self.total_number_of_events))
109
110	slacapra	1.7
111	slacapra	1.8	# read user directives
112	slacapra	1.7	eventPerJob=0
113			try:
114			eventPerJob = self.cfg_params['USER.job_number_of_events']
115	nsmirnov	1.2	except KeyError:
116			pass
117
118	slacapra	1.7	jobsPerTask=0
119	nsmirnov	1.2	try:
120	slacapra	1.7	jobsPerTask = int(self.cfg_params['USER.total_number_of_jobs'])
121	nsmirnov	1.2	except KeyError:
122	slacapra	1.7	pass
123
124			# If both the above set, complain and use event per jobs
125			if eventPerJob>0 and jobsPerTask>0:
126	nsmirnov	1.2	msg = 'Warning. '
127	slacapra	1.7	msg += 'job_number_of_events and total_number_of_jobs are both defined '
128			msg += 'Using job_number_of_events.'
129	nsmirnov	1.2	common.logger.message(msg)
130	slacapra	1.7	jobsPerTask = 0
131			if eventPerJob==0 and jobsPerTask==0:
132			msg = 'Warning. '
133			msg += 'job_number_of_events and total_number_of_jobs are not defined '
134			msg += 'Creating just one job for all events.'
135			common.logger.message(msg)
136			jobsPerTask = 1
137
138			# first case: events per job defined
139			if eventPerJob>0:
140			n=eventPerJob
141			if n == 'all' or n == '-1' or (int(n)>self.total_number_of_events and self.total_number_of_events>0):
142			common.logger.message("Asking more events than available: set it to maximum available")
143			self.job_number_of_events = self.total_number_of_events
144	slacapra	1.8	self.total_njobs = 1
145	slacapra	1.7	else:
146			self.job_number_of_events = int(n)
147	slacapra	1.8	self.total_njobs = int((self.total_number_of_events-1)/self.job_number_of_events)+1
148	slacapra	1.7	# second case: jobs per task defined
149			elif jobsPerTask>0:
150	slacapra	1.8	common.logger.debug(2,"total number of events: "+str(self.total_number_of_events)+" JobPerTask "+str(jobsPerTask))
151			self.job_number_of_events = int(math.floor((self.total_number_of_events)/jobsPerTask))
152			self.total_njobs = jobsPerTask
153	slacapra	1.7	# should not happen...
154			else:
155			raise CrabException('Somthing wrong with splitting')
156
157	slacapra	1.8	common.logger.debug(2,"total number of events: "+str(self.total_number_of_events)+
158	slacapra	1.7	" events per job: "+str(self.job_number_of_events))
159	slacapra	1.8
160	slacapra	1.7	return
161
162			def writeJobsSpecsToDB(self):
163			"""
164			Write firstEvent and maxEvents in the DB for future use
165			"""
166
167			common.jobDB.load()
168			# case one: write first and max events
169			nJobs=self.nJobs()
170
171			firstEvent=self.first_event
172			# last jobs is different...
173			for job in range(nJobs-1):
174			common.jobDB.setFirstEvent(job, firstEvent)
175			common.jobDB.setMaxEvents(job, self.job_number_of_events)
176			firstEvent=firstEvent+self.job_number_of_events
177
178			# this is the last job
179			common.jobDB.setFirstEvent(nJobs-1, firstEvent)
180	slacapra	1.8	lastJobsNumberOfEvents= (self.total_number_of_events+self.first_event)-firstEvent
181			common.jobDB.setMaxEvents(nJobs-1, lastJobsNumberOfEvents)
182	slacapra	1.7
183	slacapra	1.8	common.logger.message('Created '+str(self.total_njobs-1)+' jobs for '+str(self.job_number_of_events)+' each plus 1 for '+str(lastJobsNumberOfEvents)+' for a total of '+str(self.job_number_of_events*(self.total_njobs-1)+lastJobsNumberOfEvents)+' events')
184	slacapra	1.7	common.jobDB.save()
185
186			# case two (to be implemented) write eventCollections for each jobs
187	nsmirnov	1.2
188			return
189
190	nsmirnov	1.1	def nJobs(self):
191			return self.total_njobs
192
193			def createJobTypeObject(self):
194			file_name = 'cms_'+ string.lower(self.job_type_name)
195			klass_name = string.capitalize(self.job_type_name)
196
197			try:
198			klass = importName(file_name, klass_name)
199			except KeyError:
200			msg = 'No `class '+klass_name+'` found in file `'+file_name+'.py`'
201			raise CrabException(msg)
202			except ImportError, e:
203			msg = 'Cannot create job type '+self.job_type_name
204			msg += ' (file: '+file_name+', class '+klass_name+'):\n'
205			msg += str(e)
206			raise CrabException(msg)
207
208			self.job_type = klass(self.cfg_params)
209			return
210	nsmirnov	1.5
211			def jobType(self):
212			return self.job_type
213	nsmirnov	1.1
214			def run(self):
215	nsmirnov	1.4	"""
216			The main method of the class.
217			"""
218	nsmirnov	1.1
219	nsmirnov	1.3	common.logger.debug(5, "Creator::run() called")
220
221	nsmirnov	1.1	# Instantiate ScriptWriter
222
223			script_writer = ScriptWriter('crab_template.sh')
224
225			# Loop over jobs
226
227			njc = 0
228	nsmirnov	1.4	for nj in range(self.total_njobs):
229	nsmirnov	1.1	if njc == self.ncjobs : break
230	nsmirnov	1.4	st = common.jobDB.status(nj)
231	nsmirnov	1.1	if st != 'X': continue
232
233	nsmirnov	1.6	common.logger.message("Creating job # "+`(nj+1)`)
234	nsmirnov	1.3
235	nsmirnov	1.1	# Prepare configuration file
236
237	nsmirnov	1.4	self.job_type.modifySteeringCards(nj)
238	nsmirnov	1.1
239			# Create JDL
240	nsmirnov	1.4	# Maybe, it worths to move this call into Submitter,
241			# i.e. to create scheduler-specific file at submission time ?
242	nsmirnov	1.1
243	nsmirnov	1.4	common.scheduler.createJDL(nj)
244	nsmirnov	1.1
245			# Create script
246
247	nsmirnov	1.4	script_writer.modifyTemplateScript(nj)
248			os.chmod(common.job_list[nj].scriptFilename(), 0744)
249	nsmirnov	1.1
250	nsmirnov	1.4	common.jobDB.setStatus(nj, 'C')
251	nsmirnov	1.1	njc = njc + 1
252			pass
253
254			####
255
256			common.jobDB.save()
257
258			msg = '\nTotal of %d jobs created'%njc
259			if njc != self.ncjobs: msg = msg + ' from %d requested'%self.ncjobs
260			msg = msg + '.\n'
261			common.logger.message(msg)
262			return