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

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 = self.cfg_params['USER.first_event']
        except KeyError:
            self.first_event = 0

        maxEvents = int(self.job_type.maxEvents)

        if self.first_event>=maxEvents:
            raise CrabException('First event is bigger than maximum number of available events!')

        try:
            n = self.cfg_params['USER.total_number_of_events']
            if n == 'all': n = '-1'
            self.total_number_of_events = (maxEvents - self.first_event)
        except KeyError:
            common.logger.message("total_number_of_events not defined, set it to maximum available")
            self.total_number_of_events = (maxEvents - self.first_event)
            pass

        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
            else:
                self.job_number_of_events = int(n)
        # second case: jobs per task defined
        elif jobsPerTask>0:
            common.logger.debug(5,"total number of events: "+str(self.total_number_of_events)+" JobPerTask "+str(jobsPerTask))
            self.job_number_of_events = int((self.total_number_of_events-1)/jobsPerTask)+1
        # should not happen...
        else:
            raise CrabException('Somthing wrong with splitting')

        common.logger.debug(5,"total number of events: "+str(self.total_number_of_events)+
            " events per job: "+str(self.job_number_of_events))
        if self.job_number_of_events==-1:
            self.total_njobs = 1
        if self.total_number_of_events!=-1:
            self.total_njobs = int((self.total_number_of_events-1)/self.job_number_of_events)+1
        else:
            self.total_njobs = (maxEvents-1)/int(self.job_number_of_events)+1
        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
        print firstEvent
        # 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)
        common.jobDB.setMaxEvents(nJobs-1, self.total_number_of_events-self.first_event-firstEvent)
    
        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.7
