vendor/depot_tools/subprocess2.py

   1 # coding=utf8
   2 # Copyright (c) 2012 The Chromium Authors. All rights reserved.
   3 # Use of this source code is governed by a BSD-style license that can be
   4 # found in the LICENSE file.
   5 """Collection of subprocess wrapper functions.
   6
   7 In theory you shouldn't need anything else in subprocess, or this module failed.
   8 """
   9
  10 import cStringIO
  11 import errno
  12 import logging
  13 import os
  14 import Queue
  15 import subprocess
  16 import sys
  17 import time
  18 import threading
  19
  20
  21 # Constants forwarded from subprocess.
  22 PIPE = subprocess.PIPE
  23 STDOUT = subprocess.STDOUT
  24 # Sends stdout or stderr to os.devnull.
  25 VOID = object()
  26 # Error code when a process was killed because it timed out.
  27 TIMED_OUT = -2001
  28
  29 # Globals.
  30 # Set to True if you somehow need to disable this hack.
  31 SUBPROCESS_CLEANUP_HACKED = False
  32
  33
  34 class CalledProcessError(subprocess.CalledProcessError):
  35   """Augment the standard exception with more data."""
  36   def __init__(self, returncode, cmd, cwd, stdout, stderr):
  37     super(CalledProcessError, self).__init__(returncode, cmd, output=stdout)
  38     self.stdout = self.output  # for backward compatibility.
  39     self.stderr = stderr
  40     self.cwd = cwd
  41
  42   def __str__(self):
  43     out = 'Command %s returned non-zero exit status %s' % (
  44         ' '.join(self.cmd), self.returncode)
  45     if self.cwd:
  46       out += ' in ' + self.cwd
  47     return '\n'.join(filter(None, (out, self.stdout, self.stderr)))
  48
  49
  50 class CygwinRebaseError(CalledProcessError):
  51   """Occurs when cygwin's fork() emulation fails due to rebased dll."""
  52
  53
  54 ## Utility functions
  55
  56
  57 def kill_pid(pid):
  58   """Kills a process by its process id."""
  59   try:
  60     # Unable to import 'module'
  61     # pylint: disable=E1101,F0401
  62     import signal
  63     return os.kill(pid, signal.SIGKILL)
  64   except ImportError:
  65     pass
  66
  67
  68 def kill_win(process):
  69   """Kills a process with its windows handle.
  70
  71   Has no effect on other platforms.
  72   """
  73   try:
  74     # Unable to import 'module'
  75     # pylint: disable=F0401
  76     import win32process
  77     # Access to a protected member _handle of a client class
  78     # pylint: disable=W0212
  79     return win32process.TerminateProcess(process._handle, -1)
  80   except ImportError:
  81     pass
  82
  83
  84 def add_kill():
  85   """Adds kill() method to subprocess.Popen for python <2.6"""
  86   if hasattr(subprocess.Popen, 'kill'):
  87     return
  88
  89   if sys.platform == 'win32':
  90     subprocess.Popen.kill = kill_win
  91   else:
  92     subprocess.Popen.kill = lambda process: kill_pid(process.pid)
  93
  94
  95 def hack_subprocess():
  96   """subprocess functions may throw exceptions when used in multiple threads.
  97
  98   See http://bugs.python.org/issue1731717 for more information.
  99   """
 100   global SUBPROCESS_CLEANUP_HACKED
 101   if not SUBPROCESS_CLEANUP_HACKED and threading.activeCount() != 1:
 102     # Only hack if there is ever multiple threads.
 103     # There is no point to leak with only one thread.
 104     subprocess._cleanup = lambda: None
 105     SUBPROCESS_CLEANUP_HACKED = True
 106
 107
 108 def get_english_env(env):
 109   """Forces LANG and/or LANGUAGE to be English.
 110
 111   Forces encoding to utf-8 for subprocesses.
 112
 113   Returns None if it is unnecessary.
 114   """
 115   if sys.platform == 'win32':
 116     return None
 117   env = env or os.environ
 118
 119   # Test if it is necessary at all.
 120   is_english = lambda name: env.get(name, 'en').startswith('en')
 121
 122   if is_english('LANG') and is_english('LANGUAGE'):
 123     return None
 124
 125   # Requires modifications.
 126   env = env.copy()
 127   def fix_lang(name):
 128     if not is_english(name):
 129       env[name] = 'en_US.UTF-8'
 130   fix_lang('LANG')
 131   fix_lang('LANGUAGE')
 132   return env
 133
 134
 135 class NagTimer(object):
 136   """
 137   Triggers a callback when a time interval passes without an event being fired.
 138
 139   For example, the event could be receiving terminal output from a subprocess;
 140   and the callback could print a warning to stderr that the subprocess appeared
 141   to be hung.
 142   """
 143   def __init__(self, interval, cb):
 144     self.interval = interval
 145     self.cb = cb
 146     self.timer = threading.Timer(self.interval, self.fn)
 147     self.last_output = self.previous_last_output = 0
 148
 149   def start(self):
 150     self.last_output = self.previous_last_output = time.time()
 151     self.timer.start()
 152
 153   def event(self):
 154     self.last_output = time.time()
 155
 156   def fn(self):
 157     now = time.time()
 158     if self.last_output == self.previous_last_output:
 159       self.cb(now - self.previous_last_output)
 160     # Use 0.1 fudge factor, just in case
 161     #   (self.last_output - now) is very close to zero.
 162     sleep_time = (self.last_output - now - 0.1) % self.interval
 163     self.previous_last_output = self.last_output
 164     self.timer = threading.Timer(sleep_time + 0.1, self.fn)
 165     self.timer.start()
 166
 167   def cancel(self):
 168     self.timer.cancel()
 169
 170
 171 class Popen(subprocess.Popen):
 172   """Wraps subprocess.Popen() with various workarounds.
 173
 174   - Forces English output since it's easier to parse the stdout if it is always
 175     in English.
 176   - Sets shell=True on windows by default. You can override this by forcing
 177     shell parameter to a value.
 178   - Adds support for VOID to not buffer when not needed.
 179   - Adds self.start property.
 180
 181   Note: Popen() can throw OSError when cwd or args[0] doesn't exist. Translate
 182   exceptions generated by cygwin when it fails trying to emulate fork().
 183   """
 184   def __init__(self, args, **kwargs):
 185     # Make sure we hack subprocess if necessary.
 186     hack_subprocess()
 187     add_kill()
 188
 189     env = get_english_env(kwargs.get('env'))
 190     if env:
 191       kwargs['env'] = env
 192     if kwargs.get('shell') is None:
 193       # *Sigh*:  Windows needs shell=True, or else it won't search %PATH% for
 194       # the executable, but shell=True makes subprocess on Linux fail when it's
 195       # called with a list because it only tries to execute the first item in
 196       # the list.
 197       kwargs['shell'] = bool(sys.platform=='win32')
 198
 199     if isinstance(args, basestring):
 200       tmp_str = args
 201     elif isinstance(args, (list, tuple)):
 202       tmp_str = ' '.join(args)
 203     else:
 204       raise CalledProcessError(None, args, kwargs.get('cwd'), None, None)
 205     if kwargs.get('cwd', None):
 206       tmp_str += ';  cwd=%s' % kwargs['cwd']
 207     logging.debug(tmp_str)
 208
 209     self.stdout_cb = None
 210     self.stderr_cb = None
 211     self.stdin_is_void = False
 212     self.stdout_is_void = False
 213     self.stderr_is_void = False
 214     self.cmd_str = tmp_str
 215
 216     if kwargs.get('stdin') is VOID:
 217       kwargs['stdin'] = open(os.devnull, 'r')
 218       self.stdin_is_void = True
 219
 220     for stream in ('stdout', 'stderr'):
 221       if kwargs.get(stream) in (VOID, os.devnull):
 222         kwargs[stream] = open(os.devnull, 'w')
 223         setattr(self, stream + '_is_void', True)
 224       if callable(kwargs.get(stream)):
 225         setattr(self, stream + '_cb', kwargs[stream])
 226         kwargs[stream] = PIPE
 227
 228     self.start = time.time()
 229     self.timeout = None
 230     self.nag_timer = None
 231     self.nag_max = None
 232     self.shell = kwargs.get('shell', None)
 233     # Silence pylint on MacOSX
 234     self.returncode = None
 235
 236     try:
 237       super(Popen, self).__init__(args, **kwargs)
 238     except OSError, e:
 239       if e.errno == errno.EAGAIN and sys.platform == 'cygwin':
 240         # Convert fork() emulation failure into a CygwinRebaseError().
 241         raise CygwinRebaseError(
 242             e.errno,
 243             args,
 244             kwargs.get('cwd'),
 245             None,
 246             'Visit '
 247             'http://code.google.com/p/chromium/wiki/CygwinDllRemappingFailure '
 248             'to learn how to fix this error; you need to rebase your cygwin '
 249             'dlls')
 250       # Popen() can throw OSError when cwd or args[0] doesn't exist.
 251       raise OSError('Execution failed with error: %s.\n'
 252                     'Check that %s or %s exist and have execution permission.'
 253                     % (str(e), kwargs.get('cwd'), args[0]))
 254
 255   def _tee_threads(self, input):  # pylint: disable=W0622
 256     """Does I/O for a process's pipes using threads.
 257
 258     It's the simplest and slowest implementation. Expect very slow behavior.
 259
 260     If there is a callback and it doesn't keep up with the calls, the timeout
 261     effectiveness will be delayed accordingly.
 262     """
 263     # Queue of either of <threadname> when done or (<threadname>, data).  In
 264     # theory we would like to limit to ~64kb items to not cause large memory
 265     # usage when the callback blocks. It is not done because it slows down
 266     # processing on OSX10.6 by a factor of 2x, making it even slower than
 267     # Windows!  Revisit this decision if it becomes a problem, e.g. crash
 268     # because of memory exhaustion.
 269     queue = Queue.Queue()
 270     done = threading.Event()
 271     nag = None
 272
 273     def write_stdin():
 274       try:
 275         stdin_io = cStringIO.StringIO(input)
 276         while True:
 277           data = stdin_io.read(1024)
 278           if data:
 279             self.stdin.write(data)
 280           else:
 281             self.stdin.close()
 282             break
 283       finally:
 284         queue.put('stdin')
 285
 286     def _queue_pipe_read(pipe, name):
 287       """Queues characters read from a pipe into a queue."""
 288       try:
 289         while True:
 290           data = pipe.read(1)
 291           if not data:
 292             break
 293           if nag:
 294             nag.event()
 295           queue.put((name, data))
 296       finally:
 297         queue.put(name)
 298
 299     def timeout_fn():
 300       try:
 301         done.wait(self.timeout)
 302       finally:
 303         queue.put('timeout')
 304
 305     def wait_fn():
 306       try:
 307         self.wait()
 308       finally:
 309         queue.put('wait')
 310
 311     # Starts up to 5 threads:
 312     # Wait for the process to quit
 313     # Read stdout
 314     # Read stderr
 315     # Write stdin
 316     # Timeout
 317     threads = {
 318         'wait': threading.Thread(target=wait_fn),
 319     }
 320     if self.timeout is not None:
 321       threads['timeout'] = threading.Thread(target=timeout_fn)
 322     if self.stdout_cb:
 323       threads['stdout'] = threading.Thread(
 324           target=_queue_pipe_read, args=(self.stdout, 'stdout'))
 325     if self.stderr_cb:
 326       threads['stderr'] = threading.Thread(
 327         target=_queue_pipe_read, args=(self.stderr, 'stderr'))
 328     if input:
 329       threads['stdin'] = threading.Thread(target=write_stdin)
 330     elif self.stdin:
 331       # Pipe but no input, make sure it's closed.
 332       self.stdin.close()
 333     for t in threads.itervalues():
 334       t.start()
 335
 336     if self.nag_timer:
 337       def _nag_cb(elapsed):
 338         logging.warn('  No output for %.0f seconds from command:' % elapsed)
 339         logging.warn('    %s' % self.cmd_str)
 340         if (self.nag_max and
 341             int('%.0f' % (elapsed / self.nag_timer)) >= self.nag_max):
 342           queue.put('timeout')
 343           done.set()  # Must do this so that timeout thread stops waiting.
 344       nag = NagTimer(self.nag_timer, _nag_cb)
 345       nag.start()
 346
 347     timed_out = False
 348     try:
 349       # This thread needs to be optimized for speed.
 350       while threads:
 351         item = queue.get()
 352         if item[0] == 'stdout':
 353           self.stdout_cb(item[1])
 354         elif item[0] == 'stderr':
 355           self.stderr_cb(item[1])
 356         else:
 357           # A thread terminated.
 358           if item in threads:
 359             threads[item].join()
 360             del threads[item]
 361           if item == 'wait':
 362             # Terminate the timeout thread if necessary.
 363             done.set()
 364           elif item == 'timeout' and not timed_out and self.poll() is None:
 365             logging.debug('Timed out after %.0fs: killing' % (
 366                 time.time() - self.start))
 367             self.kill()
 368             timed_out = True
 369     finally:
 370       # Stop the threads.
 371       done.set()
 372       if nag:
 373         nag.cancel()
 374       if 'wait' in threads:
 375         # Accelerate things, otherwise it would hang until the child process is
 376         # done.
 377         logging.debug('Killing child because of an exception')
 378         self.kill()
 379       # Join threads.
 380       for thread in threads.itervalues():
 381         thread.join()
 382       if timed_out:
 383         self.returncode = TIMED_OUT
 384
 385   # pylint: disable=W0221,W0622
 386   def communicate(self, input=None, timeout=None, nag_timer=None,
 387                   nag_max=None):
 388     """Adds timeout and callbacks support.
 389
 390     Returns (stdout, stderr) like subprocess.Popen().communicate().
 391
 392     - The process will be killed after |timeout| seconds and returncode set to
 393       TIMED_OUT.
 394     - If the subprocess runs for |nag_timer| seconds without producing terminal
 395       output, print a warning to stderr.
 396     """
 397     self.timeout = timeout
 398     self.nag_timer = nag_timer
 399     self.nag_max = nag_max
 400     if (not self.timeout and not self.nag_timer and
 401         not self.stdout_cb and not self.stderr_cb):
 402       return super(Popen, self).communicate(input)
 403
 404     if self.timeout and self.shell:
 405       raise TypeError(
 406           'Using timeout and shell simultaneously will cause a process leak '
 407           'since the shell will be killed instead of the child process.')
 408
 409     stdout = None
 410     stderr = None
 411     # Convert to a lambda to workaround python's deadlock.
 412     # http://docs.python.org/library/subprocess.html#subprocess.Popen.wait
 413     # When the pipe fills up, it would deadlock this process.
 414     if self.stdout and not self.stdout_cb and not self.stdout_is_void:
 415       stdout = []
 416       self.stdout_cb = stdout.append
 417     if self.stderr and not self.stderr_cb and not self.stderr_is_void:
 418       stderr = []
 419       self.stderr_cb = stderr.append
 420     self._tee_threads(input)
 421     if stdout is not None:
 422       stdout = ''.join(stdout)
 423     if stderr is not None:
 424       stderr = ''.join(stderr)
 425     return (stdout, stderr)
 426
 427
 428 def communicate(args, timeout=None, nag_timer=None, nag_max=None, **kwargs):
 429   """Wraps subprocess.Popen().communicate() and add timeout support.
 430
 431   Returns ((stdout, stderr), returncode).
 432
 433   - The process will be killed after |timeout| seconds and returncode set to
 434     TIMED_OUT.
 435   - If the subprocess runs for |nag_timer| seconds without producing terminal
 436     output, print a warning to stderr.
 437   - Automatically passes stdin content as input so do not specify stdin=PIPE.
 438   """
 439   stdin = kwargs.pop('stdin', None)
 440   if stdin is not None:
 441     if isinstance(stdin, basestring):
 442       # When stdin is passed as an argument, use it as the actual input data and
 443       # set the Popen() parameter accordingly.
 444       kwargs['stdin'] = PIPE
 445     else:
 446       kwargs['stdin'] = stdin
 447       stdin = None
 448
 449   proc = Popen(args, **kwargs)
 450   if stdin:
 451     return proc.communicate(stdin, timeout, nag_timer), proc.returncode
 452   else:
 453     return proc.communicate(None, timeout, nag_timer), proc.returncode
 454
 455
 456 def call(args, **kwargs):
 457   """Emulates subprocess.call().
 458
 459   Automatically convert stdout=PIPE or stderr=PIPE to VOID.
 460   In no case they can be returned since no code path raises
 461   subprocess2.CalledProcessError.
 462   """
 463   if kwargs.get('stdout') == PIPE:
 464     kwargs['stdout'] = VOID
 465   if kwargs.get('stderr') == PIPE:
 466     kwargs['stderr'] = VOID
 467   return communicate(args, **kwargs)[1]
 468
 469
 470 def check_call_out(args, **kwargs):
 471   """Improved version of subprocess.check_call().
 472
 473   Returns (stdout, stderr), unlike subprocess.check_call().
 474   """
 475   out, returncode = communicate(args, **kwargs)
 476   if returncode:
 477     raise CalledProcessError(
 478         returncode, args, kwargs.get('cwd'), out[0], out[1])
 479   return out
 480
 481
 482 def check_call(args, **kwargs):
 483   """Emulate subprocess.check_call()."""
 484   check_call_out(args, **kwargs)
 485   return 0
 486
 487
 488 def capture(args, **kwargs):
 489   """Captures stdout of a process call and returns it.
 490
 491   Returns stdout.
 492
 493   - Discards returncode.
 494   - Blocks stdin by default if not specified since no output will be visible.
 495   """
 496   kwargs.setdefault('stdin', VOID)
 497
 498   # Like check_output, deny the caller from using stdout arg.
 499   return communicate(args, stdout=PIPE, **kwargs)[0][0]
 500
 501
 502 def check_output(args, **kwargs):
 503   """Emulates subprocess.check_output().
 504
 505   Captures stdout of a process call and returns stdout only.
 506
 507   - Throws if return code is not 0.
 508   - Works even prior to python 2.7.
 509   - Blocks stdin by default if not specified since no output will be visible.
 510   - As per doc, "The stdout argument is not allowed as it is used internally."
 511   """
 512   kwargs.setdefault('stdin', VOID)
 513   if 'stdout' in kwargs:
 514     raise ValueError('stdout argument not allowed, it would be overridden.')
 515   return check_call_out(args, stdout=PIPE, **kwargs)[0]