Revamped terminal output for update.

gclient_utils.py

 # Copyright (c) 2012 The Chromium Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
 """Generic utils."""
 
 import codecs
 import cStringIO
+import datetime
 import logging
 import os
 import pipes
 import platform
 import Queue
 import re
 import stat
 import subprocess
 import sys
 import tempfile
 import threading
 import time
 import urlparse
 
 import subprocess2
 
 
 RETRY_MAX = 3
 RETRY_INITIAL_SLEEP = 0.5
+START = datetime.datetime.now()
 
 
 _WARNINGS = []
 
 
 # These repos are known to cause OOM errors on 32-bit platforms, due the the
 # very large objects they contain.  It is not safe to use threaded index-pack
 # when cloning/fetching them.
 THREADED_INDEX_PACK_BLACKLIST = [
   'https://chromium.googlesource.com/chromium/reference_builds/chrome_win.git'
 ]
 
 
 class Error(Exception):
   """gclient exception class."""
   def __init__(self, msg, *args, **kwargs):
     index = getattr(threading.currentThread(), 'index', 0)
     if index:
       msg = '\n'.join('%d> %s' % (index, l) for l in msg.splitlines())
     super(Error, self).__init__(msg, *args, **kwargs)
 
 
+def Elapsed(until=None):
+  if until is None:
+    until = datetime.datetime.now()
+  return str(until - START).partition('.')[0]
+
+
 def PrintWarnings():
   """Prints any accumulated warnings."""
   if _WARNINGS:
     print >> sys.stderr, '\n\nWarnings:'
     for warning in _WARNINGS:
       print >> sys.stderr, warning
 
 
 def AddWarning(msg):
   """Adds the given warning message to the list of accumulated warnings."""
@@ skipping 416 matching lines @@
     try:
       in_byte = kid.stdout.read(1)
       if in_byte:
         if call_filter_on_first_line:
           filter_fn(None)
         in_line = ''
         while in_byte:
           output.write(in_byte)
           if print_stdout:
             stdout.write(in_byte)
-          if in_byte != '\r':
-            if in_byte != '\n':
-              in_line += in_byte
-            else:
-              filter_fn(in_line)
-              in_line = ''
+          if in_byte not in ['\r', '\n']:
+            in_line += in_byte
           else:
             filter_fn(in_line)
             in_line = ''
           in_byte = kid.stdout.read(1)
         # Flush the rest of buffered output. This is only an issue with
         # stdout/stderr not ending with a \n.
         if len(in_line):
           filter_fn(in_line)
       rv = kid.wait()
 
@@ skipping 16 matching lines @@
   raise subprocess2.CalledProcessError(
       rv, args, kwargs.get('cwd', None), None, None)
 
 
 class GitFilter(object):
   """A filter_fn implementation for quieting down git output messages.
 
   Allows a custom function to skip certain lines (predicate), and will throttle
   the output of percentage completed lines to only output every X seconds.
   """
-  PERCENT_RE = re.compile('.* ([0-9]{1,2})% .*')
+  PERCENT_RE = re.compile('(.*) ([0-9]{1,3})% .*')
 
-  def __init__(self, time_throttle=0, predicate=None):
+  def __init__(self, time_throttle=0, predicate=None, out_fh=None):
     """
     Args:
       time_throttle (int): GitFilter will throttle 'noisy' output (such as the
         XX% complete messages) to only be printed at least |time_throttle|
         seconds apart.
       predicate (f(line)): An optional function which is invoked for every line.
         The line will be skipped if predicate(line) returns False.
+      out_fh: File handle to write output to.
     """
     self.last_time = 0
     self.time_throttle = time_throttle
     self.predicate = predicate
+    self.out_fh = out_fh or sys.stdout
+    self.progress_prefix = None
 
   def __call__(self, line):
     # git uses an escape sequence to clear the line; elide it.
     esc = line.find(unichr(033))
     if esc > -1:
       line = line[:esc]
     if self.predicate and not self.predicate(line):
       return
     now = time.time()
     match = self.PERCENT_RE.match(line)
