Test and document asyncio support.

python · ericsnowcurrently · Oct 16, 2024 · Sep 25, 2024 · Sep 25, 2024 · Sep 27, 2024
commit 744dca7ee2ad8216777d8672613dc3889ab4a21e
@@ -103,7 +103,8 @@ To handle signals the event loop must be
 run in the main thread.
 
 The :meth:`loop.run_in_executor` method can be used with a
-:class:`concurrent.futures.ThreadPoolExecutor` to execute
+:class:`concurrent.futures.ThreadPoolExecutor` or
+:class:`~concurrent.futures.InterpreterPoolExecutor` to execute
 blocking code in a different OS thread without blocking the OS thread
 that the event loop runs in.
 
@@ -128,7 +129,8 @@ if a function performs a CPU-intensive calculation for 1 second,
 all concurrent asyncio Tasks and IO operations would be delayed
 by 1 second.
 
-An executor can be used to run a task in a different thread or even in
+An executor can be used to run a task in a different thread,
+including in a different interpreter, or even in
 a different process to avoid blocking the OS thread with the
 event loop.  See the :meth:`loop.run_in_executor` method for more
 details.

@@ -1305,6 +1305,12 @@ Executing code in thread or process pools
                   pool, cpu_bound)
               print('custom process pool', result)
 
+          # 4. Run in a custom interpreter pool:
+          with concurrent.futures.InterpreterPoolExecutor() as pool:
+              result = await loop.run_in_executor(
+                  pool, cpu_bound)
+              print('custom interpreter pool', result)
+
       if __name__ == '__main__':
           asyncio.run(main())
 
@@ -1329,7 +1335,8 @@ Executing code in thread or process pools
 
    Set *executor* as the default executor used by :meth:`run_in_executor`.
    *executor* must be an instance of
-   :class:`~concurrent.futures.ThreadPoolExecutor`.
+   :class:`~concurrent.futures.ThreadPoolExecutor`, which includes
+   :class:`~concurrent.futures.InterpreterPoolExecutor`.
 
    .. versionchanged:: 3.11
       *executor* must be an instance of

@@ -96,7 +96,7 @@ See also the main documentation section about the
       - Invoke a callback *at* the given time.
 
 
-.. rubric:: Thread/Process Pool
+.. rubric:: Thread/Interpreter/Process Pool
 .. list-table::
     :widths: 50 50
     :class: full-width-table

diff --git a/Lib/test/test_concurrent_futures/test_interpreter_pool.py b/Lib/test/test_concurrent_futures/test_interpreter_pool.py
@@ -1,13 +1,17 @@
+import asyncio
 import contextlib
 import io
 import os
 import pickle
 import sys
+import time
 import unittest
 from concurrent.futures.interpreter import (
     ExecutionFailed, BrokenInterpreterPool,
 )
+import _interpreters
 from test import support
+import test.test_asyncio.utils as testasyncio_utils
 from test.support.interpreters import queues
 
 from .executor import ExecutorTest, mul
@@ -34,14 +38,23 @@ def fail(exctype, msg=None):
     raise exctype(msg)
 
 
-class InterpreterPoolExecutorTest(InterpreterPoolMixin, ExecutorTest, BaseTestCase):
+def get_current_interpid(*extra):
+    interpid, _ = _interpreters.get_current()
+    return (interpid, *extra)
+
+
+class InterpretersMixin(InterpreterPoolMixin):
 
     def pipe(self):
         r, w = os.pipe()
         self.addCleanup(lambda: os.close(r))
         self.addCleanup(lambda: os.close(w))
         return r, w
 
+
+class InterpreterPoolExecutorTest(
+            InterpretersMixin, ExecutorTest, BaseTestCase):
+
     def test_init_script(self):
         msg1 = b'step: init'
         msg2 = b'step: run'
@@ -256,6 +269,64 @@ def test_idle_thread_reuse(self):
         executor.shutdown(wait=True)
 
 
+class AsyncioTest(InterpretersMixin, testasyncio_utils.TestCase):
+
+    def setUp(self):
+        super().setUp()
+        self.loop = asyncio.new_event_loop()
+        self.set_event_loop(self.loop)
+
+        self.executor = self.executor_type()
+        self.addCleanup(lambda: self.executor.shutdown())
+
+    def tearDown(self):
+        if not self.loop.is_closed():
+            testasyncio_utils.run_briefly(self.loop)
+
+        self.doCleanups()
+        support.gc_collect()
+        super().tearDown()
+
+    def test_run_in_executor(self):
+        unexpected, _ = _interpreters.get_current()
+
+        func = get_current_interpid
+        fut = self.loop.run_in_executor(self.executor, func, 'yo')
+        interpid, res = self.loop.run_until_complete(fut)
+
+        self.assertEqual(res, 'yo')
+        self.assertNotEqual(interpid, unexpected)
+
+    def test_run_in_executor_cancel(self):
+        executor = self.executor_type()
+
+        called = False
+
+        def patched_call_soon(*args):
+            nonlocal called
+            called = True
+
+        func = time.sleep
+        fut = self.loop.run_in_executor(self.executor, func, 0.05)
+        fut.cancel()
+        self.loop.run_until_complete(
+                self.loop.shutdown_default_executor())
+        self.loop.close()
+        self.loop.call_soon = patched_call_soon
+        self.loop.call_soon_threadsafe = patched_call_soon
+        time.sleep(0.4)
+        self.assertFalse(called)
+
+    def test_default_executor(self):
+        unexpected, _ = _interpreters.get_current()
+
+        self.loop.set_default_executor(self.executor)
+        fut = self.loop.run_in_executor(None, get_current_interpid)
+        interpid, = self.loop.run_until_complete(fut)
+
+        self.assertNotEqual(interpid, unexpected)
+
+
 def setUpModule():
     setup_module()