Reland "presubmit support: Run all tests in parallel."

Currently all tests in a PRESUBMIT.py file are run in parallel, but not
all tests across PRESUBMIT.py files.

This introduces a flag that will allow presubmit to run all tests across
PRESUBMIT files in parallel.

Bug: 819774
Change-Id: Idd3046cb3c85e9c28932a9789ba7b207a01d9f99
Reviewed-on: https://chromium-review.googlesource.com/994241
Reviewed-by: Aaron Gable <agable@chromium.org>
Commit-Queue: Edward Lesmes <ehmaldonado@chromium.org>

git_cl.py

@@ -1519,13 +1519,14 @@ class Changelist(object):
         new_description += foot + '\n'
     self.UpdateDescription(new_description, force)
 
-  def RunHook(self, committing, may_prompt, verbose, change):
+  def RunHook(self, committing, may_prompt, verbose, change, parallel):
     """Calls sys.exit() if the hook fails; returns a HookResults otherwise."""
     try:
       return presubmit_support.DoPresubmitChecks(change, committing,
           verbose=verbose, output_stream=sys.stdout, input_stream=sys.stdin,
           default_presubmit=None, may_prompt=may_prompt,
-          gerrit_obj=self._codereview_impl.GetGerritObjForPresubmit())
+          gerrit_obj=self._codereview_impl.GetGerritObjForPresubmit(),
+          parallel=parallel)
     except presubmit_support.PresubmitFailure as e:
       DieWithError('%s\nMaybe your depot_tools is out of date?' % e)
 
@@ -1591,7 +1592,7 @@ class Changelist(object):
       hook_results = self.RunHook(committing=False,
                                   may_prompt=not options.force,
                                   verbose=options.verbose,
-                                change=change)
+                                  change=change, parallel=options.parallel)
       if not hook_results.should_continue():
         return 1
       if not options.reviewers and hook_results.reviewers:
@@ -4793,6 +4794,9 @@ def CMDpresubmit(parser, args):
                     help='Run checks even if tree is dirty')
   parser.add_option('--all', action='store_true',
                     help='Run checks against all files, not just modified ones')
+  parser.add_option('--parallel', action='store_true',
+                    help='Run all tests specified by input_api.RunTests in all '
+                         'PRESUBMIT files in parallel.')
   auth.add_auth_options(parser)
   options, args = parser.parse_args(args)
   auth_config = auth.extract_auth_config_from_options(options)
@@ -4827,7 +4831,8 @@ def CMDpresubmit(parser, args):
       committing=not options.upload,
       may_prompt=False,
       verbose=options.verbose,
-      change=change)
+      change=change,
+      parallel=options.parallel)
   return 0
 
 
@@ -5009,6 +5014,9 @@ def CMDupload(parser, args):
                     help='Sends your change to the CQ after an approval. Only '
                          'works on repos that have the Auto-Submit label '
                          'enabled')
