ViewVC Help
View File | Revision Log | Show Annotations | Root Listing
root/cvsroot/COMP/CRAB/python/SchedulerLocal.py
Revision: 1.5
Committed: Mon Feb 4 14:42:19 2008 UTC (17 years, 2 months ago) by slacapra
Content type: text/x-python
Branch: MAIN
Changes since 1.4: +3 -8 lines
Log Message:
fix to report correctly info o dashboard

File Contents

# 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 slacapra 1.2 Scheduler.configure(self,cfg_params)
14 slacapra 1.1
15     self.jobtypeName = cfg_params['CRAB.jobtype']
16    
17     name=string.upper(self.name())
18     self.queue = cfg_params.get(name+'.queue',None)
19    
20     self.res = cfg_params.get(name+'.resource',None)
21    
22     if (cfg_params.has_key(self.name()+'.env_id')): self.environment_unique_identifier = cfg_params[self.name()+'.env_id']
23    
24     self._taskId = common.taskDB.dict('taskId')
25    
26 slacapra 1.3 self.return_data = int(cfg_params.get('USER.return_data',0))
27    
28     self.copy_data = int(cfg_params.get("USER.copy_data",0))
29     if self.copy_data == 1:
30     self._copyCommand = cfg_params.get('USER.copyCommand','rfcp')
31     self.SE_path= cfg_params.get('USER.storage_path',None)
32     self.SE_path+='/'
33     if not self.SE_path:
34     if os.environ.has_key('CASTOR_HOME'):
35     self.SE_path=os.environ['CASTOR_HOME']
36     else:
37     msg='No USER.storage_path has been provided: cannot copy_output'
38     raise CrabException(msg)
39     pass
40     pass
41    
42     if ( self.return_data == 0 and self.copy_data == 0 ):
43     msg = 'Error: return_data = 0 and copy_data = 0 ==> your exe output will be lost\n'
44     msg = msg + 'Please modify return_data and copy_data value in your crab.cfg file\n'
45     raise CrabException(msg)
46    
47     if ( self.return_data == 1 and self.copy_data == 1 ):
48     msg = 'Error: return_data and copy_data cannot be set both to 1\n'
49     msg = msg + 'Please modify return_data or copy_data value in your crab.cfg file\n'
50     raise CrabException(msg)
51 slacapra 1.1
52     ## Get local domain name
53     import socket
54     tmp=socket.gethostname()
55     dot=string.find(tmp,'.')
56     if (dot==-1):
57     msg='Unkown domain name. Cannot use local scheduler'
58     raise CrabException(msg)
59     localDomainName = string.split(tmp,'.',1)[-1]
60     ## 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     return
65    
66     def userName(self):
67     """ return the user name """
68 slacapra 1.5 import pwd,getpass
69     tmp=pwd.getpwnam(getpass.getuser())[4]
70     return "/CN="+tmp.strip()
71 slacapra 1.1
72     def wsSetupEnvironment(self):
73     """
74     Returns part of a job script which does scheduler-specific work.
75     """
76     if not self.environment_unique_identifier:
77     raise CrabException('environment_unique_identifier not set')
78    
79     txt = '# '+self.name()+' specific stuff\n'
80     txt += '# strip arguments\n'
81     txt += 'echo "strip arguments"\n'
82     txt += 'args=("$@")\n'
83     txt += 'nargs=$#\n'
84     txt += 'shift $nargs\n'
85     txt += "# job number (first parameter for job wrapper)\n"
86     txt += "NJob=${args[0]}\n"
87    
88     txt += 'MonitorJobID=`echo ${NJob}_$'+self.environment_unique_identifier+'`\n'
89     txt += 'SyncGridJobId=`echo $'+self.environment_unique_identifier+'`\n'
90     txt += 'MonitorID=`echo ' + self._taskId + '`\n'
91    
92     txt += 'echo "MonitorJobID=`echo $MonitorJobID`" | tee -a $RUNTIME_AREA/$repo \n'
93     txt += 'echo "SyncGridJobId=`echo $SyncGridJobId`" | tee -a $RUNTIME_AREA/$repo \n'
94     txt += 'echo "MonitorID=`echo $MonitorID`" | tee -a $RUNTIME_AREA/$repo\n'
95    
96     txt += 'middleware='+self.name()+' \n'
97    
98     txt += 'dumpStatus $RUNTIME_AREA/$repo \n'
99    
100     txt += '\n\n'
101    
102     return txt
103    
104 slacapra 1.3 def wsCopyOutput(self):
105     """
106     Write a CopyResults part of a job script, e.g.
107     to copy produced output into a storage element.
108     """
109     if not self.copy_data: return
110    
111     txt = '\n'
112    
113     txt += '#\n'
114     txt += '# COPY OUTPUT FILE TO '+self.SE_path
115     txt += '#\n\n'
116    
117     txt += 'export SE_PATH='+self.SE_path+'\n'
118    
119     txt += 'export CP_CMD='+self._copyCommand+'\n'
120    
121     txt += 'echo ">>> Copy output files from WN = `hostname` to PATH = $SE_PATH using $CP_CMD :"\n'
122    
123     txt += 'if [ $output_exit_status -eq 60302 ]; then\n'
124     txt += ' echo "--> No output file to copy to $SE"\n'
125     txt += ' copy_exit_status=$output_exit_status\n'
126     txt += ' echo "COPY_EXIT_STATUS = $copy_exit_status"\n'
127     txt += 'else\n'
128     txt += ' for out_file in $file_list ; do\n'
129     txt += ' echo "Trying to copy output file to $SE_PATH"\n'
130     txt += ' $CP_CMD $SOFTWARE_DIR/$out_file ${SE_PATH}/$out_file\n'
131     txt += ' copy_exit_status=$?\n'
132     txt += ' echo "COPY_EXIT_STATUS = $copy_exit_status"\n'
133     txt += ' echo "STAGE_OUT = $copy_exit_status"\n'
134     txt += ' if [ $copy_exit_status -ne 0 ]; then\n'
135     txt += ' echo "Problem copying $out_file to $SE $SE_PATH"\n'
136     txt += ' echo "StageOutExitStatus = $copy_exit_status " | tee -a $RUNTIME_AREA/$repo\n'
137 fanzago 1.4 #txt += ' copy_exit_status=60307\n'
138 slacapra 1.3 txt += ' else\n'
139     txt += ' echo "StageOutSE = $SE" | tee -a $RUNTIME_AREA/$repo\n'
140     txt += ' echo "StageOutCatalog = " | tee -a $RUNTIME_AREA/$repo\n'
141     txt += ' echo "output copied into $SE/$SE_PATH directory"\n'
142     txt += ' echo "StageOutExitStatus = 0" | tee -a $RUNTIME_AREA/$repo\n'
143     txt += ' fi\n'
144     txt += ' done\n'
145     txt += 'fi\n'
146     txt += 'exit_status=$copy_exit_status\n'
147    
148     return txt
149    
150 slacapra 1.1 def createXMLSchScript(self, nj, argsList):
151    
152     """
153     Create a XML-file for BOSS4.
154     """
155    
156     """
157     INDY
158     [begin] FIX-ME:
159     I would pass jobType instead of job
160     """
161     index = nj - 1
162     job = common.job_list[index]
163     jbt = job.type()
164     inp_sandbox = jbt.inputSandbox(index)
165     #out_sandbox = jbt.outputSandbox(index)
166     """
167     [end] FIX-ME
168     """
169    
170    
171     title = '<?xml version="1.0" encoding="UTF-8" standalone="no"?>\n'
172     jt_string = ''
173    
174     xml_fname = str(self.jobtypeName)+'.xml'
175     xml = open(common.work_space.shareDir()+'/'+xml_fname, 'a')
176    
177     #TaskName
178     dir = string.split(common.work_space.topDir(), '/')
179     taskName = dir[len(dir)-2]
180    
181     to_write = ''
182    
183     req=' '
184     req = req + jbt.getRequirements()
185    
186     #TaskName
187     dir = string.split(common.work_space.topDir(), '/')
188     taskName = dir[len(dir)-2]
189    
190     xml.write(str(title))
191    
192     #First check the X509_USER_PROXY. In not there use the default
193     xml.write('<task name="' +str(taskName)+ '" sub_path="' +common.work_space.pathForTgz() + 'share/.boss_cache"' + '>\n')
194    
195     xml.write(jt_string)
196    
197     if (to_write != ''):
198     xml.write('<extraTags\n')
199     xml.write(to_write)
200     xml.write('/>\n')
201     pass
202    
203     xml.write('<iterator>\n')
204     xml.write('\t<iteratorRule name="ITR1">\n')
205     xml.write('\t\t<ruleElement> 1:'+ str(nj) + ' </ruleElement>\n')
206     xml.write('\t</iteratorRule>\n')
207     xml.write('\t<iteratorRule name="ITR2">\n')
208     for arg in argsList:
209     xml.write('\t\t<ruleElement> <![CDATA[\n'+ arg + '\n\t\t]]> </ruleElement>\n')
210     pass
211     xml.write('\t</iteratorRule>\n')
212     #print jobList
213     xml.write('\t<iteratorRule name="ITR3">\n')
214     xml.write('\t\t<ruleElement> 1:'+ str(nj) + ':1:6 </ruleElement>\n')
215     xml.write('\t</iteratorRule>\n')
216    
217     xml.write('<chain name="' +str(taskName)+'__ITR1_" scheduler="'+str(self.name())+'">\n')
218     # xml.write('<chain scheduler="'+str(self.schedulerName)+'">\n')
219     xml.write(jt_string)
220    
221     #executable
222    
223     script = job.scriptFilename()
224     xml.write('<program>\n')
225     xml.write('<exec> ' + os.path.basename(script) +' </exec>\n')
226     xml.write(jt_string)
227    
228     xml.write('<args> <![CDATA[\n _ITR2_ \n]]> </args>\n')
229     xml.write('<program_types> crabjob </program_types>\n')
230     inp_box = common.work_space.pathForTgz() + 'job/' + jbt.scriptName + ','
231    
232     if inp_sandbox != None:
233     for fl in inp_sandbox:
234     inp_box = inp_box + '' + fl + ','
235     pass
236     pass
237    
238     if inp_box[-1] == ',' : inp_box = inp_box[:-1]
239     inp_box = '<infiles> <![CDATA[\n' + inp_box + '\n]]> </infiles>\n'
240     xml.write(inp_box)
241    
242     base = jbt.name()
243     stdout = base + '__ITR3_.stdout'
244     stderr = base + '__ITR3_.stderr'
245    
246     xml.write('<stderr> ' + stderr + '</stderr>\n')
247     xml.write('<stdout> ' + stdout + '</stdout>\n')
248    
249    
250     out_box = stdout + ',' + \
251     stderr + ',.BrokerInfo,'
252    
253     # Stuff to be returned _always_ via sandbox
254     for fl in jbt.output_file_sandbox:
255     out_box = out_box + '' + jbt.numberFile_(fl, '_ITR1_') + ','
256     pass
257     pass
258    
259     # via sandbox iif required return_data
260     if int(self.return_data) == 1:
261     for fl in jbt.output_file:
262     out_box = out_box + '' + jbt.numberFile_(fl, '_ITR1_') + ','
263     pass
264     pass
265    
266     if out_box[-1] == ',' : out_box = out_box[:-1]
267     out_box = '<outfiles> <![CDATA[\n' + out_box + '\n]]></outfiles>\n'
268     xml.write(out_box)
269    
270     xml.write('<BossAttr> crabjob.INTERNAL_ID=_ITR1_ </BossAttr>\n')
271    
272     xml.write('</program>\n')
273     xml.write('</chain>\n')
274    
275     xml.write('</iterator>\n')
276     xml.write('</task>\n')
277    
278     xml.close()
279    
280    
281     return