Merge pull request #250 in LCL/wolframclientforpython from bugfix/github-30-Quit-handling to master

* commit '0c30a10433232a4b532302475d3fc58148d3ad37':
  code formating
  Add a test for a time constrained computation followed with a terminate call
  update changelog
  update version to 1.1.7
  Future check for done instead of cancelled
  recv also check for termination event
  Remove pip as a dependency
  Add tests running Quit
  Kernel controller is more robust to terminated process

Author: Dorian Birraux

CHANGELOG.md

@@ -1,7 +1,10 @@
+# Version 1.1.7
+- kernel controller class detects terminated kernel process while waiting for an evaluation result instead of hanging.
+
 # Version 1.1.5 and 1.1.6
-    - update Wolfram Language code with latest addition from 12.3.
-    - fix various deprecated naming convention in class names.
-    
+- update Wolfram Language code with latest addition from 12.3.
+- fix various deprecated naming convention in class names.
+
 # Version 1.1.4
 - Async cloud evaluator based on `aiohttp` now use `certifi` to create a default `SSLContext` if none is provided. Other cloud evaluator are based on the `requests` module which also uses this library. 
 - Updating dependency list accordingly in `setup.py`. `certifi` was already listed as a `requests` dependency, so this should have no direct impact on user site package.

setup.py

