Merge pull request #3556 from vkarak/bugfix/proc-resource-leaks

[bugfix] Fix resource leaks in `osext.run_command()` and in `_ProcFuture`

Author: vkarak

reframe/utility/osext.py

@@ -48,6 +48,8 @@ class _ProcFuture:
 
     def __init__(self, check=False, *args, **kwargs):
         self._proc = None
+        self._stdout = None
+        self._stderr = None
         self._exitcode = None
         self._signal = None
         self._check = check
@@ -77,6 +79,17 @@ def start(self):
 
         return self
 
+    def _shutdown(self):
+        '''Shut down cleanly the underlying `Popen` object.'''
+        if self._stdout is None and self._proc.stdout:
+            self._stdout = self._proc.stdout.read()
+
+        if self._stderr is None and self._proc.stderr:
+            self._stderr = self._proc.stderr.read()
+
+        self._proc.wait()
+        self._proc.communicate()
+
     @property
     def pid(self):
         '''The PID of the spawned process.'''
@@ -179,6 +192,7 @@ def _wait(self, *, nohang=False):
         except OSError as e:
             if e.errno == errno.ECHILD:
                 self._completed = True
+                self._shutdown()
                 return self._completed
             else:
                 raise e
@@ -191,16 +205,19 @@ def _wait(self, *, nohang=False):
         elif os.WIFSIGNALED(status):
             self._signal = os.WTERMSIG(status)
 
+        # Retrieve the stdout/stderr and clean up the underlying Popen object
         self._completed = True
+        self._shutdown()
 
         # Call any done callbacks
         for func in self._done_callbacks:
             func(self)
 
         # Start the next futures in the chain
-        for fut, cond in self._next:
-            if cond(self):
-                fut.start()
+        if not self._cancelled:
+            for fut, cond in self._next:
+                if cond(self):
+                    fut.start()
 
         return self._completed
 
@@ -230,29 +247,28 @@ def exception(self):
         if not self._check:
             return
 
-        if self._proc.returncode == 0:
+        if self._exitcode == 0:
             return
 
         return SpawnedProcessError(self._proc.args,
-                                   self._proc.stdout.read(),
-                                   self._proc.stderr.read(),
-                                   self._proc.returncode)
+                                   self._stdout, self._stderr,
+                                   self._exitcode)
 
     def stdout(self):
         '''Retrieve the standard output of the spawned process.
 
         This is a blocking call and will wait until the future finishes.
         '''
         self._wait()
-        return self._proc.stdout
+        return self._stdout
 
     def stderr(self):
         '''Retrieve the standard error of the spawned process.
 
         This is a blocking call and will wait until the future finishes.
         '''
         self._wait()
-        return self._proc.stderr
+        return self._stderr
 
 
 def run_command(cmd, check=False, timeout=None, **kwargs):
@@ -281,9 +297,12 @@ def run_command(cmd, check=False, timeout=None, **kwargs):
         proc_stdout, proc_stderr = proc.communicate(timeout=timeout)
     except subprocess.TimeoutExpired as e:
         os.killpg(proc.pid, signal.SIGKILL)
-        raise SpawnedProcessTimeout(e.cmd,
-                                    proc.stdout.read(),
-                                    proc.stderr.read(), timeout) from None
+        os.waitpid(proc.pid, 0)
+        proc_stdout = proc.stdout.read()
+        proc_stderr = proc.stderr.read()
+        proc.communicate()
+        raise SpawnedProcessTimeout(e.cmd, proc_stdout,
+                                    proc_stderr, timeout) from None
 
     completed = subprocess.CompletedProcess(cmd,
                                             returncode=proc.returncode,

unittests/test_shell.py

@@ -130,7 +130,7 @@ def test_trap_signal(script_file):
 
     # We kill the whole spawned process group (we are launching a shell)
     os.killpg(proc.pid, 15)
-    proc.wait()
+    proc.communicate()
 
     f_stdout.flush()
     f_stdout.seek(0)

unittests/test_utility.py

@@ -78,6 +78,7 @@ def test_command_async():
     t_launch = time.time() - t_launch
 
     proc.wait()
+    proc.communicate()
     t_sleep = time.time() - t_sleep
 
     # Now check the timings
@@ -110,7 +111,7 @@ def test_command_futures():
     assert not proc.is_session()
 
     # stdout must block
-    assert proc.stdout().read() == 'hello\n'
+    assert proc.stdout() == 'hello\n'
     assert proc.exitcode == 0
     assert proc.signal is None
 
@@ -270,6 +271,7 @@ def test_command_futures_chain_cancel():
 
     assert proc1.started()
     proc1.cancel()
+    proc1.wait()
     assert proc1.cancelled()
     assert not proc2.started()