CRAB/python/SchedulerEdg.py

from SchedulerGrid import SchedulerGrid
from crab_logger import Logger
from crab_exceptions import *
from crab_util import *
from EdgConfig import *
from BlackWhiteListParser import BlackWhiteListParser
import common

import os, sys, time

class SchedulerEdg(SchedulerGrid):
    def __init__(self):
        SchedulerGrid.__init__(self,"EDG")

    def configure(self,cfg_params):
        SchedulerGrid.configure(self, cfg_params)
        self.environment_unique_identifier = 'EDG_WL_JOBID'

    def rb_configure(self, RB):
        edg_config = None
        edg_config_vo = None
        rb_param_file = None

        edgConfig = EdgConfig(RB)
        edg_config = edgConfig.config()
        edg_config_vo = edgConfig.configVO()

        if (edg_config and edg_config_vo):
            rb_param_file = 'RBconfig = "'+edg_config+'";\nRBconfigVO = "'+edg_config_vo+'";\n'
        return rb_param_file

    def sched_parameter(self):
        """
        Returns file with requirements and scheduler-specific parameters
        """
        index = int(common.jobDB.nJobs()) - 1
        job = common.job_list[index]
        jbt = job.type()

        lastBlock=-1
        first = []
        for n in range(common.jobDB.nJobs()):
            currBlock=common.jobDB.block(n)
            if (currBlock!=lastBlock):
                lastBlock = currBlock
                first.append(n)

        req = ''
        req = req + jbt.getRequirements()

        if self.EDG_requirements:
            if (req == ' '):
                req = req + self.EDG_requirements
            else:
                req = req +  ' && ' + self.EDG_requirements

        if self.EDG_ce_white_list:
            ce_white_list = string.split(self.EDG_ce_white_list,',')
            for i in range(len(ce_white_list)):
                if i == 0:
                    if (req == ' '):
                        req = req + '((RegExp("' + string.strip(ce_white_list[i]) + '", other.GlueCEUniqueId))'
                    else:
                        req = req +  ' && ((RegExp("' +  string.strip(ce_white_list[i]) + '", other.GlueCEUniqueId))'
                    pass
                else:
                    req = req +  ' || (RegExp("' +  string.strip(ce_white_list[i]) + '", other.GlueCEUniqueId))'
            req = req + ')'

        if self.EDG_ce_black_list:
            ce_black_list = string.split(self.EDG_ce_black_list,',')
            for ce in ce_black_list:
                if (req == ' '):
                    req = req + '(!RegExp("' + string.strip(ce) + '", other.GlueCEUniqueId))'
                else:
                    req = req +  ' && (!RegExp("' + string.strip(ce) + '", other.GlueCEUniqueId))'
                pass
        if self.EDG_clock_time:
            if (req == ' '):
                req = req + 'other.GlueCEPolicyMaxWallClockTime>='+self.EDG_clock_time
            else:
                req = req + ' && other.GlueCEPolicyMaxWallClockTime>='+self.EDG_clock_time

        if self.EDG_cpu_time:
            if (req == ' '):
                req = req + ' other.GlueCEPolicyMaxCPUTime>='+self.EDG_cpu_time
            else:
                req = req + ' && other.GlueCEPolicyMaxCPUTime>='+self.EDG_cpu_time

        for i in range(len(first)): # Add loop DS
            groupReq = req
            self.param='sched_param_'+str(i)+'.clad'
            param_file = open(common.work_space.shareDir()+'/'+self.param, 'w')

            itr4=self.findSites_(first[i])
            for arg in itr4:
                groupReq = groupReq + ' && anyMatch(other.storage.CloseSEs, ('+str(arg)+'))'
            param_file.write('Requirements = '+groupReq +';\n')

            if (self.rb_param_file):
                param_file.write(self.rb_param_file)

            if self.EDG_addJdlParam:
                if self.EDG_addJdlParam[-1] == '': self.EDG_addJdlParam= self.EDG_addJdlParam[:-1] 
                for p in self.EDG_addJdlParam:
                    param_file.write(string.strip(p)+';\n')

            param_file.close()


    def loggingInfo(self, id):
        """
        retrieve the logging info from logging and bookkeeping and return it
        """
        self.checkProxy()
        cmd = 'edg-job-get-logging-info -v 2 ' + id
        cmd_out = runCommand(cmd)
        return cmd_out

    def queryDetailedStatus(self, id):
        """ Query a detailed status of the job with id """
        cmd = 'edg-job-status '+id
        cmd_out = runCommand(cmd)
        return cmd_out

    def findSites_(self, n):
        itr4 =[]
        sites = common.jobDB.destination(n)
        if len(sites)>0 and sites[0]=="":
            return itr4

        itr = ''
        if sites != [""]:#CarlosDaniele
            ##Addedd Daniele
            replicas = self.blackWhiteListParser.checkBlackList(sites,n)
            if len(replicas)!=0:
                replicas = self.blackWhiteListParser.checkWhiteList(replicas,n)

            if len(replicas)==0:
                itr = itr + 'target.GlueSEUniqueID=="NONE" '
                #msg = 'No sites remaining that host any part of the requested data! Exiting... '
                #raise CrabException(msg)
            #####
           # for site in sites:
            for site in replicas:
                #itr = itr + 'target.GlueSEUniqueID==&quot;'+site+'&quot; || '
                itr = itr + 'target.GlueSEUniqueID=="'+site+'" || '
            itr = itr[0:-4]
            itr4.append( itr )
        return itr4

    def tOut(self, list):
        return 120


