Add --show-output option to parallel_emerge which prints output from jobs.

parallel_emerge

 #!/usr/bin/python2.6
 # Copyright (c) 2010 The Chromium OS Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
 """Program to run emerge in parallel, for significant speedup.
 
 Usage:
  ./parallel_emerge [--board=BOARD] [--workon=PKGS] [--no-workon-deps]
                    [emerge args] package"
@@ skipping 20 matching lines @@
       or "--nodeps" write only access.
   Caveats:
     * Some ebuild packages have incorrectly specified deps, and running
       them in parallel is more likely to bring out these failures.
     * Some ebuilds (especially the build part) have complex dependencies
       that are not captured well by this script (it may be necessary to
       install an old package to build, but then install a newer version
       of the same package for a runtime dep).
 """
 
+import codecs
 import copy
 import multiprocessing
 import os
 import Queue
 import shlex
+import signal
 import sys
 import tempfile
 import time
+import traceback
 import urllib2
 
 # If PORTAGE_USERNAME isn't specified, scrape it from the $HOME variable. On
 # Chromium OS, the default "portage" user doesn't have the necessary
 # permissions. It'd be easier if we could default to $USERNAME, but $USERNAME
 # is "root" here because we get called through sudo.
 #
 # We need to set this before importing any portage modules, because portage
 # looks up "PORTAGE_USERNAME" at import time.
 #
@@ skipping 148 matching lines @@
   Typical usage:
     deps = DepGraphGenerator()
     deps.Initialize(sys.argv[1:])
     deps_tree, deps_info = deps.GenDependencyTree()
     deps_graph = deps.GenDependencyGraph(deps_tree, deps_info)
     deps.PrintTree(deps_tree)
     PrintDepsMap(deps_graph)
   """
 
   __slots__ = ["board", "emerge", "mandatory_source", "no_workon_deps",
-               "package_db", "rebuild"]
+               "package_db", "rebuild", "show_output"]
 
   def __init__(self):
     self.board = None
     self.emerge = EmergeData()
     self.mandatory_source = set()
     self.no_workon_deps = False
     self.package_db = {}
     self.rebuild = False
+    self.show_output = False
 
   def ParseParallelEmergeArgs(self, argv):
     """Read the parallel emerge arguments from the command-line.
 
     We need to be compatible with emerge arg format.  We scrape arguments that
     are specific to parallel_emerge, and pass through the rest directly to
     emerge.
     Args:
       argv: arguments list
     Returns:
       Arguments that don't belong to parallel_emerge
     """
     emerge_args = []
     for arg in argv:
       # Specifically match arguments that are specific to parallel_emerge, and
       # pass through the rest.
       if arg.startswith("--board="):
         self.board = arg.replace("--board=", "")
       elif arg.startswith("--workon="):
         workon_str = arg.replace("--workon=", "")
         package_list = shlex.split(" ".join(shlex.split(workon_str)))
         self.mandatory_source.update(package_list)
       elif arg == "--no-workon-deps":
         self.no_workon_deps = True
       elif arg == "--rebuild":
         self.rebuild = True
+      elif arg == "--show-output":
+        self.show_output = True
       else:
         # Not one of our options, so pass through to emerge.
         emerge_args.append(arg)
 
     if self.rebuild:
       if self.no_workon_deps:
         print "--rebuild is not compatible with --no-workon-deps"
         sys.exit(1)
 
     return emerge_args
@@ skipping 798 matching lines @@
   """Print dependency graph, for each package list it's prerequisites."""
   for i in deps_map:
     print "%s: (%s) needs" % (i, deps_map[i]["action"])
     needs = deps_map[i]["needs"]
     for j in needs:
       print "    %s" % (j)
     if not needs:
       print "    no dependencies"
 
 
