ViewVC Help
View File | Revision Log | Show Annotations | Root Listing
root/cvsroot/COMP/CRAB/python/Submitter.py
Revision: 1.50
Committed: Wed Aug 9 15:29:50 2006 UTC (18 years, 8 months ago) by mkirn
Content type: text/x-python
Branch: MAIN
Changes since 1.49: +4 -14 lines
Log Message: 
Small bug fix on listMatch calls, changed message output

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