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, threading
14 try: from Queue import Queue
15 except ImportError: from queue import Queue
16 from waflib import Utils, Options, Errors
17 from waflib.TaskGen import EXCEPTION, CRASHED, MAXJOBS, ASK_LATER, SKIPPED, SKIP_ME, SUCCESS
21 run_old = threading.Thread.run
22 def run(*args, **kwargs):
24 run_old(*args, **kwargs)
25 except (KeyboardInterrupt, SystemExit):
28 sys.excepthook(*sys.exc_info())
29 threading.Thread.run = run
32 class TaskConsumer(object):
42 tsk.generator.bld.printout(tsk.display())
43 if tsk.__class__.stat: ret = tsk.__class__.stat(tsk)
44 # actual call to task's run() function
45 else: ret = tsk.call_run()
46 except Exception as e:
47 tsk.err_msg = Utils.ex_stack()
48 tsk.hasrun = EXCEPTION
61 except Errors.WafError:
64 tsk.err_msg = Utils.ex_stack()
65 tsk.hasrun = EXCEPTION
68 if tsk.hasrun != SUCCESS:
73 class Parallel(object):
75 keep the consumer threads busy, and avoid consuming cpu cycles
76 when no more tasks can be added (end of the build, etc)
78 def __init__(self, bld, j=2):
83 self.manager = bld.task_manager
84 self.manager.current_group = 0
86 self.total = self.manager.total()
88 # tasks waiting to be processed - IMPORTANT
90 self.maxjobs = MAXJOBS
92 # tasks that are awaiting for another task to complete
95 # tasks returned by the consumers
98 self.count = 0 # tasks not in the producer area
100 self.processed = 1 # progress indicator
102 self.stop = False # error condition to stop the build
103 self.error = False # error flag
106 "override this method to schedule the tasks in a particular order"
107 if not self.outstanding:
109 return self.outstanding.pop(0)
111 def postpone(self, tsk):
112 "override this method to schedule the tasks in a particular order"
113 # TODO consider using a deque instead
114 if random.randint(0, 1):
115 self.frozen.insert(0, tsk)
117 self.frozen.append(tsk)
119 def refill_task_list(self):
120 "called to set the next group of tasks"
122 while self.count > self.numjobs + GAP or self.count >= self.maxjobs:
125 while not self.outstanding:
130 self.outstanding += self.frozen
133 (jobs, tmp) = self.manager.get_next_set()
137 self.outstanding += tmp
141 "the tasks that are put to execute are all collected using get_out"
143 self.manager.add_finished(ret)
144 if not self.stop and getattr(ret, 'more_tasks', None):
145 self.outstanding += ret.more_tasks
146 self.total += len(ret.more_tasks)
149 def error_handler(self, tsk):
150 "by default, errors make the build stop (not thread safe so be careful)"
151 if not Options.options.keep:
160 self.refill_task_list()
162 # consider the next task
163 tsk = self.get_next()
166 # tasks may add new ones after they are run
169 # no tasks to run, no tasks running, time to exit
173 # if the task is marked as "run", just skip it
175 self.manager.add_finished(tsk)
179 st = tsk.runnable_status()
180 except Exception as e:
182 if self.stop and not Options.options.keep:
184 self.manager.add_finished(tsk)
186 self.error_handler(tsk)
187 self.manager.add_finished(tsk)
188 tsk.hasrun = EXCEPTION
189 tsk.err_msg = Utils.ex_stack()
197 self.manager.add_finished(tsk)
199 # run me: put the task in ready queue
200 tsk.position = (self.processed, self.total)
207 # self.count represents the tasks that have been made available to the consumer threads
208 # collect all the tasks after an error else the message may be incomplete
209 while self.error and self.count:
213 assert (self.count == 0 or self.stop)
218 Runner.process = process
219 Runner.Parallel = Parallel