Removing loop parameter in asynchronous evaluators

Author: Dorian Birraux

wolframclient/evaluation/base.py

@@ -143,21 +143,20 @@ def __exit__(self, type, value, traceback):
 
 class WolframAsyncEvaluator(WolframEvaluatorBase):
     """ Asynchronous evaluators are similar to synchronous ones except that they make heavy use of coroutines
-    and need an event loop. 
+    and need to run in an event loop.
     
-    Most methods from this class are similar to their counterpart from :class:`~wolframclient.evaluation.base.WolframEvaluator`,
-    except that they are coroutines. """
+    Most methods from this class are similar to their counterpart from
+    :class:`~wolframclient.evaluation.base.WolframEvaluator`, except that they are coroutines. """
 
-    def __init__(self, loop=None, **kwargs):
-        self._loop = loop or asyncio.get_event_loop()
+    def __init__(self, **kwargs):
         super().__init__(**kwargs)
 
     async def evaluate(self, expr):
         result = await self.evaluate_wrap(expr)
         return await result.get()
 
     async def evaluate_many(self, expr_list):
-        return await asyncio.gather(*map(self.evaluate, expr_list), loop=self._loop)
+        return await asyncio.gather(*map(self.evaluate, expr_list))
 
     async def evaluate_wrap(self, expr):
         raise NotImplementedError
@@ -215,7 +214,13 @@ async def __aexit__(self, type, value, traceback):
 
     def __del__(self, _warnings=warnings):
         super().__del__(_warnings=warnings)
-        if not self.stopped and self._loop and not self._loop.is_closed():
-            self._loop.call_exception_handler(
-                {self.__class__.__name__: self, "message": "Unclosed evaluator."}
-            )
+        if not self.stopped:
+            loop = None
+            try:
+                loop = asyncio.get_running_loop()
+            except RuntimeError:
+                pass
+            if loop and not loop.is_closed():
+                loop.call_exception_handler(
+                    {self.__class__.__name__: self, "message": "Unclosed evaluator."}
+                )

wolframclient/evaluation/cloud/asynccloudsession.py

@@ -38,8 +38,6 @@ class WolframCloudAsyncSession(WolframAsyncEvaluator):
         async with WolframCloudAsyncSession() as session:
             await session.call(...)
 
-    An event loop can be explicitly passed using the named parameter `loop`; otherwise, the one
-    returned by :func:`~asyncio.get_event_loop` is used.
     The initialization options of the class :class:`~wolframclient.evaluation.WolframCloudSession` are also supported by
     this class.
     """
@@ -48,14 +46,13 @@ def __init__(
         self,
         credentials=None,
         server=None,
-        loop=None,
         inputform_string_evaluation=True,
         oauth_session_class=None,
         xauth_session_class=None,
         http_sessionclass=None,
         ssl_context_class=None,
     ):
-        super().__init__(loop, inputform_string_evaluation=inputform_string_evaluation)
+        super().__init__(inputform_string_evaluation=inputform_string_evaluation)
         self.server = server or WOLFRAM_PUBLIC_CLOUD_SERVER
         self.http_session = None
         self.http_sessionclass = http_sessionclass or aiohttp.ClientSession
@@ -76,7 +73,6 @@ def duplicate(self):
         return self.__class__(
             credentials=self.credentials,
             server=self.server,
-            loop=self._loop,
             inputform_string_evaluation=self.inputform_string_evaluation,
             oauth_session_class=self.oauth_session_class,
             xauth_session_class=self.xauth_session_class,
@@ -90,8 +86,7 @@ async def start(self):
             if not self.started:
                 if self.http_session is None or self.http_session.closed:
                     self.http_session = self.http_sessionclass(
-                        headers={"User-Agent": "WolframClientForPython/1.0"}, loop=self._loop
-                    )
+                        headers={"User-Agent": "WolframClientForPython/1.0"})
                 if not self.anonymous():
                     await self._authenticate()
         except Exception as e:

wolframclient/evaluation/kernel/asyncsession.py

@@ -22,10 +22,7 @@ class WolframLanguageAsyncSession(WolframAsyncEvaluator, WolframLanguageSession)
         async with WolframLanguageAsyncSession() as session:
             await session.evaluate('Now')
 
-    An event loop can be explicitly passed using the named parameter `loop`; otherwise, the one
-    returned by :func:`~asyncio.get_event_loop` is used.
-
-    Coroutines all run in a unique thread. Since a Wolfram kernel is single threaded, there can
+    Coroutines all run in their own thread. Since a Wolfram kernel is single threaded, there can
     be only one evaluation at a time. In a sense, from the event loop point of view, evaluations 
     are atomic operations. Even when many asynchronous sessions are started, the number of 
     threads equals the number of kernel instances running and should not be problematic. Ensuring 