Revision:	1.159
Committed:	Thu Feb 7 19:49:58 2008 UTC (17 years, 2 months ago) by spiga
Content type:	text/x-python
Branch:	MAIN
Changes since 1.158:	+1 -1 lines
Log Message:	fixed sintax error while checking addJdlParam
#	User	Rev	Content
1	slacapra	1.158	from SchedulerGrid import SchedulerGrid
2	nsmirnov	1.1	from crab_logger import Logger
3			from crab_exceptions import *
4			from crab_util import *
5	slacapra	1.50	from EdgConfig import *
6	gutsche	1.130	from BlackWhiteListParser import BlackWhiteListParser
7	nsmirnov	1.1	import common
8
9	slacapra	1.113	import os, sys, time
10	nsmirnov	1.1
11	slacapra	1.158	class SchedulerEdg(SchedulerGrid):
12	nsmirnov	1.1	def __init__(self):
13	slacapra	1.158	SchedulerGrid.__init__(self,"EDG")
14	ewv	1.154
15	slacapra	1.158	def configure(self,cfg_params):
16			SchedulerGrid.configure(self, cfg_params)
17			self.environment_unique_identifier = 'EDG_WL_JOBID'
18	fanzago	1.10
19	fanzago	1.99	def rb_configure(self, RB):
20	slacapra	1.158	edg_config = None
21			edg_config_vo = None
22			rb_param_file = None
23	fanzago	1.99
24			edgConfig = EdgConfig(RB)
25	slacapra	1.158	edg_config = edgConfig.config()
26			edg_config_vo = edgConfig.configVO()
27	ewv	1.154
28	slacapra	1.158	if (edg_config and edg_config_vo):
29			rb_param_file = 'RBconfig = "'+edg_config+'";\nRBconfigVO = "'+edg_config_vo+'";\n'
30			return rb_param_file
31	fanzago	1.99
32	fanzago	1.10	def sched_parameter(self):
33			"""
34	spiga	1.88	Returns file with requirements and scheduler-specific parameters
35	fanzago	1.10	"""
36	spiga	1.88	index = int(common.jobDB.nJobs()) - 1
37			job = common.job_list[index]
38			jbt = job.type()
39	ewv	1.154
40	slacapra	1.95	lastBlock=-1
41	spiga	1.88	first = []
42			for n in range(common.jobDB.nJobs()):
43	slacapra	1.95	currBlock=common.jobDB.block(n)
44			if (currBlock!=lastBlock):
45			lastBlock = currBlock
46	spiga	1.88	first.append(n)
47	ewv	1.154
48	spiga	1.88	req = ''
49			req = req + jbt.getRequirements()
50	ewv	1.154
51	spiga	1.88	if self.EDG_requirements:
52			if (req == ' '):
53			req = req + self.EDG_requirements
54			else:
55			req = req + ' && ' + self.EDG_requirements
56	slacapra	1.95
57	spiga	1.88	if self.EDG_ce_white_list:
58			ce_white_list = string.split(self.EDG_ce_white_list,',')
59			for i in range(len(ce_white_list)):
60			if i == 0:
61			if (req == ' '):
62	spiga	1.106	req = req + '((RegExp("' + string.strip(ce_white_list[i]) + '", other.GlueCEUniqueId))'
63	spiga	1.88	else:
64	spiga	1.106	req = req + ' && ((RegExp("' + string.strip(ce_white_list[i]) + '", other.GlueCEUniqueId))'
65	spiga	1.88	pass
66			else:
67	spiga	1.106	req = req + ' \|\| (RegExp("' + string.strip(ce_white_list[i]) + '", other.GlueCEUniqueId))'
68	spiga	1.88	req = req + ')'
69	ewv	1.154
70	spiga	1.88	if self.EDG_ce_black_list:
71			ce_black_list = string.split(self.EDG_ce_black_list,',')
72			for ce in ce_black_list:
73			if (req == ' '):
74	spiga	1.106	req = req + '(!RegExp("' + string.strip(ce) + '", other.GlueCEUniqueId))'
75	spiga	1.88	else:
76	spiga	1.106	req = req + ' && (!RegExp("' + string.strip(ce) + '", other.GlueCEUniqueId))'
77	spiga	1.88	pass
78			if self.EDG_clock_time:
79			if (req == ' '):
80			req = req + 'other.GlueCEPolicyMaxWallClockTime>='+self.EDG_clock_time
81			else:
82			req = req + ' && other.GlueCEPolicyMaxWallClockTime>='+self.EDG_clock_time
83
84			if self.EDG_cpu_time:
85			if (req == ' '):
86			req = req + ' other.GlueCEPolicyMaxCPUTime>='+self.EDG_cpu_time
87			else:
88			req = req + ' && other.GlueCEPolicyMaxCPUTime>='+self.EDG_cpu_time
89	ewv	1.154
90	spiga	1.88	for i in range(len(first)): # Add loop DS
91	spiga	1.110	groupReq = req
92	spiga	1.88	self.param='sched_param_'+str(i)+'.clad'
93	fanzago	1.10	param_file = open(common.work_space.shareDir()+'/'+self.param, 'w')
94	spiga	1.88
95			itr4=self.findSites_(first[i])
96	slacapra	1.95	for arg in itr4:
97	spiga	1.110	groupReq = groupReq + ' && anyMatch(other.storage.CloseSEs, ('+str(arg)+'))'
98	ewv	1.154	param_file.write('Requirements = '+groupReq +';\n')
99
100	slacapra	1.158	if (self.rb_param_file):
101	ewv	1.154	param_file.write(self.rb_param_file)
102	spiga	1.88
103	spiga	1.159	if self.EDG_addJdlParam:
104	spiga	1.155	if self.EDG_addJdlParam[-1] == '': self.EDG_addJdlParam= self.EDG_addJdlParam[:-1]
105	slacapra	1.101	for p in self.EDG_addJdlParam:
106	spiga	1.155	param_file.write(string.strip(p)+';\n')
107	slacapra	1.101
108	ewv	1.154	param_file.close()
109	spiga	1.88
110	fanzago	1.13
111	spiga	1.23	def loggingInfo(self, id):
112	slacapra	1.7	"""
113			retrieve the logging info from logging and bookkeeping and return it
114			"""
115	slacapra	1.18	self.checkProxy()
116	fanzago	1.24	cmd = 'edg-job-get-logging-info -v 2 ' + id
117	slacapra	1.32	cmd_out = runCommand(cmd)
118	slacapra	1.7	return cmd_out
119
120	nsmirnov	1.1	def queryDetailedStatus(self, id):
121			""" Query a detailed status of the job with id """
122			cmd = 'edg-job-status '+id
123			cmd_out = runCommand(cmd)
124			return cmd_out
125
126	spiga	1.88	def findSites_(self, n):
127	ewv	1.154	itr4 =[]
128	spiga	1.88	sites = common.jobDB.destination(n)
129	spiga	1.128	if len(sites)>0 and sites[0]=="":
130	spiga	1.88	return itr4
131	spiga	1.128
132	spiga	1.88	itr = ''
133	ewv	1.154	if sites != [""]:#CarlosDaniele
134	spiga	1.128	##Addedd Daniele
135	gutsche	1.130	replicas = self.blackWhiteListParser.checkBlackList(sites,n)
136	spiga	1.128	if len(replicas)!=0:
137	gutsche	1.130	replicas = self.blackWhiteListParser.checkWhiteList(replicas,n)
138	ewv	1.154
139	spiga	1.128	if len(replicas)==0:
140	mcinquil	1.146	itr = itr + 'target.GlueSEUniqueID=="NONE" '
141			#msg = 'No sites remaining that host any part of the requested data! Exiting... '
142			#raise CrabException(msg)
143	ewv	1.154	#####
144			# for site in sites:
145			for site in replicas:
146	spiga	1.88	#itr = itr + 'target.GlueSEUniqueID=="'+site+'" \|\| '
147			itr = itr + 'target.GlueSEUniqueID=="'+site+'" \|\| '
148			itr = itr[0:-4]
149			itr4.append( itr )
150	slacapra	1.80	return itr4
151
152	mcinquil	1.151	def tOut(self, list):
153	spiga	1.121	return 120
154	ewv	1.154
155	spiga	1.121