-    if not match:
-      self.last_time = 0
-    if (now - self.last_time) >= self.time_throttle:
-      self.last_time = now
-      print line
+    if match:
+      if match.group(1) != self.progress_prefix:
+        self.progress_prefix = match.group(1)
+      elif now - self.last_time < self.time_throttle:
+        return
+    self.last_time = now
+    self.out_fh.write('[%s] ' % Elapsed())
+    print >> self.out_fh, line
 
 
 def FindGclientRoot(from_dir, filename='.gclient'):
   """Tries to find the gclient root."""
   real_from_dir = os.path.realpath(from_dir)
   path = real_from_dir
   while not os.path.exists(os.path.join(path, filename)):
     split_path = os.path.split(path)
     if not split_path[1]:
       return None
@@ skipping 109 matching lines @@
 class WorkItem(object):
   """One work item."""
   # On cygwin, creating a lock throwing randomly when nearing ~100 locks.
   # As a workaround, use a single lock. Yep you read it right. Single lock for
   # all the 100 objects.
   lock = threading.Lock()
 
   def __init__(self, name):
     # A unique string representing this work item.
     self._name = name
+    self.outbuf = cStringIO.StringIO()
+    self.start = self.finish = None
 
   def run(self, work_queue):
     """work_queue is passed as keyword argument so it should be
     the last parameters of the function when you override it."""
     pass
 
   @property
   def name(self):
     return self._name
 
 
 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, ignore_requirements):
