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):
    # marco
    def __init__(self, cfg_params, nj_list, job_type):
        self.cfg_params = cfg_params
        self.nj_list = nj_list
        self.job_type = job_type
        
        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
        # marco. Common job type parameters to be sent to ML and Daniele's Monitor
        jobtype_p = self.job_type.getParams()
        listCE = ""
        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
        
        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
        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)
                common.jobDB.setTaskId(nj, self.cfg_params['taskId'])
                njs += 1

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

                if st == 'C':
                    resFlag = 0
                elif st == 'RC':
                    resFlag = 2
                else:            
                    resFlag = 0
                    pass
                  
                try:
                    Statistic.Monitor('submit',resFlag,jid,'-----')
                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 
                        if self.job_type.name() == "ORCA" or self.job_type.name() == 'ORCA_DBSDLS':
                            listCE = ','.join(common.analisys_common_info['sites'])
                        self.cfg_params['GridName'] = runCommand("grid-proxy-info -identity")
                        common.logger.debug(5, "GRIDNAME: "+self.cfg_params['GridName'])
                        self.cfg_params['jobId'] = str(nj + 1)
                        self.cfg_params['sid'] = jid
                        nevtJob = common.jobDB.maxEvents(nj)
                        taskType = 'analysis'
                        rb = jid.split(':')[1]
                        self.cfg_params['rb'] = rb.replace('//', '')

                        params = {'jobId': str(nj + 1) + '_' + self.cfg_params['sid'] ,'taskId': self.job_type.getTaskid(), 'sid': self.cfg_params['sid'], \
                                  'nevtJob': nevtJob, 'tool': common.prog_name, 'tool_ui': os.environ['HOSTNAME'], \
                                  'scheduler': self.cfg_params['CRAB.scheduler'], 'GridName': self.cfg_params['GridName'], 'taskType': taskType, \
                                  'vo': self.cfg_params['EDG.virtual_organization'], 'broker': self.cfg_params['rb'], 'user': self.cfg_params['user'], 'TargetCE': listCE, 'bossId': common.jobDB.bossId(nj)}
                        for i in jobtype_p.iterkeys():
                            params[i] = jobtype_p[i]
#                        for j, k in params.iteritems():
#                            print "Values: %s %s"%(j, k)
                        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()
        common.logger.debug(5, "Submission Time: "+str(stop - start))
        #print "Submission Time: %d "%(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.39
