Add background parameter to servers. (#2529)

Author: janiversen

pymodbus/client/base.py

@@ -1,6 +1,7 @@
 """Base for all clients."""
 from __future__ import annotations
 
+import asyncio
 from abc import abstractmethod
 from collections.abc import Awaitable, Callable
 
@@ -57,7 +58,9 @@ async def connect(self) -> bool:
             self.ctx.comm_params.host,
             self.ctx.comm_params.port,
         )
-        return await self.ctx.connect()
+        rc = await self.ctx.connect()
+        await asyncio.sleep(0.1)
+        return rc
 
     def register(self, custom_response_class: type[ModbusPDU]) -> None:
         """Register a custom response class with the decoder (call **sync**).

pymodbus/pdu/pdu.py

@@ -5,6 +5,7 @@
 import struct
 from abc import abstractmethod
 
+from pymodbus.datastore import ModbusSlaveContext
 from pymodbus.exceptions import NotImplementedException
 from pymodbus.logging import Log
 
@@ -79,6 +80,11 @@ def encode(self) -> bytes:
     def decode(self, data: bytes) -> None:
         """Decode data part of the message."""
 
+    async def update_datastore(self, context: ModbusSlaveContext) -> ModbusPDU:
+        """Run request against a datastore."""
+        _ = context
+        return ExceptionResponse(0, 0)
+
     @classmethod
     def calculateRtuFrameSize(cls, data: bytes) -> int:
         """Calculate the size of a PDU."""

pymodbus/server/base.py

@@ -3,6 +3,7 @@
 
 import asyncio
 from collections.abc import Callable
+from contextlib import suppress
 
 from pymodbus.datastore import ModbusServerContext
 from pymodbus.device import ModbusControlBlock, ModbusDeviceIdentification
@@ -67,25 +68,27 @@ async def shutdown(self):
             self.serving.set_result(True)
         self.close()
 
-    async def serve_forever(self):
+    async def serve_forever(self, *, background: bool = False):
         """Start endless loop."""
         if self.transport:
             raise RuntimeError(
                 "Can't call serve_forever on an already running server object"
             )
         await self.listen()
         Log.info("Server listening.")
-        await self.serving
-        Log.info("Server graceful shutdown.")
+        if not background:
+            with suppress(asyncio.exceptions.CancelledError):
+                await self.serving
+            Log.info("Server graceful shutdown.")
 
     def callback_connected(self) -> None:
         """Call when connection is succcesfull."""
+        raise RuntimeError("callback_new_connection should never be called")
 
     def callback_disconnected(self, exc: Exception | None) -> None:
         """Call when connection is lost."""
-        Log.debug("callback_disconnected called: {}", exc)
+        raise RuntimeError("callback_disconnected should never be called")
 
     def callback_data(self, data: bytes, addr: tuple | None = None) -> int:
         """Handle received data."""
-        Log.debug("callback_data called: {} addr={}", data, ":hex", addr)
-        return 0
+        raise RuntimeError("callback_data should never be called")

pymodbus/server/requesthandler.py

@@ -4,7 +4,7 @@
 import asyncio
 import traceback
 
-from pymodbus.exceptions import NoSuchSlaveException
+from pymodbus.exceptions import ModbusIOException, NoSuchSlaveException
 from pymodbus.logging import Log
 from pymodbus.pdu.pdu import ExceptionResponse
 from pymodbus.transaction import TransactionManager
@@ -29,9 +29,6 @@ def __init__(self, owner):
         self.server = owner
         self.framer = self.server.framer(self.server.decoder)
         self.running = False
-        self.handler_task = None  # coroutine to be run on asyncio loop
-        self.databuffer = b''
-        self.loop = asyncio.get_running_loop()
         super().__init__(
             params,
             self.framer,
@@ -44,8 +41,7 @@ def __init__(self, owner):
 
     def callback_new_connection(self) -> ModbusProtocol:
         """Call when listener receive new connection request."""
-        Log.debug("callback_new_connection called")
-        return ServerRequestHandler(self)
+        raise RuntimeError("callback_new_connection should never be called")
 
     def callback_connected(self) -> None:
         """Call when connection is succcesfull."""
@@ -54,27 +50,11 @@ def callback_connected(self) -> None:
         if self.server.broadcast_enable:
             if 0 not in slaves:
                 slaves.append(0)
-        try:
-            self.running = True
-
-            # schedule the connection handler on the event loop
-            self.handler_task = asyncio.create_task(self.handle())
-            self.handler_task.set_name("server connection handler")
-        except Exception as exc:  # pylint: disable=broad-except
-            Log.error(
-                "Server callback_connected exception: {}; {}",
-                exc,
-                traceback.format_exc(),
-            )
 
     def callback_disconnected(self, call_exc: Exception | None) -> None:
         """Call when connection is lost."""
         super().callback_disconnected(call_exc)
         try:
-            if self.handler_task:
-                self.handler_task.cancel()
-            if hasattr(self.server, "on_connection_lost"):
-                self.server.on_connection_lost()
             if call_exc is None:
                 Log.debug(
                     "Handler for stream [{}] has been canceled", self.comm_params.comm_name
@@ -93,66 +73,46 @@ def callback_disconnected(self, call_exc: Exception | None) -> None:
                 traceback.format_exc(),
             )
 
-    async def handle(self) -> None:
-        """Coroutine which represents a single master <=> slave conversation.
-
-        Once the client connection is established, the data chunks will be
-        fed to this coroutine via the asyncio.Queue object which is fed by
-        the ServerRequestHandler class's callback Future.
-
-        This callback future gets data from either asyncio.BaseProtocol.data_received
-        or asyncio.DatagramProtocol.datagram_received.
+    def callback_data(self, data: bytes, addr: tuple | None = None) -> int:
+        """Handle received data."""
+        try:
+            used_len = super().callback_data(data, addr)
+        except ModbusIOException:
+            response = ExceptionResponse(
+                40,
+                exception_code=ExceptionResponse.ILLEGAL_FUNCTION
+            )
+            self.server_send(response, 0)
+            return(len(data))
+        if self.last_pdu:
+            if self.is_server:
+                self.loop.call_soon(self.handle_later)
+            else:
+                self.response_future.set_result(True)
+        return used_len
 
-        This function will execute without blocking in the while-loop and
-        yield to the asyncio event loop when the frame is exhausted.
-        As a result, multiple clients can be interleaved without any
-        interference between them.
-        """
-        while self.running:
-            try:
-                pdu, *addr, exc = await self.server_execute()
-                if exc:
-                    pdu = ExceptionResponse(
-                        40,
-                        exception_code=ExceptionResponse.ILLEGAL_FUNCTION
-                    )
-                    self.server_send(pdu, 0)
-                    continue
-                await self.server_async_execute(pdu, *addr)
-            except asyncio.CancelledError:
-                # catch and ignore cancellation errors
-                if self.running:
-                    Log.debug(
-                        "Handler for stream [{}] has been canceled", self.comm_params.comm_name
-                    )
-                    self.running = False
-            except Exception as exc:  # pylint: disable=broad-except
-                # force TCP socket termination as framer
-                # should handle application layer errors
-                Log.error(
-                    'Unknown exception "{}" on stream {} forcing disconnect',
-                    exc,
-                    self.comm_params.comm_name,
-                )
-                self.close()
-                self.callback_disconnected(exc)
+    def handle_later(self):
+        """Change sync (async not allowed in call_soon) to async."""
+        asyncio.run_coroutine_threadsafe(self.handle_request(), self.loop)
 
-    async def server_async_execute(self, request, *addr):
+    async def handle_request(self):
         """Handle request."""
         broadcast = False
+        if not self.last_pdu:
+            return
         try:
-            if self.server.broadcast_enable and not request.dev_id:
+            if self.server.broadcast_enable and not self.last_pdu.dev_id:
                 broadcast = True
                 # if broadcasting then execute on all slave contexts,
                 # note response will be ignored
                 for dev_id in self.server.context.slaves():
-                    response = await request.update_datastore(self.server.context[dev_id])
+                    response = await self.last_pdu.update_datastore(self.server.context[dev_id])
             else:
-                context = self.server.context[request.dev_id]
-                response = await request.update_datastore(context)
+                context = self.server.context[self.last_pdu.dev_id]
+                response = await self.last_pdu.update_datastore(context)
 
         except NoSuchSlaveException:
-            Log.error("requested slave does not exist: {}", request.dev_id)
+            Log.error("requested slave does not exist: {}", self.last_pdu.dev_id)
             if self.server.ignore_missing_slaves:
                 return  # the client will simply timeout waiting for a response
             response = ExceptionResponse(0x00, ExceptionResponse.GATEWAY_NO_RESPONSE)
@@ -165,9 +125,9 @@ async def server_async_execute(self, request, *addr):
             response = ExceptionResponse(0x00, ExceptionResponse.SLAVE_FAILURE)
         # no response when broadcasting
         if not broadcast:
-            response.transaction_id = request.transaction_id
-            response.dev_id = request.dev_id
-            self.server_send(response, *addr)
+            response.transaction_id = self.last_pdu.transaction_id
+            response.dev_id = self.last_pdu.dev_id
+            self.server_send(response, self.last_addr)
 
     def server_send(self, pdu, addr):
         """Send message."""

pymodbus/server/startstop.py

@@ -3,7 +3,6 @@
 
 import asyncio
 import os
-from contextlib import suppress
 
 from pymodbus.datastore import ModbusServerContext
 from pymodbus.pdu import ModbusPDU
@@ -27,13 +26,13 @@ async def StartAsyncTcpServer(  # pylint: disable=invalid-name
     :parameter context: Datastore object
     :parameter custom_functions: optional list of custom PDU objects
     :parameter kwargs: for parameter explanation see ModbusTcpServer
+
+    .. tip::
+        Only handles a single server !
+
+        Use ModbusTcpServer to allow multiple servers in one app.
     """
