Use subprocess42 in recipe_engine to avoid threading madness.

recipe_engine/step_runner.py

 # Copyright 2016 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.
 
 import cStringIO
 import collections
 import contextlib
 import datetime
 import json
 import os
 import re
-import subprocess
 import sys
 import tempfile
-import threading
 import traceback
 
 from . import recipe_test_api
 from . import stream
 from . import types
 from . import util
 
+from .third_party import subprocess42
+
+if sys.platform.startswith('win'):

[M-A Ruel, 2016/05/06 14:12:51]

if sys.platform == 'win32':

[iannucci, 2016/05/06 17:52:01]

Done.

+  # Windows has a bad habit of opening a dialog when a console program
+  # crashes, rather than just letting it crash.  Therefore, when a
+  # program crashes on Windows, we don't find out until the build step
+  # times out.  This code prevents the dialog from appearing, so that we
+  # find out immediately and don't waste time waiting for a user to
+  # close the dialog.
+  import ctypes
+  # SetErrorMode(SEM_NOGPFAULTERRORBOX). For more information, see:
+  # https://msdn.microsoft.com/en-us/library/windows/desktop/ms680621.aspx
+  ctypes.windll.kernel32.SetErrorMode(0x0002)

[M-A Ruel, 2016/05/06 14:12:51]

s/0x0002/2/ ? :)

That's a good idea! I hadn't thought about that.

[iannucci, 2016/05/06 17:52:01]

I was matching the text in the msdn docs :)

Looking at the docs closer, I think we should be setting all of those error
flags.
Should I move this to subprocess42?

[M-A Ruel, 2016/05/06 17:57:51]

Good question. I'm not fond of changing the global state automatically. I guess
an helper function could be useful, but the real saving is just about
copy-pasting 3 lines which I don't see as being a big deal. As you want.

+

[M-A Ruel, 2016/05/06 14:12:51]

you don't use 2 lines between file level symbols?

[iannucci, 2016/05/06 17:52:01]

lol, done

+class _streamingLinebuf(object):
+  def __init__(self):
+    self.buffedlines = []
+    self.extra = None
+
+  def ingest(self, data):
+    lines = data.splitlines()
+    if self.extra:
+      lines[0] = self.extra + lines[0]
+      self.extra = None
+    if not data.endswith("\n"):
+      self.extra = lines[-1]
+      lines = lines[:-1]
+    self.buffedlines += lines
+
+  def __iter__(self):
+    return self
+
+  def next(self):
+    if len(self.buffedlines) > 0:
+      ret = self.buffedlines[0]
+      self.buffedlines = self.buffedlines[1:]
+      return ret
+    raise StopIteration()
+
 class StepRunner(object):
   """A StepRunner is the interface to actually running steps.
 
   These can actually make subprocess calls (SubprocessStepRunner), or just
   pretend to run the steps with mock output (SimulationStepRunner).
   """
   @property
   def stream_engine(self):
     """Return the stream engine that this StepRunner uses, if meaningful.
 
@@ skipping 112 matching lines @@
       def run(inner):
         try:
           # Open file handles for IO redirection based on file names in
           # step_dict.
           handles = {
             'stdout': step_stream,
             'stderr': step_stream,
             'stdin': None,
           }
           for key in handles:
-            if key in step_dict:
-              handles[key] = open(step_dict[key],
-                                   'rb' if key == 'stdin' else 'wb')
+            fileName = step_dict.get(key)
+            if fileName:
+              handles[key] = open(fileName, 'rb' if key == 'stdin' else 'wb')
           # The subprocess will inherit and close these handles.
           retcode = self._run_cmd(
               cmd=cmd, handles=handles, env=step_env, cwd=step_dict.get('cwd'))
         except OSError:
           with step_stream.new_log_stream('exception') as l:
             trace = traceback.format_exc().splitlines()
             for line in trace:
               l.write_line(line)
           step_stream.set_step_status('EXCEPTION')
           raise
@@ skipping 14 matching lines @@
         return step_stream
 
     return ReturnOpenStep()
 
   def run_recipe(self, universe_view, recipe, properties):
     with tempfile.NamedTemporaryFile() as f:
       cmd = [sys.executable,
              universe_view.universe.package_deps.engine_recipes_py,
              '--package=%s' % universe_view.universe.config_file.path, 'run',
              '--output-result-json=%s' % f.name, recipe]
       cmd.extend(['%s=%s' % (k,repr(v)) for k, v in properties.iteritems()])

[M-A Ruel, 2016/05/06 14:12:51]

[OT] don't you want to sort the resulting list just for determinism?

[iannucci, 2016/05/06 17:52:01]

I mean... probably, but this code is going to be reworked when I do DM
integration anyhow.

 
-      retcode = subprocess.call(cmd)
+      retcode = subprocess42.call(cmd)
       result = json.load(f)
       if retcode != 0:
         raise recipe_api.StepFailure(
           'depend on %s with properties %r failed with %d.\n'
           'Recipe result: %r' % (
               recipe, properties, retcode, result))
       return result
 
   @contextlib.contextmanager
   def run_context(self):
@@ skipping 39 matching lines @@
         to direct that subprocess's filehandle to.
       env: the full environment to run the command in -- this is passed
         unaltered to subprocess.Popen.
       cwd: the working directory of the command.
     """
     fhandles = {}
 
     # If we are given StreamEngine.Streams, map them to PIPE for subprocess.
     # We will manually forward them to their corresponding stream.
     for key in ('stdout', 'stderr'):
