Merge pull request #515 from minrk/test-windows

Add Windows test coverage

Author: minrk

.github/workflows/test.yml

@@ -10,8 +10,8 @@ env:
 
 jobs:
   test:
-    runs-on: ubuntu-20.04
-    timeout-minutes: 10
+    runs-on: ${{ matrix.runs_on || 'ubuntu-20.04' }}
+    timeout-minutes: 15
 
     strategy:
       # Keep running even if one variation of the job fail
@@ -22,12 +22,14 @@ jobs:
             ssh: ssh
           - python: "3.6"
             tornado: "5.1.1"
-          - python: "3.6"
           - python: "3.7"
             controller_ip: "*"
+          - python: "3.8"
+            runs_on: windows-2019
           - python: "3.8"
             mpi: mpi
           - python: "3.9"
+            runs_on: macos-10.15
 
     steps:
       - uses: actions/checkout@v2
@@ -74,9 +76,11 @@ jobs:
           pip install --upgrade pip
           pip install --pre --upgrade .[test] distributed joblib codecov
           pip install --only-binary :all: matplotlib || echo "no matplotlib"
-          if [ "${{ matrix.tornado }}" != "" ]; then
-            pip install tornado==${{ matrix.tornado }}
-          fi
+
+      - name: Install pinned tornado
+        if: matrix.tornado
+        run: |
+          pip install tornado==${{ matrix.tornado }}
 
       - name: Show environment
         run: pip freeze

ipyparallel/cluster/launcher.py

@@ -15,8 +15,8 @@
 import sys
 import threading
 import time
+from concurrent.futures import ThreadPoolExecutor
 from functools import partial
-from signal import SIGINT
 from signal import SIGTERM
 from subprocess import check_output
 from subprocess import PIPE
@@ -85,6 +85,12 @@ class UnknownStatus(LauncherError):
 class BaseLauncher(LoggingConfigurable):
     """An abstraction for starting, stopping and signaling a process."""
 
+    stop_timeout = Integer(
+        60,
+        config=True,
+        help="The number of seconds to wait for a process to exit before raising a TimeoutError in stop",
+    )
+
     # In all of the launchers, the work_dir is where child processes will be
     # run. This will usually be the profile_dir, but may not be. any work_dir
     # passed into the __init__ method will override the config value.
