Force to close FIFO buffers instead of waiting for garbage collector

Signed-off-by: Alexey Stepanov <penguinolog@gmail.com>

exec_helpers/subprocess_runner.py

@@ -171,6 +171,13 @@ def poll_stderr():  # type: () -> None
             """Sync stderr poll."""
             result.read_stderr(src=stderr, log=logger, verbose=verbose)
 
+        def close_streams():  # type: () -> None
+            """Enforce FIFO closure."""
+            if stdout is not None and not stdout.closed:
+                stdout.close()
+            if stderr is not None and not stderr.closed:
+                stderr.close()
+
         # Store command with hidden data
         cmd_for_log = self._mask_command(cmd=command, log_mask_re=log_mask_re)
 
@@ -189,6 +196,7 @@ def poll_stderr():  # type: () -> None
         # Process closed?
         if exit_code is not None:
             result.exit_code = exit_code
+            close_streams()
             return result
         # Kill not ended process and wait for close
         try:
@@ -205,6 +213,8 @@ def poll_stderr():  # type: () -> None
                 result.exit_code = exit_code
                 return result
             raise  # Some other error
+        finally:
+            close_streams()
 
         wait_err_msg = _log_templates.CMD_WAIT_ERROR.format(result=result, timeout=timeout)
         logger.debug(wait_err_msg)

test/test_subprocess_runner.py

@@ -42,6 +42,7 @@
 class FakeFileStream(object):
     def __init__(self, *args):
         self.__src = list(args)
+        self.closed = False
 
     def __iter__(self):
         for _ in range(len(self.__src)):
@@ -50,6 +51,9 @@ def __iter__(self):
     def fileno(self):
         return hash(tuple(self.__src))
 
+    def close(self):
+        self.closed = True
+
 
 # Subprocess extra arguments.
 # Flags from: