Runner.py 16 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609
  1. #!/usr/bin/env python
  2. # encoding: utf-8
  3. # Thomas Nagy, 2005-2018 (ita)
  4. """
  5. Runner.py: Task scheduling and execution
  6. """
  7. import heapq, traceback
  8. try:
  9. from queue import Queue, PriorityQueue
  10. except ImportError:
  11. from Queue import Queue
  12. try:
  13. from Queue import PriorityQueue
  14. except ImportError:
  15. class PriorityQueue(Queue):
  16. def _init(self, maxsize):
  17. self.maxsize = maxsize
  18. self.queue = []
  19. def _put(self, item):
  20. heapq.heappush(self.queue, item)
  21. def _get(self):
  22. return heapq.heappop(self.queue)
  23. from waflib import Utils, Task, Errors, Logs
  24. GAP = 5
  25. """
  26. Wait for at least ``GAP * njobs`` before trying to enqueue more tasks to run
  27. """
  28. class PriorityTasks(object):
  29. def __init__(self):
  30. self.lst = []
  31. def __len__(self):
  32. return len(self.lst)
  33. def __iter__(self):
  34. return iter(self.lst)
  35. def __str__(self):
  36. return 'PriorityTasks: [%s]' % '\n '.join(str(x) for x in self.lst)
  37. def clear(self):
  38. self.lst = []
  39. def append(self, task):
  40. heapq.heappush(self.lst, task)
  41. def pop(self):
  42. return heapq.heappop(self.lst)
  43. def extend(self, lst):
  44. if self.lst:
  45. for x in lst:
  46. self.append(x)
  47. else:
  48. if isinstance(lst, list):
  49. self.lst = lst
  50. heapq.heapify(lst)
  51. else:
  52. self.lst = lst.lst
  53. class Consumer(Utils.threading.Thread):
  54. """
  55. Daemon thread object that executes a task. It shares a semaphore with
  56. the coordinator :py:class:`waflib.Runner.Spawner`. There is one
  57. instance per task to consume.
  58. """
  59. def __init__(self, spawner, task):
  60. Utils.threading.Thread.__init__(self)
  61. self.task = task
  62. """Task to execute"""
  63. self.spawner = spawner
  64. """Coordinator object"""
  65. self.daemon = True
  66. self.start()
  67. def run(self):
  68. """
  69. Processes a single task
  70. """
  71. try:
  72. if not self.spawner.master.stop:
  73. self.spawner.master.process_task(self.task)
  74. finally:
  75. self.spawner.sem.release()
  76. self.spawner.master.out.put(self.task)
  77. self.task = None
  78. self.spawner = None
  79. class Spawner(Utils.threading.Thread):
  80. """
  81. Daemon thread that consumes tasks from :py:class:`waflib.Runner.Parallel` producer and
  82. spawns a consuming thread :py:class:`waflib.Runner.Consumer` for each
  83. :py:class:`waflib.Task.Task` instance.
  84. """
  85. def __init__(self, master):
  86. Utils.threading.Thread.__init__(self)
  87. self.master = master
  88. """:py:class:`waflib.Runner.Parallel` producer instance"""
  89. self.sem = Utils.threading.Semaphore(master.numjobs)
  90. """Bounded semaphore that prevents spawning more than *n* concurrent consumers"""
  91. self.daemon = True
  92. self.start()
  93. def run(self):
  94. """
  95. Spawns new consumers to execute tasks by delegating to :py:meth:`waflib.Runner.Spawner.loop`
  96. """
  97. try:
  98. self.loop()
  99. except Exception:
  100. # Python 2 prints unnecessary messages when shutting down
  101. # we also want to stop the thread properly
  102. pass
  103. def loop(self):
  104. """
  105. Consumes task objects from the producer; ends when the producer has no more
  106. task to provide.
  107. """
  108. master = self.master
  109. while 1:
  110. task = master.ready.get()
  111. self.sem.acquire()
  112. if not master.stop:
  113. task.log_display(task.generator.bld)
  114. Consumer(self, task)
  115. class Parallel(object):
  116. """
  117. Schedule the tasks obtained from the build context for execution.
  118. """
  119. def __init__(self, bld, j=2):
  120. """
  121. The initialization requires a build context reference
  122. for computing the total number of jobs.
  123. """
  124. self.numjobs = j
  125. """
  126. Amount of parallel consumers to use
  127. """
  128. self.bld = bld
  129. """
  130. Instance of :py:class:`waflib.Build.BuildContext`
  131. """
  132. self.outstanding = PriorityTasks()
  133. """Heap of :py:class:`waflib.Task.Task` that may be ready to be executed"""
  134. self.postponed = PriorityTasks()
  135. """Heap of :py:class:`waflib.Task.Task` which are not ready to run for non-DAG reasons"""
  136. self.incomplete = set()
  137. """List of :py:class:`waflib.Task.Task` waiting for dependent tasks to complete (DAG)"""
  138. self.ready = PriorityQueue(0)
  139. """List of :py:class:`waflib.Task.Task` ready to be executed by consumers"""
  140. self.out = Queue(0)
  141. """List of :py:class:`waflib.Task.Task` returned by the task consumers"""
  142. self.count = 0
  143. """Amount of tasks that may be processed by :py:class:`waflib.Runner.TaskConsumer`"""
  144. self.processed = 0
  145. """Amount of tasks processed"""
  146. self.stop = False
  147. """Error flag to stop the build"""
  148. self.error = []
  149. """Tasks that could not be executed"""
  150. self.biter = None
  151. """Task iterator which must give groups of parallelizable tasks when calling ``next()``"""
  152. self.dirty = False
  153. """
  154. Flag that indicates that the build cache must be saved when a task was executed
  155. (calls :py:meth:`waflib.Build.BuildContext.store`)"""
  156. self.revdeps = Utils.defaultdict(set)
  157. """
  158. The reverse dependency graph of dependencies obtained from Task.run_after
  159. """
  160. self.spawner = None
  161. """
  162. Coordinating daemon thread that spawns thread consumers
  163. """
  164. if self.numjobs > 1:
  165. self.spawner = Spawner(self)
  166. def get_next_task(self):
  167. """
  168. Obtains the next Task instance to run
  169. :rtype: :py:class:`waflib.Task.Task`
  170. """
  171. if not self.outstanding:
  172. return None
  173. return self.outstanding.pop()
  174. def postpone(self, tsk):
  175. """
  176. Adds the task to the list :py:attr:`waflib.Runner.Parallel.postponed`.
  177. The order is scrambled so as to consume as many tasks in parallel as possible.
  178. :param tsk: task instance
  179. :type tsk: :py:class:`waflib.Task.Task`
  180. """
  181. self.postponed.append(tsk)
  182. def refill_task_list(self):
  183. """
  184. Pulls a next group of tasks to execute in :py:attr:`waflib.Runner.Parallel.outstanding`.
  185. Ensures that all tasks in the current build group are complete before processing the next one.
  186. """
  187. while self.count > self.numjobs * GAP:
  188. self.get_out()
  189. while not self.outstanding:
  190. if self.count:
  191. self.get_out()
  192. if self.outstanding:
  193. break
  194. elif self.postponed:
  195. try:
  196. cond = self.deadlock == self.processed
  197. except AttributeError:
  198. pass
  199. else:
  200. if cond:
  201. # The most common reason is conflicting build order declaration
  202. # for example: "X run_after Y" and "Y run_after X"
  203. # Another can be changing "run_after" dependencies while the build is running
  204. # for example: updating "tsk.run_after" in the "runnable_status" method
  205. lst = []
  206. for tsk in self.postponed:
  207. deps = [id(x) for x in tsk.run_after if not x.hasrun]
  208. lst.append('%s\t-> %r' % (repr(tsk), deps))
  209. if not deps:
  210. lst.append('\n task %r dependencies are done, check its *runnable_status*?' % id(tsk))
  211. raise Errors.WafError('Deadlock detected: check the task build order%s' % ''.join(lst))
  212. self.deadlock = self.processed
  213. if self.postponed:
  214. self.outstanding.extend(self.postponed)
  215. self.postponed.clear()
  216. elif not self.count:
  217. if self.incomplete:
  218. for x in self.incomplete:
  219. for k in x.run_after:
  220. if not k.hasrun:
  221. break
  222. else:
  223. # dependency added after the build started without updating revdeps
  224. self.incomplete.remove(x)
  225. self.outstanding.append(x)
  226. break
  227. else:
  228. if self.stop or self.error:
  229. break
  230. raise Errors.WafError('Broken revdeps detected on %r' % self.incomplete)
  231. else:
  232. tasks = next(self.biter)
  233. ready, waiting = self.prio_and_split(tasks)
  234. self.outstanding.extend(ready)
  235. self.incomplete.update(waiting)
  236. self.total = self.bld.total()
  237. break
  238. def add_more_tasks(self, tsk):
  239. """
  240. If a task provides :py:attr:`waflib.Task.Task.more_tasks`, then the tasks contained
  241. in that list are added to the current build and will be processed before the next build group.
  242. The priorities for dependent tasks are not re-calculated globally
  243. :param tsk: task instance
  244. :type tsk: :py:attr:`waflib.Task.Task`
  245. """
  246. if getattr(tsk, 'more_tasks', None):
  247. more = set(tsk.more_tasks)
  248. groups_done = set()
  249. def iteri(a, b):
  250. for x in a:
  251. yield x
  252. for x in b:
  253. yield x
  254. # Update the dependency tree
  255. # this assumes that task.run_after values were updated
  256. for x in iteri(self.outstanding, self.incomplete):
  257. for k in x.run_after:
  258. if isinstance(k, Task.TaskGroup):
  259. if k not in groups_done:
  260. groups_done.add(k)
  261. for j in k.prev & more:
  262. self.revdeps[j].add(k)
  263. elif k in more:
  264. self.revdeps[k].add(x)
  265. ready, waiting = self.prio_and_split(tsk.more_tasks)
  266. self.outstanding.extend(ready)
  267. self.incomplete.update(waiting)
  268. self.total += len(tsk.more_tasks)
  269. def mark_finished(self, tsk):
  270. def try_unfreeze(x):
  271. # DAG ancestors are likely to be in the incomplete set
  272. # This assumes that the run_after contents have not changed
  273. # after the build starts, else a deadlock may occur
  274. if x in self.incomplete:
  275. # TODO remove dependencies to free some memory?
  276. # x.run_after.remove(tsk)
  277. for k in x.run_after:
  278. if not k.hasrun:
  279. break
  280. else:
  281. self.incomplete.remove(x)
  282. self.outstanding.append(x)
  283. if tsk in self.revdeps:
  284. for x in self.revdeps[tsk]:
  285. if isinstance(x, Task.TaskGroup):
  286. x.prev.remove(tsk)
  287. if not x.prev:
  288. for k in x.next:
  289. # TODO necessary optimization?
  290. k.run_after.remove(x)
  291. try_unfreeze(k)
  292. # TODO necessary optimization?
  293. x.next = []
  294. else:
  295. try_unfreeze(x)
  296. del self.revdeps[tsk]
  297. if hasattr(tsk, 'semaphore'):
  298. sem = tsk.semaphore
  299. try:
  300. sem.release(tsk)
  301. except KeyError:
  302. # TODO
  303. pass
  304. else:
  305. while sem.waiting and not sem.is_locked():
  306. # take a frozen task, make it ready to run
  307. x = sem.waiting.pop()
  308. self.add_task(x)
  309. def get_out(self):
  310. """
  311. Waits for a Task that task consumers add to :py:attr:`waflib.Runner.Parallel.out` after execution.
  312. Adds more Tasks if necessary through :py:attr:`waflib.Runner.Parallel.add_more_tasks`.
  313. :rtype: :py:attr:`waflib.Task.Task`
  314. """
  315. tsk = self.out.get()
  316. if not self.stop:
  317. self.add_more_tasks(tsk)
  318. self.mark_finished(tsk)
  319. self.count -= 1
  320. self.dirty = True
  321. return tsk
  322. def add_task(self, tsk):
  323. if hasattr(tsk, 'semaphore'):
  324. sem = tsk.semaphore
  325. try:
  326. sem.acquire(tsk)
  327. except IndexError:
  328. sem.waiting.add(tsk)
  329. return
  330. self.count += 1
  331. self.processed += 1
  332. if self.numjobs == 1:
  333. tsk.log_display(tsk.generator.bld)
  334. try:
  335. self.process_task(tsk)
  336. finally:
  337. self.out.put(tsk)
  338. else:
  339. self.ready.put(tsk)
  340. def process_task(self, tsk):
  341. """
  342. Processes a task and attempts to stop the build in case of errors
  343. """
  344. tsk.process()
  345. if tsk.hasrun != Task.SUCCESS:
  346. self.error_handler(tsk)
  347. def skip(self, tsk):
  348. """
  349. Mark a task as skipped/up-to-date
  350. """
  351. tsk.hasrun = Task.SKIPPED
  352. self.mark_finished(tsk)
  353. def cancel(self, tsk):
  354. """
  355. Mark a task as failed because of unsatisfiable dependencies
  356. """
  357. tsk.hasrun = Task.CANCELED
  358. self.mark_finished(tsk)
  359. def error_handler(self, tsk):
  360. """
  361. Called when a task cannot be executed. The flag :py:attr:`waflib.Runner.Parallel.stop` is set,
  362. unless the build is executed with::
  363. $ waf build -k
  364. :param tsk: task instance
  365. :type tsk: :py:attr:`waflib.Task.Task`
  366. """
  367. if not self.bld.keep:
  368. self.stop = True
  369. self.error.append(tsk)
  370. def task_status(self, tsk):
  371. """
  372. Obtains the task status to decide whether to run it immediately or not.
  373. :return: the exit status, for example :py:attr:`waflib.Task.ASK_LATER`
  374. :rtype: integer
  375. """
  376. try:
  377. return tsk.runnable_status()
  378. except Exception:
  379. self.processed += 1
  380. tsk.err_msg = traceback.format_exc()
  381. if not self.stop and self.bld.keep:
  382. self.skip(tsk)
  383. if self.bld.keep == 1:
  384. # if -k stop on the first exception, if -kk try to go as far as possible
  385. if Logs.verbose > 1 or not self.error:
  386. self.error.append(tsk)
  387. self.stop = True
  388. else:
  389. if Logs.verbose > 1:
  390. self.error.append(tsk)
  391. return Task.EXCEPTION
  392. tsk.hasrun = Task.EXCEPTION
  393. self.error_handler(tsk)
  394. return Task.EXCEPTION
  395. def start(self):
  396. """
  397. Obtains Task instances from the BuildContext instance and adds the ones that need to be executed to
  398. :py:class:`waflib.Runner.Parallel.ready` so that the :py:class:`waflib.Runner.Spawner` consumer thread
  399. has them executed. Obtains the executed Tasks back from :py:class:`waflib.Runner.Parallel.out`
  400. and marks the build as failed by setting the ``stop`` flag.
  401. If only one job is used, then executes the tasks one by one, without consumers.
  402. """
  403. self.total = self.bld.total()
  404. while not self.stop:
  405. self.refill_task_list()
  406. # consider the next task
  407. tsk = self.get_next_task()
  408. if not tsk:
  409. if self.count:
  410. # tasks may add new ones after they are run
  411. continue
  412. else:
  413. # no tasks to run, no tasks running, time to exit
  414. break
  415. if tsk.hasrun:
  416. # if the task is marked as "run", just skip it
  417. self.processed += 1
  418. continue
  419. if self.stop: # stop immediately after a failure is detected
  420. break
  421. st = self.task_status(tsk)
  422. if st == Task.RUN_ME:
  423. self.add_task(tsk)
  424. elif st == Task.ASK_LATER:
  425. self.postpone(tsk)
  426. elif st == Task.SKIP_ME:
  427. self.processed += 1
  428. self.skip(tsk)
  429. self.add_more_tasks(tsk)
  430. elif st == Task.CANCEL_ME:
  431. # A dependency problem has occurred, and the
  432. # build is most likely run with `waf -k`
  433. if Logs.verbose > 1:
  434. self.error.append(tsk)
  435. self.processed += 1
  436. self.cancel(tsk)
  437. # self.count represents the tasks that have been made available to the consumer threads
  438. # collect all the tasks after an error else the message may be incomplete
  439. while self.error and self.count:
  440. self.get_out()
  441. self.ready.put(None)
  442. if not self.stop:
  443. assert not self.count
  444. assert not self.postponed
  445. assert not self.incomplete
  446. def prio_and_split(self, tasks):
  447. """
  448. Label input tasks with priority values, and return a pair containing
  449. the tasks that are ready to run and the tasks that are necessarily
  450. waiting for other tasks to complete.
  451. The priority system is really meant as an optional layer for optimization:
  452. dependency cycles are found quickly, and builds should be more efficient.
  453. A high priority number means that a task is processed first.
  454. This method can be overridden to disable the priority system::
  455. def prio_and_split(self, tasks):
  456. return tasks, []
  457. :return: A pair of task lists
  458. :rtype: tuple
  459. """
  460. # to disable:
  461. #return tasks, []
  462. for x in tasks:
  463. x.visited = 0
  464. reverse = self.revdeps
  465. groups_done = set()
  466. for x in tasks:
  467. for k in x.run_after:
  468. if isinstance(k, Task.TaskGroup):
  469. if k not in groups_done:
  470. groups_done.add(k)
  471. for j in k.prev:
  472. reverse[j].add(k)
  473. else:
  474. reverse[k].add(x)
  475. # the priority number is not the tree depth
  476. def visit(n):
  477. if isinstance(n, Task.TaskGroup):
  478. return sum(visit(k) for k in n.next)
  479. if n.visited == 0:
  480. n.visited = 1
  481. if n in reverse:
  482. rev = reverse[n]
  483. n.prio_order = n.tree_weight + len(rev) + sum(visit(k) for k in rev)
  484. else:
  485. n.prio_order = n.tree_weight
  486. n.visited = 2
  487. elif n.visited == 1:
  488. raise Errors.WafError('Dependency cycle found!')
  489. return n.prio_order
  490. for x in tasks:
  491. if x.visited != 0:
  492. # must visit all to detect cycles
  493. continue
  494. try:
  495. visit(x)
  496. except Errors.WafError:
  497. self.debug_cycles(tasks, reverse)
  498. ready = []
  499. waiting = []
  500. for x in tasks:
  501. for k in x.run_after:
  502. if not k.hasrun:
  503. waiting.append(x)
  504. break
  505. else:
  506. ready.append(x)
  507. return (ready, waiting)
  508. def debug_cycles(self, tasks, reverse):
  509. tmp = {}
  510. for x in tasks:
  511. tmp[x] = 0
  512. def visit(n, acc):
  513. if isinstance(n, Task.TaskGroup):
  514. for k in n.next:
  515. visit(k, acc)
  516. return
  517. if tmp[n] == 0:
  518. tmp[n] = 1
  519. for k in reverse.get(n, []):
  520. visit(k, [n] + acc)
  521. tmp[n] = 2
  522. elif tmp[n] == 1:
  523. lst = []
  524. for tsk in acc:
  525. lst.append(repr(tsk))
  526. if tsk is n:
  527. # exclude prior nodes, we want the minimum cycle
  528. break
  529. raise Errors.WafError('Task dependency cycle in "run_after" constraints: %s' % ''.join(lst))
  530. for x in tasks:
  531. visit(x, [])