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

class Submitter(Actor):
    def __init__(self, cfg_params, nj_list):
        self.cfg_params = cfg_params
        self.nj_list = nj_list
        try:
            self.ML = int(cfg_params['USER.activate_monalisa'])
        except KeyError:
            self.ML = 0
            pass

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

        totalCreatedJobs= 0
        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
        
        firstJob=self.nj_list[0]
        match = common.scheduler.listMatch(firstJob)
        if match:
            common.logger.message("Found "+str(match)+"compatible sites")
        else:
            raise CrabException("No compatible site found!")
        #########
        # Loop over jobs
        start = time.time()
        njs = 0
        try:
            for nj in self.nj_list:
                st = common.jobDB.status(nj)
#                print "nj = ", nj 
#                print "st = ", st
                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

                common.logger.message("Submitting job # "+`(nj+1)`)
                jid = common.scheduler.submit(nj)

                common.jobDB.setStatus(nj, 'S')
                common.jobDB.setJobId(nj, jid)
                njs += 1

                ############################################   

                if st == 'C':
                    resFlag = 0
                elif st == 'RC':
                    resFlag = 2
                else:            
                    resFlag = 0
                    pass
                  
                #try:
                #    print 'submitter prima '

                Statistic.Monitor('submit',resFlag,jid,'-----')  
                #    print 'submitter Dopo     '
 

                #except:
                #    pass
                
                if (self.ML==1):
                    try:
                        #List of parameters to be sent to ML monitor system
                        # Marco. Should be better to put it in the SchedulerEdg/gLite class
                        self.cfg_params['GridName'] = runCommand("grid-proxy-info -identity")
#                        self.cfg_params['taskId'] = self.cfg_params['user']+'_'+self.cfg_params['USER.dataset']+'.'+self.cfg_params['USER.owner']+'_'+str(random()*100)
                        self.cfg_params['taskId'] = self.cfg_params['user']+'_'+self.cfg_params['USER.dataset']+'.'+self.cfg_params['USER.owner']
                        self.cfg_params['jobId'] = str(nj)
                        self.cfg_params['sid'] = jid
                        user = os.getlogin()
                        try:
                            application = os.path.basename(os.environ['SCRAMRT_LOCALRT'])
                        except KeyError:
                            application = os.path.basename(os.environ['LOCALRT'])

                        nevtJob = common.jobDB.maxEvents(nj)
                        tool = common.prog_name+'_v1'
                        taskType = 'analysis'
                        rb = jid.split(':')[1]
                        self.cfg_params['rb'] = rb.replace('//', '')
                        params = {'taskId': self.cfg_params['taskId'], 'jobId': self.cfg_params['jobId'], 'sid': self.cfg_params['sid'], \
                                  'application': application, 'exe': self.cfg_params['USER.executable'], 'nevtJob': nevtJob, 'tool': tool, \
                                  'scheduler': self.cfg_params['CRAB.scheduler'], 'GridName': self.cfg_params['GridName'], 'taskType': taskType, \
                                  'vo': self.cfg_params['EDG.virtual_organization'], 'dataset': self.cfg_params['USER.dataset'],\
                                  'owner': self.cfg_params['USER.owner'], 'broker': self.cfg_params['rb'], 'user': self.cfg_params['user']}
                        self.cfg_params['apmon'].fillDict(params)
                        self.cfg_params['apmon'].sendToML()
                    except:
                        exctype, value = sys.exc_info()[:2]
                        common.logger.message("Submitter::run Exception raised: %s %s"%(exctype, value))
                        pass
                pass # use ML
        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()
#        print "Submission Time: %d "%(stop - start)
        self.cfg_params['apmon'].free() # Free the resources grabbed by apmon

        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.30
