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