CRAB/python/SchedulerSge.py

from Scheduler import Scheduler
from SchedulerLocal import SchedulerLocal
from crab_exceptions import *
from crab_util import *
import common

import os,string

#
#  Naming convention:
#  methods starting with 'ws' are responsible to provide
#  corresponding part of the job script ('ws' stands for 'write script').
#
# Author: Hartmut Stadie <stadie@mail.desy.de> Inst. f. Experimentalphysik; Universitaet Hamburg
#

class SchedulerSge(SchedulerLocal) :

    def __init__(self):
        Scheduler.__init__(self,"SGE")
        self.datasetPath   = None
        self.selectNoInput = None
        return

    def configure(self, cfg_params):
        SchedulerLocal.configure(self, cfg_params)

        try:
            tmp =  cfg_params['CMSSW.datasetpath']
            if tmp.lower() == 'none':
                self.datasetPath = None
                self.selectNoInput = 1
            else:
                self.datasetPath = tmp
                self.selectNoInput = 0
        except KeyError:
            msg = "Error: datasetpath not defined "
            raise CrabException(msg)

        self.return_data = cfg_params.get('USER.return_data', 0)
        self.copy_data   = cfg_params.get("USER.copy_data", 0)

        if ( int(self.return_data) == 0 and int(self.copy_data) == 0 ):
            msg = 'Error: return_data and copy_data cannot be set both to 0\n'
            msg = msg + 'Please modify your crab.cfg file\n'
            raise CrabException(msg)

        if ( int(self.return_data) == 1 and int(self.copy_data) == 1 ):
            msg = 'Error: return_data and copy_data cannot be set both to 1\n'
            msg = msg + 'Please modify your crab.cfg file\n'
            raise CrabException(msg)

        if ( int(self.copy_data) == 0 and int(self.publish_data) == 1 ):
            msg = 'Warning: publish_data = 1 must be used with copy_data = 1\n'
            msg = msg + 'Please modify copy_data value in your crab.cfg file\n'
            common.logger.info(msg)
            raise CrabException(msg)

        if int(self.copy_data) == 1:
            self.SE = cfg_params.get('USER.storage_element', None)
            if not self.SE:
                msg = "Error. The [USER] section has no 'storage_element'"
                common.logger.info(msg)
                raise CrabException(msg)

            self.proxyValid = 0
            self.dontCheckProxy = int(cfg_params.get("GRID.dont_check_proxy",0))
            self.proxyServer = cfg_params.get("GRID.proxy_server",'myproxy.cern.ch')
            common.logger.debug('Setting myproxy server to ' + self.proxyServer)

            self.group = cfg_params.get("GRID.group", None)
            self.role  = cfg_params.get("GRID.role", None)
            self.VO    = cfg_params.get('GRID.virtual_organization', 'cms')

            self.checkProxy()

        self.role = None

        self.pool = cfg_params.get('USER.storage_pool',None)
#        self.cpu = cfg_params.get('USER.cpu',172800)
#        self.vmem = cfg_params.get('USER.vmem',2)
        return

    def envUniqueID(self):
        id = "https://"+common.scheduler.name()+":/${JOB_ID}-"+ \
            string.replace(common._db.queryTask('name'),"_","-")
        return id

    def realSchedParams(self,cfg_params):
        """
        Return dictionary with specific parameters, to use
        with real scheduler
        """
        params = {}
        return  params

    def sched_parameter(self,i,task):
        """
        Returns parameter scheduler-specific, to use with BOSS .
        """
        index = int(common._db.nJobs()) - 1
        sched_param= ''

        for i in range(index): # Add loop DS
            sched_param= ''
            if (self.queue):
                sched_param += '-q '+self.queue +' '
                if (self.res): sched_param += ' -l '+self.res +' '
            pass

        #default is request 2G memory and 48 hours CPU time
        #sched_param += ' -V -l h_vmem=2G -l h_cpu=172800 '
