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()
        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:
                #    print 'submitter prima '

#                Statistic.Monitor('submit',resFlag,jid,'-----', jobtype_p)
                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 
                        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
            if (self.ML==1):
                self.cfg_params['apmon'].free()
        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.36
Committed:	Tue Mar 14 08:33:56 2006 UTC (19 years, 1 month ago) by corvo
Content type:	text/x-python
Branch:	MAIN
CVS Tags:	CRAB_1_0_6, CRAB_1_0_5
Changes since 1.35:	+1 -1 lines
Log Message:	Uncomment Statistic call from previous version
#	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			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.33	start = time.time()
34	slacapra	1.24	for nj in range(common.jobDB.nJobs()):
35	spiga	1.29	if (common.jobDB.status(nj)=='C') or (common.jobDB.status(nj)=='RC'): totalCreatedJobs +=1
36	slacapra	1.24	pass
37
38			if (totalCreatedJobs==0):
39			common.logger.message("No jobs to be submitted: first create them")
40	slacapra	1.28	return
41	slacapra	1.24
42	slacapra	1.7	firstJob=self.nj_list[0]
43			match = common.scheduler.listMatch(firstJob)
44	slacapra	1.8	if match:
45	slacapra	1.28	common.logger.message("Found "+str(match)+"compatible sites")
46	slacapra	1.8	else:
47	slacapra	1.28	raise CrabException("No compatible site found!")
48	fanzago	1.16	#########
49	nsmirnov	1.2	# Loop over jobs
50	nsmirnov	1.6	njs = 0
51	corvo	1.15	try:
52	corvo	1.17	for nj in self.nj_list:
53			st = common.jobDB.status(nj)
54	corvo	1.21	# print "nj = ", nj
55			# print "st = ", st
56	corvo	1.17	if st != 'C' and st != 'K' and st != 'A' and st != 'RC':
57			long_st = crabJobStatusToString(st)
58	slacapra	1.19	msg = "Job # %d not submitted: status %s"%(nj+1, long_st)
59	slacapra	1.18	common.logger.message(msg)
60	corvo	1.17	continue
61
62			common.logger.message("Submitting job # "+`(nj+1)`)
63			jid = common.scheduler.submit(nj)
64
65			common.jobDB.setStatus(nj, 'S')
66			common.jobDB.setJobId(nj, jid)
67	corvo	1.33	common.jobDB.setTaskId(nj, self.cfg_params['taskId'])
68	corvo	1.17	njs += 1
69
70			############################################
71
72			if st == 'C':
73			resFlag = 0
74			elif st == 'RC':
75			resFlag = 2
76			else:
77			resFlag = 0
78			pass
79	spiga	1.26
80			#try:
81	slacapra	1.28	# print 'submitter prima '
82	spiga	1.26
83	corvo	1.35	# Statistic.Monitor('submit',resFlag,jid,'-----', jobtype_p)
84	corvo	1.36	Statistic.Monitor('submit',resFlag,jid,'-----')
85	spiga	1.26	# print 'submitter Dopo '
86
87	corvo	1.17
88	spiga	1.26	#except:
89			# pass
90	corvo	1.17
91	slacapra	1.25	if (self.ML==1):
92	slacapra	1.22	try:
93	spiga	1.26	#List of parameters to be sent to ML monitor system
94	corvo	1.35	# Marco. Should be better to put it in the SchedulerEdg/gLite class
95			listCE = ','.join(common.analisys_common_info['sites'])
96	corvo	1.30	self.cfg_params['GridName'] = runCommand("grid-proxy-info -identity")
97	fanzago	1.32	common.logger.debug(5, "GRIDNAME: "+self.cfg_params['GridName'])
98	corvo	1.35	self.cfg_params['jobId'] = str(nj + 1)
99	corvo	1.30	self.cfg_params['sid'] = jid
100	slacapra	1.25	nevtJob = common.jobDB.maxEvents(nj)
101			taskType = 'analysis'
102	corvo	1.30	rb = jid.split(':')[1]
103			self.cfg_params['rb'] = rb.replace('//', '')
104	corvo	1.35
105			params = {'jobId': str(nj + 1) + '_' + self.cfg_params['sid'] ,'taskId': self.job_type.getTaskid(), 'sid': self.cfg_params['sid'], \
106			'nevtJob': nevtJob, 'tool': common.prog_name, 'tool_ui': os.environ['HOSTNAME'], \
107	corvo	1.30	'scheduler': self.cfg_params['CRAB.scheduler'], 'GridName': self.cfg_params['GridName'], 'taskType': taskType, \
108	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)}
109			for i in jobtype_p.iterkeys():
110			params[i] = jobtype_p[i]
111			# for j, k in params.iteritems():
112			# print "Values: %s %s"%(j, k)
113	corvo	1.30	self.cfg_params['apmon'].fillDict(params)
114			self.cfg_params['apmon'].sendToML()
115	slacapra	1.25	except:
116	corvo	1.27	exctype, value = sys.exc_info()[:2]
117			common.logger.message("Submitter::run Exception raised: %s %s"%(exctype, value))
118	slacapra	1.25	pass
119			pass # use ML
120	corvo	1.31	if (self.ML==1):
121			self.cfg_params['apmon'].free()
122	corvo	1.17	except:
123	corvo	1.27	exctype, value = sys.exc_info()[:2]
124	corvo	1.30	print "Type: %s Value: %s"%(exctype, value)
125	corvo	1.27	common.logger.message("Submitter::run Exception raised: %s %s"%(exctype, value))
126	corvo	1.17	common.jobDB.save()
127	corvo	1.30
128			stop = time.time()
129	fanzago	1.32	common.logger.debug(5, "Submission Time: "+str(stop - start))
130			#print "Submission Time: %d "%(stop - start)
131	corvo	1.17	common.jobDB.save()
132	corvo	1.15
133	nsmirnov	1.6	msg = '\nTotal of %d jobs submitted'%njs
134			if njs != len(self.nj_list) :
135			msg += ' (from %d requested).'%(len(self.nj_list))
136			pass
137			else:
138			msg += '.'
139			pass
140	nsmirnov	1.2	common.logger.message(msg)
141	nsmirnov	1.1	return