Windows: ensure exec calls don't block other greenthreads

eventlet.green.subprocess is not actually greenthread friendly on
Windows. It just uses the native subprocess.Popen in this case.

For this reason, exec calls do not yield on Windows, blocking other
greenthreads.

This change avoids this issue by wrapping the 'communicate' call
using eventlet.tpool.

We're also ensuring that subprocess.Popen uses *native* threads
internally in order to avoid deadlocks when passing data through
stdin.

Change-Id: Ic25fd1b61b5498f16e6049cbbe0877492f8aab4d
Closes-Bug: #1709586
(cherry picked from commit 3ac3c169ad)

oslo_concurrency/processutils.py

@@ -42,8 +42,20 @@ from oslo_concurrency._i18n import _
 # time module as the check because that's a monkey patched module we use
 # in combination with subprocess below, so they need to match.
 eventlet = importutils.try_import('eventlet')
-if eventlet and eventlet.patcher.is_monkey_patched(time):
-    from eventlet.green import subprocess
+eventlet_patched = eventlet and eventlet.patcher.is_monkey_patched(time)
+if eventlet_patched:
+    if os.name == 'nt':
+        # subprocess.Popen.communicate will spawn two threads consuming
+        # stdout/stderr when passing data through stdin. We need to make
+        # sure that *native* threads will be used as pipes are blocking
+        # on Windows.
+        # Recent eventlet versions actually do patch subprocess.
+        subprocess = eventlet.patcher.original('subprocess')
+        subprocess.threading = eventlet.patcher.original('threading')
+    else:
+        from eventlet.green import subprocess
+
+    from eventlet import tpool
 else:
     import subprocess
 
@@ -377,7 +389,14 @@ def execute(*cmd, **kwargs):
                 on_execute(obj)
 
             try:
-                result = obj.communicate(process_input)
+                # eventlet.green.subprocess is not really greenthread friendly
+                # on Windows. In order to avoid blocking other greenthreads,
+                # we have to wrap this call using tpool.
+                if eventlet_patched and os.name == 'nt':
+                    result = tpool.execute(obj.communicate,
+                                           process_input)
+                else:
+                    result = obj.communicate(process_input)
 
                 obj.stdin.close()  # pylint: disable=E1101
                 _returncode = obj.returncode  # pylint: disable=E1101

oslo_concurrency/tests/unit/test_processutils.py

@@ -124,6 +124,45 @@ class UtilsTest(test_base.BaseTestCase):
                 if type(e).__name__ != 'SubprocessError':
                     raise
 
+    @mock.patch.object(os, 'name', 'nt')
+    @mock.patch.object(processutils.subprocess, "Popen")
+    @mock.patch.object(processutils, 'tpool', create=True)
+    def _test_windows_execute(self, mock_tpool, mock_popen,
+                              use_eventlet=False):
+        # We want to ensure that if eventlet is used on Windows,
+        # 'communicate' calls are wrapped with eventlet.tpool.execute.
+        mock_comm = mock_popen.return_value.communicate
+        mock_comm.return_value = None
+        mock_tpool.execute.return_value = mock_comm.return_value
+
+        fake_pinput = 'fake pinput'.encode('utf-8')
+
+        with mock.patch.object(processutils, 'eventlet_patched',
+                               use_eventlet):
+            processutils.execute(
+                TRUE_UTILITY,
+                process_input=fake_pinput,
+                check_exit_code=False)
+
+        mock_popen.assert_called_once_with(
+            [TRUE_UTILITY],
+            stdin=mock.ANY, stdout=mock.ANY,
+            stderr=mock.ANY, close_fds=mock.ANY,
+            preexec_fn=mock.ANY, shell=mock.ANY,
+            cwd=mock.ANY, env=mock.ANY)
+
+        if use_eventlet:
+            mock_tpool.execute.assert_called_once_with(
+                mock_comm, fake_pinput)
+        else:
+            mock_comm.assert_called_once_with(fake_pinput)
+
+    def test_windows_execute_without_eventlet(self):
+        self._test_windows_execute()
+
+    def test_windows_execute_using_eventlet(self):
+        self._test_windows_execute(use_eventlet=True)
+
 
 class ProcessExecutionErrorTest(test_base.BaseTestCase):