#        sched_param += ' -V -l h_vmem='
#        sched_param += self.vmem.__str__()
#        sched_param += 'G -l h_cpu='
#        sched_param += self.cpu.__str__()
#        sched_param += ' '

        return sched_param

    def loggingInfo(self, id):
        """ return logging info about job nj """
        print "Warning: SchedulerSge::loggingInfo not implemented!"
        return ""

    def wsExitFunc(self):
        """
        """
        txt = '\n'

        txt += '#\n'
        txt += '# EXECUTE THIS FUNCTION BEFORE EXIT \n'
        txt += '#\n\n'

        txt += 'func_exit() { \n'
        txt += self.wsExitFunc_common()

        txt += '    cp ${SGE_STDOUT_PATH} CMSSW_${NJob}.stdout \n'
        txt += '    cp ${SGE_STDERR_PATH} CMSSW_${NJob}.stderr \n'
        txt += '    tar zcvf ${out_files}.tgz  ${filesToCheck}\n'
        txt += '    exit $job_exit_code\n'
        txt += '}\n'

        return txt

    def listMatch(self, dest, full):
        """
        """
        #if len(dest)!=0:
        sites = [self.blackWhiteListParser.cleanForBlackWhiteList(dest,'list')]
        #else:
        #    sites = [str(getLocalDomain(self))]
        return sites

    def wsCopyOutput(self):
        txt=self.wsCopyOutput_comm(self.pool)
        return txt

    def userName(self):
        """ return the user name """

        ## hack for german naf
        import pwd,getpass
        tmp=pwd.getpwnam(getpass.getuser())[4]
        tmp=tmp.rstrip(',')
        tmp=tmp.rstrip(',')
        tmp=tmp.rstrip(',')


        return "/CN="+tmp.strip()
