util/pbs/send.py

   1 #!/usr/bin/env python
   2 # Copyright (c) 2005 The Regents of The University of Michigan
   3 # All rights reserved.
   4 #
   5 # Redistribution and use in source and binary forms, with or without
   6 # modification, are permitted provided that the following conditions are
   7 # met: redistributions of source code must retain the above copyright
   8 # notice, this list of conditions and the following disclaimer;
   9 # redistributions in binary form must reproduce the above copyright
  10 # notice, this list of conditions and the following disclaimer in the
  11 # documentation and/or other materials provided with the distribution;
  12 # neither the name of the copyright holders nor the names of its
  13 # contributors may be used to endorse or promote products derived from
  14 # this software without specific prior written permission.
  15 #
  16 # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  17 # "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  18 # LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  19 # A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  20 # OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  21 # SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  22 # LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  23 # DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  24 # THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  25 # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  26 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  27 #
  28 # Authors: Ali Saidi
  29 #          Nathan Binkert
  30
  31 import os, os.path, re, socket, sys
  32 from os import environ as env, listdir
  33 from os.path import basename, isdir, isfile, islink, join as joinpath, normpath
  34 from filecmp import cmp as filecmp
  35 from shutil import copy
  36
  37 def nfspath(dir):
  38     if dir.startswith('/.automount/'):
  39         dir = '/n/%s' % dir[12:]
  40     elif not dir.startswith('/n/'):
  41         dir = '/n/%s%s' % (socket.gethostname().split('.')[0], dir)
  42     return dir
  43
  44 def syncdir(srcdir, destdir):
  45     srcdir = normpath(srcdir)
  46     destdir = normpath(destdir)
  47     if not isdir(destdir):
  48         sys.exit('destination directory "%s" does not exist' % destdir)
  49
  50     for root, dirs, files in os.walk(srcdir):
  51         root = normpath(root)
  52         prefix = os.path.commonprefix([root, srcdir])
  53         root = root[len(prefix):]
  54         if root.startswith('/'):
  55             root = root[1:]
  56         for rem in [ d for d in dirs if d.startswith('.') or d == 'SCCS']:
  57             dirs.remove(rem)
  58
  59         for entry in dirs:
  60             newdir = joinpath(destdir, root, entry)
  61             if not isdir(newdir):
  62                 os.mkdir(newdir)
  63                 print 'mkdir', newdir
  64
  65         for i,d in enumerate(dirs):
  66             if islink(joinpath(srcdir, root, d)):
  67                 dirs[i] = joinpath(d, '.')
  68
  69         for entry in files:
  70             dest = normpath(joinpath(destdir, root, entry))
  71             src = normpath(joinpath(srcdir, root, entry))
  72             if not isfile(dest) or not filecmp(src, dest):
  73                 print 'copy %s %s' % (dest, src)
  74                 copy(src, dest)
  75
  76 progpath = nfspath(sys.path[0])
  77 progname = basename(sys.argv[0])
  78 usage = """\
  79 Usage:
  80     %(progname)s [-c] [-e] [-f] [-j <jobfile>] [-q queue] [-v] <regexp>
  81     -c           clean directory if job can be run
  82     -e           only echo pbs command info, don't actually send the job
  83     -f           force the job to run regardless of state
  84     -q <queue>   submit job to the named queue
  85     -j <jobfile> specify the jobfile (default is <rootdir>/Test.py)
  86     -v           be verbose
  87
  88     %(progname)s [-j <jobfile>] -l [-v] <regexp>
  89     -j <jobfile> specify the jobfile (default is <rootdir>/Test.py)
  90     -l           list job names, don't submit
  91     -v           be verbose (list job parameters)
  92
  93     %(progname)s -h
  94     -h           display this help
  95 """ % locals()
  96
  97 try:
  98     import getopt
  99     opts, args = getopt.getopt(sys.argv[1:], '-Ccdefhj:lq:Rt:v')
 100 except getopt.GetoptError:
 101     sys.exit(usage)
 102
 103 depend = False
 104 clean = False
 105 onlyecho = False
 106 exprs = []
 107 force = False
 108 listonly = False
 109 queue = ''
 110 verbose = False
 111 jfile = 'Test.py'
 112 docpts = False
 113 doruns = True
 114 runflag = False
 115 node_type = 'FAST'
 116
 117 for opt,arg in opts:
 118     if opt == '-C':
 119         docpts = True
 120     if opt == '-c':
 121         clean = True
 122     if opt == '-d':
 123         depend = True
 124     if opt == '-e':
 125         onlyecho = True
 126     if opt == '-f':
 127         force = True
 128     if opt == '-h':
 129         print usage
 130         sys.exit(0)
 131     if opt == '-j':
 132         jfile = arg
 133     if opt == '-l':
 134         listonly = True
 135     if opt == '-q':
 136         queue = arg
 137     if opt == '-R':
 138         runflag = True
 139     if opt == '-t':
 140         node_type = arg
 141     if opt == '-v':
 142         verbose = True
 143
 144 if docpts:
 145     doruns = runflag
 146
 147 for arg in args:
 148     exprs.append(re.compile(arg))
 149
 150 import jobfile, pbs
 151 from job import JobDir, date
 152
 153 conf = jobfile.JobFile(jfile)
 154
 155 if not listonly and not onlyecho and isdir(conf.linkdir):
 156     if verbose:
 157         print 'Checking for outdated files in Link directory'
 158     if not isdir(conf.basedir):
 159         os.mkdir(conf.basedir)
 160     syncdir(conf.linkdir, conf.basedir)
 161
 162 jobnames = {}
 163 joblist = []
 164
 165 if docpts and doruns:
 166     gen = conf.alljobs()
 167 elif docpts:
 168     gen = conf.checkpoints()
 169 elif doruns:
 170     gen = conf.jobs()
 171
 172 for job in gen:
 173     if job.name in jobnames:
 174         continue
 175
 176     if exprs:
 177         for expr in exprs:
 178             if expr.match(job.name):
 179                 joblist.append(job)
 180                 break
 181     else:
 182         joblist.append(job)
 183
 184 if listonly:
 185     if verbose:
 186         for job in joblist:
 187             job.printinfo()
 188     else:
 189         for job in joblist:
 190             print job.name
 191     sys.exit(0)
 192
 193 if not onlyecho:
 194     newlist = []
 195     for job in joblist:
 196         jobdir = JobDir(joinpath(conf.rootdir, job.name))
 197         if jobdir.exists():
 198             if not force:
 199                 status = jobdir.getstatus()
 200                 if status == 'queued':
 201                     continue
 202
 203                 if status == 'running':
 204                     continue
 205
 206                 if status == 'success':
 207                     continue
 208
 209             if not clean:
 210                 sys.exit('job directory %s not clean!' % jobdir)
 211
 212             jobdir.clean()
 213         newlist.append(job)
 214     joblist = newlist
 215
 216 class NameHack(object):
 217     def __init__(self, host='pbs.pool', port=24465):
 218         self.host = host
 219         self.port = port
 220         self.socket = None
 221
 222     def setname(self, jobid, jobname):
 223         try:
 224             jobid = int(jobid)
 225         except ValueError:
 226             jobid = int(jobid.strip().split('.')[0])
 227
 228         jobname = jobname.strip()
 229         # since pbs can handle jobnames of 15 characters or less,
 230         # don't use the raj hack.
 231         if len(jobname) <= 15:
 232             return
 233
 234         if self.socket is None:
 235             import socket
 236             self.socket = socket.socket()
 237             # Connect to pbs.pool and send the jobid/jobname pair to port
 238             # 24465 (Raj didn't realize that there are only 64k ports and
 239             # setup inetd to point to port 90001)
 240             self.socket.connect((self.host, self.port))
 241
 242         self.socket.send("%s %s\n" % (jobid, jobname))
 243
 244 namehack = NameHack()
 245
 246 for job in joblist:
 247     jobdir = JobDir(joinpath(conf.rootdir, job.name))
 248     if depend:
 249         cptdir = JobDir(joinpath(conf.rootdir, job.checkpoint.name))
 250         cptjob = cptdir.readval('.pbs_jobid')
 251
 252     if not onlyecho:
 253         jobdir.create()
 254
 255     print 'Job name:       %s' % job.name
 256     print 'Job directory:  %s' % jobdir
 257
 258     qsub = pbs.qsub()
 259     qsub.pbshost = 'simpool.eecs.umich.edu'
 260     qsub.stdout = jobdir.file('jobout')
 261     qsub.name = job.name[:15]
 262     qsub.join = True
 263     qsub.node_type = node_type
 264     qsub.env['ROOTDIR'] = conf.rootdir
 265     qsub.env['JOBNAME'] = job.name
 266     if depend:
 267         qsub.afterok = cptjob
 268     if queue:
 269         qsub.queue = queue
 270     qsub.build(joinpath(progpath, 'job.py'))
 271
 272     if verbose:
 273         print 'PBS Command:    %s' % qsub.command
 274
 275     if not onlyecho:
 276         ec = qsub.do()
 277         if ec == 0:
 278             jobid = qsub.result
 279             print 'PBS Jobid:      %s' % jobid
 280             namehack.setname(jobid, job.name)
 281             queued = date()
 282             jobdir.echofile('.pbs_jobid', jobid)
 283             jobdir.echofile('.pbs_jobname', job.name)
 284             jobdir.echofile('.queued', queued)
 285             jobdir.setstatus('queued on %s' % queued)
 286         else:
 287             print 'PBS Failed'