CRAB/python/Submitter.py

from Actor import *
from crab_util import *
import common
from ApmonIf import ApmonIf
import Statistic

class Submitter(Actor):
    def __init__(self, cfg_params, nj_list):
        self.cfg_params = cfg_params
        self.nj_list = nj_list
        ############################################# 
        fileCODE1 = open(common.work_space.logDir()+"/.code","r")
        array = fileCODE1.read().split('::')
        self.ID1 = array[0]
        self.NJC = array[1]   
        self.dataset = array[2]
        self.owner = array[3]  
        fileCODE1.close()
        
        return
    
    def run(self):
        """
        The main method of the class.
        """

        common.logger.debug(5, "Submitter::run() called")
        
        # Add an instance of ApmonIf to send relevant parameters to ML
        mon = ApmonIf()
        # run a list-match on first job
        #print "lunghezza lista: ", len(self.nj_list)
        firstJob=self.nj_list[0]
        match = common.scheduler.listMatch(firstJob)
        if match:
            common.logger.message("Found compatible resources "+str(match))
        else:
            raise CrabException("No compatible resources found!")
        # Loop over jobs

        njs = 0
        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 is not submitted: status %s"%(nj+1, long_st)
                #common.logger.message(msg)
                continue

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

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

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

            destination = common.scheduler.queryDest(jid).split(":")[0]
            ID3 =  jid.split("/")[3]
            broker = jid.split("/")[2].split(":")[0]
            if st == 'C':
                resFlag = 0
            elif st == 'RC':
                resFlag = 2
            else:            
                resFlag = 0
                pass

            Statistic.notify('submit',resFlag,'-----',self.dataset,self.owner,destination,broker,ID3,self.ID1,self.NJC)
            pass

        ####
        
        common.jobDB.save()
        
        # List of parameters to be sent to monitor system
        user = os.getlogin()
        taskId = os.getlogin()+'_'+string.split(common.work_space.topDir(),'/')[-2] 
        jobId = str(nj)
        sid = jid
        try:
            application = os.path.basename(os.environ['SCRAMRT_LOCALRT'])
        except KeyError:
            application = os.path.basename(os.environ['LOCALRT'])

        nevtJob = 0
        try: 
            nevtJob = self.cfg_params['USER.job_number_of_events']
        except KeyError:
            nevtJob = 0
            pass
        exe = self.cfg_params['USER.executable']
        tool = common.prog_name
        scheduler = self.cfg_params['CRAB.scheduler']
        taskType = 'analysis'
        vo = 'cms'
        dataset = self.cfg_params['USER.dataset']
        owner = self.cfg_params['USER.owner']
        rb = sid.split(':')[1]
        rb = rb.replace('//', '')
        params = {'taskId': taskId, 'jobId': jobId, 'sid': sid, 'application': application, \
                  'exe': exe, 'nevtJob': nevtJob, 'tool': tool, 'scheduler': scheduler, \
                  'user': user, 'taskType': taskType, 'vo': vo, 'dataset': dataset, 'owner': owner, 'broker': rb}
        #for i in params.keys():
            #print "key, value: %s %s" % (i, params[i])
        mon.fillDict(params)
        mon.sendToML()

        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.14
