Test stop of server task. (#1256)

Author: janiversen

pymodbus/server/async_io.py

@@ -2,7 +2,6 @@
 # pylint: disable=missing-type-doc
 import asyncio
 import logging
-import platform
 import ssl
 import traceback
 from binascii import b2a_hex
@@ -500,7 +499,6 @@ def __init__(
         address=None,
         handler=None,
         allow_reuse_address=False,
-        allow_reuse_port=False,
         defer_start=False,
         backlog=20,
         **kwargs,
@@ -519,8 +517,6 @@ def __init__(
                         receives connection create/teardown events
         :param allow_reuse_address: Whether the server will allow the
                         reuse of an address.
-        :param allow_reuse_port: Whether the server will allow the
-                        reuse of a port.
         :param backlog:  is the maximum number of queued connections
                     passed to listen(). Defaults to 20, increase if many
                     connections are being made and broken to your Modbus slave
@@ -556,7 +552,6 @@ def __init__(
         self.server = None
         self.factory_parms = {
             "reuse_address": allow_reuse_address,
-            "reuse_port": allow_reuse_port,
             "backlog": backlog,
             "start_serving": not defer_start,
         }
@@ -621,7 +616,6 @@ def __init__(  # pylint: disable=too-many-arguments
         reqclicert=False,
         handler=None,
         allow_reuse_address=False,
-        allow_reuse_port=False,
         defer_start=False,
         backlog=20,
         **kwargs,
@@ -646,8 +640,6 @@ def __init__(  # pylint: disable=too-many-arguments
                         receives connection create/teardown events
         :param allow_reuse_address: Whether the server will allow the
                         reuse of an address.
-        :param allow_reuse_port: Whether the server will allow the
-                        reuse of a port.
         :param backlog:  is the maximum number of queued connections
                     passed to listen(). Defaults to 20, increase if many
                     connections are being made and broken to your Modbus slave
@@ -665,7 +657,6 @@ def __init__(  # pylint: disable=too-many-arguments
             address=address,
             handler=handler,
             allow_reuse_address=allow_reuse_address,
-            allow_reuse_port=allow_reuse_port,
             defer_start=defer_start,
             backlog=backlog,
             **kwargs,
@@ -689,7 +680,6 @@ def __init__(
         identity=None,
         address=None,
         handler=None,
-        allow_reuse_port=False,
         defer_start=False,  # pylint: disable=unused-argument
         backlog=20,  # pylint: disable=unused-argument
         **kwargs,
@@ -731,12 +721,10 @@ def __init__(
         self.protocol = None
         self.endpoint = None
         self.on_connection_terminated = None
-        self.stop_serving = self.loop.create_future()
         # asyncio future that will be done once server has started
         self.serving = self.loop.create_future()
         self.factory_parms = {
             "local_addr": self.address,
-            "reuse_port": allow_reuse_port,
             "allow_broadcast": True,
         }
 
@@ -749,9 +737,12 @@ async def serve_forever(self):
                     **self.factory_parms,
                 )
             except asyncio.exceptions.CancelledError:
-                pass
+                raise
+            except Exception as exc:
+                txt = f"Server unexpected exception {exc}"
+                _logger.error(txt)
+                raise RuntimeError(exc) from exc
             self.serving.set_result(True)
-            await self.stop_serving
         else:
             raise RuntimeError(
                 "Can't call serve_forever on an already running server object"
@@ -765,13 +756,10 @@ async def server_close(self):
         """Close server."""
         if self.endpoint:
             self.endpoint.running = False
-        if not self.stop_serving.done():
-            self.stop_serving.set_result(True)
         if self.endpoint is not None and self.endpoint.handler_task is not None:
             self.endpoint.handler_task.cancel()
         if self.protocol is not None:
             self.protocol.close()
-            # TBD await self.protocol.wait_closed()
             self.protocol = None
 
 
@@ -812,6 +800,7 @@ def __init__(
         :param response_manipulator: Callback method for
                     manipulating the response
         """
+        self.loop = kwargs.get("loop") or asyncio.get_event_loop()
         self.bytesize = kwargs.get("bytesize", Defaults.Bytesize)
         self.parity = kwargs.get("parity", Defaults.Parity)
         self.baudrate = kwargs.get("baudrate", Defaults.Baudrate)
@@ -862,7 +851,7 @@ async def _connect(self):
             return
         try:
             self.transport, self.protocol = await create_serial_connection(
-                asyncio.get_event_loop(),
+                self.loop,
                 lambda: self.handler(self),
                 self.device,
                 baudrate=self.baudrate,
@@ -887,44 +876,56 @@ def on_connection_lost(self):
             self.transport.close()
             self.transport = None
             self.protocol = None
-
-        self._check_reconnect()
+        if self.server is None:
+            self._check_reconnect()
 
     async def shutdown(self):
         """Terminate server."""
         if self.transport is not None:
-            self.transport.close()
-            self.transport = None
-            self.protocol = None
+            self.transport.abort()
+        if self.server is not None:
+            self.server.close()
+            await asyncio.wait_for(self.server.wait_closed(), 10)
+        self.server = None
+        self.transport = None
+        self.protocol = None
 
     def _check_reconnect(self):
         """Check reconnect."""
         txt = f"checking autoreconnect {self.auto_reconnect} {self.reconnecting_task}"
         _logger.debug(txt)
         if self.auto_reconnect and (self.reconnecting_task is None):
             _logger.debug("Scheduling serial connection reconnect")
-            loop = asyncio.get_event_loop()
-            self.reconnecting_task = loop.create_task(self._delayed_connect())
+            self.reconnecting_task = self.loop.create_task(self._delayed_connect())
 
     async def serve_forever(self):
         """Start endless loop."""
+        if self.server:
+            raise RuntimeError(
+                "Can't call serve_forever on an already running server object"
+            )
         if self.device.startswith("socket:"):
             # Socket server means listen so start a socket server
-            parts = self.device[7:].split(":")
-            host_port = ("", int(parts[1]))
-            self.server = await asyncio.get_event_loop().create_server(
+            parts = self.device[9:].split(":")
+            host_addr = (parts[0], int(parts[1]))
+            self.server = await self.loop.create_server(
                 lambda: self.handler(self),
-                *host_port,
+                *host_addr,
                 reuse_address=True,
-                reuse_port=True,
                 start_serving=True,
                 backlog=20,
             )
-            await self.server.serve_forever()
+            try:
+                await self.server.serve_forever()
+            except asyncio.exceptions.CancelledError:
+                raise
+            except Exception as exc:  # pylint: disable=broad-except
+                txt = f"Server unexpected exception {exc}"
+                _logger.error(txt)
             return
 
-        while True:
-            await asyncio.sleep(360)
+        while self.server or self.transport or self.protocol:
+            await asyncio.sleep(10)
 
 
 # --------------------------------------------------------------------------- #
@@ -951,64 +952,50 @@ def __init__(self, server, custom_functions, register):
         self.job_stop = asyncio.Event()
         self.job_is_stopped = asyncio.Event()
         self.task = None
+        self.loop = asyncio.get_event_loop()
 
     @classmethod
     def get_server(cls):
         """Get server at index."""
-        return cls._servers[-1]
+        return cls._servers[-1] if cls._servers else None
 
     def _remove(self):
         """Remove server from active list."""
         server = self._servers[-1]
         self._servers.pop()
         del server
 
+    async def _run(self):
+        """Help starting/stopping server."""
+        # self.task = asyncio.create_task(self.server.serve_forever())
+        # await self.job_stop.wait()
+        # await self.server.shutdown()
+        # await asyncio.sleep(0.1)
+        # self.task.cancel()
+        # await asyncio.sleep(0.1)
+        # try:
+        #     await asyncio.wait_for(self.task, 10)
+        # except asyncio.CancelledError:
+        #     pass
+        # self.job_is_stopped.set()
+
     async def run(self):
         """Help starting/stopping server."""
         try:
-            self.task = asyncio.create_task(self.server.serve_forever())
-        except Exception as exc:  # pylint: disable=broad-except
-            txt = f"Server caught exception: {exc}"
-            _logger.error(txt)
-        await self.job_stop.wait()
-        await self.server.shutdown()
-        await asyncio.sleep(0.1)
-        self.task.cancel()
-        await asyncio.sleep(0.1)
-        try:
-            await asyncio.wait_for(self.task, 10)
+            # await self._run()
+            await self.server.serve_forever()
         except asyncio.CancelledError:
             pass
-        if platform.system().lower() == "windows":
-            owntask = asyncio.current_task()
-            for task in asyncio.all_tasks():
-                if task != owntask:
-                    task.cancel()
-                    try:
-                        await asyncio.wait_for(task, 10)
-                    except asyncio.CancelledError:
-                        pass
-        self.job_is_stopped.set()
-
-    def request_stop(self):
-        """Request server stop."""
-        self.job_stop.set()
 
     async def async_await_stop(self):
         """Wait for server stop."""
-        try:
-            await self.job_is_stopped.wait()
-        except asyncio.exceptions.CancelledError:
-            pass
-        self._remove()
-
-    def await_stop(self):
-        """Wait for server stop."""
-        for i in range(30):  # Loop for 3 seconds
-            sleep(0.1)  # in steps of 100 milliseconds.
-            if self.job_is_stopped.is_set():
-                break
-        self._remove()
+        await self.server.shutdown()
+        # self.job_stop.set()
+        # try:
+        #    await asyncio.wait_for(self.job_is_stopped.wait(), 60)
+        # except asyncio.exceptions.CancelledError:
+        #    pass
+        # self._remove()
 
 
 async def StartAsyncTcpServer(  # pylint: disable=invalid-name,dangerous-default-value
@@ -1035,10 +1022,10 @@ async def StartAsyncTcpServer(  # pylint: disable=invalid-name,dangerous-default
     server = ModbusTcpServer(
         context, kwargs.pop("framer", ModbusSocketFramer), identity, address, **kwargs
     )
-    job = _serverList(server, custom_functions, not defer_start)
-    if defer_start:
-        return server
-    await job.run()
+    if not defer_start:
+        job = _serverList(server, custom_functions, not defer_start)
+        await job.run()
+    return server
 
 
 async def StartAsyncTlsServer(  # pylint: disable=invalid-name,dangerous-default-value,too-many-arguments
@@ -1051,7 +1038,6 @@ async def StartAsyncTlsServer(  # pylint: disable=invalid-name,dangerous-default
     password=None,
     reqclicert=False,
     allow_reuse_address=False,
-    allow_reuse_port=False,
     custom_functions=[],
     defer_start=False,
     **kwargs,
@@ -1068,7 +1054,6 @@ async def StartAsyncTlsServer(  # pylint: disable=invalid-name,dangerous-default
     :param reqclicert: Force the sever request client's certificate
     :param allow_reuse_address: Whether the server will allow the reuse of an
                                 address.
-    :param allow_reuse_port: Whether the server will allow the reuse of a port.
     :param custom_functions: An optional list of custom function classes
         supported by server instance.
     :param defer_start: if set, the server object will be returned ready to start.
@@ -1088,13 +1073,12 @@ async def StartAsyncTlsServer(  # pylint: disable=invalid-name,dangerous-default
         password,
         reqclicert,
         allow_reuse_address=allow_reuse_address,
-        allow_reuse_port=allow_reuse_port,
         **kwargs,
     )
-    job = _serverList(server, custom_functions, not defer_start)
-    if defer_start:
-        return server
-    await job.run()
+    if not defer_start:
+        job = _serverList(server, custom_functions, not defer_start)
+        await job.run()
+    return server
 
 
 async def StartAsyncUdpServer(  # pylint: disable=invalid-name,dangerous-default-value
@@ -1120,10 +1104,10 @@ async def StartAsyncUdpServer(  # pylint: disable=invalid-name,dangerous-default
     server = ModbusUdpServer(
         context, kwargs.pop("framer", ModbusSocketFramer), identity, address, **kwargs
     )
-    job = _serverList(server, custom_functions, not defer_start)
-    if defer_start:
-        return server
-    await job.run()
+    if not defer_start:
+        job = _serverList(server, custom_functions, not defer_start)
+        await job.run()
+    return server
 
 
 async def StartAsyncSerialServer(  # pylint: disable=invalid-name,dangerous-default-value
@@ -1147,11 +1131,10 @@ async def StartAsyncSerialServer(  # pylint: disable=invalid-name,dangerous-defa
     server = ModbusSerialServer(
         context, kwargs.pop("framer", ModbusAsciiFramer), identity=identity, **kwargs
     )
-    job = _serverList(server, custom_functions, not defer_start)
-    if defer_start:
-        return server
-    await server.start()
-    await job.run()
+    if not defer_start:
+        job = _serverList(server, custom_functions, not defer_start)
+        await job.run()
+    return server
 
 
 def StartSerialServer(**kwargs):  # pylint: disable=invalid-name
@@ -1176,13 +1159,18 @@ def StartUdpServer(**kwargs):  # pylint: disable=invalid-name
 
 async def ServerAsyncStop():  # pylint: disable=invalid-name
     """Terminate server."""
-    my_job = _serverList.get_server()
-    my_job.request_stop()
-    await my_job.async_await_stop()
+    if my_job := _serverList.get_server():
+        await my_job.async_await_stop()
+        await asyncio.sleep(0.1)
+    else:
+        raise RuntimeError("ServerAsyncStop called without server task active.")
 
 
 def ServerStop():  # pylint: disable=invalid-name
     """Terminate server."""
-    my_job = _serverList.get_server()
-    my_job.request_stop()
-    my_job.await_stop()
+    if my_job := _serverList.get_server():
+        if my_job.loop.is_running():
+            asyncio.run_coroutine_threadsafe(my_job.async_await_stop(), my_job.loop)
+            sleep(0.1)
+    else:
+        raise RuntimeError("ServerStop called without server task active.")