Avoid multiple dispatcher() calls for non-async

This implementation still leaks abstractions - the device now needs to
know about the difference between cothread and async calls.
Perhaps another round of refactoring is required...

Author: AlexanderWells-diamond

softioc/device.py

@@ -1,3 +1,4 @@
+from inspect import isawaitable
 import os
 import time
 import ctypes
@@ -161,7 +162,6 @@ def __init__(self, name, **kargs):
 
         self._blocking = kargs.pop('blocking', blocking)
         if self._blocking:
-            self._completion_event = Event()
             self._callback = create_callback_capsule()
 
 
@@ -185,19 +185,21 @@ def init_record(self, record):
             recGblResetAlarms(record)
         return self._epics_rc_
 
-    def __completion(self):
+    def __completion(self, record):
         if self._blocking:
-            self._completion_event.set()
+            signal_processing_complete(record, self._callback)
         pass
 
-    def __wait_for_completion(self, record):
-        self._completion_event.wait()
-        signal_processing_complete(record, self._callback)
-        self._completion_event.clear()
+    def __wrap_completion(self, value, record):
+        update = self.__on_update(value)
+        if isawaitable(update):
+            dispatcher(self._complete_update, update, record)
+        else:
+            self.__completion(record)
 
-    def __wrap_completion(self, value):
-        dispatcher(self.__on_update, value)
-        self.__completion()
+    async def _complete_update(self, future, record):
+        await future
+        self.__completion(record)
 
     def _process(self, record):
         '''Processing suitable for output records.  Performs immediate value
@@ -225,11 +227,8 @@ def _process(self, record):
             record.UDF = 0
             if self.__on_update and self.__enable_write:
                 record.PACT = self._blocking
-                dispatcher(self.__wrap_completion, python_value)
+                dispatcher(self.__wrap_completion, python_value, record)
 
-                if self._blocking:
-                    # Create a process to wait for on_update to finish
-                    dispatcher(self.__wait_for_completion, record)
             return EPICS_OK
 
 

softioc/extension.c

@@ -215,17 +215,13 @@ static PyObject *install_pv_logging(PyObject *self, PyObject *args)
 
 static void capsule_destructor(PyObject *obj)
 {
-    void *callback = PyCapsule_GetPointer(obj, CAPSULE_NAME);
-    free(callback);
+    free(PyCapsule_GetPointer(obj, CAPSULE_NAME));
 }
 
 
 static PyObject *create_callback_capsule(PyObject *self, PyObject *args)
 {
     void *callback = malloc(sizeof(CALLBACK));
-
-    printf("Created CALLBACK struct %p\n", callback);
-
     return PyCapsule_New(callback, CAPSULE_NAME, &capsule_destructor);
 }
 
@@ -240,6 +236,14 @@ static PyObject *signal_processing_complete(PyObject *self, PyObject *args)
         return NULL;
     }
 
+    if (!PyCapsule_IsValid(callback_capsule, CAPSULE_NAME))
+    {
+        return PyErr_Format(
+            PyExc_TypeError,
+            "Given object was not a capsule with name \"%s\"",
+            CAPSULE_NAME);
+    }
+
     CALLBACK *callback = PyCapsule_GetPointer(callback_capsule, CAPSULE_NAME);
 
     callbackRequestProcessCallback(callback, priority, record);