ViewVC Help
View File | Revision Log | Show Annotations | Root Listing
root/cvsroot/COMP/CRAB/python/Submitter.py
Revision: 1.52
Committed: Wed Sep 20 17:29:52 2006 UTC (18 years, 7 months ago) by slacapra
Content type: text/x-python
Branch: MAIN
Changes since 1.51: +0 -1 lines
Log Message: 
BOSS4 + sub-file splitting + taskDB

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 corvo 1.27 from random import random
7 corvo 1.30 import time
8 nsmirnov 1.1
9     class Submitter(Actor):
10 corvo 1.40 def __init__(self, cfg_params, nj_list):
11 nsmirnov 1.1 self.cfg_params = cfg_params
12 nsmirnov 1.4 self.nj_list = nj_list
13 corvo 1.35
14 slacapra 1.25 try:
15     self.ML = int(cfg_params['USER.activate_monalisa'])
16     except KeyError:
17     self.ML = 0
18     pass
19 corvo 1.30
20 nsmirnov 1.1 return
21    
22     def run(self):
23 nsmirnov 1.2 """
24     The main method of the class.
25     """
26     common.logger.debug(5, "Submitter::run() called")
27 slacapra 1.24
28 gutsche 1.42 totalCreatedJobs= 0
29 corvo 1.33 start = time.time()
30 slacapra 1.24 for nj in range(common.jobDB.nJobs()):
31 spiga 1.29 if (common.jobDB.status(nj)=='C') or (common.jobDB.status(nj)=='RC'): totalCreatedJobs +=1
32 slacapra 1.24 pass
33    
34     if (totalCreatedJobs==0):
35     common.logger.message("No jobs to be submitted: first create them")
36 slacapra 1.28 return
37 fanzago 1.16 #########
38 nsmirnov 1.2 # Loop over jobs
39 nsmirnov 1.6 njs = 0
40 corvo 1.15 try:
41 slacapra 1.49 lastDest=''
42 corvo 1.17 for nj in self.nj_list:
43 slacapra 1.49 # first check that status of the job is suitable for submission
44 corvo 1.17 st = common.jobDB.status(nj)
45     if st != 'C' and st != 'K' and st != 'A' and st != 'RC':
46     long_st = crabJobStatusToString(st)
47 slacapra 1.19 msg = "Job # %d not submitted: status %s"%(nj+1, long_st)
48 slacapra 1.18 common.logger.message(msg)
49 corvo 1.17 continue
50 slacapra 1.49 # SL perform listmatch only if destination has changed
51     currDest=common.jobDB.destination(nj)
52     if (currDest!=lastDest):
53     match = common.scheduler.listMatch(nj)
54     lastDest = currDest
55 mkirn 1.50 else:
56     common.logger.message("Sites for job "+str(nj+1)+" the same as previous job")
57 slacapra 1.49 if match:
58 mkirn 1.50 common.logger.message("Found "+str(match)+" compatible site(s) for job "+str(nj+1))
59 slacapra 1.49 else:
60 mkirn 1.50 common.logger.message("No compatible site found, will not submit job "+str(nj+1))
61 slacapra 1.49 continue
62 corvo 1.17
63     common.logger.message("Submitting job # "+`(nj+1)`)
64     jid = common.scheduler.submit(nj)
65    
66     common.jobDB.setStatus(nj, 'S')
67     common.jobDB.setJobId(nj, jid)
68 corvo 1.33 common.jobDB.setTaskId(nj, self.cfg_params['taskId'])
69 corvo 1.17 njs += 1
70    
71     ############################################
72    
73     if st == 'C':
74     resFlag = 0
75     elif st == 'RC':
76     resFlag = 2
77     else:
78     resFlag = 0
79     pass
80 spiga 1.26
81 spiga 1.38 try:
82     Statistic.Monitor('submit',resFlag,jid,'-----')
83     except:
84     pass
85 corvo 1.17
86 corvo 1.44 fl = open(common.work_space.shareDir() + '/' + self.cfg_params['apmon'].fName, 'r')
87     self.cfg_params['sid'] = jid
88 gutsche 1.47 #### FF: per il momento commentiamo nevtJob che non c'e' piu' nel jobdb
89     #nevtJob = common.jobDB.maxEvents(nj)
90 gutsche 1.46
91     # OLI: JobID treatment, special for Condor-G scheduler
92     jobId = ''
93     if common.scheduler.boss_scheduler_name == 'condor_g':
94     # create hash of cfg file
95     hash = makeCksum(common.work_space.cfgFileName())
96     jobId = str(nj + 1) + '_' + hash + '_' + self.cfg_params['sid']
97     common.logger.debug(5,'JobID for ML monitoring is created for CONDOR_G scheduler:'+jobId)
98     else:
99     jobId = str(nj + 1) + '_' + self.cfg_params['sid']
100     common.logger.debug(5,'JobID for ML monitoring is created for EDG scheduler'+jobId)
101    
102 corvo 1.44 if ( jid.find(":") != -1 ) :
103     rb = jid.split(':')[1]
104     self.cfg_params['rb'] = rb.replace('//', '')
105     else :
106     self.cfg_params['rb'] = 'OSG'
107 gutsche 1.46
108 gutsche 1.47 #### FF: per il momento commentiamo nevtJob che non c'e' piu' nel jobdb
109     #params = {'nevtJob': nevtJob, 'jobId': jobId, 'sid': self.cfg_params['sid'], \
110     # 'broker': self.cfg_params['rb'], 'bossId': common.jobDB.bossId(nj)}
111 slacapra 1.51 params = {'jobId': jobId, \
112     'sid': self.cfg_params['sid'], \
113     'broker': self.cfg_params['rb'], \
114     'bossId': common.jobDB.bossId(nj), \
115     'TargetCE': string.join((common.jobDB.destination(nj)),",")}
116    
117 corvo 1.44 for i in fl.readlines():
118     val = i.split(':')
119     params[val[0]] = string.strip(val[1])
120 corvo 1.45 self.cfg_params['apmon'].sendToML(params)
121 corvo 1.17 except:
122 corvo 1.27 exctype, value = sys.exc_info()[:2]
123 corvo 1.30 print "Type: %s Value: %s"%(exctype, value)
124 corvo 1.27 common.logger.message("Submitter::run Exception raised: %s %s"%(exctype, value))
125 corvo 1.17 common.jobDB.save()
126 corvo 1.30
127     stop = time.time()
128 gutsche 1.47 common.logger.debug(1, "Submission Time: "+str(stop - start))
129 fanzago 1.32 #print "Submission Time: %d "%(stop - start)
130 corvo 1.17 common.jobDB.save()
131 corvo 1.15
132 nsmirnov 1.6 msg = '\nTotal of %d jobs submitted'%njs
133     if njs != len(self.nj_list) :
134     msg += ' (from %d requested).'%(len(self.nj_list))
135     pass
136     else:
137     msg += '.'
138     pass
139 nsmirnov 1.2 common.logger.message(msg)
140 nsmirnov 1.1 return