Add asyncio dispatcher_callback

Author: coretl

.github/workflows/code.yml

@@ -31,23 +31,35 @@ jobs:
       matrix:
         os: [ubuntu-latest, windows-latest, macos-latest]
         python: [cp27, cp37, cp38, cp39]
+        # Install cothread and pytest-asyncio by default
+        test_requires: ['pytest-asyncio cothread aioca']
+        # Put coverage straight in the dist dir by default
+        cov_file: ['{project}/dist/coverage.xml']
 
-        include:
-        - os: macos-latest
-          TEST_REQUIRES: pytest-cov cothread
-          # Put coverage straight in the dist dir
-          COV_FILE: '{project}/dist/coverage.xml'
+        exclude:
+        - os: windows-latest
+          # No cothread or asyncio so doesn't work
+          python: cp27
 
+        include:
         - os: ubuntu-latest
-          TEST_REQUIRES: pytest-cov cothread
           # Put coverage in the output dir mounted in docker
-          COV_FILE: '/output/coverage.xml'
+          cov_file: '/output/coverage.xml'
 
         - os: windows-latest
           # cothread doesn't work on windows
-          TEST_REQUIRES: pytest-cov
-          # Put coverage straight in the dist dir
-          COV_FILE: '{project}/dist/coverage.xml'
+          test_requires: pytest-asyncio aioca
+
+        - os: ubuntu-latest
+          python: cp27
+          # asyncio doesn't work on Python2.7
+          test_requires: cothread
+
+        - os: macos-latest
+          python: cp27
+          # asyncio doesn't work on Python2.7
+          test_requires: cothread
+
 
     steps:
       - name: Checkout Source
@@ -70,8 +82,8 @@ jobs:
         run: cibuildwheel --output-dir dist
         env:
           CIBW_BUILD: ${{ matrix.python }}*64
-          CIBW_TEST_REQUIRES: ${{ matrix.TEST_REQUIRES }}
-          CIBW_TEST_COMMAND: pytest --cov=softioc {project}/tests --cov-report xml:${{ matrix.COV_FILE }}
+          CIBW_TEST_REQUIRES: pytest-cov ${{ matrix.test_requires }}
+          CIBW_TEST_COMMAND: pytest --cov=softioc {project}/tests --cov-report xml:${{ matrix.cov_file }}
           # Disable auditwheel as it isn't compatible with setuptools_dso approach
           # https://github.com/mdavidsaver/setuptools_dso/issues/17
           CIBW_REPAIR_WHEEL_COMMAND: ''

Pipfile

@@ -11,6 +11,7 @@ sphinx-rtd-theme = "*"
 sphinx-multiversion = {editable = true,git = "https://github.com/dls-controls/sphinx-multiversion.git",ref = "only-arg"}
 setuptools-dso = "*"
 aioca = "*"
+pytest-asyncio = "*"
 
 [packages]
 # All other package requirements from setup.py
@@ -20,6 +21,7 @@ epicscorelibs = "*"
 # Add some other useful extras
 cothread = "*"
 scipy = "*"
+aioca = "*"
 
 [scripts]
 # Put coverage here so we don't interfere with debugging in the IDE

Pipfile.lock

@@ -0,0 +1,19 @@
+import asyncio
+import inspect
+import threading
+
+
+class AsyncioCallback(threading.Thread):
+    def __init__(self):
+        super().__init__()
+        self.loop = asyncio.new_event_loop()
+
+    def run(self):
+        self.loop.run_forever()
+
+    def __call__(self, func, *args):
+        async def async_wrapper():
+            ret = func(*args)
+            if inspect.isawaitable(ret):
+                await ret
+        asyncio.run_coroutine_threadsafe(async_wrapper(), self.loop)

softioc/asyncio_callback.py

@@ -1,17 +1,33 @@
 import os
 import time
-import traceback
+import inspect
 from ctypes import *
 import numpy
 
-import cothread
-
 from . import alarm
 from .fields import DbfCodeToNumpy, DbrToDbfCode
 from .imports import dbLoadDatabase, recGblResetAlarms, db_put_field
 from .device_core import DeviceSupportCore, RecordLookup
 
 
+dispatcher_callback = None
+
+
+def use_cothread():
+    import cothread
+    # Create our own cothread callback queue so that our callbacks
+    # processing doesn't interfere with other callback processing.
+    global dispatcher_callback
+    dispatcher_callback = cothread.cothread._Callback()
+
+
+def use_asyncio():
+    from .asyncio_callback import AsyncioCallback
+    global dispatcher_callback
+    dispatcher_callback = AsyncioCallback()
+    dispatcher_callback.start()
+
+
 class ProcessDeviceSupportCore(DeviceSupportCore, RecordLookup):
     '''Implements canonical default processing for records with a _process
     method.  Processing typically either copies a locally set value into the
@@ -79,10 +95,6 @@ def get(self):
 class ProcessDeviceSupportOut(ProcessDeviceSupportCore):
     _link_ = 'OUT'
 
-    # Create our own cothread callback queue so that our callbacks processing
-    # doesn't interfere with other callback processing.
-    __Callback = cothread.cothread._Callback()
-
     def __init__(self, name, **kargs):
         on_update = kargs.pop('on_update', None)
         on_update_name = kargs.pop('on_update_name', None)
@@ -133,7 +145,7 @@ def _process(self, record):
 
         self._value = value
         if self.__on_update and self.__enable_write:
-            self.__Callback(self.__on_update, value)
+            dispatcher_callback(self.__on_update, value)
         return 0
 
 

softioc/device.py

@@ -6,14 +6,19 @@
 
 from epicsdbbuilder.recordset import recordset
 
-from . import imports
+from . import imports, device
 
 __all__ = ['dbLoadDatabase', 'iocInit', 'interactive_ioc']
 
 
-iocInit = imports.iocInit
 epicsExit = imports.epicsExit
 
+def iocInit():
+    if device.dispatcher_callback is None:
+        # Fallback to cothread
+        device.use_cothread()
+    imports.iocInit()
+
 def safeEpicsExit():
     '''Calls epicsExit() after ensuring Python exit handlers called.'''
     if hasattr(sys, 'exitfunc'):
@@ -26,7 +31,6 @@ def safeEpicsExit():
             del sys.exitfunc
     epicsExit()
 
-
 # The following identifiers will be exported to interactive shell.
 command_names = []