Implement asynchronous record processing

Includes a "blocking" flag on record creation.

This change allows you to use caput in asynchronous mode, where it will
wait for record processing to complete.

Author: AlexanderWells-diamond

softioc/device.py

@@ -3,10 +3,17 @@
 import ctypes
 from ctypes import *
 import numpy
+from threading import Event
 
 from . import alarm
 from . import fields
-from .imports import dbLoadDatabase, recGblResetAlarms, db_put_field
+from .imports import (
+    create_callback_capsule,
+    dbLoadDatabase,
+    signal_processing_complete,
+    recGblResetAlarms,
+    db_put_field,
+)
 from .device_core import DeviceSupportCore, RecordLookup
 
 
@@ -19,6 +26,7 @@
 # Default False to maintain behaviour from previous versions.
 blocking = False
 
+# TODO: Docs and Tests for the Blocking feature
 def set_blocking(val):
     global blocking
     blocking = val
@@ -152,6 +160,10 @@ def __init__(self, name, **kargs):
             self._value = None
 
         self._blocking = kargs.pop('blocking', blocking)
+        if self._blocking:
+            self._completion_event = Event()
+            self._callback = create_callback_capsule()
+
 
         self.__super.__init__(name, **kargs)
 
@@ -174,15 +186,26 @@ def init_record(self, record):
         return self._epics_rc_
 
     def __completion(self):
+        if self._blocking:
+            self._completion_event.set()
         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):
-        self.__on_update(value)
+        dispatcher(self.__on_update, value)
         self.__completion()
 
     def _process(self, record):
         '''Processing suitable for output records.  Performs immediate value
         validation and asynchronous update notification.'''
+
+        if record.PACT:
+            return EPICS_OK
+
         value = self._read_value(record)
         if not self.__always_update and \
                 self._compare_values(value, self._value):
@@ -201,7 +224,12 @@ def _process(self, record):
             self._value = value
             record.UDF = 0
             if self.__on_update and self.__enable_write:
+                record.PACT = self._blocking
                 dispatcher(self.__wrap_completion, python_value)
+
+                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

@@ -1,18 +1,20 @@
-
+/* Provide EPICS functions in Python format */
 #define PY_SSIZE_T_CLEAN
 #include <Python.h>
 #include <string.h>
 
 #define db_accessHFORdb_accessC     // Needed to get correct DBF_ values
 #include <dbAccess.h>
 #include <dbFldTypes.h>
+#include <callback.h>
 #include <dbStaticLib.h>
 #include <asTrapWrite.h>
 #include <epicsVersion.h>
 #include <dbChannel.h>
 #include <asTrapWrite.h>
 #include <asDbLib.h>
 
+
 /* Reference stealing version of PyDict_SetItemString */
 static void set_dict_item_steal(
     PyObject *dict, const char *name, PyObject *py_value)
@@ -209,6 +211,42 @@ static PyObject *install_pv_logging(PyObject *self, PyObject *args)
     Py_RETURN_NONE;
 }
 
+#define CAPSULE_NAME "ProcessDeviceSupportOut.callback"
+
+static void capsule_destructor(PyObject *obj)
+{
+    void *callback = PyCapsule_GetPointer(obj, CAPSULE_NAME);
+    free(callback);
+}
+
+
+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);
+}
+
+static PyObject *signal_processing_complete(PyObject *self, PyObject *args)
+{
+    int priority;
+    dbCommon *record;
+    PyObject *callback_capsule;
+
+    if (!PyArg_ParseTuple(args, "inO", &priority, &record, &callback_capsule))
+    {
+        return NULL;
+    }
+
+    CALLBACK *callback = PyCapsule_GetPointer(callback_capsule, CAPSULE_NAME);
+
+    callbackRequestProcessCallback(callback, priority, record);
+
+    Py_RETURN_NONE;
+}
+
 static struct PyMethodDef softioc_methods[] = {
     {"get_DBF_values",  get_DBF_values, METH_VARARGS,
      "Get a map of DBF names to values"},
@@ -218,6 +256,10 @@ static struct PyMethodDef softioc_methods[] = {
      "Put a database field to a value"},
     {"install_pv_logging",  install_pv_logging, METH_VARARGS,
      "Install caput logging to stdout"},
+    {"signal_processing_complete",  signal_processing_complete, METH_VARARGS,
+     "Inform EPICS that asynchronous record processing has completed"},
+     {"create_callback_capsule",  create_callback_capsule, METH_VARARGS,
+     "Create a CALLBACK structure inside a PyCapsule"},
     {NULL, NULL, 0, NULL}        /* Sentinel */
 };
 

softioc/imports.py

@@ -27,6 +27,17 @@ def install_pv_logging(acf_file):
     '''Install pv logging'''
     _extension.install_pv_logging(acf_file)
 
+def create_callback_capsule():
+    return _extension.create_callback_capsule()
+
+def signal_processing_complete(record, callback):
+    '''Signal that asynchronous record processing has completed'''
+    _extension.signal_processing_complete(
+        record.PRIO,
+        record.record.value,
+        callback
+    )
+
 def expect_success(status, function, args):
     assert status == 0, 'Expected success'
 
@@ -94,6 +105,8 @@ def from_param(cls, value):
 
 __all__ = [
     'get_field_offsets',
+    'create_callback_capsule',
+    'signal_processing_complete',
     'registryDeviceSupportAdd',
     'IOSCANPVT', 'scanIoRequest', 'scanIoInit',
     'dbLoadDatabase',

softioc/pythonSoftIoc.py

@@ -24,7 +24,7 @@ def __init__(self, builder, device, name, **fields):
         # have to maintain this separately from the corresponding device list.
         DeviceKeywords = [
             'on_update', 'on_update_name', 'validate', 'always_update',
-            'initial_value', '_wf_nelm', '_wf_dtype']
+            'initial_value', '_wf_nelm', '_wf_dtype', 'blocking']
         device_kargs = {}
         for keyword in DeviceKeywords:
             if keyword in fields:

tests/conftest.py

@@ -117,7 +117,7 @@ def enable_code_coverage():
 def select_and_recv(conn, expected_char = None):
     """Wait for the given Connection to have data to receive, and return it.
     If a character is provided check its correct before returning it."""
-    # Must use cothread's select if cothread is prsent, otherwise we'd block
+    # Must use cothread's select if cothread is present, otherwise we'd block
     # processing on all cothread processing. But we don't want to use it
     # unless we have to, as importing cothread can cause issues with forking.
     if "cothread" in sys.modules: