AsyncioDispatcher cleanup tasks atexit

.github/workflows/code.yml

@@ -54,7 +54,7 @@ jobs:
       fail-fast: false
       matrix:
         os: [ubuntu-latest, windows-latest, macos-latest]
-        python: [cp36, cp37, cp38, cp39, cp310]
+        python: [cp37, cp38, cp39, cp310]
 
         include:
           # Put coverage and results files in the project directory for mac
@@ -147,7 +147,7 @@ jobs:
       fail-fast: false
       matrix:
         os: [ubuntu-latest, windows-latest, macos-latest]
-        python: [cp36, cp37, cp38, cp39, cp310]
+        python: [cp37, cp38, cp39, cp310]
 
     runs-on: ${{ matrix.os }}
 

CHANGELOG.rst

@@ -10,6 +10,10 @@ Versioning <https://semver.org/spec/v2.0.0.html>`_.
 Unreleased_
 -----------
 
+Changed:
+
+- `AsyncioDispatcher cleanup tasks atexit <../../pull/138>`_
+
 Fixed:
 
 - `Fix conversion of ctypes pointers passed to C extension <../../pull/154>`_

setup.cfg

@@ -10,15 +10,14 @@ long_description_content_type = text/x-rst
 classifiers =
     Development Status :: 5 - Production/Stable
     License :: OSI Approved :: Apache Software License
-    Programming Language :: Python :: 3.6
     Programming Language :: Python :: 3.7
     Programming Language :: Python :: 3.8
     Programming Language :: Python :: 3.9
     Programming Language :: Python :: 3.10
 
 [options]
 packages = softioc
-python_requires = >=3.6
+python_requires = >=3.7
 
 [options.entry_points]
 # Include a command line script

softioc/asyncio_dispatcher.py

@@ -5,33 +5,61 @@
 import atexit
 
 class AsyncioDispatcher:
-    def __init__(self, loop=None):
+    def __init__(self, loop=None, debug=False):
         """A dispatcher for `asyncio` based IOCs, suitable to be passed to
         `softioc.iocInit`. Means that `on_update` callback functions can be
         async.
 
         If a ``loop`` is provided it must already be running. Otherwise a new
         Event Loop will be created and run in a dedicated thread.
+        ``debug`` is passed through to ``asyncio.run()``.
+
+        For a clean exit, call ``softioc.interactive_ioc(..., call_exit=False)``
         """
         if loop is None:
+            # will wait until worker is executing the new loop
+            started = threading.Event()
             # Make one and run it in a background thread
-            self.loop = asyncio.new_event_loop()
-            worker = threading.Thread(target=self.loop.run_forever)
+            self.__worker = threading.Thread(target=asyncio.run,
+                                             args=(self.__inloop(started),),
+                                             kwargs={'debug': debug})
             # Explicitly manage worker thread as part of interpreter shutdown.
             # Otherwise threading module will deadlock trying to join()
             # before our atexit hook runs, while the loop is still running.
-            worker.daemon = True
+            self.__worker.daemon = True
+
+            self.__worker.start()
+            started.wait()
+
+            self.__atexit = atexit.register(self.__shutdown)
+
+            assert self.loop is not None and self.loop.is_running()
 
-            @atexit.register
-            def aioJoin(worker=worker, loop=self.loop):
-                loop.call_soon_threadsafe(loop.stop)
-                worker.join()
-            worker.start()
         elif not loop.is_running():
             raise ValueError("Provided asyncio event loop is not running")
         else:
             self.loop = loop
 
+    def close(self):
+        if self.__atexit is not None:
+            atexit.unregister(self.__atexit)
+            self.__atexit = None
+
+        self.__shutdown()
+
+    async def __inloop(self, started):
+        self.loop = asyncio.get_running_loop()
+        self.__interrupt = asyncio.Event()
+        started.set()
+        del started
+        await self.__interrupt.wait()
+
+    def __shutdown(self):
+        if self.__worker is not None:
+            self.loop.call_soon_threadsafe(self.__interrupt.set)
+            self.__worker.join()
+            self.__worker = None
+
     def __call__(
             self,
             func,
@@ -48,3 +76,9 @@ async def async_wrapper():
             except Exception:
                 logging.exception("Exception when running dispatched callback")
         asyncio.run_coroutine_threadsafe(async_wrapper(), self.loop)
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, A, B, C):
+        self.close()