ViewVC Help
View File | Revision Log | Show Annotations | Root Listing
root/cvsroot/COMP/CRAB/python/Submitter.py
Revision: 1.61
Committed: Wed Oct 11 10:25:50 2006 UTC (18 years, 6 months ago) by slacapra
Content type: text/x-python
Branch: MAIN
Changes since 1.60: +2 -1 lines
Log Message: 
no progress bar if debug mode

File Contents

# User Rev Content
1 nsmirnov 1.1 from Actor import *
2 nsmirnov 1.6 from crab_util import *
3 nsmirnov 1.2 import common
4 corvo 1.9 from ApmonIf import ApmonIf
5 spiga 1.10 import Statistic
6 slacapra 1.60 #from random import random
7 corvo 1.30 import time
8 slacapra 1.60 from ProgressBar import ProgressBar
9     from TerminalController import TerminalController
10 nsmirnov 1.1
11     class Submitter(Actor):
12 corvo 1.40 def __init__(self, cfg_params, nj_list):
13 nsmirnov 1.1 self.cfg_params = cfg_params
14 nsmirnov 1.4 self.nj_list = nj_list
15 gutsche 1.58
16     if common.scheduler.boss_scheduler_name == 'condor_g':
17     # create hash of cfg file
18     self.hash = makeCksum(common.work_space.cfgFileName())
19     else:
20     self.hash = ''
21 corvo 1.30
22 nsmirnov 1.1 return
23    
24     def run(self):
25 nsmirnov 1.2 """
26 slacapra 1.53 The main method of the class: submit jobs in range self.nj_list
27 nsmirnov 1.2 """
28     common.logger.debug(5, "Submitter::run() called")
29 slacapra 1.24
30 gutsche 1.42 totalCreatedJobs= 0
31 corvo 1.33 start = time.time()
32 slacapra 1.24 for nj in range(common.jobDB.nJobs()):
33 spiga 1.29 if (common.jobDB.status(nj)=='C') or (common.jobDB.status(nj)=='RC'): totalCreatedJobs +=1
34 slacapra 1.24 pass
35    
36     if (totalCreatedJobs==0):
37     common.logger.message("No jobs to be submitted: first create them")
38 slacapra 1.28 return
39 slacapra 1.60
40 fanzago 1.16 #########
41 nsmirnov 1.2 # Loop over jobs
42 nsmirnov 1.6 njs = 0
43 corvo 1.15 try:
44 spiga 1.56 first = []
45     last = []
46 slacapra 1.57 lastBlock=-1
47 corvo 1.17 for nj in self.nj_list:
48 slacapra 1.53 same=0
49 slacapra 1.49 # first check that status of the job is suitable for submission
50 corvo 1.17 st = common.jobDB.status(nj)
51 spiga 1.56 if st != 'C' :# and st != 'K' and st != 'A' and st != 'RC': ## commentato per ora...quindi NON risotomette
52 corvo 1.17 long_st = crabJobStatusToString(st)
53 slacapra 1.19 msg = "Job # %d not submitted: status %s"%(nj+1, long_st)
54 slacapra 1.18 common.logger.message(msg)
55 corvo 1.17 continue
56 slacapra 1.57 currBlock = common.jobDB.block(nj)
57     # SL perform listmatch only if block has changed
58     if (currBlock!=lastBlock):
59 gutsche 1.54 if common.scheduler.boss_scheduler_name != "condor_g" :
60 slacapra 1.57 match = common.scheduler.listMatch(nj, currBlock)
61 gutsche 1.54 else :
62     match = "1"
63 slacapra 1.57 lastBlock = currBlock
64 mkirn 1.50 else:
65 slacapra 1.53 common.logger.debug(1,"Sites for job "+str(nj+1)+" the same as previous job")
66     same=1
67 slacapra 1.49 if match:
68 slacapra 1.53 if not same:
69 spiga 1.56 ###################### here create index list start:end
70     first.append(nj)
71     if nj != 0:last.append(nj-1)
72     ######################
73 slacapra 1.53 common.logger.message("Found "+str(match)+" compatible site(s) for job "+str(nj+1))
74     else:
75     common.logger.debug(1,"Found "+str(match)+" compatible site(s) for job "+str(nj+1))
76 slacapra 1.49 else:
77 mkirn 1.50 common.logger.message("No compatible site found, will not submit job "+str(nj+1))
78 slacapra 1.49 continue
79 slacapra 1.57
80 spiga 1.56 ############## TODO improve the follow control .....
81     if len(first)>len(last):
82     if common.jobDB.nJobs() == 1 :
83     last.append(0) # List of last index
84     else:
85     last.append(self.nj_list[len(self.nj_list)-1]) #List of last index
86     else:
87     if first[len(first)-1] > last[len(last)-1]:
88     last.remove(last[len(last)-1])
89     last.append(self.nj_list[len(self.nj_list)-1])
90    
91 slacapra 1.60 ### Progress Bar indicator
92     term = TerminalController()
93    
94 spiga 1.56 for ii in range(len(first)): # Add loop DS
95 gutsche 1.58 common.logger.message('Submitting jobs '+str(first[ii]+1)+' to '+str(last[ii]+1))
96 slacapra 1.60 try: pbar = ProgressBar(term, 'Submitting '+str(len(self.nj_list))+' jobs')
97     except: pbar = None
98 spiga 1.56 #common.logger.message("Submitting job # "+`(nj+1)`)
99     jidLista = common.scheduler.submit(first[ii],last[ii],ii)
100    
101     ####
102     for jj in range(len(jidLista)): # Add loop over SID returned from group submission DS
103     nj= int(jj+int(first[ii]))
104     jid=jidLista[jj]
105 slacapra 1.61 if pbar and not common.logger.debugLevel():
106     pbar.update(float(nj+1)/float(len(jidLista)),'please wait')
107 gutsche 1.59 common.logger.debug(1,"Submitted job # "+`(nj+1)`)
108 spiga 1.56 common.jobDB.setStatus(nj, 'S')
109     common.jobDB.setJobId(nj, jid)
110     common.jobDB.setTaskId(nj, self.cfg_params['taskId'])
111     njs += 1
112     ############################################
113    
114     if st == 'C':
115     resFlag = 0
116     elif st == 'RC':
117     resFlag = 2
118     else:
119     resFlag = 0
120     pass
121    
122     try:
123     Statistic.Monitor('submit',resFlag,jid,'-----')
124     except:
125     pass
126    
127     fl = open(common.work_space.shareDir() + '/' + self.cfg_params['apmon'].fName, 'r')
128     self.cfg_params['sid'] = jid
129     #### FF: per il momento commentiamo nevtJob che non c'e' piu' nel jobdb
130     #nevtJob = common.jobDB.maxEvents(nj)
131    
132     # OLI: JobID treatment, special for Condor-G scheduler
133     jobId = ''
134     if common.scheduler.boss_scheduler_name == 'condor_g':
135 gutsche 1.58 jobId = str(nj + 1) + '_' + self.hash + '_' + self.cfg_params['sid']
136 spiga 1.56 common.logger.debug(5,'JobID for ML monitoring is created for CONDOR_G scheduler:'+jobId)
137     else:
138     jobId = str(nj + 1) + '_' + self.cfg_params['sid']
139     common.logger.debug(5,'JobID for ML monitoring is created for EDG scheduler'+jobId)
140    
141     if ( jid.find(":") != -1 ) :
142     rb = jid.split(':')[1]
143     self.cfg_params['rb'] = rb.replace('//', '')
144     else :
145     self.cfg_params['rb'] = 'OSG'
146    
147     #### FF: per il momento commentiamo nevtJob che non c'e' piu' nel jobdb
148     #params = {'nevtJob': nevtJob, 'jobId': jobId, 'sid': self.cfg_params['sid'], \
149     # 'broker': self.cfg_params['rb'], 'bossId': common.jobDB.bossId(nj)}
150     params = {'jobId': jobId, \
151     'sid': self.cfg_params['sid'], \
152     'broker': self.cfg_params['rb'], \
153     'bossId': common.jobDB.bossId(nj), \
154     'TargetCE': string.join((common.jobDB.destination(nj)),",")}
155    
156     for i in fl.readlines():
157     val = i.split(':')
158     params[val[0]] = string.strip(val[1])
159     self.cfg_params['apmon'].sendToML(params)
160 corvo 1.17
161     except:
162 corvo 1.27 exctype, value = sys.exc_info()[:2]
163 corvo 1.30 print "Type: %s Value: %s"%(exctype, value)
164 corvo 1.27 common.logger.message("Submitter::run Exception raised: %s %s"%(exctype, value))
165 corvo 1.17 common.jobDB.save()
166 corvo 1.30
167     stop = time.time()
168 gutsche 1.47 common.logger.debug(1, "Submission Time: "+str(stop - start))
169 slacapra 1.60 common.logger.write("Submission time :"+str(stop - start))
170 corvo 1.17 common.jobDB.save()
171 corvo 1.15
172 nsmirnov 1.6 msg = '\nTotal of %d jobs submitted'%njs
173     if njs != len(self.nj_list) :
174     msg += ' (from %d requested).'%(len(self.nj_list))
175     pass
176     else:
177     msg += '.'
178     pass
179 nsmirnov 1.2 common.logger.message(msg)
180 nsmirnov 1.1 return