| #Licensed to the Apache Software Foundation (ASF) under one |
| #or more contributor license agreements. See the NOTICE file |
| #distributed with this work for additional information |
| #regarding copyright ownership. The ASF licenses this file |
| #to you under the Apache License, Version 2.0 (the |
| #"License"); you may not use this file except in compliance |
| #with the License. You may obtain a copy of the License at |
| |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| |
| #Unless required by applicable law or agreed to in writing, software |
| #distributed under the License is distributed on an "AS IS" BASIS, |
| #WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| #See the License for the specific language governing permissions and |
| #limitations under the License. |
| """Maui/Torque implementation of NodePool""" |
| # -*- python -*- |
| |
| import os, sys, csv, socket, time, re, pprint |
| |
| from hodlib.Hod.nodePool import * |
| from hodlib.Schedulers.torque import torqueInterface |
| from hodlib.Common.threads import simpleCommand |
| from hodlib.Common.util import get_exception_string, args_to_string, local_fqdn, \ |
| TORQUE_USER_LIMITS_COMMENT_FIELD |
| |
| class TorqueNodeSet(NodeSet): |
| def __init__(self, id, numNodes, preferredList, isPreemptee): |
| NodeSet.__init__(self, id, numNodes, preferredList, isPreemptee) |
| self.qsubId = None |
| self.addrList = [] |
| |
| def _setQsubId(self, qsubId): |
| self.qsubId = qsubId |
| |
| def _setAddrList(self, addrList): |
| self.addrList = addrList |
| |
| def getAddrList(self): |
| return self.addrList |
| |
| class TorquePool(NodePool): |
| def __init__(self, nodePoolDesc, cfg, log): |
| NodePool.__init__(self, nodePoolDesc, cfg, log) |
| |
| environ = os.environ.copy() |
| |
| if self._cfg['resource_manager'].has_key('pbs-server'): |
| environ['PBS_DEFAULT'] = self._cfg['resource_manager']['pbs-server'] |
| |
| self.__torque = torqueInterface( |
| self._cfg['resource_manager']['batch-home'], environ, self._log) |
| |
| def getAccountString(self): |
| account = '' |
| if self._cfg['resource_manager'].has_key('pbs-account'): |
| account = self._cfg['resource_manager']['pbs-account'] |
| return account |
| |
| def __gen_submit_params(self, nodeSet, walltime = None, qosLevel = None, |
| account = None): |
| argList = [] |
| stdinList = [] |
| |
| npd = self.nodePoolDesc |
| |
| def gen_stdin_list(): |
| # Here we are basically generating the standard input for qsub. |
| # Specifically a script to exec ringmaster. |
| stdinList.append('#!/bin/sh') |
| |
| ringBin = os.path.join(self._cfg['hod']['base-dir'], 'bin', |
| 'ringmaster') |
| ringArgs = [ringBin,] |
| ringArgs.extend(self._cfg.get_args(exclude=('hod'))) |
| |
| ringMasterCommand = args_to_string(ringArgs) |
| |
| self._log.debug("ringmaster cmd: %s" % ringMasterCommand) |
| |
| stdinList.append(ringMasterCommand) |
| |
| def gen_arg_list(): |
| def process_qsub_attributes(): |
| rawAttributes = self.nodePoolDesc.getAttrs() |
| |
| # 'W:x' is used to specify torque management extentensions ie -W x= ... |
| resourceManagementExtensions = '' |
| if 'W:x' in rawAttributes: |
| resourceManagementExtensions = rawAttributes['W:x'] |
| |
| if qosLevel: |
| if len(resourceManagementExtensions) > 0: |
| resourceManagementExtensions += ';' |
| resourceManagementExtensions += 'QOS:%s' % (qosLevel) |
| |
| rawAttributes['W:x'] = resourceManagementExtensions |
| |
| hostname = local_fqdn() |
| |
| # key values are expected to have string values. |
| rawAttributes['l:nodes'] = "%s" % nodeSet._getNumNodes() |
| |
| if walltime: |
| rawAttributes['l:walltime'] = "%s" % walltime |
| |
| #create a dict of dictionaries for |
| # various arguments of torque |
| cmds = {} |
| for key in rawAttributes: |
| value = rawAttributes[key] |
| |
| if key.find(':') == -1: |
| raise ValueError, 'Syntax error: missing colon after %s in %s=%s' % ( |
| key, key, value) |
| |
| [option, subOption] = key.split(':', 1) |
| if not option in cmds: |
| cmds[option] = {} |
| cmds[option][subOption] = value |
| |
| opts = [] |
| #create a string from this |
| #dictionary of dictionaries createde above |
| for k in cmds: |
| csv = [] |
| nv = cmds[k] |
| for n in nv: |
| v = nv[n] |
| if len(n) == 0: |
| csv.append(v) |
| else: |
| csv.append('%s=%s' % (n, v)) |
| opts.append('-%s' % (k)) |
| opts.append(','.join(csv)) |
| |
| for option in cmds: |
| commandList = [] |
| for subOption in cmds[option]: |
| value = cmds[option][subOption] |
| if len(subOption) == 0: |
| commandList.append(value) |
| else: |
| commandList.append("%s=%s" % (subOption, value)) |
| opts.append('-%s' % option) |
| opts.append(','.join(commandList)) |
| |
| return opts |
| |
| pkgdir = npd.getPkgDir() |
| |
| qsub = os.path.join(pkgdir, 'bin', 'qsub') |
| sdd = self._cfg['servicedesc'] |
| |
| gsvc = None |
| for key in sdd: |
| gsvc = sdd[key] |
| break |
| |
| argList.extend(process_qsub_attributes()) |
| |
| argList.extend(('-N', '"' + self._cfg['hod']['title'] + '"')) |
| argList.extend(('-r','n')) |
| |
| if 'pbs-user' in self._cfg['resource_manager']: |
| argList.extend(('-u', self._cfg['resource_manager']['pbs-user'])) |
| |
| argList.extend(('-d','/tmp/')) |
| if 'queue' in self._cfg['resource_manager']: |
| queue = self._cfg['resource_manager']['queue'] |
| argList.extend(('-q',queue)) |
| |
| # In HOD 0.4, we pass in an account string only if it is mentioned. |
| # Also, we don't append userid to the account string, as HOD jobs run as the |
| # user running them, not as 'HOD' user. |
| if self._cfg['resource_manager'].has_key('pbs-account'): |
| argList.extend(('-A', (self._cfg['resource_manager']['pbs-account']))) |
| |
| if 'env-vars' in self._cfg['resource_manager']: |
| qsub_envs = self._cfg['resource_manager']['env-vars'] |
| argList.extend(('-v', self.__keyValToString(qsub_envs))) |
| |
| gen_arg_list() |
| gen_stdin_list() |
| |
| return argList, stdinList |
| |
| def __keyValToString(self, keyValList): |
| ret = "" |
| for key in keyValList: |
| ret = "%s%s=%s," % (ret, key, keyValList[key]) |
| return ret[:-1] |
| |
| def newNodeSet(self, numNodes, preferred=[], isPreemptee=True, id=None): |
| if not id: |
| id = self.getNextNodeSetId() |
| |
| nodeSet = TorqueNodeSet(id, numNodes, preferred, isPreemptee) |
| |
| self.nodeSetDict[nodeSet.getId()] = nodeSet |
| |
| return nodeSet |
| |
| def submitNodeSet(self, nodeSet, walltime = None, qosLevel = None, |
| account = None): |
| |
| argList, stdinList = self.__gen_submit_params(nodeSet, walltime, qosLevel, |
| account) |
| |
| jobId, exitCode = self.__torque.qsub(argList, stdinList) |
| |
| ## UNUSED CODE: LINE ## |
| nodeSet.qsubId = jobId |
| |
| return jobId, exitCode |
| |
| def freeNodeSet(self, nodeSet): |
| |
| exitCode = self.deleteJob(nodeSet.getId()) |
| |
| del self.nodeSetDict[nodeSet.getId()] |
| |
| return exitCode |
| |
| def finalize(self): |
| status = 0 |
| exitCode = 0 |
| for nodeSet in self.nodeSetDict.values(): |
| exitCode = self.freeNodeSet(nodeSet) |
| |
| if exitCode > 0 and exitCode != 153: |
| status = 4 |
| |
| return status |
| |
| ## UNUSED METHOD ?? ## |
| def getWorkers(self): |
| hosts = [] |
| |
| qstatInfo = self.__torque(self.getServiceId()) |
| if qstatInfo: |
| hosts = qstatInfop['exec_host'] |
| |
| return hosts |
| |
| ## UNUSED METHOD ?? ## |
| def pollNodeSet(self, nodeSet): |
| status = NodeSet.COMPLETE |
| nodeSet = self.nodeSetDict[0] |
| |
| qstatInfo = self.__torque(self.getServiceId()) |
| |
| if qstatMap: |
| jobstate = qstatMap['job_state'] |
| exechost = qstatMap['exec_host'] |
| |
| if jobstate == 'Q': |
| status = NodeSet.PENDING |
| elif exechost == None: |
| status = NodeSet.COMMITTED |
| else: |
| nodeSet._setAddrList(exec_host) |
| |
| return status |
| |
| def getServiceId(self): |
| id = None |
| |
| nodeSets = self.nodeSetDict.values() |
| if len(nodeSets): |
| id = nodeSets[0].qsubId |
| |
| if id == None: |
| id = os.getenv('PBS_JOBID') |
| |
| return id |
| |
| def getJobInfo(self, jobId=None): |
| |
| jobNonExistentErrorCode = 153 |
| self.__jobInfo = { 'job_state' : False } |
| |
| if jobId == None: |
| jobId = self.getServiceId() |
| |
| qstatInfo, exitCode = self.__torque.qstat(jobId) |
| if exitCode == 0: |
| self.__jobInfo = qstatInfo |
| elif exitCode == jobNonExistentErrorCode: |
| # This really means that the job completed |
| # However, setting only job_state for now, not |
| # any other attributes, as none seem required. |
| self.__jobInfo = { 'job_state' : 'C' } |
| |
| return self.__jobInfo |
| |
| def deleteJob(self, jobId): |
| exitCode = self.__torque.qdel(jobId) |
| return exitCode |
| |
| def isJobFeasible(self): |
| comment = None |
| msg = None |
| if self.__jobInfo.has_key('comment'): |
| comment = self.__jobInfo['comment'] |
| try: |
| if comment: |
| commentField = re.compile(self._cfg['hod']['job-feasibility-attr']) |
| match = commentField.search(comment) |
| if match: |
| reqUsage = int(match.group(1)) |
| currentUsage = int(match.group(2)) |
| maxUsage = int(match.group(3)) |
| msg = "Current Usage:%s, Requested:%s, Maximum Limit:%s " % \ |
| (currentUsage, reqUsage, maxUsage) |
| if reqUsage > maxUsage: |
| return "Never", msg |
| if reqUsage + currentUsage > maxUsage: |
| return False, msg |
| except Exception, e: |
| self._log.error("Error in isJobFeasible : %s" %e) |
| raise Exception(e) |
| return True, msg |
| |
| def runWorkers(self, args): |
| return self.__torque.pbsdsh(args) |
| |
| def updateWorkerInfo(self, workerInfoMap, jobId): |
| workerInfoStr = '' |
| for key in workerInfoMap.keys(): |
| workerInfoStr = '%s,%s:%s' % (workerInfoStr, key, workerInfoMap[key]) |
| exitCode = self.__torque.qalter("notes", workerInfoStr[1:], jobId) |
| return exitCode |
| |
