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 = ''

        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

        #########
        # 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': ## commentato per ora...quindi NON risotomette
                    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" :
                        match = common.scheduler.listMatch(nj, currBlock)
                    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))
                   # job list is string because boss can't manage list  
                    list = list+str(nj+1)+',' 
                   # list.append(nj+1)
                    if nj < self.nj_list[len(self.nj_list)-1]:
                        nextBlock = common.jobDB.block(self.nj_list[count+1])
                        if  currBlock != nextBlock :
                            list_of_list.append([currBlock,list])
                            list=''    
                    else:
                        list_of_list.append([currBlock,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.message('Submitting jobs '+str(list_of_list[ii][1]))
                if not common.logger.debugLevel() :
                    try: pbar = ProgressBar(term, 'Submitting '+str(len(self.nj_list))+' jobs')
                    except: pbar = None
                #common.logger.message("Submitting job # "+`(nj+1)`)  
                jidLista = common.scheduler.submit(list_of_list[ii])
                ####
                for jj in range(len(jidLista)): # Add loop over SID returned from group submission  DS
                   # nj= int(jj+int(list[0]))
                    nj= int(str(list_of_list[ii][1]).split(',')[jj])-1
                    jid=jidLista[jj]
                    common.logger.debug(1,"Submitted job # "+`(nj+1)`)
                    common.jobDB.setStatus(nj, 'S')
                    common.jobDB.setJobId(nj, jid)
                    common.jobDB.setTaskId(nj, self.cfg_params['taskId'])
                    njs += 1
                    if not common.logger.debugLevel():
                        if pbar :
                            pbar.update(float(jj+1)/float(len(jidLista)),'please wait')
                    ############################################   
               
                    if st == 'C':
                        resFlag = 0
                    elif st == 'RC':
                        resFlag = 2
                    else:            
                        resFlag = 0
                        pass
                      
                    try:
                        Statistic.Monitor('submit',resFlag,jid,'-----')
                    except:
                        pass
                    
                    fl = open(common.work_space.shareDir() + '/' + self.cfg_params['apmon'].fName, 'r')
                    self.cfg_params['sid'] = jid
                    #### FF: per il momento commentiamo nevtJob che non c'e' piu' nel jobdb
                    #nevtJob = common.jobDB.maxEvents(nj)
               
                    # OLI: JobID treatment, special for Condor-G scheduler
                    jobId = ''
                    if common.scheduler.boss_scheduler_name == 'condor_g':
                        jobId = str(nj + 1) + '_' + self.hash + '_' + self.cfg_params['sid']
                        common.logger.debug(5,'JobID for ML monitoring is created for CONDOR_G scheduler:'+jobId)
                    else:
                        jobId = str(nj + 1) + '_' + self.cfg_params['sid']
                        common.logger.debug(5,'JobID for ML monitoring is created for EDG scheduler'+jobId)
               
                    if ( jid.find(":") != -1 ) :
                        rb = jid.split(':')[1]
                        self.cfg_params['rb'] = rb.replace('//', '')
                    else :
                        self.cfg_params['rb'] = 'OSG'
               
                    #### FF: per il momento commentiamo nevtJob che non c'e' piu' nel jobdb
                    #params = {'nevtJob': nevtJob, 'jobId': jobId, 'sid': self.cfg_params['sid'], \
                    #          'broker': self.cfg_params['rb'], 'bossId': common.jobDB.bossId(nj)}
                    params = {'jobId': jobId, \
                              'sid': self.cfg_params['sid'], \
                              'broker': self.cfg_params['rb'], \
                              'bossId': common.jobDB.bossId(nj), \
                              'TargetCE': string.join((common.jobDB.destination(nj)),",")}
               
                    for i in fl.readlines():
                        val = i.split(':')
                        params[val[0]] = string.strip(val[1])

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

        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)
        return
Revision:	1.66
Committed:	Mon Oct 16 10:07:26 2006 UTC (18 years, 6 months ago) by spiga
Content type:	text/x-python
Branch:	MAIN
CVS Tags:	CRAB_1_4_0_pre4
Changes since 1.65:	+20 -26 lines
Log Message:	add bulk resubmition and some check in clad file
#	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.66	list=''
45			list_of_list = []
46	slacapra	1.57	lastBlock=-1
47	spiga	1.66	count = 0
48	corvo	1.17	for nj in self.nj_list:
49	slacapra	1.53	same=0
50	slacapra	1.49	# first check that status of the job is suitable for submission
51	corvo	1.17	st = common.jobDB.status(nj)
52	spiga	1.66	if st != 'C' and st != 'K' and st != 'A' and st != 'RC': ## commentato per ora...quindi NON risotomette
53	corvo	1.17	long_st = crabJobStatusToString(st)
54	slacapra	1.19	msg = "Job # %d not submitted: status %s"%(nj+1, long_st)
55	slacapra	1.18	common.logger.message(msg)
56	corvo	1.17	continue
57	slacapra	1.57	currBlock = common.jobDB.block(nj)
58			# SL perform listmatch only if block has changed
59			if (currBlock!=lastBlock):
60	gutsche	1.54	if common.scheduler.boss_scheduler_name != "condor_g" :
61	slacapra	1.57	match = common.scheduler.listMatch(nj, currBlock)
62	gutsche	1.54	else :
63			match = "1"
64	slacapra	1.57	lastBlock = currBlock
65	mkirn	1.50	else:
66	slacapra	1.53	common.logger.debug(1,"Sites for job "+str(nj+1)+" the same as previous job")
67			same=1
68	slacapra	1.49	if match:
69	slacapra	1.53	if not same:
70			common.logger.message("Found "+str(match)+" compatible site(s) for job "+str(nj+1))
71			else:
72			common.logger.debug(1,"Found "+str(match)+" compatible site(s) for job "+str(nj+1))
73	spiga	1.66	# job list is string because boss can't manage list
74			list = list+str(nj+1)+','
75			# list.append(nj+1)
76			if nj < self.nj_list[len(self.nj_list)-1]:
77			nextBlock = common.jobDB.block(self.nj_list[count+1])
78			if currBlock != nextBlock :
79			list_of_list.append([currBlock,list])
80			list=''
81			else:
82			list_of_list.append([currBlock,list])
83	slacapra	1.49	else:
84	mkirn	1.50	common.logger.message("No compatible site found, will not submit job "+str(nj+1))
85	slacapra	1.49	continue
86	spiga	1.66	count += 1
87	gutsche	1.65	### Progress Bar indicator, deactivate for debug
88			if not common.logger.debugLevel() :
89			term = TerminalController()
90	slacapra	1.60
91	spiga	1.66	for ii in range(len(list_of_list)): # Add loop DS
92			common.logger.message('Submitting jobs '+str(list_of_list[ii][1]))
93	gutsche	1.65	if not common.logger.debugLevel() :
94			try: pbar = ProgressBar(term, 'Submitting '+str(len(self.nj_list))+' jobs')
95			except: pbar = None
96	spiga	1.56	#common.logger.message("Submitting job # "+`(nj+1)`)
97	spiga	1.66	jidLista = common.scheduler.submit(list_of_list[ii])
98	spiga	1.56	####
99			for jj in range(len(jidLista)): # Add loop over SID returned from group submission DS
100	spiga	1.66	# nj= int(jj+int(list[0]))
101			nj= int(str(list_of_list[ii][1]).split(',')[jj])-1
102	spiga	1.56	jid=jidLista[jj]
103	gutsche	1.59	common.logger.debug(1,"Submitted job # "+`(nj+1)`)
104	spiga	1.56	common.jobDB.setStatus(nj, 'S')
105			common.jobDB.setJobId(nj, jid)
106			common.jobDB.setTaskId(nj, self.cfg_params['taskId'])
107			njs += 1
108	gutsche	1.65	if not common.logger.debugLevel():
109			if pbar :
110			pbar.update(float(jj+1)/float(len(jidLista)),'please wait')
111	spiga	1.56	############################################
112
113			if st == 'C':
114			resFlag = 0
115			elif st == 'RC':
116			resFlag = 2
117			else:
118			resFlag = 0
119			pass
120
121			try:
122			Statistic.Monitor('submit',resFlag,jid,'-----')
123			except:
124			pass
125
126			fl = open(common.work_space.shareDir() + '/' + self.cfg_params['apmon'].fName, 'r')
127			self.cfg_params['sid'] = jid
128			#### FF: per il momento commentiamo nevtJob che non c'e' piu' nel jobdb
129			#nevtJob = common.jobDB.maxEvents(nj)
130
131			# OLI: JobID treatment, special for Condor-G scheduler
132			jobId = ''
133			if common.scheduler.boss_scheduler_name == 'condor_g':
134	gutsche	1.58	jobId = str(nj + 1) + '_' + self.hash + '_' + self.cfg_params['sid']
135	spiga	1.56	common.logger.debug(5,'JobID for ML monitoring is created for CONDOR_G scheduler:'+jobId)
136			else:
137			jobId = str(nj + 1) + '_' + self.cfg_params['sid']
138			common.logger.debug(5,'JobID for ML monitoring is created for EDG scheduler'+jobId)
139
140			if ( jid.find(":") != -1 ) :
141			rb = jid.split(':')[1]
142			self.cfg_params['rb'] = rb.replace('//', '')
143			else :
144			self.cfg_params['rb'] = 'OSG'
145
146			#### FF: per il momento commentiamo nevtJob che non c'e' piu' nel jobdb
147			#params = {'nevtJob': nevtJob, 'jobId': jobId, 'sid': self.cfg_params['sid'], \
148			# 'broker': self.cfg_params['rb'], 'bossId': common.jobDB.bossId(nj)}
149			params = {'jobId': jobId, \
150			'sid': self.cfg_params['sid'], \
151			'broker': self.cfg_params['rb'], \
152			'bossId': common.jobDB.bossId(nj), \
153			'TargetCE': string.join((common.jobDB.destination(nj)),",")}
154
155			for i in fl.readlines():
156			val = i.split(':')
157			params[val[0]] = string.strip(val[1])
158	gutsche	1.65
159			common.logger.debug(5,'Submission DashBoard report: '+str(params))
160
161	spiga	1.56	self.cfg_params['apmon'].sendToML(params)
162	corvo	1.17
163			except:
164	corvo	1.27	exctype, value = sys.exc_info()[:2]
165	corvo	1.30	print "Type: %s Value: %s"%(exctype, value)
166	corvo	1.27	common.logger.message("Submitter::run Exception raised: %s %s"%(exctype, value))
167	corvo	1.17	common.jobDB.save()
168	corvo	1.30
169			stop = time.time()
170	gutsche	1.47	common.logger.debug(1, "Submission Time: "+str(stop - start))
171	slacapra	1.60	common.logger.write("Submission time :"+str(stop - start))
172	corvo	1.17	common.jobDB.save()
173	corvo	1.15
174	nsmirnov	1.6	msg = '\nTotal of %d jobs submitted'%njs
175			if njs != len(self.nj_list) :
176			msg += ' (from %d requested).'%(len(self.nj_list))
177			pass
178			else:
179			msg += '.'
180			pass
181	nsmirnov	1.2	common.logger.message(msg)
182	nsmirnov	1.1	return