CRAB/python/Submitter.py

from Actor import *
from crab_util import *
import common
from ApmonIf import ApmonIf
import Statistic
#from random import random
import time
from ProgressBar import ProgressBar
from TerminalController import TerminalController

class Submitter(Actor):
    def __init__(self, cfg_params, nj_list):
        self.cfg_params = cfg_params
        self.nj_list = nj_list
        
        if common.scheduler.boss_scheduler_name == 'condor_g':
            # create hash of cfg file
            self.hash = makeCksum(common.work_space.cfgFileName())
        else:
            self.hash = ''

        self.UseServer=0
        try:
            self.UseServer=int(self.cfg_params['CRAB.server_mode'])
        except KeyError:
            pass

        return
    
    def run(self):
        """
        The main method of the class: submit jobs in range self.nj_list
        """
        common.logger.debug(5, "Submitter::run() called")

        totalCreatedJobs= 0
        start = time.time()
        for nj in range(common.jobDB.nJobs()):
            if (common.jobDB.status(nj)=='C') or (common.jobDB.status(nj)=='RC'): totalCreatedJobs +=1
            pass

        if (totalCreatedJobs==0):
            common.logger.message("No jobs to be submitted: first create them")
            return

        # submit pre DashBoard information
        params = {'jobId':'TaskMeta'}
               
        fl = open(common.work_space.shareDir() + '/' + self.cfg_params['apmon'].fName, 'r')
        for i in fl.readlines():
            val = i.split(':')
            params[val[0]] = string.strip(val[1])
            fl.close()

        common.logger.debug(5,'Submission DashBoard Pre-Submission report: '+str(params))
                        
        self.cfg_params['apmon'].sendToML(params)

        # modified to support server mode
        # The boss declare step is performed here 
        # only if  crab is used server mode 
        if (self.UseServer== 9999):
            if not common.scheduler.taskDeclared( common.taskDB.dict('projectName') ): #os.path.basename(os.path.split(common.work_space.topDir())[0]) ):
                common.logger.debug(5,'Declaring jobs to BOSS')
                common.scheduler.declareJob_()   #Add for BOSS4
            else:
                common.logger.debug(5,'Jobs already declared into BOSS')
            common.jobDB.save()
            common.taskDB.save()
                                                                                                                                               
        #########
        #########
        # Loop over jobs
        njs = 0
        try:
            list=[]
            list_of_list = []   
            lastBlock=-1
            count = 0
            for nj in self.nj_list:
                same=0
                # first check that status of the job is suitable for submission
                st = common.jobDB.status(nj)
                if st != 'C'  and st != 'K' and st != 'A' and st != 'RC':
                    long_st = crabJobStatusToString(st)
                    msg = "Job # %d not submitted: status %s"%(nj+1, long_st)
                    common.logger.message(msg)
                    continue
     
                currBlock = common.jobDB.block(nj)
                # SL perform listmatch only if block has changed
                if (currBlock!=lastBlock):
                    if common.scheduler.boss_scheduler_name != "condor_g" :
                        ### MATTY:  patch for white-black list with the list-mathc in glite ###
                        whiteL = []
                        blackL = []
                        if self.cfg_params['CRAB.scheduler'].find("glite") != -1:
                            if 'EDG.ce_white_list' in self.cfg_params.keys():
                                #print self.cfg_params['EDG.ce_white_list'].strip().split(",")
                                if self.cfg_params['EDG.ce_white_list'].strip() != "" and self.cfg_params['EDG.ce_white_list'] != None:
                                    for ceW in self.cfg_params['EDG.ce_white_list'].strip().split(","):
                                        if len(ceW.strip()) > 0 and ceW.strip() != None:
                                            whiteL.append(ceW.strip())
                                        #print "ADDING white ce = "+str(ceW.strip())
                            if 'EDG.ce_black_list' in self.cfg_params.keys():
                                #print self.cfg_params['EDG.ce_black_list'].strip().split(",")
                                if self.cfg_params['EDG.ce_black_list'].strip() != "" and self.cfg_params['EDG.ce_black_list'] != None:
                                    for ceB in self.cfg_params['EDG.ce_black_list'].strip().split(","):
                                        if len(ceB.strip()) > 0 and ceB.strip() != None:
                                            blackL.append(ceB.strip())
                                        #print "ADDING ce = "+str(ceB.strip())
                        match = common.scheduler.listMatch(nj, currBlock, whiteL, blackL)
                        #######################################################################
                    else :
                        match = "1"
                    lastBlock = currBlock
                else:
                    common.logger.debug(1,"Sites for job "+str(nj+1)+" the same as previous job")
                    same=1
     
                if match:
                    if not same:
                        common.logger.message("Found "+str(match)+" compatible site(s) for job "+str(nj+1))
                    else:
                        common.logger.debug(1,"Found "+str(match)+" compatible site(s) for job "+str(nj+1))
                    list.append(common.jobDB.bossId(nj))
     
                    if nj == self.nj_list[-1]: # check that is not the last job in the list
                        list_of_list.append([currBlock,list])
                    else: # check if next job has same group
                        nextBlock = common.jobDB.block(nj+1)
                        if  currBlock != nextBlock : # if not, close this group and reset
                            list_of_list.append([currBlock,list])
                            list=[]
                else:
                    common.logger.message("No compatible site found, will not submit job "+str(nj+1))
                    continue
                count += 1
            ### Progress Bar indicator, deactivate for debug
            if not common.logger.debugLevel() :
                term = TerminalController()
     
            for ii in range(len(list_of_list)): # Add loop DS
                common.logger.debug(1,'Submitting jobs '+str(list_of_list[ii][1]))
                if not common.logger.debugLevel() :
                    try: pbar = ProgressBar(term, 'Submitting '+str(len(list_of_list[ii][1]))+' jobs')
                    except: pbar = None

                jidLista, bjidLista = common.scheduler.submit(list_of_list[ii])
                bjidLista = map(int, bjidLista) # cast all bjidLista to int

                if not common.logger.debugLevel():
                    if pbar :
                        pbar.update(float(ii+1)/float(len(list_of_list)),'please wait')
     
                for jj in bjidLista: # Add loop over SID returned from group submission  DS
                    tmpNj = jj - 1

                    jid=jidLista[bjidLista.index(jj)]
                    common.logger.debug(5,"Submitted job # "+ `(jj)`)
                    common.jobDB.setStatus(tmpNj, 'S')
                    common.jobDB.setJobId(tmpNj, jid)
                    common.jobDB.setTaskId(tmpNj, self.cfg_params['taskId'])

                    njs += 1
               
                    ##### DashBoard report #####################   
                    ## To distinguish if job is direct or through the server   
                    if (self.UseServer == 0):
                        Sub_Type = 'Direct'
                    else:   
                        Sub_Type = 'Server'
               
                    try:
                        resFlag = 0
                        if st == 'RC': resFlag = 2
                        Statistic.Monitor('submit',resFlag,jid,'-----','dest')
                    except:
                        pass
                    
                    # OLI: JobID treatment, special for Condor-G scheduler
                    jobId = ''
                    if common.scheduler.boss_scheduler_name == 'condor_g':
                        jobId = str(jj) + '_' + self.hash + '_' + jid
                        common.logger.debug(5,'JobID for ML monitoring is created for CONDOR_G scheduler:'+jobId)
                    else:
                        jobId = str(jj) + '_' + jid
                        common.logger.debug(5,'JobID for ML monitoring is created for EDG scheduler'+jobId)
               
                    if ( jid.find(":") != -1 ) :
                        rb = jid.split(':')[1]
                        rb = rb.replace('//', '')
                    else :
                        rb = 'OSG'

                    if len(common.jobDB.destination(tmpNj)) <= 2 :
                        T_SE=string.join((common.jobDB.destination(tmpNj)),",")    
                    else :
                        T_SE=str(len(common.jobDB.destination(tmpNj)))+'_Selected_SE'
                    params = {'jobId': jobId, \
                              'sid': jid, \
                              'broker': rb, \
                              'bossId': jj, \
                              'SubmissionType': Sub_Type, \
                              'TargetSE': T_SE,}
                    common.logger.debug(5,str(params))
               
                    fl = open(common.work_space.shareDir() + '/' + self.cfg_params['apmon'].fName, 'r')
                    for i in fl.readlines():
                        val = i.split(':')
                        params[val[0]] = string.strip(val[1])
                    fl.close()
     
                    common.logger.debug(5,'Submission DashBoard report: '+str(params))
                        
                    self.cfg_params['apmon'].sendToML(params)
                pass
            pass

        except:
            exctype, value = sys.exc_info()[:2]
            print "Type: %s Value: %s"%(exctype, value)
            common.logger.message("Submitter::run Exception raised: %s %s"%(exctype, value))
            common.jobDB.save()
        
        stop = time.time()
        common.logger.debug(1, "Submission Time: "+str(stop - start))
        common.logger.write("Submission time :"+str(stop - start))
        common.jobDB.save()
            
        msg = '\nTotal of %d jobs submitted'%njs
        if njs != len(self.nj_list) :
            msg += ' (from %d requested).'%(len(self.nj_list))
            pass
        else:
            msg += '.'
            pass
        common.logger.message(msg)
        ## add some more verbose message in case submission is not complete
        if (njs < len(self.nj_list)):
            msg =  'Submission performed using the Requirements: \n'
            msg += common.taskDB.dict("jobtype")+' version: '+common.taskDB.dict("codeVersion")+'\n'
            try: msg += 'SE White List: '+self.cfg_params['EDG.se_white_list']+'\n'
            except KeyError: pass
            try: msg += 'SE Black List: '+self.cfg_params['EDG.se_black_list']+'\n'
            except KeyError: pass
            try: msg += 'CE White List: '+self.cfg_params['EDG.ce_white_list']+'\n'
            except KeyError: pass
            try: msg += 'CE Black List: '+self.cfg_params['EDG.ce_black_list']+'\n'
            except KeyError: pass
            msg += '(Hint: please check if '+common.taskDB.dict("jobtype")+' is available at the Sites)\n'
            
        common.logger.message(msg)
            
        return
