terminate with SIGTERM

now that engines handle SIGINT

Author: minrk

ipyparallel/cluster/cluster.py

@@ -307,12 +307,13 @@ async def stop_engines(self, engine_set_id=None):
         """Stop an engine set
 
         If engine_set_id is not given,
-        all engines are stopped"""
+        all engines are stopped.
+        """
         if engine_set_id is None:
             for engine_set_id in list(self._engine_sets):
                 await self.stop_engines(engine_set_id)
             return
-
+        self.log.info(f"Stopping engine(s): {engine_set_id}")
         engine_set = self._engine_sets[engine_set_id]
         r = engine_set.stop()
         if inspect.isawaitable(r):
@@ -352,12 +353,14 @@ async def signal_engine(self, engine_id, signum):
 
     async def signal_engines(self, engine_set_id, signum):
         """Signal all engines in a set"""
+        self.log.info(f"Sending signal {signum} to engine(s) {engine_set_id}")
         engine_set = self._engine_sets[engine_set_id]
         engine_set.signal(signum)
 
     async def stop_controller(self):
         """Stop the controller"""
         if self._controller and self._controller.running:
+            self.log.info("Stopping controller")
             r = self._controller.stop()
             if inspect.isawaitable(r):
                 await r

ipyparallel/cluster/launcher.py

@@ -17,8 +17,9 @@
 try:
     from signal import SIGKILL
 except ImportError:
-    # Windows
-    SIGKILL = SIGTERM
+    # Windows, just need a singleton.
+    # value is not relevant.
+    SIGKILL = -1
 
 try:
     # Windows >= 2.7, 3.2
@@ -317,14 +318,17 @@ def signal(self, sig):
         if self.state == 'running':
             if WINDOWS and sig != SIGINT:
                 # use Windows tree-kill for better child cleanup
-                check_output(['taskkill', '-pid', str(self.process.pid), '-t', '-f'])
+                cmd = ['taskkill', '-pid', str(self.process.pid), '-t']
+                if sig == SIGKILL:
+                    cmd.append("-f")
+                check_output(cmd)
             else:
                 self.process.send_signal(sig)
 
     def interrupt_then_kill(self, delay=2.0):
-        """Send INT, wait a delay and then send KILL."""
+        """Send TERM, wait a delay and then send KILL."""
         try:
-            self.signal(SIGINT)
+            self.signal(SIGTERM)
         except Exception:
             self.log.debug("interrupt failed")
             pass

ipyparallel/tests/test_cluster.py

@@ -1,27 +1,39 @@
 import asyncio
+import logging
 import shutil
 import signal
+import sys
 import time
 
 import pytest
+from traitlets.config import Config
 
 from .clienttest import raises_remote
 from ipyparallel import cluster
-from ipyparallel.cluster import launcher
+from ipyparallel.cluster.launcher import find_launcher_class
 
 
 @pytest.fixture
 def Cluster(request):
     """Fixture for instantiating Clusters"""
 
     def ClusterConstructor(**kwargs):
-        kwargs.setdefault('log_level', 10)
-        launcher_prefix = kwargs.pop("launcher_prefix")
-
-        if launcher_prefix == "MPI" and shutil.which("mpiexec") is None:
+        log = logging.getLogger(__file__)
+        log.setLevel(logging.DEBUG)
+        log.handlers = [logging.StreamHandler(sys.stdout)]
+        kwargs['log'] = log
+        engine_launcher_class = kwargs.get("engine_launcher_class")
+
+        if (
+            isinstance(engine_launcher_class, str)
+            and "MPI" in engine_launcher_class
+            and shutil.which("mpiexec") is None
+        ):
             pytest.skip("requires mpiexec")
-        launcher_class = launcher.find_launcher_class(launcher_prefix, "EngineSet")
-        kwargs['engine_launcher_class'] = launcher_class
+
+        cfg = kwargs.setdefault("config", Config())
+        cfg.EngineMixin.engine_args = ['--log-level=10']
+        cfg.ControllerMixin.controller_args = ['--log-level=10']
 
         c = cluster.Cluster(**kwargs)
         request.addfinalizer(c.stop_cluster_sync)
@@ -59,14 +71,14 @@ async def test_start_stop_controller(Cluster):
     # TODO: test file cleanup
 
 
-@pytest.mark.parametrize("launcher_prefix", ["Local", "MPI"])
-async def test_start_stop_engines(Cluster, launcher_prefix):
-    cluster = Cluster(launcher_prefix=launcher_prefix)
+@pytest.mark.parametrize("engine_launcher_class", ["Local", "MPI"])
+async def test_start_stop_engines(Cluster, engine_launcher_class):
+    cluster = Cluster(engine_launcher_class=engine_launcher_class)
     await cluster.start_controller()
     engine_set_id = await cluster.start_engines(n=3)
     assert engine_set_id in cluster._engine_sets
     engine_set = cluster._engine_sets[engine_set_id]
-    launcher_class = launcher_class = find_launcher_class(launcher_prefix, "EngineSet")
+    launcher_class = find_launcher_class(engine_launcher_class, "EngineSet")
     assert isinstance(engine_set, launcher_class)
     await cluster.stop_engines(engine_set_id)
     assert cluster._engine_sets == {}
@@ -76,20 +88,28 @@ async def test_start_stop_engines(Cluster, launcher_prefix):
     await cluster.stop_controller()
 
 
-@pytest.mark.parametrize("launcher_prefix", ["Local", "MPI"])
-async def test_signal_engines(Cluster, launcher_prefix):
-    cluster = Cluster(launcher_prefix=launcher_prefix)
+@pytest.mark.parametrize("engine_launcher_class", ["Local", "MPI"])
+async def test_signal_engines(Cluster, engine_launcher_class):
+    cluster = Cluster(engine_launcher_class=engine_launcher_class)
     await cluster.start_controller()
     engine_set_id = await cluster.start_engines(n=3)
     rc = cluster.connect_client()
     while len(rc) < 3:
         await asyncio.sleep(0.1)
+    # seems to be a problem if we start too soon...
+    await asyncio.sleep(1)
+    # ensure responsive
+    rc[:].apply_async(lambda: None).get(timeout=10)
+    # submit request to be interrupted
     ar = rc[:].apply_async(time.sleep, 3)
+    # wait for it to be running
     await asyncio.sleep(0.5)
+    # send signal
     await cluster.signal_engines(engine_set_id, signal.SIGINT)
 
+    # wait for result, which should raise KeyboardInterrupt
     with raises_remote(KeyboardInterrupt) as e:
-        ar.get()
+        ar.get(timeout=10)
 
     await cluster.stop_engines()
     await cluster.stop_controller()