ViewVC Help
View File | Revision Log | Show Annotations | Root Listing
root/cvsroot/COMP/CRAB/python/Creator.py
Revision: 1.33
Committed: Thu Mar 30 10:20:25 2006 UTC (19 years, 1 month ago) by corvo
Content type: text/x-python
Branch: MAIN
Changes since 1.32: +20 -0 lines
Log Message:
ML params collected here now and written to a file

File Contents

# 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 corvo 1.26 # marco
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 slacapra 1.14
24 corvo 1.30 self.createJobTypeObject()
25 nsmirnov 1.3 common.logger.debug(5, __name__+": JobType "+self.job_type.name()+" created")
26 nsmirnov 1.1
27     self.job_type.prepareSteeringCards()
28 nsmirnov 1.3 common.logger.debug(5, __name__+": Steering cards prepared")
29 nsmirnov 1.1
30 nsmirnov 1.2 self.defineTotalNumberOfJobs_()
31 nsmirnov 1.3 common.logger.debug(5, __name__+": total # of jobs = "+`self.total_njobs`)
32 nsmirnov 1.1
33     # Set number of jobs to be created
34    
35     self.ncjobs = ncjobs
36     if ncjobs == 'all' : self.ncjobs = self.total_njobs
37     if ncjobs > self.total_njobs : self.ncjobs = self.total_njobs
38 spiga 1.11
39 spiga 1.20 fileCODE1 = open(common.work_space.shareDir()+"/.code","a")
40     ### commented for FAMOS
41 gutsche 1.32 if self.job_type.name() == 'ORCA' or self.job_type.name() == 'ORCA_DBSDLS' or self.job_type.name() == 'ORCA_COMMON' :
42 spiga 1.20 self.owner = cfg_params['ORCA.owner']
43     self.dataset = cfg_params['ORCA.dataset']
44     fileCODE1.write('::'+str(self.job_type.name())+'::'+str(self.ncjobs)+'::'+str(self.dataset)+'::'+str(self.owner))
45     pass
46     elif self.job_type.name() == 'FAMOS':
47 spiga 1.27 self.inputFile = cfg_params['FAMOS.input_lfn']
48     self.executable = cfg_params['FAMOS.executable']
49     fileCODE1.write('::'+str(self.job_type.name())+'::'+str(self.ncjobs)+'::'+str(self.inputFile)+'::'+str(self.executable))
50 spiga 1.20 pass
51 spiga 1.11 fileCODE1.close()
52    
53 corvo 1.33 try:
54     fl = open(common.work_space.shareDir() + '/' + self.cfg_params['apmon'].fName, 'w')
55     self.cfg_params['GridName'] = runCommand("voms-proxy-info -identity")
56     common.logger.debug(5, "GRIDNAME: "+self.cfg_params['GridName'])
57     taskType = 'analysis'
58    
59     params = {'tool': common.prog_name, 'tool_ui': os.environ['HOSTNAME'], \
60     'scheduler': self.cfg_params['CRAB.scheduler'], 'GridName': self.cfg_params['GridName'].strip(), \
61     'taskType': taskType, 'vo': self.cfg_params['EDG.virtual_organization'], 'user': self.cfg_params['user']}
62     jtParam = self.job_type.getParams()
63     for i in jtParam.iterkeys():
64     params[i] = string.strip(jtParam[i])
65     for j, k in params.iteritems():
66     # print "Values: %s %s"%(j, k)
67     fl.write(j + ':' + k + '\n')
68     fl.close()
69     except:
70     exctype, value = sys.exc_info()[:2]
71     common.logger.message("Creator::run Exception raised: %s %s"%(exctype, value))
72     pass
73 nsmirnov 1.1
74     #TODO: deprecated code, not needed,
75     # will be eliminated when WorkSpace.saveConfiguration()
76     # will be improved.
77     #
78     # Set/Save Job Type name
79    
80     jt_fname = common.work_space.shareDir() + 'jobtype'
81     if os.path.exists(jt_fname):
82     # Read stored job type name
83     jt_file = open(jt_fname, 'r')
84     jt = jt_file.read()
85     if self.job_type_name:
86     if ( jt != self.job_type_name+'\n' ):
87     msg = 'Job Type mismatch: requested <' + self.job_type_name
88     msg += '>, found <' + jt[:-1] + '>.'
89     raise CrabException(msg)
90     pass
91     else:
92     self.job_type_name = jt[:-1]
93     pass
94     jt_file.close()
95     pass
96     else:
97     # Save job type name
98     jt_file = open(jt_fname, 'w')
99     jt_file.write(self.job_type_name+'\n')
100     jt_file.close()
101     pass
102     #end of deprecated code
103    
104 nsmirnov 1.3 common.logger.debug(5, "Creator constructor finished")
105 nsmirnov 1.1 return
106    
107 nsmirnov 1.2 def defineTotalNumberOfJobs_(self):
108     """
109     Calculates the total number of jobs to be created.
110     """
111    
112     try:
113 slacapra 1.8 self.first_event = int(self.cfg_params['USER.first_event'])
114 slacapra 1.7 except KeyError:
115     self.first_event = 0
116 slacapra 1.8 common.logger.debug(1,"First event ot be analyzed: "+str(self.first_event))
117 slacapra 1.7
118 slacapra 1.10 # TODO Could we find a better way to get this number?
119 slacapra 1.8 maxAvailableEvents = int(self.job_type.maxEvents)
120     common.logger.debug(1,"Available events: "+str(maxAvailableEvents))
121 slacapra 1.7
122 slacapra 1.8 # some sanity check
123     if self.first_event>=maxAvailableEvents:
124 slacapra 1.7 raise CrabException('First event is bigger than maximum number of available events!')
125    
126 slacapra 1.8 # the total number of events to be analyzed
127 slacapra 1.7 try:
128 nsmirnov 1.2 n = self.cfg_params['USER.total_number_of_events']
129     if n == 'all': n = '-1'
130 slacapra 1.8 if n == '-1':
131     self.total_number_of_events = (maxAvailableEvents - self.first_event)
132     common.logger.debug(1,"Analysing all available events "+str(self.total_number_of_events))
133     else:
134     if maxAvailableEvents<(int(n)+self.first_event):
135 slacapra 1.17 raise CrabException('(First event + total events)='+str(int(n)+self.first_event)+' is bigger than maximum number of available events '+str(maxAvailableEvents)+' !! Use "total_number_of_events=-1" to analyze to whole dataset')
136 slacapra 1.8 self.total_number_of_events = int(n)
137 slacapra 1.7 except KeyError:
138     common.logger.message("total_number_of_events not defined, set it to maximum available")
139 slacapra 1.8 self.total_number_of_events = (maxAvailableEvents - self.first_event)
140 slacapra 1.7 pass
141 slacapra 1.8 common.logger.message("Total number of events to be analyzed: "+str(self.total_number_of_events))
142    
143 slacapra 1.7
144 slacapra 1.8 # read user directives
145 slacapra 1.7 eventPerJob=0
146     try:
147     eventPerJob = self.cfg_params['USER.job_number_of_events']
148 nsmirnov 1.2 except KeyError:
149     pass
150    
151 slacapra 1.7 jobsPerTask=0
152 nsmirnov 1.2 try:
153 slacapra 1.7 jobsPerTask = int(self.cfg_params['USER.total_number_of_jobs'])
154 nsmirnov 1.2 except KeyError:
155 slacapra 1.7 pass
156    
157     # If both the above set, complain and use event per jobs
158     if eventPerJob>0 and jobsPerTask>0:
159 nsmirnov 1.2 msg = 'Warning. '
160 slacapra 1.7 msg += 'job_number_of_events and total_number_of_jobs are both defined '
161     msg += 'Using job_number_of_events.'
162 nsmirnov 1.2 common.logger.message(msg)
163 slacapra 1.7 jobsPerTask = 0
164     if eventPerJob==0 and jobsPerTask==0:
165     msg = 'Warning. '
166     msg += 'job_number_of_events and total_number_of_jobs are not defined '
167     msg += 'Creating just one job for all events.'
168     common.logger.message(msg)
169     jobsPerTask = 1
170    
171     # first case: events per job defined
172     if eventPerJob>0:
173     n=eventPerJob
174     if n == 'all' or n == '-1' or (int(n)>self.total_number_of_events and self.total_number_of_events>0):
175     common.logger.message("Asking more events than available: set it to maximum available")
176     self.job_number_of_events = self.total_number_of_events
177 slacapra 1.8 self.total_njobs = 1
178 slacapra 1.7 else:
179     self.job_number_of_events = int(n)
180 slacapra 1.8 self.total_njobs = int((self.total_number_of_events-1)/self.job_number_of_events)+1
181 slacapra 1.7 # second case: jobs per task defined
182     elif jobsPerTask>0:
183 slacapra 1.8 common.logger.debug(2,"total number of events: "+str(self.total_number_of_events)+" JobPerTask "+str(jobsPerTask))
184     self.job_number_of_events = int(math.floor((self.total_number_of_events)/jobsPerTask))
185     self.total_njobs = jobsPerTask
186 slacapra 1.7 # should not happen...
187     else:
188     raise CrabException('Somthing wrong with splitting')
189    
190 slacapra 1.8 common.logger.debug(2,"total number of events: "+str(self.total_number_of_events)+
191 slacapra 1.7 " events per job: "+str(self.job_number_of_events))
192 slacapra 1.8
193 slacapra 1.7 return
194    
195     def writeJobsSpecsToDB(self):
196     """
197     Write firstEvent and maxEvents in the DB for future use
198     """
199    
200     common.jobDB.load()
201     # case one: write first and max events
202     nJobs=self.nJobs()
203    
204     firstEvent=self.first_event
205 fanzago 1.21 lastJobsNumberOfEvents = self.job_number_of_events
206 fanzago 1.28
207     try:
208     self.events_management = common.analisys_common_info['events_management']
209 fanzago 1.31 #print "self.events_management = ", self.events_management
210 fanzago 1.28 except KeyError: self.events_management = 0
211    
212 slacapra 1.7 # last jobs is different...
213     for job in range(nJobs-1):
214     common.jobDB.setFirstEvent(job, firstEvent)
215     common.jobDB.setMaxEvents(job, self.job_number_of_events)
216 fanzago 1.28 ### fede checks parameter for famos and orca
217     #try:
218     # self.events_management = common.analisys_common_info['events_management']
219     # print "self.events_management = ", self.events_management
220     #except KeyError: self.events_management = 0
221 spiga 1.27 firstEvent=firstEvent+self.job_number_of_events
222 slacapra 1.7
223     # this is the last job
224     common.jobDB.setFirstEvent(nJobs-1, firstEvent)
225 fanzago 1.21 if int(self.events_management) == 1:
226     lastJobsNumberOfEvents= (self.total_number_of_events+self.first_event)-firstEvent
227 slacapra 1.8 common.jobDB.setMaxEvents(nJobs-1, lastJobsNumberOfEvents)
228 slacapra 1.7
229 slacapra 1.15 if (lastJobsNumberOfEvents!=self.job_number_of_events):
230     common.logger.message(str(self.total_njobs-1)+' jobs will be created for '+str(self.job_number_of_events)+' events each plus 1 for '+str(lastJobsNumberOfEvents)+' events for a total of '+str(self.job_number_of_events*(self.total_njobs-1)+lastJobsNumberOfEvents)+' events')
231     else:
232     common.logger.message(str(self.total_njobs)+' jobs will be created for '+str(self.job_number_of_events)+' events each for a total of '+str(self.job_number_of_events*(self.total_njobs-1)+lastJobsNumberOfEvents)+' events')
233 slacapra 1.7
234     # case two (to be implemented) write eventCollections for each jobs
235 nsmirnov 1.2
236 slacapra 1.9 # save the DB
237     common.jobDB.save()
238 nsmirnov 1.2 return
239    
240 nsmirnov 1.1 def nJobs(self):
241     return self.total_njobs
242    
243     def createJobTypeObject(self):
244     file_name = 'cms_'+ string.lower(self.job_type_name)
245     klass_name = string.capitalize(self.job_type_name)
246    
247     try:
248     klass = importName(file_name, klass_name)
249     except KeyError:
250     msg = 'No `class '+klass_name+'` found in file `'+file_name+'.py`'
251     raise CrabException(msg)
252     except ImportError, e:
253     msg = 'Cannot create job type '+self.job_type_name
254     msg += ' (file: '+file_name+', class '+klass_name+'):\n'
255     msg += str(e)
256     raise CrabException(msg)
257    
258     self.job_type = klass(self.cfg_params)
259     return
260 nsmirnov 1.5
261     def jobType(self):
262     return self.job_type
263 nsmirnov 1.1
264     def run(self):
265 nsmirnov 1.4 """
266     The main method of the class.
267     """
268 nsmirnov 1.1
269 nsmirnov 1.3 common.logger.debug(5, "Creator::run() called")
270    
271 nsmirnov 1.1 # Instantiate ScriptWriter
272    
273     script_writer = ScriptWriter('crab_template.sh')
274    
275     # Loop over jobs
276    
277     njc = 0
278 nsmirnov 1.4 for nj in range(self.total_njobs):
279 nsmirnov 1.1 if njc == self.ncjobs : break
280 nsmirnov 1.4 st = common.jobDB.status(nj)
281 nsmirnov 1.1 if st != 'X': continue
282    
283 nsmirnov 1.6 common.logger.message("Creating job # "+`(nj+1)`)
284 nsmirnov 1.3
285 nsmirnov 1.1 # Prepare configuration file
286    
287 nsmirnov 1.4 self.job_type.modifySteeringCards(nj)
288 fanzago 1.12 # Create script (sh)
289 fanzago 1.13 #print "nel for prima del modifyTemplateScript, nj vale", nj
290 nsmirnov 1.4 script_writer.modifyTemplateScript(nj)
291     os.chmod(common.job_list[nj].scriptFilename(), 0744)
292 nsmirnov 1.1
293 fanzago 1.12 # Create scheduler scripts (jdl)
294     common.scheduler.createSchScript(nj)
295    
296 nsmirnov 1.4 common.jobDB.setStatus(nj, 'C')
297 slacapra 1.9 # common: write input and output sandbox
298     common.jobDB.setInputSandbox(nj, self.job_type.inputSandbox(nj))
299    
300     outputSandbox=self.job_type.outputSandbox(nj)
301 gutsche 1.32 stdout=common.job_list[nj].stdout()
302 slacapra 1.9 stderr=common.job_list[nj].stderr()
303     outputSandbox.append(common.job_list[nj].stdout())
304     # check if out!=err
305     if stdout != stderr:
306     outputSandbox.append(common.job_list[nj].stderr())
307     common.jobDB.setOutputSandbox(nj, outputSandbox)
308 corvo 1.18 common.jobDB.setTaskId(nj, self.cfg_params['taskId'])
309 slacapra 1.9
310 nsmirnov 1.1 njc = njc + 1
311     pass
312    
313     ####
314 fanzago 1.12
315 nsmirnov 1.1 common.jobDB.save()
316    
317     msg = '\nTotal of %d jobs created'%njc
318     if njc != self.ncjobs: msg = msg + ' from %d requested'%self.ncjobs
319     msg = msg + '.\n'
320     common.logger.message(msg)
321     return