-def EmergeWorker(task_queue, done_queue, emerge, package_db):
+class EmergeJobState(object):
+  __slots__ = ["done", "filename", "last_output_seek", "last_output_timestamp",
+               "pkgname", "retcode", "start_timestamp", "target"]
+
+  def __init__(self, target, pkgname, done, filename, start_timestamp,
+               retcode=None):
+
+    # The full name of the target we're building (e.g.
+    # chromeos-base/chromeos-0.0.1-r60)
+    self.target = target
+
+    # The short name of the target we're building (e.g. chromeos-0.0.1-r60)
+    self.pkgname = pkgname
+
+    # Whether the job is done. (True if the job is done; false otherwise.)
+    self.done = done
+
+    # The filename where output is currently stored.
+    self.filename = filename
+
+    # The location (in bytes) of the end of the last complete line we printed.
+    # This starts off at zero. We use this to jump to the right place when we
+    # print output from the same ebuild multiple times.
+    self.last_output_seek = 0
+
+    # The timestamp of the last time we printed output. Since we haven't
+    # printed output yet, this starts at zero.
+    self.last_output_timestamp = 0
+
+    # The return code of our job, if the job is actually finished.
+    self.retcode = retcode
+
+    # The timestamp when our job started.
+    self.start_timestamp = start_timestamp
+
+
+def EmergeWorker(task_queue, job_queue, emerge, package_db):
   """This worker emerges any packages given to it on the task_queue.
 
   Args:
     task_queue: The queue of tasks for this worker to do.
-    done_queue: The queue of results from the worker.
+    job_queue: The queue of results from the worker.
     emerge: An EmergeData() object.
     package_db: A dict, mapping package ids to portage Package objects.
 
   It expects package identifiers to be passed to it via task_queue. When
-  the package is merged, it pushes (target, retval, outputstr) into the
-  done_queue.
+  a task is started, it pushes the (target, filename) to the started_queue.
+  The output is stored in filename. When a merge starts or finishes, we push
+  EmergeJobState objects to the job_queue.
   """
 
+  def ExitHandler(signum, frame):
+    # Remove our signal handlers so we don't get called recursively.
+    signal.signal(signal.SIGINT, signal.SIG_DFL)
+    signal.signal(signal.SIGTERM, signal.SIG_DFL)
+
+    # Try to exit cleanly
+    sys.exit(1)
+
+  # Ensure that we exit quietly and cleanly, if possible, when we receive
+  # SIGTERM or SIGINT signals. By default, when the user hits CTRL-C, all
+  # of the child processes will print details about KeyboardInterrupt
+  # exceptions, which isn't very helpful.
+  signal.signal(signal.SIGINT, ExitHandler)
+  signal.signal(signal.SIGTERM, ExitHandler)
+
   settings, trees, mtimedb = emerge.settings, emerge.trees, emerge.mtimedb
   opts, spinner = emerge.opts, emerge.spinner
   opts["--nodeps"] = True
   while True:
     # Wait for a new item to show up on the queue. This is a blocking wait,
     # so if there's nothing to do, we just sit here.
     target = task_queue.get()
-    print "Emerging", target
     db_pkg = package_db[target]
     db_pkg.root_config = emerge.root_config
     install_list = [db_pkg]
-    output = tempfile.TemporaryFile()
-    outputstr = ""
+    pkgname = db_pkg.pf
+    output = tempfile.NamedTemporaryFile(prefix=pkgname + "-", delete=False)
+    start_timestamp = time.time()
+    job = EmergeJobState(target, pkgname, False, output.name, start_timestamp)
+    job_queue.put(job)
     if "--pretend" in opts:
-      retval = 0
+      retcode = 0
     else:
       save_stdout = sys.stdout
       save_stderr = sys.stderr
       try:
         sys.stdout = output
         sys.stderr = output
         scheduler = Scheduler(settings, trees, mtimedb, opts, spinner,
                               install_list, [], emerge.scheduler_graph)
-        retval = scheduler.merge()
+        retcode = scheduler.merge()
+      except Exception:
+        traceback.print_exc(file=output)
+        retcode = 1
       finally:
         sys.stdout = save_stdout
         sys.stderr = save_stderr
-      if retval is None:
-        retval = 0
-      if retval != 0:
-        output.seek(0)
-        outputstr = output.read()
+        output.close()
+      if retcode is None:
+        retcode = 0
 
-    done_queue.put((target, retval, outputstr))
+    job = EmergeJobState(target, pkgname, True, output.name, start_timestamp,
+                         retcode)
+    job_queue.put(job)
 
 
 class EmergeQueue(object):
   """Class to schedule emerge jobs according to a dependency graph."""
 
-  def __init__(self, deps_map, emerge, package_db):
+  def __init__(self, deps_map, emerge, package_db, show_output):
     # Store the dependency graph.
     self._deps_map = deps_map
     # Initialize the running queue to empty
-    self._jobs = set()
+    self._jobs = {}
     # List of total package installs represented in deps_map.
     install_jobs = [x for x in deps_map if deps_map[x]["action"] == "merge"]
     self._total_jobs = len(install_jobs)
+    self._show_output = show_output
 
     if "--pretend" in emerge.opts:
       print "Skipping merge because of --pretend mode."
       sys.exit(0)
 
     # Setup scheduler graph object. This is used by the child processes
     # to help schedule jobs.
     emerge.scheduler_graph = emerge.depgraph.schedulerGraph()
 
     procs = min(self._total_jobs,
                 emerge.opts.get("--jobs", multiprocessing.cpu_count()))
     self._emerge_queue = multiprocessing.Queue()
