Cleanup parallel_emerge exit conditions to fix hangs.

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 22 matching lines @@
     * 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 errno
 import multiprocessing
 import os
 import Queue
 import shlex
 import signal
 import sys
 import tempfile
 import time
 import traceback
 import urllib2
@@ skipping 1248 matching lines @@
   """
 
   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()
+    if not target:
+      # If target is None, this means that the main thread wants us to quit.
+      # The other workers need to exit too, so we'll push the message back on
+      # to the queue so they'll get it too.
+      task_queue.put(target)
+      return
     db_pkg = package_db[target]
     db_pkg.root_config = emerge.root_config
     install_list = [db_pkg]
     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:
       retcode = 0
@@ skipping 84 matching lines @@
       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()
+
+  def ExitHandler(signum, frame):
+    # Switch to default signal handlers so that we'll die after two signals.
+    signal.signal(signal.SIGINT, signal.SIG_DFL)
+    signal.signal(signal.SIGTERM, signal.SIG_DFL)
+
+  # Don't exit on the first SIGINT / SIGTERM, because the parent worker will
+  # handle it and tell us when we need to exit.
+  signal.signal(signal.SIGINT, ExitHandler)
+  signal.signal(signal.SIGTERM, ExitHandler)
+
+  # seek_locations is a map indicating the position we are at in each file.
+  # It starts off empty, but is set by the various Print jobs as we go along
+  # to indicate where we left off in each file.
   seek_locations = {}
   while True:
-    job = queue.get()
-    if job:
-      job.Print(seek_locations)
-    else:
-      break
+    try:
+      job = queue.get()
+      if job:
+        job.Print(seek_locations)
+      else:
+        break
+    except IOError as ex:
+      if ex.errno == errno.EINTR:
+        # Looks like we received a signal. Keep printing.
+        continue
+      raise
 
 
 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 = {}
@@ skipping 50 matching lines @@
       signal.signal(signal.SIGTERM, signal.SIG_DFL)
 
       # Print our current job status
       for target, job in self._jobs.iteritems():
         if job:
           self._print_queue.put(JobPrinter(job, unlink=True))
 
       # Notify the user that we are exiting
       self._Print("Exiting on signal %s" % signum)
 
-      # Exit when print worker is done.
-      self._print_queue.put(None)
-      self._print_worker.join()
+      # Kill child threads, then exit.
+      self._Exit()
       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
@@ skipping 45 matching lines @@
       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)
       self._Print("Retrying emerge of %s." % target)
 
+  def _Exit(self):
+    # Tell emerge workers to exit. They all exit when 'None' is pushed
+    # to the queue.
+    self._emerge_queue.put(None)
+    self._pool.close()
+    self._pool.join()
+
+    # Now that our workers are finished, we can kill the print queue.
+    self._print_queue.put(None)
+    self._print_worker.join()
+
   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()
         else:
-          # Tell the print worker we're done, and wait for it to exit.
-          self._print_queue.put(None)
-          self._print_worker.join()
+          # Tell child threads to exit.
+          self._Exit()
 
           # The dependency map is helpful for debugging failures.
           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)
@@ skipping 40 matching lines @@
           # one retrying package actively running at a time.
           self._Retry()
 
         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()
+    # Tell child threads to exit.
+    self._Print("Merge complete")
+    self._Exit()
 
 
 def main():
 
   deps = DepGraphGenerator()
   deps.Initialize(sys.argv[1:])
   emerge = deps.emerge
 
   if emerge.action is not None:
     sys.argv = deps.ParseParallelEmergeArgs(sys.argv)
@@ skipping 77 matching lines @@
   # need to upgrade the rest of the packages. So we'll go ahead and do that.
   if portage_upgrade:
     args = sys.argv[1:] + ["--nomerge=sys-apps/portage"]
     os.execvp(os.path.realpath(sys.argv[0]), args)
 
   print "Done"
   sys.exit(0)
 
 if __name__ == "__main__":
   main()