Merge pull request #842 from wtsi-hgi/fix-832

Kill child processes when cwltool exits

Author: mr-c

cwltool.py

@@ -11,4 +11,4 @@
 from cwltool import main
 
 if __name__ == "__main__":
-    sys.exit(main.main(sys.argv[1:]))
+    main.run(sys.argv[1:])

cwltool/__main__.py

@@ -4,4 +4,4 @@
 
 from . import main
 
-sys.exit(main.main())
+main.run()

cwltool/job.py

@@ -31,7 +31,7 @@
 from .utils import bytes2str_in_dicts  # pylint: disable=unused-import
 from .utils import (  # pylint: disable=unused-import
     DEFAULT_TMP_PREFIX, Directory, copytree_with_merge, json_dump, json_dumps,
-    onWindows, subprocess)
+    onWindows, subprocess, processes_to_kill)
 from .context import (RuntimeContext,  # pylint: disable=unused-import
                       getdefault)
 if TYPE_CHECKING:
@@ -521,6 +521,7 @@ def _job_popen(
                                  stderr=stderr,
                                  env=env,
                                  cwd=cwd)
+        processes_to_kill.append(sproc)
 
         if sproc.stdin:
             sproc.stdin.close()
@@ -534,6 +535,7 @@ def terminate():
                 except OSError:
                     pass
             tm = Timer(timelimit, terminate)
+            tm.daemon = True
             tm.start()
 
         rcode = sproc.wait()
@@ -588,6 +590,7 @@ def terminate():
                 stderr=sys.stderr,
                 stdin=subprocess.PIPE,
             )
+            processes_to_kill.append(sproc)
             if sproc.stdin:
                 sproc.stdin.close()
 

cwltool/main.py

@@ -10,6 +10,7 @@
 import io
 import logging
 import os
+import signal
 import sys
 
 from typing import (IO, Any, Callable, Dict,  # pylint: disable=unused-import
@@ -47,10 +48,43 @@
 from .stdfsaccess import StdFsAccess
 from .update import ALLUPDATES, UPDATES
 from .utils import (DEFAULT_TMP_PREFIX, add_sizes, json_dumps, onWindows,
-                    versionstring, windows_default_container_id)
+                    versionstring, windows_default_container_id,
+                    processes_to_kill)
 from .context import LoadingContext, RuntimeContext, getdefault
 from .builder import HasReqsHints
 
+
+def _terminate_processes():
+    # type: () -> None
+    """Kill all spawned processes.
+
+    Processes to be killed must be appended to `utils.processes_to_kill`
+    as they are spawned.
+
+    An important caveat: since there's no supported way to kill another
+    thread in Python, this function cannot stop other threads from
+    continuing to execute while it kills the processes that they've
+    spawned. This may occasionally lead to unexpected behaviour.
+    """
+    # It's possible that another thread will spawn a new task while
+    # we're executing, so it's not safe to use a for loop here.
+    while processes_to_kill:
+        processes_to_kill.popleft().kill()
+
+
+def _signal_handler(signum, frame):
+    # type: (int, Any) -> None
+    """Kill all spawned processes and exit.
+
+    Note that it's possible for another thread to spawn a process after
+    all processes have been killed, but before Python exits.
+
+    Refer to the docstring for _terminate_processes() for other caveats.
+    """
+    _terminate_processes()
+    sys.exit(signum)
+
+
 def generate_example_input(inptype):
     # type: (Union[Text, Dict[Text, Any]]) -> Any
     defaults = {u'null': 'null',
@@ -680,5 +714,15 @@ def find_default_container(builder,                  # type: HasReqsHints
     return default_container
 
 
+def run(*args, **kwargs):
+    # type: (...) -> None
+    """Run cwltool."""
+    signal.signal(signal.SIGTERM, _signal_handler)
+    try:
+        sys.exit(main(*args, **kwargs))
+    finally:
+        _terminate_processes()
+
+
 if __name__ == "__main__":
-    sys.exit(main(sys.argv[1:]))
+    run(sys.argv[1:])

cwltool/sandboxjs.py

@@ -14,7 +14,7 @@
 import six
 from pkg_resources import resource_stream
 
-from .utils import json_dumps, onWindows, subprocess
+from .utils import json_dumps, onWindows, subprocess, processes_to_kill
 
 try:
     import queue  # type: ignore
@@ -71,7 +71,7 @@ def new_js_proc(js_text, force_docker_pull=False):
                                           stdin=subprocess.PIPE,
                                           stdout=subprocess.PIPE,
                                           stderr=subprocess.PIPE)
-
+                processes_to_kill.append(nodejs)
                 required_node_version = check_js_threshold_version(n)
                 break
         except (subprocess.CalledProcessError, OSError):
@@ -95,6 +95,7 @@ def new_js_proc(js_text, force_docker_pull=False):
                                        "--sig-proxy=true", "--interactive",
                                        "--rm", nodeimg, "node", "--eval", js_text],
                                       stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+            processes_to_kill.append(nodejs)
             docker = True
         except OSError as e:
             if e.errno == errno.ENOENT:
@@ -181,6 +182,7 @@ def terminate():
         timeout = 20
 
     tm = threading.Timer(timeout, terminate)
+    tm.daemon = True
     tm.start()
 
     stdin_text = u""

cwltool/utils.py

@@ -1,5 +1,6 @@
 from __future__ import absolute_import
 
+import collections
 import json
 import os
 import sys
@@ -9,8 +10,8 @@
 import pkg_resources
 from functools import partial  # pylint: disable=unused-import
 from typing import (IO, Any, AnyStr, Callable,  # pylint: disable=unused-import
-                    Dict, Iterable, List, Optional, Text, Tuple, TypeVar,
-                    Union)
+                    Dict, Iterable, List, Optional, Text, Tuple, TypeVar, Union)
+from typing_extensions import Deque
 
 import six
 from six.moves import urllib, zip_longest
@@ -32,6 +33,8 @@
 
 DEFAULT_TMP_PREFIX = "tmp"
 
+processes_to_kill = collections.deque()  # type: Deque[subprocess.Popen]
+
 def versionstring():
     # type: () -> Text
     '''

requirements.txt

@@ -10,3 +10,4 @@ bagit==1.6.4
 mypy-extensions
 psutil
 subprocess32 >= 3.5.0; os.name=="posix"
+typing-extensions

setup.py

@@ -60,6 +60,7 @@
           'psutil',
           'prov == 1.5.1',
           'bagit >= 1.6.4',
+          'typing-extensions',
       ],
       extras_require={
           ':os.name=="posix"': ['subprocess32 >= 3.5.0'],
@@ -72,7 +73,7 @@
       test_suite='tests',
       tests_require=['pytest', 'mock >= 2.0.0', 'arcp >= 0.2.0', 'rdflib-jsonld >= 0.4.0'],
       entry_points={
-          'console_scripts': ["cwltool=cwltool.main:main"]
+          'console_scripts': ["cwltool=cwltool.main:run"]
       },
       zip_safe=True,
       cmdclass={'egg_info': tagger},