+  def __init__(self, jobs, progress, ignore_requirements, verbose=False):
     """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 = Queue.Queue()
     # Progress status
     self.progress = progress
     if self.progress:
       self.progress.update(0)
 
     self.ignore_requirements = ignore_requirements
+    self.verbose = verbose
+    self.last_join = None
+    self.last_subproc_output = None
 
   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 out_cb(self, _):
+    self.last_subproc_output = datetime.datetime.now()
+    return True
+
+  @staticmethod
+  def format_task_output(task, comment=''):
+    if comment:
+      comment = ' (%s)' % comment
+    if task.start and task.finish:
+      elapsed = ' (Elapsed: %s)' % (
+          str(task.finish - task.start).partition('.')[0])
+    else:
+      elapsed = ''
+    return """
+%s%s%s
+----------------------------------------
+%s
+----------------------------------------""" % (
+    task.name, comment, task.outbuf.getvalue().strip(), elapsed)
+
   def flush(self, *args, **kwargs):
     """Runs all enqueued items until all are executed."""
     kwargs['work_queue'] = self
+    self.last_subproc_output = self.last_join = datetime.datetime.now()
     self.ready_cond.acquire()
     try:
       while True:
         # Check for task to run first, then wait.
         while True:
           if not self.exceptions.empty():
             # Systematically flush the queue when an exception logged.
             self.queued = []
           self._flush_terminated_threads()
           if (not self.queued and not self.running or
@@ skipping 12 matching lines @@
           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.
         try:
           self.ready_cond.wait(10)
+          # If we haven't printed to terminal for a while, but we have received
+          # spew from a suprocess, let the user know we're still progressing.
+          now = datetime.datetime.now()
+          if (now - self.last_join > datetime.timedelta(seconds=60) and
+              self.last_subproc_output > self.last_join):
+            if self.progress:
+              print >> sys.stdout, ''
+            elapsed = Elapsed()
+            print >> sys.stdout, '[%s] Still working on:' % elapsed
+            for task in self.running:
+              print >> sys.stdout, '[%s]   %s' % (elapsed, task.item.name)
         except KeyboardInterrupt:
           # Help debugging by printing some information:
           print >> sys.stderr, (
               ('\nAllowed parallel jobs: %d\n# queued: %d\nRan: %s\n'
                 'Running: %d') % (
               self.jobs,
               len(self.queued),
               ', '.join(self.ran),
               len(self.running)))
           for i in self.queued:
-            print >> sys.stderr, '%s: %s' % (i.name, ', '.join(i.requirements))
+            print >> sys.stderr, '%s (not started): %s' % (
+                i.name, ', '.join(i.requirements))
+          for i in self.running:
+            print >> sys.stderr, self.format_task_output(i.item, 'interrupted')
           raise
         # 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 not self.exceptions.empty():
+      if self.progress:
+        print >> sys.stdout, ''
       # 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.get()
+      e, task = self.exceptions.get()
+      print >> sys.stderr, self.format_task_output(task.item, 'ERROR')
       raise e[0], e[1], e[2]
-    if self.progress:
+    elif 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()
+        self.last_join = datetime.datetime.now()
         sys.stdout.flush()
+        if self.verbose:
+          print >> sys.stdout, self.format_task_output(t.item)
         if self.progress:
           self.progress.update(1, t.item.name)
         if t.item.name in self.ran:
           raise Error(
               'gclient is confused, "%s" is already in "%s"' % (
                 t.item.name, ', '.join(self.ran)))
         if not t.item.name in self.ran:
           self.ran.append(t.item.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
       new_thread = self._Worker(task_item, index, args, 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, ', '.join(t.item.name for t in self.running))
+      try:
+        task_item.start = datetime.datetime.now()
+        print >> task_item.outbuf, '[%s] Started.' % Elapsed(task_item.start)
+        task_item.run(*args, **kwargs)
+        task_item.finish = datetime.datetime.now()
+        print >> task_item.outbuf, '[%s] Finished.' % Elapsed(task_item.finish)
+        self.ran.append(task_item.name)
+        if self.verbose:
+          if self.progress:
+            print >> sys.stdout, ''
+          print >> sys.stdout, self.format_task_output(task_item)
+        if self.progress:
+          self.progress.update(1, ', '.join(t.item.name for t in self.running))
+      except KeyboardInterrupt:
+        print >> sys.stderr, self.format_task_output(task_item, 'interrupted')
+        raise
+      except Exception:
+        print >> sys.stderr, self.format_task_output(task_item, 'ERROR')
+        raise
+
 
   class _Worker(threading.Thread):
     """One thread to execute one WorkItem."""
     def __init__(self, item, index, args, kwargs):
       threading.Thread.__init__(self, name=item.name or 'Worker')
       logging.info('_Worker(%s) reqs:%s' % (item.name, item.requirements))
       self.item = item
       self.index = index
       self.args = args
       self.kwargs = kwargs
       self.daemon = True
 
     def run(self):
       """Runs in its own thread."""
       logging.debug('_Worker.run(%s)' % self.item.name)
       work_queue = self.kwargs['work_queue']
       try:
+        self.item.start = datetime.datetime.now()
+        print >> self.item.outbuf, '[%s] Started.' % Elapsed(self.item.start)
         self.item.run(*self.args, **self.kwargs)
+        self.item.finish = datetime.datetime.now()
+        print >> self.item.outbuf, '[%s] Finished.' % Elapsed(self.item.finish)
       except KeyboardInterrupt:
         logging.info('Caught KeyboardInterrupt in thread %s', self.item.name)
         logging.info(str(sys.exc_info()))
-        work_queue.exceptions.put(sys.exc_info())
+        work_queue.exceptions.put((sys.exc_info(), self))
         raise
       except Exception:
         # Catch exception location.
         logging.info('Caught exception in thread %s', self.item.name)
         logging.info(str(sys.exc_info()))
-        work_queue.exceptions.put(sys.exc_info())
+        work_queue.exceptions.put((sys.exc_info(), self))
       finally:
         logging.info('_Worker.run(%s) done', self.item.name)
         work_queue.ready_cond.acquire()
         try:
           work_queue.ready_cond.notifyAll()
         finally:
           work_queue.ready_cond.release()
 
 
 def GetEditor(git, git_editor=None):
@@ skipping 124 matching lines @@
 def DefaultIndexPackConfig(url=''):
   """Return reasonable default values for configuring git-index-pack.
 
   Experiments suggest that higher values for pack.threads don't improve
   performance."""
   cache_limit = DefaultDeltaBaseCacheLimit()
   result = ['-c', 'core.deltaBaseCacheLimit=%s' % cache_limit]
   if url in THREADED_INDEX_PACK_BLACKLIST:
     result.extend(['-c', 'pack.threads=1'])
   return result