Fix: Race condition on process kill: sometimes FIFO are still open (#107)

Log critical if readers are still running.
Allow exception on stream close (do not handle) to protect from leak.
Primary use SIGTERM for graceful shutdown, then retry with SIGKILL

Author: penguinolog

exec_helpers/__init__.py

@@ -51,7 +51,7 @@
     "ExecResult",
 )
 
-__version__ = "1.9.4"
+__version__ = "1.9.5"
 __author__ = "Alexey Stepanov"
 __author_email__ = "penguinolog@gmail.com"
 __maintainers__ = {

exec_helpers/_subprocess_helpers.py

@@ -40,15 +40,27 @@ def kill_proc_tree(pid, including_parent=True):  # type:(int, bool) -> None  # p
     :param including_parent: kill also parent process
     :type including_parent: bool
     """
+    def safe_stop(proc, kill=False):  # type: (psutil.Process, bool) -> None
+        """Do not crash on already stopped process."""
+        try:
+            if kill:
+                proc.kill()
+            proc.terminate()
+        except psutil.NoSuchProcess:
+            pass
+
     parent = psutil.Process(pid)
     children = parent.children(recursive=True)
     for child in children:  # type: psutil.Process
-        child.kill()
+        safe_stop(child)  # SIGTERM to allow cleanup
     _, alive = psutil.wait_procs(children, timeout=5)
-    for proc in alive:  # type: psutil.Process
-        proc.kill()  # 2nd shot
+    for child in alive:  # type: psutil.Process
+        safe_stop(child, kill=True)  # 2nd shot: SIGKILL
     if including_parent:
-        parent.kill()
+        safe_stop(parent)  # SIGTERM to allow cleanup
+        _, alive = psutil.wait_procs((parent,), timeout=5)
+        if alive:
+            safe_stop(parent, kill=True)  # 2nd shot: SIGKILL
         parent.wait(5)
 
 

exec_helpers/subprocess_runner.py

@@ -104,13 +104,13 @@ def _exec_command(  # type: ignore
         @threaded.threadpooled
         def poll_stdout():  # type: () -> None
             """Sync stdout poll."""
-            result.read_stdout(src=async_result.stdout, log=logger, verbose=verbose)
+            result.read_stdout(src=async_result.stdout, log=self.logger, verbose=verbose)
             async_result.interface.wait()  # wait for the end of execution
 
         @threaded.threadpooled
         def poll_stderr():  # type: () -> None
             """Sync stderr poll."""
-            result.read_stderr(src=async_result.stderr, log=logger, verbose=verbose)
+            result.read_stderr(src=async_result.stderr, log=self.logger, verbose=verbose)
 
         def close_streams():  # type: () -> None
             """Enforce FIFO closure."""
@@ -142,24 +142,34 @@ def close_streams():  # type: () -> None
         # Kill not ended process and wait for close
         try:
             # kill -9 for all subprocesses
-            _subprocess_helpers.kill_proc_tree(async_result.interface.pid, including_parent=False)
+            _subprocess_helpers.kill_proc_tree(async_result.interface.pid)
             async_result.interface.kill()  # kill -9
             # Force stop cycle if no exit code after kill
         except OSError:
             exit_code = async_result.interface.poll()
             if exit_code is not None:  # Nothing to kill
-                logger.warning("{!s} has been completed just after timeout: please validate timeout.".format(command))
+                self.logger.warning(
+                    "{!s} has been completed just after timeout: please validate timeout.".format(command))
                 concurrent.futures.wait([stdout_future, stderr_future], timeout=0.1)
                 result.exit_code = exit_code
                 return result
             raise  # Some other error
         finally:
             stdout_future.cancel()
             stderr_future.cancel()
+            _, not_done = concurrent.futures.wait([stdout_future, stderr_future], timeout=5)
+            if not_done:
+                exit_code = async_result.interface.poll()
+                if exit_code:
+                    self.logger.critical(
+                        "Process {!s} was closed with exit code {!s}, but FIFO buffers are still open".format(
+                            command, exit_code
+                        )
+                    )
             close_streams()
 
         wait_err_msg = _log_templates.CMD_WAIT_ERROR.format(result=result, timeout=timeout)
-        logger.debug(wait_err_msg)
+        self.logger.debug(wait_err_msg)
         raise exceptions.ExecHelperTimeoutError(result=result, timeout=timeout)
 
     def execute_async(

test/test_subprocess_special.py

@@ -78,11 +78,13 @@ def read_stream(stream):  # type: (FakeFileStream) -> typing.Tuple[bytes]
     "positive_verbose": dict(stdout=(b" \n", b"2\n", b"3\n", b" \n"), verbose=True),
     "no_stdout": dict(),
     "IOError_on_stdout_read": dict(stdout=(b" \n", b"2\n", IOError())),
-    "TimeoutError": dict(wait=(None,), poll=(None,), stdout=(), expect_exc=exec_helpers.ExecHelperTimeoutError),
+    "TimeoutError": dict(
+        wait=(None,), poll=(None, None, None,), stdout=(), expect_exc=exec_helpers.ExecHelperTimeoutError),
     "TimeoutError_closed": dict(
         wait=(None,), poll=(None, 0), stdout=(b" \n", b"2\n", b"3\n", b" \n"), kill=(OSError(),)
     ),
-    "TimeoutError_no_kill": dict(wait=(None,), poll=(None, None), stdout=(), kill=(OSError(),), expect_exc=OSError),
+    "TimeoutError_no_kill": dict(
+        wait=(None,), poll=(None, None, None,), stdout=(), kill=(OSError(),), expect_exc=OSError),
     "stdin_closed_PIPE_windows": dict(stdout=(b" \n", b"2\n", b"3\n", b" \n"), stdin="Warning", write=einval_exc),
     "stdin_broken_PIPE": dict(stdout=(b" \n", b"2\n", b"3\n", b" \n"), stdin="Warning", write=epipe_exc),
     "stdin_closed_PIPE": dict(stdout=(b" \n", b"2\n", b"3\n", b" \n"), stdin="Warning", write=eshutdown_exc),
@@ -152,6 +154,7 @@ def exec_result(run_parameters):
 @pytest.fixture
 def create_subprocess_shell(mocker, run_parameters):
     mocker.patch("psutil.Process")
+    mocker.patch("psutil.wait_procs", return_value=([], []))
 
     def create_mock(
         stdout=None,  # type: typing.Optional[typing.Tuple]