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()
135 if jobs != None: self.maxjobs = jobs
136 if tmp: self.outstanding += tmp
140 "the tasks that are put to execute are all collected using get_out"
142 self.manager.add_finished(ret)
143 if not self.stop and getattr(ret, 'more_tasks', None):
144 self.outstanding += ret.more_tasks
145 self.total += len(ret.more_tasks)
148 def error_handler(self, tsk):
149 "by default, errors make the build stop (not thread safe so be careful)"
150 if not Options.options.keep:
159 self.refill_task_list()
161 # consider the next task
162 tsk = self.get_next()
165 # tasks may add new ones after they are run
168 # no tasks to run, no tasks running, time to exit
172 # if the task is marked as "run", just skip it
174 self.manager.add_finished(tsk)
178 st = tsk.runnable_status()
181 if self.stop and not Options.options.keep:
183 self.manager.add_finished(tsk)
185 self.error_handler(tsk)
186 self.manager.add_finished(tsk)
187 tsk.hasrun = EXCEPTION
188 tsk.err_msg = Utils.ex_stack()
196 self.manager.add_finished(tsk)
198 # run me: put the task in ready queue
199 tsk.position = (self.processed, self.total)
206 # self.count represents the tasks that have been made available to the consumer threads
207 # collect all the tasks after an error else the message may be incomplete
208 while self.error and self.count:
212 assert (self.count == 0 or self.stop)
217 Runner.process = process
218 Runner.Parallel = Parallel