Use subprocess42 in recipe_engine to avoid threading madness.

recipe_engine/step_runner.py

Index: recipe_engine/step_runner.py
diff --git a/recipe_engine/step_runner.py b/recipe_engine/step_runner.py
index c4d166f9abc283ddd36ece58a27ecb81f8756d06..c3b5a23245daa09a04f22930207df2376feda96d 100644
--- a/recipe_engine/step_runner.py
+++ b/recipe_engine/step_runner.py
@@ -9,10 +9,8 @@ import datetime
 import json
 import os
 import re
-import subprocess
 import sys
 import tempfile
-import threading
 import traceback
 
 from . import recipe_test_api
@@ -20,6 +18,45 @@ 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.
 
@@ -152,9 +189,9 @@ class SubprocessStepRunner(StepRunner):
             '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'))
@@ -191,7 +228,7 @@ class SubprocessStepRunner(StepRunner):
              '--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(
@@ -251,39 +288,22 @@ class SubprocessStepRunner(StepRunner):
     # 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.
@@ -291,35 +311,27 @@ class SubprocessStepRunner(StepRunner):
       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
-
-    threads = []
+    outstreams = {}
+    linebufs = {}
     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]))
-
-    for th in threads:
-      th.start()
-    proc.wait()
-    for th in threads:
-      th.join()
+      handle = handles.get(key)
+      if isinstance(handle, stream.StreamEngine.Stream):
+        outstreams[key] = handle
+        linebufs[key] = _streamingLinebuf()
+
+    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()
+
     return proc.returncode
 
   def _trigger_builds(self, step, trigger_specs):