-    self._done_queue = multiprocessing.Queue()
-    args = (self._emerge_queue, self._done_queue, emerge, package_db)
+    self._job_queue = multiprocessing.Queue()
+    args = (self._emerge_queue, self._job_queue, emerge, package_db)
     self._pool = multiprocessing.Pool(procs, EmergeWorker, args)
 
     # Initialize the failed queue to empty.
     self._retry_queue = []
-    self._failed = {}
+    self._failed = set()
 
     # Print an update before we launch the merges.
     self._Status()
 
+    # Setup an exit handler so that we print nice messages if we are
+    # terminated.
+    self._SetupExitHandler()
+
+    # Schedule our jobs.
     for target, info in deps_map.items():
       if not info["needs"]:
         self._Schedule(target)
 
+  def _SetupExitHandler(self):
+
+    def ExitHandler(signum, frame):
+
+      # Kill our signal handlers so we don't get called recursively
+      signal.signal(signal.SIGINT, signal.SIG_DFL)
+      signal.signal(signal.SIGTERM, signal.SIG_DFL)
+
+      # Print our current job status
+      for target, job in self._jobs.iteritems():
+        if job:
+          self._PrintJob(job)
+          os.unlink(job.filename)
+
+      # Notify the user that we are exiting
+      print "Exiting on signal %s" % signum
+      sys.exit(1)
+
+    # Print out job status when we are killed
+    signal.signal(signal.SIGINT, ExitHandler)
+    signal.signal(signal.SIGTERM, ExitHandler)
+
   def _Schedule(self, target):
     # We maintain a tree of all deps, if this doesn't need
     # to be installed just free up it's children and continue.
     # It is possible to reinstall deps of deps, without reinstalling
     # first level deps, like so:
     # chromeos (merge) -> eselect (nomerge) -> python (merge)
     if self._deps_map[target]["action"] == "nomerge":
       self._Finish(target)
     else:
       # Kick off the build if it's marked to be built.
-      self._jobs.add(target)
+      self._jobs[target] = None
       self._emerge_queue.put(target)
 
   def _LoadAvg(self):
     loads = open("/proc/loadavg", "r").readline().split()[:3]
     return " ".join(loads)
 
+  def _PrintJob(self, job):
+    """Print output so far of specified job"""
+
+    # Calculate how long the job has been running.
+    current_time = time.time()
+    seconds = current_time - job.start_timestamp
+
+    # Note that we've printed out the job so far.
+    job.last_output_timestamp = current_time
+
+    # Note that we're starting the job
+    info = "job %s (%dm%.1fs) ===" % (job.pkgname, seconds / 60, seconds % 60)
+    if job.last_output_seek:
+      print "=== Continue output for %s " % info
+    else:
+      print "=== Start output for %s ===" % info
+
+    # Print actual output from job
+    f = codecs.open(job.filename, encoding='utf-8', errors='replace')
+    f.seek(job.last_output_seek)
+    prefix = job.pkgname + ":"
+    for line in f:
+
+      # Save off our position in the file
+      if line and line[-1] == "\n":
+        job.last_output_seek = f.tell()
+        line = line[:-1]
+
+      # Print our line
+      print prefix, line.encode('utf-8', 'replace')
+    f.close()
+
+    # Note end of output section
+    if job.done:
+      print "=== Complete: %s ===" % info
+    else:
+      print "=== Still running: %s ===" % info
+
+
   def _Status(self):
     """Print status."""
-    seconds = time.time() - GLOBAL_START
+    current_time = time.time()
+    seconds = current_time - GLOBAL_START
     line = ("Pending %s, Ready %s, Running %s, Retrying %s, Total %s "
             "[Time %dm%.1fs Load %s]")
     qsize = self._emerge_queue.qsize()
     print line % (len(self._deps_map), qsize, len(self._jobs) - qsize,
                   len(self._retry_queue), self._total_jobs,
                   seconds / 60, seconds % 60, self._LoadAvg())
 