@@ -37,7 +34,6 @@ def __init__(
         self,
         kernel=None,
         consumer=None,
-        loop=None,
         initfile=None,
         kernel_loglevel=logging.NOTSET,
         stdin=PIPE,
@@ -55,14 +51,12 @@ def __init__(
             stdout=stdout,
             stderr=stderr,
             inputform_string_evaluation=inputform_string_evaluation,
-            loop=loop,
             **kwargs
         )
 
     def duplicate(self):
         return self.__class__(
             kernel=self.kernel,
-            loop=self._loop,
             consumer=self.consumer,
             initfile=self.initfile,
             kernel_loglevel=self.kernel_loglevel,
@@ -77,7 +71,7 @@ async def do_evaluate_future(self, expr, result_update_callback=None, **kwargs):
         future = super().do_evaluate_future(
             expr, result_update_callback=result_update_callback, **kwargs
         )
-        return asyncio.wrap_future(future, loop=self._loop)
+        return asyncio.wrap_future(future)
 
     async def evaluate_future(self, expr, **kwargs):
         await self.ensure_started()
@@ -138,7 +132,7 @@ async def start(self):
         
         This method is a coroutine."""
         future = super().start_future()
-        await asyncio.wrap_future(future, loop=self._loop)
+        await asyncio.wrap_future(future)
 
     async def stop(self):
         """Asynchronously stop the session (graceful termination).
@@ -155,4 +149,4 @@ async def terminate(self):
     async def _async_terminate(self, gracefully):
         logger.info("Terminating asynchronous kernel session.")
         future = super().stop_future(gracefully=gracefully)
-        await asyncio.wrap_future(future, loop=self._loop)
+        await asyncio.wrap_future(future)

wolframclient/evaluation/pool.py

@@ -58,8 +58,6 @@ class WolframEvaluatorPool(WolframAsyncEvaluator):
     Set `load_factor` to specify how many workloads are queued per kernel before a new evaluation becomes a blocking
     operation. Values below or equal to 0 mean an infinite queue size.
 
-    Set `loop` to the event loop to use.
-
     `kwargs` are passed to :class:`~wolframclient.evaluation.WolframLanguageAsyncSession` during initialization.
     """
 
@@ -68,14 +66,13 @@ def __init__(
         async_evaluators=None,
         poolsize=4,
         load_factor=0,
-        loop=None,
         async_language_session_class=WolframLanguageAsyncSession,
         **kwargs
     ):
-        super().__init__(loop)
+        super().__init__()
         if poolsize <= 0:
             raise ValueError("Invalid pool size value %i. Expecting a positive integer." % poolsize)
-        self._queue = asyncio.Queue(load_factor * poolsize, loop=self._loop)
+        self._queue = asyncio.Queue(load_factor * poolsize)
         self.async_language_session_class = async_language_session_class
         self._evaluators = set()
         if async_evaluators is None or isinstance(async_evaluators, six.string_types):
@@ -98,7 +95,7 @@ def __init__(
     def _add_evaluator(self, evaluator, **kwargs):
         if evaluator is None or isinstance(evaluator, six.string_types):
             self._evaluators.add(
-                self.async_language_session_class(kernel=evaluator, loop=self._loop, **kwargs)
+                self.async_language_session_class(kernel=evaluator, **kwargs)
             )
         elif isinstance(evaluator, WolframAsyncEvaluator):
             if evaluator in self._evaluators:
@@ -196,11 +193,10 @@ async def start(self):
         self.stopped = False
         # keep track of the init tasks. We have to wait before terminating.
         self._kernel_start_tasks = {
-            (asyncio.ensure_future(self._async_start_kernel(kernel), loop=self._loop))
-            for kernel in self._evaluators
+            asyncio.ensure_future(self._async_start_kernel(kernel)) for kernel in self._evaluators
         }
         # uninitialized kernels are removed if they failed to start
-        # if they do start, the task (the loop) is added to _kernel_evaluation_loop_tasks.
+        # if they do start, the task is added to _kernel_evaluation_loop_tasks.
         # we need at least one working kernel.
         # we also need to keep track of start kernel tasks in case of early termination.
         while len(self._kernel_evaluation_loop_tasks) == 0:
@@ -228,15 +224,15 @@ async def stop(self):
                 for _ in range(len(self._kernel_evaluation_loop_tasks)):
                     await self._queue.put(None)
                 # wait for loop to finish before terminating the kernels
-                await asyncio.wait(self._kernel_evaluation_loop_tasks, loop=self._loop)
+                await asyncio.wait(self._kernel_evaluation_loop_tasks)
             except CancelledError:
                 pass
             except Exception as e:
                 logger.warning("Exception raised while terminating loop: %s", e)
         # terminate the kernel instances, if any started.
         tasks = {asyncio.create_task(kernel.stop()) for kernel in self._evaluators}
         # `wait` raises the first exception, but wait for all tasks to finish.
-        await asyncio.wait(tasks, loop=self._loop)
+        await asyncio.wait(tasks)
 
     async def terminate(self):
         await self.stop()
@@ -253,17 +249,17 @@ async def _put_evaluation_task(self, future, func, expr, **kwargs):
         self.eval_count += 1
 
     async def evaluate(self, expr, **kwargs):
-        future = asyncio.Future(loop=self._loop)
+        future = asyncio.Future()
         await self._put_evaluation_task(future, "evaluate", expr, **kwargs)
         return await future
 
     async def evaluate_wxf(self, expr, **kwargs):
-        future = asyncio.Future(loop=self._loop)
+        future = asyncio.Future()
         await self._put_evaluation_task(future, "evaluate_wxf", expr, **kwargs)
         return await future
 
     async def evaluate_wrap(self, expr, **kwargs):
-        future = asyncio.Future(loop=self._loop)
+        future = asyncio.Future()
         await self._put_evaluation_task(future, "evaluate_wrap", expr, **kwargs)
         return await future
 
@@ -283,7 +279,7 @@ def __len__(self):
         return len(self._kernel_evaluation_loop_tasks)
 
 
-def parallel_evaluate(expressions, evaluator_spec=None, max_evaluators=4, loop=None):
+def parallel_evaluate(expressions, evaluator_spec=None, max_evaluators=4):
     """ Start a kernel pool and evaluate the expressions in parallel. 
     
     The pool is created with the value of `evaluator_spec`. The pool is automatically stopped when it is no longer