gclient_utils: provide subproces42 based CheckCallAndFilter.

gclient_utils.py

Index: gclient_utils.py
diff --git a/gclient_utils.py b/gclient_utils.py
index 556018105cc8cd0c30e42f0ec6e235efbe900eeb..93c0a7e46cba08364401c9d6cc995fc9f212f153 100644
--- a/gclient_utils.py
+++ b/gclient_utils.py
@@ -21,7 +21,9 @@ import threading
 import time
 import urlparse
 
+# TODO(tandrii): migrate users to subprocess42.
 import subprocess2
+import subprocess42
 
 
 RETRY_MAX = 3
@@ -269,8 +271,10 @@ 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:
@@ -290,6 +294,8 @@ def CheckCallAndFilterAndHeader(args, always=False, header=None, **kwargs):
     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)
 
 
@@ -521,6 +527,7 @@ class _KillTimer(object):
         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):
@@ -613,6 +620,104 @@ def CheckCallAndFilter(args, stdout=None, filter_fn=None,
       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.