Revision:	1.9
Committed:	Tue May 26 16:53:23 2009 UTC (15 years, 11 months ago) by spiga
Content type:	text/x-python
Branch:	MAIN
CVS Tags:	CRAB_2_6_1_pre4, CRAB_2_6_1_pre3, CRAB_2_6_1_pre2, CRAB_2_6_1_pre1, CRAB_2_6_1, CRAB_2_6_0, CRAB_2_6_0_pre14, CRAB_2_6_0_pre13, CRAB_2_6_0_pre12, CRAB_2_6_0_pre11, CRAB_2_6_0_pre10, CRAB_2_6_0_pre9, CRAB_2_6_0_pre8, CRAB_2_6_0_pre7, CRAB_2_6_0_pre6, CRAB_2_6_0_pre5, CRAB_2_6_0_pre4, CRAB_2_6_0_pre3
Changes since 1.8:	+5 -5 lines
Log Message:	adapting code to EDG --> GRID migration
#	User	Rev	Content
1	spiga	1.7	from Scheduler import Scheduler
2			from SchedulerLocal import SchedulerLocal
3			from crab_exceptions import *
4			from crab_util import *
5			import common
6
7			import os,string
8
9			#
10			# Naming convention:
11			# methods starting with 'ws' are responsible to provide
12			# corresponding part of the job script ('ws' stands for 'write script').
13			#
14			# Author: Hartmut Stadie <stadie@mail.desy.de> Inst. f. Experimentalphysik; Universitaet Hamburg
15			#
16
17			class SchedulerSge(SchedulerLocal) :
18
19			def __init__(self):
20			Scheduler.__init__(self,"SGE")
21			self.datasetPath = None
22			self.selectNoInput = None
23	spiga	1.6	return
24	slacapra	1.1
25	spiga	1.7	def configure(self, cfg_params):
26			SchedulerLocal.configure(self, cfg_params)
27
28			try:
29			tmp = cfg_params['CMSSW.datasetpath']
30			if tmp.lower() == 'none':
31			self.datasetPath = None
32			self.selectNoInput = 1
33			else:
34			self.datasetPath = tmp
35			self.selectNoInput = 0
36			except KeyError:
37			msg = "Error: datasetpath not defined "
38			raise CrabException(msg)
39
40			self.return_data = cfg_params.get('USER.return_data', 0)
41			self.copy_data = cfg_params.get("USER.copy_data", 0)
42
43			if ( int(self.return_data) == 0 and int(self.copy_data) == 0 ):
44			msg = 'Error: return_data and copy_data cannot be set both to 0\n'
45			msg = msg + 'Please modify your crab.cfg file\n'
46			raise CrabException(msg)
47
48			if ( int(self.return_data) == 1 and int(self.copy_data) == 1 ):
49			msg = 'Error: return_data and copy_data cannot be set both to 1\n'
50			msg = msg + 'Please modify your crab.cfg file\n'
51			raise CrabException(msg)
52
53			if ( int(self.copy_data) == 0 and int(self.publish_data) == 1 ):
54			msg = 'Warning: publish_data = 1 must be used with copy_data = 1\n'
55			msg = msg + 'Please modify copy_data value in your crab.cfg file\n'
56	spiga	1.8	common.logger.info(msg)
57	spiga	1.7	raise CrabException(msg)
58
59			if int(self.copy_data) == 1:
60			self.SE = cfg_params.get('USER.storage_element', None)
61			if not self.SE:
62			msg = "Error. The [USER] section has no 'storage_element'"
63	spiga	1.8	common.logger.info(msg)
64	spiga	1.7	raise CrabException(msg)
65
66			self.proxyValid = 0
67	spiga	1.9	self.dontCheckProxy = int(cfg_params.get("GRID.dont_check_proxy",0))
68			self.proxyServer = cfg_params.get("GRID.proxy_server",'myproxy.cern.ch')
69	spiga	1.8	common.logger.debug('Setting myproxy server to ' + self.proxyServer)
70	spiga	1.7
71	spiga	1.9	self.group = cfg_params.get("GRID.group", None)
72			self.role = cfg_params.get("GRID.role", None)
73			self.VO = cfg_params.get('GRID.virtual_organization', 'cms')
74	spiga	1.7
75			self.checkProxy()
76
77			self.role = None
78
79			self.pool = cfg_params.get('USER.storage_pool',None)
80			# self.cpu = cfg_params.get('USER.cpu',172800)
81			# self.vmem = cfg_params.get('USER.vmem',2)
82			return
83
84			def envUniqueID(self):
85			id = "https://"+common.scheduler.name()+":/${JOB_ID}-"+ \
86			string.replace(common._db.queryTask('name'),"_","-")
87			return id
88
89			def realSchedParams(self,cfg_params):
90	spiga	1.6	"""
91	spiga	1.7	Return dictionary with specific parameters, to use
92			with real scheduler
93	spiga	1.6	"""
94	spiga	1.7	params = {}
95			return params
96	slacapra	1.1
97	spiga	1.7	def sched_parameter(self,i,task):
98	spiga	1.6	"""
99	spiga	1.7	Returns parameter scheduler-specific, to use with BOSS .
100	spiga	1.6	"""
101	spiga	1.7	index = int(common._db.nJobs()) - 1
102			sched_param= ''
103	ewv	1.5
104	spiga	1.7	for i in range(index): # Add loop DS
105			sched_param= ''
106			if (self.queue):
107			sched_param += '-q '+self.queue +' '
108			if (self.res): sched_param += ' -l '+self.res +' '
109			pass
110	slacapra	1.1
111	spiga	1.7	#default is request 2G memory and 48 hours CPU time
112			#sched_param += ' -V -l h_vmem=2G -l h_cpu=172800 '
113			# sched_param += ' -V -l h_vmem='
114			# sched_param += self.vmem.__str__()
115			# sched_param += 'G -l h_cpu='
116			# sched_param += self.cpu.__str__()
117			# sched_param += ' '
118	spiga	1.4
119	spiga	1.7	return sched_param
120	spiga	1.6
121	spiga	1.7	def loggingInfo(self, id):
122			""" return logging info about job nj """
123			print "Warning: SchedulerSge::loggingInfo not implemented!"
124			return ""
125	spiga	1.6
126	spiga	1.7	def wsExitFunc(self):
127	slacapra	1.1	"""
128			"""
129	spiga	1.7	txt = '\n'
130	slacapra	1.1
131	spiga	1.7	txt += '#\n'
132			txt += '# EXECUTE THIS FUNCTION BEFORE EXIT \n'
133			txt += '#\n\n'
134	slacapra	1.1
135	spiga	1.7	txt += 'func_exit() { \n'
136			txt += self.wsExitFunc_common()
137	spiga	1.3
138	spiga	1.7	txt += ' cp ${SGE_STDOUT_PATH} CMSSW_${NJob}.stdout \n'
139			txt += ' cp ${SGE_STDERR_PATH} CMSSW_${NJob}.stderr \n'
140			txt += ' tar zcvf ${out_files}.tgz ${filesToCheck}\n'
141			txt += ' exit $job_exit_code\n'
142			txt += '}\n'
143	slacapra	1.1
144	spiga	1.7	return txt
145	slacapra	1.1
146	spiga	1.7	def listMatch(self, dest, full):
147	slacapra	1.1	"""
148			"""
149	spiga	1.7	#if len(dest)!=0:
150			sites = [self.blackWhiteListParser.cleanForBlackWhiteList(dest,'list')]
151			#else:
152			# sites = [str(getLocalDomain(self))]
153			return sites
154	slacapra	1.1
155	spiga	1.7	def wsCopyOutput(self):
156			txt=self.wsCopyOutput_comm(self.pool)
157			return txt
158	spiga	1.2
159	spiga	1.7	def userName(self):
160			""" return the user name """
161	slacapra	1.1
162	spiga	1.7	## hack for german naf
163			import pwd,getpass
164			tmp=pwd.getpwnam(getpass.getuser())[4]
165			tmp=tmp.rstrip(',')
166			tmp=tmp.rstrip(',')
167			tmp=tmp.rstrip(',')
168	slacapra	1.1
169	ewv	1.5
170	spiga	1.7	return "/CN="+tmp.strip()