CRAB/python/SchedulerLocal.py

from Scheduler import Scheduler
from crab_exceptions import *
from crab_logger import Logger
import common

import os,string

# Base class for all local scheduler

class SchedulerLocal(Scheduler) :

    def configure(self, cfg_params):

        self.jobtypeName = cfg_params['CRAB.jobtype']

        name=string.upper(self.name())
        self.queue = cfg_params.get(name+'.queue',None)

        self.res = cfg_params.get(name+'.resource',None)

        if (cfg_params.has_key(self.name()+'.env_id')): self.environment_unique_identifier = cfg_params[self.name()+'.env_id']

        self._taskId=str("_".join(common._db.queryTask('name').split('_')[:-1]))

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

        self.copy_data = int(cfg_params.get("USER.copy_data",0))
        if self.copy_data == 1:
            self._copyCommand = cfg_params.get('USER.copyCommand','rfcp')
            self.SE_path= cfg_params.get('USER.storage_path',None)
            if not self.SE_path:
                if os.environ.has_key('CASTOR_HOME'):
                    self.SE_path=os.environ['CASTOR_HOME']
                else:
                    msg='No USER.storage_path has been provided: cannot copy_output'
                    raise CrabException(msg)
                pass
            pass
            self.SE_path+='/'

        if ( self.return_data == 0 and self.copy_data == 0 ):
           msg = 'Error: return_data = 0 and copy_data = 0 ==> your exe output will be lost\n'
           msg = msg + 'Please modify return_data and copy_data value in your crab.cfg file\n'
           raise CrabException(msg)

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

        ## Get local domain name
        import socket
        tmp=socket.gethostname()
        dot=string.find(tmp,'.')
        if (dot==-1):
            msg='Unkown domain name. Cannot use local scheduler'
            raise CrabException(msg)
        localDomainName = string.split(tmp,'.',1)[-1]
        #common.taskDB.setDict('localSite',localDomainName)
        ## is this ok?
        cfg_params['EDG.se_white_list']=localDomainName
        common.logger.message("Your domain name is "+str(localDomainName)+": only local dataset will be considered")

        Scheduler.configure(self,cfg_params)
        return

    def userName(self):
        """ return the user name """
        import pwd,getpass
        tmp=pwd.getpwnam(getpass.getuser())[4]
        return "/CN="+tmp.strip()

    def wsSetupEnvironment(self):
        """
        Returns part of a job script which does scheduler-specific work.
        """
        if not self.environment_unique_identifier:
            raise CrabException('environment_unique_identifier not set')

        index = int(common._db.nJobs())
        job = common.job_list[index-1]
        jbt = job.type()
        txt = '# '+self.name()+' specific stuff\n'
        txt += '# strip arguments\n'
        txt += 'echo "strip arguments"\n'
        txt += 'args=("$@")\n'
        txt += 'nargs=$#\n'
        txt += 'shift $nargs\n'
        txt += "# job number (first parameter for job wrapper)\n"
        txt += "NJob=${args[0]}; export NJob\n"

        txt += "out_files=out_files_${NJob}; export out_files\n"
        txt += "echo $out_files\n"
        txt += jbt.outList()

        txt += 'SyncGridJobId=`echo '+self.environment_unique_identifier+'`\n'
        txt += 'MonitorJobID=`echo ${NJob}_${SyncGridJobId}`\n'
        txt += 'MonitorID=`echo ' + self._taskId + '`\n'

        txt += 'echo "MonitorJobID=`echo $MonitorJobID`" | tee -a $RUNTIME_AREA/$repo \n'
        txt += 'echo "SyncGridJobId=`echo $SyncGridJobId`" | tee -a $RUNTIME_AREA/$repo \n'
        txt += 'echo "MonitorID=`echo $MonitorID`" | tee -a $RUNTIME_AREA/$repo\n'
        txt += 'echo "SyncCE='+self.name()+'.`hostname -d`" | tee -a $RUNTIME_AREA/$repo \n'

        txt += 'middleware='+self.name()+' \n'

        txt += 'dumpStatus $RUNTIME_AREA/$repo \n'

        txt += 'InputSandBox=${args[3]}\n'

        txt += '\n\n'

        return txt

    def wsCopyOutput(self):
        """
        Write a CopyResults part of a job script, e.g.
        to copy produced output into a storage element.
        """
        txt = '\n'
        if not self.copy_data: return txt


        txt += '#\n'
        txt += '# COPY OUTPUT FILE TO '+self.SE_path
        txt += '#\n\n'

        txt += 'export SE_PATH='+self.SE_path+'\n'

        txt += 'export CP_CMD='+self._copyCommand+'\n'

        txt += 'echo ">>> Copy output files from WN = `hostname` to PATH = $SE_PATH using $CP_CMD :"\n'

        txt += 'if [ $output_exit_status -eq 60302 ]; then\n'
        txt += '    echo "--> No output file to copy to $SE"\n'
        txt += '    copy_exit_status=$output_exit_status\n'
        txt += '    echo "COPY_EXIT_STATUS = $copy_exit_status"\n'
        txt += 'else\n'
        txt += '    for out_file in $file_list ; do\n'
        txt += '        echo "Trying to copy output file to $SE_PATH"\n'
        txt += '        $CP_CMD $SOFTWARE_DIR/$out_file ${SE_PATH}/$out_file\n'
        txt += '        copy_exit_status=$?\n'
        txt += '        echo "COPY_EXIT_STATUS = $copy_exit_status"\n'
        txt += '        echo "STAGE_OUT = $copy_exit_status"\n'
        txt += '        if [ $copy_exit_status -ne 0 ]; then\n'
        txt += '            echo "Problem copying $out_file to $SE $SE_PATH"\n'
        txt += '            echo "StageOutExitStatus = $copy_exit_status " | tee -a $RUNTIME_AREA/$repo\n'
        #txt += '            copy_exit_status=60307\n'
        txt += '        else\n'
        txt += '            echo "StageOutSE = $SE" | tee -a $RUNTIME_AREA/$repo\n'
        txt += '            echo "StageOutCatalog = " | tee -a $RUNTIME_AREA/$repo\n'
        txt += '            echo "output copied into $SE/$SE_PATH directory"\n'
        txt += '            echo "StageOutExitStatus = 0" | tee -a $RUNTIME_AREA/$repo\n'
        txt += '        fi\n'
        txt += '    done\n'
        txt += 'fi\n'
        txt += 'exit_status=$copy_exit_status\n'

        return txt