Committed:	Thu Dec 15 14:50:20 2005 UTC (19 years, 4 months ago) by corvo
Content type:	text/x-python
Branch:	MAIN
Changes since 1.29:	+24 -28 lines
Log Message:	Added a call to apmon.free() to release resources and kill threads
#	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	nsmirnov	1.4	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	slacapra	1.25	try:
14			self.ML = int(cfg_params['USER.activate_monalisa'])
15			except KeyError:
16			self.ML = 0
17			pass
18	corvo	1.30
19	nsmirnov	1.1	return
20
21			def run(self):
22	nsmirnov	1.2	"""
23			The main method of the class.
24			"""
25			common.logger.debug(5, "Submitter::run() called")
26	slacapra	1.24
27			totalCreatedJobs= 0
28			for nj in range(common.jobDB.nJobs()):
29	spiga	1.29	if (common.jobDB.status(nj)=='C') or (common.jobDB.status(nj)=='RC'): totalCreatedJobs +=1
30	slacapra	1.24	pass
31
32			if (totalCreatedJobs==0):
33			common.logger.message("No jobs to be submitted: first create them")
34	slacapra	1.28	return
35	slacapra	1.24
36	slacapra	1.7	firstJob=self.nj_list[0]
37			match = common.scheduler.listMatch(firstJob)
38	slacapra	1.8	if match:
39	slacapra	1.28	common.logger.message("Found "+str(match)+"compatible sites")
40	slacapra	1.8	else:
41	slacapra	1.28	raise CrabException("No compatible site found!")
42	fanzago	1.16	#########
43	nsmirnov	1.2	# Loop over jobs
44	corvo	1.30	start = time.time()
45	nsmirnov	1.6	njs = 0
46	corvo	1.15	try:
47	corvo	1.17	for nj in self.nj_list:
48			st = common.jobDB.status(nj)
49	corvo	1.21	# print "nj = ", nj
50			# print "st = ", st
51	corvo	1.17	if st != 'C' and st != 'K' and st != 'A' and st != 'RC':
52			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
57			common.logger.message("Submitting job # "+`(nj+1)`)
58			jid = common.scheduler.submit(nj)
59
60			common.jobDB.setStatus(nj, 'S')
61			common.jobDB.setJobId(nj, jid)
62			njs += 1
63
64			############################################
65
66			if st == 'C':
67			resFlag = 0
68			elif st == 'RC':
69			resFlag = 2
70			else:
71			resFlag = 0
72			pass
73	spiga	1.26
74			#try:
75	slacapra	1.28	# print 'submitter prima '
76	spiga	1.26
77			Statistic.Monitor('submit',resFlag,jid,'-----')
78			# print 'submitter Dopo '
79
80	corvo	1.17
81	spiga	1.26	#except:
82			# pass
83	corvo	1.17
84	slacapra	1.25	if (self.ML==1):
85	slacapra	1.22	try:
86	spiga	1.26	#List of parameters to be sent to ML monitor system
87	corvo	1.30	# Marco. Should be better to put it in the SchedulerEdg/gLite class
88			self.cfg_params['GridName'] = runCommand("grid-proxy-info -identity")
89			# self.cfg_params['taskId'] = self.cfg_params['user']+'_'+self.cfg_params['USER.dataset']+'.'+self.cfg_params['USER.owner']+'_'+str(random()*100)
90			self.cfg_params['taskId'] = self.cfg_params['user']+'_'+self.cfg_params['USER.dataset']+'.'+self.cfg_params['USER.owner']
91			self.cfg_params['jobId'] = str(nj)
92			self.cfg_params['sid'] = jid
93	slacapra	1.25	user = os.getlogin()
94			try:
95			application = os.path.basename(os.environ['SCRAMRT_LOCALRT'])
96			except KeyError:
97			application = os.path.basename(os.environ['LOCALRT'])
98
99			nevtJob = common.jobDB.maxEvents(nj)
100	corvo	1.30	tool = common.prog_name+'_v1'
101	slacapra	1.25	taskType = 'analysis'
102	corvo	1.30	rb = jid.split(':')[1]
103			self.cfg_params['rb'] = rb.replace('//', '')
104			params = {'taskId': self.cfg_params['taskId'], 'jobId': self.cfg_params['jobId'], 'sid': self.cfg_params['sid'], \
105			'application': application, 'exe': self.cfg_params['USER.executable'], 'nevtJob': nevtJob, 'tool': tool, \
106			'scheduler': self.cfg_params['CRAB.scheduler'], 'GridName': self.cfg_params['GridName'], 'taskType': taskType, \
107			'vo': self.cfg_params['EDG.virtual_organization'], 'dataset': self.cfg_params['USER.dataset'],\
108			'owner': self.cfg_params['USER.owner'], 'broker': self.cfg_params['rb'], 'user': self.cfg_params['user']}
109			self.cfg_params['apmon'].fillDict(params)
110			self.cfg_params['apmon'].sendToML()
111	slacapra	1.25	except:
112	corvo	1.27	exctype, value = sys.exc_info()[:2]
113			common.logger.message("Submitter::run Exception raised: %s %s"%(exctype, value))
114	slacapra	1.25	pass
115			pass # use ML
116	corvo	1.17	except:
117	corvo	1.27	exctype, value = sys.exc_info()[:2]
118	corvo	1.30	print "Type: %s Value: %s"%(exctype, value)
119	corvo	1.27	common.logger.message("Submitter::run Exception raised: %s %s"%(exctype, value))
120	corvo	1.17	common.jobDB.save()
121	corvo	1.30
122			stop = time.time()
123			# print "Submission Time: %d "%(stop - start)
124			self.cfg_params['apmon'].free() # Free the resources grabbed by apmon
125	slacapra	1.18
126	corvo	1.17	common.jobDB.save()
127	corvo	1.15
128	nsmirnov	1.6	msg = '\nTotal of %d jobs submitted'%njs
129			if njs != len(self.nj_list) :
130			msg += ' (from %d requested).'%(len(self.nj_list))
131			pass
132			else:
133			msg += '.'
134			pass
135	nsmirnov	1.2	common.logger.message(msg)
136	nsmirnov	1.1	return