Merge pull request #4066 from Flamefire/fix-leaked-handle

fix leaked handles in set_columns, complete_cmd, run_cmd_qa, det_terminal_size functions + tests

Author: boegel

easybuild/base/generaloption.py

@@ -43,7 +43,7 @@
 
 from easybuild.base.fancylogger import getLogger, setroot, setLogLevel, getDetailsLogLevels
 from easybuild.base.optcomplete import autocomplete, CompleterOption
-from easybuild.tools.py2vs3 import StringIO, configparser, string_type
+from easybuild.tools.py2vs3 import StringIO, configparser, string_type, subprocess_popen_text
 from easybuild.tools.utilities import mk_rst_table, nub, shell_quote
 
 try:
@@ -80,7 +80,9 @@ def set_columns(cols=None):
         stty = '/usr/bin/stty'
         if os.path.exists(stty):
             try:
-                cols = int(os.popen('%s size 2>/dev/null' % stty).read().strip().split(' ')[1])
+                with open(os.devnull, 'w') as devnull:
+                    proc = subprocess_popen_text([stty, "size"], stderr=devnull)
+                    cols = int(proc.communicate()[0].strip().split(' ')[1])
             except (AttributeError, IndexError, OSError, ValueError):
                 # do nothing
                 pass

easybuild/tools/py2vs3/py2.py

@@ -33,6 +33,7 @@
 import ConfigParser as configparser  # noqa
 import json
 import subprocess
+import time
 import urllib2 as std_urllib  # noqa
 from collections import Mapping, OrderedDict  # noqa
 from HTMLParser import HTMLParser  # noqa
@@ -55,7 +56,25 @@
 
 def subprocess_popen_text(cmd, **kwargs):
     """Call subprocess.Popen with specified named arguments."""
-    return subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, **kwargs)
+    kwargs.setdefault('stderr', subprocess.PIPE)
+    return subprocess.Popen(cmd, stdout=subprocess.PIPE, **kwargs)
+
+
+def subprocess_terminate(proc, timeout):
+    """Terminate the subprocess if it hasn't finished after the given timeout"""
+    res = None
+    for pipe in (proc.stdout, proc.stderr, proc.stdin):
+        if pipe:
+            pipe.close()
+    while timeout > 0:
+        res = proc.poll()
+        if res is not None:
+            break
+        delay = min(timeout, 0.1)
+        time.sleep(delay)
+        timeout -= delay
+    if res is None:
+        proc.terminate()
 
 
 def raise_with_traceback(exception_class, message, traceback):

easybuild/tools/py2vs3/py3.py

@@ -69,7 +69,19 @@ def json_loads(body):
 def subprocess_popen_text(cmd, **kwargs):
     """Call subprocess.Popen in text mode with specified named arguments."""
     # open stdout/stderr in text mode in Popen when using Python 3
-    return subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True, **kwargs)
+    kwargs.setdefault('stderr', subprocess.PIPE)
+    return subprocess.Popen(cmd, stdout=subprocess.PIPE, universal_newlines=True, **kwargs)
+
+
+def subprocess_terminate(proc, timeout):
+    """Terminate the subprocess if it hasn't finished after the given timeout"""
+    try:
+        proc.communicate(timeout=timeout)
+    except subprocess.TimeoutExpired:
+        for pipe in (proc.stdout, proc.stderr, proc.stdin):
+            if pipe:
+                pipe.close()
+        proc.terminate()
 
 
 def raise_with_traceback(exception_class, message, traceback):

easybuild/tools/run.py