Revision:	1.82
Committed:	Fri Oct 5 12:27:00 2007 UTC (17 years, 7 months ago) by mcinquil
Content type:	text/x-python
Branch:	MAIN
CVS Tags:	CRAB_2_0_4, CRAB_2_0_4_pre2, CRAB_2_0_4_pre1, CRAB_2_0_3, CRAB_2_0_3_pre1, CRAB_2_0_2, CRAB_2_0_2_pre6, CRAB_2_0_2_pre5, CRAB_2_0_2_pre4, CRAB_2_0_2_pre3, CRAB_2_0_2_pre2, CRAB_2_0_2_pre1, CRAB_2_0_1, CRAB_2_0_1_pre1, CRAB_2_0_0
Changes since 1.81:	+16 -15 lines
Log Message:	CE_white/black_list in glite: print adjustment + bug-fix
#	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	spiga	1.77	self.UseServer=0
23			try:
24			self.UseServer=int(self.cfg_params['CRAB.server_mode'])
25			except KeyError:
26			pass
27
28	nsmirnov	1.1	return
29
30			def run(self):
31	nsmirnov	1.2	"""
32	slacapra	1.53	The main method of the class: submit jobs in range self.nj_list
33	nsmirnov	1.2	"""
34			common.logger.debug(5, "Submitter::run() called")
35	slacapra	1.24
36	gutsche	1.42	totalCreatedJobs= 0
37	corvo	1.33	start = time.time()
38	slacapra	1.24	for nj in range(common.jobDB.nJobs()):
39	spiga	1.29	if (common.jobDB.status(nj)=='C') or (common.jobDB.status(nj)=='RC'): totalCreatedJobs +=1
40	slacapra	1.24	pass
41
42			if (totalCreatedJobs==0):
43			common.logger.message("No jobs to be submitted: first create them")
44	slacapra	1.28	return
45	slacapra	1.60
46	gutsche	1.70	# submit pre DashBoard information
47			params = {'jobId':'TaskMeta'}
48
49			fl = open(common.work_space.shareDir() + '/' + self.cfg_params['apmon'].fName, 'r')
50			for i in fl.readlines():
51			val = i.split(':')
52			params[val[0]] = string.strip(val[1])
53			fl.close()
54
55			common.logger.debug(5,'Submission DashBoard Pre-Submission report: '+str(params))
56
57			self.cfg_params['apmon'].sendToML(params)
58
59	spiga	1.77	# modified to support server mode
60			# The boss declare step is performed here
61			# only if crab is used server mode
62			if (self.UseServer== 9999):
63			if not common.scheduler.taskDeclared( common.taskDB.dict('projectName') ): #os.path.basename(os.path.split(common.work_space.topDir())[0]) ):
64			common.logger.debug(5,'Declaring jobs to BOSS')
65			common.scheduler.declareJob_() #Add for BOSS4
66			else:
67			common.logger.debug(5,'Jobs already declared into BOSS')
68			common.jobDB.save()
69			common.taskDB.save()
70
71			#########
72	fanzago	1.16	#########
73	nsmirnov	1.2	# Loop over jobs
74	nsmirnov	1.6	njs = 0
75	corvo	1.15	try:
76	slacapra	1.69	list=[]
77	spiga	1.66	list_of_list = []
78	slacapra	1.57	lastBlock=-1
79	spiga	1.66	count = 0
80	corvo	1.17	for nj in self.nj_list:
81	slacapra	1.53	same=0
82	slacapra	1.49	# first check that status of the job is suitable for submission
83	corvo	1.17	st = common.jobDB.status(nj)
84	slacapra	1.69	if st != 'C' and st != 'K' and st != 'A' and st != 'RC':
85	corvo	1.17	long_st = crabJobStatusToString(st)
86	slacapra	1.19	msg = "Job # %d not submitted: status %s"%(nj+1, long_st)
87	slacapra	1.18	common.logger.message(msg)
88	corvo	1.17	continue
89	corvo	1.72
90	slacapra	1.57	currBlock = common.jobDB.block(nj)
91			# SL perform listmatch only if block has changed
92			if (currBlock!=lastBlock):
93	gutsche	1.54	if common.scheduler.boss_scheduler_name != "condor_g" :
94	mcinquil	1.81	### MATTY: patch for white-black list with the list-mathc in glite ###
95			whiteL = []
96			blackL = []
97	mcinquil	1.82	if self.cfg_params['CRAB.scheduler'].find("glite") != -1:
98			if 'EDG.ce_white_list' in self.cfg_params.keys():
99			#print self.cfg_params['EDG.ce_white_list'].strip().split(",")
100			if self.cfg_params['EDG.ce_white_list'].strip() != "" and self.cfg_params['EDG.ce_white_list'] != None:
101			for ceW in self.cfg_params['EDG.ce_white_list'].strip().split(","):
102			if len(ceW.strip()) > 0 and ceW.strip() != None:
103			whiteL.append(ceW.strip())
104			#print "ADDING white ce = "+str(ceW.strip())
105			if 'EDG.ce_black_list' in self.cfg_params.keys():
106			#print self.cfg_params['EDG.ce_black_list'].strip().split(",")
107			if self.cfg_params['EDG.ce_black_list'].strip() != "" and self.cfg_params['EDG.ce_black_list'] != None:
108			for ceB in self.cfg_params['EDG.ce_black_list'].strip().split(","):
109			if len(ceB.strip()) > 0 and ceB.strip() != None:
110			blackL.append(ceB.strip())
111			#print "ADDING ce = "+str(ceB.strip())
112			match = common.scheduler.listMatch(nj, currBlock, whiteL, blackL)
113	mcinquil	1.81	#######################################################################
114	gutsche	1.54	else :
115			match = "1"
116	slacapra	1.57	lastBlock = currBlock
117	mkirn	1.50	else:
118	slacapra	1.53	common.logger.debug(1,"Sites for job "+str(nj+1)+" the same as previous job")
119			same=1
120	corvo	1.72
121	slacapra	1.49	if match:
122	slacapra	1.53	if not same:
123			common.logger.message("Found "+str(match)+" compatible site(s) for job "+str(nj+1))
124			else:
125			common.logger.debug(1,"Found "+str(match)+" compatible site(s) for job "+str(nj+1))
126	slacapra	1.69	list.append(common.jobDB.bossId(nj))
127	corvo	1.72
128	slacapra	1.69	if nj == self.nj_list[-1]: # check that is not the last job in the list
129			list_of_list.append([currBlock,list])
130			else: # check if next job has same group
131			nextBlock = common.jobDB.block(nj+1)
132			if currBlock != nextBlock : # if not, close this group and reset
133	spiga	1.66	list_of_list.append([currBlock,list])
134	slacapra	1.69	list=[]
135	slacapra	1.49	else:
136	mkirn	1.50	common.logger.message("No compatible site found, will not submit job "+str(nj+1))
137	slacapra	1.49	continue
138	spiga	1.66	count += 1
139	gutsche	1.65	### Progress Bar indicator, deactivate for debug
140			if not common.logger.debugLevel() :
141			term = TerminalController()
142	corvo	1.72
143	spiga	1.66	for ii in range(len(list_of_list)): # Add loop DS
144	slacapra	1.68	common.logger.debug(1,'Submitting jobs '+str(list_of_list[ii][1]))
145	corvo	1.74	if not common.logger.debugLevel() :
146			try: pbar = ProgressBar(term, 'Submitting '+str(len(list_of_list[ii][1]))+' jobs')
147			except: pbar = None
148	mcinquil	1.78
149	spiga	1.79	jidLista, bjidLista = common.scheduler.submit(list_of_list[ii])
150	slacapra	1.69	bjidLista = map(int, bjidLista) # cast all bjidLista to int
151	mcinquil	1.78
152	corvo	1.74	if not common.logger.debugLevel():
153			if pbar :
154			pbar.update(float(ii+1)/float(len(list_of_list)),'please wait')
155	corvo	1.72
156	corvo	1.67	for jj in bjidLista: # Add loop over SID returned from group submission DS
157	slacapra	1.69	tmpNj = jj - 1
158	mcinquil	1.78
159	corvo	1.67	jid=jidLista[bjidLista.index(jj)]
160	corvo	1.72	common.logger.debug(5,"Submitted job # "+ `(jj)`)
161	slacapra	1.69	common.jobDB.setStatus(tmpNj, 'S')
162			common.jobDB.setJobId(tmpNj, jid)
163			common.jobDB.setTaskId(tmpNj, self.cfg_params['taskId'])
164	mcinquil	1.78
165	spiga	1.56	njs += 1
166
167	slacapra	1.69	##### DashBoard report #####################
168	spiga	1.80	## To distinguish if job is direct or through the server
169			if (self.UseServer == 0):
170			Sub_Type = 'Direct'
171			else:
172			Sub_Type = 'Server'
173
174	slacapra	1.69	try:
175	spiga	1.56	resFlag = 0
176	slacapra	1.69	if st == 'RC': resFlag = 2
177	spiga	1.71	Statistic.Monitor('submit',resFlag,jid,'-----','dest')
178	spiga	1.56	except:
179			pass
180
181			# OLI: JobID treatment, special for Condor-G scheduler
182			jobId = ''
183			if common.scheduler.boss_scheduler_name == 'condor_g':
184	corvo	1.72	jobId = str(jj) + '_' + self.hash + '_' + jid
185	spiga	1.56	common.logger.debug(5,'JobID for ML monitoring is created for CONDOR_G scheduler:'+jobId)
186			else:
187	corvo	1.72	jobId = str(jj) + '_' + jid
188	spiga	1.56	common.logger.debug(5,'JobID for ML monitoring is created for EDG scheduler'+jobId)
189
190			if ( jid.find(":") != -1 ) :
191			rb = jid.split(':')[1]
192	slacapra	1.69	rb = rb.replace('//', '')
193	spiga	1.56	else :
194	slacapra	1.69	rb = 'OSG'
195	spiga	1.80
196			if len(common.jobDB.destination(tmpNj)) <= 2 :
197			T_SE=string.join((common.jobDB.destination(tmpNj)),",")
198			else :
199			T_SE=str(len(common.jobDB.destination(tmpNj)))+'_Selected_SE'
200	spiga	1.56	params = {'jobId': jobId, \
201	slacapra	1.69	'sid': jid, \
202			'broker': rb, \
203	corvo	1.72	'bossId': jj, \
204	spiga	1.80	'SubmissionType': Sub_Type, \
205			'TargetSE': T_SE,}
206			common.logger.debug(5,str(params))
207	spiga	1.56
208	slacapra	1.69	fl = open(common.work_space.shareDir() + '/' + self.cfg_params['apmon'].fName, 'r')
209	spiga	1.56	for i in fl.readlines():
210			val = i.split(':')
211			params[val[0]] = string.strip(val[1])
212	slacapra	1.69	fl.close()
213	corvo	1.72
214	gutsche	1.65	common.logger.debug(5,'Submission DashBoard report: '+str(params))
215
216	spiga	1.56	self.cfg_params['apmon'].sendToML(params)
217	slacapra	1.69	pass
218			pass
219	corvo	1.17
220			except:
221	corvo	1.27	exctype, value = sys.exc_info()[:2]
222	corvo	1.30	print "Type: %s Value: %s"%(exctype, value)
223	corvo	1.27	common.logger.message("Submitter::run Exception raised: %s %s"%(exctype, value))
224	corvo	1.17	common.jobDB.save()
225	corvo	1.30
226			stop = time.time()
227	gutsche	1.47	common.logger.debug(1, "Submission Time: "+str(stop - start))
228	slacapra	1.60	common.logger.write("Submission time :"+str(stop - start))
229	corvo	1.17	common.jobDB.save()
230	corvo	1.15
231	nsmirnov	1.6	msg = '\nTotal of %d jobs submitted'%njs
232			if njs != len(self.nj_list) :
233			msg += ' (from %d requested).'%(len(self.nj_list))
234			pass
235			else:
236			msg += '.'
237			pass
238	nsmirnov	1.2	common.logger.message(msg)
239	slacapra	1.73	## add some more verbose message in case submission is not complete
240			if (njs < len(self.nj_list)):
241			msg = 'Submission performed using the Requirements: \n'
242	slacapra	1.76	msg += common.taskDB.dict("jobtype")+' version: '+common.taskDB.dict("codeVersion")+'\n'
243	slacapra	1.73	try: msg += 'SE White List: '+self.cfg_params['EDG.se_white_list']+'\n'
244			except KeyError: pass
245			try: msg += 'SE Black List: '+self.cfg_params['EDG.se_black_list']+'\n'
246			except KeyError: pass
247			try: msg += 'CE White List: '+self.cfg_params['EDG.ce_white_list']+'\n'
248			except KeyError: pass
249			try: msg += 'CE Black List: '+self.cfg_params['EDG.ce_black_list']+'\n'
250			except KeyError: pass
251	slacapra	1.76	msg += '(Hint: please check if '+common.taskDB.dict("jobtype")+' is available at the Sites)\n'
252	slacapra	1.73
253			common.logger.message(msg)
254
255	nsmirnov	1.1	return