123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516 |
- #!/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 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.
- from bb import msg, data, fetch, event, mkdirhier, utils
- from sets import Set
- import bb, os, sys
- class TaskFailure(Exception):
- """Exception raised when a task in a runqueue fails"""
- def __init__(self, x):
- self.args = x
- class RunQueue:
- """
- BitBake Run Queue implementation
- """
- def __init__(self):
- self.reset_runqueue()
- def reset_runqueue(self):
- self.runq_fnid = []
- self.runq_task = []
- self.runq_depends = []
- self.runq_revdeps = []
- self.runq_weight = []
- self.prio_map = []
- def get_user_idstring(self, task, taskData):
- fn = taskData.fn_index[self.runq_fnid[task]]
- taskname = self.runq_task[task]
- return "%s, %s" % (fn, taskname)
- def prepare_runqueue(self, cfgData, dataCache, taskData, targets):
- """
- Turn a set of taskData into a RunQueue and compute data needed
- to optimise the execution order.
- targets is list of paired values - a provider name and the task to run
- """
- depends = []
- runq_weight1 = []
- runq_build = []
- runq_done = []
- bb.msg.note(1, bb.msg.domain.RunQueue, "Preparing Runqueue")
- for task in range(len(taskData.tasks_name)):
- fnid = taskData.tasks_fnid[task]
- fn = taskData.fn_index[fnid]
- task_deps = dataCache.task_deps[fn]
- if fnid not in taskData.failed_fnids:
- depends = taskData.tasks_tdepends[task]
- # Resolve Depends
- if 'deptask' in task_deps and taskData.tasks_name[task] in task_deps['deptask']:
- taskname = task_deps['deptask'][taskData.tasks_name[task]]
- for depid in taskData.depids[fnid]:
- if depid in taskData.build_targets:
- depdata = taskData.build_targets[depid][0]
- if depdata:
- dep = taskData.fn_index[depdata]
- depends.append(taskData.gettask_id(dep, taskname))
- # Resolve Runtime Depends
- if 'rdeptask' in task_deps and taskData.tasks_name[task] in task_deps['rdeptask']:
- taskname = task_deps['rdeptask'][taskData.tasks_name[task]]
- for depid in taskData.rdepids[fnid]:
- if depid in taskData.run_targets:
- depdata = taskData.run_targets[depid][0]
- if depdata:
- dep = taskData.fn_index[depdata]
- depends.append(taskData.gettask_id(dep, taskname))
- def add_recursive_build(depid):
- """
- Add build depends of depid to depends
- (if we've not see it before)
- (calls itself recursively)
- """
- if str(depid) in dep_seen:
- return
- dep_seen.append(depid)
- if depid in taskData.build_targets:
- depdata = taskData.build_targets[depid][0]
- if depdata:
- dep = taskData.fn_index[depdata]
- taskid = taskData.gettask_id(dep, taskname)
- depends.append(taskid)
- fnid = taskData.tasks_fnid[taskid]
- for nextdepid in taskData.depids[fnid]:
- if nextdepid not in dep_seen:
- add_recursive_build(nextdepid)
- for nextdepid in taskData.rdepids[fnid]:
- if nextdepid not in rdep_seen:
- add_recursive_run(nextdepid)
- def add_recursive_run(rdepid):
- """
- Add runtime depends of rdepid to depends
- (if we've not see it before)
- (calls itself recursively)
- """
- if str(rdepid) in rdep_seen:
- return
- rdep_seen.append(rdepid)
- if rdepid in taskData.run_targets:
- depdata = taskData.run_targets[rdepid][0]
- if depdata:
- dep = taskData.fn_index[depdata]
- taskid = taskData.gettask_id(dep, taskname)
- depends.append(taskid)
- fnid = taskData.tasks_fnid[taskid]
- for nextdepid in taskData.depids[fnid]:
- if nextdepid not in dep_seen:
- add_recursive_build(nextdepid)
- for nextdepid in taskData.rdepids[fnid]:
- if nextdepid not in rdep_seen:
- add_recursive_run(nextdepid)
- # Resolve Recursive Runtime Depends
- # Also includes all Build Depends (and their runtime depends)
- if 'recrdeptask' in task_deps and taskData.tasks_name[task] in task_deps['recrdeptask']:
- dep_seen = []
- rdep_seen = []
- taskname = task_deps['recrdeptask'][taskData.tasks_name[task]]
- for depid in taskData.depids[fnid]:
- add_recursive_build(depid)
- for rdepid in taskData.rdepids[fnid]:
- add_recursive_run(rdepid)
- #Prune self references
- if task in depends:
- newdep = []
- bb.msg.debug(2, bb.msg.domain.RunQueue, "Task %s (%s %s) contains self reference! %s" % (task, taskData.fn_index[taskData.tasks_fnid[task]], taskData.tasks_name[task], depends))
- for dep in depends:
- if task != dep:
- newdep.append(dep)
- depends = newdep
- self.runq_fnid.append(taskData.tasks_fnid[task])
- self.runq_task.append(taskData.tasks_name[task])
- self.runq_depends.append(Set(depends))
- self.runq_revdeps.append(Set())
- self.runq_weight.append(0)
- runq_weight1.append(0)
- runq_build.append(0)
- runq_done.append(0)
- bb.msg.note(2, bb.msg.domain.RunQueue, "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)
- for target in targets:
- targetid = taskData.getbuild_id(target[0])
- if targetid in taskData.failed_deps:
- continue
- if targetid not in taskData.build_targets:
- continue
- fnid = taskData.build_targets[targetid][0]
- if fnid in taskData.failed_fnids:
- continue
- listid = taskData.tasks_lookup[fnid][target[1]]
- mark_active(listid, 1)
- # Prune inactive tasks
- maps = []
- delcount = 0
- for listid in range(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 self.runq_weight[listid-delcount]
- del runq_weight1[listid-delcount]
- del runq_build[listid-delcount]
- del runq_done[listid-delcount]
- del self.runq_revdeps[listid-delcount]
- delcount = delcount + 1
- maps.append(-1)
- if len(self.runq_fnid) == 0:
- if not taskData.abort:
- bb.msg.note(1, bb.msg.domain.RunQueue, "All possible tasks have been run but build incomplete (--continue mode). See errors above for incomplete tasks.")
- return
- bb.msg.fatal(bb.msg.domain.RunQueue, "No active tasks and not in --continue mode?! Please report this bug.")
- bb.msg.note(2, bb.msg.domain.RunQueue, "Pruned %s inactive tasks, %s left" % (delcount, len(self.runq_fnid)))
- for listid in range(len(self.runq_fnid)):
- newdeps = []
- origdeps = self.runq_depends[listid]
- for origdep in origdeps:
- if maps[origdep] == -1:
- bb.msg.fatal(bb.msg.domain.RunQueue, "Invalid mapping - Should never happen!")
- newdeps.append(maps[origdep])
- self.runq_depends[listid] = Set(newdeps)
- bb.msg.note(2, bb.msg.domain.RunQueue, "Assign Weightings")
- for listid in range(len(self.runq_fnid)):
- for dep in self.runq_depends[listid]:
- self.runq_revdeps[dep].add(listid)
- endpoints = []
- for listid in range(len(self.runq_fnid)):
- revdeps = self.runq_revdeps[listid]
- if len(revdeps) == 0:
- runq_done[listid] = 1
- self.runq_weight[listid] = 1
- endpoints.append(listid)
- for dep in revdeps:
- if dep in self.runq_depends[listid]:
- #self.dump_data(taskData)
- bb.msg.fatal(bb.msg.domain.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]))
- runq_weight1[listid] = len(revdeps)
- bb.msg.note(2, bb.msg.domain.RunQueue, "Compute totals (have %s endpoint(s))" % len(endpoints))
- while 1:
- next_points = []
- for listid in endpoints:
- for revdep in self.runq_depends[listid]:
- self.runq_weight[revdep] = self.runq_weight[revdep] + self.runq_weight[listid]
- runq_weight1[revdep] = runq_weight1[revdep] - 1
- if runq_weight1[revdep] == 0:
- next_points.append(revdep)
- runq_done[revdep] = 1
- endpoints = next_points
- if len(next_points) == 0:
- break
- # Sanity Checks
- for task in range(len(self.runq_fnid)):
- if runq_done[task] == 0:
- seen = []
- deps_seen = []
- def print_chain(taskid, finish):
- seen.append(taskid)
- for revdep in self.runq_revdeps[taskid]:
- if runq_done[revdep] == 0 and revdep not in seen and not finish:
- bb.msg.error(bb.msg.domain.RunQueue, "Task %s (%s) (depends: %s)" % (revdep, self.get_user_idstring(revdep, taskData), self.runq_depends[revdep]))
- if revdep in deps_seen:
- bb.msg.error(bb.msg.domain.RunQueue, "Chain ends at Task %s (%s)" % (revdep, self.get_user_idstring(revdep, taskData)))
- finish = True
- return
- for dep in self.runq_depends[revdep]:
- deps_seen.append(dep)
- print_chain(revdep, finish)
- print_chain(task, False)
- bb.msg.fatal(bb.msg.domain.RunQueue, "Task %s (%s) not processed!\nThis is probably a circular dependency (the chain might be printed above)." % (task, self.get_user_idstring(task, taskData)))
- if runq_weight1[task] != 0:
- bb.msg.fatal(bb.msg.domain.RunQueue, "Task %s (%s) count not zero!" % (task, self.get_user_idstring(task, taskData)))
- # Make a weight sorted map
- from copy import deepcopy
- sortweight = deepcopy(self.runq_weight)
- sortweight.sort()
- copyweight = deepcopy(self.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()
- #self.dump_data(taskData)
- def execute_runqueue(self, cooker, cfgData, dataCache, taskData, runlist):
- """
- Run the tasks in a queue prepared by prepare_runqueue
- Upon failure, optionally try to recover the build using any alternate providers
- (if the abort on failure configuration option isn't set)
- """
- failures = 0
- while 1:
- failed_fnids = self.execute_runqueue_internal(cooker, cfgData, dataCache, taskData)
- if len(failed_fnids) == 0:
- return failures
- if taskData.abort:
- raise bb.runqueue.TaskFailure(failed_fnids)
- for fnid in failed_fnids:
- #print "Failure: %s %s %s" % (fnid, taskData.fn_index[fnid], self.runq_task[fnid])
- taskData.fail_fnid(fnid)
- failures = failures + 1
- self.reset_runqueue()
- self.prepare_runqueue(cfgData, dataCache, taskData, runlist)
- def execute_runqueue_internal(self, cooker, cfgData, dataCache, taskData):
- """
- Run the tasks in a queue prepared by prepare_runqueue
- """
- import signal
- bb.msg.note(1, bb.msg.domain.RunQueue, "Executing runqueue")
- runq_buildable = []
- runq_running = []
- runq_complete = []
- active_builds = 0
- build_pids = {}
- failed_fnids = []
- if len(self.runq_fnid) == 0:
- # nothing to do
- return
- def sigint_handler(signum, frame):
- raise KeyboardInterrupt
- def get_next_task(data):
- """
- Return the id of the highest priority task that is buildable
- """
- for task1 in range(len(data.runq_fnid)):
- task = data.prio_map[task1]
- if runq_running[task] == 1:
- continue
- if runq_buildable[task] == 1:
- return task
- return None
- def task_complete(data, task):
- """
- Mark a task as completed
- Look at the reverse dependencies and mark any task with
- completed dependencies as buildable
- """
- runq_complete[task] = 1
- for revdep in data.runq_revdeps[task]:
- if runq_running[revdep] == 1:
- continue
- if runq_buildable[revdep] == 1:
- continue
- alldeps = 1
- for dep in data.runq_depends[revdep]:
- if runq_complete[dep] != 1:
- alldeps = 0
- if alldeps == 1:
- runq_buildable[revdep] = 1
- fn = taskData.fn_index[self.runq_fnid[revdep]]
- taskname = self.runq_task[revdep]
- bb.msg.debug(1, bb.msg.domain.RunQueue, "Marking task %s (%s, %s) as buildable" % (revdep, fn, taskname))
- # Mark initial buildable tasks
- for task in range(len(self.runq_fnid)):
- runq_running.append(0)
- runq_complete.append(0)
- if len(self.runq_depends[task]) == 0:
- runq_buildable.append(1)
- else:
- runq_buildable.append(0)
- number_tasks = int(bb.data.getVar("BB_NUMBER_THREADS", cfgData) or 1)
- try:
- while 1:
- task = get_next_task(self)
- if task is not None:
- fn = taskData.fn_index[self.runq_fnid[task]]
- taskname = self.runq_task[task]
- if bb.build.stamp_is_current_cache(dataCache, fn, taskname):
- targetid = taskData.gettask_id(fn, taskname)
- if not (targetid in taskData.external_targets and cooker.configuration.force):
- bb.msg.debug(2, bb.msg.domain.RunQueue, "Stamp current task %s (%s)" % (task, self.get_user_idstring(task, taskData)))
- runq_running[task] = 1
- task_complete(self, task)
- continue
- bb.msg.debug(1, bb.msg.domain.RunQueue, "Running task %s (%s)" % (task, self.get_user_idstring(task, taskData)))
- try:
- pid = os.fork()
- except OSError, e:
- bb.msg.fatal(bb.msg.domain.RunQueue, "fork failed: %d (%s)" % (e.errno, e.strerror))
- if pid == 0:
- # Bypass finally below
- active_builds = 0
- # Stop Ctrl+C being sent to children
- # signal.signal(signal.SIGINT, signal.SIG_IGN)
- # Make the child the process group leader
- os.setpgid(0, 0)
- sys.stdin = open('/dev/null', 'r')
- cooker.configuration.cmd = taskname[3:]
- try:
- cooker.tryBuild(fn, False)
- except bb.build.EventException:
- bb.msg.error(bb.msg.domain.Build, "Build of " + fn + " " + taskname + " failed")
- sys.exit(1)
- except:
- bb.msg.error(bb.msg.domain.Build, "Build of " + fn + " " + taskname + " failed")
- raise
- sys.exit(0)
- build_pids[pid] = task
- runq_running[task] = 1
- active_builds = active_builds + 1
- if active_builds < number_tasks:
- continue
- if active_builds > 0:
- result = os.waitpid(-1, 0)
- active_builds = active_builds - 1
- task = build_pids[result[0]]
- if result[1] != 0:
- del build_pids[result[0]]
- bb.msg.error(bb.msg.domain.RunQueue, "Task %s (%s) failed" % (task, self.get_user_idstring(task, taskData)))
- failed_fnids.append(self.runq_fnid[task])
- break
- task_complete(self, task)
- del build_pids[result[0]]
- continue
- break
- finally:
- try:
- while active_builds > 0:
- bb.msg.note(1, bb.msg.domain.RunQueue, "Waiting for %s active tasks to finish" % active_builds)
- tasknum = 1
- for k, v in build_pids.iteritems():
- bb.msg.note(1, bb.msg.domain.RunQueue, "%s: %s (%s)" % (tasknum, self.get_user_idstring(v, taskData), k))
- tasknum = tasknum + 1
- result = os.waitpid(-1, 0)
- task = build_pids[result[0]]
- if result[1] != 0:
- bb.msg.error(bb.msg.domain.RunQueue, "Task %s (%s) failed" % (task, self.get_user_idstring(task, taskData)))
- failed_fnids.append(self.runq_fnid[task])
- del build_pids[result[0]]
- active_builds = active_builds - 1
- if len(failed_fnids) > 0:
- return failed_fnids
- except:
- bb.msg.note(1, bb.msg.domain.RunQueue, "Sending SIGINT to remaining %s tasks" % active_builds)
- for k, v in build_pids.iteritems():
- os.kill(-k, signal.SIGINT)
- raise
- # Sanity Checks
- for task in range(len(self.runq_fnid)):
- if runq_buildable[task] == 0:
- bb.msg.error(bb.msg.domain.RunQueue, "Task %s never buildable!" % task)
- if runq_running[task] == 0:
- bb.msg.error(bb.msg.domain.RunQueue, "Task %s never ran!" % task)
- if runq_complete[task] == 0:
- bb.msg.error(bb.msg.domain.RunQueue, "Task %s never completed!" % task)
- return failed_fnids
- def dump_data(self, taskQueue):
- """
- Dump some debug information on the internal data structures
- """
- bb.msg.debug(3, bb.msg.domain.RunQueue, "run_tasks:")
- for task in range(len(self.runq_fnid)):
- bb.msg.debug(3, bb.msg.domain.RunQueue, " (%s)%s - %s: %s Deps %s RevDeps %s" % (task,
- taskQueue.fn_index[self.runq_fnid[task]],
- self.runq_task[task],
- self.runq_weight[task],
- self.runq_depends[task],
- self.runq_revdeps[task]))
- bb.msg.debug(3, bb.msg.domain.RunQueue, "sorted_tasks:")
- for task1 in range(len(self.runq_fnid)):
- if task1 in self.prio_map:
- task = self.prio_map[task1]
- bb.msg.debug(3, bb.msg.domain.RunQueue, " (%s)%s - %s: %s Deps %s RevDeps %s" % (task,
- taskQueue.fn_index[self.runq_fnid[task]],
- self.runq_task[task],
- self.runq_weight[task],
- self.runq_depends[task],
- self.runq_revdeps[task]))
|