Check AsyncioDispatcher loop is running

Also added documentation.
Also fixed potential permanent block in tests - if the subprocess never
creates its Client, the parent process blocks forever on an accept().
This was the case in the asyncio_ioc_override, where initialization was
failing due to non-running loop.

Author: AlexanderWells-diamond

docs/reference/api.rst

@@ -134,6 +134,8 @@ Test Facilities`_ documentation for more details of each function.
     If your application uses `asyncio` then this module gives an alternative
     dispatcher for caput requests.
 
+.. autoclass:: softioc.asyncio_dispatcher.AsyncioDispatcher
+
 .. automodule:: softioc.builder
 
     Creating Records: `softioc.builder`

softioc/asyncio_dispatcher.py

@@ -8,10 +8,11 @@ class AsyncioDispatcher:
     def __init__(self, loop=None):
         """A dispatcher for `asyncio` based IOCs, suitable to be passed to
         `softioc.iocInit`. Means that `on_update` callback functions can be
-        async. If loop is None, will run an Event Loop in a thread when created.
+        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.
         """
-        #: `asyncio` event loop that the callbacks will run under.
-        self.loop = loop
         if loop is None:
             # Make one and run it in a background thread
             self.loop = asyncio.new_event_loop()
@@ -26,6 +27,10 @@ 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 __call__(self, func, *args):
         async def async_wrapper():

tests/conftest.py

@@ -49,7 +49,7 @@ def kill(self):
         if self.proc.returncode is None:
             # still running, kill it and print the output
             self.proc.kill()
-            out, err = self.proc.communicate()
+            out, err = self.proc.communicate(timeout=TIMEOUT)
             print(out.decode())
             print(err.decode())
 

tests/sim_asyncio_ioc.py

@@ -9,15 +9,14 @@
 from conftest import ADDRESS, select_and_recv
 
 if __name__ == "__main__":
-    # Being run as an IOC, so parse args and set prefix
-    parser = ArgumentParser()
-    parser.add_argument('prefix', help="The PV prefix for the records")
-    parsed_args = parser.parse_args()
-    builder.SetDeviceName(parsed_args.prefix)
-
-    import sim_records
-
     with Client(ADDRESS) as conn:
+        # Being run as an IOC, so parse args and set prefix
+        parser = ArgumentParser()
+        parser.add_argument('prefix', help="The PV prefix for the records")
+        parsed_args = parser.parse_args()
+        builder.SetDeviceName(parsed_args.prefix)
+
+        import sim_records
 
         async def callback(value):
             # Set the ao value, which will trigger on_update for it

tests/sim_asyncio_ioc_override.py

@@ -7,38 +7,42 @@
 from tempfile import NamedTemporaryFile
 from argparse import ArgumentParser
 from multiprocessing.connection import Client
+import threading
 
 from softioc import softioc, builder, asyncio_dispatcher
 
 from conftest import ADDRESS, select_and_recv
 
 if __name__ == "__main__":
-    # Being run as an IOC, so parse args and set prefix
-    parser = ArgumentParser()
-    parser.add_argument('prefix', help="The PV prefix for the records")
-    parsed_args = parser.parse_args()
-    builder.SetDeviceName(parsed_args.prefix)
-
-    # Load the base records without DTYP fields
-    with open(Path(__file__).parent / "hw_records.db", "rb") as inp:
-        with NamedTemporaryFile(suffix='.db', delete=False) as out:
-            for line in inp.readlines():
-                if not re.match(rb"\s*field\s*\(\s*DTYP", line):
-                    out.write(line)
-    softioc.dbLoadDatabase(
-        out.name, substitutions=f"device={parsed_args.prefix}")
-    os.unlink(out.name)
-
-    # Override DTYPE and OUT, and provide a callback
-    gain = builder.boolOut("GAIN", on_update=print)
-    softioc.devIocStats(parsed_args.prefix)
-
-    # Run the IOC
-    builder.LoadDatabase()
-    event_loop = asyncio.get_event_loop()
-    softioc.iocInit(asyncio_dispatcher.AsyncioDispatcher(event_loop))
-
     with Client(ADDRESS) as conn:
+        # Being run as an IOC, so parse args and set prefix
+        parser = ArgumentParser()
+        parser.add_argument('prefix', help="The PV prefix for the records")
+        parsed_args = parser.parse_args()
+        builder.SetDeviceName(parsed_args.prefix)
+
+        # Load the base records without DTYP fields
+        with open(Path(__file__).parent / "hw_records.db", "rb") as inp:
+            with NamedTemporaryFile(suffix='.db', delete=False) as out:
+                for line in inp.readlines():
+                    if not re.match(rb"\s*field\s*\(\s*DTYP", line):
+                        out.write(line)
+        softioc.dbLoadDatabase(
+            out.name, substitutions=f"device={parsed_args.prefix}")
+        os.unlink(out.name)
+
+        # Override DTYPE and OUT, and provide a callback
+        gain = builder.boolOut("GAIN", on_update=print)
+        softioc.devIocStats(parsed_args.prefix)
+
+        # Run the IOC
+        builder.LoadDatabase()
+        event_loop = asyncio.get_event_loop()
+        worker = threading.Thread(target=event_loop.run_forever)
+        worker.daemon = True
+        worker.start()
+        softioc.iocInit(asyncio_dispatcher.AsyncioDispatcher(event_loop))
+
         conn.send("R")  # "Ready"
 
         # Make sure coverage is written on epicsExit

tests/sim_cothread_ioc.py

@@ -7,21 +7,21 @@
 from conftest import ADDRESS, select_and_recv
 
 if __name__ == "__main__":
-    import cothread
+    with Client(ADDRESS) as conn:
+        import cothread
 
-    # Being run as an IOC, so parse args and set prefix
-    parser = ArgumentParser()
-    parser.add_argument('prefix', help="The PV prefix for the records")
-    parsed_args = parser.parse_args()
-    builder.SetDeviceName(parsed_args.prefix)
+        # Being run as an IOC, so parse args and set prefix
+        parser = ArgumentParser()
+        parser.add_argument('prefix', help="The PV prefix for the records")
+        parsed_args = parser.parse_args()
+        builder.SetDeviceName(parsed_args.prefix)
 
-    import sim_records
+        import sim_records
 
-    # Run the IOC
-    builder.LoadDatabase()
-    softioc.iocInit()
+        # Run the IOC
+        builder.LoadDatabase()
+        softioc.iocInit()
 
-    with Client(ADDRESS) as conn:
         conn.send("R")  # "Ready"
 
         select_and_recv(conn, "D")  # "Done"

tests/test_asyncio.py

@@ -1,10 +1,13 @@
+import asyncio
 import pytest
 import sys
 
 from multiprocessing.connection import Listener
 
 from conftest import requires_cothread, ADDRESS, select_and_recv
 
+from softioc.asyncio_dispatcher import AsyncioDispatcher
+
 @pytest.mark.asyncio
 async def test_asyncio_ioc(asyncio_ioc):
     import asyncio
@@ -112,3 +115,10 @@ async def test_asyncio_ioc_override(asyncio_ioc_override):
         print("Out:", out)
         print("Err:", err)
         raise
+
+def test_asyncio_dispatcher_event_loop():
+    """Test that passing a non-running event loop to the AsyncioDispatcher
+    raises an exception"""
+    event_loop = asyncio.get_event_loop()
+    with pytest.raises(ValueError):
+        AsyncioDispatcher(loop=event_loop)