+    # Print interim output every minute if --show-output is used. Otherwise,
+    # only print output if a job has been running for 60 minutes or more.
+    if self._show_output:
+      interval = 60
+    else:
+      interval = 60 * 60
+    for target, job in self._jobs.iteritems():
+      if job:
+        last_timestamp = max(job.start_timestamp, job.last_output_timestamp)
+        if last_timestamp + interval < current_time:
+          self._PrintJob(job)
+
   def _Finish(self, target):
     """Mark a target as completed and unblock dependecies."""
     for dep in self._deps_map[target]["provides"]:
       del self._deps_map[dep]["needs"][target]
       if not self._deps_map[dep]["needs"]:
         self._Schedule(dep)
     self._deps_map.pop(target)
 
   def _Retry(self):
     if self._retry_queue:
       target = self._retry_queue.pop(0)
       self._Schedule(target)
       print "Retrying emerge of %s." % target
 
   def Run(self):
     """Run through the scheduled ebuilds.
 
     Keep running so long as we have uninstalled packages in the
     dependency graph to merge.
     """
     while self._deps_map:
       # Check here that we are actually waiting for something.
       if (self._emerge_queue.empty() and
-          self._done_queue.empty() and
+          self._job_queue.empty() and
           not self._jobs and
           self._deps_map):
         # If we have failed on a package, retry it now.
         if self._retry_queue:
           self._Retry()
         # If we have failed a package twice, just give up.
         elif self._failed:
-          for failure, output in self._failed.items():
+          for failure in self._failed:
             print "Package failed: %s" % failure
-            print output
           PrintDepsMap(self._deps_map)
-          print "Packages failed: %s" % " ,".join(self._failed.keys())
+          print "Packages failed: %s" % " ,".join(self._failed)
           sys.exit(1)
         # If we have dependency cycles.
         else:
           print "Deadlock! Circular dependencies!"
           PrintDepsMap(self._deps_map)
           sys.exit(1)
 
       try:
-        target, retcode, output = self._done_queue.get(timeout=5)
+        job = self._job_queue.get(timeout=5)
       except Queue.Empty:
         # Print an update.
         self._Status()
         continue
 
-      self._jobs.discard(target)
+      target = job.target
 
-      # Print if necessary.
-      if retcode != 0:
-        print output
-      if retcode != 0:
+      if not job.done:
+        self._jobs[target] = job
+        print "Started %s (logged in %s)" % (target, job.filename)
+        continue
+
+      # Print output of job
+      if self._show_output or job.retcode != 0:
+        self._PrintJob(job)
+      os.unlink(job.filename)
+      del self._jobs[target]
+
+      seconds = time.time() - job.start_timestamp
+      details = "%s (in %dm%.1fs)" % (target, seconds / 60, seconds % 60)
+
+      # Complain if necessary.
+      if job.retcode != 0:
         # Handle job failure.
         if target in self._failed:
           # If this job has failed previously, give up.
-          print "Failed %s. Your build has failed." % target
+          print "Failed %s. Your build has failed." % details
         else:
           # Queue up this build to try again after a long while.
           self._retry_queue.append(target)
-          self._failed[target] = 1
-          print "Failed %s, retrying later." % target
+          self._failed.add(target)
+          print "Failed %s, retrying later." % details
       else:
         if target in self._failed and self._retry_queue:
           # If we have successfully retried a failed package, and there
           # are more failed packages, try the next one. We will only have
           # one retrying package actively running at a time.
           self._Retry()
 
-        print "Completed %s" % target
+        print "Completed %s" % details
         # Mark as completed and unblock waiting ebuilds.
         self._Finish(target)
 
         # Print an update.
         self._Status()
 
 
 def main():
 
   deps = DepGraphGenerator()
@@ skipping 30 matching lines @@
     deps.PrintTree(deps_tree)
 
   deps_graph = deps.GenDependencyGraph(deps_tree, deps_info)
 
   # OK, time to print out our progress so far.
   deps.PrintInstallPlan(deps_graph)
   if "--tree" in emerge.opts:
     PrintDepsMap(deps_graph)
 
   # Run the queued emerges.
-  scheduler = EmergeQueue(deps_graph, emerge, deps.package_db)
+  scheduler = EmergeQueue(deps_graph, emerge, deps.package_db, deps.show_output)
   scheduler.Run()
 
   # Update world.
   if ("--oneshot" not in emerge.opts and
       "--pretend" not in emerge.opts):
     world_set = emerge.root_config.sets["selected"]
     new_world_pkgs = []
     root = emerge.settings["ROOT"]
     final_db = emerge.depgraph._dynamic_config.mydbapi[root]
     for pkg in emerge.cmdline_packages:
       for db_pkg in final_db.match_pkgs(pkg):
         print "Adding %s to world" % db_pkg.cp
         new_world_pkgs.append(db_pkg.cp)
     if new_world_pkgs:
       world_set.update(new_world_pkgs)
 
   # Update environment (library cache, symlinks, etc.)
   if deps.board and "--pretend" not in emerge.opts:
     portage.env_update()
 
   print "Done"
 
 if __name__ == "__main__":
   main()