| #!/usr/bin/env python |
| # ex:ts=4:sw=4:sts=4:et |
| # -*- tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- |
| """ |
| BitBake 'RunQueue' implementation |
| |
| Handles preparation and execution of a queue of tasks |
| """ |
| |
| # Copyright (C) 2006-2007 Richard Purdie |
| # |
| # This program is free software; you can redistribute it and/or modify |
| # it under the terms of the GNU General Public License version 2 as |
| # published by the Free Software Foundation. |
| # |
| # This program is distributed in the hope that it will be useful, |
| # but WITHOUT ANY WARRANTY; without even the implied warranty of |
| # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
| # GNU General Public License for more details. |
| # |
| # You should have received a copy of the GNU General Public License along |
| # with this program; if not, write to the Free Software Foundation, Inc., |
| # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. |
| |
| import copy |
| import os |
| import sys |
| import signal |
| import stat |
| import fcntl |
| import errno |
| import logging |
| import re |
| import bb |
| from bb import msg, data, event |
| from bb import monitordisk |
| import subprocess |
| |
| try: |
| import cPickle as pickle |
| except ImportError: |
| import pickle |
| |
| bblogger = logging.getLogger("BitBake") |
| logger = logging.getLogger("BitBake.RunQueue") |
| |
| __find_md5__ = re.compile( r'(?i)(?<![a-z0-9])[a-f0-9]{32}(?![a-z0-9])' ) |
| |
| class RunQueueStats: |
| """ |
| Holds statistics on the tasks handled by the associated runQueue |
| """ |
| def __init__(self, total): |
| self.completed = 0 |
| self.skipped = 0 |
| self.failed = 0 |
| self.active = 0 |
| self.total = total |
| |
| def copy(self): |
| obj = self.__class__(self.total) |
| obj.__dict__.update(self.__dict__) |
| return obj |
| |
| def taskFailed(self): |
| self.active = self.active - 1 |
| self.failed = self.failed + 1 |
| |
| def taskCompleted(self, number = 1): |
| self.active = self.active - number |
| self.completed = self.completed + number |
| |
| def taskSkipped(self, number = 1): |
| self.active = self.active + number |
| self.skipped = self.skipped + number |
| |
| def taskActive(self): |
| self.active = self.active + 1 |
| |
| # These values indicate the next step due to be run in the |
| # runQueue state machine |
| runQueuePrepare = 2 |
| runQueueSceneInit = 3 |
| runQueueSceneRun = 4 |
| runQueueRunInit = 5 |
| runQueueRunning = 6 |
| runQueueFailed = 7 |
| runQueueCleanUp = 8 |
| runQueueComplete = 9 |
| |
| class RunQueueScheduler(object): |
| """ |
| Control the order tasks are scheduled in. |
| """ |
| name = "basic" |
| |
| def __init__(self, runqueue, rqdata): |
| """ |
| The default scheduler just returns the first buildable task (the |
| priority map is sorted by task number) |
| """ |
| self.rq = runqueue |
| self.rqdata = rqdata |
| self.numTasks = len(self.rqdata.runq_fnid) |
| |
| self.prio_map = [] |
| self.prio_map.extend(range(self.numTasks)) |
| |
| self.buildable = [] |
| self.stamps = {} |
| for taskid in xrange(self.numTasks): |
| fn = self.rqdata.taskData.fn_index[self.rqdata.runq_fnid[taskid]] |
| taskname = self.rqdata.runq_task[taskid] |
| self.stamps[taskid] = bb.build.stampfile(taskname, self.rqdata.dataCache, fn) |
| if self.rq.runq_buildable[taskid] == 1: |
| self.buildable.append(taskid) |
| |
| self.rev_prio_map = None |
| |
| def next_buildable_task(self): |
| """ |
| Return the id of the first task we find that is buildable |
| """ |
| self.buildable = [x for x in self.buildable if not self.rq.runq_running[x] == 1] |
| if not self.buildable: |
| return None |
| if len(self.buildable) == 1: |
| taskid = self.buildable[0] |
| stamp = self.stamps[taskid] |
| if stamp not in self.rq.build_stamps.itervalues(): |
| return taskid |
| |
| if not self.rev_prio_map: |
| self.rev_prio_map = range(self.numTasks) |
| for taskid in xrange(self.numTasks): |
| self.rev_prio_map[self.prio_map[taskid]] = taskid |
| |
| best = None |
| bestprio = None |
| for taskid in self.buildable: |
| prio = self.rev_prio_map[taskid] |
| if bestprio is None or bestprio > prio: |
| stamp = self.stamps[taskid] |
| if stamp in self.rq.build_stamps.itervalues(): |
| continue |
| bestprio = prio |
| best = taskid |
| |
| return best |
| |
| def next(self): |
| """ |
| Return the id of the task we should build next |
| """ |
| if self.rq.stats.active < self.rq.number_tasks: |
| return self.next_buildable_task() |
| |
| def newbuilable(self, task): |
| self.buildable.append(task) |
| |
| class RunQueueSchedulerSpeed(RunQueueScheduler): |
| """ |
| A scheduler optimised for speed. The priority map is sorted by task weight, |
| heavier weighted tasks (tasks needed by the most other tasks) are run first. |
| """ |
| name = "speed" |
| |
| def __init__(self, runqueue, rqdata): |
| """ |
| The priority map is sorted by task weight. |
| """ |
| RunQueueScheduler.__init__(self, runqueue, rqdata) |
| |
| sortweight = sorted(copy.deepcopy(self.rqdata.runq_weight)) |
| copyweight = copy.deepcopy(self.rqdata.runq_weight) |
| self.prio_map = [] |
| |
| for weight in sortweight: |
| idx = copyweight.index(weight) |
| self.prio_map.append(idx) |
| copyweight[idx] = -1 |
| |
| self.prio_map.reverse() |
| |
| class RunQueueSchedulerCompletion(RunQueueSchedulerSpeed): |
| """ |
| A scheduler optimised to complete .bb files are quickly as possible. The |
| priority map is sorted by task weight, but then reordered so once a given |
| .bb file starts to build, it's completed as quickly as possible. This works |
| well where disk space is at a premium and classes like OE's rm_work are in |
| force. |
| """ |
| name = "completion" |
| |
| def __init__(self, runqueue, rqdata): |
| RunQueueSchedulerSpeed.__init__(self, runqueue, rqdata) |
| |
| #FIXME - whilst this groups all fnids together it does not reorder the |
| #fnid groups optimally. |
| |
| basemap = copy.deepcopy(self.prio_map) |
| self.prio_map = [] |
| while (len(basemap) > 0): |
| entry = basemap.pop(0) |
| self.prio_map.append(entry) |
| fnid = self.rqdata.runq_fnid[entry] |
| todel = [] |
| for entry in basemap: |
| entry_fnid = self.rqdata.runq_fnid[entry] |
| if entry_fnid == fnid: |
| todel.append(basemap.index(entry)) |
| self.prio_map.append(entry) |
| todel.reverse() |
| for idx in todel: |
| del basemap[idx] |
| |
| class RunQueueData: |
| """ |
| BitBake Run Queue implementation |
| """ |
| def __init__(self, rq, cooker, cfgData, dataCache, taskData, targets): |
| self.cooker = cooker |
| self.dataCache = dataCache |
| self.taskData = taskData |
| self.targets = targets |
| self.rq = rq |
| self.warn_multi_bb = False |
| |
| self.stampwhitelist = cfgData.getVar("BB_STAMP_WHITELIST", True) or "" |
| self.multi_provider_whitelist = (cfgData.getVar("MULTI_PROVIDER_WHITELIST", True) or "").split() |
| |
| self.reset() |
| |
| def reset(self): |
| self.runq_fnid = [] |
| self.runq_task = [] |
| self.runq_depends = [] |
| self.runq_revdeps = [] |
| self.runq_hash = [] |
| |
| def runq_depends_names(self, ids): |
| import re |
| ret = [] |
| for id in self.runq_depends[ids]: |
| nam = os.path.basename(self.get_user_idstring(id)) |
| nam = re.sub("_[^,]*,", ",", nam) |
| ret.extend([nam]) |
| return ret |
| |
| def get_task_name(self, task): |
| return self.runq_task[task] |
| |
| def get_task_file(self, task): |
| return self.taskData.fn_index[self.runq_fnid[task]] |
| |
| def get_task_hash(self, task): |
| return self.runq_hash[task] |
| |
| def get_user_idstring(self, task, task_name_suffix = ""): |
| fn = self.taskData.fn_index[self.runq_fnid[task]] |
| taskname = self.runq_task[task] + task_name_suffix |
| return "%s, %s" % (fn, taskname) |
| |
| def get_task_id(self, fnid, taskname): |
| for listid in xrange(len(self.runq_fnid)): |
| if self.runq_fnid[listid] == fnid and self.runq_task[listid] == taskname: |
| return listid |
| return None |
| |
| def circular_depchains_handler(self, tasks): |
| """ |
| Some tasks aren't buildable, likely due to circular dependency issues. |
| Identify the circular dependencies and print them in a user readable format. |
| """ |
| from copy import deepcopy |
| |
| valid_chains = [] |
| explored_deps = {} |
| msgs = [] |
| |
| def chain_reorder(chain): |
| """ |
| Reorder a dependency chain so the lowest task id is first |
| """ |
| lowest = 0 |
| new_chain = [] |
| for entry in xrange(len(chain)): |
| if chain[entry] < chain[lowest]: |
| lowest = entry |
| new_chain.extend(chain[lowest:]) |
| new_chain.extend(chain[:lowest]) |
| return new_chain |
| |
| def chain_compare_equal(chain1, chain2): |
| """ |
| Compare two dependency chains and see if they're the same |
| """ |
| if len(chain1) != len(chain2): |
| return False |
| for index in xrange(len(chain1)): |
| if chain1[index] != chain2[index]: |
| return False |
| return True |
| |
| def chain_array_contains(chain, chain_array): |
| """ |
| Return True if chain_array contains chain |
| """ |
| for ch in chain_array: |
| if chain_compare_equal(ch, chain): |
| return True |
| return False |
| |
| def find_chains(taskid, prev_chain): |
| prev_chain.append(taskid) |
| total_deps = [] |
| total_deps.extend(self.runq_revdeps[taskid]) |
| for revdep in self.runq_revdeps[taskid]: |
| if revdep in prev_chain: |
| idx = prev_chain.index(revdep) |
| # To prevent duplicates, reorder the chain to start with the lowest taskid |
| # and search through an array of those we've already printed |
| chain = prev_chain[idx:] |
| new_chain = chain_reorder(chain) |
| if not chain_array_contains(new_chain, valid_chains): |
| valid_chains.append(new_chain) |
| msgs.append("Dependency loop #%d found:\n" % len(valid_chains)) |
| for dep in new_chain: |
| msgs.append(" Task %s (%s) (dependent Tasks %s)\n" % (dep, self.get_user_idstring(dep), self.runq_depends_names(dep))) |
| msgs.append("\n") |
| if len(valid_chains) > 10: |
| msgs.append("Aborted dependency loops search after 10 matches.\n") |
| return msgs |
| continue |
| scan = False |
| if revdep not in explored_deps: |
| scan = True |
| elif revdep in explored_deps[revdep]: |
| scan = True |
| else: |
| for dep in prev_chain: |
| if dep in explored_deps[revdep]: |
| scan = True |
| if scan: |
| find_chains(revdep, copy.deepcopy(prev_chain)) |
| for dep in explored_deps[revdep]: |
| if dep not in total_deps: |
| total_deps.append(dep) |
| |
| explored_deps[taskid] = total_deps |
| |
| for task in tasks: |
| find_chains(task, []) |
| |
| return msgs |
| |
| def calculate_task_weights(self, endpoints): |
| """ |
| Calculate a number representing the "weight" of each task. Heavier weighted tasks |
| have more dependencies and hence should be executed sooner for maximum speed. |
| |
| This function also sanity checks the task list finding tasks that are not |
| possible to execute due to circular dependencies. |
| """ |
| |
| numTasks = len(self.runq_fnid) |
| weight = [] |
| deps_left = [] |
| task_done = [] |
| |
| for listid in xrange(numTasks): |
| task_done.append(False) |
| weight.append(1) |
| deps_left.append(len(self.runq_revdeps[listid])) |
| |
| for listid in endpoints: |
| weight[listid] = 10 |
| task_done[listid] = True |
| |
| while True: |
| next_points = [] |
| for listid in endpoints: |
| for revdep in self.runq_depends[listid]: |
| weight[revdep] = weight[revdep] + weight[listid] |
| deps_left[revdep] = deps_left[revdep] - 1 |
| if deps_left[revdep] == 0: |
| next_points.append(revdep) |
| task_done[revdep] = True |
| endpoints = next_points |
| if len(next_points) == 0: |
| break |
| |
| # Circular dependency sanity check |
| problem_tasks = [] |
| for task in xrange(numTasks): |
| if task_done[task] is False or deps_left[task] != 0: |
| problem_tasks.append(task) |
| logger.debug(2, "Task %s (%s) is not buildable", task, self.get_user_idstring(task)) |
| logger.debug(2, "(Complete marker was %s and the remaining dependency count was %s)\n", task_done[task], deps_left[task]) |
| |
| if problem_tasks: |
| message = "Unbuildable tasks were found.\n" |
| message = message + "These are usually caused by circular dependencies and any circular dependency chains found will be printed below. Increase the debug level to see a list of unbuildable tasks.\n\n" |
| message = message + "Identifying dependency loops (this may take a short while)...\n" |
| logger.error(message) |
| |
| msgs = self.circular_depchains_handler(problem_tasks) |
| |
| message = "\n" |
| for msg in msgs: |
| message = message + msg |
| bb.msg.fatal("RunQueue", message) |
| |
| return weight |
| |
| def prepare(self): |
| """ |
| Turn a set of taskData into a RunQueue and compute data needed |
| to optimise the execution order. |
| """ |
| |
| runq_build = [] |
| recursivetasks = {} |
| recursiveitasks = {} |
| recursivetasksselfref = set() |
| |
| taskData = self.taskData |
| |
| if len(taskData.tasks_name) == 0: |
| # Nothing to do |
| return 0 |
| |
| logger.info("Preparing RunQueue") |
| |
| # Step A - Work out a list of tasks to run |
| # |
| # Taskdata gives us a list of possible providers for every build and run |
| # target ordered by priority. It also gives information on each of those |
| # providers. |
| # |
| # To create the actual list of tasks to execute we fix the list of |
| # providers and then resolve the dependencies into task IDs. This |
| # process is repeated for each type of dependency (tdepends, deptask, |
| # rdeptast, recrdeptask, idepends). |
| |
| def add_build_dependencies(depids, tasknames, depends): |
| for depid in depids: |
| # Won't be in build_targets if ASSUME_PROVIDED |
| if depid not in taskData.build_targets: |
| continue |
| depdata = taskData.build_targets[depid][0] |
| if depdata is None: |
| continue |
| for taskname in tasknames: |
| taskid = taskData.gettask_id_fromfnid(depdata, taskname) |
| if taskid is not None: |
| depends.add(taskid) |
| |
| def add_runtime_dependencies(depids, tasknames, depends): |
| for depid in depids: |
| if depid not in taskData.run_targets: |
| continue |
| depdata = taskData.run_targets[depid][0] |
| if depdata is None: |
| continue |
| for taskname in tasknames: |
| taskid = taskData.gettask_id_fromfnid(depdata, taskname) |
| if taskid is not None: |
| depends.add(taskid) |
| |
| def add_resolved_dependencies(depids, tasknames, depends): |
| for depid in depids: |
| for taskname in tasknames: |
| taskid = taskData.gettask_id_fromfnid(depid, taskname) |
| if taskid is not None: |
| depends.add(taskid) |
| |
| for task in xrange(len(taskData.tasks_name)): |
| depends = set() |
| fnid = taskData.tasks_fnid[task] |
| fn = taskData.fn_index[fnid] |
| task_deps = self.dataCache.task_deps[fn] |
| |
| #logger.debug(2, "Processing %s:%s", fn, taskData.tasks_name[task]) |
| |
| if fnid not in taskData.failed_fnids: |
| |
| # Resolve task internal dependencies |
| # |
| # e.g. addtask before X after Y |
| depends = set(taskData.tasks_tdepends[task]) |
| |
| # Resolve 'deptask' dependencies |
| # |
| # e.g. do_sometask[deptask] = "do_someothertask" |
| # (makes sure sometask runs after someothertask of all DEPENDS) |
| if 'deptask' in task_deps and taskData.tasks_name[task] in task_deps['deptask']: |
| tasknames = task_deps['deptask'][taskData.tasks_name[task]].split() |
| add_build_dependencies(taskData.depids[fnid], tasknames, depends) |
| |
| # Resolve 'rdeptask' dependencies |
| # |
| # e.g. do_sometask[rdeptask] = "do_someothertask" |
| # (makes sure sometask runs after someothertask of all RDEPENDS) |
| if 'rdeptask' in task_deps and taskData.tasks_name[task] in task_deps['rdeptask']: |
| tasknames = task_deps['rdeptask'][taskData.tasks_name[task]].split() |
| add_runtime_dependencies(taskData.rdepids[fnid], tasknames, depends) |
| |
| # Resolve inter-task dependencies |
| # |
| # e.g. do_sometask[depends] = "targetname:do_someothertask" |
| # (makes sure sometask runs after targetname's someothertask) |
| idepends = taskData.tasks_idepends[task] |
| for (depid, idependtask) in idepends: |
| if depid in taskData.build_targets and not depid in taskData.failed_deps: |
| # Won't be in build_targets if ASSUME_PROVIDED |
| depdata = taskData.build_targets[depid][0] |
| if depdata is not None: |
| taskid = taskData.gettask_id_fromfnid(depdata, idependtask) |
| if taskid is None: |
| bb.msg.fatal("RunQueue", "Task %s in %s depends upon non-existent task %s in %s" % (taskData.tasks_name[task], fn, idependtask, taskData.fn_index[depdata])) |
| depends.add(taskid) |
| irdepends = taskData.tasks_irdepends[task] |
| for (depid, idependtask) in irdepends: |
| if depid in taskData.run_targets: |
| # Won't be in run_targets if ASSUME_PROVIDED |
| depdata = taskData.run_targets[depid][0] |
| if depdata is not None: |
| taskid = taskData.gettask_id_fromfnid(depdata, idependtask) |
| if taskid is None: |
| bb.msg.fatal("RunQueue", "Task %s in %s rdepends upon non-existent task %s in %s" % (taskData.tasks_name[task], fn, idependtask, taskData.fn_index[depdata])) |
| depends.add(taskid) |
| |
| # Resolve recursive 'recrdeptask' dependencies (Part A) |
| # |
| # e.g. do_sometask[recrdeptask] = "do_someothertask" |
| # (makes sure sometask runs after someothertask of all DEPENDS, RDEPENDS and intertask dependencies, recursively) |
| # We cover the recursive part of the dependencies below |
| if 'recrdeptask' in task_deps and taskData.tasks_name[task] in task_deps['recrdeptask']: |
| tasknames = task_deps['recrdeptask'][taskData.tasks_name[task]].split() |
| recursivetasks[task] = tasknames |
| add_build_dependencies(taskData.depids[fnid], tasknames, depends) |
| add_runtime_dependencies(taskData.rdepids[fnid], tasknames, depends) |
| if taskData.tasks_name[task] in tasknames: |
| recursivetasksselfref.add(task) |
| |
| if 'recideptask' in task_deps and taskData.tasks_name[task] in task_deps['recideptask']: |
| recursiveitasks[task] = [] |
| for t in task_deps['recideptask'][taskData.tasks_name[task]].split(): |
| newdep = taskData.gettask_id_fromfnid(fnid, t) |
| recursiveitasks[task].append(newdep) |
| |
| self.runq_fnid.append(taskData.tasks_fnid[task]) |
| self.runq_task.append(taskData.tasks_name[task]) |
| self.runq_depends.append(depends) |
| self.runq_revdeps.append(set()) |
| self.runq_hash.append("") |
| |
| runq_build.append(0) |
| |
| # Resolve recursive 'recrdeptask' dependencies (Part B) |
| # |
| # e.g. do_sometask[recrdeptask] = "do_someothertask" |
| # (makes sure sometask runs after someothertask of all DEPENDS, RDEPENDS and intertask dependencies, recursively) |
| # We need to do this separately since we need all of self.runq_depends to be complete before this is processed |
| extradeps = {} |
| for task in recursivetasks: |
| extradeps[task] = set(self.runq_depends[task]) |
| tasknames = recursivetasks[task] |
| seendeps = set() |
| seenfnid = [] |
| |
| def generate_recdeps(t): |
| newdeps = set() |
| add_resolved_dependencies([taskData.tasks_fnid[t]], tasknames, newdeps) |
| extradeps[task].update(newdeps) |
| seendeps.add(t) |
| newdeps.add(t) |
| for i in newdeps: |
| for n in self.runq_depends[i]: |
| if n not in seendeps: |
| generate_recdeps(n) |
| generate_recdeps(task) |
| |
| if task in recursiveitasks: |
| for dep in recursiveitasks[task]: |
| generate_recdeps(dep) |
| |
| # Remove circular references so that do_a[recrdeptask] = "do_a do_b" can work |
| for task in recursivetasks: |
| extradeps[task].difference_update(recursivetasksselfref) |
| |
| for task in xrange(len(taskData.tasks_name)): |
| # Add in extra dependencies |
| if task in extradeps: |
| self.runq_depends[task] = extradeps[task] |
| # Remove all self references |
| if task in self.runq_depends[task]: |
| logger.debug(2, "Task %s (%s %s) contains self reference! %s", task, taskData.fn_index[taskData.tasks_fnid[task]], taskData.tasks_name[task], self.runq_depends[task]) |
| self.runq_depends[task].remove(task) |
| |
| # Step B - Mark all active tasks |
| # |
| # Start with the tasks we were asked to run and mark all dependencies |
| # as active too. If the task is to be 'forced', clear its stamp. Once |
| # all active tasks are marked, prune the ones we don't need. |
| |
| logger.verbose("Marking Active Tasks") |
| |
| def mark_active(listid, depth): |
| """ |
| Mark an item as active along with its depends |
| (calls itself recursively) |
| """ |
| |
| if runq_build[listid] == 1: |
| return |
| |
| runq_build[listid] = 1 |
| |
| depends = self.runq_depends[listid] |
| for depend in depends: |
| mark_active(depend, depth+1) |
| |
| self.target_pairs = [] |
| for target in self.targets: |
| targetid = taskData.getbuild_id(target[0]) |
| |
| if targetid not in taskData.build_targets: |
| continue |
| |
| if targetid in taskData.failed_deps: |
| continue |
| |
| fnid = taskData.build_targets[targetid][0] |
| fn = taskData.fn_index[fnid] |
| self.target_pairs.append((fn, target[1])) |
| |
| if fnid in taskData.failed_fnids: |
| continue |
| |
| if target[1] not in taskData.tasks_lookup[fnid]: |
| import difflib |
| close_matches = difflib.get_close_matches(target[1], taskData.tasks_lookup[fnid], cutoff=0.7) |
| if close_matches: |
| extra = ". Close matches:\n %s" % "\n ".join(close_matches) |
| else: |
| extra = "" |
| bb.msg.fatal("RunQueue", "Task %s does not exist for target %s%s" % (target[1], target[0], extra)) |
| |
| listid = taskData.tasks_lookup[fnid][target[1]] |
| |
| mark_active(listid, 1) |
| |
| # Step C - Prune all inactive tasks |
| # |
| # Once all active tasks are marked, prune the ones we don't need. |
| |
| maps = [] |
| delcount = 0 |
| for listid in xrange(len(self.runq_fnid)): |
| if runq_build[listid-delcount] == 1: |
| maps.append(listid-delcount) |
| else: |
| del self.runq_fnid[listid-delcount] |
| del self.runq_task[listid-delcount] |
| del self.runq_depends[listid-delcount] |
| del runq_build[listid-delcount] |
| del self.runq_revdeps[listid-delcount] |
| del self.runq_hash[listid-delcount] |
| delcount = delcount + 1 |
| maps.append(-1) |
| |
| # |
| # Step D - Sanity checks and computation |
| # |
| |
| # Check to make sure we still have tasks to run |
| if len(self.runq_fnid) == 0: |
| if not taskData.abort: |
| bb.msg.fatal("RunQueue", "All buildable tasks have been run but the build is incomplete (--continue mode). Errors for the tasks that failed will have been printed above.") |
| else: |
| bb.msg.fatal("RunQueue", "No active tasks and not in --continue mode?! Please report this bug.") |
| |
| logger.verbose("Pruned %s inactive tasks, %s left", delcount, len(self.runq_fnid)) |
| |
| # Remap the dependencies to account for the deleted tasks |
| # Check we didn't delete a task we depend on |
| for listid in xrange(len(self.runq_fnid)): |
| newdeps = [] |
| origdeps = self.runq_depends[listid] |
| for origdep in origdeps: |
| if maps[origdep] == -1: |
| bb.msg.fatal("RunQueue", "Invalid mapping - Should never happen!") |
| newdeps.append(maps[origdep]) |
| self.runq_depends[listid] = set(newdeps) |
| |
| logger.verbose("Assign Weightings") |
| |
| # Generate a list of reverse dependencies to ease future calculations |
| for listid in xrange(len(self.runq_fnid)): |
| for dep in self.runq_depends[listid]: |
| self.runq_revdeps[dep].add(listid) |
| |
| # Identify tasks at the end of dependency chains |
| # Error on circular dependency loops (length two) |
| endpoints = [] |
| for listid in xrange(len(self.runq_fnid)): |
| revdeps = self.runq_revdeps[listid] |
| if len(revdeps) == 0: |
| endpoints.append(listid) |
| for dep in revdeps: |
| if dep in self.runq_depends[listid]: |
| #self.dump_data(taskData) |
| bb.msg.fatal("RunQueue", "Task %s (%s) has circular dependency on %s (%s)" % (taskData.fn_index[self.runq_fnid[dep]], self.runq_task[dep], taskData.fn_index[self.runq_fnid[listid]], self.runq_task[listid])) |
| |
| logger.verbose("Compute totals (have %s endpoint(s))", len(endpoints)) |
| |
| # Calculate task weights |
| # Check of higher length circular dependencies |
| self.runq_weight = self.calculate_task_weights(endpoints) |
| |
| # Sanity Check - Check for multiple tasks building the same provider |
| prov_list = {} |
| seen_fn = [] |
| for task in xrange(len(self.runq_fnid)): |
| fn = taskData.fn_index[self.runq_fnid[task]] |
| if fn in seen_fn: |
| continue |
| seen_fn.append(fn) |
| for prov in self.dataCache.fn_provides[fn]: |
| if prov not in prov_list: |
| prov_list[prov] = [fn] |
| elif fn not in prov_list[prov]: |
| prov_list[prov].append(fn) |
| for prov in prov_list: |
| if len(prov_list[prov]) > 1 and prov not in self.multi_provider_whitelist: |
| seen_pn = [] |
| # If two versions of the same PN are being built its fatal, we don't support it. |
| for fn in prov_list[prov]: |
| pn = self.dataCache.pkg_fn[fn] |
| if pn not in seen_pn: |
| seen_pn.append(pn) |
| else: |
| bb.fatal("Multiple versions of %s are due to be built (%s). Only one version of a given PN should be built in any given build. You likely need to set PREFERRED_VERSION_%s to select the correct version or don't depend on multiple versions." % (pn, " ".join(prov_list[prov]), pn)) |
| msg = "Multiple .bb files are due to be built which each provide %s (%s)." % (prov, " ".join(prov_list[prov])) |
| if self.warn_multi_bb: |
| logger.warn(msg) |
| else: |
| msg += "\n This usually means one provides something the other doesn't and should." |
| logger.error(msg) |
| |
| # Create a whitelist usable by the stamp checks |
| stampfnwhitelist = [] |
| for entry in self.stampwhitelist.split(): |
| entryid = self.taskData.getbuild_id(entry) |
| if entryid not in self.taskData.build_targets: |
| continue |
| fnid = self.taskData.build_targets[entryid][0] |
| fn = self.taskData.fn_index[fnid] |
| stampfnwhitelist.append(fn) |
| self.stampfnwhitelist = stampfnwhitelist |
| |
| # Iterate over the task list looking for tasks with a 'setscene' function |
| self.runq_setscene = [] |
| if not self.cooker.configuration.nosetscene: |
| for task in range(len(self.runq_fnid)): |
| setscene = taskData.gettask_id(self.taskData.fn_index[self.runq_fnid[task]], self.runq_task[task] + "_setscene", False) |
| if not setscene: |
| continue |
| self.runq_setscene.append(task) |
| |
| def invalidate_task(fn, taskname, error_nostamp): |
| taskdep = self.dataCache.task_deps[fn] |
| fnid = self.taskData.getfn_id(fn) |
| if taskname not in taskData.tasks_lookup[fnid]: |
| logger.warn("Task %s does not exist, invalidating this task will have no effect" % taskname) |
| if 'nostamp' in taskdep and taskname in taskdep['nostamp']: |
| if error_nostamp: |
| bb.fatal("Task %s is marked nostamp, cannot invalidate this task" % taskname) |
| else: |
| bb.debug(1, "Task %s is marked nostamp, cannot invalidate this task" % taskname) |
| else: |
| logger.verbose("Invalidate task %s, %s", taskname, fn) |
| bb.parse.siggen.invalidate_task(taskname, self.dataCache, fn) |
| |
| # Invalidate task if force mode active |
| if self.cooker.configuration.force: |
| for (fn, target) in self.target_pairs: |
| invalidate_task(fn, target, False) |
| |
| # Invalidate task if invalidate mode active |
| if self.cooker.configuration.invalidate_stamp: |
| for (fn, target) in self.target_pairs: |
| for st in self.cooker.configuration.invalidate_stamp.split(','): |
| if not st.startswith("do_"): |
| st = "do_%s" % st |
| invalidate_task(fn, st, True) |
| |
| # Create and print to the logs a virtual/xxxx -> PN (fn) table |
| virtmap = taskData.get_providermap() |
| virtpnmap = {} |
| for v in virtmap: |
| virtpnmap[v] = self.dataCache.pkg_fn[virtmap[v]] |
| bb.debug(2, "%s resolved to: %s (%s)" % (v, virtpnmap[v], virtmap[v])) |
| if hasattr(bb.parse.siggen, "tasks_resolved"): |
| bb.parse.siggen.tasks_resolved(virtmap, virtpnmap, self.dataCache) |
| |
| # Iterate over the task list and call into the siggen code |
| dealtwith = set() |
| todeal = set(range(len(self.runq_fnid))) |
| while len(todeal) > 0: |
| for task in todeal.copy(): |
| if len(self.runq_depends[task] - dealtwith) == 0: |
| dealtwith.add(task) |
| todeal.remove(task) |
| procdep = [] |
| for dep in self.runq_depends[task]: |
| procdep.append(self.taskData.fn_index[self.runq_fnid[dep]] + "." + self.runq_task[dep]) |
| self.runq_hash[task] = bb.parse.siggen.get_taskhash(self.taskData.fn_index[self.runq_fnid[task]], self.runq_task[task], procdep, self.dataCache) |
| |
| return len(self.runq_fnid) |
| |
| def dump_data(self, taskQueue): |
| """ |
| Dump some debug information on the internal data structures |
| """ |
| logger.debug(3, "run_tasks:") |
| for task in xrange(len(self.rqdata.runq_task)): |
| logger.debug(3, " (%s)%s - %s: %s Deps %s RevDeps %s", task, |
| taskQueue.fn_index[self.rqdata.runq_fnid[task]], |
| self.rqdata.runq_task[task], |
| self.rqdata.runq_weight[task], |
| self.rqdata.runq_depends[task], |
| self.rqdata.runq_revdeps[task]) |
| |
| logger.debug(3, "sorted_tasks:") |
| for task1 in xrange(len(self.rqdata.runq_task)): |
| if task1 in self.prio_map: |
| task = self.prio_map[task1] |
| logger.debug(3, " (%s)%s - %s: %s Deps %s RevDeps %s", task, |
| taskQueue.fn_index[self.rqdata.runq_fnid[task]], |
| self.rqdata.runq_task[task], |
| self.rqdata.runq_weight[task], |
| self.rqdata.runq_depends[task], |
| self.rqdata.runq_revdeps[task]) |
| |
| class RunQueue: |
| def __init__(self, cooker, cfgData, dataCache, taskData, targets): |
| |
| self.cooker = cooker |
| self.cfgData = cfgData |
| self.rqdata = RunQueueData(self, cooker, cfgData, dataCache, taskData, targets) |
| |
| self.stamppolicy = cfgData.getVar("BB_STAMP_POLICY", True) or "perfile" |
| self.hashvalidate = cfgData.getVar("BB_HASHCHECK_FUNCTION", True) or None |
| self.setsceneverify = cfgData.getVar("BB_SETSCENE_VERIFY_FUNCTION", True) or None |
| self.depvalidate = cfgData.getVar("BB_SETSCENE_DEPVALID", True) or None |
| |
| self.state = runQueuePrepare |
| |
| # For disk space monitor |
| self.dm = monitordisk.diskMonitor(cfgData) |
| |
| self.rqexe = None |
| self.worker = None |
| self.workerpipe = None |
| self.fakeworker = None |
| self.fakeworkerpipe = None |
| |
| def _start_worker(self, fakeroot = False, rqexec = None): |
| logger.debug(1, "Starting bitbake-worker") |
| magic = "decafbad" |
| if self.cooker.configuration.profile: |
| magic = "decafbadbad" |
| if fakeroot: |
| fakerootcmd = self.cfgData.getVar("FAKEROOTCMD", True) |
| fakerootenv = (self.cfgData.getVar("FAKEROOTBASEENV", True) or "").split() |
| env = os.environ.copy() |
| for key, value in (var.split('=') for var in fakerootenv): |
| env[key] = value |
| worker = subprocess.Popen([fakerootcmd, "bitbake-worker", magic], stdout=subprocess.PIPE, stdin=subprocess.PIPE, env=env) |
| else: |
| worker = subprocess.Popen(["bitbake-worker", magic], stdout=subprocess.PIPE, stdin=subprocess.PIPE) |
| bb.utils.nonblockingfd(worker.stdout) |
| workerpipe = runQueuePipe(worker.stdout, None, self.cfgData, self, rqexec) |
| |
| workerdata = { |
| "taskdeps" : self.rqdata.dataCache.task_deps, |
| "fakerootenv" : self.rqdata.dataCache.fakerootenv, |
| "fakerootdirs" : self.rqdata.dataCache.fakerootdirs, |
| "fakerootnoenv" : self.rqdata.dataCache.fakerootnoenv, |
| "sigdata" : bb.parse.siggen.get_taskdata(), |
| "runq_hash" : self.rqdata.runq_hash, |
| "logdefaultdebug" : bb.msg.loggerDefaultDebugLevel, |
| "logdefaultverbose" : bb.msg.loggerDefaultVerbose, |
| "logdefaultverboselogs" : bb.msg.loggerVerboseLogs, |
| "logdefaultdomain" : bb.msg.loggerDefaultDomains, |
| "prhost" : self.cooker.prhost, |
| "buildname" : self.cfgData.getVar("BUILDNAME", True), |
| "date" : self.cfgData.getVar("DATE", True), |
| "time" : self.cfgData.getVar("TIME", True), |
| } |
| |
| worker.stdin.write("<cookerconfig>" + pickle.dumps(self.cooker.configuration) + "</cookerconfig>") |
| worker.stdin.write("<workerdata>" + pickle.dumps(workerdata) + "</workerdata>") |
| worker.stdin.flush() |
| |
| return worker, workerpipe |
| |
| def _teardown_worker(self, worker, workerpipe): |
| if not worker: |
| return |
| logger.debug(1, "Teardown for bitbake-worker") |
| try: |
| worker.stdin.write("<quit></quit>") |
| worker.stdin.flush() |
| except IOError: |
| pass |
| while worker.returncode is None: |
| workerpipe.read() |
| worker.poll() |
| while workerpipe.read(): |
| continue |
| workerpipe.close() |
| |
| def start_worker(self): |
| if self.worker: |
| self.teardown_workers() |
| self.teardown = False |
| self.worker, self.workerpipe = self._start_worker() |
| |
| def start_fakeworker(self, rqexec): |
| if not self.fakeworker: |
| self.fakeworker, self.fakeworkerpipe = self._start_worker(True, rqexec) |
| |
| def teardown_workers(self): |
| self.teardown = True |
| self._teardown_worker(self.worker, self.workerpipe) |
| self.worker = None |
| self.workerpipe = None |
| self._teardown_worker(self.fakeworker, self.fakeworkerpipe) |
| self.fakeworker = None |
| self.fakeworkerpipe = None |
| |
| def read_workers(self): |
| self.workerpipe.read() |
| if self.fakeworkerpipe: |
| self.fakeworkerpipe.read() |
| |
| def active_fds(self): |
| fds = [] |
| if self.workerpipe: |
| fds.append(self.workerpipe.input) |
| if self.fakeworkerpipe: |
| fds.append(self.fakeworkerpipe.input) |
| return fds |
| |
| def check_stamp_task(self, task, taskname = None, recurse = False, cache = None): |
| def get_timestamp(f): |
| try: |
| if not os.access(f, os.F_OK): |
| return None |
| return os.stat(f)[stat.ST_MTIME] |
| except: |
| return None |
| |
| if self.stamppolicy == "perfile": |
| fulldeptree = False |
| else: |
| fulldeptree = True |
| stampwhitelist = [] |
| if self.stamppolicy == "whitelist": |
| stampwhitelist = self.rqdata.stampfnwhitelist |
| |
| fn = self.rqdata.taskData.fn_index[self.rqdata.runq_fnid[task]] |
| if taskname is None: |
| taskname = self.rqdata.runq_task[task] |
| |
| stampfile = bb.build.stampfile(taskname, self.rqdata.dataCache, fn) |
| |
| # If the stamp is missing, it's not current |
| if not os.access(stampfile, os.F_OK): |
| logger.debug(2, "Stampfile %s not available", stampfile) |
| return False |
| # If it's a 'nostamp' task, it's not current |
| taskdep = self.rqdata.dataCache.task_deps[fn] |
| if 'nostamp' in taskdep and taskname in taskdep['nostamp']: |
| logger.debug(2, "%s.%s is nostamp\n", fn, taskname) |
| return False |
| |
| if taskname != "do_setscene" and taskname.endswith("_setscene"): |
| return True |
| |
| if cache is None: |
| cache = {} |
| |
| iscurrent = True |
| t1 = get_timestamp(stampfile) |
| for dep in self.rqdata.runq_depends[task]: |
| if iscurrent: |
| fn2 = self.rqdata.taskData.fn_index[self.rqdata.runq_fnid[dep]] |
| taskname2 = self.rqdata.runq_task[dep] |
| stampfile2 = bb.build.stampfile(taskname2, self.rqdata.dataCache, fn2) |
| stampfile3 = bb.build.stampfile(taskname2 + "_setscene", self.rqdata.dataCache, fn2) |
| t2 = get_timestamp(stampfile2) |
| t3 = get_timestamp(stampfile3) |
| if t3 and t3 > t2: |
| continue |
| if fn == fn2 or (fulldeptree and fn2 not in stampwhitelist): |
| if not t2: |
| logger.debug(2, 'Stampfile %s does not exist', stampfile2) |
| iscurrent = False |
| if t1 < t2: |
| logger.debug(2, 'Stampfile %s < %s', stampfile, stampfile2) |
| iscurrent = False |
| if recurse and iscurrent: |
| if dep in cache: |
| iscurrent = cache[dep] |
| if not iscurrent: |
| logger.debug(2, 'Stampfile for dependency %s:%s invalid (cached)' % (fn2, taskname2)) |
| else: |
| iscurrent = self.check_stamp_task(dep, recurse=True, cache=cache) |
| cache[dep] = iscurrent |
| if recurse: |
| cache[task] = iscurrent |
| return iscurrent |
| |
| def _execute_runqueue(self): |
| """ |
| Run the tasks in a queue prepared by rqdata.prepare() |
| Upon failure, optionally try to recover the build using any alternate providers |
| (if the abort on failure configuration option isn't set) |
| """ |
| |
| retval = True |
| |
| if self.state is runQueuePrepare: |
| self.rqexe = RunQueueExecuteDummy(self) |
| if self.rqdata.prepare() == 0: |
| self.state = runQueueComplete |
| else: |
| self.state = runQueueSceneInit |
| |
| # we are ready to run, see if any UI client needs the dependency info |
| if bb.cooker.CookerFeatures.SEND_DEPENDS_TREE in self.cooker.featureset: |
| depgraph = self.cooker.buildDependTree(self, self.rqdata.taskData) |
| bb.event.fire(bb.event.DepTreeGenerated(depgraph), self.cooker.data) |
| |
| if self.state is runQueueSceneInit: |
| dump = self.cooker.configuration.dump_signatures |
| if dump: |
| if 'printdiff' in dump: |
| invalidtasks = self.print_diffscenetasks() |
| self.dump_signatures(dump) |
| if 'printdiff' in dump: |
| self.write_diffscenetasks(invalidtasks) |
| self.state = runQueueComplete |
| else: |
| self.start_worker() |
| self.rqexe = RunQueueExecuteScenequeue(self) |
| |
| if self.state in [runQueueSceneRun, runQueueRunning, runQueueCleanUp]: |
| self.dm.check(self) |
| |
| if self.state is runQueueSceneRun: |
| retval = self.rqexe.execute() |
| |
| if self.state is runQueueRunInit: |
| logger.info("Executing RunQueue Tasks") |
| self.rqexe = RunQueueExecuteTasks(self) |
| self.state = runQueueRunning |
| |
| if self.state is runQueueRunning: |
| retval = self.rqexe.execute() |
| |
| if self.state is runQueueCleanUp: |
| retval = self.rqexe.finish() |
| |
| if (self.state is runQueueComplete or self.state is runQueueFailed) and self.rqexe: |
| self.teardown_workers() |
| if self.rqexe.stats.failed: |
| logger.info("Tasks Summary: Attempted %d tasks of which %d didn't need to be rerun and %d failed.", self.rqexe.stats.completed + self.rqexe.stats.failed, self.rqexe.stats.skipped, self.rqexe.stats.failed) |
| else: |
| # Let's avoid the word "failed" if nothing actually did |
| logger.info("Tasks Summary: Attempted %d tasks of which %d didn't need to be rerun and all succeeded.", self.rqexe.stats.completed, self.rqexe.stats.skipped) |
| |
| if self.state is runQueueFailed: |
| if not self.rqdata.taskData.tryaltconfigs: |
| raise bb.runqueue.TaskFailure(self.rqexe.failed_fnids) |
| for fnid in self.rqexe.failed_fnids: |
| self.rqdata.taskData.fail_fnid(fnid) |
| self.rqdata.reset() |
| |
| if self.state is runQueueComplete: |
| # All done |
| return False |
| |
| # Loop |
| return retval |
| |
| def execute_runqueue(self): |
| # Catch unexpected exceptions and ensure we exit when an error occurs, not loop. |
| try: |
| return self._execute_runqueue() |
| except bb.runqueue.TaskFailure: |
| raise |
| except SystemExit: |
| raise |
| except bb.BBHandledException: |
| try: |
| self.teardown_workers() |
| except: |
| pass |
| self.state = runQueueComplete |
| raise |
| except: |
| logger.error("An uncaught exception occured in runqueue, please see the failure below:") |
| try: |
| self.teardown_workers() |
| except: |
| pass |
| self.state = runQueueComplete |
| raise |
| |
| def finish_runqueue(self, now = False): |
| if not self.rqexe: |
| self.state = runQueueComplete |
| return |
| |
| if now: |
| self.rqexe.finish_now() |
| else: |
| self.rqexe.finish() |
| |
| def dump_signatures(self, options): |
| done = set() |
| bb.note("Reparsing files to collect dependency data") |
| for task in range(len(self.rqdata.runq_fnid)): |
| if self.rqdata.runq_fnid[task] not in done: |
| fn = self.rqdata.taskData.fn_index[self.rqdata.runq_fnid[task]] |
| the_data = bb.cache.Cache.loadDataFull(fn, self.cooker.collection.get_file_appends(fn), self.cooker.data) |
| done.add(self.rqdata.runq_fnid[task]) |
| |
| bb.parse.siggen.dump_sigs(self.rqdata.dataCache, options) |
| |
| return |
| |
| def print_diffscenetasks(self): |
| |
| valid = [] |
| sq_hash = [] |
| sq_hashfn = [] |
| sq_fn = [] |
| sq_taskname = [] |
| sq_task = [] |
| noexec = [] |
| stamppresent = [] |
| valid_new = set() |
| |
| for task in xrange(len(self.rqdata.runq_fnid)): |
| fn = self.rqdata.taskData.fn_index[self.rqdata.runq_fnid[task]] |
| taskname = self.rqdata.runq_task[task] |
| taskdep = self.rqdata.dataCache.task_deps[fn] |
| |
| if 'noexec' in taskdep and taskname in taskdep['noexec']: |
| noexec.append(task) |
| continue |
| |
| sq_fn.append(fn) |
| sq_hashfn.append(self.rqdata.dataCache.hashfn[fn]) |
| sq_hash.append(self.rqdata.runq_hash[task]) |
| sq_taskname.append(taskname) |
| sq_task.append(task) |
| locs = { "sq_fn" : sq_fn, "sq_task" : sq_taskname, "sq_hash" : sq_hash, "sq_hashfn" : sq_hashfn, "d" : self.cooker.expanded_data } |
| try: |
| call = self.hashvalidate + "(sq_fn, sq_task, sq_hash, sq_hashfn, d, siginfo=True)" |
| valid = bb.utils.better_eval(call, locs) |
| # Handle version with no siginfo parameter |
| except TypeError: |
| call = self.hashvalidate + "(sq_fn, sq_task, sq_hash, sq_hashfn, d)" |
| valid = bb.utils.better_eval(call, locs) |
| for v in valid: |
| valid_new.add(sq_task[v]) |
| |
| # Tasks which are both setscene and noexec never care about dependencies |
| # We therefore find tasks which are setscene and noexec and mark their |
| # unique dependencies as valid. |
| for task in noexec: |
| if task not in self.rqdata.runq_setscene: |
| continue |
| for dep in self.rqdata.runq_depends[task]: |
| hasnoexecparents = True |
| for dep2 in self.rqdata.runq_revdeps[dep]: |
| if dep2 in self.rqdata.runq_setscene and dep2 in noexec: |
| continue |
| hasnoexecparents = False |
| break |
| if hasnoexecparents: |
| valid_new.add(dep) |
| |
| invalidtasks = set() |
| for task in xrange(len(self.rqdata.runq_fnid)): |
| if task not in valid_new and task not in noexec: |
| invalidtasks.add(task) |
| |
| found = set() |
| processed = set() |
| for task in invalidtasks: |
| toprocess = set([task]) |
| while toprocess: |
| next = set() |
| for t in toprocess: |
| for dep in self.rqdata.runq_depends[t]: |
| if dep in invalidtasks: |
| found.add(task) |
| if dep not in processed: |
| processed.add(dep) |
| next.add(dep) |
| toprocess = next |
| if task in found: |
| toprocess = set() |
| |
| tasklist = [] |
| for task in invalidtasks.difference(found): |
| tasklist.append(self.rqdata.get_user_idstring(task)) |
| |
| if tasklist: |
| bb.plain("The differences between the current build and any cached tasks start at the following tasks:\n" + "\n".join(tasklist)) |
| |
| return invalidtasks.difference(found) |
| |
| def write_diffscenetasks(self, invalidtasks): |
| |
| # Define recursion callback |
| def recursecb(key, hash1, hash2): |
| hashes = [hash1, hash2] |
| hashfiles = bb.siggen.find_siginfo(key, None, hashes, self.cfgData) |
| |
| recout = [] |
| if len(hashfiles) == 2: |
| out2 = bb.siggen.compare_sigfiles(hashfiles[hash1], hashfiles[hash2], recursecb) |
| recout.extend(list(' ' + l for l in out2)) |
| else: |
| recout.append("Unable to find matching sigdata for %s with hashes %s or %s" % (key, hash1, hash2)) |
| |
| return recout |
| |
| |
| for task in invalidtasks: |
| fn = self.rqdata.taskData.fn_index[self.rqdata.runq_fnid[task]] |
| pn = self.rqdata.dataCache.pkg_fn[fn] |
| taskname = self.rqdata.runq_task[task] |
| h = self.rqdata.runq_hash[task] |
| matches = bb.siggen.find_siginfo(pn, taskname, [], self.cfgData) |
| match = None |
| for m in matches: |
| if h in m: |
| match = m |
| if match is None: |
| bb.fatal("Can't find a task we're supposed to have written out? (hash: %s)?" % h) |
| matches = {k : v for k, v in matches.iteritems() if h not in k} |
| if matches: |
| latestmatch = sorted(matches.keys(), key=lambda f: matches[f])[-1] |
| prevh = __find_md5__.search(latestmatch).group(0) |
| output = bb.siggen.compare_sigfiles(latestmatch, match, recursecb) |
| bb.plain("\nTask %s:%s couldn't be used from the cache because:\n We need hash %s, closest matching task was %s\n " % (pn, taskname, h, prevh) + '\n '.join(output)) |
| |
| class RunQueueExecute: |
| |
| def __init__(self, rq): |
| self.rq = rq |
| self.cooker = rq.cooker |
| self.cfgData = rq.cfgData |
| self.rqdata = rq.rqdata |
| |
| self.number_tasks = int(self.cfgData.getVar("BB_NUMBER_THREADS", True) or 1) |
| self.scheduler = self.cfgData.getVar("BB_SCHEDULER", True) or "speed" |
| |
| self.runq_buildable = [] |
| self.runq_running = [] |
| self.runq_complete = [] |
| |
| self.build_stamps = {} |
| self.build_stamps2 = [] |
| self.failed_fnids = [] |
| |
| self.stampcache = {} |
| |
| rq.workerpipe.setrunqueueexec(self) |
| if rq.fakeworkerpipe: |
| rq.fakeworkerpipe.setrunqueueexec(self) |
| |
| if self.number_tasks <= 0: |
| bb.fatal("Invalid BB_NUMBER_THREADS %s" % self.number_tasks) |
| |
| def runqueue_process_waitpid(self, task, status): |
| |
| # self.build_stamps[pid] may not exist when use shared work directory. |
| if task in self.build_stamps: |
| self.build_stamps2.remove(self.build_stamps[task]) |
| del self.build_stamps[task] |
| |
| if status != 0: |
| self.task_fail(task, status) |
| else: |
| self.task_complete(task) |
| return True |
| |
| def finish_now(self): |
| |
| for worker in [self.rq.worker, self.rq.fakeworker]: |
| if not worker: |
| continue |
| try: |
| worker.stdin.write("<finishnow></finishnow>") |
| worker.stdin.flush() |
| except IOError: |
| # worker must have died? |
| pass |
| |
| if len(self.failed_fnids) != 0: |
| self.rq.state = runQueueFailed |
| return |
| |
| self.rq.state = runQueueComplete |
| return |
| |
| def finish(self): |
| self.rq.state = runQueueCleanUp |
| |
| if self.stats.active > 0: |
| bb.event.fire(runQueueExitWait(self.stats.active), self.cfgData) |
| self.rq.read_workers() |
| return self.rq.active_fds() |
| |
| if len(self.failed_fnids) != 0: |
| self.rq.state = runQueueFailed |
| return True |
| |
| self.rq.state = runQueueComplete |
| return True |
| |
| def check_dependencies(self, task, taskdeps, setscene = False): |
| if not self.rq.depvalidate: |
| return False |
| |
| taskdata = {} |
| taskdeps.add(task) |
| for dep in taskdeps: |
| if setscene: |
| depid = self.rqdata.runq_setscene[dep] |
| else: |
| depid = dep |
| fn = self.rqdata.taskData.fn_index[self.rqdata.runq_fnid[depid]] |
| pn = self.rqdata.dataCache.pkg_fn[fn] |
| taskname = self.rqdata.runq_task[depid] |
| taskdata[dep] = [pn, taskname, fn] |
| call = self.rq.depvalidate + "(task, taskdata, notneeded, d)" |
| locs = { "task" : task, "taskdata" : taskdata, "notneeded" : self.scenequeue_notneeded, "d" : self.cooker.expanded_data } |
| valid = bb.utils.better_eval(call, locs) |
| return valid |
| |
| class RunQueueExecuteDummy(RunQueueExecute): |
| def __init__(self, rq): |
| self.rq = rq |
| self.stats = RunQueueStats(0) |
| |
| def finish(self): |
| self.rq.state = runQueueComplete |
| return |
| |
| class RunQueueExecuteTasks(RunQueueExecute): |
| def __init__(self, rq): |
| RunQueueExecute.__init__(self, rq) |
| |
| self.stats = RunQueueStats(len(self.rqdata.runq_fnid)) |
| |
| self.stampcache = {} |
| |
| initial_covered = self.rq.scenequeue_covered.copy() |
| |
| # Mark initial buildable tasks |
| for task in xrange(self.stats.total): |
| self.runq_running.append(0) |
| self.runq_complete.append(0) |
| if len(self.rqdata.runq_depends[task]) == 0: |
| self.runq_buildable.append(1) |
| else: |
| self.runq_buildable.append(0) |
| if len(self.rqdata.runq_revdeps[task]) > 0 and self.rqdata.runq_revdeps[task].issubset(self.rq.scenequeue_covered) and task not in self.rq.scenequeue_notcovered: |
| self.rq.scenequeue_covered.add(task) |
| |
| found = True |
| while found: |
| found = False |
| for task in xrange(self.stats.total): |
| if task in self.rq.scenequeue_covered: |
| continue |
| logger.debug(1, 'Considering %s (%s): %s' % (task, self.rqdata.get_user_idstring(task), str(self.rqdata.runq_revdeps[task]))) |
| |
| if len(self.rqdata.runq_revdeps[task]) > 0 and self.rqdata.runq_revdeps[task].issubset(self.rq.scenequeue_covered) and task not in self.rq.scenequeue_notcovered: |
| found = True |
| self.rq.scenequeue_covered.add(task) |
| |
| logger.debug(1, 'Skip list (pre setsceneverify) %s', sorted(self.rq.scenequeue_covered)) |
| |
| # Allow the metadata to elect for setscene tasks to run anyway |
| covered_remove = set() |
| if self.rq.setsceneverify: |
| invalidtasks = [] |
| for task in xrange(len(self.rqdata.runq_task)): |
| fn = self.rqdata.taskData.fn_index[self.rqdata.runq_fnid[task]] |
| taskname = self.rqdata.runq_task[task] |
| taskdep = self.rqdata.dataCache.task_deps[fn] |
| |
| if 'noexec' in taskdep and taskname in taskdep['noexec']: |
| continue |
| if self.rq.check_stamp_task(task, taskname + "_setscene", cache=self.stampcache): |
| logger.debug(2, 'Setscene stamp current for task %s(%s)', task, self.rqdata.get_user_idstring(task)) |
| continue |
| if self.rq.check_stamp_task(task, taskname, recurse = True, cache=self.stampcache): |
| logger.debug(2, 'Normal stamp current for task %s(%s)', task, self.rqdata.get_user_idstring(task)) |
| continue |
| invalidtasks.append(task) |
| |
| call = self.rq.setsceneverify + "(covered, tasknames, fnids, fns, d, invalidtasks=invalidtasks)" |
| call2 = self.rq.setsceneverify + "(covered, tasknames, fnids, fns, d)" |
| locs = { "covered" : self.rq.scenequeue_covered, "tasknames" : self.rqdata.runq_task, "fnids" : self.rqdata.runq_fnid, "fns" : self.rqdata.taskData.fn_index, "d" : self.cooker.expanded_data, "invalidtasks" : invalidtasks } |
| # Backwards compatibility with older versions without invalidtasks |
| try: |
| covered_remove = bb.utils.better_eval(call, locs) |
| except TypeError: |
| covered_remove = bb.utils.better_eval(call2, locs) |
| |
| def removecoveredtask(task): |
| fn = self.rqdata.taskData.fn_index[self.rqdata.runq_fnid[task]] |
| taskname = self.rqdata.runq_task[task] + '_setscene' |
| bb.build.del_stamp(taskname, self.rqdata.dataCache, fn) |
| self.rq.scenequeue_covered.remove(task) |
| |
| toremove = covered_remove |
| for task in toremove: |
| logger.debug(1, 'Not skipping task %s due to setsceneverify', task) |
| while toremove: |
| covered_remove = [] |
| for task in toremove: |
| removecoveredtask(task) |
| for deptask in self.rqdata.runq_depends[task]: |
| if deptask not in self.rq.scenequeue_covered: |
| continue |
| if deptask in toremove or deptask in covered_remove or deptask in initial_covered: |
| continue |
| logger.debug(1, 'Task %s depends on task %s so not skipping' % (task, deptask)) |
| covered_remove.append(deptask) |
| toremove = covered_remove |
| |
| logger.debug(1, 'Full skip list %s', self.rq.scenequeue_covered) |
| |
| event.fire(bb.event.StampUpdate(self.rqdata.target_pairs, self.rqdata.dataCache.stamp), self.cfgData) |
| |
| schedulers = self.get_schedulers() |
| for scheduler in schedulers: |
| if self.scheduler == scheduler.name: |
| self.sched = scheduler(self, self.rqdata) |
| logger.debug(1, "Using runqueue scheduler '%s'", scheduler.name) |
| break |
| else: |
| bb.fatal("Invalid scheduler '%s'. Available schedulers: %s" % |
| (self.scheduler, ", ".join(obj.name for obj in schedulers))) |
| |
| def get_schedulers(self): |
| schedulers = set(obj for obj in globals().values() |
| if type(obj) is type and |
| issubclass(obj, RunQueueScheduler)) |
| |
| user_schedulers = self.cfgData.getVar("BB_SCHEDULERS", True) |
| if user_schedulers: |
| for sched in user_schedulers.split(): |
| if not "." in sched: |
| bb.note("Ignoring scheduler '%s' from BB_SCHEDULERS: not an import" % sched) |
| continue |
| |
| modname, name = sched.rsplit(".", 1) |
| try: |
| module = __import__(modname, fromlist=(name,)) |
| except ImportError as exc: |
| logger.critical("Unable to import scheduler '%s' from '%s': %s" % (name, modname, exc)) |
| raise SystemExit(1) |
| else: |
| schedulers.add(getattr(module, name)) |
| return schedulers |
| |
| def setbuildable(self, task): |
| self.runq_buildable[task] = 1 |
| self.sched.newbuilable(task) |
| |
| def task_completeoutright(self, task): |
| """ |
| Mark a task as completed |
| Look at the reverse dependencies and mark any task with |
| completed dependencies as buildable |
| """ |
| self.runq_complete[task] = 1 |
| for revdep in self.rqdata.runq_revdeps[task]: |
| if self.runq_running[revdep] == 1: |
| continue |
| if self.runq_buildable[revdep] == 1: |
| continue |
| alldeps = 1 |
| for dep in self.rqdata.runq_depends[revdep]: |
| if self.runq_complete[dep] != 1: |
| alldeps = 0 |
| if alldeps == 1: |
| self.setbuildable(revdep) |
| fn = self.rqdata.taskData.fn_index[self.rqdata.runq_fnid[revdep]] |
| taskname = self.rqdata.runq_task[revdep] |
| logger.debug(1, "Marking task %s (%s, %s) as buildable", revdep, fn, taskname) |
| |
| def task_complete(self, task): |
| self.stats.taskCompleted() |
| bb.event.fire(runQueueTaskCompleted(task, self.stats, self.rq), self.cfgData) |
| self.task_completeoutright(task) |
| |
| def task_fail(self, task, exitcode): |
| """ |
| Called when a task has failed |
| Updates the state engine with the failure |
| """ |
| self.stats.taskFailed() |
| fnid = self.rqdata.runq_fnid[task] |
| self.failed_fnids.append(fnid) |
| bb.event.fire(runQueueTaskFailed(task, self.stats, exitcode, self.rq), self.cfgData) |
| if self.rqdata.taskData.abort: |
| self.rq.state = runQueueCleanUp |
| |
| def task_skip(self, task, reason): |
| self.runq_running[task] = 1 |
| self.setbuildable(task) |
| bb.event.fire(runQueueTaskSkipped(task, self.stats, self.rq, reason), self.cfgData) |
| self.task_completeoutright(task) |
| self.stats.taskCompleted() |
| self.stats.taskSkipped() |
| |
| def execute(self): |
| """ |
| Run the tasks in a queue prepared by rqdata.prepare() |
| """ |
| |
| self.rq.read_workers() |
| |
| |
| if self.stats.total == 0: |
| # nothing to do |
| self.rq.state = runQueueCleanUp |
| |
| task = self.sched.next() |
| if task is not None: |
| fn = self.rqdata.taskData.fn_index[self.rqdata.runq_fnid[task]] |
| taskname = self.rqdata.runq_task[task] |
| |
| if task in self.rq.scenequeue_covered: |
| logger.debug(2, "Setscene covered task %s (%s)", task, |
| self.rqdata.get_user_idstring(task)) |
| self.task_skip(task, "covered") |
| return True |
| |
| if self.rq.check_stamp_task(task, taskname, cache=self.stampcache): |
| logger.debug(2, "Stamp current task %s (%s)", task, |
| self.rqdata.get_user_idstring(task)) |
| self.task_skip(task, "existing") |
| return True |
| |
| taskdep = self.rqdata.dataCache.task_deps[fn] |
| if 'noexec' in taskdep and taskname in taskdep['noexec']: |
| startevent = runQueueTaskStarted(task, self.stats, self.rq, |
| noexec=True) |
| bb.event.fire(startevent, self.cfgData) |
| self.runq_running[task] = 1 |
| self.stats.taskActive() |
| if not self.cooker.configuration.dry_run: |
| bb.build.make_stamp(taskname, self.rqdata.dataCache, fn) |
| self.task_complete(task) |
| return True |
| else: |
| startevent = runQueueTaskStarted(task, self.stats, self.rq) |
| bb.event.fire(startevent, self.cfgData) |
| |
| taskdepdata = self.build_taskdepdata(task) |
| |
| taskdep = self.rqdata.dataCache.task_deps[fn] |
| if 'fakeroot' in taskdep and taskname in taskdep['fakeroot'] and not self.cooker.configuration.dry_run: |
| if not self.rq.fakeworker: |
| try: |
| self.rq.start_fakeworker(self) |
| except OSError as exc: |
| logger.critical("Failed to spawn fakeroot worker to run %s:%s: %s" % (fn, taskname, str(exc))) |
| self.rq.state = runQueueFailed |
| return True |
| self.rq.fakeworker.stdin.write("<runtask>" + pickle.dumps((fn, task, taskname, False, self.cooker.collection.get_file_appends(fn), taskdepdata)) + "</runtask>") |
| self.rq.fakeworker.stdin.flush() |
| else: |
| self.rq.worker.stdin.write("<runtask>" + pickle.dumps((fn, task, taskname, False, self.cooker.collection.get_file_appends(fn), taskdepdata)) + "</runtask>") |
| self.rq.worker.stdin.flush() |
| |
| self.build_stamps[task] = bb.build.stampfile(taskname, self.rqdata.dataCache, fn) |
| self.build_stamps2.append(self.build_stamps[task]) |
| self.runq_running[task] = 1 |
| self.stats.taskActive() |
| if self.stats.active < self.number_tasks: |
| return True |
| |
| if self.stats.active > 0: |
| self.rq.read_workers() |
| return self.rq.active_fds() |
| |
| if len(self.failed_fnids) != 0: |
| self.rq.state = runQueueFailed |
| return True |
| |
| # Sanity Checks |
| for task in xrange(self.stats.total): |
| if self.runq_buildable[task] == 0: |
| logger.error("Task %s never buildable!", task) |
| if self.runq_running[task] == 0: |
| logger.error("Task %s never ran!", task) |
| if self.runq_complete[task] == 0: |
| logger.error("Task %s never completed!", task) |
| self.rq.state = runQueueComplete |
| |
| return True |
| |
| def build_taskdepdata(self, task): |
| taskdepdata = {} |
| next = self.rqdata.runq_depends[task] |
| next.add(task) |
| while next: |
| additional = [] |
| for revdep in next: |
| fn = self.rqdata.taskData.fn_index[self.rqdata.runq_fnid[revdep]] |
| pn = self.rqdata.dataCache.pkg_fn[fn] |
| taskname = self.rqdata.runq_task[revdep] |
| deps = self.rqdata.runq_depends[revdep] |
| provides = self.rqdata.dataCache.fn_provides[fn] |
| taskdepdata[revdep] = [pn, taskname, fn, deps, provides] |
| for revdep2 in deps: |
| if revdep2 not in taskdepdata: |
| additional.append(revdep2) |
| next = additional |
| |
| #bb.note("Task %s: " % task + str(taskdepdata).replace("], ", "],\n")) |
| return taskdepdata |
| |
| class RunQueueExecuteScenequeue(RunQueueExecute): |
| def __init__(self, rq): |
| RunQueueExecute.__init__(self, rq) |
| |
| self.scenequeue_covered = set() |
| self.scenequeue_notcovered = set() |
| self.scenequeue_notneeded = set() |
| |
| # If we don't have any setscene functions, skip this step |
| if len(self.rqdata.runq_setscene) == 0: |
| rq.scenequeue_covered = set() |
| rq.state = runQueueRunInit |
| return |
| |
| self.stats = RunQueueStats(len(self.rqdata.runq_setscene)) |
| |
| sq_revdeps = [] |
| sq_revdeps_new = [] |
| sq_revdeps_squash = [] |
| self.sq_harddeps = {} |
| |
| # We need to construct a dependency graph for the setscene functions. Intermediate |
| # dependencies between the setscene tasks only complicate the code. This code |
| # therefore aims to collapse the huge runqueue dependency tree into a smaller one |
| # only containing the setscene functions. |
| |
| for task in xrange(self.stats.total): |
| self.runq_running.append(0) |
| self.runq_complete.append(0) |
| self.runq_buildable.append(0) |
| |
| # First process the chains up to the first setscene task. |
| endpoints = {} |
| for task in xrange(len(self.rqdata.runq_fnid)): |
| sq_revdeps.append(copy.copy(self.rqdata.runq_revdeps[task])) |
| sq_revdeps_new.append(set()) |
| if (len(self.rqdata.runq_revdeps[task]) == 0) and task not in self.rqdata.runq_setscene: |
| endpoints[task] = set() |
| |
| # Secondly process the chains between setscene tasks. |
| for task in self.rqdata.runq_setscene: |
| for dep in self.rqdata.runq_depends[task]: |
| if dep not in endpoints: |
| endpoints[dep] = set() |
| endpoints[dep].add(task) |
| |
| def process_endpoints(endpoints): |
| newendpoints = {} |
| for point, task in endpoints.items(): |
| tasks = set() |
| if task: |
| tasks |= task |
| if sq_revdeps_new[point]: |
| tasks |= sq_revdeps_new[point] |
| sq_revdeps_new[point] = set() |
| if point in self.rqdata.runq_setscene: |
| sq_revdeps_new[point] = tasks |
| for dep in self.rqdata.runq_depends[point]: |
| if point in sq_revdeps[dep]: |
| sq_revdeps[dep].remove(point) |
| if tasks: |
| sq_revdeps_new[dep] |= tasks |
| if (len(sq_revdeps[dep]) == 0 or len(sq_revdeps_new[dep]) != 0) and dep not in self.rqdata.runq_setscene: |
| newendpoints[dep] = task |
| if len(newendpoints) != 0: |
| process_endpoints(newendpoints) |
| |
| process_endpoints(endpoints) |
| |
| # Build a list of setscene tasks which are "unskippable" |
| # These are direct endpoints referenced by the build |
| endpoints2 = {} |
| sq_revdeps2 = [] |
| sq_revdeps_new2 = [] |
| def process_endpoints2(endpoints): |
| newendpoints = {} |
| for point, task in endpoints.items(): |
| tasks = set([point]) |
| if task: |
| tasks |= task |
| if sq_revdeps_new2[point]: |
| tasks |= sq_revdeps_new2[point] |
| sq_revdeps_new2[point] = set() |
| if point in self.rqdata.runq_setscene: |
| sq_revdeps_new2[point] = tasks |
| for dep in self.rqdata.runq_depends[point]: |
| if point in sq_revdeps2[dep]: |
| sq_revdeps2[dep].remove(point) |
| if tasks: |
| sq_revdeps_new2[dep] |= tasks |
| if (len(sq_revdeps2[dep]) == 0 or len(sq_revdeps_new2[dep]) != 0) and dep not in self.rqdata.runq_setscene: |
| newendpoints[dep] = tasks |
| if len(newendpoints) != 0: |
| process_endpoints2(newendpoints) |
| for task in xrange(len(self.rqdata.runq_fnid)): |
| sq_revdeps2.append(copy.copy(self.rqdata.runq_revdeps[task])) |
| sq_revdeps_new2.append(set()) |
| if (len(self.rqdata.runq_revdeps[task]) == 0) and task not in self.rqdata.runq_setscene: |
| endpoints2[task] = set() |
| process_endpoints2(endpoints2) |
| self.unskippable = [] |
| for task in self.rqdata.runq_setscene: |
| if sq_revdeps_new2[task]: |
| self.unskippable.append(self.rqdata.runq_setscene.index(task)) |
| |
| for task in xrange(len(self.rqdata.runq_fnid)): |
| if task in self.rqdata.runq_setscene: |
| deps = set() |
| for dep in sq_revdeps_new[task]: |
| deps.add(self.rqdata.runq_setscene.index(dep)) |
| sq_revdeps_squash.append(deps) |
| elif len(sq_revdeps_new[task]) != 0: |
| bb.msg.fatal("RunQueue", "Something went badly wrong during scenequeue generation, aborting. Please report this problem.") |
| |
| # Resolve setscene inter-task dependencies |
| # e.g. do_sometask_setscene[depends] = "targetname:do_someothertask_setscene" |
| # Note that anything explicitly depended upon will have its reverse dependencies removed to avoid circular dependencies |
| for task in self.rqdata.runq_setscene: |
| realid = self.rqdata.taskData.gettask_id(self.rqdata.taskData.fn_index[self.rqdata.runq_fnid[task]], self.rqdata.runq_task[task] + "_setscene", False) |
| idepends = self.rqdata.taskData.tasks_idepends[realid] |
| for (depid, idependtask) in idepends: |
| if depid not in self.rqdata.taskData.build_targets: |
| continue |
| |
| depdata = self.rqdata.taskData.build_targets[depid][0] |
| if depdata is None: |
| continue |
| dep = self.rqdata.taskData.fn_index[depdata] |
| taskid = self.rqdata.get_task_id(self.rqdata.taskData.getfn_id(dep), idependtask.replace("_setscene", "")) |
| if taskid is None: |
| bb.msg.fatal("RunQueue", "Task %s_setscene depends upon non-existent task %s:%s" % (self.rqdata.get_user_idstring(task), dep, idependtask)) |
| |
| if not self.rqdata.runq_setscene.index(taskid) in self.sq_harddeps: |
| self.sq_harddeps[self.rqdata.runq_setscene.index(taskid)] = set() |
| self.sq_harddeps[self.rqdata.runq_setscene.index(taskid)].add(self.rqdata.runq_setscene.index(task)) |
| |
| sq_revdeps_squash[self.rqdata.runq_setscene.index(task)].add(self.rqdata.runq_setscene.index(taskid)) |
| # Have to zero this to avoid circular dependencies |
| sq_revdeps_squash[self.rqdata.runq_setscene.index(taskid)] = set() |
| |
| for task in self.sq_harddeps: |
| for dep in self.sq_harddeps[task]: |
| sq_revdeps_squash[dep].add(task) |
| |
| #for task in xrange(len(sq_revdeps_squash)): |
| # realtask = self.rqdata.runq_setscene[task] |
| # bb.warn("Task %s: %s_setscene is %s " % (task, self.rqdata.get_user_idstring(realtask) , sq_revdeps_squash[task])) |
| |
| self.sq_deps = [] |
| self.sq_revdeps = sq_revdeps_squash |
| self.sq_revdeps2 = copy.deepcopy(self.sq_revdeps) |
| |
| for task in xrange(len(self.sq_revdeps)): |
| self.sq_deps.append(set()) |
| for task in xrange(len(self.sq_revdeps)): |
| for dep in self.sq_revdeps[task]: |
| self.sq_deps[dep].add(task) |
| |
| for task in xrange(len(self.sq_revdeps)): |
| if len(self.sq_revdeps[task]) == 0: |
| self.runq_buildable[task] = 1 |
| |
| self.outrightfail = [] |
| if self.rq.hashvalidate: |
| sq_hash = [] |
| sq_hashfn = [] |
| sq_fn = [] |
| sq_taskname = [] |
| sq_task = [] |
| noexec = [] |
| stamppresent = [] |
| for task in xrange(len(self.sq_revdeps)): |
| realtask = self.rqdata.runq_setscene[task] |
| fn = self.rqdata.taskData.fn_index[self.rqdata.runq_fnid[realtask]] |
| taskname = self.rqdata.runq_task[realtask] |
| taskdep = self.rqdata.dataCache.task_deps[fn] |
| |
| if 'noexec' in taskdep and taskname in taskdep['noexec']: |
| noexec.append(task) |
| self.task_skip(task) |
| bb.build.make_stamp(taskname + "_setscene", self.rqdata.dataCache, fn) |
| continue |
| |
| if self.rq.check_stamp_task(realtask, taskname + "_setscene", cache=self.stampcache): |
| logger.debug(2, 'Setscene stamp current for task %s(%s)', task, self.rqdata.get_user_idstring(realtask)) |
| stamppresent.append(task) |
| self.task_skip(task) |
| continue |
| |
| if self.rq.check_stamp_task(realtask, taskname, recurse = True, cache=self.stampcache): |
| logger.debug(2, 'Normal stamp current for task %s(%s)', task, self.rqdata.get_user_idstring(realtask)) |
| stamppresent.append(task) |
| self.task_skip(task) |
| continue |
| |
| sq_fn.append(fn) |
| sq_hashfn.append(self.rqdata.dataCache.hashfn[fn]) |
| sq_hash.append(self.rqdata.runq_hash[realtask]) |
| sq_taskname.append(taskname) |
| sq_task.append(task) |
| call = self.rq.hashvalidate + "(sq_fn, sq_task, sq_hash, sq_hashfn, d)" |
| locs = { "sq_fn" : sq_fn, "sq_task" : sq_taskname, "sq_hash" : sq_hash, "sq_hashfn" : sq_hashfn, "d" : self.cooker.expanded_data } |
| valid = bb.utils.better_eval(call, locs) |
| |
| valid_new = stamppresent |
| for v in valid: |
| valid_new.append(sq_task[v]) |
| |
| for task in xrange(len(self.sq_revdeps)): |
| if task not in valid_new and task not in noexec: |
| realtask = self.rqdata.runq_setscene[task] |
| logger.debug(2, 'No package found, so skipping setscene task %s', |
| self.rqdata.get_user_idstring(realtask)) |
| self.outrightfail.append(task) |
| |
| logger.info('Executing SetScene Tasks') |
| |
| self.rq.state = runQueueSceneRun |
| |
| def scenequeue_updatecounters(self, task, fail = False): |
| for dep in self.sq_deps[task]: |
| if fail and task in self.sq_harddeps and dep in self.sq_harddeps[task]: |
| realtask = self.rqdata.runq_setscene[task] |
| realdep = self.rqdata.runq_setscene[dep] |
| logger.debug(2, "%s was unavailable and is a hard dependency of %s so skipping" % (self.rqdata.get_user_idstring(realtask), self.rqdata.get_user_idstring(realdep))) |
| self.scenequeue_updatecounters(dep, fail) |
| continue |
| if task not in self.sq_revdeps2[dep]: |
| # May already have been removed by the fail case above |
| continue |
| self.sq_revdeps2[dep].remove(task) |
| if len(self.sq_revdeps2[dep]) == 0: |
| self.runq_buildable[dep] = 1 |
| |
| def task_completeoutright(self, task): |
| """ |
| Mark a task as completed |
| Look at the reverse dependencies and mark any task with |
| completed dependencies as buildable |
| """ |
| |
| index = self.rqdata.runq_setscene[task] |
| logger.debug(1, 'Found task %s which could be accelerated', |
| self.rqdata.get_user_idstring(index)) |
| |
| self.scenequeue_covered.add(task) |
| self.scenequeue_updatecounters(task) |
| |
| def task_complete(self, task): |
| self.stats.taskCompleted() |
| bb.event.fire(sceneQueueTaskCompleted(task, self.stats, self.rq), self.cfgData) |
| self.task_completeoutright(task) |
| |
| def task_fail(self, task, result): |
| self.stats.taskFailed() |
| bb.event.fire(sceneQueueTaskFailed(task, self.stats, result, self), self.cfgData) |
| self.scenequeue_notcovered.add(task) |
| self.scenequeue_updatecounters(task, True) |
| |
| def task_failoutright(self, task): |
| self.runq_running[task] = 1 |
| self.runq_buildable[task] = 1 |
| self.stats.taskCompleted() |
| self.stats.taskSkipped() |
| index = self.rqdata.runq_setscene[task] |
| self.scenequeue_notcovered.add(task) |
| self.scenequeue_updatecounters(task, True) |
| |
| def task_skip(self, task): |
| self.runq_running[task] = 1 |
| self.runq_buildable[task] = 1 |
| self.task_completeoutright(task) |
| self.stats.taskCompleted() |
| self.stats.taskSkipped() |
| |
| def execute(self): |
| """ |
| Run the tasks in a queue prepared by prepare_runqueue |
| """ |
| |
| self.rq.read_workers() |
| |
| task = None |
| if self.stats.active < self.number_tasks: |
| # Find the next setscene to run |
| for nexttask in xrange(self.stats.total): |
| if self.runq_buildable[nexttask] == 1 and self.runq_running[nexttask] != 1: |
| if nexttask in self.unskippable: |
| logger.debug(2, "Setscene task %s is unskippable" % self.rqdata.get_user_idstring(self.rqdata.runq_setscene[nexttask])) |
| if nexttask not in self.unskippable and len(self.sq_revdeps[nexttask]) > 0 and self.sq_revdeps[nexttask].issubset(self.scenequeue_covered) and self.check_dependencies(nexttask, self.sq_revdeps[nexttask], True): |
| realtask = self.rqdata.runq_setscene[nexttask] |
| fn = self.rqdata.taskData.fn_index[self.rqdata.runq_fnid[realtask]] |
| foundtarget = False |
| for target in self.rqdata.target_pairs: |
| if target[0] == fn and target[1] == self.rqdata.runq_task[realtask]: |
| foundtarget = True |
| break |
| if not foundtarget: |
| logger.debug(2, "Skipping setscene for task %s" % self.rqdata.get_user_idstring(self.rqdata.runq_setscene[nexttask])) |
| self.task_skip(nexttask) |
| self.scenequeue_notneeded.add(nexttask) |
| return True |
| if nexttask in self.outrightfail: |
| self.task_failoutright(nexttask) |
| return True |
| task = nexttask |
| break |
| if task is not None: |
| realtask = self.rqdata.runq_setscene[task] |
| fn = self.rqdata.taskData.fn_index[self.rqdata.runq_fnid[realtask]] |
| |
| taskname = self.rqdata.runq_task[realtask] + "_setscene" |
| if self.rq.check_stamp_task(realtask, self.rqdata.runq_task[realtask], recurse = True, cache=self.stampcache): |
| logger.debug(2, 'Stamp for underlying task %s(%s) is current, so skipping setscene variant', |
| task, self.rqdata.get_user_idstring(realtask)) |
| self.task_failoutright(task) |
| return True |
| |
| if self.cooker.configuration.force: |
| for target in self.rqdata.target_pairs: |
| if target[0] == fn and target[1] == self.rqdata.runq_task[realtask]: |
| self.task_failoutright(task) |
| return True |
| |
| if self.rq.check_stamp_task(realtask, taskname, cache=self.stampcache): |
| logger.debug(2, 'Setscene stamp current task %s(%s), so skip it and its dependencies', |
| task, self.rqdata.get_user_idstring(realtask)) |
| self.task_skip(task) |
| return True |
| |
| startevent = sceneQueueTaskStarted(task, self.stats, self.rq) |
| bb.event.fire(startevent, self.cfgData) |
| |
| taskdep = self.rqdata.dataCache.task_deps[fn] |
| if 'fakeroot' in taskdep and taskname in taskdep['fakeroot']: |
| if not self.rq.fakeworker: |
| self.rq.start_fakeworker(self) |
| self.rq.fakeworker.stdin.write("<runtask>" + pickle.dumps((fn, realtask, taskname, True, self.cooker.collection.get_file_appends(fn), None)) + "</runtask>") |
| self.rq.fakeworker.stdin.flush() |
| else: |
| self.rq.worker.stdin.write("<runtask>" + pickle.dumps((fn, realtask, taskname, True, self.cooker.collection.get_file_appends(fn), None)) + "</runtask>") |
| self.rq.worker.stdin.flush() |
| |
| self.runq_running[task] = 1 |
| self.stats.taskActive() |
| if self.stats.active < self.number_tasks: |
| return True |
| |
| if self.stats.active > 0: |
| self.rq.read_workers() |
| return self.rq.active_fds() |
| |
| #for task in xrange(self.stats.total): |
| # if self.runq_running[task] != 1: |
| # buildable = self.runq_buildable[task] |
| # revdeps = self.sq_revdeps[task] |
| # bb.warn("Found we didn't run %s %s %s %s" % (task, buildable, str(revdeps), self.rqdata.get_user_idstring(self.rqdata.runq_setscene[task]))) |
| |
| # Convert scenequeue_covered task numbers into full taskgraph ids |
| oldcovered = self.scenequeue_covered |
| self.rq.scenequeue_covered = set() |
| for task in oldcovered: |
| self.rq.scenequeue_covered.add(self.rqdata.runq_setscene[task]) |
| self.rq.scenequeue_notcovered = set() |
| for task in self.scenequeue_notcovered: |
| self.rq.scenequeue_notcovered.add(self.rqdata.runq_setscene[task]) |
| |
| logger.debug(1, 'We can skip tasks %s', sorted(self.rq.scenequeue_covered)) |
| |
| self.rq.state = runQueueRunInit |
| |
| completeevent = sceneQueueComplete(self.stats, self.rq) |
| bb.event.fire(completeevent, self.cfgData) |
| |
| return True |
| |
| def runqueue_process_waitpid(self, task, status): |
| task = self.rq.rqdata.runq_setscene.index(task) |
| |
| RunQueueExecute.runqueue_process_waitpid(self, task, status) |
| |
| class TaskFailure(Exception): |
| """ |
| Exception raised when a task in a runqueue fails |
| """ |
| def __init__(self, x): |
| self.args = x |
| |
| |
| class runQueueExitWait(bb.event.Event): |
| """ |
| Event when waiting for task processes to exit |
| """ |
| |
| def __init__(self, remain): |
| self.remain = remain |
| self.message = "Waiting for %s active tasks to finish" % remain |
| bb.event.Event.__init__(self) |
| |
| class runQueueEvent(bb.event.Event): |
| """ |
| Base runQueue event class |
| """ |
| def __init__(self, task, stats, rq): |
| self.taskid = task |
| self.taskstring = rq.rqdata.get_user_idstring(task) |
| self.taskname = rq.rqdata.get_task_name(task) |
| self.taskfile = rq.rqdata.get_task_file(task) |
| self.taskhash = rq.rqdata.get_task_hash(task) |
| self.stats = stats.copy() |
| bb.event.Event.__init__(self) |
| |
| class sceneQueueEvent(runQueueEvent): |
| """ |
| Base sceneQueue event class |
| """ |
| def __init__(self, task, stats, rq, noexec=False): |
| runQueueEvent.__init__(self, task, stats, rq) |
| realtask = rq.rqdata.runq_setscene[task] |
| self.taskstring = rq.rqdata.get_user_idstring(realtask, "_setscene") |
| self.taskname = rq.rqdata.get_task_name(realtask) + "_setscene" |
| self.taskfile = rq.rqdata.get_task_file(realtask) |
| self.taskhash = rq.rqdata.get_task_hash(realtask) |
| |
| class runQueueTaskStarted(runQueueEvent): |
| """ |
| Event notifying a task was started |
| """ |
| def __init__(self, task, stats, rq, noexec=False): |
| runQueueEvent.__init__(self, task, stats, rq) |
| self.noexec = noexec |
| |
| class sceneQueueTaskStarted(sceneQueueEvent): |
| """ |
| Event notifying a setscene task was started |
| """ |
| def __init__(self, task, stats, rq, noexec=False): |
| sceneQueueEvent.__init__(self, task, stats, rq) |
| self.noexec = noexec |
| |
| class runQueueTaskFailed(runQueueEvent): |
| """ |
| Event notifying a task failed |
| """ |
| def __init__(self, task, stats, exitcode, rq): |
| runQueueEvent.__init__(self, task, stats, rq) |
| self.exitcode = exitcode |
| |
| class sceneQueueTaskFailed(sceneQueueEvent): |
| """ |
| Event notifying a setscene task failed |
| """ |
| def __init__(self, task, stats, exitcode, rq): |
| sceneQueueEvent.__init__(self, task, stats, rq) |
| self.exitcode = exitcode |
| |
| class sceneQueueComplete(sceneQueueEvent): |
| """ |
| Event when all the sceneQueue tasks are complete |
| """ |
| def __init__(self, stats, rq): |
| self.stats = stats.copy() |
| bb.event.Event.__init__(self) |
| |
| class runQueueTaskCompleted(runQueueEvent): |
| """ |
| Event notifying a task completed |
| """ |
| |
| class sceneQueueTaskCompleted(sceneQueueEvent): |
| """ |
| Event notifying a setscene task completed |
| """ |
| |
| class runQueueTaskSkipped(runQueueEvent): |
| """ |
| Event notifying a task was skipped |
| """ |
| def __init__(self, task, stats, rq, reason): |
| runQueueEvent.__init__(self, task, stats, rq) |
| self.reason = reason |
| |
| class runQueuePipe(): |
| """ |
| Abstraction for a pipe between a worker thread and the server |
| """ |
| def __init__(self, pipein, pipeout, d, rq, rqexec): |
| self.input = pipein |
| if pipeout: |
| pipeout.close() |
| bb.utils.nonblockingfd(self.input) |
| self.queue = "" |
| self.d = d |
| self.rq = rq |
| self.rqexec = rqexec |
| |
| def setrunqueueexec(self, rqexec): |
| self.rqexec = rqexec |
| |
| def read(self): |
| for w in [self.rq.worker, self.rq.fakeworker]: |
| if not w: |
| continue |
| w.poll() |
| if w.returncode is not None and not self.rq.teardown: |
| name = None |
| if self.rq.worker and w.pid == self.rq.worker.pid: |
| name = "Worker" |
| elif self.rq.fakeworker and w.pid == self.rq.fakeworker.pid: |
| name = "Fakeroot" |
| bb.error("%s process (%s) exited unexpectedly (%s), shutting down..." % (name, w.pid, str(w.returncode))) |
| self.rq.finish_runqueue(True) |
| |
| start = len(self.queue) |
| try: |
| self.queue = self.queue + self.input.read(102400) |
| except (OSError, IOError) as e: |
| if e.errno != errno.EAGAIN: |
| raise |
| end = len(self.queue) |
| found = True |
| while found and len(self.queue): |
| found = False |
| index = self.queue.find("</event>") |
| while index != -1 and self.queue.startswith("<event>"): |
| try: |
| event = pickle.loads(self.queue[7:index]) |
| except ValueError as e: |
| bb.msg.fatal("RunQueue", "failed load pickle '%s': '%s'" % (e, self.queue[7:index])) |
| bb.event.fire_from_worker(event, self.d) |
| found = True |
| self.queue = self.queue[index+8:] |
| index = self.queue.find("</event>") |
| index = self.queue.find("</exitcode>") |
| while index != -1 and self.queue.startswith("<exitcode>"): |
| try: |
| task, status = pickle.loads(self.queue[10:index]) |
| except ValueError as e: |
| bb.msg.fatal("RunQueue", "failed load pickle '%s': '%s'" % (e, self.queue[10:index])) |
| self.rqexec.runqueue_process_waitpid(task, status) |
| found = True |
| self.queue = self.queue[index+11:] |
| index = self.queue.find("</exitcode>") |
| return (end > start) |
| |
| def close(self): |
| while self.read(): |
| continue |
| if len(self.queue) > 0: |
| print("Warning, worker left partial message: %s" % self.queue) |
| self.input.close() |