Discard the late response of a cancelled request

pymodbus/transaction/transaction.py

@@ -65,6 +65,7 @@ def __init__(
             self._lock = asyncio.Lock()
             self.low_level_send = self.send
             self.response_future: asyncio.Future = asyncio.Future()
+            self.cancelled_requests: dict[int, tuple[asyncio.Future, float]] = {}
             self.last_pdu: ModbusPDU | None = None
             self.last_addr: tuple | None = None
 
@@ -181,13 +182,13 @@ def sync_execute(self, no_response_expected: bool, request: ModbusPDU) -> Modbus
             Log.error(txt)
             raise self._io_exception_from_request(txt, request)
 
-    async def execute(
+    async def execute(  # noqa: C901
         self, no_response_expected: bool, request: ModbusPDU
     ) -> ModbusPDU:
         """Execute requests asynchronously.
 
-        REMARK: this method is identical to sync_execute, apart from the lock and try/except.
-                any changes in either method MUST be mirrored !!!
+        REMARK: this method is identical to sync_execute, apart from the lock, try/except
+                and the wait after a cancel. any changes in either method MUST be mirrored !!!
         """
         if not self.transport:
             Log.warning("Not connected, trying to connect!")
@@ -196,6 +197,9 @@ async def execute(
                     "Client cannot connect (automatic retry continuing) !!"
                 )
         async with self._lock:
+            if cancelled := self.cancelled_requests.pop(request.dev_id, None):
+                late_response, until = cancelled
+                await asyncio.wait([late_response], timeout=until - monotonic())
             request.transaction_id = self.getNextTID()
             count_retries = 0
             while count_retries <= self.retries:
@@ -228,6 +232,18 @@ async def execute(
                     return response
                 except asyncio.exceptions.TimeoutError:
                     count_retries += 1
+                except asyncio.exceptions.CancelledError:
+                    # RTU/ASCII responses carry no transaction id, so the next
+                    # request to this device first waits for the response to
+                    # this one. handleFrame skips a response from another
+                    # device, so requests to other devices are not delayed.
+                    if not request.transaction_id:
+                        self.response_future = asyncio.Future()
+                        self.cancelled_requests[request.dev_id] = (
+                            self.response_future,
+                            monotonic() + self.comm_params.timeout_connect,
+                        )
+                    raise
             if self.count_until_disconnect < 0:
                 if self.comm_params.comm_type != CommType.SERIAL:
                     self.connection_lost(asyncio.TimeoutError("Server not responding"))

test/transaction/test_transaction.py

@@ -11,6 +11,10 @@
 from pymodbus.framer import FramerRTU, FramerSocket, FramerType
 from pymodbus.pdu import DecodePDU, ExceptionResponse
 from pymodbus.pdu.bit_message import ReadCoilsRequest, ReadCoilsResponse
+from pymodbus.pdu.register_message import (
+    ReadHoldingRegistersRequest,
+    ReadHoldingRegistersResponse,
+)
 from pymodbus.transaction import TransactionManager
 from pymodbus.transport import CommType
 
@@ -294,6 +298,81 @@ async def test_transaction_execute(self, use_clc, scenario):  # noqa: C901
             assert exc_info.value.dev_id == request.dev_id
             assert exc_info.value.transaction_id == request.transaction_id
 
+    @pytest.mark.parametrize("late_response", [True, False])
+    async def test_transaction_execute_cancel(self, use_clc, late_response):
+        """Test next request does not get the response of a cancelled request."""
+        transact = TransactionManager(
+            use_clc,
+            FramerRTU(DecodePDU(False)),
+            0,
+            False,
+            None,
+            None,
+            None,
+        )
+        transport = mock.Mock()
+        transact.connection_made(transport)
+        transact.comm_params.timeout_connect = 0.5
+        resp = asyncio.create_task(
+            transact.execute(False, ReadHoldingRegistersRequest(dev_id=1, count=4))
+        )
+        await asyncio.sleep(0.1)
+        resp.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await resp
+        resp = asyncio.create_task(
+            transact.execute(False, ReadHoldingRegistersRequest(dev_id=1, count=5))
+        )
+        await asyncio.sleep(0.1)
+        if late_response:
+            late = ReadHoldingRegistersResponse(dev_id=1, registers=[1, 2, 3, 4])
+            transact.data_received(transact.framer.buildFrame(late))
+        while transport.write.call_count < 2:
+            await asyncio.sleep(0.1)
+        response = ReadHoldingRegistersResponse(dev_id=1, registers=[1, 2, 3, 4, 5])
+        transact.data_received(transact.framer.buildFrame(response))
+        assert (await resp).registers == response.registers
+
+    @pytest.mark.parametrize(("framer", "dev_id"), [(FramerRTU, 2), (FramerSocket, 1)])
+    async def test_transaction_execute_cancel_no_wait(self, use_clc, framer, dev_id):
+        """Test no wait after a cancel for a request that cannot get its response.
+
+        That is a request to another device, or any request with a transaction id.
+        """
+        transact = TransactionManager(
+            use_clc,
+            framer(DecodePDU(False)),
+            0,
+            False,
+            None,
+            None,
+            None,
+        )
+        transport = mock.Mock()
+        transact.connection_made(transport)
+        transact.comm_params.timeout_connect = 5
+        resp = asyncio.create_task(
+            transact.execute(False, ReadHoldingRegistersRequest(dev_id=1, count=4))
+        )
+        await asyncio.sleep(0.1)
+        resp.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await resp
+        resp = asyncio.create_task(
+            transact.execute(False, ReadHoldingRegistersRequest(dev_id=dev_id, count=5))
+        )
+        await asyncio.sleep(0.1)
+        assert transport.write.call_count == 2
+        late = ReadHoldingRegistersResponse(
+            dev_id=1, transaction_id=1, registers=[1, 2, 3, 4]
+        )
+        transact.data_received(transact.framer.buildFrame(late))
+        response = ReadHoldingRegistersResponse(
+            dev_id=dev_id, transaction_id=2, registers=[1, 2, 3, 4, 5]
+        )
+        transact.data_received(transact.framer.buildFrame(response))
+        assert (await resp).registers == response.registers
+
     async def test_transaction_receiver(self, use_clc):
         """Test tracers in disconnect."""
         transact = TransactionManager(