2 # Thomas Nagy, 2005-2008 (ita)
4 # this replaces the core of Runner.py in waf with a varient that works
5 # on systems with completely broken threading (such as Python 2.5.x on
6 # AIX). For simplicity we enable this when JOBS=1, which is triggered
7 # by the compatibility makefile used for the waf build. That also ensures
8 # this code is tested, as it means it is used in the build farm, and by
9 # anyone using 'make' to build Samba with waf
13 import sys, random, time, threading, traceback, os
14 try: from Queue import Queue
15 except ImportError: from queue import Queue
16 import Build, Utils, Logs, Options
17 from Logs import debug, error
18 from Constants import *
22 run_old = threading.Thread.run
23 def run(*args, **kwargs):
25 run_old(*args, **kwargs)
26 except (KeyboardInterrupt, SystemExit):
29 sys.excepthook(*sys.exc_info())
30 threading.Thread.run = run
33 class TaskConsumer(object):
43 tsk.generator.bld.printout(tsk.display())
44 if tsk.__class__.stat: ret = tsk.__class__.stat(tsk)
45 # actual call to task's run() function
46 else: ret = tsk.call_run()
48 tsk.err_msg = Utils.ex_stack()
49 tsk.hasrun = EXCEPTION
62 except Utils.WafError:
65 tsk.err_msg = Utils.ex_stack()
66 tsk.hasrun = EXCEPTION
69 if tsk.hasrun != SUCCESS:
74 class Parallel(object):
76 keep the consumer threads busy, and avoid consuming cpu cycles
77 when no more tasks can be added (end of the build, etc)
79 def __init__(self, bld, j=2):
84 self.manager = bld.task_manager
85 self.manager.current_group = 0
87 self.total = self.manager.total()
89 # tasks waiting to be processed - IMPORTANT
91 self.maxjobs = MAXJOBS
93 # tasks that are awaiting for another task to complete
96 # tasks returned by the consumers
99 self.count = 0 # tasks not in the producer area
101 self.processed = 1 # progress indicator
103 self.stop = False # error condition to stop the build
104 self.error = False # error flag
107 "override this method to schedule the tasks in a particular order"
108 if not self.outstanding:
110 return self.outstanding.pop(0)
112 def postpone(self, tsk):
113 "override this method to schedule the tasks in a particular order"
114 # TODO consider using a deque instead
115 if random.randint(0, 1):
116 self.frozen.insert(0, tsk)
118 self.frozen.append(tsk)
120 def refill_task_list(self):
121 "called to set the next group of tasks"
123 while self.count > self.numjobs + GAP or self.count >= self.maxjobs:
126 while not self.outstanding:
131 self.outstanding += self.frozen
134 (jobs, tmp) = self.manager.get_next_set()
138 self.outstanding += tmp
142 "the tasks that are put to execute are all collected using get_out"
144 self.manager.add_finished(ret)
145 if not self.stop and getattr(ret, 'more_tasks', None):
146 self.outstanding += ret.more_tasks
147 self.total += len(ret.more_tasks)
150 def error_handler(self, tsk):
151 "by default, errors make the build stop (not thread safe so be careful)"
152 if not Options.options.keep:
161 self.refill_task_list()
163 # consider the next task
164 tsk = self.get_next()
167 # tasks may add new ones after they are run
170 # no tasks to run, no tasks running, time to exit
174 # if the task is marked as "run", just skip it
176 self.manager.add_finished(tsk)
180 st = tsk.runnable_status()
183 if self.stop and not Options.options.keep:
185 self.manager.add_finished(tsk)
187 self.error_handler(tsk)
188 self.manager.add_finished(tsk)
189 tsk.hasrun = EXCEPTION
190 tsk.err_msg = Utils.ex_stack()
198 self.manager.add_finished(tsk)
200 # run me: put the task in ready queue
201 tsk.position = (self.processed, self.total)
208 # self.count represents the tasks that have been made available to the consumer threads
209 # collect all the tasks after an error else the message may be incomplete
210 while self.error and self.count:
214 assert (self.count == 0 or self.stop)
219 Runner.process = process
220 Runner.Parallel = Parallel