CRAB/python/SchedulerSge.py

from Scheduler import Scheduler
from SchedulerLocal import SchedulerLocal
from crab_exceptions import *
from crab_util import *
from crab_logger import Logger
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.message(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.message(msg)
                raise CrabException(msg)

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

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