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")

        return

    def configure(self, cfg_params):
        SchedulerLocal.configure(self, cfg_params)
        self.environment_unique_identifier = "https://"+common.scheduler.name()+":/${JOB_ID}-"+ \
            string.replace(common._db.queryTask('name'),"_","-")

        #self.role = cfg_params.get("EDG.role", None)
        self.role = None
        
        self.pool = cfg_params.get('USER.storage_pool',None)
        ## default is 48 hours CPU time, 2G memrory
        self.cpu = cfg_params.get('USER.cpu',172800)
        self.vmem = cfg_params.get('USER.vmem',2)
        return

    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 += ' -R '+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.3
Committed:	Wed Feb 11 10:01:48 2009 UTC (16 years, 2 months ago) by spiga
Content type:	text/x-python
Branch:	MAIN
CVS Tags:	CRAB_2_5_0_pre5, CRAB_2_5_0_pre4
Changes since 1.2:	+40 -2 lines
Log Message:	sge support upgrade
#	User	Rev	Content
1	slacapra	1.1	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
23			return
24
25			def configure(self, cfg_params):
26			SchedulerLocal.configure(self, cfg_params)
27			self.environment_unique_identifier = "https://"+common.scheduler.name()+":/${JOB_ID}-"+ \
28			string.replace(common._db.queryTask('name'),"_","-")
29
30	spiga	1.3	#self.role = cfg_params.get("EDG.role", None)
31			self.role = None
32
33			self.pool = cfg_params.get('USER.storage_pool',None)
34			## default is 48 hours CPU time, 2G memrory
35			self.cpu = cfg_params.get('USER.cpu',172800)
36			self.vmem = cfg_params.get('USER.vmem',2)
37	slacapra	1.1	return
38
39			def realSchedParams(self,cfg_params):
40			"""
41			Return dictionary with specific parameters, to use
42			with real scheduler
43			"""
44			params = {}
45			return params
46
47			def sched_parameter(self,i,task):
48			"""
49			Returns parameter scheduler-specific, to use with BOSS .
50			"""
51			index = int(common._db.nJobs()) - 1
52			sched_param= ''
53
54			for i in range(index): # Add loop DS
55			sched_param= ''
56			if (self.queue):
57			sched_param += '-q '+self.queue +' '
58			if (self.res): sched_param += ' -R '+self.res +' '
59			pass
60
61	spiga	1.3	#default is request 2G memory and 48 hours CPU time
62			#sched_param += ' -V -l h_vmem=2G -l h_cpu=172800 '
63			sched_param += ' -V -l h_vmem='
64			sched_param += self.vmem.__str__()
65			sched_param += 'G -l h_cpu='
66			sched_param += self.cpu.__str__()
67			sched_param += ' '
68
69	slacapra	1.1	return sched_param
70
71			def loggingInfo(self, id):
72			""" return logging info about job nj """
73			print "Warning: SchedulerSge::loggingInfo not implemented!"
74			return ""
75
76			def wsExitFunc(self):
77			"""
78			"""
79			txt = '\n'
80
81			txt += '#\n'
82			txt += '# EXECUTE THIS FUNCTION BEFORE EXIT \n'
83			txt += '#\n\n'
84
85			txt += 'func_exit() { \n'
86	spiga	1.2	txt += self.wsExitFunc_common()
87
88	slacapra	1.1	txt += ' cp ${SGE_STDOUT_PATH} CMSSW_${NJob}.stdout \n'
89			txt += ' cp ${SGE_STDERR_PATH} CMSSW_${NJob}.stderr \n'
90			txt += ' tar zcvf ${out_files}.tgz ${filesToCheck}\n'
91			txt += ' exit $job_exit_code\n'
92			txt += '}\n'
93
94			return txt
95
96	spiga	1.3	def listMatch(self, dest, full):
97			"""
98			"""
99			#if len(dest)!=0:
100			sites = [self.blackWhiteListParser.cleanForBlackWhiteList(dest,'list')]
101			#else:
102			# sites = [str(getLocalDomain(self))]
103			return sites
104
105			def wsCopyOutput(self):
106			txt=self.wsCopyOutput_comm(self.pool)
107			return txt
108
109			def userName(self):
110			""" return the user name """
111
112			## hack for german naf
113			import pwd,getpass
114			tmp=pwd.getpwnam(getpass.getuser())[4]
115			tmp=tmp.rstrip(',')
116			tmp=tmp.rstrip(',')
117			tmp=tmp.rstrip(',')
118
119
120			return "/CN="+tmp.strip()