code.vuplus.com Git - vuplus_bitbake/blob - lib/bb/runqueue.py

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