@@ -60,7 +60,6 @@ def load_tests():
     test_suite='setup.load_tests',
     python_requires='>=3.5.3',
     install_requires = [
-        'pip',
         'numpy',
         'pytz',
         'requests',

wolframclient/about.py

@@ -2,6 +2,6 @@
 
 __name__ = "wolframclient"
 __description__ = "A Python library with various tools to interact with the Wolfram Language and the Wolfram Cloud."
-__version__ = "1.1.6"
+__version__ = "1.1.7"
 __author__ = "Wolfram Research"
 __author_email__ = "support@wolfram.com, dorianb@wolfram.com, riccardod@wolfram.com"

wolframclient/evaluation/kernel/kernelcontroller.py

@@ -434,7 +434,7 @@ def _kernel_start(self):
             # on the kernel side.
             response = self.kernel_socket_in.recv_abortable(
                 timeout=self.get_parameter("STARTUP_TIMEOUT"),
-                abort_event=_StartEvent(self.kernel_proc, self.trigger_termination_requested),
+                abort_event=self._new_running_event(),
             )
             if response == self._KERNEL_OK:
                 if logger.isEnabledFor(logging.INFO):
@@ -483,9 +483,15 @@ def terminate(self):
             if not self.started:
                 future.set_result(True)
                 return future
-            self.enqueue_task(self.STOP, future, None)
+            # only enqueue task if the Event is not triggered.
+            # when trigger_termination_requested is set, the run function is
+            # already dealing with the exception and is about to terminate.
+            if not self.trigger_termination_requested.is_set():
+                self.enqueue_task(self.STOP, future, None)
+                self.trigger_termination_requested.set()
+
+            future.set_result(True)
             self._state_terminated = True
-            self.trigger_termination_requested.set()
         return future
 
     def join(self, timeout=None):
@@ -496,13 +502,37 @@ def join(self, timeout=None):
     def evaluate_future(self, wxf, future, result_update_callback=None, **kwargs):
         self.enqueue_task(wxf, future, result_update_callback)
 
+    def _new_running_event(self):
+        """
+        Create a new event that triggers when the kernel process has terminated or when termination was requested.
+        :return:
+        """
+        return _ProcessAliveNotAbortedEvent(
+            self.kernel_proc, self.trigger_termination_requested
+        )
+
+    def _recv_check_process(self, copy=False):
+        """
+        Call recv on the kernel input socket. Regularly check that the kernel process
+        is running.
+        :param copy: whether or not to copy the socket data. Default is False.
+        :return:
+        """
+        try:
+            return self.kernel_socket_in.recv_abortable(
+                copy=copy, abort_event=self._new_running_event()
+            )
+        except SocketAborted:
+            logger.info("Kernel process is not running anymore.")
+            raise WolframKernelException("Kernel is not running anymore.")
+
     def _do_evaluate(self, wxf, future, result_update_callback):
         start = time.perf_counter()
         self.kernel_socket_out.send(zmq.Frame(wxf))
         if logger.isEnabledFor(logging.DEBUG):
             logger.debug("Expression sent to kernel in %.06fsec", time.perf_counter() - start)
             start = time.perf_counter()
-        wxf_eval_data = self.kernel_socket_in.recv_abortable(copy=False)
+        wxf_eval_data = self._recv_check_process()
         if logger.isEnabledFor(logging.DEBUG):
             logger.debug(
                 "Expression received from kernel after %.06fsec", time.perf_counter() - start
@@ -562,7 +592,7 @@ def run(self):
             raise e
         except Exception as e:
             self.trigger_termination_requested.set()
-            if future and not future.cancelled():
+            if future and not future.done():
                 future.set_exception(e)
                 future = None
             else:
@@ -577,11 +607,11 @@ def run(self):
                 else:
                     self._kernel_stop()
             except Exception as e:
-                if future:
+                if future and not future.done():
                     future.set_exception(e)
                     future = None
             finally:
-                if future:
+                if future and not future.done():
                     future.set_result(True)
 
     def _cancel_tasks(self):
@@ -604,10 +634,18 @@ def __repr__(self):
             return '<%s[%s ❌], "%s">' % (self.__class__.__name__, self.name, self.kernel)
 
 
-class _StartEvent(object):
+class _ProcessAliveNotAbortedEvent(object):
     def __init__(self, subprocess, abort_event):
         self.subprocess = subprocess
         self.abort_event = abort_event
 
     def is_set(self):
         return self.subprocess.poll() is not None or self.abort_event.is_set()
+
+
+class _KernelProcessDied(object):
+    def __init__(self, subprocess):
+        self.subprocess = subprocess
+
+    def is_set(self):
+        return self.subprocess.poll() is not None

wolframclient/evaluation/kernel/localsession.py

@@ -170,13 +170,16 @@ def start_future(self):
         return future
 
     def stop(self):
-        """ Request the Wolfram kernel to stop gracefully. """
+        """ Request the Wolfram kernel to stop gracefully.
+        When recovering from an error state, it's recommended to use terminate() which
+        is more resilient.
+        """
         self._stop(gracefully=True)
 
     def terminate(self):
         """ Request the Wolfram kernel to stop immediately.
-        
-        Ongoing evaluations may be cancelled. """
+        Ongoing evaluations are likely to be cancelled.
+        """
         self._stop(gracefully=False)
 
     def _stop(self, gracefully=True):

wolframclient/tests/evaluation/test_coroutine.py

@@ -9,6 +9,7 @@
     WolframLanguageAsyncSession,
     parallel_evaluate,
 )
+from wolframclient.exception import WolframKernelException
 from wolframclient.language import wl, wlexpr
 from wolframclient.tests.configure import (
     kernel_path,
@@ -111,6 +112,25 @@ async def test_eval_parallel(self):
         res = await asyncio.gather(*tasks)
         self.assertEqual(res, list(range(1, 11)))
 
+    @run_in_loop
+    async def test_quit_restart(self):
+        try:
+            async_session = WolframLanguageAsyncSession(
+                kernel_path, kernel_loglevel=logging.INFO
+            )
+            pid1 = await async_session.evaluate("$ProcessID")
+            self.assertEqual(pid1, async_session.kernel_controller.pid)
+            with self.assertRaises(WolframKernelException):
+                await async_session.evaluate("Quit[]")
+            await async_session.terminate()
+            pid2 = await async_session.evaluate("$ProcessID")
+            self.assertEqual(pid2, async_session.kernel_controller.pid)
+            self.assertNotEqual(pid1, pid2)
+        finally:
+            if async_session:
+                await async_session.terminate()
+            self.assertTrue(async_session.stopped)
+
     def test_kwargs_parameters(self):
         TestKernelBase.class_kwargs_parameters(self, WolframLanguageAsyncSession)
 
@@ -207,7 +227,7 @@ async def test_pool_from_mixed_kernel_cloud_path(self):
             kernel_path,
         )
         async with WolframEvaluatorPool(
-            sessions, kernel_loglevel=logging.INFO, STARTUP_TIMEOUT=5, TERMINATE_TIMEOUT=3
+            sessions, kernel_loglevel=logging.INFO, STARTUP_TIMEOUT=20, TERMINATE_TIMEOUT=3
         ) as pool:
             await self._pool_evaluation_check(pool)
         for session in sessions:

wolframclient/tests/evaluation/test_kernel.py

@@ -1,6 +1,8 @@
 from __future__ import absolute_import, print_function, unicode_literals
 
 import logging
+from concurrent.futures import TimeoutError
+from time import time
 
 from wolframclient.deserializers import WXFConsumer, binary_deserialize
 from wolframclient.evaluation import WolframLanguageSession
@@ -178,6 +180,35 @@ def test_auto_start_session(self):
             session.terminate()
             self.assertTrue(session.stopped)
 
+    def test_kernel_dies_restart(self):
+        try:
+            session = WolframLanguageSession(kernel_path)
+            pid1 = session.evaluate("$ProcessID")
+            self.assertEqual(pid1, session.kernel_controller.pid)
+            with self.assertRaises(WolframKernelException):
+                session.evaluate("Quit[]")
+            session.terminate()
+            pid2 = session.evaluate("$ProcessID")
+            self.assertEqual(pid2, session.kernel_controller.pid)
+            self.assertNotEqual(pid1, pid2)
+        finally:
+            session.terminate()
+            self.assertTrue(session.stopped)
+
+    def test_kernel_abort_restart(self):
+        try:
+            session = WolframLanguageSession(kernel_path)
+            session.start()
+            start = time()
+            future = session.evaluate_future("Pause[10]")
+            with self.assertRaises(TimeoutError):
+                future.result(timeout=1.0)
+            session.terminate()
+            self.assertTrue((time() - start) < 10)
+        finally:
+            session.terminate()
+            self.assertTrue(session.stopped)
+
     def test_pure_function_inputform(self):
         f = self.kernel_session.function("#+1&")
         self.assertEqual(f(3), 4)
@@ -298,7 +329,7 @@ def test_throw(self):
 class TestSessionTimeout(TestCaseSettings):
     def test_evaluate_async_basic_inputform(self):
         future = self.kernel_session.evaluate_future("1+1")
-        self.assertEqual(future.result(timeout=2), 2)
+        self.assertEqual(future.result(timeout=5), 2)
 
     def test_evaluate_async_basic_wl(self):
         future = self.kernel_session.evaluate_future(wl.Plus(1, 2))

wolframclient/tests/externalevaluate/ev_loop.py

@@ -9,6 +9,7 @@
 from wolframclient.utils.externalevaluate import EXPORT_KWARGS, start_zmq_loop
 from wolframclient.utils.tests import TestCase as BaseTestCase
 
+
 class TestCase(BaseTestCase):
     def compare(self, string_version, result):
         self.assertEqual(string_version, export(result, **EXPORT_KWARGS))