@@ -33,6 +33,7 @@
 :author: Toon Willems (Ghent University)
 :author: Ward Poelmans (Ghent University)
 """
+import contextlib
 import functools
 import os
 import re
@@ -315,21 +316,24 @@ def complete_cmd(proc, cmd, owd, start_time, cmd_log, log_ok=True, log_all=False
 
     stdouterr = output
 
-    ec = proc.poll()
-    while ec is None:
-        # need to read from time to time.
-        # - otherwise the stdout/stderr buffer gets filled and it all stops working
-        output = get_output_from_process(proc, read_size=read_size)
-        if cmd_log:
-            cmd_log.write(output)
-        if stream_output:
-            sys.stdout.write(output)
-        stdouterr += output
+    try:
         ec = proc.poll()
+        while ec is None:
+            # need to read from time to time.
+            # - otherwise the stdout/stderr buffer gets filled and it all stops working
+            output = get_output_from_process(proc, read_size=read_size)
+            if cmd_log:
+                cmd_log.write(output)
+            if stream_output:
+                sys.stdout.write(output)
+            stdouterr += output
+            ec = proc.poll()
+
+        # read remaining data (all of it)
+        output = get_output_from_process(proc)
+    finally:
+        proc.stdout.close()
 
-    # read remaining data (all of it)
-    output = get_output_from_process(proc)
-    proc.stdout.close()
     if cmd_log:
         cmd_log.write(output)
         cmd_log.close()
@@ -397,6 +401,8 @@ def run_cmd_qa(cmd, qa, no_qa=None, log_ok=True, log_all=False, simple=False, re
             path = cwd
         dry_run_msg("  running interactive command \"%s\"" % cmd, silent=build_option('silent'))
         dry_run_msg("  (in %s)" % path, silent=build_option('silent'))
+        if cmd_log:
+            cmd_log.close()
         if simple:
             return True
         else:
@@ -442,6 +448,8 @@ def check_answers_list(answers):
         if isinstance(answers, string_type):
             answers = [answers]
         elif not isinstance(answers, list):
+            if cmd_log:
+                cmd_log.close()
             raise EasyBuildError("Invalid type for answer on %s, no string or list: %s (%s)",
                                  question, type(answers), answers)
         # list is manipulated when answering matching question, so return a copy
@@ -479,47 +487,48 @@ def check_answers_list(answers):
     if cmd_log:
         cmd_log.write("# output for interactive command: %s\n\n" % cmd)
 
-    try:
-        proc = asyncprocess.Popen(cmd, shell=True, stdout=asyncprocess.PIPE, stderr=asyncprocess.STDOUT,
-                                  stdin=asyncprocess.PIPE, close_fds=True, executable='/bin/bash')
-    except OSError as err:
-        raise EasyBuildError("run_cmd_qa init cmd %s failed:%s", cmd, err)
-
-    ec = proc.poll()
-    stdout_err = ''
-    old_len_out = -1
-    hit_count = 0
-
-    while ec is None:
-        # need to read from time to time.
-        # - otherwise the stdout/stderr buffer gets filled and it all stops working
+    # Make sure we close the proc handles and the cmd_log file
+    @contextlib.contextmanager
+    def get_proc():
         try:
-            out = get_output_from_process(proc, asynchronous=True)
-
+            proc = asyncprocess.Popen(cmd, shell=True, stdout=asyncprocess.PIPE, stderr=asyncprocess.STDOUT,
+                                      stdin=asyncprocess.PIPE, close_fds=True, executable='/bin/bash')
+        except OSError as err:
             if cmd_log:
-                cmd_log.write(out)
-            stdout_err += out
-        # recv_some used by get_output_from_process for getting asynchronous output may throw exception
-        except (IOError, Exception) as err:
-            _log.debug("run_cmd_qa cmd %s: read failed: %s", cmd, err)
-            out = None
-
-        hit = False
-        for question, answers in new_qa.items():
-            res = question.search(stdout_err)
-            if out and res:
-                fa = answers[0] % res.groupdict()
-                # cycle through list of answers
-                last_answer = answers.pop(0)
-                answers.append(last_answer)
-                _log.debug("List of answers for question %s after cycling: %s", question.pattern, answers)
-
-                _log.debug("run_cmd_qa answer %s question %s out %s", fa, question.pattern, stdout_err[-50:])
-                asyncprocess.send_all(proc, fa)
-                hit = True
-                break
-        if not hit:
-            for question, answers in new_std_qa.items():
+                cmd_log.close()
+            raise EasyBuildError("run_cmd_qa init cmd %s failed:%s", cmd, err)
+        try:
+            yield proc
+        finally:
+            if proc.stdout:
+                proc.stdout.close()
+            if proc.stdin:
+                proc.stdin.close()
+            if cmd_log:
+                cmd_log.close()
+
+    with get_proc() as proc:
+        ec = proc.poll()
+        stdout_err = ''
+        old_len_out = -1
+        hit_count = 0
+
+        while ec is None:
+            # need to read from time to time.
+            # - otherwise the stdout/stderr buffer gets filled and it all stops working
+            try:
+                out = get_output_from_process(proc, asynchronous=True)
+
+                if cmd_log:
+                    cmd_log.write(out)
+                stdout_err += out
+            # recv_some used by get_output_from_process for getting asynchronous output may throw exception
+            except (IOError, Exception) as err:
+                _log.debug("run_cmd_qa cmd %s: read failed: %s", cmd, err)
+                out = None
+
+            hit = False
+            for question, answers in new_qa.items():
                 res = question.search(stdout_err)
                 if out and res:
                     fa = answers[0] % res.groupdict()
@@ -528,51 +537,65 @@ def check_answers_list(answers):
                     answers.append(last_answer)
                     _log.debug("List of answers for question %s after cycling: %s", question.pattern, answers)
 
-                    _log.debug("run_cmd_qa answer %s std question %s out %s", fa, question.pattern, stdout_err[-50:])
+                    _log.debug("run_cmd_qa answer %s question %s out %s", fa, question.pattern, stdout_err[-50:])
                     asyncprocess.send_all(proc, fa)
                     hit = True
                     break
             if not hit:
-                if len(stdout_err) > old_len_out:
-                    old_len_out = len(stdout_err)
+                for question, answers in new_std_qa.items():
+                    res = question.search(stdout_err)
+                    if out and res:
+                        fa = answers[0] % res.groupdict()
+                        # cycle through list of answers
+                        last_answer = answers.pop(0)
+                        answers.append(last_answer)
+                        _log.debug("List of answers for question %s after cycling: %s", question.pattern, answers)
+
+                        _log.debug("run_cmd_qa answer %s std question %s out %s",
+                                   fa, question.pattern, stdout_err[-50:])
+                        asyncprocess.send_all(proc, fa)
+                        hit = True
+                        break
+                if not hit:
+                    if len(stdout_err) > old_len_out:
+                        old_len_out = len(stdout_err)
+                    else:
+                        noqa = False
+                        for r in new_no_qa:
+                            if r.search(stdout_err):
+                                _log.debug("runqanda: noQandA found for out %s", stdout_err[-50:])
+                                noqa = True
+                        if not noqa:
+                            hit_count += 1
                 else:
-                    noqa = False
-                    for r in new_no_qa:
-                        if r.search(stdout_err):
-                            _log.debug("runqanda: noQandA found for out %s", stdout_err[-50:])
-                            noqa = True
-                    if not noqa:
-                        hit_count += 1
+                    hit_count = 0
             else:
                 hit_count = 0
-        else:
-            hit_count = 0
 
-        if hit_count > maxhits:
-            # explicitly kill the child process before exiting
-            try:
-                os.killpg(proc.pid, signal.SIGKILL)
-                os.kill(proc.pid, signal.SIGKILL)
-            except OSError as err:
-                _log.debug("run_cmd_qa exception caught when killing child process: %s", err)
-            _log.debug("run_cmd_qa: full stdouterr: %s", stdout_err)
-            raise EasyBuildError("run_cmd_qa: cmd %s : Max nohits %s reached: end of output %s",
-                                 cmd, maxhits, stdout_err[-500:])
-
-        # the sleep below is required to avoid exiting on unknown 'questions' too early (see above)
-        time.sleep(1)
-        ec = proc.poll()
-
-    # Process stopped. Read all remaining data
-    try:
-        if proc.stdout:
-            out = get_output_from_process(proc)
-            stdout_err += out
-            if cmd_log:
-                cmd_log.write(out)
-                cmd_log.close()
-    except IOError as err:
-        _log.debug("runqanda cmd %s: remaining data read failed: %s", cmd, err)
+            if hit_count > maxhits:
+                # explicitly kill the child process before exiting
+                try:
+                    os.killpg(proc.pid, signal.SIGKILL)
+                    os.kill(proc.pid, signal.SIGKILL)
+                except OSError as err:
+                    _log.debug("run_cmd_qa exception caught when killing child process: %s", err)
+                _log.debug("run_cmd_qa: full stdouterr: %s", stdout_err)
+                raise EasyBuildError("run_cmd_qa: cmd %s : Max nohits %s reached: end of output %s",
+                                     cmd, maxhits, stdout_err[-500:])
+
+            # the sleep below is required to avoid exiting on unknown 'questions' too early (see above)
+            time.sleep(1)
+            ec = proc.poll()
+
+        # Process stopped. Read all remaining data
+        try:
+            if proc.stdout:
+                out = get_output_from_process(proc)
+                stdout_err += out
+                if cmd_log:
+                    cmd_log.write(out)
+        except IOError as err:
+            _log.debug("runqanda cmd %s: remaining data read failed: %s", cmd, err)
 
     if trace:
         trace_msg("interactive command completed: exit %s, ran in %s" % (ec, time_str_since(start_time)))

easybuild/tools/systemtools.py

@@ -41,6 +41,7 @@
 import termios
 from ctypes.util import find_library
 from socket import gethostname
+from easybuild.tools.py2vs3 import subprocess_popen_text
 
 # pkg_resources is provided by the setuptools Python package,
 # which we really want to keep as an *optional* dependency
@@ -1201,7 +1202,7 @@ def det_terminal_size():
     except Exception as err:
         _log.warning("First attempt to determine terminal size failed: %s", err)
         try:
-            height, width = [int(x) for x in os.popen("stty size").read().strip().split()]
+            height, width = [int(x) for x in subprocess_popen_text("stty size").communicate()[0].strip().split()]
         except Exception as err:
             _log.warning("Second attempt to determine terminal size failed, going to return defaults: %s", err)
             height, width = 25, 80

test/framework/asyncprocess.py

@@ -35,6 +35,7 @@
 
 import easybuild.tools.asyncprocess as p
 from easybuild.tools.asyncprocess import Popen
+from easybuild.tools.py2vs3 import subprocess_terminate
 
 
 class AsyncProcessTest(EnhancedTestCase):
@@ -62,6 +63,7 @@ def runTest(self):
 
     def tearDown(self):
         """cleanup"""
+        subprocess_terminate(self.shell, timeout=1)
         super(AsyncProcessTest, self).tearDown()