runqueue.py 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516
  1. #!/usr/bin/env python
  2. # ex:ts=4:sw=4:sts=4:et
  3. # -*- tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*-
  4. """
  5. BitBake 'RunQueue' implementation
  6. Handles preparation and execution of a queue of tasks
  7. """
  8. # Copyright (C) 2006 Richard Purdie
  9. #
  10. # This program is free software; you can redistribute it and/or modify
  11. # it under the terms of the GNU General Public License version 2 as
  12. # published by the Free Software Foundation.
  13. #
  14. # This program is distributed in the hope that it will be useful,
  15. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  16. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  17. # GNU General Public License for more details.
  18. #
  19. # You should have received a copy of the GNU General Public License along
  20. # with this program; if not, write to the Free Software Foundation, Inc.,
  21. # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  22. from bb import msg, data, fetch, event, mkdirhier, utils
  23. from sets import Set
  24. import bb, os, sys
  25. class TaskFailure(Exception):
  26. """Exception raised when a task in a runqueue fails"""
  27. def __init__(self, x):
  28. self.args = x
  29. class RunQueue:
  30. """
  31. BitBake Run Queue implementation
  32. """
  33. def __init__(self):
  34. self.reset_runqueue()
  35. def reset_runqueue(self):
  36. self.runq_fnid = []
  37. self.runq_task = []
  38. self.runq_depends = []
  39. self.runq_revdeps = []
  40. self.runq_weight = []
  41. self.prio_map = []
  42. def get_user_idstring(self, task, taskData):
  43. fn = taskData.fn_index[self.runq_fnid[task]]
  44. taskname = self.runq_task[task]
  45. return "%s, %s" % (fn, taskname)
  46. def prepare_runqueue(self, cfgData, dataCache, taskData, targets):
  47. """
  48. Turn a set of taskData into a RunQueue and compute data needed
  49. to optimise the execution order.
  50. targets is list of paired values - a provider name and the task to run
  51. """
  52. depends = []
  53. runq_weight1 = []
  54. runq_build = []
  55. runq_done = []
  56. bb.msg.note(1, bb.msg.domain.RunQueue, "Preparing Runqueue")
  57. for task in range(len(taskData.tasks_name)):
  58. fnid = taskData.tasks_fnid[task]
  59. fn = taskData.fn_index[fnid]
  60. task_deps = dataCache.task_deps[fn]
  61. if fnid not in taskData.failed_fnids:
  62. depends = taskData.tasks_tdepends[task]
  63. # Resolve Depends
  64. if 'deptask' in task_deps and taskData.tasks_name[task] in task_deps['deptask']:
  65. taskname = task_deps['deptask'][taskData.tasks_name[task]]
  66. for depid in taskData.depids[fnid]:
  67. if depid in taskData.build_targets:
  68. depdata = taskData.build_targets[depid][0]
  69. if depdata:
  70. dep = taskData.fn_index[depdata]
  71. depends.append(taskData.gettask_id(dep, taskname))
  72. # Resolve Runtime Depends
  73. if 'rdeptask' in task_deps and taskData.tasks_name[task] in task_deps['rdeptask']:
  74. taskname = task_deps['rdeptask'][taskData.tasks_name[task]]
  75. for depid in taskData.rdepids[fnid]:
  76. if depid in taskData.run_targets:
  77. depdata = taskData.run_targets[depid][0]
  78. if depdata:
  79. dep = taskData.fn_index[depdata]
  80. depends.append(taskData.gettask_id(dep, taskname))
  81. def add_recursive_build(depid):
  82. """
  83. Add build depends of depid to depends
  84. (if we've not see it before)
  85. (calls itself recursively)
  86. """
  87. if str(depid) in dep_seen:
  88. return
  89. dep_seen.append(depid)
  90. if depid in taskData.build_targets:
  91. depdata = taskData.build_targets[depid][0]
  92. if depdata:
  93. dep = taskData.fn_index[depdata]
  94. taskid = taskData.gettask_id(dep, taskname)
  95. depends.append(taskid)
  96. fnid = taskData.tasks_fnid[taskid]
  97. for nextdepid in taskData.depids[fnid]:
  98. if nextdepid not in dep_seen:
  99. add_recursive_build(nextdepid)
  100. for nextdepid in taskData.rdepids[fnid]:
  101. if nextdepid not in rdep_seen:
  102. add_recursive_run(nextdepid)
  103. def add_recursive_run(rdepid):
  104. """
  105. Add runtime depends of rdepid to depends
  106. (if we've not see it before)
  107. (calls itself recursively)
  108. """
  109. if str(rdepid) in rdep_seen:
  110. return
  111. rdep_seen.append(rdepid)
  112. if rdepid in taskData.run_targets:
  113. depdata = taskData.run_targets[rdepid][0]
  114. if depdata:
  115. dep = taskData.fn_index[depdata]
  116. taskid = taskData.gettask_id(dep, taskname)
  117. depends.append(taskid)
  118. fnid = taskData.tasks_fnid[taskid]
  119. for nextdepid in taskData.depids[fnid]:
  120. if nextdepid not in dep_seen:
  121. add_recursive_build(nextdepid)
  122. for nextdepid in taskData.rdepids[fnid]:
  123. if nextdepid not in rdep_seen:
  124. add_recursive_run(nextdepid)
  125. # Resolve Recursive Runtime Depends
  126. # Also includes all Build Depends (and their runtime depends)
  127. if 'recrdeptask' in task_deps and taskData.tasks_name[task] in task_deps['recrdeptask']:
  128. dep_seen = []
  129. rdep_seen = []
  130. taskname = task_deps['recrdeptask'][taskData.tasks_name[task]]
  131. for depid in taskData.depids[fnid]:
  132. add_recursive_build(depid)
  133. for rdepid in taskData.rdepids[fnid]:
  134. add_recursive_run(rdepid)
  135. #Prune self references
  136. if task in depends:
  137. newdep = []
  138. 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))
  139. for dep in depends:
  140. if task != dep:
  141. newdep.append(dep)
  142. depends = newdep
  143. self.runq_fnid.append(taskData.tasks_fnid[task])
  144. self.runq_task.append(taskData.tasks_name[task])
  145. self.runq_depends.append(Set(depends))
  146. self.runq_revdeps.append(Set())
  147. self.runq_weight.append(0)
  148. runq_weight1.append(0)
  149. runq_build.append(0)
  150. runq_done.append(0)
  151. bb.msg.note(2, bb.msg.domain.RunQueue, "Marking Active Tasks")
  152. def mark_active(listid, depth):
  153. """
  154. Mark an item as active along with its depends
  155. (calls itself recursively)
  156. """
  157. if runq_build[listid] == 1:
  158. return
  159. runq_build[listid] = 1
  160. depends = self.runq_depends[listid]
  161. for depend in depends:
  162. mark_active(depend, depth+1)
  163. for target in targets:
  164. targetid = taskData.getbuild_id(target[0])
  165. if targetid in taskData.failed_deps:
  166. continue
  167. if targetid not in taskData.build_targets:
  168. continue
  169. fnid = taskData.build_targets[targetid][0]
  170. if fnid in taskData.failed_fnids:
  171. continue
  172. listid = taskData.tasks_lookup[fnid][target[1]]
  173. mark_active(listid, 1)
  174. # Prune inactive tasks
  175. maps = []
  176. delcount = 0
  177. for listid in range(len(self.runq_fnid)):
  178. if runq_build[listid-delcount] == 1:
  179. maps.append(listid-delcount)
  180. else:
  181. del self.runq_fnid[listid-delcount]
  182. del self.runq_task[listid-delcount]
  183. del self.runq_depends[listid-delcount]
  184. del self.runq_weight[listid-delcount]
  185. del runq_weight1[listid-delcount]
  186. del runq_build[listid-delcount]
  187. del runq_done[listid-delcount]
  188. del self.runq_revdeps[listid-delcount]
  189. delcount = delcount + 1
  190. maps.append(-1)
  191. if len(self.runq_fnid) == 0:
  192. if not taskData.abort:
  193. 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.")
  194. return
  195. bb.msg.fatal(bb.msg.domain.RunQueue, "No active tasks and not in --continue mode?! Please report this bug.")
  196. bb.msg.note(2, bb.msg.domain.RunQueue, "Pruned %s inactive tasks, %s left" % (delcount, len(self.runq_fnid)))
  197. for listid in range(len(self.runq_fnid)):
  198. newdeps = []
  199. origdeps = self.runq_depends[listid]
  200. for origdep in origdeps:
  201. if maps[origdep] == -1:
  202. bb.msg.fatal(bb.msg.domain.RunQueue, "Invalid mapping - Should never happen!")
  203. newdeps.append(maps[origdep])
  204. self.runq_depends[listid] = Set(newdeps)
  205. bb.msg.note(2, bb.msg.domain.RunQueue, "Assign Weightings")
  206. for listid in range(len(self.runq_fnid)):
  207. for dep in self.runq_depends[listid]:
  208. self.runq_revdeps[dep].add(listid)
  209. endpoints = []
  210. for listid in range(len(self.runq_fnid)):
  211. revdeps = self.runq_revdeps[listid]
  212. if len(revdeps) == 0:
  213. runq_done[listid] = 1
  214. self.runq_weight[listid] = 1
  215. endpoints.append(listid)
  216. for dep in revdeps:
  217. if dep in self.runq_depends[listid]:
  218. #self.dump_data(taskData)
  219. 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]))
  220. runq_weight1[listid] = len(revdeps)
  221. bb.msg.note(2, bb.msg.domain.RunQueue, "Compute totals (have %s endpoint(s))" % len(endpoints))
  222. while 1:
  223. next_points = []
  224. for listid in endpoints:
  225. for revdep in self.runq_depends[listid]:
  226. self.runq_weight[revdep] = self.runq_weight[revdep] + self.runq_weight[listid]
  227. runq_weight1[revdep] = runq_weight1[revdep] - 1
  228. if runq_weight1[revdep] == 0:
  229. next_points.append(revdep)
  230. runq_done[revdep] = 1
  231. endpoints = next_points
  232. if len(next_points) == 0:
  233. break
  234. # Sanity Checks
  235. for task in range(len(self.runq_fnid)):
  236. if runq_done[task] == 0:
  237. seen = []
  238. deps_seen = []
  239. def print_chain(taskid, finish):
  240. seen.append(taskid)
  241. for revdep in self.runq_revdeps[taskid]:
  242. if runq_done[revdep] == 0 and revdep not in seen and not finish:
  243. bb.msg.error(bb.msg.domain.RunQueue, "Task %s (%s) (depends: %s)" % (revdep, self.get_user_idstring(revdep, taskData), self.runq_depends[revdep]))
  244. if revdep in deps_seen:
  245. bb.msg.error(bb.msg.domain.RunQueue, "Chain ends at Task %s (%s)" % (revdep, self.get_user_idstring(revdep, taskData)))
  246. finish = True
  247. return
  248. for dep in self.runq_depends[revdep]:
  249. deps_seen.append(dep)
  250. print_chain(revdep, finish)
  251. print_chain(task, False)
  252. 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)))
  253. if runq_weight1[task] != 0:
  254. bb.msg.fatal(bb.msg.domain.RunQueue, "Task %s (%s) count not zero!" % (task, self.get_user_idstring(task, taskData)))
  255. # Make a weight sorted map
  256. from copy import deepcopy
  257. sortweight = deepcopy(self.runq_weight)
  258. sortweight.sort()
  259. copyweight = deepcopy(self.runq_weight)
  260. self.prio_map = []
  261. for weight in sortweight:
  262. idx = copyweight.index(weight)
  263. self.prio_map.append(idx)
  264. copyweight[idx] = -1
  265. self.prio_map.reverse()
  266. #self.dump_data(taskData)
  267. def execute_runqueue(self, cooker, cfgData, dataCache, taskData, runlist):
  268. """
  269. Run the tasks in a queue prepared by prepare_runqueue
  270. Upon failure, optionally try to recover the build using any alternate providers
  271. (if the abort on failure configuration option isn't set)
  272. """
  273. failures = 0
  274. while 1:
  275. failed_fnids = self.execute_runqueue_internal(cooker, cfgData, dataCache, taskData)
  276. if len(failed_fnids) == 0:
  277. return failures
  278. if taskData.abort:
  279. raise bb.runqueue.TaskFailure(failed_fnids)
  280. for fnid in failed_fnids:
  281. #print "Failure: %s %s %s" % (fnid, taskData.fn_index[fnid], self.runq_task[fnid])
  282. taskData.fail_fnid(fnid)
  283. failures = failures + 1
  284. self.reset_runqueue()
  285. self.prepare_runqueue(cfgData, dataCache, taskData, runlist)
  286. def execute_runqueue_internal(self, cooker, cfgData, dataCache, taskData):
  287. """
  288. Run the tasks in a queue prepared by prepare_runqueue
  289. """
  290. import signal
  291. bb.msg.note(1, bb.msg.domain.RunQueue, "Executing runqueue")
  292. runq_buildable = []
  293. runq_running = []
  294. runq_complete = []
  295. active_builds = 0
  296. build_pids = {}
  297. failed_fnids = []
  298. if len(self.runq_fnid) == 0:
  299. # nothing to do
  300. return
  301. def sigint_handler(signum, frame):
  302. raise KeyboardInterrupt
  303. def get_next_task(data):
  304. """
  305. Return the id of the highest priority task that is buildable
  306. """
  307. for task1 in range(len(data.runq_fnid)):
  308. task = data.prio_map[task1]
  309. if runq_running[task] == 1:
  310. continue
  311. if runq_buildable[task] == 1:
  312. return task
  313. return None
  314. def task_complete(data, task):
  315. """
  316. Mark a task as completed
  317. Look at the reverse dependencies and mark any task with
  318. completed dependencies as buildable
  319. """
  320. runq_complete[task] = 1
  321. for revdep in data.runq_revdeps[task]:
  322. if runq_running[revdep] == 1:
  323. continue
  324. if runq_buildable[revdep] == 1:
  325. continue
  326. alldeps = 1
  327. for dep in data.runq_depends[revdep]:
  328. if runq_complete[dep] != 1:
  329. alldeps = 0
  330. if alldeps == 1:
  331. runq_buildable[revdep] = 1
  332. fn = taskData.fn_index[self.runq_fnid[revdep]]
  333. taskname = self.runq_task[revdep]
  334. bb.msg.debug(1, bb.msg.domain.RunQueue, "Marking task %s (%s, %s) as buildable" % (revdep, fn, taskname))
  335. # Mark initial buildable tasks
  336. for task in range(len(self.runq_fnid)):
  337. runq_running.append(0)
  338. runq_complete.append(0)
  339. if len(self.runq_depends[task]) == 0:
  340. runq_buildable.append(1)
  341. else:
  342. runq_buildable.append(0)
  343. number_tasks = int(bb.data.getVar("BB_NUMBER_THREADS", cfgData) or 1)
  344. try:
  345. while 1:
  346. task = get_next_task(self)
  347. if task is not None:
  348. fn = taskData.fn_index[self.runq_fnid[task]]
  349. taskname = self.runq_task[task]
  350. if bb.build.stamp_is_current_cache(dataCache, fn, taskname):
  351. targetid = taskData.gettask_id(fn, taskname)
  352. if not (targetid in taskData.external_targets and cooker.configuration.force):
  353. bb.msg.debug(2, bb.msg.domain.RunQueue, "Stamp current task %s (%s)" % (task, self.get_user_idstring(task, taskData)))
  354. runq_running[task] = 1
  355. task_complete(self, task)
  356. continue
  357. bb.msg.debug(1, bb.msg.domain.RunQueue, "Running task %s (%s)" % (task, self.get_user_idstring(task, taskData)))
  358. try:
  359. pid = os.fork()
  360. except OSError, e:
  361. bb.msg.fatal(bb.msg.domain.RunQueue, "fork failed: %d (%s)" % (e.errno, e.strerror))
  362. if pid == 0:
  363. # Bypass finally below
  364. active_builds = 0
  365. # Stop Ctrl+C being sent to children
  366. # signal.signal(signal.SIGINT, signal.SIG_IGN)
  367. # Make the child the process group leader
  368. os.setpgid(0, 0)
  369. sys.stdin = open('/dev/null', 'r')
  370. cooker.configuration.cmd = taskname[3:]
  371. try:
  372. cooker.tryBuild(fn, False)
  373. except bb.build.EventException:
  374. bb.msg.error(bb.msg.domain.Build, "Build of " + fn + " " + taskname + " failed")
  375. sys.exit(1)
  376. except:
  377. bb.msg.error(bb.msg.domain.Build, "Build of " + fn + " " + taskname + " failed")
  378. raise
  379. sys.exit(0)
  380. build_pids[pid] = task
  381. runq_running[task] = 1
  382. active_builds = active_builds + 1
  383. if active_builds < number_tasks:
  384. continue
  385. if active_builds > 0:
  386. result = os.waitpid(-1, 0)
  387. active_builds = active_builds - 1
  388. task = build_pids[result[0]]
  389. if result[1] != 0:
  390. del build_pids[result[0]]
  391. bb.msg.error(bb.msg.domain.RunQueue, "Task %s (%s) failed" % (task, self.get_user_idstring(task, taskData)))
  392. failed_fnids.append(self.runq_fnid[task])
  393. break
  394. task_complete(self, task)
  395. del build_pids[result[0]]
  396. continue
  397. break
  398. finally:
  399. try:
  400. while active_builds > 0:
  401. bb.msg.note(1, bb.msg.domain.RunQueue, "Waiting for %s active tasks to finish" % active_builds)
  402. tasknum = 1
  403. for k, v in build_pids.iteritems():
  404. bb.msg.note(1, bb.msg.domain.RunQueue, "%s: %s (%s)" % (tasknum, self.get_user_idstring(v, taskData), k))
  405. tasknum = tasknum + 1
  406. result = os.waitpid(-1, 0)
  407. task = build_pids[result[0]]
  408. if result[1] != 0:
  409. bb.msg.error(bb.msg.domain.RunQueue, "Task %s (%s) failed" % (task, self.get_user_idstring(task, taskData)))
  410. failed_fnids.append(self.runq_fnid[task])
  411. del build_pids[result[0]]
  412. active_builds = active_builds - 1
  413. if len(failed_fnids) > 0:
  414. return failed_fnids
  415. except:
  416. bb.msg.note(1, bb.msg.domain.RunQueue, "Sending SIGINT to remaining %s tasks" % active_builds)
  417. for k, v in build_pids.iteritems():
  418. os.kill(-k, signal.SIGINT)
  419. raise
  420. # Sanity Checks
  421. for task in range(len(self.runq_fnid)):
  422. if runq_buildable[task] == 0:
  423. bb.msg.error(bb.msg.domain.RunQueue, "Task %s never buildable!" % task)
  424. if runq_running[task] == 0:
  425. bb.msg.error(bb.msg.domain.RunQueue, "Task %s never ran!" % task)
  426. if runq_complete[task] == 0:
  427. bb.msg.error(bb.msg.domain.RunQueue, "Task %s never completed!" % task)
  428. return failed_fnids
  429. def dump_data(self, taskQueue):
  430. """
  431. Dump some debug information on the internal data structures
  432. """
  433. bb.msg.debug(3, bb.msg.domain.RunQueue, "run_tasks:")
  434. for task in range(len(self.runq_fnid)):
  435. bb.msg.debug(3, bb.msg.domain.RunQueue, " (%s)%s - %s: %s Deps %s RevDeps %s" % (task,
  436. taskQueue.fn_index[self.runq_fnid[task]],
  437. self.runq_task[task],
  438. self.runq_weight[task],
  439. self.runq_depends[task],
  440. self.runq_revdeps[task]))
  441. bb.msg.debug(3, bb.msg.domain.RunQueue, "sorted_tasks:")
  442. for task1 in range(len(self.runq_fnid)):
  443. if task1 in self.prio_map:
  444. task = self.prio_map[task1]
  445. bb.msg.debug(3, bb.msg.domain.RunQueue, " (%s)%s - %s: %s Deps %s RevDeps %s" % (task,
  446. taskQueue.fn_index[self.runq_fnid[task]],
  447. self.runq_task[task],
  448. self.runq_weight[task],
  449. self.runq_depends[task],
  450. self.runq_revdeps[task]))