Refactor dispatchers to remove implementation leak

By creating/modifying the dispatchers we can easily handle the
__completion being called after the __on_update.
Without doing this there seem to be unavoidable implementation leaks,
where the device.py file would have to care about the difference
between cothread and asyncio.

Author: AlexanderWells-diamond

softioc/asyncio_dispatcher.py

@@ -32,12 +32,18 @@ def aioJoin(worker=worker, loop=self.loop):
         else:
             self.loop = loop
 
-    def __call__(self, func, *args):
+    def __call__(self, func, completion, func_args=(), completion_args=()):
         async def async_wrapper():
             try:
-                ret = func(*args)
-                if inspect.isawaitable(ret):
-                    await ret
+                if inspect.iscoroutinefunction(func):
+                    await func(*func_args)
+                else:
+                    ret = func(*func_args)
+                    # Handle the case of a synchronous function that returns a
+                    # coroutine, like the lambda for on_update_name does
+                    if inspect.isawaitable(ret):
+                        await ret
+                completion(*completion_args)
             except Exception:
                 logging.exception("Exception when awaiting callback")
         asyncio.run_coroutine_threadsafe(async_wrapper(), self.loop)

softioc/cothread_dispatcher.py

@@ -0,0 +1,17 @@
+
+class CothreadDispatcher:
+    def __init__(self):
+        """A dispatcher for `cothread` based IOCs, suitable to be passed to
+        `softioc.iocInit`. """
+        # Import here to ensure we don't instantiate any of cothread's global
+        # state unless we have to
+        import cothread
+        # Create our own cothread callback queue so that our callbacks
+        # processing doesn't interfere with other callback processing.
+        self.__dispatcher = cothread.cothread._Callback()
+
+    def __call__(self, func, completion, func_args=(), completion_args=()):
+        def wrapper():
+            func(*func_args)
+            completion(*completion_args)
+        self.__dispatcher(wrapper)

softioc/device.py

@@ -150,19 +150,6 @@ def __init__(self, name, **kargs):
         else:
             self.__on_update = None
 
-        # We cannot simply use `iscoroutinefunction` at the point of calling
-        # on_update as the lambda used for on_update_name is NOT a coroutine.
-        # There's no way to examine the lambda to see if it'd return a
-        # coroutine without calling it, so we must keep track of it ourselves.
-        # This is an unfortunate, unavoidable, leak of implementation detail
-        # that really should be contained in the dispatcher.
-        self.__on_update_is_coroutine = False
-        if (
-            iscoroutinefunction(on_update)
-            or iscoroutinefunction(on_update_name)
-        ):
-            self.__on_update_is_coroutine = True
-
         self.__validate = kargs.pop('validate', None)
         self.__always_update = kargs.pop('always_update', False)
         self.__enable_write = True
@@ -203,14 +190,6 @@ def __completion(self, record):
             signal_processing_complete(record, self._callback)
         pass
 
-    def __wrap_completion(self, value, record):
-        self.__on_update(value)
-        self.__completion(record)
-
-    async def __async_wrap_completion(self, future, record):
-        await future
-        self.__completion(record)
-
     def _process(self, record):
         '''Processing suitable for output records.  Performs immediate value
         validation and asynchronous update notification.'''
@@ -237,21 +216,12 @@ def _process(self, record):
             record.UDF = 0
             if self.__on_update and self.__enable_write:
                 record.PACT = self._blocking
-
-                if self.__on_update_is_coroutine:
-                    # This is an unfortunate, but unavoidable, leak of
-                    # implementation detail that really should be kept within
-                    # the dispatcher, but cannot be. This is due to asyncio not
-                    # allowing its event loop to be nested, thus either
-                    # requiring an additional call to the dispatcher once you
-                    # acquire the Future from the coroutine, or doing this.
-                    dispatcher(
-                        self.__async_wrap_completion,
-                        self.__on_update(python_value),
-                        record
-                    )
-                else:
-                    dispatcher(self.__wrap_completion, python_value, record)
+                dispatcher(
+                    self.__on_update,
+                    self.__completion,
+                    func_args=(python_value,),
+                    completion_args=(record,)
+                )
 
             return EPICS_OK
 

softioc/softioc.py

@@ -7,6 +7,7 @@
 from epicsdbbuilder.recordset import recordset
 
 from . import imports, device
+from . import cothread_dispatcher
 
 __all__ = ['dbLoadDatabase', 'iocInit', 'interactive_ioc']
 
@@ -31,10 +32,7 @@ def iocInit(dispatcher=None):
     '''
     if dispatcher is None:
         # Fallback to cothread
-        import cothread
-        # Create our own cothread callback queue so that our callbacks
-        # processing doesn't interfere with other callback processing.
-        dispatcher = cothread.cothread._Callback()
+        dispatcher = cothread_dispatcher.CothreadDispatcher()
     # Set the dispatcher for record processing callbacks
     device.dispatcher = dispatcher
     imports.iocInit()

tests/sim_asyncio_ioc.py

@@ -36,6 +36,17 @@ async def callback(value):
         # Create a record to set the alarm
         t_ao = builder.aOut('ALARM', on_update=callback)
 
+        async def on_update_name_callback(value, name):
+            print(name, "value", value)
+
+        builder.longOut(
+            "NAME-CALLBACK",
+            initial_value = 3,
+            always_update=True,
+            on_update_name=on_update_name_callback,
+            blocking=True
+        )
+
         # Run the IOC
         builder.LoadDatabase()
         softioc.iocInit(asyncio_dispatcher.AsyncioDispatcher())

tests/test_asyncio.py

@@ -39,6 +39,8 @@ async def test_asyncio_ioc(asyncio_ioc):
 
         await caput(pre + ":ALARM", 3, wait=True)
 
+        await caput(pre + ":NAME-CALLBACK", 12, wait=True)
+
         # Confirm the ALARM callback has completed
         select_and_recv(conn, "C")  # "Complete"
 
@@ -68,6 +70,7 @@ async def test_asyncio_ioc(asyncio_ioc):
         assert "%s:ALARM.VAL 0 -> 3" % pre in out
         assert 'on_update %s:AO : 3.0' % pre in out
         assert 'async update 3.0 (23.45)' in out
+        assert "%s:NAME-CALLBACK value 12" % pre in out
         assert 'Starting iocInit' in err
         assert 'iocRun: All initialization complete' in err
     except Exception: