util/batch/send.py

   1 #!/usr/bin/env python
   2 # Copyright (c) 2006 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: Kevin Lim
  29
  30 import os, os.path, re, socket, sys
  31 from os import environ as env, listdir
  32 from os.path import basename, isdir, isfile, islink, join as joinpath, normpath
  33 from filecmp import cmp as filecmp
  34 from shutil import copy
  35
  36 def nfspath(dir):
  37     if dir.startswith('/.automount/'):
  38         dir = '/n/%s' % dir[12:]
  39     elif not dir.startswith('/n/'):
  40         dir = '/n/%s%s' % (socket.gethostname().split('.')[0], dir)
  41     return dir
  42
  43 def syncdir(srcdir, destdir):
  44     srcdir = normpath(srcdir)
  45     destdir = normpath(destdir)
  46     if not isdir(destdir):
  47         sys.exit('destination directory "%s" does not exist' % destdir)
  48
  49     for root, dirs, files in os.walk(srcdir):
  50         root = normpath(root)
  51         prefix = os.path.commonprefix([root, srcdir])
  52         root = root[len(prefix):]
  53         if root.startswith('/'):
  54             root = root[1:]
  55         for rem in [ d for d in dirs if d.startswith('.') or d == 'SCCS']:
  56             dirs.remove(rem)
  57
  58         for entry in dirs:
  59             newdir = joinpath(destdir, root, entry)
  60             if not isdir(newdir):
  61                 os.mkdir(newdir)
  62                 print 'mkdir', newdir
  63
  64         for i,d in enumerate(dirs):
  65             if islink(joinpath(srcdir, root, d)):
  66                 dirs[i] = joinpath(d, '.')
  67
  68         for entry in files:
  69             dest = normpath(joinpath(destdir, root, entry))
  70             src = normpath(joinpath(srcdir, root, entry))
  71             if not isfile(dest) or not filecmp(src, dest):
  72                 print 'copy %s %s' % (dest, src)
  73                 copy(src, dest)
  74
  75 progpath = nfspath(sys.path[0])
  76 progname = basename(sys.argv[0])
  77 usage = """\
  78 Usage:
  79     %(progname)s [-c] [-e] [-f] [-j <jobfile>] [-q queue] [-v] <regexp>
  80     -c           clean directory if job can be run
  81     -C           submit the checkpointing runs
  82     -d           Make jobs be dependent on the completion of the checkpoint runs
  83     -e           only echo pbs command info, don't actually send the job
  84     -f           force the job to run regardless of state
  85     -q <queue>   submit job to the named queue
  86     -j <jobfile> specify the jobfile (default is <rootdir>/Test.py)
  87     -v           be verbose
  88
  89     %(progname)s [-j <jobfile>] -l [-v] <regexp>
  90     -j <jobfile> specify the jobfile (default is <rootdir>/Test.py)
  91     -l           list job names, don't submit
  92     -v           be verbose (list job parameters)
  93
  94     %(progname)s -h
  95     -h           display this help
  96 """ % locals()
  97
  98 try:
  99     import getopt
 100     opts, args = getopt.getopt(sys.argv[1:], '-Ccdefhj:lnq:Rt:v')
 101 except getopt.GetoptError:
 102     sys.exit(usage)
 103
 104 depend = False
 105 clean = False
 106 onlyecho = False
 107 exprs = []
 108 force = False
 109 listonly = False
 110 queue = ''
 111 verbose = False
 112 jfile = 'Test.py'
 113 docpts = False
 114 doruns = True
 115 runflag = False
 116 node_type = 'FAST'
 117 update = True
 118
 119 for opt,arg in opts:
 120     if opt == '-C':
 121         docpts = True
 122     if opt == '-c':
 123         clean = True
 124     if opt == '-d':
 125         depend = True
 126     if opt == '-e':
 127         onlyecho = True
 128     if opt == '-f':
 129         force = True
 130     if opt == '-h':
 131         print usage
 132         sys.exit(0)
 133     if opt == '-j':
 134         jfile = arg
 135     if opt == '-l':
 136         listonly = True
 137     if opt == '-n':
 138         update = False
 139     if opt == '-q':
 140         queue = arg
 141     if opt == '-R':
 142         runflag = True
 143     if opt == '-t':
 144         node_type = arg
 145     if opt == '-v':
 146         verbose = True
 147
 148 if docpts:
 149     doruns = runflag
 150
 151 for arg in args:
 152     exprs.append(re.compile(arg))
 153
 154 import jobfile, batch
 155 from job import JobDir, date
 156
 157 conf = jobfile.JobFile(jfile)
 158
 159 if update and not listonly and not onlyecho and isdir(conf.linkdir):
 160     if verbose:
 161         print 'Checking for outdated files in Link directory'
 162     if not isdir(conf.basedir):
 163         os.mkdir(conf.basedir)
 164     syncdir(conf.linkdir, conf.basedir)
 165
 166 jobnames = {}
 167 joblist = []
 168
 169 if docpts and doruns:
 170     gen = conf.alljobs()
 171 elif docpts:
 172     gen = conf.checkpoints()
 173 elif doruns:
 174     gen = conf.jobs()
 175
 176 for job in gen:
 177     if job.name in jobnames:
 178         continue
 179
 180     if exprs:
 181         for expr in exprs:
 182             if expr.match(job.name):
 183                 joblist.append(job)
 184                 break
 185     else:
 186         joblist.append(job)
 187
 188 if listonly:
 189     if verbose:
 190         for job in joblist:
 191             job.printinfo()
 192     else:
 193         for job in joblist:
 194             print job.name
 195     sys.exit(0)
 196
 197 if not onlyecho:
 198     newlist = []
 199     for job in joblist:
 200         jobdir = JobDir(joinpath(conf.rootdir, job.name))
 201         if jobdir.exists():
 202             if not force:
 203                 status = jobdir.getstatus()
 204                 if status == 'queued':
 205                     continue
 206
 207                 if status == 'running':
 208                     continue
 209
 210                 if status == 'success':
 211                     continue
 212
 213             if not clean:
 214                 sys.exit('job directory %s not clean!' % jobdir)
 215
 216             jobdir.clean()
 217         newlist.append(job)
 218     joblist = newlist
 219
 220 class NameHack(object):
 221     def __init__(self, host='pbs.pool', port=24465):
 222         self.host = host
 223         self.port = port
 224         self.socket = None
 225
 226     def setname(self, jobid, jobname):
 227         try:
 228             jobid = int(jobid)
 229         except ValueError:
 230             jobid = int(jobid.strip().split('.')[0])
 231
 232         jobname = jobname.strip()
 233         # since pbs can handle jobnames of 15 characters or less,
 234         # don't use the raj hack.
 235         if len(jobname) <= 15:
 236             return
 237
 238         if self.socket is None:
 239             import socket
 240             self.socket = socket.socket()
 241             # Connect to pbs.pool and send the jobid/jobname pair to port
 242             # 24465 (Raj didn't realize that there are only 64k ports and
 243             # setup inetd to point to port 90001)
 244             self.socket.connect((self.host, self.port))
 245
 246         self.socket.send("%s %s\n" % (jobid, jobname))
 247
 248 namehack = NameHack()
 249
 250 rootdir = conf.rootdir
 251 script = joinpath(rootdir, 'Base', 'job.py')
 252
 253 for job in joblist:
 254     jobdir = JobDir(joinpath(rootdir, job.name))
 255     if depend:
 256         cptdir = JobDir(joinpath(rootdir, job.checkpoint.name))
 257         path = str(cptdir)
 258         if not isdir(path) or not isfile(joinpath(path, '.success')):
 259             continue
 260
 261         cptjob = cptdir.readval('.batch_jobid')
 262
 263     if not onlyecho:
 264         jobdir.create()
 265         os.chdir(str(jobdir))
 266         os.environ['PWD'] = str(jobdir)
 267
 268     print 'Job name:       %s' % job.name
 269     print 'Job directory:  %s' % jobdir
 270
 271
 272     qsub = batch.oarsub()
 273     qsub.oarhost = 'poolfs.eecs.umich.edu'
 274     #qsub.stdout = jobdir.file('jobout')
 275     qsub.name = job.name
 276     qsub.walltime = '50'
 277     #qsub.join = True
 278     #qsub.node_type = node_type
 279     #qsub.env['ROOTDIR'] = conf.rootdir
 280     #qsub.env['JOBNAME'] = job.name
 281     #if depend:
 282     #    qsub.afterok = cptjob
 283     #if queue:
 284     #    qsub.queue = queue
 285     qsub.properties = "64bit = 'Yes' or 64bit = 'No'"
 286     qsub.build(script)
 287
 288     if verbose:
 289         print 'cwd:    %s' % qsub.command
 290         print 'PBS Command:    %s' % qsub.command
 291
 292     if not onlyecho:
 293         ec = qsub.do()
 294         if ec == 0:
 295             jobid = qsub.result
 296             print 'OAR Jobid:      %s' % jobid
 297             #namehack.setname(jobid, job.name)
 298             queued = date()
 299             jobdir.echofile('.batch_jobid', jobid)
 300             jobdir.echofile('.batch_jobname', job.name)
 301             jobdir.echofile('.queued', queued)
 302             jobdir.setstatus('queued on %s' % queued)
 303         else:
 304             print 'OAR Failed'
 305     print
 306     print