Committed:	Wed Jul 20 10:03:22 2005 UTC (19 years, 9 months ago) by slacapra
Content type:	text/x-python
Branch:	MAIN
Changes since 1.6:	+87 -9 lines
Log Message:	too many changes to list them here...
#	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			import os, string
13
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.7	self.first_event = self.cfg_params['USER.first_event']
82			except KeyError:
83			self.first_event = 0
84
85			maxEvents = int(self.job_type.maxEvents)
86
87			if self.first_event>=maxEvents:
88			raise CrabException('First event is bigger than maximum number of available events!')
89
90			try:
91	nsmirnov	1.2	n = self.cfg_params['USER.total_number_of_events']
92			if n == 'all': n = '-1'
93	slacapra	1.7	self.total_number_of_events = (maxEvents - self.first_event)
94			except KeyError:
95			common.logger.message("total_number_of_events not defined, set it to maximum available")
96			self.total_number_of_events = (maxEvents - self.first_event)
97			pass
98
99			eventPerJob=0
100			try:
101			eventPerJob = self.cfg_params['USER.job_number_of_events']
102	nsmirnov	1.2	except KeyError:
103			pass
104
105	slacapra	1.7	jobsPerTask=0
106	nsmirnov	1.2	try:
107	slacapra	1.7	jobsPerTask = int(self.cfg_params['USER.total_number_of_jobs'])
108	nsmirnov	1.2	except KeyError:
109	slacapra	1.7	pass
110
111			# If both the above set, complain and use event per jobs
112			if eventPerJob>0 and jobsPerTask>0:
113	nsmirnov	1.2	msg = 'Warning. '
114	slacapra	1.7	msg += 'job_number_of_events and total_number_of_jobs are both defined '
115			msg += 'Using job_number_of_events.'
116	nsmirnov	1.2	common.logger.message(msg)
117	slacapra	1.7	jobsPerTask = 0
118			if eventPerJob==0 and jobsPerTask==0:
119			msg = 'Warning. '
120			msg += 'job_number_of_events and total_number_of_jobs are not defined '
121			msg += 'Creating just one job for all events.'
122			common.logger.message(msg)
123			jobsPerTask = 1
124
125			# first case: events per job defined
126
127			if eventPerJob>0:
128			n=eventPerJob
129			if n == 'all' or n == '-1' or (int(n)>self.total_number_of_events and self.total_number_of_events>0):
130			common.logger.message("Asking more events than available: set it to maximum available")
131			self.job_number_of_events = self.total_number_of_events
132			else:
133			self.job_number_of_events = int(n)
134			# second case: jobs per task defined
135			elif jobsPerTask>0:
136			common.logger.debug(5,"total number of events: "+str(self.total_number_of_events)+" JobPerTask "+str(jobsPerTask))
137			self.job_number_of_events = int((self.total_number_of_events-1)/jobsPerTask)+1
138			# should not happen...
139			else:
140			raise CrabException('Somthing wrong with splitting')
141
142			common.logger.debug(5,"total number of events: "+str(self.total_number_of_events)+
143			" events per job: "+str(self.job_number_of_events))
144			if self.job_number_of_events==-1:
145			self.total_njobs = 1
146			if self.total_number_of_events!=-1:
147			self.total_njobs = int((self.total_number_of_events-1)/self.job_number_of_events)+1
148			else:
149			self.total_njobs = (maxEvents-1)/int(self.job_number_of_events)+1
150			return
151
152			def writeJobsSpecsToDB(self):
153			"""
154			Write firstEvent and maxEvents in the DB for future use
155			"""
156
157			common.jobDB.load()
158			# case one: write first and max events
159			nJobs=self.nJobs()
160
161			firstEvent=self.first_event
162			print firstEvent
163			# last jobs is different...
164			for job in range(nJobs-1):
165			common.jobDB.setFirstEvent(job, firstEvent)
166			common.jobDB.setMaxEvents(job, self.job_number_of_events)
167			firstEvent=firstEvent+self.job_number_of_events
168
169			# this is the last job
170			common.jobDB.setFirstEvent(nJobs-1, firstEvent)
171			common.jobDB.setMaxEvents(nJobs-1, self.total_number_of_events-self.first_event-firstEvent)
172
173			common.jobDB.save()
174
175			# case two (to be implemented) write eventCollections for each jobs
176	nsmirnov	1.2
177			return
178
179	nsmirnov	1.1	def nJobs(self):
180			return self.total_njobs
181
182			def createJobTypeObject(self):
183			file_name = 'cms_'+ string.lower(self.job_type_name)
184			klass_name = string.capitalize(self.job_type_name)
185
186			try:
187			klass = importName(file_name, klass_name)
188			except KeyError:
189			msg = 'No `class '+klass_name+'` found in file `'+file_name+'.py`'
190			raise CrabException(msg)
191			except ImportError, e:
192			msg = 'Cannot create job type '+self.job_type_name
193			msg += ' (file: '+file_name+', class '+klass_name+'):\n'
194			msg += str(e)
195			raise CrabException(msg)
196
197			self.job_type = klass(self.cfg_params)
198			return
199	nsmirnov	1.5
200			def jobType(self):
201			return self.job_type
202	nsmirnov	1.1
203			def run(self):
204	nsmirnov	1.4	"""
205			The main method of the class.
206			"""
207	nsmirnov	1.1
208	nsmirnov	1.3	common.logger.debug(5, "Creator::run() called")
209
210	nsmirnov	1.1	# Instantiate ScriptWriter
211
212			script_writer = ScriptWriter('crab_template.sh')
213
214			# Loop over jobs
215
216			njc = 0
217	nsmirnov	1.4	for nj in range(self.total_njobs):
218	nsmirnov	1.1	if njc == self.ncjobs : break
219	nsmirnov	1.4	st = common.jobDB.status(nj)
220	nsmirnov	1.1	if st != 'X': continue
221
222	nsmirnov	1.6	common.logger.message("Creating job # "+`(nj+1)`)
223	nsmirnov	1.3
224	nsmirnov	1.1	# Prepare configuration file
225
226	nsmirnov	1.4	self.job_type.modifySteeringCards(nj)
227	nsmirnov	1.1
228			# Create JDL
229	nsmirnov	1.4	# Maybe, it worths to move this call into Submitter,
230			# i.e. to create scheduler-specific file at submission time ?
231	nsmirnov	1.1
232	nsmirnov	1.4	common.scheduler.createJDL(nj)
233	nsmirnov	1.1
234			# Create script
235
236	nsmirnov	1.4	script_writer.modifyTemplateScript(nj)
237			os.chmod(common.job_list[nj].scriptFilename(), 0744)
238	nsmirnov	1.1
239	nsmirnov	1.4	common.jobDB.setStatus(nj, 'C')
240	nsmirnov	1.1	njc = njc + 1
241			pass
242
243			####
244
245			common.jobDB.save()
246
247			msg = '\nTotal of %d jobs created'%njc
248			if njc != self.ncjobs: msg = msg + ' from %d requested'%self.ncjobs
249			msg = msg + '.\n'
250			common.logger.message(msg)
251			return