@@ -249,6 +255,10 @@ def signal(self, sig):
         """
         raise NotImplementedError('signal must be implemented in a subclass')
 
+    def join(self, timeout=None):
+        """Wait for the process to finish"""
+        raise NotImplementedError('join must be implemented in a subclass')
+
     output_limit = Integer(
         100,
         config=True,
@@ -376,6 +386,12 @@ def _default_output_file(self):
         os.makedirs(log_dir, exist_ok=True)
         return os.path.join(log_dir, f'{self.identifier}.log')
 
+    stop_seconds_until_kill = Integer(
+        5,
+        config=True,
+        help="""The number of seconds to wait for a process to exit after sending SIGTERM before sending SIGKILL""",
+    )
+
     stdout = None
     stderr = None
     process = None
@@ -446,6 +462,18 @@ def start(self):
         if self.log.level <= logging.DEBUG:
             self._start_streaming()
 
+    async def join(self, timeout=None):
+        """Wait for the process to exit"""
+        with ThreadPoolExecutor(1) as pool:
+            try:
+                await asyncio.wrap_future(
+                    pool.submit(partial(self.process.wait, timeout))
+                )
+            except psutil.TimeoutExpired:
+                raise TimeoutError(
+                    f"Process {self.pid} did not complete in {timeout} seconds."
+                )
+
     def _stream_file(self, path):
         """Stream one file"""
         with open(path, 'r') as f:
@@ -460,7 +488,7 @@ def _stream_file(self, path):
                     time.sleep(0.1)
 
     def _start_streaming(self):
-        t = threading.Thread(
+        self._stream_thread = t = threading.Thread(
             target=partial(self._stream_file, self.output_file),
             name=f"Stream Output {self.identifier}",
             daemon=True,
@@ -483,35 +511,46 @@ def get_output(self, remove=False):
 
         if remove and os.path.isfile(self.output_file):
             self.log.debug(f"Removing {self.output_file}")
-            os.remove(self.output_file)
+            try:
+                os.remove(self.output_file)
+            except Exception as e:
+                # don't crash on failure to remove a file,
+                # e.g. due to another processing having it open
+                self.log.error(f"Failed to remove {self.output_file}: {e}")
 
         return self._output
 
-    def stop(self):
-        return self.interrupt_then_kill()
+    async def stop(self):
+        try:
+            self.signal(SIGTERM)
+        except Exception as e:
+            self.log.debug(f"TERM failed: {e!r}")
+
+        try:
+            await self.join(timeout=self.stop_seconds_until_kill)
+        except TimeoutError:
+            self.log.warning(
+                f"Process {self.pid} did not exit in {self.stop_seconds_until_kill} seconds after TERM"
+            )
+        else:
+            return
+
+        try:
+            self.signal(SIGKILL)
+        except Exception as e:
+            self.log.debug(f"KILL failed: {e!r}")
+
+        await self.join(timeout=self.stop_timeout)
 
     def signal(self, sig):
         if self.state == 'running':
-            if WINDOWS and sig != SIGINT:
+            if WINDOWS and sig == SIGKILL:
                 # use Windows tree-kill for better child cleanup
-                cmd = ['taskkill', '-pid', str(self.process.pid), '-t']
-                if sig == SIGKILL:
-                    cmd.append("-f")
+                cmd = ['taskkill', '/pid', str(self.process.pid), '/t', '/F']
                 check_output(cmd)
             else:
                 self.process.send_signal(sig)
 
-    def interrupt_then_kill(self, delay=2.0):
-        """Send TERM, wait a delay and then send KILL."""
-        try:
-            self.signal(SIGTERM)
-        except Exception as e:
-            self.log.debug(f"interrupt failed: {e!r}")
-            pass
-        self.killer = asyncio.get_event_loop().call_later(
-            delay, lambda: self.signal(SIGKILL)
-        )
-
     # callbacks, etc:
 
     def handle_stdout(self, fd, events):
@@ -637,21 +676,18 @@ def find_args(self):
         return ['engine set']
 
     def signal(self, sig):
-        dlist = []
-        for el in itervalues(self.launchers):
-            d = el.signal(sig)
-            dlist.append(d)
-        return dlist
+        for el in list(self.launchers.values()):
+            el.signal(sig)
 
-    def interrupt_then_kill(self, delay=1.0):
-        dlist = []
-        for el in itervalues(self.launchers):
-            d = el.interrupt_then_kill(delay)
-            dlist.append(d)
-        return dlist
+    async def stop(self):
+        futures = []
+        for el in list(self.launchers.values()):
+            f = el.stop()
+            if inspect.isawaitable(f):
+                futures.append(asyncio.ensure_future(f))
 
-    def stop(self):
-        return self.interrupt_then_kill()
+        if futures:
+            await asyncio.gather(*futures)
 
     def _notice_engine_stopped(self, data):
         identifier = data['identifier']
@@ -1146,6 +1182,12 @@ def wait_one(self, timeout):
             raise TimeoutError("still running")
         return int(values.get("exit_code", -1))
 
+    async def join(self, timeout=None):
+        with ThreadPoolExecutor(1) as pool:
+            await asyncio.wrap_future(
+                pool.submit(partial(self.wait_one, timeout=timeout))
+            )
+
     def signal(self, sig):
         if self.state == 'running':
             check_output(
@@ -1306,7 +1348,7 @@ def start(self, n):
         return dlist
 
 
-class SSHProxyEngineSetLauncher(SSHLauncher):
+class SSHProxyEngineSetLauncher(SSHLauncher, EngineLauncher):
     """Launcher for calling
     `ipcluster engines` on a remote machine.
 

ipyparallel/tests/__init__.py

@@ -1,12 +1,10 @@
 """toplevel setup/teardown for parallel tests."""
 from __future__ import print_function
 
+import asyncio
 import os
-import tempfile
 import time
-from subprocess import PIPE
 from subprocess import Popen
-from subprocess import STDOUT
 
 from IPython.paths import get_ipython_dir
 
@@ -125,7 +123,9 @@ def teardown():
         p = launchers.pop()
         if p.poll() is None:
             try:
-                p.stop()
+                f = p.stop()
+                if f:
+                    asyncio.run(f)
             except Exception as e:
                 print(e)
                 pass

ipyparallel/tests/conftest.py

@@ -3,6 +3,7 @@
 import logging
 import os
 import sys
+from contextlib import contextmanager
 from subprocess import check_call
 from subprocess import check_output
 from tempfile import TemporaryDirectory
@@ -21,16 +22,33 @@
 from . import teardown
 
 
-@pytest.fixture(autouse=True, scope="session")
-def ipython_dir():
-    with TemporaryDirectory(suffix="dotipython") as td:
-        with mock.patch.dict(os.environ, {"IPYTHONDIR": td}):
-            assert IPython.paths.get_ipython_dir() == td
-            pd = ProfileDir.create_profile_dir_by_name(td, name="default")
-            # configure fast heartbeats for quicker tests with small numbers of local engines
-            with open(os.path.join(pd.location, "ipcontroller_config.py"), "w") as f:
-                f.write("c.HeartMonitor.period = 200")
+@contextmanager
+def temporary_ipython_dir():
+    # FIXME: cleanup has issues on Windows
+    # this is *probably* a real bug of holding open files,
+    # but it is preventing feedback about test failures
+    td_obj = TemporaryDirectory(suffix=".ipython")
+    td = td_obj.name
+
+    with mock.patch.dict(os.environ, {"IPYTHONDIR": td}):
+        assert IPython.paths.get_ipython_dir() == td
+        pd = ProfileDir.create_profile_dir_by_name(td, name="default")
+        # configure fast heartbeats for quicker tests with small numbers of local engines
+        with open(os.path.join(pd.location, "ipcontroller_config.py"), "w") as f:
+            f.write("c.HeartMonitor.period = 200")
+        try:
             yield td
+        finally:
+            try:
+                td_obj.cleanup()
+            except Exception as e:
+                print(f"Failed to cleanup {td}: {e}", file=sys.stderr)
+
+
+@pytest.fixture(autouse=True, scope="module")
+def ipython_dir(request):
+    with temporary_ipython_dir() as ipython_dir:
+        yield ipython_dir
 
 
 def pytest_collection_modifyitems(items):
@@ -46,19 +64,23 @@ def pytest_collection_modifyitems(items):
         assert not inspect.isasyncgenfunction(item.obj)
 
 
-@pytest.fixture(scope="session")
-def cluster(request):
+@pytest.fixture(scope="module")
+def cluster(request, ipython_dir):
     """Setup IPython parallel cluster"""
     setup()
-    request.addfinalizer(teardown)
+    try:
+        yield
+    finally:
+        teardown()
 
 
-@pytest.fixture(scope='session')
-def ipython():
+@pytest.fixture(scope='module')
+def ipython(ipython_dir):
     config = default_config()
     config.TerminalInteractiveShell.simple_prompt = True
     shell = TerminalInteractiveShell.instance(config=config)
-    return shell
+    yield shell
+    TerminalInteractiveShell.clear_instance()
 
 
 @pytest.fixture()
@@ -81,7 +103,7 @@ def Context():
 
 
 @pytest.fixture
-def Cluster(request, io_loop):
+def Cluster(request, ipython_dir, io_loop):
     """Fixture for instantiating Clusters"""
 
     def ClusterConstructor(**kwargs):

ipyparallel/tests/test_client.py

@@ -659,9 +659,18 @@ def test_wait_for_engines(self):
         self.add_engines(1)
         assert f.result() is None
 
+    @pytest.mark.skipif(
+        sys.platform.startswith("win"), reason="Signal tests don't pass on Windows yet"
+    )
     def test_signal_engines(self):
+
         view = self.client[:]
-        for sig in (signal.SIGINT, 'SIGINT'):
+        if sys.platform.startswith("win"):
+            signame = 'CTRL_C_EVENT'
+        else:
+            signame = 'SIGINT'
+        signum = getattr(signal, signame)
+        for sig in (signum, signame):
             ar = view.apply_async(time.sleep, 10)
             # FIXME: use status:busy to wait for tasks to start
             time.sleep(1)