-      if (key in handles and
-          isinstance(handles[key], stream.StreamEngine.Stream)):
-        fhandles[key] = subprocess.PIPE
+      handle = handles.get(key)
+      if isinstance(handle, stream.StreamEngine.Stream):
+        fhandles[key] = subprocess42.PIPE
       else:
-        fhandles[key] = handles[key]
+        fhandles[key] = handle
 
     # stdin must be a real handle, if it exists
     fhandles['stdin'] = handles.get('stdin')
 
-    if sys.platform.startswith('win'):
-      # Windows has a bad habit of opening a dialog when a console program
-      # crashes, rather than just letting it crash.  Therefore, when a
-      # program crashes on Windows, we don't find out until the build step
-      # times out.  This code prevents the dialog from appearing, so that we
-      # find out immediately and don't waste time waiting for a user to
-      # close the dialog.
-      import ctypes
-      # SetErrorMode(SEM_NOGPFAULTERRORBOX). For more information, see:
-      # https://msdn.microsoft.com/en-us/library/windows/desktop/ms680621.aspx
-      ctypes.windll.kernel32.SetErrorMode(0x0002)
-      # CREATE_NO_WINDOW. For more information, see:
-      # https://msdn.microsoft.com/en-us/library/windows/desktop/ms684863.aspx
-      creationflags = 0x8000000
-    else:
-      creationflags = 0
-
     with _modify_lookup_path(env.get('PATH')):
-      proc = subprocess.Popen(
+      proc = subprocess42.Popen(
           cmd,
           env=env,
           cwd=cwd,
           universal_newlines=True,
-          creationflags=creationflags,
+          detached=True,
           **fhandles)
 
     # Safe to close file handles now that subprocess has inherited them.
     for handle in fhandles.itervalues():
       if isinstance(handle, file):
         handle.close()
 
-    outlock = threading.Lock()
-    def make_pipe_thread(inhandle, outstream):
-      def body():
-        while True:
-          line = inhandle.readline()
-          if not line:
-            break
-          line = line[:-1] # Strip newline for write_line's expectations
-                           # (universal_newlines is on, so it's only \n)
-          outlock.acquire()
-          try:
-            outstream.write_line(line)
-          finally:
-            outlock.release()
-      t = threading.Thread(target=body, args=())
-      t.daemon = True
-      return t
+    outstreams = {}
+    linebufs = {}
+    for key in ('stdout', 'stderr'):
+      handle = handles.get(key)
+      if isinstance(handle, stream.StreamEngine.Stream):
+        outstreams[key] = handle
+        linebufs[key] = _streamingLinebuf()
 
-    threads = []
-    for key in ('stdout', 'stderr'):
-      if (key in handles and
-          isinstance(handles[key], stream.StreamEngine.Stream)):
-        threads.append(make_pipe_thread(getattr(proc, key), handles[key]))
+    if len(linebufs) > 0:

[M-A Ruel, 2016/05/06 14:12:52]

if linebufs:

[iannucci, 2016/05/06 17:52:01]

Done.

+      for pipe, data in proc.yield_any(timeout=1):
+        if pipe is None:
+          continue
+        buf = linebufs.get(pipe)
+        if not buf:
+          continue
+        buf.ingest(data)
+        for line in buf:
+          outstreams[pipe].write_line(line)
+    else:
+      proc.wait()
 
-    for th in threads:
-      th.start()
-    proc.wait()
-    for th in threads:
-      th.join()
     return proc.returncode
 
   def _trigger_builds(self, step, trigger_specs):
     assert trigger_specs is not None
     for trig in trigger_specs:
       builder_name = trig.get('builder_name')
       if not builder_name:
         raise ValueError('Trigger spec: builder_name is not set')
 
       changes = trig.get('buildbot_changes', [])
@@ skipping 279 matching lines @@
   supplied command, and only uses the |env| kwarg for modifying the environment
   of the child process.
   """
   saved_path = os.environ['PATH']
   try:
     if path is not None:
       os.environ['PATH'] = path
     yield
   finally:
     os.environ['PATH'] = saved_path