diff options
Diffstat (limited to 'waflib/Runner.py')
| m--------- | waflib | 0 | ||||
| -rw-r--r-- | waflib/Runner.py | 617 |
2 files changed, 0 insertions, 617 deletions
diff --git a/waflib b/waflib new file mode 160000 +Subproject 2314e236ca6e7d94a26c3c17091da0f25f5867f diff --git a/waflib/Runner.py b/waflib/Runner.py deleted file mode 100644 index 5d276698..00000000 --- a/waflib/Runner.py +++ /dev/null @@ -1,617 +0,0 @@ -#!/usr/bin/env python -# encoding: utf-8 -# Thomas Nagy, 2005-2018 (ita) - -""" -Runner.py: Task scheduling and execution -""" - -import heapq, traceback -try: - from queue import Queue, PriorityQueue -except ImportError: - from Queue import Queue - try: - from Queue import PriorityQueue - except ImportError: - class PriorityQueue(Queue): - def _init(self, maxsize): - self.maxsize = maxsize - self.queue = [] - def _put(self, item): - heapq.heappush(self.queue, item) - def _get(self): - return heapq.heappop(self.queue) - -from waflib import Utils, Task, Errors, Logs - -GAP = 5 -""" -Wait for at least ``GAP * njobs`` before trying to enqueue more tasks to run -""" - -class PriorityTasks(object): - def __init__(self): - self.lst = [] - def __len__(self): - return len(self.lst) - def __iter__(self): - return iter(self.lst) - def __str__(self): - return 'PriorityTasks: [%s]' % '\n '.join(str(x) for x in self.lst) - def clear(self): - self.lst = [] - def append(self, task): - heapq.heappush(self.lst, task) - def appendleft(self, task): - "Deprecated, do not use" - heapq.heappush(self.lst, task) - def pop(self): - return heapq.heappop(self.lst) - def extend(self, lst): - if self.lst: - for x in lst: - self.append(x) - else: - if isinstance(lst, list): - self.lst = lst - heapq.heapify(lst) - else: - self.lst = lst.lst - -class Consumer(Utils.threading.Thread): - """ - Daemon thread object that executes a task. It shares a semaphore with - the coordinator :py:class:`waflib.Runner.Spawner`. There is one - instance per task to consume. - """ - def __init__(self, spawner, task): - Utils.threading.Thread.__init__(self) - self.task = task - """Task to execute""" - self.spawner = spawner - """Coordinator object""" - self.setDaemon(1) - self.start() - def run(self): - """ - Processes a single task - """ - try: - if not self.spawner.master.stop: - self.spawner.master.process_task(self.task) - finally: - self.spawner.sem.release() - self.spawner.master.out.put(self.task) - self.task = None - self.spawner = None - -class Spawner(Utils.threading.Thread): - """ - Daemon thread that consumes tasks from :py:class:`waflib.Runner.Parallel` producer and - spawns a consuming thread :py:class:`waflib.Runner.Consumer` for each - :py:class:`waflib.Task.Task` instance. - """ - def __init__(self, master): - Utils.threading.Thread.__init__(self) - self.master = master - """:py:class:`waflib.Runner.Parallel` producer instance""" - self.sem = Utils.threading.Semaphore(master.numjobs) - """Bounded semaphore that prevents spawning more than *n* concurrent consumers""" - self.setDaemon(1) - self.start() - def run(self): - """ - Spawns new consumers to execute tasks by delegating to :py:meth:`waflib.Runner.Spawner.loop` - """ - try: - self.loop() - except Exception: - # Python 2 prints unnecessary messages when shutting down - # we also want to stop the thread properly - pass - def loop(self): - """ - Consumes task objects from the producer; ends when the producer has no more - task to provide. - """ - master = self.master - while 1: - task = master.ready.get() - self.sem.acquire() - if not master.stop: - task.log_display(task.generator.bld) - Consumer(self, task) - -class Parallel(object): - """ - Schedule the tasks obtained from the build context for execution. - """ - def __init__(self, bld, j=2): - """ - The initialization requires a build context reference - for computing the total number of jobs. - """ - - self.numjobs = j - """ - Amount of parallel consumers to use - """ - - self.bld = bld - """ - Instance of :py:class:`waflib.Build.BuildContext` - """ - - self.outstanding = PriorityTasks() - """Heap of :py:class:`waflib.Task.Task` that may be ready to be executed""" - - self.postponed = PriorityTasks() - """Heap of :py:class:`waflib.Task.Task` which are not ready to run for non-DAG reasons""" - - self.incomplete = set() - """List of :py:class:`waflib.Task.Task` waiting for dependent tasks to complete (DAG)""" - - self.ready = PriorityQueue(0) - """List of :py:class:`waflib.Task.Task` ready to be executed by consumers""" - - self.out = Queue(0) - """List of :py:class:`waflib.Task.Task` returned by the task consumers""" - - self.count = 0 - """Amount of tasks that may be processed by :py:class:`waflib.Runner.TaskConsumer`""" - - self.processed = 0 - """Amount of tasks processed""" - - self.stop = False - """Error flag to stop the build""" - - self.error = [] - """Tasks that could not be executed""" - - self.biter = None - """Task iterator which must give groups of parallelizable tasks when calling ``next()``""" - - self.dirty = False - """ - Flag that indicates that the build cache must be saved when a task was executed - (calls :py:meth:`waflib.Build.BuildContext.store`)""" - - self.revdeps = Utils.defaultdict(set) - """ - The reverse dependency graph of dependencies obtained from Task.run_after - """ - - self.spawner = None - """ - Coordinating daemon thread that spawns thread consumers - """ - if self.numjobs > 1: - self.spawner = Spawner(self) - - def get_next_task(self): - """ - Obtains the next Task instance to run - - :rtype: :py:class:`waflib.Task.Task` - """ - if not self.outstanding: - return None - return self.outstanding.pop() - - def postpone(self, tsk): - """ - Adds the task to the list :py:attr:`waflib.Runner.Parallel.postponed`. - The order is scrambled so as to consume as many tasks in parallel as possible. - - :param tsk: task instance - :type tsk: :py:class:`waflib.Task.Task` - """ - self.postponed.append(tsk) - - def refill_task_list(self): - """ - Pulls a next group of tasks to execute in :py:attr:`waflib.Runner.Parallel.outstanding`. - Ensures that all tasks in the current build group are complete before processing the next one. - """ - while self.count > self.numjobs * GAP: - self.get_out() - - while not self.outstanding: - if self.count: - self.get_out() - if self.outstanding: - break - elif self.postponed: - try: - cond = self.deadlock == self.processed - except AttributeError: - pass - else: - if cond: - # The most common reason is conflicting build order declaration - # for example: "X run_after Y" and "Y run_after X" - # Another can be changing "run_after" dependencies while the build is running - # for example: updating "tsk.run_after" in the "runnable_status" method - lst = [] - for tsk in self.postponed: - deps = [id(x) for x in tsk.run_after if not x.hasrun] - lst.append('%s\t-> %r' % (repr(tsk), deps)) - if not deps: - lst.append('\n task %r dependencies are done, check its *runnable_status*?' % id(tsk)) - raise Errors.WafError('Deadlock detected: check the task build order%s' % ''.join(lst)) - self.deadlock = self.processed - - if self.postponed: - self.outstanding.extend(self.postponed) - self.postponed.clear() - elif not self.count: - if self.incomplete: - for x in self.incomplete: - for k in x.run_after: - if not k.hasrun: - break - else: - # dependency added after the build started without updating revdeps - self.incomplete.remove(x) - self.outstanding.append(x) - break - else: - if self.stop or self.error: - break - raise Errors.WafError('Broken revdeps detected on %r' % self.incomplete) - else: - tasks = next(self.biter) - ready, waiting = self.prio_and_split(tasks) - self.outstanding.extend(ready) - self.incomplete.update(waiting) - self.total = self.bld.total() - break - - def add_more_tasks(self, tsk): - """ - If a task provides :py:attr:`waflib.Task.Task.more_tasks`, then the tasks contained - in that list are added to the current build and will be processed before the next build group. - - The priorities for dependent tasks are not re-calculated globally - - :param tsk: task instance - :type tsk: :py:attr:`waflib.Task.Task` - """ - if getattr(tsk, 'more_tasks', None): - more = set(tsk.more_tasks) - groups_done = set() - def iteri(a, b): - for x in a: - yield x - for x in b: - yield x - - # Update the dependency tree - # this assumes that task.run_after values were updated - for x in iteri(self.outstanding, self.incomplete): - for k in x.run_after: - if isinstance(k, Task.TaskGroup): - if k not in groups_done: - groups_done.add(k) - for j in k.prev & more: - self.revdeps[j].add(k) - elif k in more: - self.revdeps[k].add(x) - - ready, waiting = self.prio_and_split(tsk.more_tasks) - self.outstanding.extend(ready) - self.incomplete.update(waiting) - self.total += len(tsk.more_tasks) - - def mark_finished(self, tsk): - def try_unfreeze(x): - # DAG ancestors are likely to be in the incomplete set - # This assumes that the run_after contents have not changed - # after the build starts, else a deadlock may occur - if x in self.incomplete: - # TODO remove dependencies to free some memory? - # x.run_after.remove(tsk) - for k in x.run_after: - if not k.hasrun: - break - else: - self.incomplete.remove(x) - self.outstanding.append(x) - - if tsk in self.revdeps: - for x in self.revdeps[tsk]: - if isinstance(x, Task.TaskGroup): - x.prev.remove(tsk) - if not x.prev: - for k in x.next: - # TODO necessary optimization? - k.run_after.remove(x) - try_unfreeze(k) - # TODO necessary optimization? - x.next = [] - else: - try_unfreeze(x) - del self.revdeps[tsk] - - if hasattr(tsk, 'semaphore'): - sem = tsk.semaphore - sem.release(tsk) - while sem.waiting and not sem.is_locked(): - # take a frozen task, make it ready to run - x = sem.waiting.pop() - self._add_task(x) - - def get_out(self): - """ - Waits for a Task that task consumers add to :py:attr:`waflib.Runner.Parallel.out` after execution. - Adds more Tasks if necessary through :py:attr:`waflib.Runner.Parallel.add_more_tasks`. - - :rtype: :py:attr:`waflib.Task.Task` - """ - tsk = self.out.get() - if not self.stop: - self.add_more_tasks(tsk) - self.mark_finished(tsk) - - self.count -= 1 - self.dirty = True - return tsk - - def add_task(self, tsk): - """ - Enqueue a Task to :py:attr:`waflib.Runner.Parallel.ready` so that consumers can run them. - - :param tsk: task instance - :type tsk: :py:attr:`waflib.Task.Task` - """ - # TODO change in waf 2.1 - self.ready.put(tsk) - - def _add_task(self, tsk): - if hasattr(tsk, 'semaphore'): - sem = tsk.semaphore - try: - sem.acquire(tsk) - except IndexError: - sem.waiting.add(tsk) - return - - self.count += 1 - self.processed += 1 - if self.numjobs == 1: - tsk.log_display(tsk.generator.bld) - try: - self.process_task(tsk) - finally: - self.out.put(tsk) - else: - self.add_task(tsk) - - def process_task(self, tsk): - """ - Processes a task and attempts to stop the build in case of errors - """ - tsk.process() - if tsk.hasrun != Task.SUCCESS: - self.error_handler(tsk) - - def skip(self, tsk): - """ - Mark a task as skipped/up-to-date - """ - tsk.hasrun = Task.SKIPPED - self.mark_finished(tsk) - - def cancel(self, tsk): - """ - Mark a task as failed because of unsatisfiable dependencies - """ - tsk.hasrun = Task.CANCELED - self.mark_finished(tsk) - - def error_handler(self, tsk): - """ - Called when a task cannot be executed. The flag :py:attr:`waflib.Runner.Parallel.stop` is set, - unless the build is executed with:: - - $ waf build -k - - :param tsk: task instance - :type tsk: :py:attr:`waflib.Task.Task` - """ - if not self.bld.keep: - self.stop = True - self.error.append(tsk) - - def task_status(self, tsk): - """ - Obtains the task status to decide whether to run it immediately or not. - - :return: the exit status, for example :py:attr:`waflib.Task.ASK_LATER` - :rtype: integer - """ - try: - return tsk.runnable_status() - except Exception: - self.processed += 1 - tsk.err_msg = traceback.format_exc() - if not self.stop and self.bld.keep: - self.skip(tsk) - if self.bld.keep == 1: - # if -k stop on the first exception, if -kk try to go as far as possible - if Logs.verbose > 1 or not self.error: - self.error.append(tsk) - self.stop = True - else: - if Logs.verbose > 1: - self.error.append(tsk) - return Task.EXCEPTION - - tsk.hasrun = Task.EXCEPTION - self.error_handler(tsk) - - return Task.EXCEPTION - - def start(self): - """ - Obtains Task instances from the BuildContext instance and adds the ones that need to be executed to - :py:class:`waflib.Runner.Parallel.ready` so that the :py:class:`waflib.Runner.Spawner` consumer thread - has them executed. Obtains the executed Tasks back from :py:class:`waflib.Runner.Parallel.out` - and marks the build as failed by setting the ``stop`` flag. - If only one job is used, then executes the tasks one by one, without consumers. - """ - self.total = self.bld.total() - - while not self.stop: - - self.refill_task_list() - - # consider the next task - tsk = self.get_next_task() - 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 - continue - - if self.stop: # stop immediately after a failure is detected - break - - st = self.task_status(tsk) - if st == Task.RUN_ME: - self._add_task(tsk) - elif st == Task.ASK_LATER: - self.postpone(tsk) - elif st == Task.SKIP_ME: - self.processed += 1 - self.skip(tsk) - self.add_more_tasks(tsk) - elif st == Task.CANCEL_ME: - # A dependency problem has occurred, and the - # build is most likely run with `waf -k` - if Logs.verbose > 1: - self.error.append(tsk) - self.processed += 1 - self.cancel(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() - - self.ready.put(None) - if not self.stop: - assert not self.count - assert not self.postponed - assert not self.incomplete - - def prio_and_split(self, tasks): - """ - Label input tasks with priority values, and return a pair containing - the tasks that are ready to run and the tasks that are necessarily - waiting for other tasks to complete. - - The priority system is really meant as an optional layer for optimization: - dependency cycles are found quickly, and builds should be more efficient. - A high priority number means that a task is processed first. - - This method can be overridden to disable the priority system:: - - def prio_and_split(self, tasks): - return tasks, [] - - :return: A pair of task lists - :rtype: tuple - """ - # to disable: - #return tasks, [] - for x in tasks: - x.visited = 0 - - reverse = self.revdeps - - groups_done = set() - for x in tasks: - for k in x.run_after: - if isinstance(k, Task.TaskGroup): - if k not in groups_done: - groups_done.add(k) - for j in k.prev: - reverse[j].add(k) - else: - reverse[k].add(x) - - # the priority number is not the tree depth - def visit(n): - if isinstance(n, Task.TaskGroup): - return sum(visit(k) for k in n.next) - - if n.visited == 0: - n.visited = 1 - - if n in reverse: - rev = reverse[n] - n.prio_order = n.tree_weight + len(rev) + sum(visit(k) for k in rev) - else: - n.prio_order = n.tree_weight - - n.visited = 2 - elif n.visited == 1: - raise Errors.WafError('Dependency cycle found!') - return n.prio_order - - for x in tasks: - if x.visited != 0: - # must visit all to detect cycles - continue - try: - visit(x) - except Errors.WafError: - self.debug_cycles(tasks, reverse) - - ready = [] - waiting = [] - for x in tasks: - for k in x.run_after: - if not k.hasrun: - waiting.append(x) - break - else: - ready.append(x) - return (ready, waiting) - - def debug_cycles(self, tasks, reverse): - tmp = {} - for x in tasks: - tmp[x] = 0 - - def visit(n, acc): - if isinstance(n, Task.TaskGroup): - for k in n.next: - visit(k, acc) - return - if tmp[n] == 0: - tmp[n] = 1 - for k in reverse.get(n, []): - visit(k, [n] + acc) - tmp[n] = 2 - elif tmp[n] == 1: - lst = [] - for tsk in acc: - lst.append(repr(tsk)) - if tsk is n: - # exclude prior nodes, we want the minimum cycle - break - raise Errors.WafError('Task dependency cycle in "run_after" constraints: %s' % ''.join(lst)) - for x in tasks: - visit(x, []) - |