Move printing of output to a separate thread.

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 1101 matching lines @@
     # 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.
-    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
-  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 SetupWorkerSignals():
   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)
 
+
+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.
+    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
+  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.
+  """
+
+  SetupWorkerSignals()
   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()
     db_pkg = package_db[target]
     db_pkg.root_config = emerge.root_config
     install_list = [db_pkg]
@@ skipping 21 matching lines @@
         sys.stderr = save_stderr
         output.close()
       if retcode is None:
         retcode = 0
 
     job = EmergeJobState(target, pkgname, True, output.name, start_timestamp,
                          retcode)
     job_queue.put(job)
 
 
+class LinePrinter(object):
+  """Helper object to print a single line."""
+
+  def __init__(self, line):
+    self.line = line
+
+  def Print(self, seek_locations):
+    print self.line
+
+
+class JobPrinter(object):
+  """Helper object to print output of a job."""
+
+  def __init__(self, job, unlink=False):
+    """Print output of job.
+
+    If unlink is True, unlink the job output file when done."""
+    self.current_time = time.time()
+    self.job = job
+    self.unlink = unlink
+
+  def Print(self, seek_locations):
+
+    job = self.job
+
+    # Calculate how long the job has been running.
+    seconds = self.current_time - job.start_timestamp
+
+    # Note that we've printed out the job so far.
+    job.last_output_timestamp = self.current_time
+
+    # Note that we're starting the job
+    info = "job %s (%dm%.1fs)" % (job.pkgname, seconds / 60, seconds % 60)
+    last_output_seek = seek_locations.get(job.filename, 0)
+    if 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(last_output_seek)
+    prefix = job.pkgname + ":"
+    for line in f:
+
+      # Save off our position in the file
+      if line and line[-1] == "\n":
+        last_output_seek = f.tell()
+        line = line[:-1]
+
+      # Print our line
+      print prefix, line.encode('utf-8', 'replace')
+    f.close()
+
+    # Save our last spot in the file so that we don't print out the same
+    # location twice.
+    seek_locations[job.filename] = last_output_seek
+
+    # Note end of output section
+    if job.done:
+      print "=== Complete: %s ===" % info
+    else:
+      print "=== Still running: %s ===" % info
+
+    if self.unlink:
+      os.unlink(job.filename)
+
+
+def PrintWorker(queue):
+  """A worker that prints stuff to the screen as requested."""
+  SetupWorkerSignals()
+  seek_locations = {}
+  while True:
+    job = queue.get()
+    if job:
+      job.Print(seek_locations)
+    else:
+      break
+
+
 class EmergeQueue(object):
   """Class to schedule emerge jobs according to a dependency graph."""
 
   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 = {}
     # 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._job_queue = multiprocessing.Queue()
+    self._print_queue = multiprocessing.Queue()
     args = (self._emerge_queue, self._job_queue, emerge, package_db)
     self._pool = multiprocessing.Pool(procs, EmergeWorker, args)
+    self._print_worker = multiprocessing.Process(target=PrintWorker,
+                                                 args=[self._print_queue])
+    self._print_worker.start()
 
     # Initialize the failed queue to empty.
     self._retry_queue = []
     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)
+          self._print_queue.put(JobPrinter(job, unlink=True))
 
       # Notify the user that we are exiting
-      print "Exiting on signal %s" % signum
+      self._Print("Exiting on signal %s" % signum)
+
+      # Exit when print worker is done.
+      self._print_queue.put(None)
+      self._print_worker.join()
       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[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 _Print(self, line):
+    """Print a single line."""
+    self._print_queue.put(LinePrinter(line))
 
   def _Status(self):
     """Print status."""
     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())
+    self._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)
+          self._print_queue.put(JobPrinter(job))
+          job.last_output_timestamp = current_time
 
   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
+      self._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._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 in self._failed:
-            print "Package failed: %s" % failure
+        else:
+          # Tell the print worker we're done, and wait for it to exit.
+          self._print_queue.put(None)
+          self._print_worker.join()
+
+          # The dependency map is helpful for debugging failures.
           PrintDepsMap(self._deps_map)
-          print "Packages failed: %s" % " ,".join(self._failed)
-          sys.exit(1)
-        # If we have dependency cycles.
-        else:
-          print "Deadlock! Circular dependencies!"
-          PrintDepsMap(self._deps_map)
+
+          # Tell the user why we're exiting.
+          if self._failed:
+            print "Packages failed: %s" % " ,".join(self._failed)
+          else:
+            print "Deadlock! Circular dependencies!"
           sys.exit(1)
 
       try:
         job = self._job_queue.get(timeout=5)
       except Queue.Empty:
         # Print an update.
         self._Status()
         continue
 
       target = job.target
 
       if not job.done:
         self._jobs[target] = job
-        print "Started %s (logged in %s)" % (target, job.filename)
+        self._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)
+        self._print_queue.put(JobPrinter(job, unlink=True))
+      else:
+        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." % details
+          self._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.add(target)
-          print "Failed %s, retrying later." % details
+          self._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" % details
+        self._Print("Completed %s" % details)
         # Mark as completed and unblock waiting ebuilds.
         self._Finish(target)
 
         # Print an update.
         self._Status()
 
+    # Tell the print worker we're done, and wait for it to exit.
+    self._print_queue.put(None)
+    self._print_worker.join()
+
 
 def main():
 
   deps = DepGraphGenerator()
   deps.Initialize(sys.argv[1:])
   emerge = deps.emerge
 
   if emerge.action is not None:
     sys.argv = deps.ParseParallelEmergeArgs(sys.argv)
     sys.exit(emerge_main())
@@ skipping 49 matching lines @@
       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()