Committed:	Mon Sep 26 15:22:57 2005 UTC (19 years, 7 months ago) by fanzago
Content type:	text/x-python
Branch:	MAIN
CVS Tags:	CRAB_1_0_0_pre1_boss
Changes since 1.13:	+2 -2 lines
Log Message:	commented common.logger.message(msg) at 48 line
#	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	nsmirnov	1.1
7			class Submitter(Actor):
8	nsmirnov	1.4	def __init__(self, cfg_params, nj_list):
9	nsmirnov	1.1	self.cfg_params = cfg_params
10	nsmirnov	1.4	self.nj_list = nj_list
11	spiga	1.10	#############################################
12			fileCODE1 = open(common.work_space.logDir()+"/.code","r")
13			array = fileCODE1.read().split('::')
14			self.ID1 = array[0]
15			self.NJC = array[1]
16			self.dataset = array[2]
17			self.owner = array[3]
18			fileCODE1.close()
19
20	nsmirnov	1.1	return
21
22			def run(self):
23	nsmirnov	1.2	"""
24			The main method of the class.
25			"""
26
27			common.logger.debug(5, "Submitter::run() called")
28	slacapra	1.7
29	corvo	1.9	# Add an instance of ApmonIf to send relevant parameters to ML
30			mon = ApmonIf()
31	slacapra	1.7	# run a list-match on first job
32	fanzago	1.13	#print "lunghezza lista: ", len(self.nj_list)
33	slacapra	1.7	firstJob=self.nj_list[0]
34			match = common.scheduler.listMatch(firstJob)
35	slacapra	1.8	if match:
36			common.logger.message("Found compatible resources "+str(match))
37			else:
38			raise CrabException("No compatible resources found!")
39	nsmirnov	1.2	# Loop over jobs
40
41	nsmirnov	1.6	njs = 0
42	nsmirnov	1.4	for nj in self.nj_list:
43	nsmirnov	1.2	st = common.jobDB.status(nj)
44	fanzago	1.13	#print "nj = ", nj
45			#print "st = ", st
46	spiga	1.10	if st != 'C' and st != 'K' and st != 'A' and st != 'RC':
47	nsmirnov	1.6	long_st = crabJobStatusToString(st)
48	fanzago	1.14	#msg = "Job # %d is not submitted: status %s"%(nj+1, long_st)
49			#common.logger.message(msg)
50	nsmirnov	1.6	continue
51	nsmirnov	1.2
52	nsmirnov	1.6	common.logger.message("Submitting job # "+`(nj+1)`)
53	fanzago	1.13	print "submitting job!"
54	nsmirnov	1.2	jid = common.scheduler.submit(nj)
55
56			common.jobDB.setStatus(nj, 'S')
57			common.jobDB.setJobId(nj, jid)
58	nsmirnov	1.6	njs += 1
59	spiga	1.10
60			############################################
61
62			destination = common.scheduler.queryDest(jid).split(":")[0]
63			ID3 = jid.split("/")[3]
64			broker = jid.split("/")[2].split(":")[0]
65			if st == 'C':
66			resFlag = 0
67			elif st == 'RC':
68			resFlag = 2
69			else:
70			resFlag = 0
71			pass
72
73	corvo	1.11	Statistic.notify('submit',resFlag,'-----',self.dataset,self.owner,destination,broker,ID3,self.ID1,self.NJC)
74	nsmirnov	1.2	pass
75
76			####
77
78			common.jobDB.save()
79	corvo	1.9
80			# List of parameters to be sent to monitor system
81			user = os.getlogin()
82			taskId = os.getlogin()+'_'+string.split(common.work_space.topDir(),'/')[-2]
83			jobId = str(nj)
84			sid = jid
85			try:
86			application = os.path.basename(os.environ['SCRAMRT_LOCALRT'])
87			except KeyError:
88			application = os.path.basename(os.environ['LOCALRT'])
89	slacapra	1.12
90			nevtJob = 0
91	corvo	1.9	try:
92			nevtJob = self.cfg_params['USER.job_number_of_events']
93			except KeyError:
94	fanzago	1.13	nevtJob = 0
95	corvo	1.9	pass
96			exe = self.cfg_params['USER.executable']
97			tool = common.prog_name
98			scheduler = self.cfg_params['CRAB.scheduler']
99			taskType = 'analysis'
100			vo = 'cms'
101			dataset = self.cfg_params['USER.dataset']
102			owner = self.cfg_params['USER.owner']
103			rb = sid.split(':')[1]
104			rb = rb.replace('//', '')
105			params = {'taskId': taskId, 'jobId': jobId, 'sid': sid, 'application': application, \
106			'exe': exe, 'nevtJob': nevtJob, 'tool': tool, 'scheduler': scheduler, \
107	corvo	1.11	'user': user, 'taskType': taskType, 'vo': vo, 'dataset': dataset, 'owner': owner, 'broker': rb}
108	fanzago	1.13	#for i in params.keys():
109			#print "key, value: %s %s" % (i, params[i])
110	corvo	1.9	mon.fillDict(params)
111			mon.sendToML()
112	nsmirnov	1.2
113	nsmirnov	1.6	msg = '\nTotal of %d jobs submitted'%njs
114			if njs != len(self.nj_list) :
115			msg += ' (from %d requested).'%(len(self.nj_list))
116			pass
117			else:
118			msg += '.'
119			pass
120	nsmirnov	1.2	common.logger.message(msg)
121	nsmirnov	1.1	return
122