Committed:	Tue Mar 21 16:07:19 2006 UTC (19 years, 1 month ago) by corvo
Content type:	text/x-python
Branch:	MAIN
Changes since 1.38:	+0 -2 lines
Log Message:	Removed the call to free() for apmon
#	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.35	# marco
11			def __init__(self, cfg_params, nj_list, job_type):
12	nsmirnov	1.1	self.cfg_params = cfg_params
13	nsmirnov	1.4	self.nj_list = nj_list
14	corvo	1.35	self.job_type = job_type
15
16	slacapra	1.25	try:
17			self.ML = int(cfg_params['USER.activate_monalisa'])
18			except KeyError:
19			self.ML = 0
20			pass
21	corvo	1.30
22	nsmirnov	1.1	return
23
24			def run(self):
25	nsmirnov	1.2	"""
26			The main method of the class.
27			"""
28			common.logger.debug(5, "Submitter::run() called")
29	slacapra	1.24
30	corvo	1.37	totalCreatedJobs = 0
31	corvo	1.35	# marco. Common job type parameters to be sent to ML and Daniele's Monitor
32			jobtype_p = self.job_type.getParams()
33	corvo	1.37	listCE = ""
34	corvo	1.33	start = time.time()
35	slacapra	1.24	for nj in range(common.jobDB.nJobs()):
36	spiga	1.29	if (common.jobDB.status(nj)=='C') or (common.jobDB.status(nj)=='RC'): totalCreatedJobs +=1
37	slacapra	1.24	pass
38
39			if (totalCreatedJobs==0):
40			common.logger.message("No jobs to be submitted: first create them")
41	slacapra	1.28	return
42	slacapra	1.24
43	slacapra	1.7	firstJob=self.nj_list[0]
44			match = common.scheduler.listMatch(firstJob)
45	slacapra	1.8	if match:
46	slacapra	1.28	common.logger.message("Found "+str(match)+"compatible sites")
47	slacapra	1.8	else:
48	slacapra	1.28	raise CrabException("No compatible site found!")
49	fanzago	1.16	#########
50	nsmirnov	1.2	# Loop over jobs
51	nsmirnov	1.6	njs = 0
52	corvo	1.15	try:
53	corvo	1.17	for nj in self.nj_list:
54			st = common.jobDB.status(nj)
55	corvo	1.21	# print "nj = ", nj
56			# print "st = ", st
57	corvo	1.17	if st != 'C' and st != 'K' and st != 'A' and st != 'RC':
58			long_st = crabJobStatusToString(st)
59	slacapra	1.19	msg = "Job # %d not submitted: status %s"%(nj+1, long_st)
60	slacapra	1.18	common.logger.message(msg)
61	corvo	1.17	continue
62
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	slacapra	1.25	if (self.ML==1):
87	slacapra	1.22	try:
88	spiga	1.26	#List of parameters to be sent to ML monitor system
89	corvo	1.35	# Marco. Should be better to put it in the SchedulerEdg/gLite class
90	spiga	1.38	if self.job_type.name() == "ORCA" or self.job_type.name() == 'ORCA_DBSDLS':
91	corvo	1.37	listCE = ','.join(common.analisys_common_info['sites'])
92	corvo	1.30	self.cfg_params['GridName'] = runCommand("grid-proxy-info -identity")
93	fanzago	1.32	common.logger.debug(5, "GRIDNAME: "+self.cfg_params['GridName'])
94	corvo	1.35	self.cfg_params['jobId'] = str(nj + 1)
95	corvo	1.30	self.cfg_params['sid'] = jid
96	slacapra	1.25	nevtJob = common.jobDB.maxEvents(nj)
97			taskType = 'analysis'
98	corvo	1.30	rb = jid.split(':')[1]
99			self.cfg_params['rb'] = rb.replace('//', '')
100	corvo	1.35
101			params = {'jobId': str(nj + 1) + '_' + self.cfg_params['sid'] ,'taskId': self.job_type.getTaskid(), 'sid': self.cfg_params['sid'], \
102			'nevtJob': nevtJob, 'tool': common.prog_name, 'tool_ui': os.environ['HOSTNAME'], \
103	corvo	1.30	'scheduler': self.cfg_params['CRAB.scheduler'], 'GridName': self.cfg_params['GridName'], 'taskType': taskType, \
104	corvo	1.35	'vo': self.cfg_params['EDG.virtual_organization'], 'broker': self.cfg_params['rb'], 'user': self.cfg_params['user'], 'TargetCE': listCE, 'bossId': common.jobDB.bossId(nj)}
105			for i in jobtype_p.iterkeys():
106			params[i] = jobtype_p[i]
107			# for j, k in params.iteritems():
108			# print "Values: %s %s"%(j, k)
109	corvo	1.30	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	spiga	1.38	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	fanzago	1.32	common.logger.debug(5, "Submission Time: "+str(stop - start))
124			#print "Submission Time: %d "%(stop - start)
125	corvo	1.17	common.jobDB.save()
126	corvo	1.15
127	nsmirnov	1.6	msg = '\nTotal of %d jobs submitted'%njs
128			if njs != len(self.nj_list) :
129			msg += ' (from %d requested).'%(len(self.nj_list))
130			pass
131			else:
132			msg += '.'
133			pass
134	nsmirnov	1.2	common.logger.message(msg)
135	nsmirnov	1.1	return