blob: 4f87d4ef4b95e35c51ba29d2e9c74a818679d881 [file] [log] [blame]
mblighb0fab822007-07-25 16:40:19 +00001__author__ = """Copyright Andy Whitcroft, Martin J. Bligh - 2006, 2007"""
2
mbligh42ff92f2007-11-24 19:13:15 +00003import sys, os, subprocess, traceback
mblighb0fab822007-07-25 16:40:19 +00004
5
6def parallel(tasklist):
7 """Run an set of predefined subcommands in parallel"""
8 pids = []
9 error = False
10 for task in tasklist:
11 task.fork_start()
12 for task in tasklist:
13 status = task.fork_waitfor()
14 if status != 0:
15 error = True
16 if error:
mbligh4d6feff2008-01-14 16:48:56 +000017 raise AutoservError('One or more subcommands failed')
mblighb0fab822007-07-25 16:40:19 +000018
19
mbligh84c0ab12007-10-24 21:28:58 +000020def parallel_simple(function, arglist, log=True):
mblighb0fab822007-07-25 16:40:19 +000021 """Each element in the arglist used to create a subcommand object,
22 where that arg is used both as a subdir name, and a single argument
23 to pass to "function".
24 We create a subcommand object for each element in the list,
25 then execute those subcommand objects in parallel."""
mblighdd3235b2008-01-14 16:44:19 +000026
27 # Bypass the multithreading if only one machine.
28 if len (arglist) == 1:
29 function(arglist[0])
30 return
31
mblighb0fab822007-07-25 16:40:19 +000032 subcommands = []
33 for arg in arglist:
mbligh0c9e7822007-07-25 22:47:51 +000034 args = [arg]
mbligh84c0ab12007-10-24 21:28:58 +000035 if log:
36 subdir = str(arg)
37 else:
38 subdir = None
mbligh0c9e7822007-07-25 22:47:51 +000039 subcommands.append(subcommand(function, args, subdir))
mblighb0fab822007-07-25 16:40:19 +000040 parallel(subcommands)
41
42
mblighcee25b12007-08-31 08:53:05 +000043def _where_art_thy_filehandles():
mblighb0fab822007-07-25 16:40:19 +000044 os.system("ls -l /proc/%d/fd >> /dev/tty" % os.getpid())
45
mblighdc735a22007-08-02 16:54:37 +000046
mblighcee25b12007-08-31 08:53:05 +000047def _print_to_tty(string):
mblighb0fab822007-07-25 16:40:19 +000048 open('/dev/tty', 'w').write(string + '\n')
49
50
mblighcee25b12007-08-31 08:53:05 +000051def _redirect_stream(fd, output):
mblighb0fab822007-07-25 16:40:19 +000052 newfd = os.open(output, os.O_WRONLY | os.O_CREAT)
53 os.dup2(newfd, fd)
54 os.close(newfd)
55 if fd == 1:
56 sys.stdout = os.fdopen(fd, 'w')
57 if fd == 2:
58 sys.stderr = os.fdopen(fd, 'w')
59
60
mblighcee25b12007-08-31 08:53:05 +000061def _redirect_stream_tee(fd, output, tag):
mblighb0fab822007-07-25 16:40:19 +000062 """Use the low-level fork & pipe operations here to get a fd,
63 not a filehandle. This ensures that we get both the
64 filehandle and fd for stdout/stderr redirected correctly."""
65 r, w = os.pipe()
66 pid = os.fork()
67 if pid: # Parent
68 os.dup2(w, fd)
69 os.close(r)
70 os.close(w)
71 if fd == 1:
72 sys.stdout = os.fdopen(fd, 'w', 1)
73 if fd == 2:
74 sys.stderr = os.fdopen(fd, 'w', 1)
75 return
76 else: # Child
77 os.close(w)
78 log = open(output, 'w')
79 f = os.fdopen(r, 'r')
80 for line in iter(f.readline, ''):
81 # Tee straight to file
82 log.write(line)
83 log.flush()
84 # Prepend stdout with the tag
85 print tag + ' : ' + line,
86 sys.stdout.flush()
87 log.close()
88 os._exit(0)
89
90
91class subcommand:
mblighd7685d32007-08-10 22:08:42 +000092 def __init__(self, func, args, subdir = None, stdprint = True):
mblighb0fab822007-07-25 16:40:19 +000093 # func(args) - the subcommand to run
94 # subdir - the subdirectory to log results in
mblighd7685d32007-08-10 22:08:42 +000095 # stdprint - whether to print results to stdout/stderr
96 if subdir:
97 self.subdir = os.path.abspath(subdir)
98 if os.path.exists(self.subdir):
99 os.system("rm -rf %s" % self.subdir)
100 os.mkdir(self.subdir)
101 self.debug = os.path.join(self.subdir, 'debug')
mblighc526da32007-08-28 10:07:27 +0000102 os.mkdir(self.debug)
mblighd7685d32007-08-10 22:08:42 +0000103 self.stdout = os.path.join(self.debug, 'stdout')
104 self.stderr = os.path.join(self.debug, 'stderr')
105 else:
106 self.subdir = None
107 self.debug = '/dev/null'
108 self.stdout = '/dev/null'
109 self.stderr = '/dev/null'
110
mblighb0fab822007-07-25 16:40:19 +0000111 self.func = func
112 self.args = args
113 self.lambda_function = lambda: func(*args)
114 self.pid = None
mblighd7685d32007-08-10 22:08:42 +0000115 self.stdprint = stdprint
mblighb0fab822007-07-25 16:40:19 +0000116
117
118 def redirect_output(self):
mblighd7685d32007-08-10 22:08:42 +0000119 if self.stdprint:
120 if self.subdir:
121 tag = os.path.basename(self.subdir)
mblighcee25b12007-08-31 08:53:05 +0000122 _redirect_stream_tee(1, self.stdout, tag)
123 _redirect_stream_tee(2, self.stderr, tag)
mblighb491d022007-08-09 23:04:56 +0000124 else:
mblighcee25b12007-08-31 08:53:05 +0000125 _redirect_stream(1, self.stdout)
126 _redirect_stream(2, self.stderr)
mblighb0fab822007-07-25 16:40:19 +0000127
128
129 def fork_start(self):
130 sys.stdout.flush()
131 sys.stderr.flush()
132 self.pid = os.fork()
133
134 if self.pid: # I am the parent
135 return
136
137 # We are the child from this point on. Never return.
mblighd7685d32007-08-10 22:08:42 +0000138 if self.subdir:
139 os.chdir(self.subdir)
mblighb0fab822007-07-25 16:40:19 +0000140 self.redirect_output()
141
142 try:
143 self.lambda_function()
144
145 except:
mbligh42ff92f2007-11-24 19:13:15 +0000146 traceback.print_exc()
mblighb0fab822007-07-25 16:40:19 +0000147 sys.stdout.flush()
148 sys.stderr.flush()
149 os._exit(1)
150
151 sys.stdout.flush()
152 sys.stderr.flush()
153 os._exit(0)
154
155
156 def fork_waitfor(self):
157 (pid, status) = os.waitpid(self.pid, 0)
158
159 if status != 0:
160 print "subcommand failed pid %d" % pid
161 print "%s(%s)" % (self.func, self.args)
162 print "rc=%d" % status
163 print
164 if os.path.exists(self.stderr):
165 for line in open(self.stderr).readlines():
166 print line,
167 print "\n--------------------------------------------\n"
168 return status