torque.py
上传用户:quxuerui
上传日期:2018-01-08
资源大小:41811k
文件大小:10k
- #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