Add the infrastructure necessary to support annotated stdout.

gclient_utils.py

 # Copyright 2009 Google Inc.  All Rights Reserved.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
 # You may obtain a copy of the License at
 #
 #      http://www.apache.org/licenses/LICENSE-2.0
 #
 # Unless required by applicable law or agreed to in writing, software
 # distributed under the License is distributed on an "AS IS" BASIS,
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
 """Generic utils."""
 
+import copy
 import errno
 import logging
 import os
+import Queue
 import re
 import stat
 import subprocess
 import sys
 import threading
 import time
 import xml.dom.minidom
 import xml.parsers.expat
 
 
@@ skipping 363 matching lines @@
       exec(FileRead(entries_filename), scope)
     except SyntaxError, e:
       SyntaxErrorToError(filename, e)
     all_directories = scope['entries'].keys()
     path_to_check = real_from_dir[len(path)+1:]
     while path_to_check:
       if path_to_check in all_directories:
         return path
       path_to_check = os.path.dirname(path_to_check)
     return None
-    
+
   logging.info('Found gclient root at ' + path)
   return path
 
 
 def PathDifference(root, subpath):
   """Returns the difference subpath minus root."""
   root = os.path.realpath(root)
   subpath = os.path.realpath(subpath)
   if not subpath.startswith(root):
     return None
@@ skipping 34 matching lines @@
   return config_dir, env['entries']
 
 
 class WorkItem(object):
   """One work item."""
   # A list of string, each being a WorkItem name.
   requirements = []
   # A unique string representing this work item.
   name = None
 
-  def run(self):
+  def run(self, work_queue, options):
+    """work_queue and options are passed as keyword arguments so they should be
+    the last parameters of the function when you override it."""
     pass
 
 
 class ExecutionQueue(object):
   """Runs a set of WorkItem that have interdependencies and were WorkItem are
   added as they are processed.
 
   In gclient's case, Dependencies sometime needs to be run out of order due to
   From() keyword. This class manages that all the required dependencies are run
   before running each one.
 
   Methods of this class are thread safe.
   """
   def __init__(self, jobs, progress):
     """jobs specifies the number of concurrent tasks to allow. progress is a
     Progress instance."""
     # Set when a thread is done or a new item is enqueued.
     self.ready_cond = threading.Condition()
     # Maximum number of concurrent tasks.
     self.jobs = jobs
     # List of WorkItem, for gclient, these are Dependency instances.
     self.queued = []
     # List of strings representing each Dependency.name that was run.
     self.ran = []
     # List of items currently running.
     self.running = []
     # Exceptions thrown if any.
-    self.exceptions = []
+    self.exceptions = Queue.Queue()
+    # Progress status
     self.progress = progress
     if self.progress:
-      self.progress.update()
+      self.progress.update(0)
 
   def enqueue(self, d):
     """Enqueue one Dependency to be executed later once its requirements are
     satisfied.
     """
     assert isinstance(d, WorkItem)
     self.ready_cond.acquire()
     try:
       self.queued.append(d)
       total = len(self.queued) + len(self.ran) + len(self.running)
       logging.debug('enqueued(%s)' % d.name)
       if self.progress:
         self.progress._total = total + 1
         self.progress.update(0)
       self.ready_cond.notifyAll()
     finally:
       self.ready_cond.release()
 
   def flush(self, *args, **kwargs):
     """Runs all enqueued items until all are executed."""
+    kwargs['work_queue'] = self
     self.ready_cond.acquire()
     try:
       while True:
         # Check for task to run first, then wait.
         while True:
-          if self.exceptions:
-            # Systematically flush the queue when there is an exception logged
-            # in.
+          if not self.exceptions.empty():
+            # Systematically flush the queue when an exception logged.
             self.queued = []
-          # Flush threads that have terminated.
-          self.running = [t for t in self.running if t.isAlive()]
-          if not self.queued and not self.running:
+          self._flush_terminated_threads()
+          if (not self.queued and not self.running or
+              self.jobs == len(self.running)):
+            # No more worker threads or can't queue anything.
             break
