Convert adapters to use asyncio instead of the deprecated asynchat/asyncore

doc/conf.py

@@ -4,6 +4,7 @@
 # sphinx-quickstart on Wed Nov  9 16:42:53 2016.
 import os
 import sys
+
 sys.path.insert(0, os.path.abspath("../lewis"))
 
 
@@ -22,11 +23,11 @@
 ]
 templates_path = ["_templates"]
 # General information about the project.
-project = u"lewis"
-language = 'en'
+project = "lewis"
+language = "en"
 exclude_patterns = ["_build", "Thumbs.db", ".DS_Store"]
 # -- Options for HTML output ---------------------------------------------
-suppress_warnings =["docutils"]
+suppress_warnings = ["docutils"]
 html_theme = "sphinx_rtd_theme"
 html_logo = "resources/logo/lewis-logo.png"
 html_context = {

doc/developer_guide/framework_details.md

@@ -72,3 +72,24 @@ statemachine:
 -  Implicit: Implement handlers in the device class, with standard names
    like `on_entry_init` for a state called "init", and call
    `bindHandlersByName()`
+
+## Adapter Concurrency
+
+Adapters performing network I/O for communicating with client applications
+make use of python's [asyncio](https://docs.python.org/3/library/asyncio.html)
+library.
+
+-  Lewis is a multi-threaded application, each adapter is moved on its own
+   dedicated thread, which is isolated from the main simulation thread. 
+-  The main thread uses the following two synchronization tools:
+   -  device lock: ensures that the device is only accessed from one 
+       thread at a time
+   -  is_running event: sends stop request to the adapter thread
+-  Adapters have to implement the following three 
+   [async coroutines](https://docs.python.org/3/library/asyncio-task.html), 
+   which will be scheduled as tasks by their respective async event loops:
+   -  start_server: starts the server, handles client connections
+   -  stop_server: gracefully closes client connections and stops the server
+   -  handle: synchronizes with the simulation steps
+
+![The adapter concurrency diagram.](../resources/diagrams/AdapterConcurrency.drawio.png)

lewis/adapters/epics.py

@@ -1,4 +1,4 @@
 # -*- coding: utf-8 -*-
 # *********************************************************************
 # lewis - a library for creating hardware device simulators
 # Copyright (C) 2016-2021 European Spallation Source ERIC
@@ -90,7 +90,7 @@
     def value(self, new_value) -> None:
         if self.read_only:
             raise AccessViolationException(
                 "The property {} is read only.".format(self._pv.property)
             )
 
         setattr(self._target, self._pv.property, new_value)
@@ -339,9 +339,9 @@
 
         if not self._function_has_n_args(final_callable, 0):
             raise RuntimeError(
                 "The function '{}' does not look like a getter function. A valid getter "
                 "function has no arguments that do not have a default. The self-argument of "
                 "methods does not count towards that number.".format(final_callable.__name__)
             )
 
         @wraps(final_callable)
@@ -367,9 +367,9 @@
 
         if not self._function_has_n_args(func, 1):
             raise RuntimeError(
                 "The function '{}' does not look like a setter function. A valid setter "
                 "function has exactly one argument without a default. The self-argument of "
                 "methods does not count towards that number.".format(func.__name__)
             )
 
         def setter(obj, value) -> None:
@@ -393,8 +393,8 @@
 
             if not func:
                 raise AttributeError(
                     "No method with the name '{}' could be found on any of the target objects "
                     "(device, interface). Please check the spelling.".format(func_name)
                 )
 
         return func
@@ -416,13 +416,13 @@
 @has_log
 class PropertyExposingDriver(Driver):
     def __init__(self, interface, device_lock) -> None:
         super(PropertyExposingDriver, self).__init__()
 
         self._interface = interface
         self._device_lock = device_lock
         self._set_logging_context(interface)
 
         self._timers = {k: 0.0 for k in self._interface.bound_pvs.keys()}
         self._last_update_call = None
 
     def write(self, pv, value) -> bool:
@@ -441,14 +441,14 @@
                 return True
         except LimitViolationException as e:
             self.log.warning(
-                "Rejected writing value %s to PV %s due to limit " "violation. %s",
+                "Rejected writing value %s to PV %s due to limit violation. %s",
                 value,
                 pv,
                 e,
             )
         except AccessViolationException:
             self.log.warning(
-                "Rejected writing value %s to PV %s due to access " "violation, PV is read-only.",
+                "Rejected writing value %s to PV %s due to access violation, PV is read-only.",
                 value,
                 pv,
             )
@@ -483,7 +483,7 @@
 
         :param force: If True, will force updates to all PVs regardless of timers.
         """
         dt = seconds_since(self._last_update_call or datetime.now())
 
         # Cache details of PVs that need to update
         value_updates = []
@@ -509,14 +509,14 @@
         self._process_value_updates(value_updates)
         self._process_meta_updates(meta_updates)
 
         self._last_update_call = datetime.now()
 
     def _process_value_updates(self, updates) -> None:
         if updates:
             update_log = []
             for pv, value in updates:
                 self.setParam(pv, value)
                 update_log.append("{}={}".format(pv, value))
 
             self.log.info("Processed PV updates: %s", ", ".join(update_log))
 
@@ -573,7 +573,7 @@
 
         return "\n\n".join([inspect.getdoc(self.interface) or "", "PVs\n==="] + pvs)
 
-    def start_server(self) -> None:
+    async def start_server(self) -> None:
         """
         Creates a pcaspy-server.
 
@@ -597,15 +597,15 @@
                 ", ".join((self._options.prefix + pv for pv in self.interface.bound_pvs.keys())),
             )
 
-    def stop_server(self) -> None:
+    async def stop_server(self) -> None:
         self._driver = None
         self._server = None
 
     @property
     def is_running(self):
         return self._server is not None
 
-    def handle(self, cycle_delay=0.1) -> None:
+    async def handle(self, cycle_delay=0.1) -> None:
         """
         Call this method to spend about ``cycle_delay`` seconds processing
         requests in the pcaspy server. Under load, for example when running ``caget`` at a

lewis/adapters/modbus.py

@@ -32,8 +32,7 @@
     at lewis/examples/modbus_device.
 """
 
-import asyncore
-import socket
+import asyncio
 import struct
 from copy import deepcopy
 from math import ceil
@@ -237,7 +236,7 @@ def create_exception(self, code):
         frame = deepcopy(self)
         frame.length = 3
         frame.fcode += 0x80
-        frame.data = bytearray(chr(code))
+        frame.data = bytearray([code])
         return frame
 
     def create_response(self, data=None):
@@ -260,23 +259,22 @@ class ModbusProtocol:
     This class implements the Modbus TCP Protocol.
 
     The user of this class should provide a ModbusDataStore instance that will be used to
-    fulfill read and write requests, and a callable `sender` which accepts one bytearray
-    parameter. The `sender` will be called whenever a response frame is generated, with a
-    bytearray containing the response frame as the parameter.
+    fulfill read and write requests. The `writer` will be called whenever a response frame
+    is generated, with a bytearray containing the response frame as the parameter.
 
     Processing occurs when the user calls ModbusProtocol.process(), passing in the raw frame
     data to process as a bytearray. The data may include multiple frames and partial frame
     fragments. Any data that could not be processed (due to incomplete frames) is buffered for
     the next call to process.
 
-    :param sender: callable that accepts one bytearray parameter, called to send responses.
+    :param writer: asyncio.StreamWriter, called to send responses.
     :param datastore: ModbusDataStore instance to reference when processing requests
     """
 
-    def __init__(self, sender, datastore) -> None:
+    def __init__(self, writer: asyncio.StreamWriter, datastore: ModbusDataStore) -> None:
         self._buffer = bytearray()
         self._datastore = datastore
-        self._send = lambda req: sender(req.to_bytearray())
+        self._writer = writer
 
         # Lookup table to handle requests as per Modbus Application Protocol v1.1b3, Section 6.
         self._fcode_handler_map = {
@@ -290,7 +288,11 @@ def __init__(self, sender, datastore) -> None:
             0x10: self._handle_write_multiple_registers,
         }
 
-    def process(self, data, device_lock) -> None:
+    async def _send(self, response) -> None:
+        self._writer.write(response.to_bytearray())
+        await self._writer.drain()
+
+    async def process(self, data, device_lock) -> None:
         """
         Process as much of given data as possible.
 
@@ -302,22 +304,22 @@ def process(self, data, device_lock) -> None:
         """
         self._buffer.extend(bytearray(data))
 
+        responses = []
         with device_lock:
             for request in self._buffered_requests():
-                self.log.debug(
-                    "Request: %s",
-                    str(["{:#04x}".format(c) for c in request.to_bytearray()]),
-                )
-
                 handler = self._get_handler(request.fcode)
-                response = handler(request)
-
-                self.log.debug(
-                    "Response: %s",
-                    str(["{:#04x}".format(c) for c in response.to_bytearray()]),
-                )
+                responses.append((request, handler(request)))
 
-                self._send(response)
+        for request, response in responses:
+            self.log.debug(
+                "Request: %s",
+                str(["{:#04x}".format(c) for c in request.to_bytearray()]),
+            )
+            self.log.debug(
+                "Response: %s",
+                str(["{:#04x}".format(c) for c in response.to_bytearray()]),
+            )
+            await self._send(response)
 
     def _buffered_requests(self):
         """Generator to yield all complete modbus requests in the internal buffer"""
@@ -529,59 +531,104 @@ def _handle_write_multiple_registers(self, request):
 
 
 @has_log
-class ModbusHandler(asyncore.dispatcher_with_send):
-    def __init__(self, sock, interface, server) -> None:
-        asyncore.dispatcher_with_send.__init__(self, sock=sock)
+class ModbusHandler:
+    def __init__(
+        self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter, interface, server
+    ) -> None:
         self._datastore = ModbusDataStore(interface.di, interface.co, interface.ir, interface.hr)
-        self._modbus = ModbusProtocol(self.send, self._datastore)
+        self._modbus = ModbusProtocol(writer, self._datastore)
         self._server = server
+        self._reader = reader
+        self._writer = writer
+        self._closing = False
 
         self._set_logging_context(interface)
-        self.log.info("Client connected from %s:%s", *sock.getpeername())
 
-    def handle_read(self) -> None:
-        data = self.recv(8192)
-        self._modbus.process(data, self._server.device_lock)
-
-    def handle_close(self) -> None:
-        self.log.info("Closing connection to client %s:%s", *self.socket.getpeername())
+    async def handle_client(self) -> None:
+        try:
+            while True:
+                data = await self._reader.read(8192)
+                if data:
+                    await self._modbus.process(data, self._server.device_lock)
+                else:
+                    break
+        except OSError as e:
+            self.log.error("Connection error: %s", e)
+        finally:
+            await self.handle_close()
+
+    async def handle_close(self) -> None:
+        if self._closing:
+            return
+        self._closing = True
+        sock = self._writer.get_extra_info("socket")
+        if sock is not None:
+            try:
+                self.log.info("Closing connection to client %s:%s", *sock.getpeername())
+            except OSError:
+                self.log.info("Closing connection to client (peer address unavailable)")
+        if not self._writer.is_closing():
+            self._writer.close()
+        try:
+            await self._writer.wait_closed()
+        except OSError:
+            self.log.debug("Connection reset by peer while waiting for close")
         self._server.remove_handler(self)
-        self.close()
 
 
 @has_log
-class ModbusServer(asyncore.dispatcher):
+class ModbusServer:
     def __init__(self, host, port, interface, device_lock) -> None:
-        asyncore.dispatcher.__init__(self)
+        self.host = host
+        self.port = port
         self.device_lock = device_lock
         self.interface = interface
-        self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
-        self.set_reuse_addr()
-        self.bind((host, port))
-        self.listen(5)
+        self._server = None
 
         self._set_logging_context(interface)
-        self.log.info("Listening on %s:%s", host, port)
 
         self._accepted_connections = []
 
-    def handle_accept(self) -> None:
-        pair = self.accept()
-        if pair is not None:
-            sock, _ = pair
-            handler = ModbusHandler(sock, self.interface, self)
-            self._accepted_connections.append(handler)
+    async def start(self):
+        self._server = await asyncio.start_server(
+            self._handle_accept,
+            host=self.host,
+            port=self.port,
+            backlog=5,
+            reuse_address=True,
+            start_serving=True,
+        )
+        self.log.info("Listening on %s:%s", self.host, self.port)
+
+    async def _handle_accept(
+        self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter
+    ) -> None:
+        sock = writer.get_extra_info("socket")
+        if sock is not None:
+            try:
+                self.log.info("Client connected from %s:%s", *sock.getpeername())
+            except OSError:
+                self.log.info("Client connected (peer address unavailable)")
+        handler = ModbusHandler(reader, writer, self.interface, self)
+        self._accepted_connections.append(handler)
+        await handler.handle_client()
 
     def remove_handler(self, handler) -> None:
-        self._accepted_connections.remove(handler)
+        try:
+            self._accepted_connections.remove(handler)
+        except ValueError:
+            pass  # Removed from another path
 
-    def handle_close(self) -> None:
-        self.log.info("Shutting down server, closing all remaining client connections.")
+    async def close(self) -> None:
+        if self._server is not None:
+            self.log.info("Shutting down server, closing all remaining client connections.")
+            self._server.close()
 
-        for handler in self._accepted_connections:
-            handler.close()
-        self._accepted_connections = []
-        self.close()
+            for handler in list(self._accepted_connections):
+                await handler.handle_close()
+
+            self._accepted_connections = []
+            await self._server.wait_closed()
 
 
 class ModbusAdapter(Adapter):
@@ -591,25 +638,27 @@ def __init__(self, options=None) -> None:
         super(ModbusAdapter, self).__init__(options)
         self._server = None
 
-    def start_server(self) -> None:
+    async def start_server(self) -> None:
         self._server = ModbusServer(
             self._options.bind_address,
             self._options.port,
             self.interface,
             self.device_lock,
         )
 
-    def stop_server(self) -> None:
+        await self._server.start()
+
+    async def stop_server(self) -> None:
         if self._server is not None:
-            self._server.close()
+            await self._server.close()
             self._server = None
 
     @property
     def is_running(self):
         return self._server is not None
 
-    def handle(self, cycle_delay=0.1) -> None:
-        asyncore.loop(cycle_delay, count=1)
+    async def handle(self, cycle_delay=0.1) -> None:
+        await asyncio.sleep(cycle_delay)
 
 
 class ModbusInterface(InterfaceBase):