gclient_utils: provide subproces42 based CheckCallAndFilter.

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
 
+# TODO(tandrii): migrate users to subprocess42.
 import subprocess2
+import subprocess42
 
 
 RETRY_MAX = 3
 RETRY_INITIAL_SLEEP = 0.5
 START = datetime.datetime.now()
 
 
 _WARNINGS = []
 
 
@@ skipping 227 matching lines @@
 
 def CommandToStr(args):
   """Converts an arg list into a shell escaped string."""
   return ' '.join(pipes.quote(arg) for arg in args)
 
 
 def CheckCallAndFilterAndHeader(args, always=False, header=None, **kwargs):
   """Adds 'header' support to CheckCallAndFilter.
 
   If |always| is True, a message indicating what is being done
-  is printed to stdout all the time even if not output is generated. Otherwise
+  is printed to stdout all the time even if no output is generated. Otherwise
   the message header is printed only if the call generated any ouput.
+
+  If use_v42=True is given, uses subprocess42 through CheckCallAndFilter42.
   """
   stdout = kwargs.setdefault('stdout', sys.stdout)
   if header is None:
     header = "\n________ running '%s' in '%s'\n" % (
                  ' '.join(args), kwargs.get('cwd', '.'))
 
   if always:
     stdout.write(header)
   else:
     filter_fn = kwargs.get('filter_fn')
     def filter_msg(line):
       if line is None:
         stdout.write(header)
       elif filter_fn:
         filter_fn(line)
     kwargs['filter_fn'] = filter_msg
     kwargs['call_filter_on_first_line'] = True
   # Obviously.
   kwargs.setdefault('print_stdout', True)
+  if kwargs.pop('use_v42', False):
+    return CheckCallAndFilter42(args, **kwargs)
   return CheckCallAndFilter(args, **kwargs)
 
 
 class Wrapper(object):
   """Wraps an object, acting as a transparent proxy for all properties by
   default.
   """
   def __init__(self, wrapped):
     self._wrapped = wrapped
 
@@ skipping 211 matching lines @@
         try:
           logging.warn('killing child %s %s because of no output for %fs',
                        self._child.pid, self._child_info, self._timeout)
           self._killing_attempted = True
           self._child.kill()
         except OSError:
           logging.exception('failed to kill child %s', self._child)
         return
 
 
+#TODO(tandrii): all users of this should be converted to CheckCallAndFilter42.
 def CheckCallAndFilter(args, stdout=None, filter_fn=None,
                        print_stdout=None, call_filter_on_first_line=False,
                        retry=False, kill_timeout=None, **kwargs):
   """Runs a command and calls back a filter function if needed.
 
   Accepts all subprocess2.Popen() parameters plus:
     print_stdout: If True, the command's stdout is forwarded to stdout.
     filter_fn: A function taking a single string argument called with each line
                of the subprocess2's output. Each line has the trailing newline
                character trimmed.
@@ skipping 72 matching lines @@
     if timeout_killer.killing_attempted:
       print('The subprocess above was likely killed because it looked hung. '
             'Output thus far:\n> %s' %
             ('\n> '.join(output.getvalue().splitlines())))
     time.sleep(sleep_interval)
     sleep_interval *= 2
   raise subprocess2.CalledProcessError(
       rv, args, kwargs.get('cwd', None), None, None)
 
 
+def CheckCallAndFilter42(args, stdout=None, filter_fn=None,
+                       print_stdout=None, call_filter_on_first_line=False,
+                       retry=False, kill_timeout=None, **kwargs):
+  """Runs a command and calls back a filter function if needed.
+
+  Accepts all subprocess2.Popen() parameters plus:
+    print_stdout: If True, the command's stdout is forwarded to stdout.
+    filter_fn: A function taking a single string argument called with each line
+               of the subprocess2's output. Each line has the trailing newline
+               character trimmed.
+    stdout: Can be any bufferable output.
+    retry: If the process exits non-zero, sleep for a brief interval and try
+           again, up to RETRY_MAX times.
+    kill_timeout: (float) if given, number of seconds after which spawned
+           process would be killed. This applies to each retry separately.
+
+  stderr is always redirected to stdout.
+  """
+  assert print_stdout or filter_fn
+  if kill_timeout:
+    assert isinstance(kill_timeout, (int, float)), kill_timeout
+
+  stdout = stdout or sys.stdout
+  output = cStringIO.StringIO()
+  filter_fn = filter_fn or (lambda x: None)
+
+  sleep_interval = RETRY_INITIAL_SLEEP
+  run_cwd = kwargs.get('cwd', os.getcwd())
+  debug_child_info = "'%s' in %s" % (' '.join('"%s"' % x for x in args), run_cwd)
+
+  for _ in xrange(RETRY_MAX + 1):
+    child = subprocess42.Popen(
+        args, bufsize=0, stdout=subprocess2.PIPE, stderr=subprocess2.STDOUT,
+        **kwargs)
+
+    GClientChildren.add(child)
+
+    # Do a flush of stdout before we begin reading from the subprocess42's
+    # stdout.
+    stdout.flush()
+
+    # Also, we need to forward stdout to prevent weird re-ordering of output.
+    try:
+      first_line = True
+      in_line = ''
+      # Don't block for more than 1 second.
+      for pipe, data in child.yield_any(maxsize=256, timeout=1):
+        if pipe is None:
+          # No new data.
+          if kill_timeout and child.duration() > kill_timeout:
+            print('ERROR: killing process %s running for %.0fs (timeout: %.0fs)'
+                  % (debug_child_info, child.duration(), kill_timeout))
+            child.kill()
+          continue
+
+        assert pipe == 'stdout'
+        if first_line:
+          first_line = False
+          if call_filter_on_first_line:
+            filter_fn(None)
+
+        output.write(data)
+        if print_stdout:
+          stdout.write(data)
+        for byte in data:
+          if byte not in ['\r', '\n']:
+            in_line += byte
+          else:
+            filter_fn(in_line)
+            in_line = ''
+      # 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 = child.wait()
+
+      # Don't put this in a 'finally,' since the child may still run if we get
+      # an exception.
+      GClientChildren.remove(child)
+
+    except KeyboardInterrupt:
+      print >> sys.stderr, 'Failed while running "%s"' % ' '.join(args)
+      raise
+
+    if rv == 0:
+      return output.getvalue()
+    if not retry:
+      break
+    print ("WARNING: subprocess %s failed; will retry after a short nap..." %
+           debug_child_info)
+    time.sleep(sleep_interval)
+    sleep_interval *= 2
+  # TODO(tandrii): change this to subprocess.CalledProcessError,
+  # beacuse subprocess42 doesn't have a class unlike subprocess2.
+  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,3})% .*')
 
   def __init__(self, time_throttle=0, predicate=None, out_fh=None):
     """
@@ skipping 668 matching lines @@
     # Just incase we have some ~/blah paths.
     target = os.path.abspath(os.path.expanduser(target))
     if os.path.isfile(target) and os.access(target, os.X_OK):
       return target
     if sys.platform.startswith('win'):
       for suffix in ('.bat', '.cmd', '.exe'):
         alt_target = target + suffix
         if os.path.isfile(alt_target) and os.access(alt_target, os.X_OK):
           return alt_target
   return None