-    server = ModbusTcpServer(context, **kwargs)
-    if custom_functions:
-        for func in custom_functions:
-            server.decoder.register(func)
-    with suppress(asyncio.exceptions.CancelledError):
-        await server.serve_forever()
+    await ModbusTcpServer(context, custom_pdu=custom_functions, **kwargs).serve_forever()
 
 
 def StartTcpServer(  # pylint: disable=invalid-name
@@ -46,8 +45,13 @@ def StartTcpServer(  # pylint: disable=invalid-name
     :parameter context: Datastore object
     :parameter custom_functions: optional list of custom PDU objects
     :parameter kwargs: for parameter explanation see ModbusTcpServer
+
+    .. tip::
+        Only handles a single server !
+
+        Use ModbusTcpServer to allow multiple servers in one app.
     """
-    return asyncio.run(StartAsyncTcpServer(context, custom_functions=custom_functions, **kwargs))
+    asyncio.run(StartAsyncTcpServer(context, custom_functions=custom_functions, **kwargs))
 
 
 async def StartAsyncTlsServer(  # pylint: disable=invalid-name
@@ -60,13 +64,13 @@ async def StartAsyncTlsServer(  # pylint: disable=invalid-name
     :parameter context: Datastore object
     :parameter custom_functions: optional list of custom PDU objects
     :parameter kwargs: for parameter explanation see ModbusTlsServer
+
+    .. tip::
+        Only handles a single server !
+
+        Use ModbusTlsServer to allow multiple servers in one app.
     """
-    server = ModbusTlsServer(context, **kwargs)
-    if custom_functions:
-        for func in custom_functions:
-            server.decoder.register(func)
-    with suppress(asyncio.exceptions.CancelledError):
-        await server.serve_forever()
+    await ModbusTlsServer(context, custom_pdu=custom_functions, **kwargs).serve_forever()
 
 
 def StartTlsServer(  # pylint: disable=invalid-name
@@ -79,6 +83,11 @@ def StartTlsServer(  # pylint: disable=invalid-name
     :parameter context: Datastore object
     :parameter custom_functions: optional list of custom PDU objects
     :parameter kwargs: for parameter explanation see ModbusTlsServer
+
+    .. tip::
+        Only handles a single server !
+
+        Use ModbusTlsServer to allow multiple servers in one app.
     """
     asyncio.run(StartAsyncTlsServer(context, custom_functions=custom_functions, **kwargs))
 
@@ -93,13 +102,13 @@ async def StartAsyncUdpServer(  # pylint: disable=invalid-name
     :parameter context: Datastore object
     :parameter custom_functions: optional list of custom PDU objects
     :parameter kwargs: for parameter explanation see ModbusUdpServer
+
+    .. tip::
+        Only handles a single server !
+
+        Use ModbusUdpServer to allow multiple servers in one app.
     """
-    server = ModbusUdpServer(context, **kwargs)
-    if custom_functions:
-        for func in custom_functions:
-            server.decoder.register(func)
-    with suppress(asyncio.exceptions.CancelledError):
-        await server.serve_forever()
+    await ModbusUdpServer(context, custom_pdu=custom_functions, **kwargs).serve_forever()
 
 
 def StartUdpServer(  # pylint: disable=invalid-name
@@ -112,6 +121,11 @@ def StartUdpServer(  # pylint: disable=invalid-name
     :parameter context: Datastore object
     :parameter custom_functions: optional list of custom PDU objects
     :parameter kwargs: for parameter explanation see ModbusUdpServer
+
+    .. tip::
+        Only handles a single server !
+
+        Use ModbusUdpServer to allow multiple servers in one app.
     """
     asyncio.run(StartAsyncUdpServer(context, custom_functions=custom_functions, **kwargs))
 
@@ -126,13 +140,13 @@ async def StartAsyncSerialServer(  # pylint: disable=invalid-name
     :parameter context: Datastore object
     :parameter custom_functions: optional list of custom PDU objects
     :parameter kwargs: for parameter explanation see ModbusSerialServer
+
+    .. tip::
+        Only handles a single server !
+
+        Use ModbusSerialServer to allow multiple servers in one app.
     """
-    server = ModbusSerialServer(context, **kwargs)
-    if custom_functions:
-        for func in custom_functions:
-            server.decoder.register(func)
-    with suppress(asyncio.exceptions.CancelledError):
-        await server.serve_forever()
+    await ModbusSerialServer(context, custom_pdu=custom_functions, **kwargs).serve_forever()
 
 
 def StartSerialServer(  # pylint: disable=invalid-name
@@ -145,6 +159,11 @@ def StartSerialServer(  # pylint: disable=invalid-name
     :parameter context: Datastore object
     :parameter custom_functions: optional list of custom PDU objects
     :parameter kwargs: for parameter explanation see ModbusSerialServer
+
+    .. tip::
+        Only handles a single server !
+
+        Use ModbusSerialServer to allow multiple servers in one app.
     """
     asyncio.run(StartAsyncSerialServer(context, custom_functions=custom_functions, **kwargs))