Revision:	1.14
Committed:	Sun Apr 20 09:34:40 2008 UTC (17 years ago) by spiga
Content type:	text/x-python
Branch:	MAIN
CVS Tags:	CRAB_2_2_0_pre19, CRAB_2_2_0_pre18, CRAB_2_2_0_pre17, CRAB_2_2_0_pre16, CRAB_2_2_0_pre15, CRAB_2_2_0_pre13, CRAB_2_2_0_pre12, CRAB_2_2_0_pre11, CRAB_2_2_0_pre10
Changes since 1.13:	+1 -1 lines
Log Message:	reimplemented dashboard communication sending again both pre and post submssion infos, plus many Submitter code reorganization
#	User	Rev	Content
1	slacapra	1.1	from Scheduler import Scheduler
2			from crab_exceptions import *
3			from crab_logger import Logger
4			import common
5
6			import os,string
7
8	slacapra	1.2	# Base class for all local scheduler
9	slacapra	1.1
10			class SchedulerLocal(Scheduler) :
11
12			def configure(self, cfg_params):
13
14			self.jobtypeName = cfg_params['CRAB.jobtype']
15
16			name=string.upper(self.name())
17			self.queue = cfg_params.get(name+'.queue',None)
18
19			self.res = cfg_params.get(name+'.resource',None)
20
21			if (cfg_params.has_key(self.name()+'.env_id')): self.environment_unique_identifier = cfg_params[self.name()+'.env_id']
22
23	spiga	1.14	self._taskId=str("_".join(common._db.queryTask('name').split('_')[:-1]))
24	slacapra	1.1
25	slacapra	1.3	self.return_data = int(cfg_params.get('USER.return_data',0))
26
27			self.copy_data = int(cfg_params.get("USER.copy_data",0))
28			if self.copy_data == 1:
29			self._copyCommand = cfg_params.get('USER.copyCommand','rfcp')
30			self.SE_path= cfg_params.get('USER.storage_path',None)
31			if not self.SE_path:
32			if os.environ.has_key('CASTOR_HOME'):
33			self.SE_path=os.environ['CASTOR_HOME']
34			else:
35			msg='No USER.storage_path has been provided: cannot copy_output'
36			raise CrabException(msg)
37			pass
38			pass
39	slacapra	1.10	self.SE_path+='/'
40	slacapra	1.3
41			if ( self.return_data == 0 and self.copy_data == 0 ):
42			msg = 'Error: return_data = 0 and copy_data = 0 ==> your exe output will be lost\n'
43			msg = msg + 'Please modify return_data and copy_data value in your crab.cfg file\n'
44			raise CrabException(msg)
45
46			if ( self.return_data == 1 and self.copy_data == 1 ):
47			msg = 'Error: return_data and copy_data cannot be set both to 1\n'
48			msg = msg + 'Please modify return_data or copy_data value in your crab.cfg file\n'
49			raise CrabException(msg)
50	slacapra	1.1
51			## Get local domain name
52			import socket
53			tmp=socket.gethostname()
54			dot=string.find(tmp,'.')
55			if (dot==-1):
56			msg='Unkown domain name. Cannot use local scheduler'
57			raise CrabException(msg)
58			localDomainName = string.split(tmp,'.',1)[-1]
59	slacapra	1.11	#common.taskDB.setDict('localSite',localDomainName)
60	slacapra	1.1	## is this ok?
61			cfg_params['EDG.se_white_list']=localDomainName
62			common.logger.message("Your domain name is "+str(localDomainName)+": only local dataset will be considered")
63
64	slacapra	1.13	Scheduler.configure(self,cfg_params)
65	slacapra	1.1	return
66
67			def userName(self):
68			""" return the user name """
69	slacapra	1.5	import pwd,getpass
70			tmp=pwd.getpwnam(getpass.getuser())[4]
71			return "/CN="+tmp.strip()
72	slacapra	1.1
73			def wsSetupEnvironment(self):
74			"""
75			Returns part of a job script which does scheduler-specific work.
76			"""
77			if not self.environment_unique_identifier:
78			raise CrabException('environment_unique_identifier not set')
79
80	slacapra	1.12	index = int(common._db.nJobs())
81			job = common.job_list[index-1]
82			jbt = job.type()
83	slacapra	1.1	txt = '# '+self.name()+' specific stuff\n'
84			txt += '# strip arguments\n'
85			txt += 'echo "strip arguments"\n'
86			txt += 'args=("$@")\n'
87			txt += 'nargs=$#\n'
88			txt += 'shift $nargs\n'
89			txt += "# job number (first parameter for job wrapper)\n"
90	ewv	1.9	txt += "NJob=${args[0]}; export NJob\n"
91	slacapra	1.1
92	slacapra	1.12	txt += "out_files=out_files_${NJob}; export out_files\n"
93			txt += "echo $out_files\n"
94			txt += jbt.outList()
95
96	slacapra	1.6	txt += 'SyncGridJobId=`echo '+self.environment_unique_identifier+'`\n'
97			txt += 'MonitorJobID=`echo ${NJob}_${SyncGridJobId}`\n'
98	slacapra	1.1	txt += 'MonitorID=`echo ' + self._taskId + '`\n'
99
100			txt += 'echo "MonitorJobID=`echo $MonitorJobID`" \| tee -a $RUNTIME_AREA/$repo \n'
101			txt += 'echo "SyncGridJobId=`echo $SyncGridJobId`" \| tee -a $RUNTIME_AREA/$repo \n'
102			txt += 'echo "MonitorID=`echo $MonitorID`" \| tee -a $RUNTIME_AREA/$repo\n'
103	slacapra	1.7	txt += 'echo "SyncCE='+self.name()+'.`hostname -d`" \| tee -a $RUNTIME_AREA/$repo \n'
104	slacapra	1.1
105			txt += 'middleware='+self.name()+' \n'
106
107			txt += 'dumpStatus $RUNTIME_AREA/$repo \n'
108
109	slacapra	1.12	txt += 'InputSandBox=${args[3]}\n'
110
111	slacapra	1.1	txt += '\n\n'
112
113			return txt
114
115	slacapra	1.3	def wsCopyOutput(self):
116			"""
117			Write a CopyResults part of a job script, e.g.
118			to copy produced output into a storage element.
119			"""
120	slacapra	1.8	txt = '\n'
121			if not self.copy_data: return txt
122	slacapra	1.3
123
124			txt += '#\n'
125			txt += '# COPY OUTPUT FILE TO '+self.SE_path
126			txt += '#\n\n'
127
128			txt += 'export SE_PATH='+self.SE_path+'\n'
129
130			txt += 'export CP_CMD='+self._copyCommand+'\n'
131
132			txt += 'echo ">>> Copy output files from WN = `hostname` to PATH = $SE_PATH using $CP_CMD :"\n'
133
134			txt += 'if [ $output_exit_status -eq 60302 ]; then\n'
135			txt += ' echo "--> No output file to copy to $SE"\n'
136			txt += ' copy_exit_status=$output_exit_status\n'
137			txt += ' echo "COPY_EXIT_STATUS = $copy_exit_status"\n'
138			txt += 'else\n'
139			txt += ' for out_file in $file_list ; do\n'
140			txt += ' echo "Trying to copy output file to $SE_PATH"\n'
141			txt += ' $CP_CMD $SOFTWARE_DIR/$out_file ${SE_PATH}/$out_file\n'
142			txt += ' copy_exit_status=$?\n'
143			txt += ' echo "COPY_EXIT_STATUS = $copy_exit_status"\n'
144			txt += ' echo "STAGE_OUT = $copy_exit_status"\n'
145			txt += ' if [ $copy_exit_status -ne 0 ]; then\n'
146			txt += ' echo "Problem copying $out_file to $SE $SE_PATH"\n'
147			txt += ' echo "StageOutExitStatus = $copy_exit_status " \| tee -a $RUNTIME_AREA/$repo\n'
148	fanzago	1.4	#txt += ' copy_exit_status=60307\n'
149	slacapra	1.3	txt += ' else\n'
150			txt += ' echo "StageOutSE = $SE" \| tee -a $RUNTIME_AREA/$repo\n'
151			txt += ' echo "StageOutCatalog = " \| tee -a $RUNTIME_AREA/$repo\n'
152			txt += ' echo "output copied into $SE/$SE_PATH directory"\n'
153			txt += ' echo "StageOutExitStatus = 0" \| tee -a $RUNTIME_AREA/$repo\n'
154			txt += ' fi\n'
155			txt += ' done\n'
156			txt += 'fi\n'
157			txt += 'exit_status=$copy_exit_status\n'
158
159			return txt