-          if self.jobs == len(self.running):
-            break
+
+          # Check for new tasks to start.
           for i in xrange(len(self.queued)):
             # Verify its requirements.
             for r in self.queued[i].requirements:
               if not r in self.ran:
                 # Requirement not met.
                 break
             else:
               # Start one work item: all its requirements are satisfied.
-              d = self.queued.pop(i)
-              new_thread = self._Worker(self, d, args=args, kwargs=kwargs)
-              if self.jobs > 1:
-                # Start the thread.
-                self.running.append(new_thread)
-                new_thread.start()
-              else:
-                # Run the 'thread' inside the main thread.
-                new_thread.run()
+              self._run_one_task(self.queued.pop(i), args, kwargs)
               break
           else:
             # Couldn't find an item that could run. Break out the outher loop.
             break
+
         if not self.queued and not self.running:
+          # We're done.
           break
         # We need to poll here otherwise Ctrl-C isn't processed.
         self.ready_cond.wait(10)
         # Something happened: self.enqueue() or a thread terminated. Loop again.
     finally:
       self.ready_cond.release()
+
     assert not self.running, 'Now guaranteed to be single-threaded'
-    if self.exceptions:
+    if not self.exceptions.empty():
       # To get back the stack location correctly, the raise a, b, c form must be
       # used, passing a tuple as the first argument doesn't work.
-      e = self.exceptions.pop(0)
+      e = self.exceptions.get()
       raise e[0], e[1], e[2]
     if self.progress:
       self.progress.end()
 
+  def _flush_terminated_threads(self):
+    """Flush threads that have terminated."""
+    running = self.running
+    self.running = []
+    for t in running:
+      if t.isAlive():
+        self.running.append(t)
+      else:
+        t.join()
+        t.kwargs['options'].stdout.flush()
+        if self.progress:
+          self.progress.update(1)
+        assert not t.name in self.ran
+        if not t.name in self.ran:
+          self.ran.append(t.name)
+
+  def _run_one_task(self, task_item, args, kwargs):
+    if self.jobs > 1:
+      # Start the thread.
+      index = len(self.ran) + len(self.running) + 1
+      # Copy 'options' just to be safe.
+      task_kwargs = kwargs.copy()
+      task_kwargs['options'] = copy.copy(task_kwargs['options'])
+      new_thread = self._Worker(task_item, args, task_kwargs)
+      self.running.append(new_thread)
+      new_thread.start()
+    else:
+      # Run the 'thread' inside the main thread. Don't try to catch any
+      # exception.
+      task_item.run(*args, **kwargs)
+      self.ran.append(task_item.name)
+      if self.progress:
+        self.progress.update(1)
+
   class _Worker(threading.Thread):
     """One thread to execute one WorkItem."""
-    def __init__(self, parent, item, args=(), kwargs=None):
+    def __init__(self, item, args, kwargs):
       threading.Thread.__init__(self, name=item.name or 'Worker')
+      logging.info(item.name)
+      self.item = item
       self.args = args
-      self.kwargs = kwargs or {}
-      self.item = item
-      self.parent = parent
+      self.kwargs = kwargs
 
     def run(self):
       """Runs in its own thread."""
       logging.debug('running(%s)' % self.item.name)
-      exception = None
+      work_queue = self.kwargs['work_queue']
       try:
         self.item.run(*self.args, **self.kwargs)
       except Exception:
         # Catch exception location.
-        exception = sys.exc_info()
+        logging.info('Caught exception in thread %s' % self.item.name)
+        logging.info(str(sys.exc_info()))
+        work_queue.exceptions.put(sys.exc_info())
+      logging.info('Task %s done' % self.item.name)
 
-      # This assumes the following code won't throw an exception. Bad.
-      self.parent.ready_cond.acquire()
+      work_queue.ready_cond.acquire()
       try:
-        if exception:
-          self.parent.exceptions.append(exception)
-        if self.parent.progress:
-          self.parent.progress.update(1)
-        assert not self.item.name in self.parent.ran
-        if not self.item.name in self.parent.ran:
-          self.parent.ran.append(self.item.name)
+        work_queue.ready_cond.notifyAll()
       finally:
-        self.parent.ready_cond.notifyAll()
-        self.parent.ready_cond.release()
+        work_queue.ready_cond.release()