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 import bb, os, sys
  24
  25 class TaskFailure(Exception):
  26     """Exception raised when a task in a runqueue fails"""
  27
  28     def __init__(self, fnid, taskname):
  29         self.args = fnid, taskname
  30
  31 class RunQueue:
  32     """
  33     BitBake Run Queue implementation
  34     """
  35     def __init__(self):
  36         self.reset_runqueue()
  37
  38     def reset_runqueue(self):
  39         self.runq_fnid = []
  40         self.runq_task = []
  41         self.runq_depends = []
  42         self.runq_revdeps = []
  43         self.runq_weight = []
  44         self.prio_map = []
  45
  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
  53         runq_weight1 = []
  54         runq_build = []
  55         runq_done = []
  56
  57         bb.msg.note(1, bb.msg.domain.RunQueue, "Preparing Runqueue")
  58
  59         for task in range(len(taskData.tasks_name)):
  60             fnid = taskData.tasks_fnid[task]
  61             fn = taskData.fn_index[fnid]
  62             task_deps = dataCache.task_deps[fn]
  63
  64             if fnid not in taskData.failed_fnids:
  65
  66                 depends = taskData.tasks_tdepends[task]
  67
  68                 # Resolve Depends
  69                 if 'deptask' in task_deps and taskData.tasks_name[task] in task_deps['deptask']:
  70                     taskname = task_deps['deptask'][taskData.tasks_name[task]]
  71                     for depid in taskData.depids[fnid]:
  72                         if depid in taskData.build_targets:
  73                             depdata = taskData.build_targets[depid][0]
  74                             if depdata:
  75                                 dep = taskData.fn_index[depdata]
  76                                 depends.append(taskData.gettask_id(dep, taskname))
  77
  78                 # Resolve Runtime Depends
  79                 if 'rdeptask' in task_deps and taskData.tasks_name[task] in task_deps['rdeptask']:
  80                     taskname = task_deps['rdeptask'][taskData.tasks_name[task]]
  81                     for depid in taskData.rdepids[fnid]:
  82                         if depid in taskData.run_targets:
  83                             depdata = taskData.run_targets[depid][0]
  84                             if depdata:
  85                                 dep = taskData.fn_index[depdata]
  86                                 depends.append(taskData.gettask_id(dep, taskname))
  87
  88                 def add_recursive_run(rdepid):
  89                     """
  90                     Add runtime depends of rdepid to depends, if
  91                     we've not see it before
  92                     (calls itself recursively)
  93                     """
  94                     if str(rdepid) in rdep_seen:
  95                         return
  96                     rdep_seen.append(rdepid)
  97                     if rdepid in taskData.run_targets:
  98                         depdata = taskData.run_targets[rdepid][0]
  99                         if depdata:
 100                             dep = taskData.fn_index[depdata]
 101                             taskid = taskData.gettask_id(dep, taskname)
 102                             depends.append(taskid)
 103                             fnid = taskData.tasks_fnid[taskid]
 104                             for nextdepid in taskData.rdepids[fnid]:
 105                                 if nextdepid not in rdep_seen:
 106                                     add_recursive_run(nextdepid)
 107
 108                 # Resolve Recursive Runtime Depends
 109                 if 'recrdeptask' in task_deps and taskData.tasks_name[task] in task_deps['recrdeptask']:
 110                     rdep_seen = []
 111                     taskname = task_deps['recrdeptask'][taskData.tasks_name[task]]
 112                     for rdepid in taskData.rdepids[fnid]:
 113                         add_recursive_run(rdepid)
 114
 115                 #Prune self references
 116                 if task in depends:
 117                     newdep = []
 118                     bb.debug(2, "Task %s (%s %s) contains self reference! %s" % (task, taskData.fn_index[taskData.tasks_fnid[task]], taskData.tasks_name[task], depends))
 119                     for dep in depends:
 120                        if task != dep:
 121                           newdep.append(dep)
 122                     depends = newdep
 123
 124
 125             self.runq_fnid.append(taskData.tasks_fnid[task])
 126             self.runq_task.append(taskData.tasks_name[task])
 127             self.runq_depends.append(depends)
 128             self.runq_revdeps.append([])
 129             self.runq_weight.append(0)
 130
 131             runq_weight1.append(0)
 132             runq_build.append(0)
 133             runq_done.append(0)
 134
 135         bb.msg.note(2, bb.msg.domain.RunQueue, "Marking Active Tasks")
 136
 137         def mark_active(listid, depth):
 138             """
 139             Mark an item as active along with its depends
 140             (calls itself recursively)
 141             """
 142
 143             if runq_build[listid] == 1:
 144                 return
 145
 146             runq_build[listid] = 1
 147
 148             depends = self.runq_depends[listid]
 149             for depend in depends:
 150                 mark_active(depend, depth+1)
 151
 152         for target in targets:
 153             targetid = taskData.getbuild_id(target[0])
 154             if targetid in taskData.failed_deps:
 155                 continue
 156
 157             fnid = taskData.build_targets[targetid][0]
 158             if fnid in taskData.failed_fnids:
 159                 continue
 160
 161             fnids = taskData.matches_in_list(self.runq_fnid, fnid)
 162             tasks = taskData.matches_in_list(self.runq_task, target[1])
 163
 164             listid = taskData.both_contain(fnids, tasks)
 165
 166             mark_active(listid, 1)
 167
 168         # Prune inactive tasks
 169         maps = []
 170         delcount = 0
 171         for listid in range(len(self.runq_fnid)):
 172             if runq_build[listid-delcount] == 1:
 173                 maps.append(listid-delcount)
 174             else:
 175                 del self.runq_fnid[listid-delcount]
 176                 del self.runq_task[listid-delcount]
 177                 del self.runq_depends[listid-delcount]
 178                 del self.runq_weight[listid-delcount]
 179                 del runq_weight1[listid-delcount]
 180                 del runq_build[listid-delcount]
 181                 del runq_done[listid-delcount]
 182                 del self.runq_revdeps[listid-delcount]
 183                 delcount = delcount + 1
 184                 maps.append(-1)
 185
 186         bb.msg.note(2, bb.msg.domain.RunQueue, "Pruned %s inactive tasks, %s left" % (delcount, len(self.runq_fnid)))
 187
 188         for listid in range(len(self.runq_fnid)):
 189             newdeps = []
 190             origdeps = self.runq_depends[listid]
 191             for origdep in origdeps:
 192                 if maps[origdep] == -1:
 193                     bb.fatal(bb.msg.domain.RunQueue, "Invalid mapping - Should never happen!")
 194                 newdeps.append(maps[origdep])
 195             self.runq_depends[listid] = newdeps
 196
 197         bb.msg.note(2, bb.msg.domain.RunQueue, "Assign Weightings")
 198
 199         for listid in range(len(self.runq_fnid)):
 200             for dep in self.runq_depends[listid]:
 201                 if dep not in self.runq_revdeps[dep]:
 202                     self.runq_revdeps[dep].append(listid)
 203
 204         endpoints = []
 205         for listid in range(len(self.runq_fnid)):
 206             revdeps = self.runq_revdeps[listid]
 207             if len(revdeps) == 0:
 208                 runq_done[listid] = 1
 209                 self.runq_weight[listid] = 1
 210                 endpoints.append(listid)
 211             for dep in revdeps:
 212                 if dep in self.runq_depends[listid]:
 213                     self.dump_data(taskData)
 214                     bb.fatal("Task %s has circular dependency on %s" % (dep, listid))
 215             runq_weight1[listid] = len(revdeps)
 216
 217         bb.msg.note(2, bb.msg.domain.RunQueue, "Compute totals (have %s endpoint(s))" % len(endpoints))
 218
 219         while 1:
 220             next_points = []
 221             for listid in endpoints:
 222                 for revdep in self.runq_depends[listid]:
 223                     self.runq_weight[revdep] = self.runq_weight[revdep] + self.runq_weight[listid]
 224                     runq_weight1[revdep] = runq_weight1[revdep] - 1
 225                     if runq_weight1[revdep] == 0:
 226                         next_points.append(revdep)
 227                         runq_done[revdep] = 1
 228             endpoints = next_points
 229             if len(next_points) == 0:
 230                 break
 231
 232         # Sanity Checks
 233         for task in range(len(self.runq_fnid)):
 234             if runq_done[task] == 0:
 235                 bb.fatal("Task %s not processed!" % task)
 236             if runq_weight1[task] != 0:
 237                 bb.fatal("Task %s count not zero!" % task)
 238
 239         # Make a weight sorted map
 240         from copy import deepcopy
 241
 242         sortweight = deepcopy(self.runq_weight)
 243         sortweight.sort()
 244         copyweight = deepcopy(self.runq_weight)
 245         self.prio_map = []
 246
 247         for weight in sortweight:
 248             idx = copyweight.index(weight)
 249             self.prio_map.append(idx)
 250             copyweight[idx] = -1
 251         self.prio_map.reverse()
 252
 253     def execute_runqueue(self, cooker, cfgData, dataCache, taskData, runlist):
 254         """
 255         Run the tasks in a queue prepared by prepare_runqueue
 256         Upon failure, optionally try to recover the build using any alternate providers
 257         (if the abort on failure configuration option isn't set)
 258         """
 259
 260         failures = 0
 261         while 1:
 262             try:
 263                 self.execute_runqueue_internal(cooker, cfgData, dataCache, taskData)
 264                 return failures
 265             except bb.runqueue.TaskFailure, (fnid, taskname):
 266                 taskData.fail_fnid(fnid)
 267                 self.reset_runqueue()
 268                 self.prepare_runqueue(cfgData, dataCache, taskData, runlist)
 269                 failures = failures + 1
 270
 271     def execute_runqueue_internal(self, cooker, cfgData, dataCache, taskData):
 272         """
 273         Run the tasks in a queue prepared by prepare_runqueue
 274         """
 275
 276         bb.msg.note(1, bb.msg.domain.RunQueue, "Executing runqueue")
 277
 278         runq_buildable = []
 279         runq_running = []
 280         runq_complete = []
 281         active_builds = 0
 282         build_pids = {}
 283
 284         def get_next_task(data):
 285             """
 286             Return the id of the highest priority task that is buildable
 287             """
 288             for task1 in range(len(data.runq_fnid)):
 289                 task = data.prio_map[task1]
 290                 if runq_running[task] == 1:
 291                     continue
 292                 if runq_buildable[task] == 1:
 293                     return task
 294             return None
 295
 296         def task_complete(data, task):
 297             """
 298             Mark a task as completed
 299             Look at the reverse dependencies and mark any task with
 300             completed dependencies as buildable
 301             """
 302             runq_complete[task] = 1
 303             for revdep in data.runq_revdeps[task]:
 304                 if runq_running[revdep] == 1:
 305                     continue
 306                 if runq_buildable[revdep] == 1:
 307                     continue
 308                 alldeps = 1
 309                 for dep in data.runq_depends[revdep]:
 310                     if runq_complete[dep] != 1:
 311                         alldeps = 0
 312                 if alldeps == 1:
 313                     runq_buildable[revdep] = 1
 314                     fn = taskData.fn_index[self.runq_fnid[revdep]]
 315                     taskname = self.runq_task[revdep]
 316                     bb.msg.debug(1, bb.msg.domain.RunQueue, "Marking task %s (%s, %s) as buildable" % (revdep, fn, taskname))
 317
 318         # Mark initial buildable tasks
 319         for task in range(len(self.runq_fnid)):
 320             runq_running.append(0)
 321             runq_complete.append(0)
 322             if len(self.runq_depends[task]) == 0:
 323                 runq_buildable.append(1)
 324             else:
 325                 runq_buildable.append(0)
 326
 327         def get_user_idstring(task):
 328             fn = taskData.fn_index[self.runq_fnid[task]]
 329             taskname = self.runq_task[task]
 330             return "%s, %s" % (fn, taskname)
 331
 332         number_tasks = bb.data.getVar("BB_NUMBER_THREADS", cfgData)
 333         if not number_tasks:
 334             number_tasks = 1
 335
 336         try:
 337             while 1:
 338                 task = get_next_task(self)
 339                 if task is not None:
 340                     fn = taskData.fn_index[self.runq_fnid[task]]
 341                     taskname = self.runq_task[task]
 342
 343                     if bb.build.stamp_is_current_cache(dataCache, fn, taskname):
 344                         bb.msg.debug(2, bb.msg.domain.RunQueue, "Stamp current task %s (%s)" % (task, get_user_idstring(task)))
 345                         runq_running[task] = 1
 346                         task_complete(self, task)
 347                         continue
 348
 349                     bb.msg.debug(1, bb.msg.domain.RunQueue, "Running task %s (%s)" % (task, get_user_idstring(task)))
 350                     try:
 351                         pid = os.fork()
 352                     except OSError, e:
 353                         bb.msg.fatal(bb.msg.domain.RunQueue, "fork failed: %d (%s)" % (e.errno, e.strerror))
 354                     if pid == 0:
 355                         cooker.configuration.cmd = taskname[3:]
 356                         try:
 357                             cooker.tryBuild(fn, False)
 358                         except bb.build.EventException, e:
 359                             bb.msg.error(bb.msg.domain.Build, "Build of " + fn + " " + taskname + " failed")
 360                             sys.exit(1)
 361                         except:
 362                             sys.exit(1)
 363                         sys.exit(0)
 364                     build_pids[pid] = task
 365                     runq_running[task] = 1
 366                     active_builds = active_builds + 1
 367                     if active_builds < number_tasks:
 368                         continue
 369                 if active_builds > 0:
 370                     result = os.waitpid(-1, 0)
 371                     active_builds = active_builds - 1
 372                     if result[1] != 0:
 373                         bb.msg.error(bb.msg.domain.RunQueue, "Task %s (%s) failed" % (build_pids[result[0]], get_user_idstring(build_pids[result[0]])))
 374                         raise bb.runqueue.TaskFailure(self.runq_fnid[build_pids[result[0]]], self.runq_task[build_pids[result[0]]])
 375                     task_complete(self, build_pids[result[0]])
 376                     continue
 377                 break
 378         except SystemExit:
 379             raise
 380         except:
 381             bb.error("Exception received")
 382             if active_builds > 0:
 383                 while active_builds > 0:
 384                     bb.note("Waiting for %s active tasks to finish" % active_builds)
 385                     os.waitpid(-1, 0)
 386                     active_builds = active_builds - 1
 387             if cooker.configuration.abort:
 388                 sys.exit(1)
 389             raise
 390
 391         # Sanity Checks
 392         for task in range(len(self.runq_fnid)):
 393             if runq_buildable[task] == 0:
 394                 bb.msg.error(bb.msg.domain.RunQueue, "Task %s never buildable!" % task)
 395             if runq_running[task] == 0:
 396                 bb.msg.error(bb.msg.domain.RunQueue, "Task %s never ran!" % task)
 397             if runq_complete[task] == 0:
 398                 bb.msg.error(bb.msg.domain.RunQueue, "Task %s never completed!" % task)
 399
 400         return 0
 401
 402     def dump_data(self, taskQueue):
 403         """
 404         Dump some debug information on the internal data structures
 405         """
 406         bb.msg.debug(3, bb.msg.domain.RunQueue, "run_tasks:")
 407         for task in range(len(self.runq_fnid)):
 408                 bb.msg.debug(3, bb.msg.domain.RunQueue, " (%s)%s - %s: %s   Deps %s RevDeps %s" % (task,
 409                         taskQueue.fn_index[self.runq_fnid[task]],
 410                         self.runq_task[task],
 411                         self.runq_weight[task],
 412                         self.runq_depends[task],
 413                         self.runq_revdeps[task]))
 414
 415         bb.msg.debug(3, bb.msg.domain.RunQueue, "sorted_tasks:")
 416         for task1 in range(len(self.runq_fnid)):
 417             task = self.prio_map[task1]
 418             bb.msg.debug(3, bb.msg.domain.RunQueue, " (%s)%s - %s: %s   Deps %s RevDeps %s" % (task,
 419                         taskQueue.fn_index[self.runq_fnid[task]],
 420                         self.runq_task[task],
 421                         self.runq_weight[task],
 422                         self.runq_depends[task],
 423                         self.runq_revdeps[task]))