summaryrefslogtreecommitdiff
path: root/buildtools/wafsamba/nothreads.py
diff options
context:
space:
mode:
Diffstat (limited to 'buildtools/wafsamba/nothreads.py')
-rw-r--r--buildtools/wafsamba/nothreads.py362
1 files changed, 181 insertions, 181 deletions
diff --git a/buildtools/wafsamba/nothreads.py b/buildtools/wafsamba/nothreads.py
index 9054a57af5..b39aeb5a63 100644
--- a/buildtools/wafsamba/nothreads.py
+++ b/buildtools/wafsamba/nothreads.py
@@ -22,195 +22,195 @@ GAP = 15
run_old = threading.Thread.run
def run(*args, **kwargs):
- try:
- run_old(*args, **kwargs)
- except (KeyboardInterrupt, SystemExit):
- raise
- except:
- sys.excepthook(*sys.exc_info())
+ try:
+ run_old(*args, **kwargs)
+ except (KeyboardInterrupt, SystemExit):
+ raise
+ except:
+ sys.excepthook(*sys.exc_info())
threading.Thread.run = run
class TaskConsumer(object):
- consumers = 1
+ consumers = 1
def process(tsk):
- m = tsk.master
- if m.stop:
- m.out.put(tsk)
- return
-
- try:
- tsk.generator.bld.printout(tsk.display())
- if tsk.__class__.stat: ret = tsk.__class__.stat(tsk)
- # actual call to task's run() function
- else: ret = tsk.call_run()
- except Exception, e:
- tsk.err_msg = Utils.ex_stack()
- tsk.hasrun = EXCEPTION
-
- # TODO cleanup
- m.error_handler(tsk)
- m.out.put(tsk)
- return
-
- if ret:
- tsk.err_code = ret
- tsk.hasrun = CRASHED
- else:
- try:
- tsk.post_run()
- except Utils.WafError:
- pass
- except Exception:
- tsk.err_msg = Utils.ex_stack()
- tsk.hasrun = EXCEPTION
- else:
- tsk.hasrun = SUCCESS
- if tsk.hasrun != SUCCESS:
- m.error_handler(tsk)
-
- m.out.put(tsk)
+ m = tsk.master
+ if m.stop:
+ m.out.put(tsk)
+ return
+
+ try:
+ tsk.generator.bld.printout(tsk.display())
+ if tsk.__class__.stat: ret = tsk.__class__.stat(tsk)
+ # actual call to task's run() function
+ else: ret = tsk.call_run()
+ except Exception, e:
+ tsk.err_msg = Utils.ex_stack()
+ tsk.hasrun = EXCEPTION
+
+ # TODO cleanup
+ m.error_handler(tsk)
+ m.out.put(tsk)
+ return
+
+ if ret:
+ tsk.err_code = ret
+ tsk.hasrun = CRASHED
+ else:
+ try:
+ tsk.post_run()
+ except Utils.WafError:
+ pass
+ except Exception:
+ tsk.err_msg = Utils.ex_stack()
+ tsk.hasrun = EXCEPTION
+ else:
+ tsk.hasrun = SUCCESS
+ if tsk.hasrun != SUCCESS:
+ m.error_handler(tsk)
+
+ m.out.put(tsk)
class Parallel(object):
- """
- keep the consumer threads busy, and avoid consuming cpu cycles
- when no more tasks can be added (end of the build, etc)
- """
- def __init__(self, bld, j=2):
-
- # number of consumers
- self.numjobs = j
-
- self.manager = bld.task_manager
- self.manager.current_group = 0
-
- self.total = self.manager.total()
-
- # tasks waiting to be processed - IMPORTANT
- self.outstanding = []
- self.maxjobs = MAXJOBS
-
- # tasks that are awaiting for another task to complete
- self.frozen = []
-
- # tasks returned by the consumers
- self.out = Queue(0)
-
- self.count = 0 # tasks not in the producer area
-
- self.processed = 1 # progress indicator
-
- self.stop = False # error condition to stop the build
- self.error = False # error flag
-
- def get_next(self):
- "override this method to schedule the tasks in a particular order"
- if not self.outstanding:
- return None
- return self.outstanding.pop(0)
-
- def postpone(self, tsk):
- "override this method to schedule the tasks in a particular order"
- # TODO consider using a deque instead
- if random.randint(0, 1):
- self.frozen.insert(0, tsk)
- else:
- self.frozen.append(tsk)
-
- def refill_task_list(self):
- "called to set the next group of tasks"
-
- while self.count > self.numjobs + GAP or self.count >= self.maxjobs:
- self.get_out()
-
- while not self.outstanding:
- if self.count:
- self.get_out()
-
- if self.frozen:
- self.outstanding += self.frozen
- self.frozen = []
- elif not self.count:
- (jobs, tmp) = self.manager.get_next_set()
- if jobs != None: self.maxjobs = jobs
- if tmp: self.outstanding += tmp
- break
-
- def get_out(self):
- "the tasks that are put to execute are all collected using get_out"
- ret = self.out.get()
- self.manager.add_finished(ret)
- if not self.stop and getattr(ret, 'more_tasks', None):
- self.outstanding += ret.more_tasks
- self.total += len(ret.more_tasks)
- self.count -= 1
-
- def error_handler(self, tsk):
- "by default, errors make the build stop (not thread safe so be careful)"
- if not Options.options.keep:
- self.stop = True
- self.error = True
-
- def start(self):
- "execute the tasks"
-
- while not self.stop:
-
- self.refill_task_list()
-
- # consider the next task
- tsk = self.get_next()
- if not tsk:
- if self.count:
- # tasks may add new ones after they are run
- continue
- else:
- # no tasks to run, no tasks running, time to exit
- break
-
- if tsk.hasrun:
- # if the task is marked as "run", just skip it
- self.processed += 1
- self.manager.add_finished(tsk)
- continue
-
- try:
- st = tsk.runnable_status()
- except Exception, e:
- self.processed += 1
- if self.stop and not Options.options.keep:
- tsk.hasrun = SKIPPED
- self.manager.add_finished(tsk)
- continue
- self.error_handler(tsk)
- self.manager.add_finished(tsk)
- tsk.hasrun = EXCEPTION
- tsk.err_msg = Utils.ex_stack()
- continue
-
- if st == ASK_LATER:
- self.postpone(tsk)
- elif st == SKIP_ME:
- self.processed += 1
- tsk.hasrun = SKIPPED
- self.manager.add_finished(tsk)
- else:
- # run me: put the task in ready queue
- tsk.position = (self.processed, self.total)
- self.count += 1
- self.processed += 1
- tsk.master = self
-
- process(tsk)
-
- # self.count represents the tasks that have been made available to the consumer threads
- # collect all the tasks after an error else the message may be incomplete
- while self.error and self.count:
- self.get_out()
-
- #print loop
- assert (self.count == 0 or self.stop)
+ """
+ keep the consumer threads busy, and avoid consuming cpu cycles
+ when no more tasks can be added (end of the build, etc)
+ """
+ def __init__(self, bld, j=2):
+
+ # number of consumers
+ self.numjobs = j
+
+ self.manager = bld.task_manager
+ self.manager.current_group = 0
+
+ self.total = self.manager.total()
+
+ # tasks waiting to be processed - IMPORTANT
+ self.outstanding = []
+ self.maxjobs = MAXJOBS
+
+ # tasks that are awaiting for another task to complete
+ self.frozen = []
+
+ # tasks returned by the consumers
+ self.out = Queue(0)
+
+ self.count = 0 # tasks not in the producer area
+
+ self.processed = 1 # progress indicator
+
+ self.stop = False # error condition to stop the build
+ self.error = False # error flag
+
+ def get_next(self):
+ "override this method to schedule the tasks in a particular order"
+ if not self.outstanding:
+ return None
+ return self.outstanding.pop(0)
+
+ def postpone(self, tsk):
+ "override this method to schedule the tasks in a particular order"
+ # TODO consider using a deque instead
+ if random.randint(0, 1):
+ self.frozen.insert(0, tsk)
+ else:
+ self.frozen.append(tsk)
+
+ def refill_task_list(self):
+ "called to set the next group of tasks"
+
+ while self.count > self.numjobs + GAP or self.count >= self.maxjobs:
+ self.get_out()
+
+ while not self.outstanding:
+ if self.count:
+ self.get_out()
+
+ if self.frozen:
+ self.outstanding += self.frozen
+ self.frozen = []
+ elif not self.count:
+ (jobs, tmp) = self.manager.get_next_set()
+ if jobs != None: self.maxjobs = jobs
+ if tmp: self.outstanding += tmp
+ break
+
+ def get_out(self):
+ "the tasks that are put to execute are all collected using get_out"
+ ret = self.out.get()
+ self.manager.add_finished(ret)
+ if not self.stop and getattr(ret, 'more_tasks', None):
+ self.outstanding += ret.more_tasks
+ self.total += len(ret.more_tasks)
+ self.count -= 1
+
+ def error_handler(self, tsk):
+ "by default, errors make the build stop (not thread safe so be careful)"
+ if not Options.options.keep:
+ self.stop = True
+ self.error = True
+
+ def start(self):
+ "execute the tasks"
+
+ while not self.stop:
+
+ self.refill_task_list()
+
+ # consider the next task
+ tsk = self.get_next()
+ if not tsk:
+ if self.count:
+ # tasks may add new ones after they are run
+ continue
+ else:
+ # no tasks to run, no tasks running, time to exit
+ break
+
+ if tsk.hasrun:
+ # if the task is marked as "run", just skip it
+ self.processed += 1
+ self.manager.add_finished(tsk)
+ continue
+
+ try:
+ st = tsk.runnable_status()
+ except Exception, e:
+ self.processed += 1
+ if self.stop and not Options.options.keep:
+ tsk.hasrun = SKIPPED
+ self.manager.add_finished(tsk)
+ continue
+ self.error_handler(tsk)
+ self.manager.add_finished(tsk)
+ tsk.hasrun = EXCEPTION
+ tsk.err_msg = Utils.ex_stack()
+ continue
+
+ if st == ASK_LATER:
+ self.postpone(tsk)
+ elif st == SKIP_ME:
+ self.processed += 1
+ tsk.hasrun = SKIPPED
+ self.manager.add_finished(tsk)
+ else:
+ # run me: put the task in ready queue
+ tsk.position = (self.processed, self.total)
+ self.count += 1
+ self.processed += 1
+ tsk.master = self
+
+ process(tsk)
+
+ # self.count represents the tasks that have been made available to the consumer threads
+ # collect all the tasks after an error else the message may be incomplete
+ while self.error and self.count:
+ self.get_out()
+
+ #print loop
+ assert (self.count == 0 or self.stop)
# enable nothreads
generated by cgit (git 2.34.1) at 2024-12-28 14:33:04 +0000