Try to remove background thread

Author: marcelldls

.gitignore

@@ -45,6 +45,7 @@ coverage.xml
 cov.xml
 .pytest_cache/
 .mypy_cache/
+.benchmarks/
 
 # Translations
 *.mo

.vscode/launch.json

@@ -16,7 +16,7 @@
             "console": "integratedTerminal",
             "env": {
                 // Enable break on exception when debugging tests (see: tests/conftest.py)
-                "PYTEST_RAISE": "1",
+                "PYTEST_RAISE": "1"
             },
         }
     ]

pyproject.toml

@@ -34,6 +34,7 @@ dev = [
     "pydata-sphinx-theme>=0.12",
     "pyright",
     "pytest",
+    "pytest-benchmark",
     "pytest-cov",
     "pytest-mock",
     "pytest-asyncio",
@@ -69,7 +70,7 @@ reportMissingImports = false  # Ignore missing stubs in imported modules
 [tool.pytest.ini_options]
 # Run pytest with all our checkers, and don't spam us with massive tracebacks on error
 addopts = """
-    --tb=native -vv --doctest-modules --doctest-glob="*.rst"
+    --tb=native -vv --doctest-modules --doctest-glob="*.rst" --benchmark-sort=mean --benchmark-autosave --benchmark-columns="mean, min, max, outliers, ops, rounds"
     """
 # https://iscinumpy.gitlab.io/post/bound-version-constraints/#watch-for-warnings
 filterwarnings = "error"

src/fastcs/attributes.py

@@ -69,9 +69,10 @@ def __init__(
         allowed_values: list[T] | None = None,
         description: str | None = None,
     ) -> None:
-        assert (
-            datatype.dtype in ATTRIBUTE_TYPES
-        ), f"Attr type must be one of {ATTRIBUTE_TYPES}, received type {datatype.dtype}"
+        assert datatype.dtype in ATTRIBUTE_TYPES, (
+            f"Attr type must be one of {ATTRIBUTE_TYPES}"
+            f", received type {datatype.dtype}"
+        )
         self._datatype: DataType[T] = datatype
         self._access_mode: AttrMode = access_mode
         self._group = group

src/fastcs/backend.py

@@ -1,11 +1,8 @@
 import asyncio
 from collections import defaultdict
 from collections.abc import Callable
-from concurrent.futures import Future
 from types import MethodType
 
-from softioc.asyncio_dispatcher import AsyncioDispatcher
-
 from .attributes import AttrR, AttrW, Sender, Updater
 from .controller import Controller, SingleMapping
 from .exceptions import FastCSException
@@ -15,19 +12,15 @@ class Backend:
     def __init__(
         self,
         controller: Controller,
-        loop: asyncio.AbstractEventLoop | None = None,
+        loop: asyncio.AbstractEventLoop,
     ):
-        self.dispatcher = AsyncioDispatcher(loop)
-        self._loop = self.dispatcher.loop
+        self._loop = loop
         self._controller = controller
 
         self._initial_coros = [controller.connect]
-        self._scan_futures: set[Future] = set()
-
-        asyncio.run_coroutine_threadsafe(
-            self._controller.initialise(), self._loop
-        ).result()
+        self._scan_tasks: set[asyncio.Task] = set()
 
+        loop.run_until_complete(self._controller.initialise())
         self._link_process_tasks()
 
     def _link_process_tasks(self):
@@ -36,28 +29,26 @@ def _link_process_tasks(self):
             _link_attribute_sender_class(single_mapping)
 
     def __del__(self):
-        self.stop_scan_futures()
+        self._stop_scan_tasks()
 
-    def run(self):
-        self._run_initial_futures()
-        self.start_scan_futures()
+    async def serve(self):
+        await self._run_initial_tasks()
+        await self._start_scan_tasks()
 
-    def _run_initial_futures(self):
+    async def _run_initial_tasks(self):
         for coro in self._initial_coros:
-            future = asyncio.run_coroutine_threadsafe(coro(), self._loop)
-            future.result()
+            await coro()
 
-    def start_scan_futures(self):
-        self._scan_futures = {
-            asyncio.run_coroutine_threadsafe(coro(), self._loop)
-            for coro in _get_scan_coros(self._controller)
+    async def _start_scan_tasks(self):
+        self._scan_tasks = {
+            self._loop.create_task(coro()) for coro in _get_scan_coros(self._controller)
         }
 
-    def stop_scan_futures(self):
-        for future in self._scan_futures:
-            if not future.done():
+    def _stop_scan_tasks(self):
+        for task in self._scan_tasks:
+            if not task.done():
                 try:
-                    future.cancel()
+                    task.cancel()
                 except asyncio.CancelledError:
                     pass
 
@@ -83,9 +74,9 @@ def _link_attribute_sender_class(single_mapping: SingleMapping) -> None:
     for attr_name, attribute in single_mapping.attributes.items():
         match attribute:
             case AttrW(sender=Sender()):
-                assert (
-                    not attribute.has_process_callback()
-                ), f"Cannot assign both put method and Sender object to {attr_name}"
+                assert not attribute.has_process_callback(), (
+                    f"Cannot assign both put method and Sender object to {attr_name}"
+                )
 
                 callback = _create_sender_callback(attribute, single_mapping.controller)
                 attribute.set_process_callback(callback)

src/fastcs/launch.py

@@ -1,3 +1,4 @@
+import asyncio
 import inspect
 import json
 from pathlib import Path
@@ -19,7 +20,9 @@
 from .transport.tango.options import TangoOptions
 
 # Define a type alias for transport options
-TransportOptions: TypeAlias = EpicsOptions | TangoOptions | RestOptions | GraphQLOptions
+TransportOptions: TypeAlias = list[
+    EpicsOptions | TangoOptions | RestOptions | GraphQLOptions
+]
 
 
 class FastCS:
@@ -28,48 +31,63 @@ def __init__(
         controller: Controller,
         transport_options: TransportOptions,
     ):
-        self._backend = Backend(controller)
-        self._transport: TransportAdapter
-        match transport_options:
-            case EpicsOptions():
-                from .transport.epics.adapter import EpicsTransport
-
-                self._transport = EpicsTransport(
-                    controller,
-                    self._backend.dispatcher,
-                    transport_options,
-                )
-            case GraphQLOptions():
-                from .transport.graphQL.adapter import GraphQLTransport
-
-                self._transport = GraphQLTransport(
-                    controller,
-                    transport_options,
-                )
-            case TangoOptions():
-                from .transport.tango.adapter import TangoTransport
-
-                self._transport = TangoTransport(
-                    controller,
-                    transport_options,
-                )
-            case RestOptions():
-                from .transport.rest.adapter import RestTransport
-
-                self._transport = RestTransport(
-                    controller,
-                    transport_options,
-                )
+        self._loop = asyncio.get_event_loop()
+        self._backend = Backend(controller, self._loop)
+        transport: TransportAdapter
+        self._transports: list[TransportAdapter] = []
+        for option in transport_options:
+            match option:
+                case EpicsOptions():
+                    from .transport.epics.adapter import EpicsTransport
+
+                    transport = EpicsTransport(
+                        controller,
+                        self._loop,
+                        option,
+                    )
+                case TangoOptions():
+                    from .transport.tango.adapter import TangoTransport
+
+                    transport = TangoTransport(
+                        controller,
+                        self._loop,
+                        option,
+                    )
+                case RestOptions():
+                    from .transport.rest.adapter import RestTransport
+
+                    transport = RestTransport(
+                        controller,
+                        option,
+                    )
+                case GraphQLOptions():
+                    from .transport.graphQL.adapter import GraphQLTransport
+
+                    transport = GraphQLTransport(
+                        controller,
+                        option,
+                    )
+            self._transports.append(transport)
 
     def create_docs(self) -> None:
-        self._transport.create_docs()
+        for transport in self._transports:
+            if hasattr(transport.options, "docs"):
+                transport.create_docs()
 
     def create_gui(self) -> None:
-        self._transport.create_gui()
+        for transport in self._transports:
+            if hasattr(transport.options, "gui"):
+                transport.create_docs()
 
-    def run(self) -> None:
-        self._backend.run()
-        self._transport.run()
+    def run(self):
+        self._loop.run_until_complete(
+            self.serve(),
+        )
+
+    async def serve(self) -> None:
+        coros = [self._backend.serve()]
+        coros.extend([transport.serve() for transport in self._transports])
+        await asyncio.gather(*coros)
 
 
 def launch(
@@ -158,10 +176,8 @@ def run(
             instance_options.transport,
         )
 
-        if "gui" in options_yaml["transport"]:
-            instance.create_gui()
-        if "docs" in options_yaml["transport"]:
-            instance.create_docs()
+        instance.create_gui()
+        instance.create_docs()
         instance.run()
 
     @launch_typer.command(name="version", help=f"{controller_class.__name__} version")

src/fastcs/transport/adapter.py

@@ -1,9 +1,15 @@
 from abc import ABC, abstractmethod
+from typing import Any
 
 
 class TransportAdapter(ABC):
+    @property
     @abstractmethod
-    def run(self) -> None:
+    def options(self) -> Any:
+        pass
+
+    @abstractmethod
+    async def serve(self) -> None:
         pass
 
     @abstractmethod

src/fastcs/transport/epics/adapter.py

@@ -1,4 +1,4 @@
-from softioc.asyncio_dispatcher import AsyncioDispatcher
+import asyncio
 
 from fastcs.controller import Controller
 from fastcs.transport.adapter import TransportAdapter
@@ -13,20 +13,30 @@ class EpicsTransport(TransportAdapter):
     def __init__(
         self,
         controller: Controller,
-        dispatcher: AsyncioDispatcher,
+        loop: asyncio.AbstractEventLoop,
         options: EpicsOptions | None = None,
     ) -> None:
-        self.options = options or EpicsOptions()
         self._controller = controller
-        self._dispatcher = dispatcher
+        self._loop = loop
+        self._options = options or EpicsOptions()
         self._pv_prefix = self.options.ioc.pv_prefix
-        self._ioc = EpicsIOC(self.options.ioc.pv_prefix, controller)
+        self._ioc = EpicsIOC(
+            self.options.ioc.pv_prefix,
+            controller,
+            self._options.ioc,
+        )
+
+    @property
+    def options(self) -> EpicsOptions:
+        return self._options
 
     def create_docs(self) -> None:
         EpicsDocs(self._controller).create_docs(self.options.docs)
 
     def create_gui(self) -> None:
         EpicsGUI(self._controller, self._pv_prefix).create_gui(self.options.gui)
 
-    def run(self):
-        self._ioc.run(self._dispatcher)
+    async def serve(self) -> None:
+        self._ioc.run(self._loop)
+        while True:
+            await asyncio.sleep(1)

src/fastcs/transport/epics/ioc.py

@@ -1,3 +1,4 @@
+import asyncio
 from collections.abc import Callable
 from dataclasses import asdict
 from types import MethodType
@@ -50,7 +51,7 @@ def __init__(
         controller: Controller,
         options: EpicsIOCOptions | None = None,
     ):
-        self.options = options or EpicsIOCOptions()
+        self._options = options or EpicsIOCOptions()
         self._controller = controller
         _add_pvi_info(f"{pv_prefix}:PVI")
         _add_sub_controller_pvi_info(pv_prefix, controller)
@@ -60,18 +61,12 @@ def __init__(
 
     def run(
         self,
-        dispatcher: AsyncioDispatcher,
+        loop: asyncio.AbstractEventLoop,
     ) -> None:
+        dispatcher = AsyncioDispatcher(loop)  # Needs running loop
         builder.LoadDatabase()
         softioc.iocInit(dispatcher)
 
-        if self.options.terminal:
-            context = {
-                "dispatcher": dispatcher,
-                "controller": self._controller,
-            }
-            softioc.interactive_ioc(context)
-
 
 def _add_pvi_info(
     pvi: str,

src/fastcs/transport/epics/options.py

@@ -23,7 +23,6 @@ class EpicsGUIOptions:
 
 @dataclass
 class EpicsIOCOptions:
-    terminal: bool = True
     pv_prefix: str = "MY-DEVICE-PREFIX"