--- /dev/null
+#!/usr/bin/env python
+
+# DHARMA Project
+# Copyright (C) 2003-2004 Yun Mao, University of Pennsylvania
+#
+# This library is free software; you can redistribute it and/or
+# modify it under the terms of version 2.1 of the GNU Lesser General Public
+# License as published by the Free Software Foundation.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public
+# License along with this program; if not, write to the Free Software
+# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
+
+
+"""
+vxargs: Visualized xargs with redirected output
+
+"""
+version = "0.3.1"
+import os, sys, time, signal, errno
+import curses, random
+import getopt
+
+update_rate = 1
+
+final_stats = {}
+gsl = None
+stopping = 0
+
+def getListFromFile(f):
+ """I'll ignore the line starting with #
+
+ @param f: file object of the host list file
+ @return: a list of hostnames (or IPs)
+ """
+ hostlist = []
+ for line in f:
+ if line[0]!='#':
+ if line.strip():
+ hostlist.append([line.strip(),''])
+ elif hostlist and hostlist[-1][1]=='':
+ hostlist[-1][1] = line.strip()[1:]
+ return hostlist
+
+def get_last_line(fn):
+ #equ to tail -n1 fn
+ try:
+ lines = open(fn,'r').readlines()
+ if len(lines)>0:
+ return (0, lines[-1].strip())
+ except IOError:
+ pass
+ return (1,'')
+
+class Slot:
+ def __init__(self, outdir, num, screen, timeout, name, count):
+ self.outdir = outdir
+ self.slotnum = num
+ self.screen = screen
+ self.comment = ""
+ self.startTime = time.time()
+ self.timeout = timeout
+ self.name = name
+ self.count = count
+
+ def drawLine(self, comment='', done = False):
+ if self.screen is None: return
+ if comment == '':
+ comment = self.comment
+ else:
+ self.comment = comment
+ stdscr = self.screen
+ elapsed = time.time()-self.startTime
+ try:
+ y,x = stdscr.getmaxyx()
+ spaces = ' '*x
+ stdscr.addstr(self.slotnum+2, 0, spaces) #title occupies two lines
+ if done:
+ stdscr.addstr(self.slotnum+2,0, comment[:x])
+ else:
+ #construct the string
+ output = "(%3ds)%3d: %s " % ( round(elapsed), self.count, self.name )
+ spaceleft = x - len(output)
+ if self.outdir and spaceleft>1:
+ outfn = os.path.join(self.outdir, '%s.out' % self.name)
+ errfn = os.path.join(self.outdir, '%s.err' % self.name)
+ lout = get_last_line(outfn)
+ lerr = get_last_line(errfn)
+ if lerr[0]==0 and lerr[1]:
+ output += lerr[1]
+ elif lout[0]==0 and lout[1]:
+ output += lout[1]
+ else:
+ output += comment
+ else:
+ output += comment
+ stdscr.addstr(self.slotnum+2, 0, output[:x] )
+ stdscr.refresh()
+ except curses.error: #some of them will be out of screen, ignore it
+ pass
+ def update(self, pid):
+ self.drawLine()
+ if self.timeout >0:
+ self.kill(pid)
+
+ def kill(self, pid):
+ overtime = time.time()-self.startTime - self.timeout
+ try:
+ if overtime > 3: #expired more than 3 seconds, send -9
+ os.kill(-pid, signal.SIGKILL)
+ elif overtime > 2: #expired more than 2 seconds, send -15
+ os.kill(-pid, signal.SIGTERM)
+ elif overtime >= 0:
+ os.kill(-pid, signal.SIGINT)
+ except OSError, e:
+ if e.errno != errno.ESRCH: # No such process
+ raise e
+
+ def stop(self, pid):
+ """stop current pid b/c we caught SIGINT twice
+ """
+ self.startTime = time.time() - self.timeout
+ self.kill(pid)
+
+class Slots:
+ pids = {}
+ def __init__(self, max, screen, timeout, outdir):
+ self.maxChild = max
+ self.slots = range(self.maxChild)
+ self.screen = screen
+ self.t = timeout
+ self.outdir = outdir
+
+ def getSlot(self, name, count):
+ if not self.slots:
+ #it's empty, wait until other jobs finish
+ slot = self.waitJobs().slotnum
+ else:
+ slot = self.slots[0]
+ self.slots.remove(slot)
+ return Slot(self.outdir, slot, self.screen, self.t, name, count)
+
+ def mapPID(self, pid, slot):
+ """@param slot: slot object
+ """
+ self.pids[pid] = slot
+
+ def waitJobs(self):
+ while 1:
+ try:
+ pid, status = os.wait()
+ break
+ except OSError, e:
+ if e.errno == errno.ECHILD: #no child processes
+ raise RuntimeError('no child processes when waiting')
+ slot = self.pids[pid]
+ if self.outdir:
+ open(os.path.join(self.outdir, '%s.status' % slot.name),'w').write('%d' % (status>>8))
+ if (status & 0xFF) !=0:
+ open(os.path.join(self.outdir, 'killed_list'),'a').write('%s\n' % (slot.name))
+ if status >>8:
+ open(os.path.join(self.outdir, 'abnormal_list'),'a').write('%s\n' % (slot.name))
+ del self.pids[pid]
+ s = status >> 8
+ if final_stats.has_key(s):
+ final_stats[s]+= 1
+ else:
+ final_stats[s]=1
+ return slot
+ def update(self):
+ for k,v in self.pids.items():
+ v.update(k)
+ def timeout(self):
+ self.update()
+ signal.alarm(update_rate)
+
+ def drawTitle(self, stuff):
+ if self.screen:
+ y,x = self.screen.getmaxyx()
+ spaces = ' '*(x*2)
+ self.screen.addstr(0,0, spaces)
+ self.screen.addstr(0,0, stuff[:x*2])
+ self.screen.refresh()
+ else:
+ print stuff
+ def stop(self):
+ if stopping ==1:
+ msg = 'Stopping -- Waiting current jobs done. Press Ctrl-C again to kill current jobs.'
+ else:
+ msg = 'Stopping -- Killing current jobs'
+ self.drawTitle(msg)
+ if stopping >1:
+ for k,v in self.pids.items():
+ v.stop(k)
+ return
+
+def handler(signum, frame_unused):
+ global gsl
+ if signum==signal.SIGALRM:
+ gsl.timeout()
+ if signum==signal.SIGINT:
+ global stopping
+ stopping += 1
+ gsl.stop()
+
+def generateCommands(cmd_line, args):
+ return [per_arg.replace('{}', args[0]) for per_arg in cmd_line]
+
+def spawn(cmdline, outfn, errfn, setpgrp = False):
+ """A cleverer spawn that lets you redirect stdout and stderr to
+ outfn and errfn. Returns pid of child.
+ You can't do this with os.spawn, sadly.
+ """
+ pid = os.fork()
+ if pid==0: #child
+ out = open(outfn, 'w')
+ os.dup2(out.fileno() ,sys.stdout.fileno())
+ err = open(errfn, 'w')
+ os.dup2(err.fileno(), sys.stderr.fileno())
+ if setpgrp:
+ os.setpgrp()
+ try:
+ os.execvp(cmdline[0], cmdline)
+ except OSError,e:
+ print >> sys.stderr, "error before execution:",e
+ sys.exit(255)
+ #father process
+ return pid
+
+def start(win, max_child, hlist, outdir, randomize, command_line, timeout):
+
+ total = len(hlist)
+
+ if randomize:
+ random.shuffle(hlist)
+
+ signal.signal(signal.SIGALRM, handler)
+ signal.signal(signal.SIGINT, handler)
+ signal.alarm(update_rate)
+
+ sl = Slots(max_child, win, timeout, outdir)
+ global gsl
+ global stopping
+ gsl = sl
+ count = 0
+ for i in hlist:
+ slot = sl.getSlot(i[0], count)
+ if stopping>0:
+ slot.drawLine('Done', done=True)
+ break
+
+ count += 1
+ slot.drawLine(i[1])
+ x = generateCommands(command_line, i)
+
+ sl.drawTitle("%d/%d:%s" %(count, total,' '.join(x)))
+
+ outpath = '/dev/null'
+ errpath = '/dev/null'
+ if outdir:
+ outpath = os.path.join(outdir, '%s.out'%i[0])
+ errpath = os.path.join(outdir, '%s.err'%i[0])
+
+ pid = spawn(x, outpath, errpath, setpgrp = True)
+ sl.mapPID(pid, slot)
+
+ while sl.pids:
+ try:
+ slot = sl.waitJobs()
+ except RuntimeError:
+ print >> sys.stderr, 'Warning: lost tracking of %d jobs' % len(sl.pids)
+ return
+ slot.drawLine('Done', done = True) #Done
+
+def get_output(outdir, argument_list, out= True, err=False, status=False):
+ """
+
+ For post processing the output dir.
+
+ @param out: decide whether to process *.out files
+ @param err: decide whether to process *.err files
+ @param status: decide whether to process *.status files
+
+ @return: (out, err, status): out is a hash table, in which the
+ keys are the arguments, and the values are the string of the
+ output, if available. err is similar. the values of hash table
+ status is the value of exit status in int.
+
+ """
+ if not out and not err and not status:
+ raise RuntimeError("one of out, err and status has to be True")
+
+ result = ({},{},{})
+ mapping = ('out','err','status')
+ p = []
+ if out: p.append(0)
+ if err: p.append(1)
+ if status: p.append(2)
+ for arg in argument_list:
+ basefn = os.path.join(outdir, arg)
+ for i in p:
+ fn = '.'.join( (basefn, mapping[i]) ) #basefn.ext
+ try:
+ lines = open(fn).readlines()
+ result[i][arg]=''.join(lines)
+ except IOError:
+ pass
+ if not status: return result
+ int_status = {}
+ for k,v in result[2].items():
+ try:
+ int_status[k] = int(v.strip())
+ except ValueError:
+ pass
+ return result[0], result[1], int_status
+
+def main():
+ options = 'hP:ra:o:yt:pn'
+ long_opts = ['help','max-procs=','randomize','args=','output=','noprompt','timeout=','plain', 'version','no-exec']
+ try:
+ opts,args = getopt.getopt(sys.argv[1:], options,long_opts)
+ except getopt.GetoptError:
+ print "Unknown options"
+ usage()
+ sys.exit(1)
+ #set default values
+ ask_prompt = True
+ maxchild = 30
+ randomize = False
+ hostfile = sys.stdin
+ outdir = ''
+ timeout = 0
+ plain = False
+ no_exec = False
+ if os.environ.has_key('VXARGS_OUTDIR'):
+ outdir = os.environ['VXARGS_OUTDIR']
+ for o,a in opts:
+ if o in ['--version']:
+ print "vxargs version",version
+ print "Copyright (c) 2004 Yun Mao (maoy@cis.upenn.edu)"
+ print "Freely distributed under GNU LGPL License"
+ sys.exit(1)
+ elif o in ['-h','--help']:
+ usage()
+ sys.exit(1)
+ elif o in ['-r','--randomize']:
+ randomize = True
+ elif o in ['-P','--max-procs']:
+ maxchild = int(a)
+ elif o in ['-a','--args']:
+ try:
+ hostfile = open(a,'r')
+ except IOError, e:
+ print "argument file %s has error: %s" % ( a, str(e) )
+ sys.exit(3)
+ elif o in ['-o','--output']:
+ outdir = a
+ if a =='/dev/null': outdir = ''
+ elif o in ['-y','--noprompt']:
+ ask_prompt = False
+ elif o in ['-t','--timeout']:
+ timeout = int(a)
+ elif o in ['-p','--plain']:
+ plain = True
+ elif o in ['-n','--no-exec']:
+ no_exec = True
+ else:
+ print 'Unknown options'
+ usage()
+ sys.exit(1)
+ if len(args)<1:
+ print "No command given."
+ usage()
+ sys.exit(1)
+ #now test outdir
+ if outdir:
+ if os.path.exists(outdir):
+ if not os.path.isdir(outdir):
+ print "%s exists and is not a dir, won't continue" % outdir
+ sys.exit(3)
+ elif no_exec:
+ print "%s is the destination dir and would be destroyed." % (outdir)
+ elif ask_prompt:
+ if hostfile == sys.stdin:
+ print "You must specify --noprompt (-y) option if no --args (-a) or --no-exec (-n) is given. Doing so will destroy folder %s." % (outdir)
+ sys.exit(3)
+ else:
+ result = raw_input("%s exists. Continue will destroy everything in it. Are you sure? (y/n) " % (outdir))
+ if result not in ['y','Y']:
+ sys.exit(3)
+ os.system('rm -f %s' % (os.path.join(outdir,'*')))
+ else:
+ if not no_exec:
+ os.system('mkdir -p %s' % outdir)
+
+ hlist = getListFromFile(hostfile)
+ if no_exec:
+ for i in hlist:
+ real_cmdline = generateCommands(args, i)
+ print ' '.join(real_cmdline)
+ sys.exit(0)
+
+ if plain: # no fancy output
+ return start(None, maxchild, hlist, outdir, randomize, args, timeout)
+ else:
+ # use fancy curses-based animation
+ try:
+ curses.wrapper(start, maxchild, hlist, outdir, randomize, args, timeout)
+ except curses.error:
+ sys.exit(4)
+ #post execution, output some stats
+ total = 0
+ for k,v in final_stats.items():
+ print "exit code %d: %d job(s)" % (k,v)
+ total += v
+ print "total number of jobs:", total
+def usage():
+ print """\
+NAME
+
+ vxargs - build and execute command lines from an argument list file
+ with visualization and parallelism, and output redirection.
+
+DESCRIPTION
+
+ vxargs reads a list of arguments from a txt file or standard input,
+ delimited by newlines, and executes the command one or more times
+ with initial arguments in which {} is substituted by the arguments
+ read from the file or standard input. The current executing commands
+ and progress will be dynamically updated on the screen. Stdout and
+ stderr of each command will be redirected to separate files. A list
+ of all processes with a non-zero exit status is generated in file
+ abnormal_list. A list of all timeout processes is generated in file
+ killed_list.
+
+SYNOPSIS
+
+ vxargs [OPTIONS] command [initial-arguments]
+
+OPTIONS
+
+ --help
+ Print a summary of the options to vxargs and exit.
+
+ --max-procs=max-procs, -P max-procs
+ Run up to max-procs processes at a time; the default is 30.
+
+ --randomize, -r [OPTIONAL]
+ Randomize the host list before all execution.
+
+ --args=filename, -a filename
+ The arguments file. If unspecified, the arguments will be read
+ from standard input, and -y option must be specified.
+
+ --output=outdir, -o outdir
+ output directory for stdout and stderr files
+ The default value is specified by the environment variable VXARGS_OUTDIR.
+ If it is unspecified, both stdout and stderr will be redirected
+ to /dev/null.
+ Note that if the directory existed before execution, everything
+ inside will be wiped.
+
+ --timeout=timeout, -t timeout
+ The maximal time in second for each command to execute. timeout=0
+ means infinite. 0 (i.e. infinite) is the default value. When the time is up,
+ vxargs will send signal SIGINT to the process. If the process does not
+ stop after 2 seconds, vxargs will send SIGTERM signal, and send SIGKILL
+ if it still keeps running after 3 seconds.
+
+ --noprompt, -y
+ Wipe out the outdir without confirmation.
+
+ --no-exec, -n
+ Print the commands that would be executed, but do not execute them.
+
+ --plain, -p
+ Don't use curses-based output, but plain output to stdout
+ instead. It will be less exciting, but will do the same job
+ effectively. It is useful if one wants to start vxargs from cron
+ or by another program that doesn't want to see the output.
+ By default, vxargs uses the curses-based output.
+
+ --version
+ Display current version and copyright information.
+
+EXAMPLES:
+ Suppose the iplist.txt file has following content:
+$ cat iplist.txt
+216.165.109.79
+#planetx.scs.cs.nyu.edu
+158.130.6.254
+#planetlab1.cis.upenn.edu
+158.130.6.253
+#planetlab2.cis.upenn.edu
+128.232.103.203
+#planetlab3.xeno.cl.cam.ac.uk
+
+Note that lines starting with '#' will be interpreted as comment for
+the previous lines, which is optional, for visualization purpose only.
+
+$ vxargs -a iplist.txt -o /tmp/result -P 10 ssh upenn_dharma@{} "hostname;uptime"
+
+...[ UI output]...
+
+$ cat /tmp/result/*
+planetlab3.xeno.cl.cam.ac.uk
+ 03:13:21 up 4 days, 14:36, 0 users, load average: 0.36, 0.44, 0.44
+planetlab2.cis.upenn.edu
+ 03:13:20 up 26 days, 16:19, 0 users, load average: 8.11, 7.41, 7.41
+planetlab1.cis.upenn.edu
+ 03:13:19 up 22 days, 20:02, 0 users, load average: 13.60, 12.55, 12.59
+ssh: connect to host 216.165.109.79 port 22: Connection timed out
+$
+
+other examples:
+cat iplist.txt | vxargs -o /tmp/result rsync -az -e ssh --delete mirror $SLICE@{}:
+
+vxargs -a iplist.txt -o /tmp/result ssh {} killall -9 java
+
+For more information, please visit http://dharma.cis.upenn.edu/planetlab/vxargs/
+"""
+if __name__=='__main__':
+ main()
+
git://git.onelab.eu / monitor.git / blobdiff