1 |
mcinquil |
1.1 |
from Scheduler import Scheduler
|
2 |
|
|
from SchedulerLocal import SchedulerLocal
|
3 |
|
|
from crab_exceptions import *
|
4 |
|
|
from crab_util import *
|
5 |
|
|
import common
|
6 |
|
|
|
7 |
|
|
import os,string
|
8 |
|
|
|
9 |
|
|
# PBS/torque interface for CRAB (dave.newbold@cern.ch, June 09)
|
10 |
|
|
#
|
11 |
|
|
#
|
12 |
|
|
# In the [CRAB] section of crab.cfg, use:
|
13 |
|
|
#
|
14 |
|
|
# scheduler = pbs
|
15 |
|
|
#
|
16 |
|
|
# In the [PBS] section of crab.cfg, use the following optional parameters
|
17 |
|
|
#
|
18 |
|
|
# queue= pbs_queue_to_use [default, use the default queue in your local PBS config]
|
19 |
|
|
# resources = resource_1=value,resource_2=value, etc [like qsub -l syntax]
|
20 |
|
|
#
|
21 |
|
|
# NB: - the scheduler uses a wrapper script to create a local dir (see BossLite scheduler module)
|
22 |
|
|
# Both wrapper stdout/stderr and job script output files are placed in your crab_*/res directory by default
|
23 |
|
|
#
|
24 |
|
|
|
25 |
|
|
#
|
26 |
|
|
# Naming convention:
|
27 |
|
|
# methods starting with 'ws' are responsible to provide
|
28 |
|
|
# corresponding part of the job script ('ws' stands for 'write script').
|
29 |
|
|
#
|
30 |
|
|
|
31 |
|
|
class SchedulerPbs(SchedulerLocal) :
|
32 |
|
|
|
33 |
|
|
def __init__(self):
|
34 |
|
|
Scheduler.__init__(self,"PBS")
|
35 |
|
|
|
36 |
|
|
def configure(self, cfg_params):
|
37 |
|
|
SchedulerLocal.configure(self, cfg_params)
|
38 |
|
|
if "PBS.queue" in cfg_params:
|
39 |
|
|
if len(cfg_params["PBS.queue"]) == 0 or cfg_params["PBS.queue"] == "default":
|
40 |
|
|
common.logger.info(" The default queue of local PBS configuration will be used")
|
41 |
|
|
else:
|
42 |
|
|
common.logger.info(" The default queue of local PBS configuration will be used")
|
43 |
|
|
|
44 |
|
|
#self.return_data = cfg_params.get('USER.return_data', 0)
|
45 |
|
|
#self.copy_data = cfg_params.get("USER.copy_data", 0)
|
46 |
|
|
|
47 |
|
|
#if ( int(self.return_data) == 0 and int(self.copy_data) == 0 ):
|
48 |
|
|
# msg = 'Error: return_data and copy_data cannot be set both to 0\n'
|
49 |
|
|
# msg = msg + 'Please modify your crab.cfg file\n'
|
50 |
|
|
# raise CrabException(msg)
|
51 |
|
|
|
52 |
|
|
#if ( int(self.return_data) == 1 and int(self.copy_data) == 1 ):
|
53 |
|
|
# msg = 'Error: return_data and copy_data cannot be set both to 1\n'
|
54 |
|
|
# msg = msg + 'Please modify your crab.cfg file\n'
|
55 |
|
|
# raise CrabException(msg)
|
56 |
|
|
|
57 |
|
|
#if ( int(self.copy_data) == 0 and int(self.publish_data) == 1 ):
|
58 |
|
|
# msg = 'Warning: publish_data = 1 must be used with copy_data = 1\n'
|
59 |
|
|
# msg = msg + 'Please modify copy_data value in your crab.cfg file\n'
|
60 |
|
|
# common.logger.info(msg)
|
61 |
|
|
# raise CrabException(msg)
|
62 |
|
|
|
63 |
|
|
#if int(self.copy_data) == 1:
|
64 |
|
|
# self.SE = cfg_params.get('USER.storage_element', None)
|
65 |
|
|
# if not self.SE:
|
66 |
|
|
# msg = "Error. The [USER] section has no 'storage_element'"
|
67 |
|
|
# common.logger.info(msg)
|
68 |
|
|
# raise CrabException(msg)
|
69 |
|
|
|
70 |
|
|
# self.proxyValid = 0
|
71 |
|
|
# self.dontCheckProxy = int(cfg_params.get("GRID.dont_check_proxy",0))
|
72 |
|
|
# self.proxyServer = cfg_params.get("GRID.proxy_server",'myproxy.cern.ch')
|
73 |
|
|
# common.logger.debug('Setting myproxy server to ' + self.proxyServer)
|
74 |
|
|
|
75 |
|
|
# self.group = cfg_params.get("GRID.group", None)
|
76 |
|
|
# self.role = cfg_params.get("GRID.role", None)
|
77 |
|
|
# self.VO = cfg_params.get('GRID.virtual_organization', 'cms')
|
78 |
|
|
|
79 |
|
|
# self.checkProxy()
|
80 |
|
|
|
81 |
|
|
return
|
82 |
|
|
|
83 |
|
|
|
84 |
|
|
def envUniqueID(self):
|
85 |
|
|
id = "https://"+common.scheduler.name()+":/${PBS_JOBID}-"+ \
|
86 |
|
|
string.replace(common._db.queryTask('name'),"_","-")
|
87 |
|
|
return id
|
88 |
|
|
|
89 |
|
|
def realSchedParams(self,cfg_params):
|
90 |
|
|
"""
|
91 |
|
|
Return dictionary with specific parameters, to use
|
92 |
|
|
with real scheduler
|
93 |
|
|
"""
|
94 |
|
|
|
95 |
|
|
params={'jobScriptDir':common.work_space.jobDir(),
|
96 |
|
|
'jobResDir':common.work_space.resDir()
|
97 |
|
|
}
|
98 |
|
|
# 'use_proxy': 0}
|
99 |
|
|
|
100 |
|
|
for s in ('resources', 'queue'):
|
101 |
|
|
params.update({s:cfg_params.get(self.name().upper()+'.'+s,'')})
|
102 |
|
|
|
103 |
|
|
#if 'PBS.use_proxy' in cfg_params:
|
104 |
|
|
# if cfg_params['PBS.use_proxy'] == "1" or cfg_params['PBS.use_proxy'] == "0":
|
105 |
|
|
# params['use_proxy'] = int(cfg_params['PBS.use_proxy'])
|
106 |
|
|
# if params['use_proxy'] == 1:
|
107 |
|
|
# import os
|
108 |
|
|
# params['user_proxy'] = os.path.join(params['jobScriptDir'],'pbs_proxy')
|
109 |
|
|
|
110 |
|
|
return params
|
111 |
|
|
|
112 |
|
|
def listMatch(self, dest, full):
|
113 |
|
|
return [str(getLocalDomain(self))]
|
114 |
|
|
|
115 |
|
|
def wsCopyOutput(self):
|
116 |
|
|
return self.wsCopyOutput_comm()
|
117 |
|
|
|
118 |
|
|
def wsExitFunc(self):
|
119 |
|
|
"""
|
120 |
|
|
"""
|
121 |
|
|
s=[]
|
122 |
|
|
s.append('func_exit(){')
|
123 |
|
|
s.append(self.wsExitFunc_common())
|
124 |
|
|
s.append('tar zcvf '+common.work_space.resDir()+'${out_files}.tgz ${filesToCheck}')
|
125 |
|
|
s.append('exit $job_exit_code')
|
126 |
|
|
s.append('}')
|
127 |
|
|
return '\n'.join(s)
|
128 |
|
|
|
129 |
|
|
def envUniqueID(self):
|
130 |
|
|
id = "https://"+common.scheduler.name()+":/${PBS_JOBID}-"+ \
|
131 |
|
|
string.replace(common._db.queryTask('name'),"_","-")
|
132 |
|
|
return id
|
133 |
|
|
|