+  parser.add_option('--parallel', action='store_true',
+                    help='Run all tests specified by input_api.RunTests in all '
+                         'PRESUBMIT files in parallel.')
 
   # TODO: remove Rietveld flags
   parser.add_option('--private', action='store_true',

presubmit_support.py

@@ -30,8 +30,10 @@ import os  # Somewhat exposed through the API.
 import pickle  # Exposed through the API.
 import random
 import re  # Exposed through the API.
+import signal
 import sys  # Parts exposed through API.
 import tempfile  # Exposed through the API.
+import threading
 import time
 import traceback  # Exposed through the API.
 import types
@@ -64,11 +66,154 @@ class CommandData(object):
   def __init__(self, name, cmd, kwargs, message):
     self.name = name
     self.cmd = cmd
+    self.stdin = kwargs.get('stdin', None)
     self.kwargs = kwargs
+    self.kwargs['stdout'] = subprocess.PIPE
+    self.kwargs['stderr'] = subprocess.STDOUT
+    self.kwargs['stdin'] = subprocess.PIPE
     self.message = message
     self.info = None
 
 
+# Adapted from
+# https://github.com/google/gtest-parallel/blob/master/gtest_parallel.py#L37
+#
+# An object that catches SIGINT sent to the Python process and notices
+# if processes passed to wait() die by SIGINT (we need to look for
+# both of those cases, because pressing Ctrl+C can result in either
+# the main process or one of the subprocesses getting the signal).
+#
+# Before a SIGINT is seen, wait(p) will simply call p.wait() and
+# return the result. Once a SIGINT has been seen (in the main process
+# or a subprocess, including the one the current call is waiting for),
+# wait(p) will call p.terminate() and raise ProcessWasInterrupted.
+class SigintHandler(object):
+  class ProcessWasInterrupted(Exception):
+    pass
+
+  sigint_returncodes = {-signal.SIGINT,  # Unix
+                        -1073741510,     # Windows
+                        }
+  def __init__(self):
+    self.__lock = threading.Lock()
+    self.__processes = set()
+    self.__got_sigint = False
+    signal.signal(signal.SIGINT, lambda signal_num, frame: self.interrupt())
+
+  def __on_sigint(self):
+    self.__got_sigint = True
+    while self.__processes:
+      try:
+        self.__processes.pop().terminate()
+      except OSError:
+        pass
+
+  def interrupt(self):
+    with self.__lock:
+      self.__on_sigint()
+
+  def got_sigint(self):
+    with self.__lock:
+      return self.__got_sigint
+
+  def wait(self, p, stdin):
+    with self.__lock:
+      if self.__got_sigint:
+        p.terminate()
+      self.__processes.add(p)
+    stdout, stderr = p.communicate(stdin)
+    code = p.returncode
+    with self.__lock:
+      self.__processes.discard(p)
+      if code in self.sigint_returncodes:
+        self.__on_sigint()
+      if self.__got_sigint:
+        raise self.ProcessWasInterrupted
+    return stdout, stderr
+
+sigint_handler = SigintHandler()
+
+
+class ThreadPool(object):
+  def __init__(self, pool_size=None):
+    self._pool_size = pool_size or multiprocessing.cpu_count()
+    self._messages = []
+    self._messages_lock = threading.Lock()
+    self._tests = []
+    self._tests_lock = threading.Lock()
+    self._nonparallel_tests = []
+
+  def CallCommand(self, test):
+    """Runs an external program.
+
+    This function converts invocation of .py files and invocations of "python"
+    to vpython invocations.
+    """
+    vpython = 'vpython.bat' if sys.platform == 'win32' else 'vpython'
+
+    cmd = test.cmd
+    if cmd[0] == 'python':
+      cmd = list(cmd)
+      cmd[0] = vpython
+    elif cmd[0].endswith('.py'):
+      cmd = [vpython] + cmd
+
+    try:
+      start = time.time()
+      p = subprocess.Popen(cmd, **test.kwargs)
+      stdout, _ = sigint_handler.wait(p, test.stdin)
+      duration = time.time() - start
+    except OSError as e:
+      duration = time.time() - start
+      return test.message(
+          '%s exec failure (%4.2fs)\n   %s' % (test.name, duration, e))
+    if p.returncode != 0:
+      return test.message(
+          '%s (%4.2fs) failed\n%s' % (test.name, duration, stdout))
+    if test.info:
+      return test.info('%s (%4.2fs)' % (test.name, duration))
+
+  def AddTests(self, tests, parallel=True):
+    if parallel:
+      self._tests.extend(tests)
+    else:
+      self._nonparallel_tests.extend(tests)
+
+  def RunAsync(self):
+    self._messages = []
+
+    def _WorkerFn():
+      while True:
+        test = None
+        with self._tests_lock:
+          if not self._tests:
+            break
+          test = self._tests.pop()
+        result = self.CallCommand(test)
+        if result:
+          with self._messages_lock:
+            self._messages.append(result)
+
+    def _StartDaemon():
+      t = threading.Thread(target=_WorkerFn)
+      t.daemon = True
+      t.start()
+      return t
+
+    while self._nonparallel_tests:
+      test = self._nonparallel_tests.pop()
+      result = self.CallCommand(test)
+      if result:
+        self._messages.append(result)
+
+    if self._tests:
+      threads = [_StartDaemon() for _ in range(self._pool_size)]
+      for worker in threads:
+        worker.join()
+
+    return self._messages
+
+
 def normpath(path):
   '''Version of os.path.normpath that also changes backward slashes to
   forward slashes when not running on Windows.
@@ -388,7 +533,7 @@ class InputApi(object):
   )
 
   def __init__(self, change, presubmit_path, is_committing,
-      verbose, gerrit_obj, dry_run=None):
+      verbose, gerrit_obj, dry_run=None, thread_pool=None, parallel=False):
     """Builds an InputApi object.
 
     Args:
@@ -397,6 +542,8 @@ class InputApi(object):
       is_committing: True if the change is about to be committed.
       gerrit_obj: provides basic Gerrit codereview functionality.
       dry_run: if true, some Checks will be skipped.
+      parallel: if true, all tests reported via input_api.RunTests for all
+                PRESUBMIT files will be run in parallel.
     """
     # Version number of the presubmit_support script.
     self.version = [int(x) for x in __version__.split('.')]
@@ -405,6 +552,9 @@ class InputApi(object):
     self.gerrit = gerrit_obj
     self.dry_run = dry_run
 
+    self.parallel = parallel
+    self.thread_pool = thread_pool or ThreadPool()
+
     # We expose various modules and functions as attributes of the input_api
     # so that presubmit scripts don't have to import them.
     self.ast = ast
@@ -444,12 +594,6 @@ class InputApi(object):
 
     self.cpu_count = multiprocessing.cpu_count()
 
-    # this is done here because in RunTests, the current working directory has
-    # changed, which causes Pool() to explode fantastically when run on windows
-    # (because it tries to load the __main__ module, which imports lots of
-    # things relative to the current working directory).
-    self._run_tests_pool = multiprocessing.Pool(self.cpu_count)
-
     # The local path of the currently-being-processed presubmit script.
     self._current_presubmit_path = os.path.dirname(presubmit_path)
 
@@ -627,27 +771,23 @@ class InputApi(object):
     return 'TBR' in self.change.tags or self.change.TBRsFromDescription()
 
   def RunTests(self, tests_mix, parallel=True):
+    # RunTests doesn't actually run tests. It adds them to a ThreadPool that
+    # will run all tests once all PRESUBMIT files are processed.
     tests = []
     msgs = []
     for t in tests_mix:
-      if isinstance(t, OutputApi.PresubmitResult):
+      if isinstance(t, OutputApi.PresubmitResult) and t:
         msgs.append(t)
       else:
         assert issubclass(t.message, _PresubmitResult)
         tests.append(t)
         if self.verbose:
           t.info = _PresubmitNotifyResult
-    if len(tests) > 1 and parallel:
+        t.kwargs['cwd'] = self.PresubmitLocalPath()
-      # async recipe works around multiprocessing bug handling Ctrl-C
+    self.thread_pool.AddTests(tests, parallel)
-      msgs.extend(self._run_tests_pool.map_async(CallCommand, tests).get(99999))
+    if not self.parallel:
-    else:
+      msgs.extend(self.thread_pool.RunAsync())
-      msgs.extend(map(CallCommand, tests))
+    return msgs
-    return [m for m in msgs if m]
-
-  def ShutdownPool(self):
-    self._run_tests_pool.close()
-    self._run_tests_pool.join()
-    self._run_tests_pool = None
 
 
 class _DiffCache(object):
@@ -1265,13 +1405,15 @@ def DoPostUploadExecuter(change,
 
 class PresubmitExecuter(object):
   def __init__(self, change, committing, verbose,
-               gerrit_obj, dry_run=None):
+               gerrit_obj, dry_run=None, thread_pool=None, parallel=False):
     """
     Args:
       change: The Change object.
       committing: True if 'git cl land' is running, False if 'git cl upload' is.
       gerrit_obj: provides basic Gerrit codereview functionality.
       dry_run: if true, some Checks will be skipped.
+      parallel: if true, all tests reported via input_api.RunTests for all
+                PRESUBMIT files will be run in parallel.
     """
     self.change = change
     self.committing = committing
@@ -1279,6 +1421,8 @@ class PresubmitExecuter(object):
     self.verbose = verbose
     self.dry_run = dry_run
     self.more_cc = []
+    self.thread_pool = thread_pool
+    self.parallel = parallel
 
   def ExecPresubmitScript(self, script_text, presubmit_path):
     """Executes a single presubmit script.
@@ -1299,7 +1443,8 @@ class PresubmitExecuter(object):
     # Load the presubmit script into context.
     input_api = InputApi(self.change, presubmit_path, self.committing,
                          self.verbose, gerrit_obj=self.gerrit,
-                         dry_run=self.dry_run)
+                         dry_run=self.dry_run, thread_pool=self.thread_pool,
+                         parallel=self.parallel)
     output_api = OutputApi(self.committing)
     context = {}
     try:
@@ -1334,8 +1479,6 @@ class PresubmitExecuter(object):
     else:
       result = ()  # no error since the script doesn't care about current event.
 
-    input_api.ShutdownPool()
-
     # Return the process to the original working directory.
     os.chdir(main_path)
     return result
@@ -1348,7 +1491,8 @@ def DoPresubmitChecks(change,
                       default_presubmit,
                       may_prompt,
                       gerrit_obj,
-                      dry_run=None):
+                      dry_run=None,
+                      parallel=False):
   """Runs all presubmit checks that apply to the files in the change.
 
   This finds all PRESUBMIT.py files in directories enclosing the files in the
@@ -1369,6 +1513,8 @@ def DoPresubmitChecks(change,
                 any questions are answered with yes by default.
     gerrit_obj: provides basic Gerrit codereview functionality.
     dry_run: if true, some Checks will be skipped.
+    parallel: if true, all tests specified by input_api.RunTests in all
+              PRESUBMIT files will be run in parallel.
 
   Warning:
     If may_prompt is true, output_stream SHOULD be sys.stdout and input_stream
@@ -1395,8 +1541,9 @@ def DoPresubmitChecks(change,
     if not presubmit_files and verbose:
       output.write("Warning, no PRESUBMIT.py found.\n")
     results = []
+    thread_pool = ThreadPool()
     executer = PresubmitExecuter(change, committing, verbose,
-                                 gerrit_obj, dry_run)
+                                 gerrit_obj, dry_run, thread_pool)
     if default_presubmit:
       if verbose:
         output.write("Running default presubmit script.\n")
@@ -1410,6 +1557,8 @@ def DoPresubmitChecks(change,
       presubmit_script = gclient_utils.FileRead(filename, 'rU')
       results += executer.ExecPresubmitScript(presubmit_script, filename)
 
+    results += thread_pool.RunAsync()
+
     output.more_cc.extend(executer.more_cc)
     errors = []
     notifications = []
@@ -1517,41 +1666,6 @@ def canned_check_filter(method_names):
       setattr(presubmit_canned_checks, name, method)
 
 
-def CallCommand(cmd_data):
-  """Runs an external program, potentially from a child process created by the
-  multiprocessing module.
-
-  multiprocessing needs a top level function with a single argument.
-
-  This function converts invocation of .py files and invocations of "python" to
-  vpython invocations.
-  """
-  vpython = 'vpython.bat' if sys.platform == 'win32' else 'vpython'
-
-  cmd = cmd_data.cmd
-  if cmd[0] == 'python':
-    cmd = list(cmd)
-    cmd[0] = vpython
-  elif cmd[0].endswith('.py'):
-    cmd = [vpython] + cmd
-
-  cmd_data.kwargs['stdout'] = subprocess.PIPE
-  cmd_data.kwargs['stderr'] = subprocess.STDOUT
-  try:
-    start = time.time()
-    (out, _), code = subprocess.communicate(cmd, **cmd_data.kwargs)
-    duration = time.time() - start
-  except OSError as e:
-    duration = time.time() - start
-    return cmd_data.message(
-        '%s exec failure (%4.2fs)\n   %s' % (cmd_data.name, duration, e))
-  if code != 0:
-    return cmd_data.message(
-        '%s (%4.2fs) failed\n%s' % (cmd_data.name, duration, out))
-  if cmd_data.info:
-    return cmd_data.info('%s (%4.2fs)' % (cmd_data.name, duration))
-
-
 def main(argv=None):
   parser = optparse.OptionParser(usage="%prog [options] <files...>",
                                  version="%prog " + str(__version__))
@@ -1587,6 +1701,9 @@ def main(argv=None):
   parser.add_option("--gerrit_url", help=optparse.SUPPRESS_HELP)
   parser.add_option("--gerrit_fetch", action='store_true',
                     help=optparse.SUPPRESS_HELP)
+  parser.add_option('--parallel', action='store_true',
+                    help='Run all tests specified by input_api.RunTests in all '
+                         'PRESUBMIT files in parallel.')
 
   options, args = parser.parse_args(argv)
 
@@ -1630,7 +1747,8 @@ def main(argv=None):
           options.default_presubmit,
           options.may_prompt,
           gerrit_obj,
-          options.dry_run)
+          options.dry_run,
+          options.parallel)
     return not results.should_continue()
   except PresubmitFailure, e:
     print >> sys.stderr, e

tests/presubmit_unittest.py

@@ -22,6 +22,7 @@ _ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 sys.path.insert(0, _ROOT)
 
 from testing_support.super_mox import mox, SuperMoxTestBase
+from third_party import mock
 
 import owners
 import owners_finder
@@ -172,16 +173,18 @@ class PresubmitUnittest(PresubmitTestsBase):
     self.mox.ReplayAll()
     members = [
       'AffectedFile', 'Change', 'DoPostUploadExecuter', 'DoPresubmitChecks',
-      'GetPostUploadExecuter', 'GitAffectedFile', 'CallCommand', 'CommandData',
+      'GetPostUploadExecuter', 'GitAffectedFile', 'CommandData',
       'GitChange', 'InputApi', 'ListRelevantPresubmitFiles', 'main',
       'OutputApi', 'ParseFiles',
       'PresubmitFailure', 'PresubmitExecuter', 'PresubmitOutput', 'ScanSubDirs',
+      'SigintHandler', 'ThreadPool',
       'ast', 'cPickle', 'cpplint', 'cStringIO', 'contextlib',
       'canned_check_filter', 'fix_encoding', 'fnmatch', 'gclient_utils',
       'git_footers', 'glob', 'inspect', 'json', 'load_files', 'logging',
       'marshal', 'normpath', 'optparse', 'os', 'owners', 'owners_finder',
       'pickle', 'presubmit_canned_checks', 'random', 're', 'scm',
-      'subprocess', 'sys', 'tempfile',
+      'sigint_handler', 'signal',
+      'subprocess', 'sys', 'tempfile', 'threading',
       'time', 'traceback', 'types', 'unittest',
       'urllib2', 'warn', 'multiprocessing', 'DoGetTryMasters',
       'GetTryMastersExecuter', 'itertools', 'urlparse', 'gerrit_util',
@@ -893,7 +896,7 @@ def CheckChangeOnCommit(input_api, output_api):
     presubmit.DoPresubmitChecks(mox.IgnoreArg(), False, False,
                                 mox.IgnoreArg(),
                                 mox.IgnoreArg(),
-                                None, False, None, None).AndReturn(output)
+                                None, False, None, None, None).AndReturn(output)
     self.mox.ReplayAll()
 
     self.assertEquals(
@@ -942,7 +945,6 @@ class InputApiUnittest(PresubmitTestsBase):
         'PresubmitLocalPath',
         'ReadFile',
         'RightHandSideLines',
-        'ShutdownPool',
         'ast',
         'basename',
         'cPickle',
@@ -965,6 +967,7 @@ class InputApiUnittest(PresubmitTestsBase):
         'os_stat',
         'owners_db',
         'owners_finder',
+        'parallel',
         'pickle',
         'platform',
         'python_executable',
@@ -972,6 +975,7 @@ class InputApiUnittest(PresubmitTestsBase):
         'subprocess',
         'tbr',
         'tempfile',
+        'thread_pool',
         'time',
         'traceback',
         'unittest',
@@ -1644,19 +1648,29 @@ class ChangeUnittest(PresubmitTestsBase):
     self.assertEquals('bar,baz,foo', change.TBR)
 
 
-def CommHelper(input_api, cmd, ret=None, **kwargs):
+class CannedChecksUnittest(PresubmitTestsBase):
+  """Tests presubmit_canned_checks.py."""
+  def CommHelper(self, input_api, cmd, stdin=None, ret=None, **kwargs):
     ret = ret or (('', None), 0)
-  input_api.subprocess.communicate(
+    kwargs.setdefault('cwd', mox.IgnoreArg())
-      cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, **kwargs
+    kwargs.setdefault('stdin', subprocess.PIPE)
-      ).AndReturn(ret)
 
+    mock_process = input_api.mox.CreateMockAnything()
+    mock_process.returncode = ret[1]
 
-class CannedChecksUnittest(PresubmitTestsBase):
+    input_api.PresubmitLocalPath().AndReturn(self.fake_root_dir)
-  """Tests presubmit_canned_checks.py."""
+    input_api.subprocess.Popen(
+        cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, **kwargs
+    ).AndReturn(mock_process)
+
+    presubmit.sigint_handler.wait(mock_process, stdin).AndReturn(ret[0])
 
   def MockInputApi(self, change, committing):
     # pylint: disable=no-self-use
     input_api = self.mox.CreateMock(presubmit.InputApi)
+    input_api.mox = self.mox
+    input_api.thread_pool = presubmit.ThreadPool()
+    input_api.parallel = False
     input_api.cStringIO = presubmit.cStringIO
     input_api.json = presubmit.json
     input_api.logging = logging
@@ -1689,6 +1703,7 @@ class CannedChecksUnittest(PresubmitTestsBase):
     input_api.Command = presubmit.CommandData
     input_api.RunTests = functools.partial(
         presubmit.InputApi.RunTests, input_api)
+    presubmit.sigint_handler = self.mox.CreateMock(presubmit.SigintHandler)
     return input_api
 
   def testMembersChanged(self):
@@ -2198,14 +2213,15 @@ class CannedChecksUnittest(PresubmitTestsBase):
   def testRunPythonUnitTestsNoTest(self):
     input_api = self.MockInputApi(None, False)
     self.mox.ReplayAll()
-    results = presubmit_canned_checks.RunPythonUnitTests(
+    presubmit_canned_checks.RunPythonUnitTests(
         input_api, presubmit.OutputApi, [])
+    results = input_api.thread_pool.RunAsync()
     self.assertEquals(results, [])
 
   def testRunPythonUnitTestsNonExistentUpload(self):
     input_api = self.MockInputApi(None, False)
-    CommHelper(input_api, ['pyyyyython', '-m', '_non_existent_module'],
+    self.CommHelper(input_api, ['pyyyyython', '-m', '_non_existent_module'],
-                    ret=(('foo', None), 1), cwd=None, env=None)
+                    ret=(('foo', None), 1), env=None)
     self.mox.ReplayAll()
 
     results = presubmit_canned_checks.RunPythonUnitTests(
@@ -2216,8 +2232,8 @@ class CannedChecksUnittest(PresubmitTestsBase):
 
   def testRunPythonUnitTestsNonExistentCommitting(self):
     input_api = self.MockInputApi(None, True)
-    CommHelper(input_api, ['pyyyyython', '-m', '_non_existent_module'],
+    self.CommHelper(input_api, ['pyyyyython', '-m', '_non_existent_module'],
-                    ret=(('foo', None), 1), cwd=None, env=None)
+                    ret=(('foo', None), 1), env=None)
     self.mox.ReplayAll()
 
     results = presubmit_canned_checks.RunPythonUnitTests(
@@ -2229,8 +2245,8 @@ class CannedChecksUnittest(PresubmitTestsBase):
     input_api = self.MockInputApi(None, False)
     input_api.unittest = self.mox.CreateMock(unittest)
     input_api.cStringIO = self.mox.CreateMock(presubmit.cStringIO)
-    CommHelper(input_api, ['pyyyyython', '-m', 'test_module'],
+    self.CommHelper(input_api, ['pyyyyython', '-m', 'test_module'],
-                    ret=(('foo', None), 1), cwd=None, env=None)
+                    ret=(('foo', None), 1), env=None)
     self.mox.ReplayAll()
 
     results = presubmit_canned_checks.RunPythonUnitTests(
@@ -2242,8 +2258,8 @@ class CannedChecksUnittest(PresubmitTestsBase):
 
   def testRunPythonUnitTestsFailureCommitting(self):
     input_api = self.MockInputApi(None, True)
-    CommHelper(input_api, ['pyyyyython', '-m', 'test_module'],
+    self.CommHelper(input_api, ['pyyyyython', '-m', 'test_module'],
-                    ret=(('foo', None), 1), cwd=None, env=None)
+                    ret=(('foo', None), 1), env=None)
     self.mox.ReplayAll()
 
     results = presubmit_canned_checks.RunPythonUnitTests(
@@ -2256,13 +2272,13 @@ class CannedChecksUnittest(PresubmitTestsBase):
     input_api = self.MockInputApi(None, False)
     input_api.cStringIO = self.mox.CreateMock(presubmit.cStringIO)
     input_api.unittest = self.mox.CreateMock(unittest)
-    CommHelper(input_api, ['pyyyyython', '-m', 'test_module'],
+    self.CommHelper(input_api, ['pyyyyython', '-m', 'test_module'], env=None)
-                    cwd=None, env=None)
     self.mox.ReplayAll()
 
-    results = presubmit_canned_checks.RunPythonUnitTests(
+    presubmit_canned_checks.RunPythonUnitTests(
         input_api, presubmit.OutputApi, ['test_module'])
-    self.assertEquals(len(results), 0)
+    results = input_api.thread_pool.RunAsync()
+    self.assertEquals(results, [])
 
   def testCannedRunPylint(self):
     input_api = self.MockInputApi(None, True)
@@ -2275,20 +2291,21 @@ class CannedChecksUnittest(PresubmitTestsBase):
     pylint = os.path.join(_ROOT, 'third_party', 'pylint.py')
     pylintrc = os.path.join(_ROOT, 'pylintrc')
 
-    CommHelper(input_api,
+    self.CommHelper(input_api,
         ['pyyyyython', pylint, '--args-on-stdin'],
         env=mox.IgnoreArg(), stdin=
-               '--rcfile=%s\n--disable=cyclic-import\n--jobs=2\nfile1.py'
+               '--rcfile=%s\n--disable=all\n--enable=cyclic-import\nfile1.py'
                % pylintrc)
-    CommHelper(input_api,
+    self.CommHelper(input_api,
         ['pyyyyython', pylint, '--args-on-stdin'],
         env=mox.IgnoreArg(), stdin=
-               '--rcfile=%s\n--disable=all\n--enable=cyclic-import\nfile1.py'
+               '--rcfile=%s\n--disable=cyclic-import\n--jobs=2\nfile1.py'
                % pylintrc)
     self.mox.ReplayAll()
 
     results = presubmit_canned_checks.RunPylint(
         input_api, presubmit.OutputApi)
+
     self.assertEquals([], results)
     self.checkstdout('')
 
@@ -2681,13 +2698,13 @@ class CannedChecksUnittest(PresubmitTestsBase):
     unit_tests = ['allo', 'bar.py']
     input_api.PresubmitLocalPath().AndReturn(self.fake_root_dir)
     input_api.PresubmitLocalPath().AndReturn(self.fake_root_dir)
-    CommHelper(input_api, ['allo', '--verbose'], cwd=self.fake_root_dir)
     cmd = ['bar.py', '--verbose']
     if input_api.platform == 'win32':
       cmd.insert(0, 'vpython.bat')
     else:
       cmd.insert(0, 'vpython')
-    CommHelper(input_api, cmd, cwd=self.fake_root_dir, ret=(('', None), 1))
+    self.CommHelper(input_api, cmd, cwd=self.fake_root_dir, ret=(('', None), 1))
+    self.CommHelper(input_api, ['allo', '--verbose'], cwd=self.fake_root_dir)
 
     self.mox.ReplayAll()
     results = presubmit_canned_checks.RunUnitTests(
@@ -2696,9 +2713,9 @@ class CannedChecksUnittest(PresubmitTestsBase):
         unit_tests)
     self.assertEqual(2, len(results))
     self.assertEqual(
-        presubmit.OutputApi.PresubmitNotifyResult, results[0].__class__)
+        presubmit.OutputApi.PresubmitNotifyResult, results[1].__class__)
     self.assertEqual(
-        presubmit.OutputApi.PresubmitPromptWarning, results[1].__class__)
+        presubmit.OutputApi.PresubmitPromptWarning, results[0].__class__)
     self.checkstdout('')
 
   def testCannedRunUnitTestsInDirectory(self):
@@ -2712,7 +2729,7 @@ class CannedChecksUnittest(PresubmitTestsBase):
     path = presubmit.os.path.join(self.fake_root_dir, 'random_directory')
     input_api.os_listdir(path).AndReturn(['.', '..', 'a', 'b', 'c'])
     input_api.os_path.isfile = lambda x: not x.endswith('.')
-    CommHelper(
+    self.CommHelper(
         input_api,
         [presubmit.os.path.join('random_directory', 'b'), '--verbose'],
         cwd=self.fake_root_dir)
@@ -2779,7 +2796,11 @@ class CannedChecksUnittest(PresubmitTestsBase):
         input_api, presubmit.OutputApi, path='/path/to/foo')
     self.assertEquals(command.cmd,
         ['cipd', 'ensure-file-verify', '-ensure-file', '/path/to/foo'])
-    self.assertEquals(command.kwargs, {})
+    self.assertEquals(command.kwargs, {
+        'stdin': subprocess.PIPE,
+        'stdout': subprocess.PIPE,
+        'stderr': subprocess.STDOUT,
+    })
 
   def testCheckCIPDManifest_content(self):
     input_api = self.MockInputApi(None, False)
@@ -2790,7 +2811,12 @@ class CannedChecksUnittest(PresubmitTestsBase):
         input_api, presubmit.OutputApi, content='manifest_content')
     self.assertEquals(command.cmd,
         ['cipd', 'ensure-file-verify', '-log-level', 'debug', '-ensure-file=-'])
-    self.assertEquals(command.kwargs, {'stdin': 'manifest_content'})
+    self.assertEquals(command.stdin, 'manifest_content')
+    self.assertEquals(command.kwargs, {
+        'stdin': subprocess.PIPE,
+        'stdout': subprocess.PIPE,
+        'stderr': subprocess.STDOUT,
+    })
 
   def testCheckCIPDPackages(self):
     content = '\n'.join([
@@ -2812,7 +2838,12 @@ class CannedChecksUnittest(PresubmitTestsBase):
         })
     self.assertEquals(command.cmd,
         ['cipd', 'ensure-file-verify', '-ensure-file=-'])
-    self.assertEquals(command.kwargs, {'stdin': content})
+    self.assertEquals(command.stdin, content)
+    self.assertEquals(command.kwargs, {
+        'stdin': subprocess.PIPE,
+        'stdout': subprocess.PIPE,
+        'stderr': subprocess.STDOUT,
+    })
 
   def testCannedCheckVPythonSpec(self):
     change = presubmit.Change('a', 'b', self.fake_root_dir, None, 0, 0, None)
@@ -2835,7 +2866,12 @@ class CannedChecksUnittest(PresubmitTestsBase):
       '-vpython-tool', 'verify'
     ])
     self.assertDictEqual(
-        commands[0].kwargs, {'stderr': input_api.subprocess.STDOUT})
+        commands[0].kwargs,
+        {
+            'stderr': input_api.subprocess.STDOUT,
+            'stdout': input_api.subprocess.PIPE,
+            'stdin': input_api.subprocess.PIPE,
+        })
     self.assertEqual(commands[0].message, presubmit.OutputApi.PresubmitError)
     self.assertIsNone(commands[0].info)